pyrefcnt: reimplement QueuedObjects with static table

This is a simpler version of the "QueuedObjects" list for biased
reference counting with a statically allocated hashtable and a
per-bucket lock. The design is similar to parking_lot.c.

This also addresses the bug where an object could be deallocated
while still in the queue.

Fixes #66

Author: colesbury

Include/internal/pycore_pystate.h

@@ -9,6 +9,7 @@ extern "C" {
 #endif
 
 #include "pycore_gil.h"       /* struct _gil_runtime_state  */
+#include "pycore_llist.h"
 #include "pycore_pymem.h"     /* struct _gc_runtime_state */
 #include "pycore_warnings.h"  /* struct _warnings_runtime_state */
 #include "lock.h"
@@ -127,9 +128,6 @@ struct _is {
 
     PyObject *dict;  /* Stores per-interpreter state */
 
-    struct _Py_hashtable_t *object_queues;
-    pthread_rwlock_t object_queues_lk;
-
     PyObject *builtins_copy;
     PyObject *import_func;
     /* Initialized to PyEval_EvalFrameDefault(). */
@@ -364,6 +362,8 @@ struct PyThreadStateWaiter {
     int64_t time_to_be_fair;
 };
 
+struct brc_queued_object;
+
 struct PyThreadStateOS {
     struct PyThreadStateWaiter waiter;
 
@@ -373,6 +373,12 @@ struct PyThreadStateOS {
     PyCOND_T waiter_cond;
     int waiter_counter;
 
+    struct _PyBrcState {
+        struct llist_node node;
+        uintptr_t thread_id;
+        struct brc_queued_object *queue;
+    } brc;
+
     /* DEBUG info */
     PyThreadState *last_notifier;
     const char *last_notifier_msg;

Include/internal/pycore_refcnt.h

@@ -15,6 +15,7 @@ void _Py_queue_object(PyObject *ob);
 void _Py_queue_process(PyThreadState *tstate);
 void _Py_queue_create(PyThreadState *tstate);
 void _Py_queue_destroy(PyThreadState *tstate);
+void _Py_queue_after_fork(PyThreadState *tstate);
 
 #ifdef __cplusplus
 }

Include/object.h

@@ -418,7 +418,7 @@ PyAPI_FUNC(void) _Py_NO_INLINE _Py_IncRefShared(PyObject *);
 PyAPI_FUNC(void) _Py_NO_INLINE _Py_DecRefShared(PyObject *);
 PyAPI_FUNC(int) _Py_TryIncRefShared(PyObject *);
 PyAPI_FUNC(void) _Py_MergeZeroRefcount(PyObject *);
-void _Py_ExplicitMergeRefcount(PyObject *);
+Py_ssize_t _Py_ExplicitMergeRefcount(PyObject *);
 
 static inline uintptr_t
 _Py_ThreadId(void)

Objects/object.c

@@ -2216,27 +2216,29 @@ _Py_Dealloc(PyObject *op)
 void
 _Py_MergeZeroRefcount(PyObject *op)
 {
-    Py_ssize_t refcount;
-    int queued, merged;
-    int ok;
+    assert(_Py_atomic_load_uint32_relaxed(&op->ob_ref_local) == 0);
 
-    assert(!_Py_REF_IS_IMMORTAL(op->ob_ref_local));
+    _Py_atomic_store_uintptr_relaxed(&op->ob_tid, 0);
 
-    do {
-        uint32_t shared;
-        shared = _Py_atomic_load_uint32_relaxed(&op->ob_ref_shared);
+    Py_ssize_t refcount;
+    for (;;) {
+        uint32_t shared = _Py_atomic_load_uint32_relaxed(&op->ob_ref_shared);
+
+        int queued, merged;
         _PyRef_UnpackShared(shared, &refcount, &queued, &merged);
+
+        // The object can't have negative shared reference count
+        // at this point because that would imply a negative total reference
+        // count (since the local refcount is zero).
+        assert(refcount >= 0);
         assert(!merged);
 
         if (queued) {
-            /* Note that the object can't have negative shared reference count
-            * at this point because that would imply a negative total reference
-            * count (since the local refcount is zero). However, it may still
-            * be queued, if the shared reference count was temporarily negative
-            * and hasn't been proceessed yet. We need to fix this case, because
-            * currently the queue points to garbage objects. FIXME(sgross).
-            */
-            Py_FatalError("UH OH OBJECT IS QUEUED");
+            // Note that the object may still  be queued, if the shared reference
+            // count was temporarily negative and hasn't been proceessed yet.
+            // We don't want to merge it yet because that might result in the
+            // object being freed while it's still in the queue.
+            break;
         }
 
         if (refcount == 0 &&
@@ -2253,21 +2255,22 @@ _Py_MergeZeroRefcount(PyObject *op)
             return;
         }
 
-        ok = _Py_atomic_compare_exchange_uint32(
+        int ok = _Py_atomic_compare_exchange_uint32(
             &op->ob_ref_shared,
             shared,
             shared | _Py_REF_MERGED_MASK);
-    } while (!ok);
 
-    _Py_atomic_store_uintptr_relaxed(&op->ob_tid, 0);
-    _Py_atomic_store_uint32_relaxed(&op->ob_ref_local, 0);
+        if (ok) {
+            break;
+        }
+    }
 
     if (refcount == 0) {
         _Py_Dealloc(op);
     }
 }
 
-void
+Py_ssize_t
 _Py_ExplicitMergeRefcount(PyObject *op)
 {
     uint32_t old_shared;
@@ -2283,7 +2286,8 @@ _Py_ExplicitMergeRefcount(PyObject *op)
         _PyRef_UnpackShared(old_shared, &refcount, &queued, &merged);
 
         if (merged) {
-            return;
+            assert(refcount > 0);
+            return refcount;
         }
 
         // TODO(sgross): implementation defined behavior!
@@ -2304,10 +2308,7 @@ _Py_ExplicitMergeRefcount(PyObject *op)
 
     _Py_atomic_store_uint32_relaxed(&op->ob_ref_local, 0);
     _Py_atomic_store_uintptr_relaxed(&op->ob_tid, 0);
-
-    if (refcount == 0) {
-        _Py_Dealloc(op);
-    }
+    return refcount;
 }
 
 void
@@ -2375,7 +2376,10 @@ _Py_DecRefShared(PyObject *op)
     if (_Py_REF_IS_QUEUED(new_shared) != _Py_REF_IS_QUEUED(old_shared)) {
         PyThreadState *tstate = _PyThreadState_GET();
         if (tstate->interp->gc.collecting) {
-            _Py_ExplicitMergeRefcount(op);
+            Py_ssize_t refcount = _Py_ExplicitMergeRefcount(op);
+            if (refcount == 0) {
+                _Py_Dealloc(op);
+            }
         }
         else {
             _Py_queue_object(op);

Python/ceval.c

@@ -312,11 +312,6 @@ _PyEval_ReInitThreads(_PyRuntimeState *runtime)
         Py_FatalError("Can't initialize threads for pending calls");
     }
 
-    PyInterpreterState *interp = current_tstate->interp;
-    if (pthread_rwlock_init(&interp->object_queues_lk, NULL) != 0) {
-        Py_FatalError("Can't initialize object queues mutex");
-    }
-
     /* Destroy all threads except the current one */
     _PyThreadState_DeleteExcept(runtime, current_tstate);
 }