Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/joblib/externals/loky/process_executor.py: 19%

1############################################################################### 

2# Re-implementation of the ProcessPoolExecutor more robust to faults 

3# 

4# author: Thomas Moreau and Olivier Grisel 

5# 

6# adapted from concurrent/futures/process_pool_executor.py (17/02/2017) 

7# * Add an extra management thread to detect executor_manager_thread failures, 

8# * Improve the shutdown process to avoid deadlocks, 

9# * Add timeout for workers, 

10# * More robust pickling process. 

11# 

12# Copyright 2009 Brian Quinlan. All Rights Reserved. 

13# Licensed to PSF under a Contributor Agreement. 

14 

15"""Implements ProcessPoolExecutor. 

16 

17The follow diagram and text describe the data-flow through the system: 

18 

19|======================= In-process =====================|== Out-of-process ==| 

20 

21+----------+ +----------+ +--------+ +-----------+ +---------+ 

22| | => | Work Ids | | | | Call Q | | Process | 

23| | +----------+ | | +-----------+ | Pool | 

24| | | ... | | | | ... | +---------+ 

25| | | 6 | => | | => | 5, call() | => | | 

26| | | 7 | | | | ... | | | 

27| Process | | ... | | Local | +-----------+ | Process | 

28| Pool | +----------+ | Worker | | #1..n | 

29| Executor | | Thread | | | 

30| | +----------- + | | +-----------+ | | 

31| | <=> | Work Items | <=> | | <= | Result Q | <= | | 

32| | +------------+ | | +-----------+ | | 

33| | | 6: call() | | | | ... | | | 

34| | | future | +--------+ | 4, result | | | 

35| | | ... | | 3, except | | | 

36+----------+ +------------+ +-----------+ +---------+ 

37 

38Executor.submit() called: 

39- creates a uniquely numbered _WorkItem and adds it to the "Work Items" dict 

40- adds the id of the _WorkItem to the "Work Ids" queue 

41 

42Local worker thread: 

43- reads work ids from the "Work Ids" queue and looks up the corresponding 

44 WorkItem from the "Work Items" dict: if the work item has been cancelled then 

45 it is simply removed from the dict, otherwise it is repackaged as a 

46 _CallItem and put in the "Call Q". New _CallItems are put in the "Call Q" 

47 until "Call Q" is full. NOTE: the size of the "Call Q" is kept small because 

48 calls placed in the "Call Q" can no longer be cancelled with Future.cancel(). 

49- reads _ResultItems from "Result Q", updates the future stored in the 

50 "Work Items" dict and deletes the dict entry 

51 

52Process #1..n: 

53- reads _CallItems from "Call Q", executes the calls, and puts the resulting 

54 _ResultItems in "Result Q" 

55""" 

56 

57 

58__author__ = "Thomas Moreau (thomas.moreau.2010@gmail.com)" 

59 

60 

61import faulthandler 

62import os 

63import gc 

64import sys 

65import queue 

66import struct 

67import weakref 

68import warnings 

69import itertools 

70import traceback 

71import threading 

72from time import time, sleep 

73import multiprocessing as mp 

74from functools import partial 

75from pickle import PicklingError 

76from concurrent.futures import Executor, InvalidStateError 

77from concurrent.futures._base import LOGGER 

78from concurrent.futures.process import BrokenProcessPool as _BPPException 

79from multiprocessing.connection import wait 

80 

81from ._base import Future 

82from .backend import get_context 

83from .backend.context import cpu_count, _MAX_WINDOWS_WORKERS 

84from .backend.queues import Queue, SimpleQueue 

85from .backend.reduction import set_loky_pickler, get_loky_pickler_name 

86from .backend.utils import kill_process_tree, get_exitcodes_terminated_worker 

87from .initializers import _prepare_initializer 

88 

89 

90# Mechanism to prevent infinite process spawning. When a worker of a 

91# ProcessPoolExecutor nested in MAX_DEPTH Executor tries to create a new 

92# Executor, a LokyRecursionError is raised 

93MAX_DEPTH = int(os.environ.get("LOKY_MAX_DEPTH", 10)) 

94_CURRENT_DEPTH = 0 

95 

96# Minimum time interval between two consecutive memory leak protection checks. 

97_MEMORY_LEAK_CHECK_DELAY = 1.0 

98 

99# Number of bytes of memory usage allowed over the reference process size. 

100_MAX_MEMORY_LEAK_SIZE = int(3e8) 

101 

102 

103try: 

104 from psutil import Process 

105 

106 _USE_PSUTIL = True 

107 

108 def _get_memory_usage(pid, force_gc=False): 

