bpo-33608: Factor out a private, per-interpreter _Py_AddPendingCall(). (gh-12360)

This is effectively an un-revert of #11617 and #12024 (reverted in #12159). Portions of those were merged in other PRs (with lower risk) and this represents the remainder. Note that I found 3 different bugs in the original PRs and have fixed them here.

4 files changed, 118 insertions, 97 deletions

diff --git a/Python/ceval.c b/Python/ceval.c
index 28e9232..b2fa20d 100644
--- a/Python/ceval.c
+++ b/Python/ceval.c
@@ -96,61 +96,61 @@ static long dxp[256];
 /* This can set eval_breaker to 0 even though gil_drop_request became
    1.  We believe this is all right because the eval loop will release
    the GIL eventually anyway. */
-#define COMPUTE_EVAL_BREAKER() \
+#define COMPUTE_EVAL_BREAKER(interp) \
     _Py_atomic_store_relaxed( \
-        &_PyRuntime.ceval.eval_breaker, \
+        &interp->ceval.eval_breaker, \
         GIL_REQUEST | \
         _Py_atomic_load_relaxed(&_PyRuntime.ceval.signals_pending) | \
-        _Py_atomic_load_relaxed(&_PyRuntime.ceval.pending.calls_to_do) | \
-        _PyRuntime.ceval.pending.async_exc)
+        _Py_atomic_load_relaxed(&interp->ceval.pending.calls_to_do) | \
+        interp->ceval.pending.async_exc)
 
-#define SET_GIL_DROP_REQUEST() \
+#define SET_GIL_DROP_REQUEST(interp) \
     do { \
         _Py_atomic_store_relaxed(&_PyRuntime.ceval.gil_drop_request, 1); \
-        _Py_atomic_store_relaxed(&_PyRuntime.ceval.eval_breaker, 1); \
+        _Py_atomic_store_relaxed(&interp->ceval.eval_breaker, 1); \
     } while (0)
 
-#define RESET_GIL_DROP_REQUEST() \
+#define RESET_GIL_DROP_REQUEST(interp) \
     do { \
         _Py_atomic_store_relaxed(&_PyRuntime.ceval.gil_drop_request, 0); \
-        COMPUTE_EVAL_BREAKER(); \
+        COMPUTE_EVAL_BREAKER(interp); \
     } while (0)
 
 /* Pending calls are only modified under pending_lock */
-#define SIGNAL_PENDING_CALLS() \
+#define SIGNAL_PENDING_CALLS(interp) \
     do { \
-        _Py_atomic_store_relaxed(&_PyRuntime.ceval.pending.calls_to_do, 1); \
-        _Py_atomic_store_relaxed(&_PyRuntime.ceval.eval_breaker, 1); \
+        _Py_atomic_store_relaxed(&interp->ceval.pending.calls_to_do, 1); \
+        _Py_atomic_store_relaxed(&interp->ceval.eval_breaker, 1); \
     } while (0)
 
-#define UNSIGNAL_PENDING_CALLS() \
+#define UNSIGNAL_PENDING_CALLS(interp) \
     do { \
-        _Py_atomic_store_relaxed(&_PyRuntime.ceval.pending.calls_to_do, 0); \
-        COMPUTE_EVAL_BREAKER(); \
+        _Py_atomic_store_relaxed(&interp->ceval.pending.calls_to_do, 0); \
+        COMPUTE_EVAL_BREAKER(interp); \
     } while (0)
 
 #define SIGNAL_PENDING_SIGNALS() \
     do { \
         _Py_atomic_store_relaxed(&_PyRuntime.ceval.signals_pending, 1); \
-        _Py_atomic_store_relaxed(&_PyRuntime.ceval.eval_breaker, 1); \
+        _Py_atomic_store_relaxed(&_PyRuntime.interpreters.main->ceval.eval_breaker, 1); \
     } while (0)
 
 #define UNSIGNAL_PENDING_SIGNALS() \
     do { \
         _Py_atomic_store_relaxed(&_PyRuntime.ceval.signals_pending, 0); \
-        COMPUTE_EVAL_BREAKER(); \
+        COMPUTE_EVAL_BREAKER(_PyRuntime.interpreters.main); \
     } while (0)
 
