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>
Remove daemon threads from :mod:`concurrent.futures` by adding
an internal `threading._register_atexit()`, which calls registered functions
prior to joining all non-daemon threads. This allows for compatibility
with subinterpreters, which don't support daemon threads.
As reported initially by @rad-pat in #6084, the following script causes a deadlock.
```
from concurrent.futures import ProcessPoolExecutor
class ObjectWithPickleError():
"""Triggers a RuntimeError when sending job to the workers"""
def __reduce__(self):
raise RuntimeError()
if __name__ == "__main__":
e = ProcessPoolExecutor()
f = e.submit(id, ObjectWithPickleError())
e.shutdown(wait=False)
f.result() # Deadlock on get
```
This is caused by the fact that the main process is closing communication channels that might be necessary to the `queue_management_thread` later. To avoid this, this PR let the `queue_management_thread` manage all the closing.
https://bugs.python.org/issue39104
Automerge-Triggered-By: @pitrou
* Fixes issue 24882
* Add news file entry for change.
* Change test_concurrent_futures.ThreadPoolShutdownTest
Adjust the shutdown test so that, after submitting three jobs
to the executor, the test checks for less than three threads,
instead of looking for exactly three threads.
If idle threads are being recycled properly, then we should have
less than three threads.
* Switched idle count to semaphor, Updated tests
As suggested by reviewer tomMoral, swapped lock-protected counter
with a semaphore to track the number of unused threads.
Adjusted test_threads_terminate to wait for completiton of the
previous future before submitting a new one (and checking the
number of threads used).
Also added a new test to confirm the thread pool can be saturated.
* Updates tests as requested by pitrou.
* Correct minor whitespace error.
* Make test_saturation faster
Wrap the callback call within the `add_done_callback` function within concurrent.futures, in order to behave in an identical manner to callbacks added to a running future are triggered once it has completed.
Deprecated passing the following arguments as keyword arguments:
- "func" in functools.partialmethod(), weakref.finalize(),
profile.Profile.runcall(), cProfile.Profile.runcall(),
bdb.Bdb.runcall(), trace.Trace.runfunc() and
curses.wrapper().
- "function" in unittest.addModuleCleanup() and
unittest.TestCase.addCleanup().
- "fn" in the submit() method of concurrent.futures.ThreadPoolExecutor
and concurrent.futures.ProcessPoolExecutor.
- "callback" in contextlib.ExitStack.callback(),
contextlib.AsyncExitStack.callback() and
contextlib.AsyncExitStack.push_async_callback().
- "c" and "typeid" in the create() method of multiprocessing.managers.Server
and multiprocessing.managers.SharedMemoryServer.
- "obj" in weakref.finalize().
Also allowed to pass arbitrary keyword arguments (even "self" and "func")
if the above arguments are passed as positional argument.
ProcessPoolExecutor workers will hold the return value of their last task in memory until the next task is received. Since the return value has already been propagated to the parent process's Future (or has been discarded by this point), the object can be safely released.
Two kind of mistakes:
1. Missed space. After concatenating there is no space between words.
2. Missed comma. Causes unintentional concatenating in a list of strings.
Use a monotonic clock to compute timeouts in :meth:`Executor.map` and :func:`as_completed`, in order to prevent timeouts from deviating when the system clock is adjusted.
This may not be sufficient on all systems. On POSIX for example, the actual waiting (e.g. in ``sem_timedwait``) is specified to rely on the CLOCK_REALTIME clock.
Future.set_result and Future.set_exception now raise InvalidStateError
if the futures are not pending or running. This mirrors the behavior
of asyncio.Future, and prevents AssertionErrors in asyncio.wrap_future
when set_result is called multiple times.
elfstrom
Heinz-Alexander Fuetterer
Victor Stinner
Louis Paulot
Andrii Kuzmin
yonatanp
Alex Waygood
Gregory P. Smith
Thomas Grainger
themylogin
Yiannis Hadjicharalambous
Serhiy Storchaka
Kumar Aditya
Jakub Kulík
Logan Jones
nullptr
Machinexa2
Jesús Cea
Asheesh Laroia
Shantanu
Kyle Stanley
Batuhan Taşkaya
Thomas Moreau
Brian Quinlan
ubordignon
Inada Naoki
Sean
Sam Martin
Dave Chevell
orlnub123
jhaydaman