109 if force_gc: 

110 gc.collect() 

111 

112 mem_size = Process(pid).memory_info().rss 

113 mp.util.debug(f"psutil return memory size: {mem_size}") 

114 return mem_size 

115 

116except ImportError: 

117 _USE_PSUTIL = False 

118 

119 

120class _ThreadWakeup: 

121 def __init__(self): 

122 self._closed = False 

123 self._reader, self._writer = mp.Pipe(duplex=False) 

124 

125 def close(self): 

126 if not self._closed: 

127 self._closed = True 

128 self._writer.close() 

129 self._reader.close() 

130 

131 def wakeup(self): 

132 if not self._closed: 

133 self._writer.send_bytes(b"") 

134 

135 def clear(self): 

136 if not self._closed: 

137 while self._reader.poll(): 

138 self._reader.recv_bytes() 

139 

140 

141class _ExecutorFlags: 

142 """necessary references to maintain executor states without preventing gc 

143 

144 It permits to keep the information needed by executor_manager_thread 

145 and crash_detection_thread to maintain the pool without preventing the 

146 garbage collection of unreferenced executors. 

147 """ 

148 

149 def __init__(self, shutdown_lock): 

150 

151 self.shutdown = False 

152 self.broken = None 

153 self.kill_workers = False 

154 self.shutdown_lock = shutdown_lock 

155 

156 def flag_as_shutting_down(self, kill_workers=None): 

157 with self.shutdown_lock: 

158 self.shutdown = True 

159 if kill_workers is not None: 

160 self.kill_workers = kill_workers 

161 

162 def flag_as_broken(self, broken): 

163 with self.shutdown_lock: 

164 self.shutdown = True 

165 self.broken = broken 

166 

167 

168# Prior to 3.9, executor_manager_thread is created as daemon thread. This means 

169# that it is not joined automatically when the interpreter is shutting down. 

170# To work around this problem, an exit handler is installed to tell the 

171# thread to exit when the interpreter is shutting down and then waits until 

172# it finishes. The thread needs to be daemonized because the atexit hooks are 

173# called after all non daemonized threads are joined. 

174# 

175# Starting 3.9, there exists a specific atexit hook to be called before joining 

176# the threads so the executor_manager_thread does not need to be daemonized 

177# anymore. 

178# 

179# The atexit hooks are registered when starting the first ProcessPoolExecutor 

180# to avoid import having an effect on the interpreter. 

181 

182_global_shutdown = False 

183_global_shutdown_lock = threading.Lock() 

184_threads_wakeups = weakref.WeakKeyDictionary() 

185 

186 

187def _python_exit(): 

188 global _global_shutdown 

189 _global_shutdown = True 

190 

191 # Materialize the list of items to avoid error due to iterating over 

192 # changing size dictionary. 

193 items = list(_threads_wakeups.items()) 

194 if len(items) > 0: 

195 mp.util.debug( 

196 f"Interpreter shutting down. Waking up {len(items)}" 

197 f"executor_manager_thread:\n{items}" 

198 ) 

199 

200 # Wake up the executor_manager_thread's so they can detect the interpreter 

201 # is shutting down and exit. 

202 for _, (shutdown_lock, thread_wakeup) in items: 

203 with shutdown_lock: 

204 thread_wakeup.wakeup() 

205 

206 # Collect the executor_manager_thread's to make sure we exit cleanly. 

207 for thread, _ in items: 

208 # This locks is to prevent situations where an executor is gc'ed in one 

209 # thread while the atexit finalizer is running in another thread. 

210 with _global_shutdown_lock: 

211 thread.join() 

212 

213 

214# With the fork context, _thread_wakeups is propagated to children. 

215# Clear it after fork to avoid some situation that can cause some 

216# freeze when joining the workers. 

217mp.util.register_after_fork(_threads_wakeups, lambda obj: obj.clear()) 

218 

219 

220# Module variable to register the at_exit call 

221process_pool_executor_at_exit = None 

222 

223# Controls how many more calls than processes will be queued in the call queue. 

224# A smaller number will mean that processes spend more time idle waiting for 

225# work while a larger number will make Future.cancel() succeed less frequently 

226# (Futures in the call queue cannot be cancelled). 

227EXTRA_QUEUED_CALLS = 1 

228 

229 

230class _RemoteTraceback(Exception): 

231 """Embed stringification of remote traceback in local traceback""" 

232 

233 def __init__(self, tb=None): 

234 self.tb = f'\n"""\n{tb}"""' 

235 

236 def __str__(self): 

237 return self.tb 

238 