-#define SIGNAL_ASYNC_EXC() \
+#define SIGNAL_ASYNC_EXC(interp) \
     do { \
-        _PyRuntime.ceval.pending.async_exc = 1; \
-        _Py_atomic_store_relaxed(&_PyRuntime.ceval.eval_breaker, 1); \
+        interp->ceval.pending.async_exc = 1; \
+        _Py_atomic_store_relaxed(&interp->ceval.eval_breaker, 1); \
     } while (0)
 
-#define UNSIGNAL_ASYNC_EXC() \
+#define UNSIGNAL_ASYNC_EXC(interp) \
     do { \
-        _PyRuntime.ceval.pending.async_exc = 0; \
-        COMPUTE_EVAL_BREAKER(); \
+        interp->ceval.pending.async_exc = 0; \
+        COMPUTE_EVAL_BREAKER(interp); \
     } while (0)
 
 
@@ -177,10 +177,7 @@ PyEval_InitThreads(void)
     create_gil();
     take_gil(_PyThreadState_GET());
 
-    _PyRuntime.ceval.pending.lock = PyThread_allocate_lock();
-    if (_PyRuntime.ceval.pending.lock == NULL) {
-        Py_FatalError("Can't initialize threads for pending calls");
-    }
+    // The pending calls mutex is initialized in PyInterpreterState_New().
 }
 
 void
@@ -192,11 +189,6 @@ _PyEval_FiniThreads(void)
 
     destroy_gil();
     assert(!gil_created());
-
-    if (_PyRuntime.ceval.pending.lock != NULL) {
-        PyThread_free_lock(_PyRuntime.ceval.pending.lock);
-        _PyRuntime.ceval.pending.lock = NULL;
-    }
 }
 
 void
@@ -256,8 +248,10 @@ PyEval_ReInitThreads(void)
     recreate_gil();
     take_gil(current_tstate);
 
