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

Activity

1. vstinner

   added

   testsTests in the Lib/test dirTests in the Lib/test dir

   on Sep 26, 2023
2. vstinner

   changed the title [-]test_concurrent_futures.test_deadlock failed with 15 min timeout (hang?) on ARM64 macOS 3.x[/-] [+]test_concurrent_futures: test_gh105829_should_not_deadlock_if_wakeup_pipe_full() hangs on ARM64 macOS 3.x[/+] on Sep 26, 2023
3. elfstrom commented on Sep 27, 2023

   

   elfstrom

   on Sep 27, 2023

   Contributor

   More actions

   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.
4. elfstrom commented on Sep 27, 2023

   

   elfstrom

   on Sep 27, 2023

   Contributor

   More actions

   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.
5. vstinner commented on Sep 29, 2023

   

   vstinner

   on Sep 29, 2023

   MemberAuthor

   More actions

   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.
6. vstinner commented on Sep 29, 2023

   

   vstinner

   on Sep 29, 2023

   MemberAuthor

   More actions

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

   added a commit that references this issue on Sep 29, 2023

   pythongh-109917, pythongh-105829: Fix concurrent.futures _ThreadWakeu…

   ...

   531aca8
8. vstinner commented on Sep 29, 2023

   

   vstinner

   on Sep 29, 2023

   MemberAuthor

   More actions

   I proposed PR gh-110129 to fix the issue.
9. elfstrom commented on Sep 30, 2023

   

   elfstrom

   on Sep 30, 2023

   Contributor

   More actions

   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.
10. vstinner

    added 3 commits that reference this issue on Sep 30, 2023

    pythongh-109917, pythongh-105829: Fix concurrent.futures _ThreadWakeu…

    ...

    1aa24ee

    pythongh-109917, pythongh-105829: Fix concurrent.futures _ThreadWakeu…

    ...

    5e8706f

    pythongh-109917: Fix concurrent.future ProcessPoolExecutor locks

    ...

    fb905f5
11. vstinner commented on Sep 30, 2023

    

    vstinner

    on Sep 30, 2023

    MemberAuthor

    More actions

    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.

    Hum. You are describing a bad synchronization between two threads. Usually the solution to synchronization is adding a new lock. I wrote PR gh-110137 to add a lock to _ThreadWakeup to make it thread safe. Currently, the shutdown lock is used to access _ThreadWakeup, but IMO it's a bad idea to attempt to protect too many things with an unique lock.
12. elfstrom commented on Sep 30, 2023

    

    elfstrom

    on Sep 30, 2023

    Last edited by elfstrom

    Contributor

    More actions

    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.

    Hum. You are describing a bad synchronization between two threads. Usually the solution to synchronization is adding a new lock. I wrote PR gh-110137 to add a lock to _ThreadWakeup to make it thread safe. Currently, the shutdown lock is used to access _ThreadWakeup, but IMO it's a bad idea to attempt to protect too many things with an unique lock.

    The purpose of the wakeup object is not about synchronization (atomicity or mutual exclusiveness). The only goal is to communicate that new work may be available. This is currently done by sending messages on a Pipe which is thread safe. The only time the main thread and the _ExecutorManagerThread reads/writes shared state is during shutdown (the _closed bool) and this is protected by the shutdown lock so it is also thread safe. I may have missed something but to me it looks like there are no threading issues with the current wakeup implementation.
13. elfstrom

    added a commit that references this issue on Oct 3, 2023

    pythongh-109917: Fix test instability in test_concurrent_futures

    ...

    3522747
14. cjw296

    added a commit that references this issue on Oct 3, 2023

    gh-109917: Fix test instability in test_concurrent_futures (#110306)

    ...

    a376a72
15. miss-islington

    added 2 commits that reference this issue on Oct 3, 2023

    pythongh-109917: Fix test instability in test_concurrent_futures (pyt…

    ...

    0597c30

    pythongh-109917: Fix test instability in test_concurrent_futures (pyt…

    ...

    43f846e
16. cjw296

    added 2 commits that reference this issue on Oct 3, 2023

    [3.11] gh-109917: Fix test instability in test_concurrent_futures (GH…

    ...

    4e44927

    [3.12] gh-109917: Fix test instability in test_concurrent_futures (GH…

    ...

    414f562
17. vstinner commented on Oct 4, 2023

    

    vstinner

    on Oct 4, 2023

    MemberAuthor

    More actions

    Fixed by a376a72
18. vstinner

    closed this as completedon Oct 4, 2023
19. Glyphack

    added a commit that references this issue on Sep 2, 2024

    pythongh-109917: Fix test instability in test_concurrent_futures (pyt…

    ...

    74b57a7