239 

240# Do not inherit from BaseException to mirror 

241# concurrent.futures.process._ExceptionWithTraceback 

242class _ExceptionWithTraceback: 

243 def __init__(self, exc): 

244 tb = getattr(exc, "__traceback__", None) 

245 if tb is None: 

246 _, _, tb = sys.exc_info() 

247 tb = traceback.format_exception(type(exc), exc, tb) 

248 tb = "".join(tb) 

249 self.exc = exc 

250 self.tb = tb 

251 

252 def __reduce__(self): 

253 return _rebuild_exc, (self.exc, self.tb) 

254 

255 

256def _rebuild_exc(exc, tb): 

257 exc.__cause__ = _RemoteTraceback(tb) 

258 return exc 

259 

260 

261class _WorkItem: 

262 

263 __slots__ = ["future", "fn", "args", "kwargs"] 

264 

265 def __init__(self, future, fn, args, kwargs): 

266 self.future = future 

267 self.fn = fn 

268 self.args = args 

269 self.kwargs = kwargs 

270 

271 

272class _ResultItem: 

273 def __init__(self, work_id, exception=None, result=None): 

274 self.work_id = work_id 

275 self.exception = exception 

276 self.result = result 

277 

278 

279class _CallItem: 

280 def __init__(self, work_id, fn, args, kwargs): 

281 self.work_id = work_id 

282 self.fn = fn 

283 self.args = args 

284 self.kwargs = kwargs 

285 

286 # Store the current loky_pickler so it is correctly set in the worker 

287 self.loky_pickler = get_loky_pickler_name() 

288 

289 def __call__(self): 

290 set_loky_pickler(self.loky_pickler) 

291 return self.fn(*self.args, **self.kwargs) 

292 

293 def __repr__(self): 

294 return ( 

295 f"CallItem({self.work_id}, {self.fn}, {self.args}, {self.kwargs})" 

296 ) 

297 

298 

299class _SafeQueue(Queue): 

300 """Safe Queue set exception to the future object linked to a job""" 

301 

302 def __init__( 

303 self, 

304 max_size=0, 

305 ctx=None, 

306 pending_work_items=None, 

307 running_work_items=None, 

308 thread_wakeup=None, 

309 shutdown_lock=None, 

310 reducers=None, 

311 ): 

312 self.thread_wakeup = thread_wakeup 

313 self.shutdown_lock = shutdown_lock 

314 self.pending_work_items = pending_work_items 

315 self.running_work_items = running_work_items 

316 super().__init__(max_size, reducers=reducers, ctx=ctx) 

317 

318 def _on_queue_feeder_error(self, e, obj): 

319 if isinstance(obj, _CallItem): 

320 # format traceback only works on python3 

321 if isinstance(e, struct.error): 

322 raised_error = RuntimeError( 

323 "The task could not be sent to the workers as it is too " 

324 "large for `send_bytes`." 

325 ) 

326 else: 

327 raised_error = PicklingError( 

328 "Could not pickle the task to send it to the workers." 

329 ) 

330 tb = traceback.format_exception( 

331 type(e), e, getattr(e, "__traceback__", None) 

332 ) 

333 raised_error.__cause__ = _RemoteTraceback("".join(tb)) 

334 work_item = self.pending_work_items.pop(obj.work_id, None) 

335 self.running_work_items.remove(obj.work_id) 

336 # work_item can be None if another process terminated. In this 

337 # case, the executor_manager_thread fails all work_items with 

338 # BrokenProcessPool 

339 if work_item is not None: 

340 work_item.future.set_exception(raised_error) 

341 del work_item 

342 with self.shutdown_lock: 

343 self.thread_wakeup.wakeup() 

344 else: 

345 super()._on_queue_feeder_error(e, obj) 

346 

347 

348def _get_chunks(chunksize, *iterables): 

349 """Iterates over zip()ed iterables in chunks.""" 

350 it = zip(*iterables) 

351 while True: 

352 chunk = tuple(itertools.islice(it, chunksize)) 

353 if not chunk: 

354 return 

355 yield chunk 

356 

357 

358def _process_chunk(fn, chunk): 

359 """Processes a chunk of an iterable passed to map. 

360 

361 Runs the function passed to map() on a chunk of the 

362 iterable passed to map. 

363 

364 This function is run in a separate process. 

365 

366 """ 

367 return [fn(*args) for args in chunk] 

368 

369 

370def _sendback_result(result_queue, work_id, result=None, exception=None): 

371 """Safely send back the given result or exception""" 

372 try: 

373 result_queue.put( 

374 _ResultItem(work_id, result=result, exception=exception) 

375 ) 