-    _PyRuntime.ceval.pending.lock = PyThread_allocate_lock();
-    if (_PyRuntime.ceval.pending.lock == NULL) {
+    // Only the main interpreter remains, so ignore the rest.
+    PyInterpreterState *interp = _PyRuntime.interpreters.main;
+    interp->ceval.pending.lock = PyThread_allocate_lock();
+    if (interp->ceval.pending.lock == NULL) {
         Py_FatalError("Can't initialize threads for pending calls");
     }
 
@@ -269,9 +263,9 @@ PyEval_ReInitThreads(void)
    raised. */
 
 void
-_PyEval_SignalAsyncExc(void)
+_PyEval_SignalAsyncExc(PyInterpreterState *interp)
 {
-    SIGNAL_ASYNC_EXC();
+    SIGNAL_ASYNC_EXC(interp);
 }
 
 PyThreadState *
@@ -339,7 +333,7 @@ _PyEval_SignalReceived(void)
 
 /* Push one item onto the queue while holding the lock. */
 static int
-_push_pending_call(struct _pending_calls *pending,
+_push_pending_call(struct _pending_calls *pending, unsigned long thread_id,
                    int (*func)(void *), void *arg)
 {
     int i = pending->last;
@@ -347,6 +341,7 @@ _push_pending_call(struct _pending_calls *pending,
     if (j == pending->first) {
         return -1; /* Queue full */
     }
+    pending->calls[i].thread_id = thread_id;
     pending->calls[i].func = func;
     pending->calls[i].arg = arg;
     pending->last = j;
@@ -355,7 +350,7 @@ _push_pending_call(struct _pending_calls *pending,
 
 /* Pop one item off the queue while holding the lock. */
 static void
-_pop_pending_call(struct _pending_calls *pending,
+_pop_pending_call(struct _pending_calls *pending, unsigned long *thread_id,
                   int (**func)(void *), void **arg)
 {
     int i = pending->first;
@@ -365,6 +360,7 @@ _pop_pending_call(struct _pending_calls *pending,
 
     *func = pending->calls[i].func;
     *arg = pending->calls[i].arg;
+    *thread_id = pending->calls[i].thread_id;
     pending->first = (i + 1) % NPENDINGCALLS;
 }
 
@@ -374,9 +370,10 @@ _pop_pending_call(struct _pending_calls *pending,
  */
 
 int
-Py_AddPendingCall(int (*func)(void *), void *arg)
+_Py_AddPendingCall(PyInterpreterState *interp, unsigned long thread_id,
+                   int (*func)(void *), void *arg)
 {
-    struct _pending_calls *pending = &_PyRuntime.ceval.pending;
+    struct _pending_calls *pending = &interp->ceval.pending;
 
     PyThread_acquire_lock(pending->lock, WAIT_LOCK);
     if (pending->finishing) {
@@ -391,14 +388,23 @@ Py_AddPendingCall(int (*func)(void *), void *arg)
         PyErr_Restore(exc, val, tb);
         return -1;
     }
-    int result = _push_pending_call(pending, func, arg);
+    int result = _push_pending_call(pending, thread_id, func, arg);
+    /* signal main loop */
+    SIGNAL_PENDING_CALLS(interp);
     PyThread_release_lock(pending->lock);
 
-    /* signal main loop */
-    SIGNAL_PENDING_CALLS();
     return result;
 }
 
+/* Py_AddPendingCall() is a simple wrapper for the sake
+   of backward-compatibility. */
+int
+Py_AddPendingCall(int (*func)(void *), void *arg)
+{
+    PyInterpreterState *interp = _PyRuntime.interpreters.main;
+    return _Py_AddPendingCall(interp, _PyRuntime.main_thread, func, arg);
+}
+
 static int
 handle_signals(void)
 {
@@ -425,15 +431,11 @@ handle_signals(void)
 }
 
 static int
-make_pending_calls(struct _pending_calls* pending)
+make_pending_calls(PyInterpreterState *interp)
 {
+    struct _pending_calls *pending = &interp->ceval.pending;
     static int busy = 0;
 
-    /* only service pending calls on main thread */
-    if (PyThread_get_thread_ident() != _PyRuntime.main_thread) {
-        return 0;
-    }
-
     /* don't perform recursive pending calls */
     if (busy) {
         return 0;
@@ -441,19 +443,27 @@ make_pending_calls(struct _pending_calls* pending)
     busy = 1;
     /* unsignal before starting to call callbacks, so that any callback
        added in-between re-signals */
-    UNSIGNAL_PENDING_CALLS();
+    UNSIGNAL_PENDING_CALLS(interp);
     int res = 0;
 
     /* perform a bounded number of calls, in case of recursion */
+    unsigned long thread_id = 0;
     for (int i=0; i<NPENDINGCALLS; i++) {
         int (*func)(void *) = NULL;
         void *arg = NULL;
 
         /* pop one item off the queue while holding the lock */
         PyThread_acquire_lock(pending->lock, WAIT_LOCK);
-        _pop_pending_call(pending, &func, &arg);
+        _pop_pending_call(pending, &thread_id, &func, &arg);
         PyThread_release_lock(pending->lock);
 
+        if (thread_id && PyThread_get_thread_ident() != thread_id) {
+            // Thread mismatch, so move it to the end of the list
+            // and start over.
+            _Py_AddPendingCall(interp, thread_id, func, arg);
+            goto error;
+        }
+
         /* having released the lock, perform the callback */
         if (func == NULL) {
             break;
@@ -469,14 +479,14 @@ make_pending_calls(struct _pending_calls* pending)
 
 error:
     busy = 0;
-    SIGNAL_PENDING_CALLS();
+    SIGNAL_PENDING_CALLS(interp); /* We're not done yet */
     return res;
 }
 
 void
-_Py_FinishPendingCalls(void)
+_Py_FinishPendingCalls(PyInterpreterState *interp)
 {
-    struct _pending_calls *pending = &_PyRuntime.ceval.pending;
+    struct _pending_calls *pending = &interp->ceval.pending;
 
     assert(PyGILState_Check());
 
@@ -488,7 +498,7 @@ _Py_FinishPendingCalls(void)
         return;
     }
 
-    if (make_pending_calls(pending) < 0) {
+    if (make_pending_calls(interp) < 0) {
         PyObject *exc, *val, *tb;
         PyErr_Fetch(&exc, &val, &tb);
         PyErr_BadInternalCall();
@@ -497,6 +507,14 @@ _Py_FinishPendingCalls(void)
     }
 }
 
+int
+_Py_MakePendingCalls(PyInterpreterState *interp)
+{
+    assert(PyGILState_Check());
+
+    return make_pending_calls(interp);
+}
+
 /* Py_MakePendingCalls() is a simple wrapper for the sake
    of backward-compatibility. */
 int
@@ -511,12 +529,8 @@ Py_MakePendingCalls(void)
         return res;
     }
 
-    res = make_pending_calls(&_PyRuntime.ceval.pending);
-    if (res != 0) {
-        return res;
-    }
-
-    return 0;
+    PyInterpreterState *interp = _PyRuntime.interpreters.main;
+    return make_pending_calls(interp);
 }
 
 /* The interpreter's recursion limit */
@@ -638,7 +652,7 @@ _PyEval_EvalFrameDefault(PyFrameObject *f, int throwflag)
     PyObject **fastlocals, **freevars;
     PyObject *retval = NULL;            /* Return value */
     PyThreadState *tstate = _PyThreadState_GET();
-    _Py_atomic_int *eval_breaker = &_PyRuntime.ceval.eval_breaker;
+    _Py_atomic_int *eval_breaker = &tstate->interp->ceval.eval_breaker;
     PyCodeObject *co;
 
     /* when tracing we set things up so that
@@ -1059,9 +1073,9 @@ main_loop:
                 }
             }
             if (_Py_atomic_load_relaxed(
-                        &_PyRuntime.ceval.pending.calls_to_do))
+                        &tstate->interp->ceval.pending.calls_to_do))
             {
-                if (make_pending_calls(&_PyRuntime.ceval.pending) != 0) {
+                if (make_pending_calls(tstate->interp) != 0) {
                     goto error;
                 }
             }
@@ -1093,7 +1107,7 @@ main_loop:
             if (tstate->async_exc != NULL) {
                 PyObject *exc = tstate->async_exc;
                 tstate->async_exc = NULL;
-                UNSIGNAL_ASYNC_EXC();
+                UNSIGNAL_ASYNC_EXC(tstate->interp);
                 PyErr_SetNone(exc);
                 Py_DECREF(exc);
                 goto error;
diff --git a/Python/ceval_gil.h b/Python/ceval_gil.h
index f2d5fdb..d9ad361 100644
--- a/Python/ceval_gil.h
+++ b/Python/ceval_gil.h
@@ -176,7 +176,7 @@ static void drop_gil(PyThreadState *tstate)
                     &_PyRuntime.ceval.gil.last_holder)
             ) == tstate)
         {
-        RESET_GIL_DROP_REQUEST();
+        RESET_GIL_DROP_REQUEST(tstate->interp);
             /* NOTE: if COND_WAIT does not atomically start waiting when
                releasing the mutex, another thread can run through, take
                the GIL and drop it again, and reset the condition
@@ -213,7 +213,7 @@ static void take_gil(PyThreadState *tstate)
         if (timed_out &&
             _Py_atomic_load_relaxed(&_PyRuntime.ceval.gil.locked) &&
             _PyRuntime.ceval.gil.switch_number == saved_switchnum) {
-            SET_GIL_DROP_REQUEST();
+            SET_GIL_DROP_REQUEST(tstate->interp);
         }
     }
 _ready:
@@ -239,10 +239,10 @@ _ready:
     MUTEX_UNLOCK(_PyRuntime.ceval.gil.switch_mutex);
 #endif
     if (_Py_atomic_load_relaxed(&_PyRuntime.ceval.gil_drop_request)) {
-        RESET_GIL_DROP_REQUEST();
+        RESET_GIL_DROP_REQUEST(tstate->interp);
     }
     if (tstate->async_exc != NULL) {
-        _PyEval_SignalAsyncExc();
+        _PyEval_SignalAsyncExc(tstate->interp);
     }
 
     MUTEX_UNLOCK(_PyRuntime.ceval.gil.mutex);
diff --git a/Python/pylifecycle.c b/Python/pylifecycle.c
index ad14472..44acba2 100644
--- a/Python/pylifecycle.c
+++ b/Python/pylifecycle.c
@@ -1146,7 +1146,7 @@ Py_FinalizeEx(void)
     interp = tstate->interp;
 
     // Make any remaining pending calls.
-    _Py_FinishPendingCalls();
+    _Py_FinishPendingCalls(interp);
 
     /* The interpreter is still entirely intact at this point, and the
      * exit funcs may be relying on that.  In particular, if some thread
@@ -1552,6 +1552,9 @@ Py_EndInterpreter(PyThreadState *tstate)
     // Wrap up existing "threading"-module-created, non-daemon threads.
     wait_for_thread_shutdown();
 
+    // Make any remaining pending calls.
+    _Py_FinishPendingCalls(interp);
+
     call_py_exitfuncs(interp);
 
     if (tstate != interp->tstate_head || tstate->next != NULL)
diff --git a/Python/pystate.c b/Python/pystate.c
index a2464b6..fee3501 100644
--- a/Python/pystate.c
+++ b/Python/pystate.c
@@ -173,6 +173,14 @@ PyInterpreterState_New(void)
     memset(interp, 0, sizeof(*interp));
     interp->id_refcount = -1;
     interp->check_interval = 100;
+
+    interp->ceval.pending.lock = PyThread_allocate_lock();
+    if (interp->ceval.pending.lock == NULL) {
+        PyErr_SetString(PyExc_RuntimeError,
+                        "failed to create interpreter ceval pending mutex");
+        return NULL;
+    }
+
     interp->core_config = _PyCoreConfig_INIT;
     interp->eval_frame = _PyEval_EvalFrameDefault;
 #ifdef HAVE_DLOPEN
@@ -279,6 +287,9 @@ PyInterpreterState_Delete(PyInterpreterState *interp)
     if (interp->id_mutex != NULL) {
         PyThread_free_lock(interp->id_mutex);
     }
+    if (interp->ceval.pending.lock != NULL) {
+        PyThread_free_lock(interp->ceval.pending.lock);
+    }
     PyMem_RawFree(interp);
 }
 
@@ -928,7 +939,7 @@ PyThreadState_SetAsyncExc(unsigned long id, PyObject *exc)
             p->async_exc = exc;
             HEAD_UNLOCK();
             Py_XDECREF(old_exc);
-            _PyEval_SignalAsyncExc();
+            _PyEval_SignalAsyncExc(interp);
             return 1;
         }
     }
@@ -1342,7 +1353,7 @@ _PyObject_GetCrossInterpreterData(PyObject *obj, _PyCrossInterpreterData *data)
     return 0;
 }
 
-static void
+static int
 _release_xidata(void *arg)
 {
     _PyCrossInterpreterData *data = (_PyCrossInterpreterData *)arg;
@@ -1350,30 +1361,8 @@ _release_xidata(void *arg)
         data->free(data->data);
     }
     Py_XDECREF(data->obj);
-}
-
-static void
-_call_in_interpreter(PyInterpreterState *interp,
-                     void (*func)(void *), void *arg)
-{
-    /* We would use Py_AddPendingCall() if it weren't specific to the
-     * main interpreter (see bpo-33608).  In the meantime we take a
-     * naive approach.
-     */
-    PyThreadState *save_tstate = NULL;
-    if (interp != _PyInterpreterState_Get()) {
-        // XXX Using the "head" thread isn't strictly correct.
-        PyThreadState *tstate = PyInterpreterState_ThreadHead(interp);
-        // XXX Possible GILState issues?
-        save_tstate = PyThreadState_Swap(tstate);
-    }
-
-    func(arg);
-
-    // Switch back.
-    if (save_tstate != NULL) {
-        PyThreadState_Swap(save_tstate);
-    }
+    PyMem_Free(data);
+    return 0;
 }
 
 void
@@ -1384,7 +1373,7 @@ _PyCrossInterpreterData_Release(_PyCrossInterpreterData *data)
         return;
     }
 
-    // Switch to the original interpreter.
+    // Get the original interpreter.
     PyInterpreterState *interp = _PyInterpreterState_LookUpID(data->interp);
     if (interp == NULL) {
         // The intepreter was already destroyed.
@@ -1393,9 +1382,24 @@ _PyCrossInterpreterData_Release(_PyCrossInterpreterData *data)
         }
         return;
     }
+    // XXX There's an ever-so-slight race here...
+    if (interp->finalizing) {
+        // XXX Someone leaked some memory...
+        return;
+    }
 
     // "Release" the data and/or the object.
-    _call_in_interpreter(interp, _release_xidata, data);
+    _PyCrossInterpreterData *copied = PyMem_Malloc(sizeof(_PyCrossInterpreterData));
+    if (copied == NULL) {
+        PyErr_SetString(PyExc_MemoryError,
+                        "Not enough memory to preserve cross-interpreter data");
+        PyErr_Print();
+        return;
+    }
+    memcpy(copied, data, sizeof(_PyCrossInterpreterData));
+    if (_Py_AddPendingCall(interp, 0, _release_xidata, copied) != 0) {
+        // XXX Queue full or couldn't get lock.  Try again somehow?
+    }
 }
 
 PyObject *