merge 3.3-slp (Stackless python#120)

Author: Anselm Kruis

Python/pystate.c

@@ -48,6 +48,20 @@ static int autoTLSkey = 0;
 #define HEAD_UNLOCK() /* Nothing */
 #endif
 
+#ifdef STACKLESS
+#ifdef WITH_THREAD
+void
+slp_head_lock(void) {
+    HEAD_LOCK();
+}
+
+void
+slp_head_unlock(void) {
+    HEAD_UNLOCK();
+}
+#endif
+#endif
+
 static PyInterpreterState *interp_head = NULL;
 
 /* Assuming the current thread holds the GIL, this is the

Stackless/changelog.txt

@@ -19,6 +19,10 @@ What's New in Stackless 3.X.X?
   weakref-callback runs Python code, Python used to leak a reference to a
   C-stack object.
 
+- https://bitbucket.org/stackless-dev/stackless/issues/120
+  Stackless now correctly grabs the head_mutex, when it iterates over the list
+  of thread states.
+
 - https://bitbucket.org/stackless-dev/stackless/issues/119
   Fix a rare bug in the stack unwinding mechanism, that caused a SystemError
   exception or an assertion violation, if a __del__()-method or a weakref

Stackless/core/stackless_impl.h

@@ -788,6 +788,17 @@ PyObject * slp_get_channel_callback(void);
      (task)->flags.ignore_nesting || \
      (ts->st.runflags & PY_WATCHDOG_IGNORE_NESTING))
 
+/* Interpreter shutdown and thread state access */
+PyObject * slp_getthreads(PyObject *self);
+#ifdef WITH_THREAD
+void slp_head_lock(void);
+void slp_head_unlock(void);
+#define SLP_HEAD_LOCK() slp_head_lock()
+#define SLP_HEAD_UNLOCK() slp_head_unlock()
+#else
+#define SLP_HEAD_LOCK() assert(1)
+#define SLP_HEAD_UNLOCK() assert(1)
+#endif
 
 #include "stackless_api.h"
 

Stackless/core/stacklesseval.c

@@ -384,44 +384,61 @@ slp_eval_frame(PyFrameObject *f)
 }
 
 static void
-kill_pending_current_main_and_watchdogs(PyThreadState *ts)
+get_current_main_and_watchdogs(PyThreadState *ts, PyObject *list)
 {
     PyTaskletObject *t;
 
     assert(ts != PyThreadState_GET());  /* don't kill ourself */
+    assert(PyList_CheckExact(list));
 
     /* kill watchdogs */
     if (ts->st.watchdogs && PyList_CheckExact(ts->st.watchdogs)) {
         Py_ssize_t i;
         /* we don't kill the "intterupt" slot, number 0 */
         for(i = PyList_GET_SIZE(ts->st.watchdogs) - 1; i > 0; i--) {
-            PyObject * item = PyList_GET_ITEM(ts->st.watchdogs, i);
+            PyObject *item = PyList_GET_ITEM(ts->st.watchdogs, i);
             assert(item && PyTasklet_Check(item));
-            t = (PyTaskletObject *) item;
-            Py_INCREF(t);  /* it is a borrowed ref */
-            PyTasklet_KillEx(t, 1);
+            Py_INCREF(item);  /* it is a borrowed ref */
+            PyList_Append(list, item);
             PyErr_Clear();
-            Py_DECREF(t);
+            Py_DECREF(item);
         }
     }
     /* kill main */
     t = ts->st.main;
     if (t != NULL) {
         Py_INCREF(t);  /* it is a borrowed ref */
-        PyTasklet_KillEx(t, 1);
+        PyList_Append(list, (PyObject *)t);
         PyErr_Clear();
         Py_DECREF(t);
     }
     /* kill current */
     t = ts->st.current;
     if (t != NULL) {
         Py_INCREF(t);  /* it is a borrowed ref */
-        PyTasklet_KillEx(t, 1);
+        PyList_Append(list, (PyObject *)t);
         PyErr_Clear();
         Py_DECREF(t);
     }
 }
 