376 except BaseException as e: 

377 exc = _ExceptionWithTraceback(e) 

378 result_queue.put(_ResultItem(work_id, exception=exc)) 

379 

380 

381def _enable_faulthandler_if_needed(): 

382 if "PYTHONFAULTHANDLER" in os.environ: 

383 # Respect the environment variable to configure faulthandler. This 

384 # makes it possible to never enable faulthandler in the loky workers by 

385 # setting PYTHONFAULTHANDLER=0 explicitly in the environment. 

386 mp.util.debug( 

387 f"faulthandler explicitly configured by environment variable: " 

388 f"PYTHONFAULTHANDLER={os.environ['PYTHONFAULTHANDLER']}." 

389 ) 

390 else: 

391 if faulthandler.is_enabled(): 

392 # Fault handler is already enabled, possibly via a custom 

393 # initializer to customize the behavior. 

394 mp.util.debug("faulthandler already enabled.") 

395 else: 

396 # Enable faulthandler by default with default paramaters otherwise. 

397 mp.util.debug( 

398 "Enabling faulthandler to report tracebacks on worker crashes." 

399 ) 

400 faulthandler.enable() 

401 

402 

403def _process_worker( 

404 call_queue, 

405 result_queue, 

406 initializer, 

407 initargs, 

408 processes_management_lock, 

409 timeout, 

410 worker_exit_lock, 

411 current_depth, 

412): 

413 """Evaluates calls from call_queue and places the results in result_queue. 

414 

415 This worker is run in a separate process. 

416 

417 Args: 

418 call_queue: A ctx.Queue of _CallItems that will be read and 

419 evaluated by the worker. 

420 result_queue: A ctx.Queue of _ResultItems that will written 

421 to by the worker. 

422 initializer: A callable initializer, or None 

423 initargs: A tuple of args for the initializer 

424 processes_management_lock: A ctx.Lock avoiding worker timeout while 

425 some workers are being spawned. 

426 timeout: maximum time to wait for a new item in the call_queue. If that 

427 time is expired, the worker will shutdown. 

428 worker_exit_lock: Lock to avoid flagging the executor as broken on 

429 workers timeout. 

430 current_depth: Nested parallelism level, to avoid infinite spawning. 

431 """ 

432 if initializer is not None: 

433 try: 

434 initializer(*initargs) 

435 except BaseException: 

436 LOGGER.critical("Exception in initializer:", exc_info=True) 

437 # The parent will notice that the process stopped and 

438 # mark the pool broken 

439 return 

440 

441 # set the global _CURRENT_DEPTH mechanism to limit recursive call 

442 global _CURRENT_DEPTH 

443 _CURRENT_DEPTH = current_depth 

444 _process_reference_size = None 

445 _last_memory_leak_check = None 

446 pid = os.getpid() 

447 

448 mp.util.debug(f"Worker started with timeout={timeout}") 

449 _enable_faulthandler_if_needed() 

450 

451 while True: 

452 try: 

453 call_item = call_queue.get(block=True, timeout=timeout) 

454 if call_item is None: 

455 mp.util.info("Shutting down worker on sentinel") 

456 except queue.Empty: 

457 mp.util.info(f"Shutting down worker after timeout {timeout:0.3f}s") 

458 if processes_management_lock.acquire(block=False): 

459 processes_management_lock.release() 

460 call_item = None 

461 else: 

462 mp.util.info("Could not acquire processes_management_lock") 

463 continue 

464 except BaseException: 

465 previous_tb = traceback.format_exc() 

466 try: 

467 result_queue.put(_RemoteTraceback(previous_tb)) 

468 except BaseException: 

469 # If we cannot format correctly the exception, at least print 

470 # the traceback. 

471 print(previous_tb) 

472 mp.util.debug("Exiting with code 1") 

473 sys.exit(1) 

474 if call_item is None: 

475 # Notify queue management thread about worker shutdown 

476 result_queue.put(pid) 

477 is_clean = worker_exit_lock.acquire(True, timeout=30) 

478 

479 # Early notify any loky executor running in this worker process 

480 # (nested parallelism) that this process is about to shutdown to 

481 # avoid a deadlock waiting undifinitely for the worker to finish. 

482 _python_exit() 

483 

484 if is_clean: 

485 mp.util.debug("Exited cleanly") 

486 else: 

487 mp.util.info("Main process did not release worker_exit") 

488 return 

489 try: 

490 r = call_item() 

491 except BaseException as e: 

492 exc = _ExceptionWithTraceback(e) 

