gh-117293: Fix race condition in run_workers.py #117298
Conversation
The worker thread may still be alive after it enqueues it's last result, which can lead to a delay of 30 seconds after the test finishes. This happens much more frequently in the free-threaded build with the GIL disabled. This changes run_workers.py to track of live workers by enqueueing a `WorkerExited()` instance before the worker exits.
Co-authored-by: Alex Waygood <Alex.Waygood@Gmail.com>
| @@ -511,14 +518,18 @@ def _get_result(self) -> QueueOutput | None: | |||
|
|
|||
| # bpo-46205: check the status of workers every iteration to avoid | |||
| # waiting forever on an empty queue. | |||
| while any(worker.is_alive() for worker in self.workers): | |||
| while self.live_worker_count > 0: | |||
There was a problem hiding this comment.
worker.is_alive() is alive is used in other places, I would prefer to use the same logic to list alive workers in all places. I would prefer that you remove this live_worker_count attribute.
If any(worker.is_alive() for worker in self.workers) is inefficient, we can design something else, like a list of alive workers and trim this list when a worker exits. But there number of workers should be less than 1000, so it should be fine.
There was a problem hiding this comment.
I don't understand what design you'd like. We can't continue to rely on the is_alive() method for the termination condition because that is the source of the race condition: it may happen before or after the processing of the results queue.
There are only two other uses of is_alive(), but it makes sense in those places, but doesn't make sense here.
In __repr__, where use it for displaying the worker status.
cpython/Lib/test/libregrtest/run_workers.py
Lines 118 to 119 in 9a1e55b
In wait_stopped() it's paired with join(). Only is_alive() makes sense here because of the use of join().
cpython/Lib/test/libregrtest/run_workers.py
Lines 428 to 430 in 9a1e55b
There was a problem hiding this comment.
You can ignore my comment, your change is fine.
I'm thinking at get_running() function, but I was wrong: this function doesn't call the is_alive() method but only relies on worker.test_name attribute to decide if a thread is "running" or not. There is already a nested finally: self.test_name = None which makes sure that the attribute is cleared in all cases.
| @@ -511,14 +518,18 @@ def _get_result(self) -> QueueOutput | None: | |||
|
|
|||
| # bpo-46205: check the status of workers every iteration to avoid | |||
| # waiting forever on an empty queue. | |||
| while any(worker.is_alive() for worker in self.workers): | |||
| while self.live_worker_count > 0: | |||
There was a problem hiding this comment.
You can ignore my comment, your change is fine.
I'm thinking at get_running() function, but I was wrong: this function doesn't call the is_alive() method but only relies on worker.test_name attribute to decide if a thread is "running" or not. There is already a nested finally: self.test_name = None which makes sure that the attribute is cleared in all cases.
The worker thread may still be alive after it enqueues it's last result, which can lead to a delay of 30 seconds after the test finishes. This happens much more frequently in the free-threaded build with the GIL disabled. This changes run_workers.py to track of live workers by enqueueing a `WorkerExited()` instance before the worker exits.
The worker thread may still be alive after it enqueues it's last result, which can lead to a delay of 30 seconds after the test finishes while waiting on an empty queue. This happens much more frequently in the free-threaded build with the GIL disabled.
This changes run_workers.py to track of live workers by enqueueing a
WorkerExited()instance before the worker exits and using those instances to decrement the count of live workers.