Stackless issue python#90: raise RuntimeError on unbinding main tasklets.

Unbinding (tasklet.bind(None) ) of main tasklets caused an assertion violation.
Now it raises RuntimError.

https://bitbucket.org/stackless-dev/stackless/issues/90
(grafted from 58b930a8a1d931da2f5bc834681af5d1452e1733 and 46335aa61c7c)

Author: Anselm Kruis

Stackless/changelog.txt

@@ -10,6 +10,10 @@ What's New in Stackless 3.X.X?
 
 *Release date: 20XX-XX-XX*
 
+- https://bitbucket.org/stackless-dev/stackless/issues/90
+  Stackless now raises a RuntimeError, if you try to unbind (tlet.bind(None))
+  a main-tasklet.
+
 - https://bitbucket.org/stackless-dev/stackless/issues/78
   Documentation update: if you iterate over a channel, the
   sender must send StopIteration manually. See

Stackless/module/taskletobject.c

@@ -251,6 +251,9 @@ PyTasklet_BindEx(PyTaskletObject *task, PyObject *func, PyObject *args, PyObject
     if (tasklet_has_c_stack(task)) {
         RUNTIME_ERROR("tasklet has C state on its stack", -1);
     }
+    if (ts && task == ts->st.main && args == NULL && kwargs == NULL) {
+        RUNTIME_ERROR("can't unbind the main tasklet", -1);
+    }
 
     tasklet_clear_frames(task);
     assert(task->f.frame == NULL);

Stackless/unittests/test_miscell.py

@@ -8,6 +8,7 @@
 import contextlib
 import weakref
 import types
+import _thread as thread
 
 from support import StacklessTestCase, AsTaskletTestCase
 try:
@@ -876,6 +877,58 @@ def test_bind_args_not_runnable(self):
         self.assertFalse(t.scheduled)
         t.run()
 
+    def test_unbind_main(self):
+        self.skipUnlessSoftswitching()
+
+        done = []
+
+        def other():
+            main = stackless.main
+            self.assertRaisesRegex(RuntimeError, "can't unbind the main tasklet", main.bind, None)
+
+        # the initial nesting level depends on the test runner.
+        # We need a main tasklet with nesting_level == 0. Therefore we
+        # use a thread
+        def other_thread():
+            self.assertEqual(stackless.current.nesting_level, 0)
+            self.assertIs(stackless.current, stackless.main)
+            stackless.tasklet(other)().switch()
+            done.append(True)
+
+        t = threading.Thread(target=other_thread, name="other thread")
+        t.start()
+        t.join()
+        self.assertTrue(done[0])
+
+    def test_rebind_main(self):
+        # rebind the main tasklet of a thread. This is highly discouraged,
+        # because it will deadlock, if the thread is a threading.Thread.
+        self.skipUnlessSoftswitching()
+
+        ready = thread.allocate_lock()
+        ready.acquire()
+
+        self.target_called = False
+        self.main_returned = False
+
+        def target():
+            self.target_called = True
+            ready.release()
+
+        def other_thread_main():
+            self.assertTrue(stackless.current.is_main)
+            try:
+                stackless.tasklet(stackless.main.bind)(target, ()).switch()
+            finally:
+                self.main_returned = True
+                ready.release()
+
+        thread.start_new_thread(other_thread_main, ())
+        ready.acquire()
+
+        self.assertTrue(self.target_called)
+        self.assertFalse(self.main_returned)
+
 
 class TestSwitch(StacklessTestCase):
     """Test the new tasklet.switch() method, which allows