493 result_queue.put(_ResultItem(call_item.work_id, exception=exc)) 

494 else: 

495 _sendback_result(result_queue, call_item.work_id, result=r) 

496 del r 

497 

498 # Free the resource as soon as possible, to avoid holding onto 

499 # open files or shared memory that is not needed anymore 

500 del call_item 

501 

502 if _USE_PSUTIL: 

503 if _process_reference_size is None: 

504 # Make reference measurement after the first call 

505 _process_reference_size = _get_memory_usage(pid, force_gc=True) 

506 _last_memory_leak_check = time() 

507 continue 

508 if time() - _last_memory_leak_check > _MEMORY_LEAK_CHECK_DELAY: 

509 mem_usage = _get_memory_usage(pid) 

510 _last_memory_leak_check = time() 

511 if mem_usage - _process_reference_size < _MAX_MEMORY_LEAK_SIZE: 

512 # Memory usage stays within bounds: everything is fine. 

513 continue 

514 

515 # Check again memory usage; this time take the measurement 

516 # after a forced garbage collection to break any reference 

517 # cycles. 

518 mem_usage = _get_memory_usage(pid, force_gc=True) 

519 _last_memory_leak_check = time() 

520 if mem_usage - _process_reference_size < _MAX_MEMORY_LEAK_SIZE: 

521 # The GC managed to free the memory: everything is fine. 

522 continue 

523 

524 # The process is leaking memory: let the master process 

525 # know that we need to start a new worker. 

526 mp.util.info("Memory leak detected: shutting down worker") 

527 result_queue.put(pid) 

528 with worker_exit_lock: 

529 mp.util.debug("Exit due to memory leak") 

530 return 

531 else: 

532 # if psutil is not installed, trigger gc.collect events 

533 # regularly to limit potential memory leaks due to reference cycles 

534 if _last_memory_leak_check is None or ( 

535 time() - _last_memory_leak_check > _MEMORY_LEAK_CHECK_DELAY 

536 ): 

537 gc.collect() 

538 _last_memory_leak_check = time() 

539 

540 

541class _ExecutorManagerThread(threading.Thread): 

542 """Manages the communication between this process and the worker processes. 

543 

544 The manager is run in a local thread. 

545 

546 Args: 

547 executor: A reference to the ProcessPoolExecutor that owns 

548 this thread. A weakref will be own by the manager as well as 

549 references to internal objects used to introspect the state of 

550 the executor. 

551 """ 

552 

553 def __init__(self, executor): 

554 # Store references to necessary internals of the executor. 

555 

556 # A _ThreadWakeup to allow waking up the executor_manager_thread from 

557 # the main Thread and avoid deadlocks caused by permanently 

558 # locked queues. 

559 self.thread_wakeup = executor._executor_manager_thread_wakeup 

560 self.shutdown_lock = executor._shutdown_lock 

561 

562 # A weakref.ref to the ProcessPoolExecutor that owns this thread. Used 

563 # to determine if the ProcessPoolExecutor has been garbage collected 

564 # and that the manager can exit. 

565 # When the executor gets garbage collected, the weakref callback 

566 # will wake up the queue management thread so that it can terminate 

567 # if there is no pending work item. 

568 def weakref_cb( 

569 _, 

570 thread_wakeup=self.thread_wakeup, 

571 shutdown_lock=self.shutdown_lock, 

572 ): 

573 if mp is not None: 

574 # At this point, the multiprocessing module can already be 

575 # garbage collected. We only log debug info when still 

576 # possible. 

577 mp.util.debug( 

578 "Executor collected: triggering callback for" 

579 " QueueManager wakeup" 

580 ) 

581 with shutdown_lock: 

582 thread_wakeup.wakeup() 

583 

584 self.executor_reference = weakref.ref(executor, weakref_cb) 

585 

586 # The flags of the executor 

587 self.executor_flags = executor._flags 

588 

589 # A list of the ctx.Process instances used as workers. 

590 self.processes = executor._processes 

591 

592 # A ctx.Queue that will be filled with _CallItems derived from 

593 # _WorkItems for processing by the process workers. 

594 self.call_queue = executor._call_queue 

595 

596 # A ctx.SimpleQueue of _ResultItems generated by the process workers. 

597 self.result_queue = executor._result_queue 

598 

599 # A queue.Queue of work ids e.g. Queue([5, 6, ...]). 

600 self.work_ids_queue = executor._work_ids 

601 

602 # A dict mapping work ids to _WorkItems e.g. 

603 # {5: <_WorkItem...>, 6: <_WorkItem...>, ...} 

604 self.pending_work_items = executor._pending_work_items 