+static void
+kill_pending(PyObject *list)
+{
+    Py_ssize_t i, len;
+
+    assert(list && PyList_CheckExact(list));
+
+    len = PyList_GET_SIZE(list);
+    for (i=0; i < len; i++) {
+        PyTaskletObject *t = (PyTaskletObject *) PyList_GET_ITEM(list, i);
+        assert(PyTasklet_Check(t));
+        PyTasklet_KillEx(t, 1);
+        PyErr_Clear();
+        assert(len == PyList_GET_SIZE(list));
+    }
+}
+
 static void
 run_other_threads(PyObject **sleep, Py_ssize_t count)
 {
@@ -584,7 +601,7 @@ void slp_kill_tasks_with_stacks(PyThreadState *target_ts)
 other_threads:
     /* Step II of III
      *
-     * A separate simple loop to kill tasklets on foreign threads.
+     * Kill tasklets on foreign threads:.
      * Since foreign tasklets are scheduled in their own good time,
      * there is no guarantee that they are actually dead when we
      * exit this function.  Therefore, we also can't clear their thread
@@ -595,20 +612,57 @@ void slp_kill_tasks_with_stacks(PyThreadState *target_ts)
         PyTaskletObject *t;
         PyObject *sleepfunc = NULL;
         Py_ssize_t count;
+        PyObject *tasklet_list = PyList_New(0);
+        if (tasklet_list == NULL) {
+            PyErr_Clear();
+        }
 
-        /* other threads, first pass: kill (pending) current, main and watchdog tasklets */
-        if (target_ts == NULL) {
+        /* Other threads, first pass: kill (pending) current, main and watchdog tasklets
+         * Iterating over the threads requires the HEAD lock. In order to prevent dead locks,
+         * we try to avoid interpreter recursions (= no Py_DECREF), while we hold the lock.
+         */
+        if (target_ts == NULL && tasklet_list) {
             PyThreadState *ts;
+            PyObject *threadid_list = NULL;
             count = 0;
+
+            /* build a list of tasklets to be killed */
+            SLP_HEAD_LOCK();
             for (ts = cts->interp->tstate_head; ts != NULL; ts = ts->next) {
                 if (ts != cts) {
                     /* Inactivate thread ts. In case the thread is active,
                      * it will be killed. If the thread is sleping, it
                      * continues to sleep.
                      */
                     count++;
-                    kill_pending_current_main_and_watchdogs(ts);
+                    get_current_main_and_watchdogs(ts, tasklet_list);
+                }
+            }
+            SLP_HEAD_UNLOCK();
+
+            /* get the list of thread ids */
+            if (PyExc_TaskletExit)
+                threadid_list = slp_getthreads(Py_None); /* requires the HEAD lock */
+
+            /* kill the tasklets */
+            kill_pending(tasklet_list);
+            /* kill the threads */
+            if (threadid_list != NULL) {
+                Py_ssize_t i, len;
+                assert(PyList_CheckExact(threadid_list));
+                len = PyList_GET_SIZE(threadid_list);
+                for (i=0; i < len; i++) {
+                    long thread_id;
+                    PyObject *item = PyList_GET_ITEM(threadid_list, i);
+                    assert(PyLong_CheckExact(item));
+                    thread_id = PyLong_AsLong(item);
+                    if (thread_id != cts->thread_id) {
+                        /* requires the HEAD lock */
+                        PyThreadState_SetAsyncExc(thread_id, PyExc_TaskletExit);
+                        PyErr_Clear();
+                    }
                 }
+                Py_DECREF(threadid_list);
             }
             /* We must not release the GIL while we might hold the HEAD-lock.
              * Otherwise another thread (usually the thread of the killed tasklet)
@@ -619,10 +673,12 @@ void slp_kill_tasks_with_stacks(PyThreadState *target_ts)
             /* The other threads might have modified the thread state chain, but fortunately we
              * are done with it.
              */
-        } else if (target_ts != cts) {
-            kill_pending_current_main_and_watchdogs(target_ts);
+        } else if (target_ts != cts && tasklet_list) {
+            get_current_main_and_watchdogs(target_ts, tasklet_list);
+            kill_pending(tasklet_list);
             /* Here it is not safe to release the GIL. */
         }
+        Py_XDECREF(tasklet_list);
 
         /* other threads, second pass: kill tasklets with nesting-level > 0 and
          * clear tstate if target_ts != NULL && target_ts != cts. */

Stackless/module/scheduling.c

@@ -687,10 +687,14 @@ check_for_deadlock(void)
     PyInterpreterState *interp = ts->interp;
 
     /* see if anybody else will be able to run */
-
-    for (ts = interp->tstate_head; ts != NULL; ts = ts->next)
-        if (is_thread_runnable(ts))
+    SLP_HEAD_LOCK();
+    for (ts = interp->tstate_head; ts != NULL; ts = ts->next) {
+        if (is_thread_runnable(ts)) {
+            SLP_HEAD_UNLOCK();
             return 0;
+        }
+    }
+    SLP_HEAD_UNLOCK();
     return 1;
 }
 

Stackless/module/stacklessmodule.c

@@ -732,10 +732,12 @@ get_thread_info(PyObject *self, PyObject *args)
     if (!PyArg_ParseTuple(args, "|lk:get_thread_info", &id, &flags))
         return NULL;
     if (id != -1) {
+        SLP_HEAD_LOCK();
         for (ts = interp->tstate_head; id && ts != NULL; ts = ts->next) {
             if (ts->thread_id == id)
                 break;
         }
+        SLP_HEAD_UNLOCK();
         if (ts == NULL)
             RUNTIME_ERROR("Thread id not found", NULL);
     }
@@ -1529,8 +1531,8 @@ These might need to be killed manually in order to free memory,\n\
 since their C stack might prevent garbage collection.\n\
 Note that a tasklet is reported for every C stacks it has.");
 
-static PyObject *
-slpmodule_getthreads(PyObject *self)
+PyObject *
+slp_getthreads(PyObject *self)
 {
     PyObject *lis = PyList_New(0);
     PyThreadState *ts = PyThreadState_GET();
@@ -1539,20 +1541,27 @@ slpmodule_getthreads(PyObject *self)
     if (lis == NULL)
         return NULL;
 
+    SLP_HEAD_LOCK();
     for (ts = interp->tstate_head; ts != NULL; ts = ts->next) {
         PyObject *id = PyLong_FromLong(ts->thread_id);
         if (id == NULL) {
+            SLP_HEAD_UNLOCK();
             Py_DECREF(lis);
             return NULL;
         }
         if (PyList_Append(lis, id)) {
+            SLP_HEAD_UNLOCK();
             Py_DECREF(lis);
             Py_DECREF(id);
             return NULL;
         }
         Py_DECREF(id);
     }
-    PyList_Reverse(lis);
+    SLP_HEAD_UNLOCK();
+    if(PyList_Reverse(lis)) {
+        Py_DECREF(lis);
+        return NULL;
+    }
     return lis;
 }
 
@@ -1625,7 +1634,7 @@ static PyMethodDef stackless_methods[] = {
     slpmodule_getdebug__doc__},
     {"getuncollectables",          (PCF)slpmodule_getuncollectables,    METH_NOARGS,
     slpmodule_getuncollectables__doc__},
-    {"getthreads",                 (PCF)slpmodule_getthreads,    METH_NOARGS,
+    {"getthreads",                 (PCF)slp_getthreads,    METH_NOARGS,
     slpmodule_getthreads__doc__},
     {NULL,                          NULL}       /* sentinel */
 };

Stackless/unittests/test_shutdown.py

@@ -360,6 +360,7 @@ def test_other_thread_Py_Exit(self):
 
             def exit():
                 # print("Calling Py_Exit(42)")
+                time.sleep(0.1)
                 sys.stdout.flush()
                 Py_Exit(42)
 
@@ -545,7 +546,7 @@ def __init__(self, out, checks, tasklets):
         self.checks = checks
         self.tasklets = tasklets
 
-        # In Py_Finalize() the PyImport_Cleanup() runs shortly after 
+        # In Py_Finalize() the PyImport_Cleanup() runs shortly after
         # slp_kill_tasks_with_stacks(NULL).
         # As very first action of PyImport_Cleanup() the Python
         # interpreter sets builtins._ to None.