Most importantly, this resolves the issues with functions and types defined in __main__.
It also expands the number of supported objects and simplifies the implementation.
PEP-734 has been accepted (for 3.14).
(FTR, I'm opposed to putting this under the concurrent package, but
doing so is the SC condition under which the module can land in 3.14.)
When shutdown is called with wait=False, the executor thread keeps running
even after the ProcessPoolExecutor's state is reset. The executor then tries
to replenish the worker processes pool resulting in an error and a potential hang
when it comes across a worker that has died. Fixed the issue by having
_adjust_process_count() return without doing anything if the ProcessPoolExecutor's
state has been reset.
Added unit tests to validate two scenarios:
max_workers < num_tasks (exception)
max_workers > num_tasks (exception + hang)
`concurrent.futures.Executor.map` now supports limiting the number of submitted
tasks whose results have not yet been yielded via the new `buffersize` parameter.
---------
Co-authored-by: Bénédikt Tran <10796600+picnixz@users.noreply.github.com>
This adds two new methods to `multiprocessing`'s `ProcessPoolExecutor`:
- **`terminate_workers()`**: forcefully terminates worker processes using `Process.terminate()`
- **`kill_workers()`**: forcefully kills worker processes using `Process.kill()`
These methods provide users with a direct way to stop worker processes without `shutdown()` or relying on implementation details, addressing situations where immediate termination is needed.
Co-authored-by: Bénédikt Tran <10796600+picnixz@users.noreply.github.com>
Co-authored-by: blurb-it[bot] <43283697+blurb-it[bot]@users.noreply.github.com>
Co-authored-by: Sam Gross @colesbury
Commit-message-mostly-authored-by: Claude Sonnet 3.7 (because why not -greg)
This adds two new methods to `multiprocessing`'s `ProcessPoolExecutor`:
- **`terminate_workers()`**: forcefully terminates worker processes using `Process.terminate()`
- **`kill_workers()`**: forcefully kills worker processes using `Process.kill()`
These methods provide users with a direct way to stop worker processes without `shutdown()` or relying on implementation details, addressing situations where immediate termination is needed.
Co-authored-by: Bénédikt Tran <10796600+picnixz@users.noreply.github.com>
Co-authored-by: blurb-it[bot] <43283697+blurb-it[bot]@users.noreply.github.com>
Commit-message-mostly-authored-by: Claude Sonnet 3.7 (because why not -greg)
Threads are gone after fork, so clear the queues too. Otherwise the
child process (here created via multiprocessing.Process) crashes on
interpreter exit.
Co-authored-by: Serhiy Storchaka <storchaka@gmail.com>
This is an implementation of InterpreterPoolExecutor that builds on ThreadPoolExecutor.
(Note that this is not tied to PEP 734, which is strictly about adding a new stdlib module.)
Possible future improvements:
* support passing a script for the initializer or to submit()
* support passing (most) arbitrary functions without pickling
* support passing closures
* optionally exec functions against __main__ instead of the their original module
There was a deadlock when `ProcessPoolExecutor` shuts down at the same
time that a queueing thread handles an error processing a task.
Don't use `_shutdown_lock` to protect the `_ThreadWakeup` pipes -- use
an internal lock instead. This fixes the ordering deadlock where the
`ExecutorManagerThread` holds the `_shutdown_lock` and joins the
queueing thread, while the queueing thread is attempting to acquire the
`_shutdown_lock` while closing the `_ThreadWakeup`.
Tools such as ruff can ignore "imported but unused" warnings if a
line ends with "# noqa: F401". It avoids the temptation to remove
an import which is used effectively.
Deferred reference counting is not fully implemented yet. As a temporary
measure, we immortalize objects that would use deferred reference
counting to avoid multi-threaded scaling bottlenecks.
This is only performed in the free-threaded build once the first
non-main thread is started. Additionally, some tests, including refleak
tests, suppress this behavior.
concurrent.futures: The *executor manager thread* now catches
exceptions when adding an item to the *call queue*. During Python
finalization, creating a new thread can now raise RuntimeError. Catch
the exception and call terminate_broken() in this case.
Add test_python_finalization_error() to test_concurrent_futures.
concurrent.futures._ExecutorManagerThread changes:
* terminate_broken() no longer calls shutdown_workers() since the
call queue is no longer working anymore (read and write ends of
the queue pipe are closed).
* terminate_broken() now terminates child processes, not only
wait until they complete.
* _ExecutorManagerThread.terminate_broken() now holds shutdown_lock
to prevent race conditons with ProcessPoolExecutor.submit().
multiprocessing.Queue changes:
* Add _terminate_broken() method.
* _start_thread() sets _thread to None on exception to prevent
leaking "dangling threads" even if the thread was not started
yet.
Follow-up of gh-107219.
* Only close the connection writer on Windows.
* Also use existing constant _winapi.ERROR_OPERATION_ABORTED instead of
WSA_OPERATION_ABORTED.
This fixes issue #105829, https://github.com/python/cpython/issues/105829
Co-authored-by: blurb-it[bot] <43283697+blurb-it[bot]@users.noreply.github.com>
Co-authored-by: Antoine Pitrou <antoine@python.org>
Co-authored-by: Chris Withers <chris@withers.org>
Co-authored-by: Thomas Moreau <thomas.moreau.2010@gmail.com>
Fix a race condition in concurrent.futures. When a process in the
process pool was terminated abruptly (while the future was running or
pending), close the connection write end. If the call queue is
blocked on sending bytes to a worker process, closing the connection
write end interrupts the send, so the queue can be closed.
Changes:
* _ExecutorManagerThread.terminate_broken() now closes
call_queue._writer.
* multiprocessing PipeConnection.close() now interrupts
WaitForMultipleObjects() in _send_bytes() by cancelling the
overlapped operation.
Fix a race condition in _ExecutorManagerThread.terminate_broken():
ignore the InvalidStateError on future.set_exception(). It can happen
if the future is cancelled before the caller.
Moreover, test_crash_big_data() now waits explicitly until the
executor completes.
Fixes a hang in multiprocessing process pool executor when a child process crashes and code could otherwise block on writing to the pipe. See GH-94777 for more details.
Reduced _idle_semaphore.release calls in concurrent.futures.thread._worker
_idle_semaphore.release() is now only called if only work_queue is empty.
---------
Co-authored-by: blurb-it[bot] <43283697+blurb-it[bot]@users.noreply.github.com>
Co-authored-by: Shantanu <12621235+hauntsaninja@users.noreply.github.com>
This starts the process. Users who don't specify their own start method
and use the default on platforms where it is 'fork' will see a
DeprecationWarning upon multiprocessing.Pool() construction or upon
multiprocessing.Process.start() or concurrent.futures.ProcessPool use.
See the related issue and documentation within this change for details.
Do not spawn ProcessPool workers on demand when they spawn via fork.
This avoids potential deadlocks in the child processes due to forking from
a multithreaded process.
Prevent `max_tasks_per_child` use with a "fork" mp_context to avoid deadlocks.
Also defaults to "spawn" when no mp_context is supplied for safe convenience.
* bpo-20369: concurrent.futures.wait() now deduplicates futures given as arg.
* 📜🤖 Added by blurb_it.
Co-authored-by: blurb-it[bot] <43283697+blurb-it[bot]@users.noreply.github.com>
In the case of multiprocessing.synchronize() being missing, the
test_concurrent_futures test suite now skips only the tests that
require multiprocessing.synchronize().
Validate that multiprocessing.synchronize exists as part of
_check_system_limits(), allowing ProcessPoolExecutor to raise
NotImplementedError during __init__, rather than crashing with
ImportError during __init__ when creating a lock imported from
multiprocessing.synchronize.
Use _check_system_limits() to disable tests of
ProcessPoolExecutor on systems without multiprocessing.synchronize.
Running the test suite without multiprocessing.synchronize reveals
that Lib/compileall.py crashes when it uses a ProcessPoolExecutor.
Therefore, change Lib/compileall.py to call _check_system_limits()
before creating the ProcessPoolExecutor.
Note that both Lib/compileall.py and Lib/test/test_compileall.py
were attempting to sanity-check ProcessPoolExecutor by expecting
ImportError. In multiprocessing.resource_tracker, sem_unlink() is also absent
on platforms where POSIX semaphores aren't available. Avoid using
sem_unlink() if it, too, does not exist.
Co-authored-by: Pablo Galindo <Pablogsal@gmail.com>