605 

606 # A list of the work_ids that are currently running 

607 self.running_work_items = executor._running_work_items 

608 

609 # A lock to avoid concurrent shutdown of workers on timeout and spawn 

610 # of new processes or shut down 

611 self.processes_management_lock = executor._processes_management_lock 

612 

613 super().__init__(name="ExecutorManagerThread") 

614 if sys.version_info < (3, 9): 

615 self.daemon = True 

616 

617 def run(self): 

618 # Main loop for the executor manager thread. 

619 

620 while True: 

621 self.add_call_item_to_queue() 

622 

623 result_item, is_broken, bpe = self.wait_result_broken_or_wakeup() 

624 

625 if is_broken: 

626 self.terminate_broken(bpe) 

627 return 

628 if result_item is not None: 

629 self.process_result_item(result_item) 

630 # Delete reference to result_item to avoid keeping references 

631 # while waiting on new results. 

632 del result_item 

633 

634 if self.is_shutting_down(): 

635 self.flag_executor_shutting_down() 

636 

637 # When only canceled futures remain in pending_work_items, our 

638 # next call to wait_result_broken_or_wakeup would hang forever. 

639 # This makes sure we have some running futures or none at all. 

640 self.add_call_item_to_queue() 

641 

642 # Since no new work items can be added, it is safe to shutdown 

643 # this thread if there are no pending work items. 

644 if not self.pending_work_items: 

645 self.join_executor_internals() 

646 return 

647 

648 def add_call_item_to_queue(self): 

649 # Fills call_queue with _WorkItems from pending_work_items. 

650 # This function never blocks. 

651 while True: 

652 if self.call_queue.full(): 

653 return 

654 try: 

655 work_id = self.work_ids_queue.get(block=False) 

656 except queue.Empty: 

657 return 

658 else: 

659 work_item = self.pending_work_items[work_id] 

660 

661 if work_item.future.set_running_or_notify_cancel(): 

662 self.running_work_items += [work_id] 

663 self.call_queue.put( 

664 _CallItem( 

665 work_id, 

666 work_item.fn, 

667 work_item.args, 

668 work_item.kwargs, 

669 ), 

670 block=True, 

671 ) 

672 else: 

673 del self.pending_work_items[work_id] 

674 continue 

675 

676 def wait_result_broken_or_wakeup(self): 

677 # Wait for a result to be ready in the result_queue while checking 

678 # that all worker processes are still running, or for a wake up 

679 # signal send. The wake up signals come either from new tasks being 

680 # submitted, from the executor being shutdown/gc-ed, or from the 

681 # shutdown of the python interpreter. 

682 result_reader = self.result_queue._reader 

683 wakeup_reader = self.thread_wakeup._reader 

684 readers = [result_reader, wakeup_reader] 

685 worker_sentinels = [p.sentinel for p in list(self.processes.values())] 

686 ready = wait(readers + worker_sentinels) 

687 

688 bpe = None 

689 is_broken = True 

690 result_item = None 

691 if result_reader in ready: 

692 try: 

693 result_item = result_reader.recv() 

694 if isinstance(result_item, _RemoteTraceback): 

695 bpe = BrokenProcessPool( 

696 "A task has failed to un-serialize. Please ensure that" 

697 " the arguments of the function are all picklable." 

698 ) 

699 bpe.__cause__ = result_item 

700 else: 

701 is_broken = False 

702 except BaseException as e: 

703 bpe = BrokenProcessPool( 

704 "A result has failed to un-serialize. Please ensure that " 

705 "the objects returned by the function are always " 

706 "picklable." 

707 ) 

708 tb = traceback.format_exception( 

709 type(e), e, getattr(e, "__traceback__", None) 

710 ) 

711 bpe.__cause__ = _RemoteTraceback("".join(tb)) 

712 

713 elif wakeup_reader in ready: 

714 # This is simply a wake-up event that might either trigger putting 

715 # more tasks in the queue or trigger the clean up of resources. 

716 is_broken = False 

717 else: 

718 # A worker has terminated and we don't know why, set the state of 

719 # the executor as broken 

720 exit_codes = "" 

721 if sys.platform != "win32": 

722 # In Windows, introspecting terminated workers exitcodes seems 

723 # unstable, therefore they are not appended in the exception 

724 # message. 

725 exit_codes = ( 

726 "\nThe exit codes of the workers are " 

727 f"{get_exitcodes_terminated_worker(self.processes)}" 

728 ) 

