Queue decrefs

Author: DinoV

Include/internal/pycore_pymem.h

@@ -132,6 +132,13 @@ static inline void _PyObject_XDecRefDelayed(PyObject *obj)
 // Periodically process delayed free requests.
 extern void _PyMem_ProcessDelayed(PyThreadState *tstate);
 
+
+// Periodically process delayed free requests when the world is stopped.
+// Notify of any objects whic should be freeed.
+typedef void (*delayed_dealloc_cb)(PyObject *, void *);
+extern void _PyMem_ProcessDelayedNoDealloc(PyThreadState *tstate, 
+                                           delayed_dealloc_cb cb, void *state);
+
 // Abandon all thread-local delayed free requests and push them to the
 // interpreter's queue.
 extern void _PyMem_AbandonDelayed(PyThreadState *tstate);

Objects/obmalloc.c

@@ -1093,10 +1093,20 @@ struct _mem_work_chunk {
 };
 
 static void
-free_work_item(uintptr_t ptr)
+free_work_item(uintptr_t ptr, delayed_dealloc_cb cb, void *state)
 {
     if (ptr & 0x01) {
-        Py_DECREF((PyObject*)(char *)(ptr - 1));
+        PyObject *obj = (PyObject*)(char *)(ptr - 1);
+        if (cb == NULL) {
+            assert(!_PyInterpreterState_GET()->stoptheworld.world_stopped);
+            Py_DECREF(obj);
+            return;
+        }
+
+        Py_ssize_t refcount = _Py_ExplicitMergeRefcount(obj, -1);
+        if (refcount == 0) {
+            cb(obj, state);
+        }
     }
     else {
         PyMem_Free((void *)ptr);
@@ -1107,15 +1117,16 @@ static void
 free_delayed(uintptr_t ptr)
 {
 #ifndef Py_GIL_DISABLED
-    free_work_item(ptr);
+    free_work_item(ptr, NULL, NULL);
 #else
     PyInterpreterState *interp = _PyInterpreterState_GET();
     if (_PyInterpreterState_GetFinalizing(interp) != NULL ||
         interp->stoptheworld.world_stopped)
     {
         // Free immediately during interpreter shutdown or if the world is
         // stopped.
-        free_work_item(ptr);
+        assert(!interp->stoptheworld.world_stopped || !(ptr & 0x01));
+        free_work_item(ptr, NULL, NULL);
         return;
     }
 
@@ -1142,7 +1153,8 @@ free_delayed(uintptr_t ptr)
     if (buf == NULL) {
         // failed to allocate a buffer, free immediately
         _PyEval_StopTheWorld(tstate->base.interp);
-        free_work_item(ptr);
+        // TODO: Fix me
+        free_work_item(ptr, NULL, NULL);
         _PyEval_StartTheWorld(tstate->base.interp);
         return;
     }
@@ -1185,7 +1197,7 @@ work_queue_first(struct llist_node *head)
 
 static void
 process_queue(struct llist_node *head, struct _qsbr_thread_state *qsbr,
-              bool keep_empty)
+              bool keep_empty, delayed_dealloc_cb cb, void *state)
 {
     while (!llist_empty(head)) {
         struct _mem_work_chunk *buf = work_queue_first(head);
@@ -1196,7 +1208,7 @@ process_queue(struct llist_node *head, struct _qsbr_thread_state *qsbr,
                 return;
             }
 
-            free_work_item(item->ptr);
+            free_work_item(item->ptr, cb, state);
             buf->rd_idx++;
         }
 
@@ -1214,15 +1226,16 @@ process_queue(struct llist_node *head, struct _qsbr_thread_state *qsbr,
 
 static void
 process_interp_queue(struct _Py_mem_interp_free_queue *queue,
-                     struct _qsbr_thread_state *qsbr)
+                     struct _qsbr_thread_state *qsbr, delayed_dealloc_cb cb,
+                     void *state)
 {
     if (!_Py_atomic_load_int_relaxed(&queue->has_work)) {
         return;
     }
 
     // Try to acquire the lock, but don't block if it's already held.
     if (_PyMutex_LockTimed(&queue->mutex, 0, 0) == PY_LOCK_ACQUIRED) {
-        process_queue(&queue->head, qsbr, false);
+        process_queue(&queue->head, qsbr, false, cb, state);
 
         int more_work = !llist_empty(&queue->head);
         _Py_atomic_store_int_relaxed(&queue->has_work, more_work);
@@ -1238,10 +1251,23 @@ _PyMem_ProcessDelayed(PyThreadState *tstate)
     _PyThreadStateImpl *tstate_impl = (_PyThreadStateImpl *)tstate;
 
     // Process thread-local work
-    process_queue(&tstate_impl->mem_free_queue, tstate_impl->qsbr, true);
+    process_queue(&tstate_impl->mem_free_queue, tstate_impl->qsbr, true, NULL, NULL);
+
+    // Process shared interpreter work
+    process_interp_queue(&interp->mem_free_queue, tstate_impl->qsbr, NULL, NULL);
+}
+
+void
+_PyMem_ProcessDelayedNoDealloc(PyThreadState *tstate, delayed_dealloc_cb cb, void *state)
+{
+    PyInterpreterState *interp = tstate->interp;
+    _PyThreadStateImpl *tstate_impl = (_PyThreadStateImpl *)tstate;
+
+    // Process thread-local work
+    process_queue(&tstate_impl->mem_free_queue, tstate_impl->qsbr, true, cb, state);
 
     // Process shared interpreter work
-    process_interp_queue(&interp->mem_free_queue, tstate_impl->qsbr);
+    process_interp_queue(&interp->mem_free_queue, tstate_impl->qsbr, cb, state);
 }
 
 void
@@ -1283,7 +1309,7 @@ _PyMem_FiniDelayed(PyInterpreterState *interp)
             // Free the remaining items immediately. There should be no other
             // threads accessing the memory at this point during shutdown.
             struct _mem_work_item *item = &buf->array[buf->rd_idx];
-            free_work_item(item->ptr);
+            free_work_item(item->ptr, NULL, NULL);
             buf->rd_idx++;
         }
 

Python/gc_free_threading.c

@@ -417,7 +417,25 @@ merge_queued_objects(_PyThreadStateImpl *tstate, struct collection_state *state)
 }
 
 static void
-process_delayed_frees(PyInterpreterState *interp)
+queue_freed_object(PyObject *obj, void *arg)
+{
+    struct collection_state *state = (struct collection_state *)arg;
+
+    // GC objects with zero refcount are handled subsequently by the
+    // GC as if they were cyclic trash, but we have to handle dead
+    // non-GC objects here. Add one to the refcount so that we can
+    // decref and deallocate the object once we start the world again.
+    if (!_PyObject_GC_IS_TRACKED(obj)) {
+        obj->ob_ref_shared += (1 << _Py_REF_SHARED_SHIFT);
+    #ifdef Py_REF_DEBUG
+        _Py_IncRefTotal(_PyThreadState_GET());
+    #endif
+        worklist_push(&state->objs_to_decref, obj);
+    }
+}
+
+static void
+process_delayed_frees(PyInterpreterState *interp, struct collection_state *state)
 {
     // In STW status, we can observe the latest write sequence by
     // advancing the write sequence immediately.
@@ -426,8 +444,9 @@ process_delayed_frees(PyInterpreterState *interp)
     _Py_qsbr_quiescent_state(current_tstate->qsbr);
     HEAD_LOCK(&_PyRuntime);
     PyThreadState *tstate = interp->threads.head;
+
     while (tstate != NULL) {
-        _PyMem_ProcessDelayed(tstate);
+        _PyMem_ProcessDelayedNoDealloc(tstate, queue_freed_object, state);
         tstate = (PyThreadState *)tstate->next;
     }
     HEAD_UNLOCK(&_PyRuntime);
@@ -1233,7 +1252,7 @@ gc_collect_internal(PyInterpreterState *interp, struct collection_state *state,
     }
     HEAD_UNLOCK(&_PyRuntime);
 
-    process_delayed_frees(interp);
+    process_delayed_frees(interp, state);
 
     // Find unreachable objects
     int err = deduce_unreachable_heap(interp, state);