test_concurrent_futures: test_gh105829_should_not_deadlock_if_wakeup_pipe_full() hangs on ARM64 macOS 3.x

[vstinner]

ARM64 macOS 3.x:

0:19:31 load avg: 2.16 [460/460/2] test.test_concurrent_futures.test_deadlock process crashed (Exit code 1)
Timeout (0:15:00)!
Thread 0x000000016ff7b000 (most recent call first):
  File "/Users/buildbot/buildarea/3.x.pablogsal-macos-m1.macos-with-brew/build/Lib/threading.py", line 348 in wait
  File "/Users/buildbot/buildarea/3.x.pablogsal-macos-m1.macos-with-brew/build/Lib/multiprocessing/queues.py", line 231 in _feed
  File "/Users/buildbot/buildarea/3.x.pablogsal-macos-m1.macos-with-brew/build/Lib/threading.py", line 1003 in run
  File "/Users/buildbot/buildarea/3.x.pablogsal-macos-m1.macos-with-brew/build/Lib/threading.py", line 1066 in _bootstrap_inner
  File "/Users/buildbot/buildarea/3.x.pablogsal-macos-m1.macos-with-brew/build/Lib/threading.py", line 1023 in _bootstrap

Thread 0x000000016ef6f000 (most recent call first):
  File "/Users/buildbot/buildarea/3.x.pablogsal-macos-m1.macos-with-brew/build/Lib/selectors.py", line 398 in select
  File "/Users/buildbot/buildarea/3.x.pablogsal-macos-m1.macos-with-brew/build/Lib/multiprocessing/connection.py", line 1135 in wait
  File "/Users/buildbot/buildarea/3.x.pablogsal-macos-m1.macos-with-brew/build/Lib/concurrent/futures/process.py", line 419 in wait_result_broken_or_wakeup
  File "/Users/buildbot/buildarea/3.x.pablogsal-macos-m1.macos-with-brew/build/Lib/concurrent/futures/process.py", line 346 in run
  File "/Users/buildbot/buildarea/3.x.pablogsal-macos-m1.macos-with-brew/build/Lib/test/test_concurrent_futures/test_deadlock.py", line 276 in mock_run
  File "/Users/buildbot/buildarea/3.x.pablogsal-macos-m1.macos-with-brew/build/Lib/threading.py", line 1066 in _bootstrap_inner
  File "/Users/buildbot/buildarea/3.x.pablogsal-macos-m1.macos-with-brew/build/Lib/threading.py", line 1023 in _bootstrap

Thread 0x00000001e2671300 (most recent call first):
  File "/Users/buildbot/buildarea/3.x.pablogsal-macos-m1.macos-with-brew/build/Lib/threading.py", line 348 in wait
  File "/Users/buildbot/buildarea/3.x.pablogsal-macos-m1.macos-with-brew/build/Lib/queue.py", line 140 in put
  File "/Users/buildbot/buildarea/3.x.pablogsal-macos-m1.macos-with-brew/build/Lib/test/test_concurrent_futures/test_deadlock.py", line 285 in wakeup
  File "/Users/buildbot/buildarea/3.x.pablogsal-macos-m1.macos-with-brew/build/Lib/concurrent/futures/process.py", line 857 in shutdown
  File "/Users/buildbot/buildarea/3.x.pablogsal-macos-m1.macos-with-brew/build/Lib/concurrent/futures/_base.py", line 647 in __exit__
  File "/Users/buildbot/buildarea/3.x.pablogsal-macos-m1.macos-with-brew/build/Lib/test/test_concurrent_futures/test_deadlock.py", line 299 in test_gh105829_should_not_deadlock_if_wakeup_pipe_full
  (...)

The test passed when re-run in verbose mode:

0:19:45 load avg: 2.10 [2/2/1] test.test_concurrent_futures.test_deadlock passed

build: https://buildbot.python.org/all/#/builders/725/builds/5749

Linked PRs

• gh-109917, gh-105829: Fix concurrent.futures _ThreadWakeup.wakeup() #110129
• gh-109917: Fix concurrent.future ProcessPoolExecutor locks #110137
• gh-109917: Fix test instability in test_concurrent_futures #110306
• [3.12] gh-109917: Fix test instability in test_concurrent_futures (GH-110306) #110315
• [3.11] gh-109917: Fix test instability in test_concurrent_futures (GH-110306) #110316

[elfstrom]

I have only looked very briefly but I suspect this is a symptom of a race due to the wakeup mock that is used to force blocking behaviour. I will investigate further.

[elfstrom]

If my guess for what goes wrong is correct then I think this would fix it:

--- a/Lib/test/test_concurrent_futures/test_deadlock.py
+++ b/Lib/test/test_concurrent_futures/test_deadlock.py
@@ -280,11 +280,12 @@ def wakeup(self):
                 super().wakeup()
 
             def clear(self):