729 mp.util.debug( 

730 "A worker unexpectedly terminated. Workers that " 

731 "might have caused the breakage: " 

732 + str( 

733 { 

734 p.name: p.exitcode 

735 for p in list(self.processes.values()) 

736 if p is not None and p.sentinel in ready 

737 } 

738 ) 

739 ) 

740 bpe = TerminatedWorkerError( 

741 "A worker process managed by the executor was unexpectedly " 

742 "terminated. This could be caused by a segmentation fault " 

743 "while calling the function or by an excessive memory usage " 

744 "causing the Operating System to kill the worker.\n" 

745 f"{exit_codes}\n" 

746 "Detailed tracebacks of the workers should have been printed " 

747 "to stderr in the executor process if faulthandler was not " 

748 "disabled." 

749 ) 

750 

751 self.thread_wakeup.clear() 

752 

753 return result_item, is_broken, bpe 

754 

755 def process_result_item(self, result_item): 

756 # Process the received a result_item. This can be either the PID of a 

757 # worker that exited gracefully or a _ResultItem 

758 

759 if isinstance(result_item, int): 

760 # Clean shutdown of a worker using its PID, either on request 

761 # by the executor.shutdown method or by the timeout of the worker 

762 # itself: we should not mark the executor as broken. 

763 with self.processes_management_lock: 

764 p = self.processes.pop(result_item, None) 

765 

766 # p can be None if the executor is concurrently shutting down. 

767 if p is not None: 

768 p._worker_exit_lock.release() 

769 mp.util.debug( 

770 f"joining {p.name} when processing {p.pid} as result_item" 

771 ) 

772 p.join() 

773 del p 

774 

775 # Make sure the executor have the right number of worker, even if a 

776 # worker timeout while some jobs were submitted. If some work is 

777 # pending or there is less processes than running items, we need to 

778 # start a new Process and raise a warning. 

779 n_pending = len(self.pending_work_items) 

780 n_running = len(self.running_work_items) 

781 if n_pending - n_running > 0 or n_running > len(self.processes): 

782 executor = self.executor_reference() 

783 if ( 

784 executor is not None 

785 and len(self.processes) < executor._max_workers 

786 ): 

787 warnings.warn( 

788 "A worker stopped while some jobs were given to the " 

789 "executor. This can be caused by a too short worker " 

790 "timeout or by a memory leak.", 

791 UserWarning, 

792 ) 

793 with executor._processes_management_lock: 

794 executor._adjust_process_count() 

795 executor = None 

796 else: 

797 # Received a _ResultItem so mark the future as completed. 

798 work_item = self.pending_work_items.pop(result_item.work_id, None) 

799 # work_item can be None if another process terminated (see above) 

800 if work_item is not None: 

801 if result_item.exception is not None: 

802 work_item.future.set_exception(result_item.exception) 

803 else: 

804 work_item.future.set_result(result_item.result) 

805 self.running_work_items.remove(result_item.work_id) 

806 

807 def is_shutting_down(self): 

808 # Check whether we should start shutting down the executor. 

809 executor = self.executor_reference() 

810 # No more work items can be added if: 

811 # - The interpreter is shutting down OR 

812 # - The executor that owns this thread is not broken AND 

813 # * The executor that owns this worker has been collected OR 

814 # * The executor that owns this worker has been shutdown. 

815 # If the executor is broken, it should be detected in the next loop. 

816 return _global_shutdown or ( 

817 (executor is None or self.executor_flags.shutdown) 

818 and not self.executor_flags.broken 

819 ) 

820 

821 def terminate_broken(self, bpe): 

822 # Terminate the executor because it is in a broken state. The bpe 

823 # argument can be used to display more information on the error that 

824 # lead the executor into becoming broken. 

825 

826 # Mark the process pool broken so that submits fail right now. 

827 self.executor_flags.flag_as_broken(bpe) 

828 

829 # Mark pending tasks as failed. 

830 for work_item in self.pending_work_items.values(): 

831 try: 

832 work_item.future.set_exception(bpe) 

833 except InvalidStateError: 

834 # set_exception() fails if the future is cancelled: ignore it. 

835 # Trying to check if the future is cancelled before calling 

836 # set_exception() would leave a race condition if the future is 

837 # cancelled between the check and set_exception(). 

838 pass 

839 # Delete references to object. See issue16284 

840 del work_item 

841 self.pending_work_items.clear() 

842 

843 # Terminate remaining workers forcibly: the queues or their 

844 # locks may be in a dirty state and block forever. 

845 self.kill_workers(reason="broken executor") 

846 

847 # clean up resources 

848 self.join_executor_internals() 

849 

850 def flag_executor_shutting_down(self): 

