Home - Summaries: (main) : (py3.11) : Everything - Nightly builds - Benchmarks - RPython - Builders - About

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]