+                super().clear()
                 try:
                     while True:
                         self._dummy_queue.get_nowait()
                 except queue.Empty:
-                    super().clear()
+                    pass
 
         with (unittest.mock.patch.object(futures.process._ExecutorManagerThread,
                                          'run', mock_run),

I cant reproduce the failure locally though so I might be completely wrong about what's going on.

[vstinner]

The only purpose of _ThreadWakeup is to notify _ExecutorManagerThread when something bad happen:

    def wait_result_broken_or_wakeup(self):
        ...
        wakeup_reader = self.thread_wakeup._reader
        ...
        ready = mp.connection.wait(readers + worker_sentinels)

        ...
        elif wakeup_reader in ready:
            is_broken = False

The number of wakeup() calls doesn't matter: 1 or 1000 is the same.

I don't get test_gh105829_should_not_deadlock_if_wakeup_pipe_full(). It blocks if wakeup() is called more than once? Why?

_ThreadWakeup.wakeup() has a strange implementation: it works 0 bytes into the writer side of the pipe:

class _ThreadWakeup:
    def __init__(self):
        self._closed = False
        self._reader, self._writer = mp.Pipe(duplex=False)

    def wakeup(self):
        if not self._closed:
            self._writer.send_bytes(b"")

Why not sending a byte and ignoring when the pipe is full?

asyncio has a similar "self-pipe" pattern, but it just ignores when the pipe is full:

class BaseSelectorEventLoop(base_events.BaseEventLoop):
    def _write_to_self(self):
        try:
            self._csock.send(b'\0')
        except OSError:
            if self._debug:
                logger.debug("Fail to write a null byte into the "
                             "self-pipe socket",
                             exc_info=True)

Also, there is a simple way to not fill the pipe: write a single byte and then recall in an attribute that a byte was written.

[vstinner]

See also issue gh-105829: concurrent.futures.ProcessPoolExecutor pool deadlocks when submitting many tasks.

[vstinner]

I proposed PR gh-110129 to fix the issue.

[elfstrom]

The only purpose of _ThreadWakeup is to notify _ExecutorManagerThread when something bad happen:

    def wait_result_broken_or_wakeup(self):
        ...
        wakeup_reader = self.thread_wakeup._reader
        ...
        ready = mp.connection.wait(readers + worker_sentinels)

        ...
        elif wakeup_reader in ready:
            is_broken = False

The number of wakeup() calls doesn't matter: 1 or 1000 is the same.

This is not true. When the main thread pushes a new work item it needs to make sure the _ExecutorManagerThread wakes up and moves the work item into the worker process call queue. The push of the work item is not sufficient for that, the _ExecutorManagerThread is only waiting on results and worker process sentinels and this wakeup pipe that can be used to "manually" trigger wakeup. In the current implementation the main thread pushes one object into the wakeup pipe for every new work item and the wakeup pipe is then fully drained when the _ExecutorManagerThread wakes up. This makes sure new work items are moved into the call queue and made available to the workers.

I don't get test_gh105829_should_not_deadlock_if_wakeup_pipe_full(). It blocks if wakeup() is called more than once? Why?

_ThreadWakeup.wakeup() has a strange implementation: it works 0 bytes into the writer side of the pipe:

class _ThreadWakeup:
    def __init__(self):
        self._closed = False
        self._reader, self._writer = mp.Pipe(duplex=False)

    def wakeup(self):
        if not self._closed:
            self._writer.send_bytes(b"")

Why not sending a byte and ignoring when the pipe is full?

We started with that type of fix in #108513, please read through the commits and discussion for the full background on why we ended up with the current state.

asyncio has a similar "self-pipe" pattern, but it just ignores when the pipe is full:

class BaseSelectorEventLoop(base_events.BaseEventLoop):
    def _write_to_self(self):
        try:
            self._csock.send(b'\0')
        except OSError:
            if self._debug:
                logger.debug("Fail to write a null byte into the "
                             "self-pipe socket",
                             exc_info=True)

Also, there is a simple way to not fill the pipe: write a single byte and then recall in an attribute that a byte was written.

The tradeoff done was to prefer removing the lock vs avoiding the write of the pipe. @tomMoral thinks it is possible to do both and that is likely true but we went with the safer option for the time being to get the deadlock fix out the door. There are some interesting corner cases surrounding what happens when the call queue is full and the management thread cannot not actually move any new work at that time. It is likely ok (assuming it is ok without the change) but at least I didn't feel like risking breaking anything.

One thing is very clear however, there is a lot of contention while new work is submitted from the main thread and the workers can easily get starved during this period because the _ExecutorManagerThread cannot move work into their queue fast enough. Improvements here is probably good but would require better tests and benchmarks.