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

pypy/module/thread/test/test_fork.py::AppTestFork::()::test_nested_import_lock_fork

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

    def test_nested_import_lock_fork(self):
            """Check fork() in main thread works while the main thread is doing an import"""
            # Issue 9573: this used to trigger RuntimeError in the child process
            import imp
            import os
            import time
    
            if not hasattr(os, 'fork'):
                skip("No fork on this platform")
    
            def fork_with_import_lock(level):
                release = 0
                in_child = False
                try:
                    try:
                        for i in range(level):
                            imp.acquire_lock()
                            release += 1
                        pid = os.fork()
                        in_child = not pid
                    finally:
                        for i in range(release):
                            imp.release_lock()
                except RuntimeError:
                    if in_child:
                        if verbose > 1:
                            print("RuntimeError in child")
                        os._exit(1)
                    raise
                if in_child:
                    os._exit(0)
    
                for i in range(10):
                    spid, status = os.waitpid(pid, os.WNOHANG)
                    if spid == pid:
                        break
                    time.sleep(1.0)
                assert spid == pid
                assert status == 0
    
            # Check this works with various levels of nested
            # import in the main thread
            for level in range(5):
>               fork_with_import_lock(level)

[/Users/matti/build-worker-arm64/own-macos-arm64/build/pypy/module/thread/test/test_fork.py:93]:44: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

level = W_IntObject(2)

    def fork_with_import_lock(level):
        release = 0
        in_child = False
        try:
            try:
                for i in range(level):
                    imp.acquire_lock()
                    release += 1
                pid = os.fork()
                in_child = not pid
            finally:
                for i in range(release):
                    imp.release_lock()
        except RuntimeError:
            if in_child:
                if verbose > 1:
                    print("RuntimeError in child")
                os._exit(1)
            raise
        if in_child:
            os._exit(0)
    
        for i in range(10):
            spid, status = os.waitpid(pid, os.WNOHANG)
            if spid == pid:
                break
            time.sleep(1.0)
        assert spid == pid
>       assert status == 0
E       (application-level) AssertionError

[/Users/matti/build-worker-arm64/own-macos-arm64/build/pypy/module/thread/test/test_fork.py:93]:39: AssertionError
builder: own-macos-arm64 build #1313+
test: pypy/module/thread/test/test_fork.py::AppTestFork::()::test_nested_import_lock_fork