induces a segfault if delay time sleepdelay import faulthandler faulthandler disable faulthandler sigsegv def crashwithdatadata induces a sys exit with exitcode 1 sys exit1 def raiseerrorerr function that raises an exception in process and ignores stderr import io sys stderr io stringio raise err def returninstancecls bad object that triggers a segfault at pickling time def reduceself crash class crashatunpickleobject bad object that triggers a process exit at pickling time def reduceself exit class exitatunpickleobject bad object that triggers an error at pickling time def reduceself from pickle import picklingerror raise picklingerrorerror in pickle class erroratunpickleobject if we did not recover before timeout seconds consider that the executor is in a deadlock state and forcefully clean all its composants this should be safe to call executor shutdown here as all possible deadlocks should have been broken test for deadlock caused by crashes in a pool if we did not recover before timeout seconds consider that the executor is in a deadlock state check problem occurring while pickling a task in the taskhandler thread check problem occurring while unpickling a task on workers gh109832 restore stderr overriden by raiseerrorignorestderr check problem occurring while unpickling a task on workers check problem occurring while unpickling a task on workers check problem occurring during func execution on workers check problem occurring during func execution on workers check problem occurring during func execution on workers check problem occurring while pickling a task result on workers check problem occurring while pickling a task result on workers check problem occurring while pickling a task result on workers gh109832 restore stderr overriden by raiseerrorignorestderr check problem occurring while unpickling a task in the resulthandler thread check problem occurring while unpickling a task in the resulthandler thread test that the pool calling shutdown do not cause deadlock if a worker fails after the shutdown call test that the pool calling shutdown with waitfalse does not cause a deadlock if a task fails at pickle after the shutdown call reported in bpo39104 start the executor and get the executormanagerthread to collect the threads and avoid dangling thread that should be cleaned up asynchronously submit a task that fails at pickle and shutdown the executor without waiting make sure the executor is eventually shutdown and do not leave dangling threads test that there is a clean exception instad of a deadlock when a child process crashes while some data is being written into the queue https github compythoncpythonissues94777 issue 105829 the executormanagerthread wakeup pipe could fill up and block see https github compythoncpythonissues105829 lots of cargo culting while writing this test apologies if something is really stupid delay thread startup so the wakeup pipe can fill up and block mock wakeup object to force the wakeup to block def initself super init self dummyqueue queue queuemaxsize1 def wakeupself self dummyqueue putnone blocktrue super wakeup def clearself super clear try while true self dummyqueue getnowait except queue empty pass with unittest mock patch objectfutures process executormanagerthread run mockrun unittest mock patch concurrent futures process threadwakeup mockwakeup with self executortypemaxworkers2 mpcontextself getcontext as executor self executor executor allow clean up in failondeadlock jobnum 100 jobdata rangejobnum need to use sigalarm for timeout detection because executor submit is not guarded by any timeout both self workids putself queuecount and self executormanagerthreadwakeup wakeup might timeout maybe more in this specific case it was the wakeup call that deadlocked on a blocking pipe oldhandler signal signalsignal sigalrm timeout try signal alarmintself timeout self assertequaljobnum lenlistexecutor mapint jobdata finally signal alarm0 signal signalsignal sigalrm oldhandler createexecutortestsglobals executordeadlocktest executormixinsprocesspoolforkmixin processpoolforkservermixin processpoolspawnmixin def setupmodule setupmodule if name main unittest main induces a segfault induces a segfault with dummy data in input induces a sys exit with exitcode 1 function that raises an exception in process function that raises an exception in process and ignores stderr function that returns a instance of cls bad object that triggers a segfault at pickling time bad object that triggers a segfault at unpickling time bad object that triggers a process exit at pickling time bad object that triggers a process exit at unpickling time bad object that triggers an error at pickling time bad object that triggers an error at unpickling time if we did not recover before timeout seconds consider that the executor is in a deadlock state and forcefully clean all its composants this should be safe to call executor shutdown here as all possible deadlocks should have been broken test for deadlock caused by crashes in a pool if we did not recover before timeout seconds consider that the executor is in a deadlock state check problem occurring while pickling a task in the task_handler thread check problem occurring while unpickling a task on workers gh 109832 restore stderr overriden by _raise_error_ignore_stderr check problem occurring while unpickling a task on workers check problem occurring while unpickling a task on workers check problem occurring during func execution on workers check problem occurring during func execution on workers check problem occurring during func execution on workers check problem occurring while pickling a task result on workers check problem occurring while pickling a task result on workers check problem occurring while pickling a task result on workers gh 109832 restore stderr overriden by _raise_error_ignore_stderr check problem occurring while unpickling a task in the result_handler thread check problem occurring while unpickling a task in the result_handler thread test that the pool calling shutdown do not cause deadlock if a worker fails after the shutdown call allow clean up in fail_on_deadlock test that the pool calling shutdown with wait false does not cause a deadlock if a task fails at pickle after the shutdown call reported in bpo 39104 allow clean up in fail_on_deadlock start the executor and get the executor_manager_thread to collect the threads and avoid dangling thread that should be cleaned up asynchronously submit a task that fails at pickle and shutdown the executor without waiting make sure the executor is eventually shutdown and do not leave dangling threads test that there is a clean exception instad of a deadlock when a child process crashes while some data is being written into the queue https github com python cpython issues 94777 allow clean up in fail_on_deadlock issue 105829 the _executormanagerthread wakeup pipe could fill up and block see https github com python cpython issues 105829 lots of cargo culting while writing this test apologies if something is really stupid delay thread startup so the wakeup pipe can fill up and block mock wakeup object to force the wakeup to block allow clean up in fail_on_deadlock need to use sigalarm for timeout detection because executor submit is not guarded by any timeout both self _work_ids put self _queue_count and self _executor_manager_thread_wakeup wakeup might timeout maybe more in this specific case it was the wakeup call that deadlocked on a blocking pipe
import contextlib
import queue
import signal
import sys
import time
import unittest
import unittest.mock
from pickle import PicklingError
from concurrent import futures
from concurrent.futures.process import BrokenProcessPool, _ThreadWakeup