851 # Flag the executor as shutting down and cancel remaining tasks if 

852 # requested as early as possible if it is not gc-ed yet. 

853 self.executor_flags.flag_as_shutting_down() 

854 

855 # Cancel pending work items if requested. 

856 if self.executor_flags.kill_workers: 

857 while self.pending_work_items: 

858 _, work_item = self.pending_work_items.popitem() 

859 work_item.future.set_exception( 

860 ShutdownExecutorError( 

861 "The Executor was shutdown with `kill_workers=True` " 

862 "before this job could complete." 

863 ) 

864 ) 

865 del work_item 

866 

867 # Kill the remaining worker forcibly to no waste time joining them 

868 self.kill_workers(reason="executor shutting down") 

869 

870 def kill_workers(self, reason=""): 

871 # Terminate the remaining workers using SIGKILL. This function also 

872 # terminates descendant workers of the children in case there is some 

873 # nested parallelism. 

874 while self.processes: 

875 _, p = self.processes.popitem() 

876 mp.util.debug(f"terminate process {p.name}, reason: {reason}") 

877 try: 

878 kill_process_tree(p) 

879 except ProcessLookupError: # pragma: no cover 

880 pass 

881 

882 def shutdown_workers(self): 

883 # shutdown all workers in self.processes 

884 

885 # Create a list to avoid RuntimeError due to concurrent modification of 

886 # processes. nb_children_alive is thus an upper bound. Also release the 

887 # processes' _worker_exit_lock to accelerate the shutdown procedure, as 

888 # there is no need for hand-shake here. 

889 with self.processes_management_lock: 

890 n_children_to_stop = 0 

891 for p in list(self.processes.values()): 

892 mp.util.debug(f"releasing worker exit lock on {p.name}") 

893 p._worker_exit_lock.release() 

894 n_children_to_stop += 1 

895 

896 mp.util.debug(f"found {n_children_to_stop} processes to stop") 

897 

898 # Send the right number of sentinels, to make sure all children are 

899 # properly terminated. Do it with a mechanism that avoid hanging on 

900 # Full queue when all workers have already been shutdown. 

901 n_sentinels_sent = 0 

902 cooldown_time = 0.001 

903 while ( 

904 n_sentinels_sent < n_children_to_stop 

905 and self.get_n_children_alive() > 0 

906 ): 

907 for _ in range(n_children_to_stop - n_sentinels_sent): 

908 try: 

909 self.call_queue.put_nowait(None) 

910 n_sentinels_sent += 1 

911 except queue.Full as e: 

912 if cooldown_time > 5.0: 

913 mp.util.info( 

914 "failed to send all sentinels and exit with error." 

915 f"\ncall_queue size={self.call_queue._maxsize}; " 

916 f" full is {self.call_queue.full()}; " 

917 ) 

918 raise e 

919 mp.util.info( 

920 "full call_queue prevented to send all sentinels at " 

921 "once, waiting..." 

922 ) 

923 sleep(cooldown_time) 

924 cooldown_time *= 1.2 

925 break 

926 

927 mp.util.debug(f"sent {n_sentinels_sent} sentinels to the call queue") 

928 

929 def join_executor_internals(self): 

930 self.shutdown_workers() 

931 

932 # Release the queue's resources as soon as possible. Flag the feeder 

933 # thread for clean exit to avoid having the crash detection thread flag 

934 # the Executor as broken during the shutdown. This is safe as either: 

935 # * We don't need to communicate with the workers anymore 

936 # * There is nothing left in the Queue buffer except None sentinels 

937 mp.util.debug("closing call_queue") 

938 self.call_queue.close() 

939 self.call_queue.join_thread() 

940 

941 # Closing result_queue 

942 mp.util.debug("closing result_queue") 

943 self.result_queue.close() 

944 

945 mp.util.debug("closing thread_wakeup") 

946 with self.shutdown_lock: 

947 self.thread_wakeup.close() 

948 

949 # If .join() is not called on the created processes then 

950 # some ctx.Queue methods may deadlock on macOS. 

951 with self.processes_management_lock: 

952 mp.util.debug(f"joining {len(self.processes)} processes") 

953 n_joined_processes = 0 

954 while True: 

955 try: 

956 pid, p = self.processes.popitem() 

957 mp.util.debug(f"joining process {p.name} with pid {pid}") 

958 p.join() 

959 n_joined_processes += 1 

960 except KeyError: 

961 break 

962 

963 mp.util.debug( 

964 "executor management thread clean shutdown of " 

965 f"{n_joined_processes} workers" 

966 ) 

967 

968 def get_n_children_alive(self):