GH-109978: Allow multiprocessing finalizers to run on a separate thread#110510
GH-109978: Allow multiprocessing finalizers to run on a separate thread#110510pitrou wants to merge 1 commit intopython:mainfrom
Conversation
pitrou
force-pushed
the
gh109978-mp-finalizer-thread
branch
from
a7d3581 to
3664a79
Compare
| if thread_was_running: | ||
| # HACK: os.fork() queries the system for the number of running threads. | ||
| # However, Thread.join() only ensures that the Python thread state | ||
| # was destroyed, while the system thread could still be running. | ||
| # Give it time to exit. | ||
| time.sleep(0.001) |
|
I think that it will skip my turn for this one. I touched too many multiprocessing code recently, and I was bitten. Maybe @serhiy-storchaka wants to have a look, he did something like that recently (I recall vaguely). |
gpshead
left a comment
gpshead
left a comment
There was a problem hiding this comment.
I haven't tried to understand the whole thing yet, but i've given it a once over to look at the issues this is running into.
| if thread_was_running: | ||
| # HACK: os.fork() queries the system for the number of running threads. | ||
| # However, Thread.join() only ensures that the Python thread state | ||
| # was destroyed, while the system thread could still be running. | ||
| # Give it time to exit. | ||
| time.sleep(0.001) |
There was a problem hiding this comment.
Agreed, this is effectively because our join() has never truely been a join... I filed #110829 to track that, it is probably time to fix that.
| } | ||
| // XXX This counts the number of system threads, but Python code can only | ||
| // call threading.Thread.join(), which can return before the system thread ended. | ||
| // This function could therefore print a spurious warning in unlucky cases. |
There was a problem hiding this comment.
It's not spurious though, the existence of a thread doing anything is accurately identified as a potential problem. The annoyance that led to this though is that Python didn't provide a concrete way to guarantee a thread has exited.
| while self._pool: | ||
| self._pool.pop().join() |
There was a problem hiding this comment.
What is the difference?
| with self._lock(): | ||
| return self._stopped |
There was a problem hiding this comment.
Why is the lock needed for reading an attribute?
|
|
||
| def wait_until_idle(self): | ||
| with self._queue_drained_cond: | ||
| self._queue_drained_cond.wait_for(lambda: self._queue.empty()) |
There was a problem hiding this comment.
| self._queue_drained_cond.wait_for(lambda: self._queue.empty()) | |
| self._queue_drained_cond.wait_for(self._queue.empty) |
| # from the work loop. | ||
| assert callable(cb) | ||
| with self._lock: | ||
| if not self._stopped and self._thread is not None: |
There was a problem hiding this comment.
Isn't self._thread always None when self._stopped is True?
| finally: | ||
| cb = None |
There was a problem hiding this comment.
Maybe add a reference to _work_loop() or copy the comment? I was confused when I saw this.
Hmm, _work_queue.enqueue_task() is always called from the global enqueue_task()which keeps a reference tocb`, so it perhaps does not help.
| sub_debug('finalizer ignored because different process') | ||
| res = None | ||
| else: | ||
| try: |
There was a problem hiding this comment.
There is already one try a level above. The second try may be not needed, you can simply add finally at the same level after else. The difference is that the finally block will be exececuted in the except case, but I do not see what can be wrong with this.
|
FWIW, I'm prioritizing #110848 now, as it should make this PR slightly more robust. |
4 participants
Uh oh!
There was an error while loading. Please reload this page.