from test import support

from .util import (
    create_executor_tests, setup_module,
    ProcessPoolForkMixin, ProcessPoolForkserverMixin, ProcessPoolSpawnMixin)


def _crash(delay=None):
    
    if delay:
        time.sleep(delay)
    import faulthandler
    faulthandler.disable()
    faulthandler._sigsegv()


def _crash_with_data(data):
    
    _crash()


def _exit():
    
    sys.exit(1)


def _raise_error(Err):
    
    raise Err()


def _raise_error_ignore_stderr(Err):
    
    import io
    sys.stderr = io.StringIO()
    raise Err()


def _return_instance(cls):
    
    return cls()


class CrashAtPickle(object):
    
    def __reduce__(self):
        _crash()


class CrashAtUnpickle(object):
    
    def __reduce__(self):
        return _crash, ()


class ExitAtPickle(object):
    
    def __reduce__(self):
        _exit()


class ExitAtUnpickle(object):
    
    def __reduce__(self):
        return _exit, ()


class ErrorAtPickle(object):
    
    def __reduce__(self):
        from pickle import PicklingError
        raise PicklingError("Error in pickle")


class ErrorAtUnpickle(object):
    
    def __reduce__(self):
        from pickle import UnpicklingError
        return _raise_error_ignore_stderr, (UnpicklingError, )


class ExecutorDeadlockTest:
    TIMEOUT = support.LONG_TIMEOUT

    def _fail_on_deadlock(self, executor):
        
        
        
        import faulthandler
        from tempfile import TemporaryFile
        with TemporaryFile(mode="w+") as f:
            faulthandler.dump_traceback(file=f)
            f.seek(0)
            tb = f.read()
        for p in executor._processes.values():
            p.terminate()
        
        
        executor.shutdown(wait=True)
        print(f"\nTraceback:\n {tb}", file=sys.__stderr__)
        self.fail(f"Executor deadlock:\n\n{tb}")


    def _check_crash(self, error, func, *args, ignore_stderr=False):
        
        self.executor.shutdown(wait=True)

        executor = self.executor_type(
            max_workers=2, mp_context=self.get_context())
        res = executor.submit(func, *args)

        if ignore_stderr:
            cm = support.captured_stderr()
        else:
            cm = contextlib.nullcontext()

        try:
            with self.assertRaises(error):
                with cm:
                    res.result(timeout=self.TIMEOUT)
        except futures.TimeoutError:
            
            
            self._fail_on_deadlock(executor)
        executor.shutdown(wait=True)

    def test_error_at_task_pickle(self):
        
        
        self._check_crash(PicklingError, id, ErrorAtPickle())

    def test_exit_at_task_unpickle(self):
        
        self._check_crash(BrokenProcessPool, id, ExitAtUnpickle())

    def test_error_at_task_unpickle(self):
        
        self.addCleanup(setattr, sys, 'stderr', sys.stderr)

        
        self._check_crash(BrokenProcessPool, id, ErrorAtUnpickle())

    def test_crash_at_task_unpickle(self):
        
        self._check_crash(BrokenProcessPool, id, CrashAtUnpickle())

    def test_crash_during_func_exec_on_worker(self):
        
        self._check_crash(BrokenProcessPool, _crash)

    def test_exit_during_func_exec_on_worker(self):
        
        self._check_crash(SystemExit, _exit)

    def test_error_during_func_exec_on_worker(self):
        
        self._check_crash(RuntimeError, _raise_error, RuntimeError)

    def test_crash_during_result_pickle_on_worker(self):
        
        
        self._check_crash(BrokenProcessPool, _return_instance, CrashAtPickle)

    def test_exit_during_result_pickle_on_worker(self):
        
        
        self._check_crash(SystemExit, _return_instance, ExitAtPickle)

    def test_error_during_result_pickle_on_worker(self):
        
        
        self._check_crash(PicklingError, _return_instance, ErrorAtPickle)

    def test_error_during_result_unpickle_in_result_handler(self):
        
        self.addCleanup(setattr, sys, 'stderr', sys.stderr)

        
        
        self._check_crash(BrokenProcessPool,
                          _return_instance, ErrorAtUnpickle,
                          ignore_stderr=True)

    def test_exit_during_result_unpickle_in_result_handler(self):
        
        
        self._check_crash(BrokenProcessPool, _return_instance, ExitAtUnpickle)

    def test_shutdown_deadlock(self):
        
        
        self.executor.shutdown(wait=True)
        with self.executor_type(max_workers=2,
                                mp_context=self.get_context()) as executor:
            self.executor = executor  
            f = executor.submit(_crash, delay=.1)
            executor.shutdown(wait=True)
            with self.assertRaises(BrokenProcessPool):
                f.result()

    def test_shutdown_deadlock_pickle(self):
        
        
        
        self.executor.shutdown(wait=True)
        with self.executor_type(max_workers=2,
                                mp_context=self.get_context()) as executor:
            self.executor = executor  

            
            
            
            executor.submit(id, 42).result()
            executor_manager = executor._executor_manager_thread

            
            
            f = executor.submit(id, ErrorAtPickle())
            executor.shutdown(wait=False)
            with self.assertRaises(PicklingError):
                f.result()

        
        
        executor_manager.join()

    def test_crash_big_data(self):
        
        
        
        
        self.executor.shutdown(wait=True)
        data = "a" * support.PIPE_MAX_SIZE
        with self.executor_type(max_workers=2,
                                mp_context=self.get_context()) as executor:
            self.executor = executor  
            with self.assertRaises(BrokenProcessPool):
                list(executor.map(_crash_with_data, [data] * 10))

        executor.shutdown(wait=True)

    def test_gh105829_should_not_deadlock_if_wakeup_pipe_full(self):
        
        

        
        

        self.executor.shutdown(wait=True)

        if not hasattr(signal, 'alarm'):
            raise unittest.SkipTest(
                "Tested platform does not support the alarm signal")

        def timeout(_signum, _frame):
            import faulthandler
            faulthandler.dump_traceback()

            raise RuntimeError("timed out while submitting jobs?")

        thread_run = futures.process._ExecutorManagerThread.run
        def mock_run(self):
            
            time.sleep(3)
            thread_run(self)

        class MockWakeup(_ThreadWakeup):
            
            def __init__(self):
                super().__init__()
                self._dummy_queue = queue.Queue(maxsize=1)

            def wakeup(self):
                self._dummy_queue.put(None, block=True)
                super().wakeup()

            def clear(self):
                super().clear()
                try:
                    while True:
                        self._dummy_queue.get_nowait()
                except queue.Empty:
                    pass

        with (unittest.mock.patch.object(futures.process._ExecutorManagerThread,
                                         'run', mock_run),
              unittest.mock.patch('concurrent.futures.process._ThreadWakeup',
                                  MockWakeup)):
            with self.executor_type(max_workers=2,
                                    mp_context=self.get_context()) as executor:
                self.executor = executor  

                job_num = 100
                job_data = range(job_num)

                
                
                
                
                
                
                old_handler = signal.signal(signal.SIGALRM, timeout)
                try:
                    signal.alarm(int(self.TIMEOUT))
                    self.assertEqual(job_num, len(list(executor.map(int, job_data))))
                finally:
                    signal.alarm(0)
                    signal.signal(signal.SIGALRM, old_handler)


create_executor_tests(globals(), ExecutorDeadlockTest,
                      executor_mixins=(ProcessPoolForkMixin,
                                       ProcessPoolForkserverMixin,
                                       ProcessPoolSpawnMixin))

def setUpModule():
    setup_module()


if __name__ == "__main__":
    unittest.main()