Commit ebb8901f authored by Armin Rigo's avatar Armin Rigo

Tests (passing) for _continulet switching to a different thread

parent bcd408a1ef3d
......@@ -24,7 +24,7 @@ class W_Continulet(W_Root):
ec = self.space.getexecutioncontext()
if ec.stacklet_thread is not self.sthread:
global_state.clear()
raise geterror(self.space, "inter-thread support is missing")
raise geterror(self.space, "cannot switch to a different thread")
def descr_init(self, w_callable, __args__):
if self.sthread is not None:
......
......@@ -117,6 +117,86 @@ class AppTestWrapper:
_vmprof.disable()
f.close()
def test_thread_switch_to_sub(self):
try:
import thread, time
except ImportError:
py.test.skip("no threads")
c_list = []
lock = thread.allocate_lock()
lock.acquire()
lock2 = thread.allocate_lock()
lock2.acquire()
#
def fn():
c = _continuation.continulet(lambda c_main: c_main.switch())
c.switch()
c_list.append(c)
lock.release()
lock2.acquire()
#
thread.start_new_thread(fn, ())
lock.acquire()
[c] = c_list
py.test.raises(_continuation.error, c.switch)
#
lock2.release()
time.sleep(0.5)
py.test.raises(_continuation.error, c.switch)
def test_thread_switch_to_sub_nonstarted(self):
try:
import thread, time
except ImportError:
py.test.skip("no threads")
c_list = []
lock = thread.allocate_lock()
lock.acquire()
lock2 = thread.allocate_lock()
lock2.acquire()
#
def fn():
c = _continuation.continulet(lambda c_main: None)
c_list.append(c)
lock.release()
lock2.acquire()
#
thread.start_new_thread(fn, ())
lock.acquire()
[c] = c_list
py.test.raises(_continuation.error, c.switch)
#
lock2.release()
time.sleep(0.5)
py.test.raises(_continuation.error, c.switch)
def test_thread_switch_to_main(self):
try:
import thread, time
except ImportError:
py.test.skip("no threads")
c_list = []
lock = thread.allocate_lock()
lock.acquire()
lock2 = thread.allocate_lock()
lock2.acquire()
#
def fn():
def in_continulet(c_main):
c_list.append(c_main)
lock.release()
lock2.acquire()
c = _continuation.continulet(in_continulet)
c.switch()
#
thread.start_new_thread(fn, ())
lock.acquire()
[c] = c_list
py.test.raises(_continuation.error, c.switch)
#
lock2.release()
time.sleep(0.5)
py.test.raises(_continuation.error, c.switch)
def _setup():
for _i in range(20):
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment