pypy/module/_multiprocessing/test/test_interp_semaphore.py::test_semlock_release[spaceconfig0]
space = StdObjSpace
@pytest.mark.parametrize('spaceconfig', [
{'usemodules': ['_multiprocessing', 'thread']}])
@pytest.mark.skipif(sys.platform == 'darwin', reason="Hangs on macOSX")
def test_semlock_release(space):
# trigger the setup() code in time.moduledef
space.getbuiltinmodule('time')
sem_name = '/test8'
_handle = create_semaphore(space, sem_name, 1, 1)
try:
w_lock = W_SemLock(space, _handle, 0, 1)
created = []
successful = []
N_THREADS = 16
def run(space):
w_sentinel = allocate_lock(space)
yield_thread()
w_sentinel.descr_lock_acquire(space) # releases GIL
try:
yield_thread()
created.append(w_sentinel)
w_got = w_lock.acquire(space, w_timeout=space.newfloat(5.)) # releases GIL
if space.is_true(w_got):
yield_thread()
w_lock.release(space)
successful.append(w_sentinel)
w_sentinel.descr_lock_release(space)
except:
import traceback
traceback.print_exc()
w_sentinel.descr_lock_release(space)
raise
w_run = space.wrap(interp2app(run))
w_lock.acquire(space)
for _ in range(N_THREADS):
start_new_thread(space, w_run, space.newtuple([])) # releases GIL
deadline = time.time() + 5.
while len(created) < N_THREADS:
assert time.time() < deadline
yield_thread()
w_lock.release(space)
for w_sentinel in created:
# Join thread
w_sentinel.descr_lock_acquire(space) # releases GIL
w_sentinel.descr_lock_release(space)
> assert len(successful) == N_THREADS
E assert 14 == 16
E + where 14 = len([<pypy.module.thread.os_lock.Lock object at 0x000001e0d1c77da8>, <pypy.module.thread.os_lock.Lock object at 0x000001e0...thread.os_lock.Lock object at 0x000001e0d1c5ebf0>, <pypy.module.thread.os_lock.Lock object at 0x000001e0d1c3f788>, ...])
module\_multiprocessing\test\test_interp_semaphore.py:59: AssertionError
builder: own-win-x86-64 build #2225
test: pypy/module/_multiprocessing/test/test_interp_semaphore/py/test_semlock_release[spaceconfig0]