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

pypy/module/cpyext/test/test_thread.py::AppTestThread::()::test_timed_lock

self = <pypy.interpreter.typedef.W_ObjectObjectUserDictWeakrefable object at 0x000001d1c0a7b9f0>

    def test_timed_lock(self):
            module = self.import_extension('foo', [
                ("timed_acquire_lock", "METH_O",
                 """
    #ifndef PyThread_acquire_lock_timed
    #error "seems we are not accessing PyPy's functions"
    #endif
                    PyLockStatus acquire_result;
                    int microseconds = PyLong_AsLong(args);
                    Py_BEGIN_ALLOW_THREADS
                    acquire_result = PyThread_acquire_lock_timed(global_lock, microseconds, 0);
                    Py_END_ALLOW_THREADS
                    if (acquire_result == PY_LOCK_ACQUIRED) {
                        Py_RETURN_TRUE;
                    } else {
                        Py_RETURN_FALSE;
                    }
                 """),
                ("release_lock", "METH_NOARGS",
                  """
                  PyThread_release_lock(global_lock);
                  Py_RETURN_NONE;
                  """),
                ("cleanup_lock", "METH_NOARGS",
                  """
                  // cleanup isn't part of the test, but I'd feel guilty otherwise
                  PyThread_free_lock(global_lock); global_lock=NULL;
                  Py_RETURN_NONE;
                  """
                 )
                ],
                prologue="static PyThread_type_lock global_lock;",
                more_init="global_lock = PyThread_allocate_lock();"
            )
            try:
                import threading
                import time
                failure = []
                assert module.timed_acquire_lock(-1)
                main_thread_should_release_the_lock_barrier = threading.Barrier(2)
    
                def thread_func():
                    try:
                        if module.timed_acquire_lock(0):
                            failure.append("Lock should be held elsewhere")
                            return
                        if module.timed_acquire_lock(1):
                            failure.append("Lock should be held elsewhere")
                            return
                    finally:
                        main_thread_should_release_the_lock_barrier.wait()
                    if not module.timed_acquire_lock(1000000):
                        failure.append("Lock should have become available")
                        return
                    module.release_lock()
                thread = threading.Thread(target=thread_func)
                thread.start()
                main_thread_should_release_the_lock_barrier.wait()
                # At this point, thread_func should be waiting 1s for the lock,
                # so sleep a short amount of time then let it have the lock.
                time.sleep(0.01)
                module.release_lock()
    
                thread.join()
>               assert not failure, failure
E               (application-level) AssertionError: ['Lock should have become available']

[d:\pypy_stuff\buildbot64\slave\own-win-x86-64\build\pypy\module\cpyext\test\test_thread.py:84]:65: AssertionError
---------- Captured stdout call ----------
source_0.c
   Creating library d:\systemtemp\pytest\usession-py3.11-1001\foo-2\foo.lib and object d:\systemtemp\pytest\usession-py3.11-1001\foo-2\foo.exp
Generating code
Finished generating code
---------- Captured stderr call ----------
<function PyModule_Create2 at 0x000001d1bdbe82f0>  DONE
<function PyLong_AsLong at 0x000001d1bc2bb6a0>  DONE
<function PyEval_SaveThread at 0x000001d1bd00fbc8>  DONE
<function PyEval_RestoreThread at 0x000001d1bd00fad8>  DONE
<function PyLong_AsLong at 0x000001d1bc2bb6a0>  DONE
<function PyEval_SaveThread at 0x000001d1bd00fbc8>  DONE
<function PyEval_RestoreThread at 0x000001d1bd00fad8>  DONE
<function PyLong_AsLong at 0x000001d1bc2bb6a0>  DONE
<function PyEval_SaveThread at 0x000001d1bd00fbc8>  DONE
<function PyEval_RestoreThread at 0x000001d1bd00fad8>  DONE
<function PyLong_AsLong at 0x000001d1bc2bb6a0>  DONE
<function PyEval_SaveThread at 0x000001d1bd00fbc8>  DONE
<function PyEval_RestoreThread at 0x000001d1bd00fad8>  DONE
builder: own-win-x86-64 build #2232
test: pypy/module/cpyext/test/test_thread/py/AppTestThread/()/test_timed_lock