Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/joblib/parallel.py: 15%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""
2Helpers for embarrassingly parallel code.
3"""
4# Author: Gael Varoquaux < gael dot varoquaux at normalesup dot org >
5# Copyright: 2010, Gael Varoquaux
6# License: BSD 3 clause
8from __future__ import division
10import collections
11import functools
12import itertools
13import os
14import queue
15import sys
16import threading
17import time
18import warnings
19import weakref
20from contextlib import nullcontext
21from math import floor, log10, sqrt
22from multiprocessing import TimeoutError
23from numbers import Integral
24from uuid import uuid4
26from ._multiprocessing_helpers import mp
28# Make sure that those two classes are part of the public joblib.parallel API
29# so that 3rd party backend implementers can import them from here.
30from ._parallel_backends import (
31 AutoBatchingMixin, # noqa
32 FallbackToBackend,
33 LokyBackend,
34 MultiprocessingBackend,
35 ParallelBackendBase, # noqa
36 SequentialBackend,
37 ThreadingBackend,
38 cpu_count, # noqa
39 set_thread_local_cpu_limit, # noqa
40)
41from ._utils import _Sentinel, eval_expr
42from .disk import memstr_to_bytes
43from .logger import Logger, short_format_time
45BACKENDS = {
46 "threading": ThreadingBackend,
47 "sequential": SequentialBackend,
48}
49# name of the backend used by default by Parallel outside of any context
50# managed by ``parallel_config`` or ``parallel_backend``.
52# threading is the only backend that is always everywhere
53DEFAULT_BACKEND = "threading"
54DEFAULT_THREAD_BACKEND = "threading"
55DEFAULT_PROCESS_BACKEND = "threading"
57MAYBE_AVAILABLE_BACKENDS = {"multiprocessing", "loky"}
59# if multiprocessing is available, so is loky, we set it as the default
60# backend
61if mp is not None:
62 BACKENDS["multiprocessing"] = MultiprocessingBackend
64 BACKENDS["loky"] = LokyBackend
65 DEFAULT_BACKEND = "loky"
66 DEFAULT_PROCESS_BACKEND = "loky"
68# Thread local value that can be overridden by the ``parallel_config`` context
69# manager
70_backend = threading.local()
73def _register_dask():
74 """Register Dask Backend if called with parallel_config(backend="dask")"""
75 try:
76 from ._dask import DaskDistributedBackend
78 register_parallel_backend("dask", DaskDistributedBackend)
79 except ImportError as e:
80 msg = (
81 "To use the dask.distributed backend you must install both "
82 "the `dask` and distributed modules.\n\n"
83 "See https://dask.pydata.org/en/latest/install.html for more "
84 "information."
85 )
86 raise ImportError(msg) from e
89EXTERNAL_BACKENDS = {
90 "dask": _register_dask,
91}
94# Sentinels for the default values of the Parallel constructor and
95# the parallel_config and parallel_backend context managers
96default_parallel_config = {
97 "backend": _Sentinel(default_value=None),
98 "n_jobs": _Sentinel(default_value=None),
99 "verbose": _Sentinel(default_value=0),
100 "temp_folder": _Sentinel(default_value=None),
101 "max_nbytes": _Sentinel(default_value="1M"),
102 "mmap_mode": _Sentinel(default_value="r"),
103 "prefer": _Sentinel(default_value=None),
104 "require": _Sentinel(default_value=None),
105}
108VALID_BACKEND_HINTS = ("processes", "threads", None)
109VALID_BACKEND_CONSTRAINTS = ("sharedmem", None)
112def _get_config_param(param, context_config, key):
113 """Return the value of a parallel config parameter
115 Explicitly setting it in Parallel has priority over setting in a
116 parallel_(config/backend) context manager.
117 """
118 if param is not default_parallel_config[key]:
119 # param is explicitly set, return it
120 return param
122 if context_config[key] is not default_parallel_config[key]:
123 # there's a context manager and the key is set, return it
124 return context_config[key]
126 # Otherwise, we are in the default_parallel_config,
127 # return the default value
128 return param.default_value
131def get_active_backend(
132 prefer=default_parallel_config["prefer"],
133 require=default_parallel_config["require"],
134 verbose=default_parallel_config["verbose"],
135):
136 """Return the active default backend"""
137 backend, config = _get_active_backend(prefer, require, verbose)
138 n_jobs = _get_config_param(default_parallel_config["n_jobs"], config, "n_jobs")
139 return backend, n_jobs
142def _get_active_backend(
143 prefer=default_parallel_config["prefer"],
144 require=default_parallel_config["require"],
145 verbose=default_parallel_config["verbose"],
146):
147 """Return the active default backend"""
149 backend_config = getattr(_backend, "config", default_parallel_config)
151 backend = _get_config_param(
152 default_parallel_config["backend"], backend_config, "backend"
153 )
155 prefer = _get_config_param(prefer, backend_config, "prefer")
156 require = _get_config_param(require, backend_config, "require")
157 verbose = _get_config_param(verbose, backend_config, "verbose")
159 if prefer not in VALID_BACKEND_HINTS:
160 raise ValueError(
161 f"prefer={prefer} is not a valid backend hint, "
162 f"expected one of {VALID_BACKEND_HINTS}"
163 )
164 if require not in VALID_BACKEND_CONSTRAINTS:
165 raise ValueError(
166 f"require={require} is not a valid backend constraint, "
167 f"expected one of {VALID_BACKEND_CONSTRAINTS}"
168 )
169 if prefer == "processes" and require == "sharedmem":
170 raise ValueError(
171 "prefer == 'processes' and require == 'sharedmem' are inconsistent settings"
172 )
174 explicit_backend = True
175 if backend is None:
176 # We are either outside of the scope of any parallel_(config/backend)
177 # context manager or the context manager did not set a backend.
178 # create the default backend instance now.
179 backend = BACKENDS[DEFAULT_BACKEND](nesting_level=0)
180 explicit_backend = False
182 # Try to use the backend set by the user with the context manager.
184 nesting_level = backend.nesting_level
185 uses_threads = getattr(backend, "uses_threads", False)
186 supports_sharedmem = getattr(backend, "supports_sharedmem", False)
187 # Force to use thread-based backend if the provided backend does not
188 # match the shared memory constraint or if the backend is not explicitly
189 # given and threads are preferred.
190 force_threads = (require == "sharedmem" and not supports_sharedmem) or (
191 not explicit_backend and prefer == "threads" and not uses_threads
192 )
193 force_processes = not explicit_backend and prefer == "processes" and uses_threads
195 if force_threads:
196 # This backend does not match the shared memory constraint:
197 # fallback to the default thead-based backend.
198 sharedmem_backend = BACKENDS[DEFAULT_THREAD_BACKEND](
199 nesting_level=nesting_level
200 )
201 # Warn the user if we forced the backend to thread-based, while the
202 # user explicitly specified a non-thread-based backend.
203 if verbose >= 10 and explicit_backend:
204 print(
205 f"Using {sharedmem_backend.__class__.__name__} as "
206 f"joblib backend instead of {backend.__class__.__name__} "
207 "as the latter does not provide shared memory semantics."
208 )
209 # Force to n_jobs=1 by default
210 thread_config = backend_config.copy()
211 thread_config["n_jobs"] = 1
212 return sharedmem_backend, thread_config
214 if force_processes:
215 # This backend does not match the prefer="processes" constraint:
216 # fallback to the default process-based backend.
217 process_backend = BACKENDS[DEFAULT_PROCESS_BACKEND](nesting_level=nesting_level)
219 return process_backend, backend_config.copy()
221 return backend, backend_config
224class parallel_config:
225 """Set the default backend or configuration for :class:`~joblib.Parallel`.
227 This is an alternative to directly passing keyword arguments to the
228 :class:`~joblib.Parallel` class constructor. It is particularly useful when
229 calling into library code that uses joblib internally but does not expose
230 the various parallel configuration arguments in its own API.
232 Parameters
233 ----------
234 backend: str or ParallelBackendBase instance, default=None
235 If ``backend`` is a string it must match a previously registered
236 implementation using the :func:`~register_parallel_backend` function.
238 By default the following backends are available:
240 - 'loky': single-host, process-based parallelism (used by default),
241 - 'threading': single-host, thread-based parallelism,
242 - 'multiprocessing': legacy single-host, process-based parallelism.
244 'loky' is recommended to run functions that manipulate Python objects.
245 'threading' is a low-overhead alternative that is most efficient for
246 functions that release the Global Interpreter Lock: e.g. I/O-bound
247 code or CPU-bound code in a few calls to native code that explicitly
248 releases the GIL. Note that on some rare systems (such as pyodide),
249 multiprocessing and loky may not be available, in which case joblib
250 defaults to threading.
252 In addition, if the ``dask`` and ``distributed`` Python packages are
253 installed, it is possible to use the 'dask' backend for better
254 scheduling of nested parallel calls without over-subscription and
255 potentially distribute parallel calls over a networked cluster of
256 several hosts.
258 It is also possible to use the distributed 'ray' backend for
259 distributing the workload to a cluster of nodes. See more details
260 in the Examples section below.
262 Alternatively the backend can be passed directly as an instance.
264 n_jobs: int, default=None
265 The maximum number of concurrently running jobs, such as the number
266 of Python worker processes when ``backend="loky"`` or the size of the
267 thread-pool when ``backend="threading"``.
268 This argument is converted to an integer, rounded below for float.
269 If -1 is given, `joblib` tries to use all available CPUs. The number
270 of CPUs ``n_cpus`` is obtained with :func:`~cpu_count`. ``n_cpus``
271 may be further limited if this instance is being started inside a
272 worker for another :class:`~Parallel`, to ensure the number of
273 workers (threads or processes) doesn't exceed the number of cores
274 available to the top-level process.
275 For n_jobs below -1, (n_cpus + 1 + n_jobs) are used. For instance,
276 using ``n_jobs=-2`` will result in all available CPUs but one being
277 used. n_jobs can also go above ``n_cpus``, which will cause
278 oversubscription. In some cases, slight oversubscription can be
279 beneficial, e.g., for tasks with large I/O operations.
280 If 1 is given, no parallel computing code is used at all, and the
281 behavior amounts to a simple python `for` loop. This mode is not
282 compatible with `timeout`.
283 None is a marker for 'unset' that will be interpreted as n_jobs=1
284 unless the call is performed under a :func:`~parallel_config`
285 context manager that sets another value for ``n_jobs``.
286 If n_jobs = 0 then a ValueError is raised.
288 verbose: int, default=0
289 The verbosity level: if non zero, progress messages are
290 printed. Above 50, the output is sent to stdout.
291 The frequency of the messages increases with the verbosity level.
292 If it more than 10, all iterations are reported.
294 temp_folder: str or None, default=None
295 Folder to be used by the pool for memmapping large arrays
296 for sharing memory with worker processes. If None, this will try in
297 order:
299 - a folder pointed by the ``JOBLIB_TEMP_FOLDER`` environment
300 variable,
301 - ``/dev/shm`` if the folder exists and is writable: this is a
302 RAM disk filesystem available by default on modern Linux
303 distributions,
304 - the default system temporary folder that can be
305 overridden with ``TMP``, ``TMPDIR`` or ``TEMP`` environment
306 variables, typically ``/tmp`` under Unix operating systems.
308 max_nbytes: int, str, or None, optional, default='1M'
309 Threshold on the size of arrays passed to the workers that
310 triggers automated memory mapping in temp_folder. Can be an int
311 in Bytes, or a human-readable string, e.g., '1M' for 1 megabyte.
312 Use None to disable memmapping of large arrays.
314 mmap_mode: {None, 'r+', 'r', 'w+', 'c'}, default='r'
315 Memmapping mode for numpy arrays passed to workers. None will
316 disable memmapping, other modes defined in the numpy.memmap doc:
317 https://numpy.org/doc/stable/reference/generated/numpy.memmap.html
318 Also, see 'max_nbytes' parameter documentation for more details.
320 prefer: str in {'processes', 'threads'} or None, default=None
321 Soft hint to choose the default backend.
322 The default process-based backend is 'loky' and the default
323 thread-based backend is 'threading'. Ignored if the ``backend``
324 parameter is specified.
326 require: 'sharedmem' or None, default=None
327 Hard constraint to select the backend. If set to 'sharedmem',
328 the selected backend will be single-host and thread-based.
330 inner_max_num_threads: int, default=None
331 If not None, overwrites the limit set on the number of threads
332 usable in some third-party library threadpools like OpenBLAS,
333 MKL or OpenMP. This is only used with the ``loky`` backend.
335 backend_params: dict
336 Additional parameters to pass to the backend constructor when
337 backend is a string.
339 Notes
340 -----
341 Joblib tries to limit the oversubscription by limiting the number of
342 threads usable in some third-party library threadpools like OpenBLAS, MKL
343 or OpenMP. The default limit in each worker is set to
344 ``max(cpu_count() // effective_n_jobs, 1)`` but this limit can be
345 overwritten with the ``inner_max_num_threads`` argument which will be used
346 to set this limit in the child processes.
348 .. versionadded:: 1.3
350 Examples
351 --------
352 >>> from operator import neg
353 >>> with parallel_config(backend='threading'):
354 ... print(Parallel()(delayed(neg)(i + 1) for i in range(5)))
355 ...
356 [-1, -2, -3, -4, -5]
358 To use the 'ray' joblib backend add the following lines:
360 >>> from ray.util.joblib import register_ray # doctest: +SKIP
361 >>> register_ray() # doctest: +SKIP
362 >>> with parallel_config(backend="ray"): # doctest: +SKIP
363 ... print(Parallel()(delayed(neg)(i + 1) for i in range(5)))
364 [-1, -2, -3, -4, -5]
366 """
368 def __init__(
369 self,
370 backend=default_parallel_config["backend"],
371 *,
372 n_jobs=default_parallel_config["n_jobs"],
373 verbose=default_parallel_config["verbose"],
374 temp_folder=default_parallel_config["temp_folder"],
375 max_nbytes=default_parallel_config["max_nbytes"],
376 mmap_mode=default_parallel_config["mmap_mode"],
377 prefer=default_parallel_config["prefer"],
378 require=default_parallel_config["require"],
379 inner_max_num_threads=None,
380 **backend_params,
381 ):
382 # Save the parallel info and set the active parallel config
383 self.old_parallel_config = getattr(_backend, "config", default_parallel_config)
385 backend = self._check_backend(backend, inner_max_num_threads, **backend_params)
387 new_config = {
388 "n_jobs": n_jobs,
389 "verbose": verbose,
390 "temp_folder": temp_folder,
391 "max_nbytes": max_nbytes,
392 "mmap_mode": mmap_mode,
393 "prefer": prefer,
394 "require": require,
395 "backend": backend,
396 }
397 self.parallel_config = self.old_parallel_config.copy()
398 self.parallel_config.update(
399 {k: v for k, v in new_config.items() if not isinstance(v, _Sentinel)}
400 )
402 setattr(_backend, "config", self.parallel_config)
404 def _check_backend(self, backend, inner_max_num_threads, **backend_params):
405 if backend is default_parallel_config["backend"]:
406 if inner_max_num_threads is not None or len(backend_params) > 0:
407 raise ValueError(
408 "inner_max_num_threads and other constructor "
409 "parameters backend_params are only supported "
410 "when backend is not None."
411 )
412 return backend
414 if isinstance(backend, str):
415 # Handle non-registered or missing backends
416 if backend not in BACKENDS:
417 if backend in EXTERNAL_BACKENDS:
418 register = EXTERNAL_BACKENDS[backend]
419 register()
420 elif backend in MAYBE_AVAILABLE_BACKENDS:
421 warnings.warn(
422 f"joblib backend '{backend}' is not available on "
423 f"your system, falling back to {DEFAULT_BACKEND}.",
424 UserWarning,
425 stacklevel=2,
426 )
427 BACKENDS[backend] = BACKENDS[DEFAULT_BACKEND]
428 else:
429 raise ValueError(
430 f"Invalid backend: {backend}, expected one of "
431 f"{sorted(BACKENDS.keys())}"
432 )
434 backend = BACKENDS[backend](**backend_params)
435 else:
436 if len(backend_params) > 0:
437 raise ValueError(
438 "Constructor parameters backend_params are only "
439 "supported when backend is a string."
440 )
442 if inner_max_num_threads is not None:
443 msg = (
444 f"{backend.__class__.__name__} does not accept setting the "
445 "inner_max_num_threads argument."
446 )
447 assert backend.supports_inner_max_num_threads, msg
448 backend.inner_max_num_threads = inner_max_num_threads
450 # If the nesting_level of the backend is not set previously, use the
451 # nesting level from the previous active_backend to set it
452 if backend.nesting_level is None:
453 parent_backend = self.old_parallel_config["backend"]
454 if parent_backend is default_parallel_config["backend"]:
455 nesting_level = 0
456 else:
457 nesting_level = parent_backend.nesting_level
458 backend.nesting_level = nesting_level
460 return backend
462 def __enter__(self):
463 return self.parallel_config
465 def __exit__(self, type, value, traceback):
466 self.unregister()
468 def unregister(self):
469 setattr(_backend, "config", self.old_parallel_config)
472class parallel_backend(parallel_config):
473 """Change the default backend used by Parallel inside a with block.
475 .. warning::
476 It is advised to use the :class:`~joblib.parallel_config` context
477 manager instead, which allows more fine-grained control over the
478 backend configuration.
480 If ``backend`` is a string it must match a previously registered
481 implementation using the :func:`~register_parallel_backend` function.
483 By default the following backends are available:
485 - 'loky': single-host, process-based parallelism (used by default),
486 - 'threading': single-host, thread-based parallelism,
487 - 'multiprocessing': legacy single-host, process-based parallelism.
489 'loky' is recommended to run functions that manipulate Python objects.
490 'threading' is a low-overhead alternative that is most efficient for
491 functions that release the Global Interpreter Lock: e.g. I/O-bound code or
492 CPU-bound code in a few calls to native code that explicitly releases the
493 GIL. Note that on some rare systems (such as Pyodide),
494 multiprocessing and loky may not be available, in which case joblib
495 defaults to threading.
497 You can also use the `Dask <https://docs.dask.org/en/stable/>`_ joblib
498 backend to distribute work across machines. This works well with
499 scikit-learn estimators with the ``n_jobs`` parameter, for example::
501 >>> import joblib # doctest: +SKIP
502 >>> from sklearn.model_selection import GridSearchCV # doctest: +SKIP
503 >>> from dask.distributed import Client, LocalCluster # doctest: +SKIP
505 >>> # create a local Dask cluster
506 >>> cluster = LocalCluster() # doctest: +SKIP
507 >>> client = Client(cluster) # doctest: +SKIP
508 >>> grid_search = GridSearchCV(estimator, param_grid, n_jobs=-1)
509 ... # doctest: +SKIP
510 >>> with joblib.parallel_backend("dask", scatter=[X, y]): # doctest: +SKIP
511 ... grid_search.fit(X, y)
513 It is also possible to use the distributed 'ray' backend for distributing
514 the workload to a cluster of nodes. To use the 'ray' joblib backend add
515 the following lines::
517 >>> from ray.util.joblib import register_ray # doctest: +SKIP
518 >>> register_ray() # doctest: +SKIP
519 >>> with parallel_backend("ray"): # doctest: +SKIP
520 ... print(Parallel()(delayed(neg)(i + 1) for i in range(5)))
521 [-1, -2, -3, -4, -5]
523 Alternatively the backend can be passed directly as an instance.
525 By default all available workers will be used (``n_jobs=-1``) unless the
526 caller passes an explicit value for the ``n_jobs`` parameter.
528 This is an alternative to passing a ``backend='backend_name'`` argument to
529 the :class:`~Parallel` class constructor. It is particularly useful when
530 calling into library code that uses joblib internally but does not expose
531 the backend argument in its own API.
533 >>> from operator import neg
534 >>> with parallel_backend('threading'):
535 ... print(Parallel()(delayed(neg)(i + 1) for i in range(5)))
536 ...
537 [-1, -2, -3, -4, -5]
539 Joblib also tries to limit the oversubscription by limiting the number of
540 threads usable in some third-party library threadpools like OpenBLAS, MKL
541 or OpenMP. The default limit in each worker is set to
542 ``max(cpu_count() // effective_n_jobs, 1)`` but this limit can be
543 overwritten with the ``inner_max_num_threads`` argument which will be used
544 to set this limit in the child processes.
546 .. versionadded:: 0.10
548 See Also
549 --------
550 joblib.parallel_config: context manager to change the backend configuration.
551 """
553 def __init__(
554 self, backend, n_jobs=-1, inner_max_num_threads=None, **backend_params
555 ):
556 super().__init__(
557 backend=backend,
558 n_jobs=n_jobs,
559 inner_max_num_threads=inner_max_num_threads,
560 **backend_params,
561 )
563 if self.old_parallel_config is None:
564 self.old_backend_and_jobs = None
565 else:
566 self.old_backend_and_jobs = (
567 self.old_parallel_config["backend"],
568 self.old_parallel_config["n_jobs"],
569 )
570 self.new_backend_and_jobs = (
571 self.parallel_config["backend"],
572 self.parallel_config["n_jobs"],
573 )
575 def __enter__(self):
576 return self.new_backend_and_jobs
579# Under Linux or OS X the default start method of multiprocessing
580# can cause third party libraries to crash. Under Python 3.4+ it is possible
581# to set an environment variable to switch the default start method from
582# 'fork' to 'forkserver' or 'spawn' to avoid this issue albeit at the cost
583# of causing semantic changes and some additional pool instantiation overhead.
584DEFAULT_MP_CONTEXT = None
585if hasattr(mp, "get_context"):
586 method = os.environ.get("JOBLIB_START_METHOD", "").strip() or None
587 if method is not None:
588 DEFAULT_MP_CONTEXT = mp.get_context(method=method)
591class BatchedCalls(object):
592 """Wrap a sequence of (func, args, kwargs) tuples as a single callable"""
594 def __init__(
595 self, iterator_slice, backend_and_jobs, reducer_callback=None, pickle_cache=None
596 ):
597 self.items = list(iterator_slice)
598 self._size = len(self.items)
599 self._reducer_callback = reducer_callback
600 if isinstance(backend_and_jobs, tuple):
601 self._backend, self._n_jobs = backend_and_jobs
602 else:
603 # this is for backward compatibility purposes. Before 0.12.6,
604 # nested backends were returned without n_jobs indications.
605 self._backend, self._n_jobs = backend_and_jobs, None
606 self._pickle_cache = pickle_cache if pickle_cache is not None else {}
608 def __call__(self):
609 # Set the default nested backend to self._backend but do not set the
610 # change the default number of processes to -1
611 with parallel_config(backend=self._backend, n_jobs=self._n_jobs):
612 return [func(*args, **kwargs) for func, args, kwargs in self.items]
614 def __reduce__(self):
615 if self._reducer_callback is not None:
616 self._reducer_callback()
617 # no need to pickle the callback.
618 return (
619 BatchedCalls,
620 (self.items, (self._backend, self._n_jobs), None, self._pickle_cache),
621 )
623 def __len__(self):
624 return self._size
627# Possible exit status for a task
628TASK_DONE = "Done"
629TASK_ERROR = "Error"
630TASK_PENDING = "Pending"
633###############################################################################
634# For verbosity
637def _verbosity_filter(index, verbose):
638 """Returns False for indices increasingly apart, the distance
639 depending on the value of verbose.
641 We use a lag increasing as the square of index
642 """
643 if not verbose:
644 return True
645 elif verbose > 10:
646 return False
647 if index == 0:
648 return False
649 verbose = 0.5 * (11 - verbose) ** 2
650 scale = sqrt(index / verbose)
651 next_scale = sqrt((index + 1) / verbose)
652 return int(next_scale) == int(scale)
655###############################################################################
656def delayed(function):
657 """Decorator used to capture the arguments of a function.
659 Parameters
660 ----------
661 function: callable
662 The function to be decorated.
664 Returns
665 -------
666 callable
667 A new function ``F`` such that calling ``F(*args, **kwargs)``
668 returns a tuple ``(function, args, kwargs)``, allowing the later
669 execution of ``function(*args, **kwargs)``.
671 Notes
672 -----
673 Be careful about the order in which decorators are applied, especially
674 when using :meth:`Memory.cache<joblib.Memory.cache>`. For instance,
675 ``Memory.cache(delayed(func))`` will cache the outputs of
676 ``delayed(func)``, that is, tuples of the form ``(func, args, kwargs)``.
677 To cache the outputs of ``func`` itself, you must instead use
678 ``delayed(Memory.cache(func))``.
679 """
681 def delayed_function(*args, **kwargs):
682 return function, args, kwargs
684 try:
685 delayed_function = functools.wraps(function)(delayed_function)
686 except AttributeError:
687 " functools.wraps fails on some callable objects "
688 return delayed_function
691###############################################################################
692class BatchCompletionCallBack(object):
693 """Callback to keep track of completed results and schedule the next tasks.
695 This callable is executed by the parent process whenever a worker process
696 has completed a batch of tasks.
698 It is used for progress reporting, to update estimate of the batch
699 processing duration and to schedule the next batch of tasks to be
700 processed.
702 It is assumed that this callback will always be triggered by the backend
703 right after the end of a task, in case of success as well as in case of
704 failure.
705 """
707 ##########################################################################
708 # METHODS CALLED BY THE MAIN THREAD #
709 ##########################################################################
710 def __init__(self, dispatch_timestamp, batch_size, parallel):
711 self.dispatch_timestamp = dispatch_timestamp
712 self.batch_size = batch_size
713 self.parallel = parallel
714 self.parallel_call_id = parallel._call_id
715 self._completion_timeout_counter = None
717 # Internals to keep track of the status and outcome of the task.
719 # Used to hold a reference to the future-like object returned by the
720 # backend after launching this task
721 # This will be set later when calling `register_job`, as it is only
722 # created once the task has been submitted.
723 self.job = None
725 if not parallel._backend.supports_retrieve_callback:
726 # The status is only used for asynchronous result retrieval in the
727 # callback.
728 self.status = None
729 else:
730 # The initial status for the job is TASK_PENDING.
731 # Once it is done, it will be either TASK_DONE, or TASK_ERROR.
732 self.status = TASK_PENDING
734 def register_job(self, job):
735 """Register the object returned by `submit`."""
736 self.job = job
738 def get_result(self, timeout):
739 """Returns the raw result of the task that was submitted.
741 If the task raised an exception rather than returning, this same
742 exception will be raised instead.
744 If the backend supports the retrieval callback, it is assumed that this
745 method is only called after the result has been registered. It is
746 ensured by checking that `self.status(timeout)` does not return
747 TASK_PENDING. In this case, `get_result` directly returns the
748 registered result (or raise the registered exception).
750 For other backends, there are no such assumptions, but `get_result`
751 still needs to synchronously retrieve the result before it can
752 return it or raise. It will block at most `self.timeout` seconds
753 waiting for retrieval to complete, after that it raises a TimeoutError.
754 """
756 backend = self.parallel._backend
758 if backend.supports_retrieve_callback:
759 # We assume that the result has already been retrieved by the
760 # callback thread, and is stored internally. It's just waiting to
761 # be returned.
762 return self._return_or_raise()
764 # For other backends, the main thread needs to run the retrieval step.
765 try:
766 result = backend.retrieve_result(self.job, timeout=timeout)
767 outcome = dict(result=result, status=TASK_DONE)
768 except BaseException as e:
769 outcome = dict(result=e, status=TASK_ERROR)
770 self._register_outcome(outcome)
772 return self._return_or_raise()
774 def _return_or_raise(self):
775 try:
776 if self.status == TASK_ERROR:
777 raise self._result
778 return self._result
779 finally:
780 del self._result
782 def get_status(self, timeout):
783 """Get the status of the task.
785 This function also checks if the timeout has been reached and register
786 the TimeoutError outcome when it is the case.
787 """
788 if timeout is None or self.status != TASK_PENDING:
789 return self.status
791 # The computation are running and the status is pending.
792 # Check that we did not wait for this jobs more than `timeout`.
793 now = time.time()
794 if self._completion_timeout_counter is None:
795 self._completion_timeout_counter = now
797 if (now - self._completion_timeout_counter) > timeout:
798 outcome = dict(result=TimeoutError(), status=TASK_ERROR)
799 self._register_outcome(outcome)
801 return self.status
803 ##########################################################################
804 # METHODS CALLED BY CALLBACK THREADS #
805 ##########################################################################
806 def __call__(self, *args, **kwargs):
807 """Function called by the callback thread after a job is completed."""
809 # If the backend doesn't support callback retrievals, the next batch of
810 # tasks is dispatched regardless. The result will be retrieved by the
811 # main thread when calling `get_result`.
812 if not self.parallel._backend.supports_retrieve_callback:
813 self._dispatch_new()
814 return
816 # If the backend supports retrieving the result in the callback, it
817 # registers the task outcome (TASK_ERROR or TASK_DONE), and schedules
818 # the next batch if needed.
819 with self.parallel._lock:
820 # Edge case where while the task was processing, the `parallel`
821 # instance has been reset and a new call has been issued, but the
822 # worker managed to complete the task and trigger this callback
823 # call just before being aborted by the reset.
824 if self.parallel._call_id != self.parallel_call_id:
825 return
827 # When aborting, stop as fast as possible and do not retrieve the
828 # result as it won't be returned by the Parallel call.
829 if self.parallel._aborting:
830 return
832 # Retrieves the result of the task in the main process and dispatch
833 # a new batch if needed.
834 job_succeeded = self._retrieve_result(*args, **kwargs)
836 if job_succeeded:
837 self._dispatch_new()
839 def _dispatch_new(self):
840 """Schedule the next batch of tasks to be processed."""
842 # This steps ensure that auto-batching works as expected.
843 this_batch_duration = time.time() - self.dispatch_timestamp
844 self.parallel._backend.batch_completed(self.batch_size, this_batch_duration)
846 # Schedule the next batch of tasks.
847 with self.parallel._lock:
848 self.parallel.n_completed_tasks += self.batch_size
849 self.parallel.print_progress()
850 if self.parallel._original_iterator is not None:
851 self.parallel.dispatch_next()
853 def _retrieve_result(self, out):
854 """Fetch and register the outcome of a task.
856 Return True if the task succeeded, False otherwise.
857 This function is only called by backends that support retrieving
858 the task result in the callback thread.
859 """
860 try:
861 result = self.parallel._backend.retrieve_result_callback(out)
862 outcome = dict(status=TASK_DONE, result=result)
863 except BaseException as e:
864 # Avoid keeping references to parallel in the error.
865 e.__traceback__ = None
866 outcome = dict(result=e, status=TASK_ERROR)
868 self._register_outcome(outcome)
869 return outcome["status"] != TASK_ERROR
871 ##########################################################################
872 # This method can be called either in the main thread #
873 # or in the callback thread. #
874 ##########################################################################
875 def _register_outcome(self, outcome):
876 """Register the outcome of a task.
878 This method can be called only once, future calls will be ignored.
879 """
880 # Covers the edge case where the main thread tries to register a
881 # `TimeoutError` while the callback thread tries to register a result
882 # at the same time.
883 with self.parallel._lock:
884 if self.status not in (TASK_PENDING, None):
885 return
886 self.status = outcome["status"]
888 self._result = outcome["result"]
890 # Once the result and the status are extracted, the last reference to
891 # the job can be deleted.
892 self.job = None
894 # As soon as an error as been spotted, early stopping flags are sent to
895 # the `parallel` instance.
896 if self.status == TASK_ERROR:
897 self.parallel._exception = True
898 self.parallel._aborting = True
900 if self.parallel.return_ordered:
901 return
903 with self.parallel._lock:
904 # For `return_as=generator_unordered`, append the job to the queue
905 # in the order of completion instead of submission.
906 self.parallel._jobs.append(self)
909###############################################################################
910def register_parallel_backend(name, factory, make_default=False):
911 """Register a new Parallel backend factory.
913 The new backend can then be selected by passing its name as the backend
914 argument to the :class:`~Parallel` class. Moreover, the default backend can
915 be overwritten globally by setting make_default=True.
917 The factory can be any callable that takes no argument and return an
918 instance of ``ParallelBackendBase``.
920 Warning: this function is experimental and subject to change in a future
921 version of joblib.
923 .. versionadded:: 0.10
924 """
925 BACKENDS[name] = factory
926 if make_default:
927 global DEFAULT_BACKEND
928 DEFAULT_BACKEND = name
931def effective_n_jobs(n_jobs=-1):
932 """Determine the number of jobs that can actually run in parallel
934 n_jobs is the number of workers requested by the callers. Passing n_jobs=-1
935 means requesting all available workers for instance matching the number of
936 CPU cores on the worker host(s).
938 This method should return a guesstimate of the number of workers that can
939 actually perform work concurrently with the currently enabled default
940 backend. The primary use case is to make it possible for the caller to know
941 in how many chunks to slice the work.
943 In general working on larger data chunks is more efficient (less scheduling
944 overhead and better use of CPU cache prefetching heuristics) as long as all
945 the workers have enough work to do.
947 Warning: this function is experimental and subject to change in a future
948 version of joblib.
950 .. versionadded:: 0.10
951 """
952 if n_jobs == 1:
953 return 1
955 backend, backend_n_jobs = get_active_backend()
956 if n_jobs is None:
957 n_jobs = backend_n_jobs
958 return backend.effective_n_jobs(n_jobs=n_jobs)
961###############################################################################
962class Parallel(Logger):
963 """Helper class for readable parallel mapping.
965 Read more in the :ref:`User Guide <parallel>`.
967 Parameters
968 ----------
969 n_jobs: int, default=None
970 The maximum number of concurrently running jobs, such as the number
971 of Python worker processes when ``backend="loky"`` or the size of
972 the thread-pool when ``backend="threading"``.
973 This argument is converted to an integer, rounded below for float.
974 If -1 is given, `joblib` tries to use all available CPUs. The number
975 of CPUs ``n_cpus`` is obtained with :func:`~cpu_count`. ``n_cpus``
976 may be further limited if this instance is being started inside a
977 worker for another :class:`~Parallel`, to ensure the number of
978 workers (threads or processes) doesn't exceed the number of cores
979 available to the top-level process.
980 For n_jobs below -1, (n_cpus + 1 + n_jobs) are used. For instance,
981 using ``n_jobs=-2`` will result in all available CPUs but one being
982 used. n_jobs can also go above ``n_cpus``, which will cause
983 oversubscription. In some cases, slight oversubscription can be
984 beneficial, e.g., for tasks with large I/O operations.
985 If 1 is given, no parallel computing code is used at all, and the
986 behavior amounts to a simple python `for` loop. This mode is not
987 compatible with ``timeout``.
988 None is a marker for 'unset' that will be interpreted as n_jobs=1
989 unless the call is performed under a :func:`~parallel_config`
990 context manager that sets another value for ``n_jobs``.
991 If n_jobs = 0 then a ValueError is raised.
992 backend: str, ParallelBackendBase instance or None, default='loky'
993 Specify the parallelization backend implementation.
994 Supported backends are:
996 - "loky" used by default, can induce some
997 communication and memory overhead when exchanging input and
998 output data with the worker Python processes. On some rare
999 systems (such as Pyiodide), the loky backend may not be
1000 available.
1001 - "multiprocessing" previous process-based backend based on
1002 `multiprocessing.Pool`. Less robust than `loky`.
1003 - "threading" is a very low-overhead backend but it suffers
1004 from the Python Global Interpreter Lock if the called function
1005 relies a lot on Python objects. "threading" is mostly useful
1006 when the execution bottleneck is a compiled extension that
1007 explicitly releases the GIL (for instance a Cython loop wrapped
1008 in a "with nogil" block or an expensive call to a library such
1009 as NumPy).
1010 - finally, you can register backends by calling
1011 :func:`~register_parallel_backend`. This will allow you to
1012 implement a backend of your liking.
1014 It is not recommended to hard-code the backend name in a call to
1015 :class:`~Parallel` in a library. Instead it is recommended to set
1016 soft hints (prefer) or hard constraints (require) so as to make it
1017 possible for library users to change the backend from the outside
1018 using the :func:`~parallel_config` context manager.
1019 return_as: str in {'list', 'generator', 'generator_unordered'}, default='list'
1020 If 'list', calls to this instance will return a list, only when
1021 all results have been processed and retrieved.
1022 If 'generator', it will return a generator that yields the results
1023 as soon as they are available, in the order the tasks have been
1024 submitted with.
1025 If 'generator_unordered', the generator will immediately yield
1026 available results independently of the submission order. The output
1027 order is not deterministic in this case because it depends on the
1028 concurrency of the workers.
1029 prefer: str in {'processes', 'threads'} or None, default=None
1030 Soft hint to choose the default backend if no specific backend
1031 was selected with the :func:`~parallel_config` context manager.
1032 The default process-based backend is 'loky' and the default
1033 thread-based backend is 'threading'. Ignored if the ``backend``
1034 parameter is specified.
1035 require: 'sharedmem' or None, default=None
1036 Hard constraint to select the backend. If set to 'sharedmem',
1037 the selected backend will be single-host and thread-based even
1038 if the user asked for a non-thread based backend with
1039 :func:`~joblib.parallel_config`.
1040 verbose: int, default=0
1041 The verbosity level: if non zero, progress messages are
1042 printed. Above 50, the output is sent to stdout.
1043 The frequency of the messages increases with the verbosity level.
1044 If it more than 10, all iterations are reported.
1045 timeout: float or None, default=None
1046 Timeout limit for each task to complete. If any task takes longer
1047 a TimeOutError will be raised. Only applied when n_jobs != 1
1048 pre_dispatch: {'all', positive integer, or expression, as in '3*n_jobs'}, default='2*n_jobs'
1049 The number of batches (of tasks) to be pre-dispatched.
1050 Default is '2*n_jobs'. When batch_size="auto" this is reasonable
1051 default and the workers should never starve. Note that only basic
1052 arithmetic are allowed here and no modules can be used in this
1053 expression.
1054 batch_size: int or 'auto', default='auto'
1055 The number of atomic tasks to dispatch at once to each
1056 worker. When individual evaluations are very fast, dispatching
1057 calls to workers can be slower than sequential computation because
1058 of the overhead. Batching fast computations together can mitigate
1059 this.
1060 The ``'auto'`` strategy keeps track of the time it takes for a
1061 batch to complete, and dynamically adjusts the batch size to keep
1062 the time on the order of half a second, using a heuristic. The
1063 initial batch size is 1.
1064 ``batch_size="auto"`` with ``backend="threading"`` will dispatch
1065 batches of a single task at a time as the threading backend has
1066 very little overhead and using larger batch size has not proved to
1067 bring any gain in that case.
1068 temp_folder: str or None, default=None
1069 Folder to be used by the pool for memmapping large arrays
1070 for sharing memory with worker processes. If None, this will try in
1071 order:
1073 - a folder pointed by the JOBLIB_TEMP_FOLDER environment
1074 variable,
1075 - /dev/shm if the folder exists and is writable: this is a
1076 RAM disk filesystem available by default on modern Linux
1077 distributions,
1078 - the default system temporary folder that can be
1079 overridden with TMP, TMPDIR or TEMP environment
1080 variables, typically /tmp under Unix operating systems.
1082 Only active when ``backend="loky"`` or ``"multiprocessing"``.
1083 max_nbytes int, str, or None, optional, default='1M'
1084 Threshold on the size of arrays passed to the workers that
1085 triggers automated memory mapping in temp_folder. Can be an int
1086 in Bytes, or a human-readable string, e.g., '1M' for 1 megabyte.
1087 Use None to disable memmapping of large arrays.
1088 Only active when ``backend="loky"`` or ``"multiprocessing"``.
1089 mmap_mode: {None, 'r+', 'r', 'w+', 'c'}, default='r'
1090 Memmapping mode for numpy arrays passed to workers. None will
1091 disable memmapping, other modes defined in the numpy.memmap doc:
1092 https://numpy.org/doc/stable/reference/generated/numpy.memmap.html
1093 Also, see 'max_nbytes' parameter documentation for more details.
1094 backend_kwargs: dict, optional
1095 Additional parameters to pass to the backend `configure` method.
1097 Notes
1098 -----
1100 This object uses workers to compute in parallel the application of a
1101 function to many different arguments. The main functionality it brings
1102 in addition to using the raw multiprocessing or concurrent.futures API
1103 are (see examples for details):
1105 * More readable code, in particular since it avoids
1106 constructing list of arguments.
1108 * Easier debugging:
1109 - informative tracebacks even when the error happens on
1110 the client side
1111 - using 'n_jobs=1' enables to turn off parallel computing
1112 for debugging without changing the codepath
1113 - early capture of pickling errors
1115 * An optional progress meter.
1117 * Interruption of multiprocesses jobs with 'Ctrl-C'
1119 * Flexible pickling control for the communication to and from
1120 the worker processes.
1122 * Ability to use shared memory efficiently with worker
1123 processes for large numpy-based datastructures.
1125 Note that the intended usage is to run one call at a time. Multiple
1126 calls to the same Parallel object will result in a ``RuntimeError``
1128 Examples
1129 --------
1131 A simple example:
1133 >>> from math import sqrt
1134 >>> from joblib import Parallel, delayed
1135 >>> Parallel(n_jobs=1)(delayed(sqrt)(i**2) for i in range(10))
1136 [0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0]
1138 Reshaping the output when the function has several return
1139 values:
1141 >>> from math import modf
1142 >>> from joblib import Parallel, delayed
1143 >>> r = Parallel(n_jobs=1)(delayed(modf)(i/2.) for i in range(10))
1144 >>> res, i = zip(*r)
1145 >>> res
1146 (0.0, 0.5, 0.0, 0.5, 0.0, 0.5, 0.0, 0.5, 0.0, 0.5)
1147 >>> i
1148 (0.0, 0.0, 1.0, 1.0, 2.0, 2.0, 3.0, 3.0, 4.0, 4.0)
1150 The progress meter: the higher the value of `verbose`, the more
1151 messages:
1153 >>> from time import sleep
1154 >>> from joblib import Parallel, delayed
1155 >>> r = Parallel(n_jobs=2, verbose=10)(
1156 ... delayed(sleep)(.2) for _ in range(10)) #doctest: +SKIP
1157 [Parallel(n_jobs=2)]: Done 1 tasks | elapsed: 0.6s
1158 [Parallel(n_jobs=2)]: Done 4 tasks | elapsed: 0.8s
1159 [Parallel(n_jobs=2)]: Done 10 out of 10 | elapsed: 1.4s finished
1161 Traceback example, note how the line of the error is indicated
1162 as well as the values of the parameter passed to the function that
1163 triggered the exception, even though the traceback happens in the
1164 child process:
1166 >>> from heapq import nlargest
1167 >>> from joblib import Parallel, delayed
1168 >>> Parallel(n_jobs=2)(
1169 ... delayed(nlargest)(2, n) for n in (range(4), 'abcde', 3))
1170 ... # doctest: +SKIP
1171 -----------------------------------------------------------------------
1172 Sub-process traceback:
1173 -----------------------------------------------------------------------
1174 TypeError Mon Nov 12 11:37:46 2012
1175 PID: 12934 Python 2.7.3: /usr/bin/python
1176 ........................................................................
1177 /usr/lib/python2.7/heapq.pyc in nlargest(n=2, iterable=3, key=None)
1178 419 if n >= size:
1179 420 return sorted(iterable, key=key, reverse=True)[:n]
1180 421
1181 422 # When key is none, use simpler decoration
1182 423 if key is None:
1183 --> 424 it = izip(iterable, count(0,-1)) # decorate
1184 425 result = _nlargest(n, it)
1185 426 return map(itemgetter(0), result) # undecorate
1186 427
1187 428 # General case, slowest method
1188 TypeError: izip argument #1 must support iteration
1189 _______________________________________________________________________
1192 Using pre_dispatch in a producer/consumer situation, where the
1193 data is generated on the fly. Note how the producer is first
1194 called 3 times before the parallel loop is initiated, and then
1195 called to generate new data on the fly:
1197 >>> from math import sqrt
1198 >>> from joblib import Parallel, delayed
1199 >>> def producer():
1200 ... for i in range(6):
1201 ... print('Produced %s' % i)
1202 ... yield i
1203 >>> out = Parallel(n_jobs=2, verbose=100, pre_dispatch='1.5*n_jobs')(
1204 ... delayed(sqrt)(i) for i in producer()) #doctest: +SKIP
1205 Produced 0
1206 Produced 1
1207 Produced 2
1208 [Parallel(n_jobs=2)]: Done 1 jobs | elapsed: 0.0s
1209 Produced 3
1210 [Parallel(n_jobs=2)]: Done 2 jobs | elapsed: 0.0s
1211 Produced 4
1212 [Parallel(n_jobs=2)]: Done 3 jobs | elapsed: 0.0s
1213 Produced 5
1214 [Parallel(n_jobs=2)]: Done 4 jobs | elapsed: 0.0s
1215 [Parallel(n_jobs=2)]: Done 6 out of 6 | elapsed: 0.0s remaining: 0.0s
1216 [Parallel(n_jobs=2)]: Done 6 out of 6 | elapsed: 0.0s finished
1218 """ # noqa: E501
1220 def __init__(
1221 self,
1222 n_jobs=default_parallel_config["n_jobs"],
1223 backend=default_parallel_config["backend"],
1224 return_as="list",
1225 verbose=default_parallel_config["verbose"],
1226 timeout=None,
1227 pre_dispatch="2 * n_jobs",
1228 batch_size="auto",
1229 temp_folder=default_parallel_config["temp_folder"],
1230 max_nbytes=default_parallel_config["max_nbytes"],
1231 mmap_mode=default_parallel_config["mmap_mode"],
1232 prefer=default_parallel_config["prefer"],
1233 require=default_parallel_config["require"],
1234 **backend_kwargs,
1235 ):
1236 # Initiate parent Logger class state
1237 super().__init__()
1239 # Interpret n_jobs=None as 'unset'
1240 if n_jobs is None:
1241 n_jobs = default_parallel_config["n_jobs"]
1243 active_backend, context_config = _get_active_backend(
1244 prefer=prefer, require=require, verbose=verbose
1245 )
1247 nesting_level = active_backend.nesting_level
1249 self.verbose = _get_config_param(verbose, context_config, "verbose")
1250 self.timeout = timeout
1251 self.pre_dispatch = pre_dispatch
1253 if return_as not in {"list", "generator", "generator_unordered"}:
1254 raise ValueError(
1255 'Expected `return_as` parameter to be a string equal to "list"'
1256 f',"generator" or "generator_unordered", but got {return_as} '
1257 "instead."
1258 )
1259 self.return_as = return_as
1260 self.return_generator = return_as != "list"
1261 self.return_ordered = return_as != "generator_unordered"
1263 # Check if we are under a parallel_config or parallel_backend
1264 # context manager and use the config from the context manager
1265 # for arguments that are not explicitly set.
1266 self._backend_kwargs = {
1267 **backend_kwargs,
1268 **{
1269 k: _get_config_param(param, context_config, k)
1270 for param, k in [
1271 (max_nbytes, "max_nbytes"),
1272 (temp_folder, "temp_folder"),
1273 (mmap_mode, "mmap_mode"),
1274 (prefer, "prefer"),
1275 (require, "require"),
1276 (verbose, "verbose"),
1277 ]
1278 },
1279 }
1281 if isinstance(self._backend_kwargs["max_nbytes"], str):
1282 self._backend_kwargs["max_nbytes"] = memstr_to_bytes(
1283 self._backend_kwargs["max_nbytes"]
1284 )
1285 self._backend_kwargs["verbose"] = max(0, self._backend_kwargs["verbose"] - 50)
1287 if DEFAULT_MP_CONTEXT is not None:
1288 self._backend_kwargs["context"] = DEFAULT_MP_CONTEXT
1289 elif hasattr(mp, "get_context"):
1290 self._backend_kwargs["context"] = mp.get_context()
1292 if backend is default_parallel_config["backend"] or backend is None:
1293 backend = active_backend
1295 elif isinstance(backend, ParallelBackendBase):
1296 # Use provided backend as is, with the current nesting_level if it
1297 # is not set yet.
1298 if backend.nesting_level is None:
1299 backend.nesting_level = nesting_level
1301 elif hasattr(backend, "Pool") and hasattr(backend, "Lock"):
1302 # Make it possible to pass a custom multiprocessing context as
1303 # backend to change the start method to forkserver or spawn or
1304 # preload modules on the forkserver helper process.
1305 self._backend_kwargs["context"] = backend
1306 backend = MultiprocessingBackend(nesting_level=nesting_level)
1308 elif backend not in BACKENDS and backend in MAYBE_AVAILABLE_BACKENDS:
1309 warnings.warn(
1310 f"joblib backend '{backend}' is not available on "
1311 f"your system, falling back to {DEFAULT_BACKEND}.",
1312 UserWarning,
1313 stacklevel=2,
1314 )
1315 BACKENDS[backend] = BACKENDS[DEFAULT_BACKEND]
1316 backend = BACKENDS[DEFAULT_BACKEND](nesting_level=nesting_level)
1318 else:
1319 try:
1320 backend_factory = BACKENDS[backend]
1321 except KeyError as e:
1322 raise ValueError(
1323 "Invalid backend: %s, expected one of %r"
1324 % (backend, sorted(BACKENDS.keys()))
1325 ) from e
1326 backend = backend_factory(nesting_level=nesting_level)
1328 n_jobs = _get_config_param(n_jobs, context_config, "n_jobs")
1329 if n_jobs is None:
1330 # No specific context override and no specific value request:
1331 # default to the default of the backend.
1332 n_jobs = backend.default_n_jobs
1333 try:
1334 n_jobs = int(n_jobs)
1335 except ValueError:
1336 raise ValueError("n_jobs could not be converted to int")
1337 self.n_jobs = n_jobs
1339 if require == "sharedmem" and not getattr(backend, "supports_sharedmem", False):
1340 raise ValueError("Backend %s does not support shared memory" % backend)
1342 if batch_size == "auto" or isinstance(batch_size, Integral) and batch_size > 0:
1343 self.batch_size = batch_size
1344 else:
1345 raise ValueError(
1346 "batch_size must be 'auto' or a positive integer, got: %r" % batch_size
1347 )
1349 if not isinstance(backend, SequentialBackend):
1350 if self.return_generator and not backend.supports_return_generator:
1351 raise ValueError(
1352 "Backend {} does not support return_as={}".format(
1353 backend, return_as
1354 )
1355 )
1356 # This lock is used to coordinate the main thread of this process
1357 # with the async callback thread of our the pool.
1358 self._lock = threading.RLock()
1359 self._jobs = collections.deque()
1360 self._jobs_set = set()
1361 self._pending_outputs = list()
1362 self._ready_batches = queue.Queue()
1363 self._reducer_callback = None
1365 # Internal variables
1366 self._backend = backend
1367 self._running = False
1368 self._managed_backend = False
1369 self._id = uuid4().hex
1370 self._call_ref = None
1372 def __enter__(self):
1373 self._managed_backend = True
1374 self._calling = False
1375 self._initialize_backend()
1376 return self
1378 def __exit__(self, exc_type, exc_value, traceback):
1379 self._managed_backend = False
1380 if self.return_generator and self._calling:
1381 self._abort()
1382 self._terminate_and_reset()
1384 def _initialize_backend(self):
1385 """Build a process or thread pool and return the number of workers"""
1386 try:
1387 n_jobs = self._backend.configure(
1388 n_jobs=self.n_jobs, parallel=self, **self._backend_kwargs
1389 )
1390 if self.timeout is not None and not self._backend.supports_timeout:
1391 warnings.warn(
1392 "The backend class {!r} does not support timeout. "
1393 "You have set 'timeout={}' in Parallel but "
1394 "the 'timeout' parameter will not be used.".format(
1395 self._backend.__class__.__name__, self.timeout
1396 )
1397 )
1399 except FallbackToBackend as e:
1400 # Recursively initialize the backend in case of requested fallback.
1401 self._backend = e.backend
1402 n_jobs = self._initialize_backend()
1404 return n_jobs
1406 def _effective_n_jobs(self):
1407 if self._backend:
1408 return self._backend.effective_n_jobs(self.n_jobs)
1409 return 1
1411 def _terminate_and_reset(self):
1412 if hasattr(self._backend, "stop_call") and self._calling:
1413 self._backend.stop_call()
1414 self._calling = False
1415 if not self._managed_backend:
1416 self._backend.terminate()
1418 def _dispatch(self, batch):
1419 """Queue the batch for computing, with or without multiprocessing
1421 WARNING: this method is not thread-safe: it should be only called
1422 indirectly via dispatch_one_batch.
1424 """
1425 # If job.get() catches an exception, it closes the queue:
1426 if self._aborting:
1427 return
1429 batch_size = len(batch)
1431 self.n_dispatched_tasks += batch_size
1432 self.n_dispatched_batches += 1
1434 dispatch_timestamp = time.time()
1436 batch_tracker = BatchCompletionCallBack(dispatch_timestamp, batch_size, self)
1438 self._register_new_job(batch_tracker)
1440 # If return_ordered is False, the batch_tracker is not stored in the
1441 # jobs queue at the time of submission. Instead, it will be appended to
1442 # the queue by itself as soon as the callback is triggered to be able
1443 # to return the results in the order of completion.
1445 job = self._backend.submit(batch, callback=batch_tracker)
1446 batch_tracker.register_job(job)
1448 def _register_new_job(self, batch_tracker):
1449 if self.return_ordered:
1450 self._jobs.append(batch_tracker)
1451 else:
1452 self._jobs_set.add(batch_tracker)
1454 def dispatch_next(self):
1455 """Dispatch more data for parallel processing
1457 This method is meant to be called concurrently by the multiprocessing
1458 callback. We rely on the thread-safety of dispatch_one_batch to protect
1459 against concurrent consumption of the unprotected iterator.
1460 """
1461 if not self.dispatch_one_batch(self._original_iterator):
1462 self._iterating = False
1463 self._original_iterator = None
1465 def dispatch_one_batch(self, iterator):
1466 """Prefetch the tasks for the next batch and dispatch them.
1468 The effective size of the batch is computed here.
1469 If there are no more jobs to dispatch, return False, else return True.
1471 The iterator consumption and dispatching is protected by the same
1472 lock so calling this function should be thread safe.
1474 """
1476 if self._aborting:
1477 return False
1479 batch_size = self._get_batch_size()
1481 with self._lock:
1482 # to ensure an even distribution of the workload between workers,
1483 # we look ahead in the original iterators more than batch_size
1484 # tasks - However, we keep consuming only one batch at each
1485 # dispatch_one_batch call. The extra tasks are stored in a local
1486 # queue, _ready_batches, that is looked-up prior to re-consuming
1487 # tasks from the origal iterator.
1488 try:
1489 tasks = self._ready_batches.get(block=False)
1490 except queue.Empty:
1491 # slice the iterator n_jobs * batchsize items at a time. If the
1492 # slice returns less than that, then the current batchsize puts
1493 # too much weight on a subset of workers, while other may end
1494 # up starving. So in this case, re-scale the batch size
1495 # accordingly to distribute evenly the last items between all
1496 # workers.
1497 n_jobs = self._cached_effective_n_jobs
1498 big_batch_size = batch_size * n_jobs
1500 try:
1501 islice = list(itertools.islice(iterator, big_batch_size))
1502 except Exception as e:
1503 # Handle the fact that the generator of task raised an
1504 # exception. As this part of the code can be executed in
1505 # a thread internal to the backend, register a task with
1506 # an error that will be raised in the user's thread.
1507 if isinstance(e.__context__, queue.Empty):
1508 # Suppress the cause of the exception if it is
1509 # queue.Empty to avoid cluttered traceback. Only do it
1510 # if the __context__ is really empty to avoid messing
1511 # with causes of the original error.
1512 e.__cause__ = None
1513 batch_tracker = BatchCompletionCallBack(0, batch_size, self)
1514 self._register_new_job(batch_tracker)
1515 batch_tracker._register_outcome(dict(result=e, status=TASK_ERROR))
1516 return True
1518 if len(islice) == 0:
1519 return False
1520 elif (
1521 iterator is self._original_iterator and len(islice) < big_batch_size
1522 ):
1523 # We reached the end of the original iterator (unless
1524 # iterator is the ``pre_dispatch``-long initial slice of
1525 # the original iterator) -- decrease the batch size to
1526 # account for potential variance in the batches running
1527 # time.
1528 final_batch_size = max(1, len(islice) // (10 * n_jobs))
1529 else:
1530 final_batch_size = max(1, len(islice) // n_jobs)
1532 # enqueue n_jobs batches in a local queue
1533 for i in range(0, len(islice), final_batch_size):
1534 tasks = BatchedCalls(
1535 islice[i : i + final_batch_size],
1536 self._backend.get_nested_backend(),
1537 self._reducer_callback,
1538 self._pickle_cache,
1539 )
1540 self._ready_batches.put(tasks)
1542 # finally, get one task.
1543 tasks = self._ready_batches.get(block=False)
1544 if len(tasks) == 0:
1545 # No more tasks available in the iterator: tell caller to stop.
1546 return False
1547 else:
1548 self._dispatch(tasks)
1549 return True
1551 def _get_batch_size(self):
1552 """Returns the effective batch size for dispatch"""
1553 if self.batch_size == "auto":
1554 return self._backend.compute_batch_size()
1555 else:
1556 # Fixed batch size strategy
1557 return self.batch_size
1559 def _print(self, msg):
1560 """Display the message on stout or stderr depending on verbosity"""
1561 # XXX: Not using the logger framework: need to
1562 # learn to use logger better.
1563 if not self.verbose:
1564 return
1565 if self.verbose < 50:
1566 writer = sys.stderr.write
1567 else:
1568 writer = sys.stdout.write
1569 writer(f"[{self}]: {msg}\n")
1571 def _is_completed(self):
1572 """Check if all tasks have been completed"""
1573 return self.n_completed_tasks == self.n_dispatched_tasks and not (
1574 self._iterating or self._aborting
1575 )
1577 def print_progress(self):
1578 """Display the process of the parallel execution only a fraction
1579 of time, controlled by self.verbose.
1580 """
1582 if not self.verbose:
1583 return
1585 if self.n_tasks is not None and self.n_tasks > 0:
1586 width = floor(log10(self.n_tasks)) + 1
1587 else:
1588 width = 3
1589 elapsed_time = time.time() - self._start_time
1591 if self._is_completed():
1592 # Make sure that we get a last message telling us we are done
1593 self._print(
1594 f"Done {self.n_completed_tasks:{width}d} out of "
1595 f"{self.n_completed_tasks:{width}d} | elapsed: "
1596 f"{short_format_time(elapsed_time)} finished"
1597 )
1598 return
1600 # Original job iterator becomes None once it has been fully
1601 # consumed: at this point we know the total number of jobs and we are
1602 # able to display an estimation of the remaining time based on already
1603 # completed jobs. Otherwise, we simply display the number of completed
1604 # tasks.
1605 elif self._original_iterator is not None:
1606 if _verbosity_filter(self.n_dispatched_batches, self.verbose):
1607 return
1608 fmt_time = f"| elapsed: {short_format_time(elapsed_time)}"
1609 index = self.n_completed_tasks
1610 if self.n_tasks is not None:
1611 self._print(
1612 f"Done {index:{width}d} out of {self.n_tasks:{width}d} {fmt_time}"
1613 )
1614 else:
1615 pad = " " * (len("out of ") + width - len("tasks"))
1616 self._print(f"Done {index:{width}d} tasks {pad}{fmt_time}")
1617 else:
1618 index = self.n_completed_tasks
1619 # We are finished dispatching
1620 total_tasks = self.n_dispatched_tasks
1621 # We always display the first loop
1622 if index != 0:
1623 # Display depending on the number of remaining items
1624 # A message as soon as we finish dispatching, cursor is 0
1625 cursor = total_tasks - index + 1 - self._pre_dispatch_amount
1626 frequency = (total_tasks // self.verbose) + 1
1627 is_last_item = index + 1 == total_tasks
1628 if is_last_item or cursor % frequency:
1629 return
1630 remaining_time = (elapsed_time / max(index, 1)) * (
1631 self.n_dispatched_tasks - index
1632 )
1633 # only display status if remaining time is greater or equal to 0
1634 self._print(
1635 f"Done {index:{width}d} out of {total_tasks:{width}d} "
1636 f"| elapsed: {short_format_time(elapsed_time)} remaining: "
1637 f"{short_format_time(remaining_time)}"
1638 )
1640 def _abort(self):
1641 # Stop dispatching new jobs in the async callback thread
1642 self._aborting = True
1644 # If the backend allows it, cancel or kill remaining running
1645 # tasks without waiting for the results as we will raise
1646 # the exception we got back to the caller instead of returning
1647 # any result.
1648 backend = self._backend
1649 if not self._aborted and hasattr(backend, "abort_everything"):
1650 # If the backend is managed externally we need to make sure
1651 # to leave it in a working state to allow for future jobs
1652 # scheduling.
1653 ensure_ready = self._managed_backend
1654 backend.abort_everything(ensure_ready=ensure_ready)
1655 self._aborted = True
1657 def _start(self, iterator, pre_dispatch):
1658 # Only set self._iterating to True if at least a batch
1659 # was dispatched. In particular this covers the edge
1660 # case of Parallel used with an exhausted iterator. If
1661 # self._original_iterator is None, then this means either
1662 # that pre_dispatch == "all", n_jobs == 1 or that the first batch
1663 # was very quick and its callback already dispatched all the
1664 # remaining jobs.
1665 self._iterating = False
1666 if self.dispatch_one_batch(iterator):
1667 self._iterating = self._original_iterator is not None
1669 while self.dispatch_one_batch(iterator):
1670 pass
1672 if pre_dispatch == "all":
1673 # The iterable was consumed all at once by the above for loop.
1674 # No need to wait for async callbacks to trigger to
1675 # consumption.
1676 self._iterating = False
1678 def _get_outputs(self, iterator, pre_dispatch):
1679 """Iterator returning the tasks' output as soon as they are ready."""
1680 dispatch_thread_id = threading.get_ident()
1681 detach_generator_exit = False
1682 try:
1683 self._start(iterator, pre_dispatch)
1684 # first yield returns None, for internal use only. This ensures
1685 # that we enter the try/except block and start dispatching the
1686 # tasks.
1687 yield
1689 with self._backend.retrieval_context():
1690 yield from self._retrieve()
1692 except GeneratorExit:
1693 # The generator has been garbage collected before being fully
1694 # consumed. This aborts the remaining tasks if possible and warn
1695 # the user if necessary.
1696 self._exception = True
1698 # In some interpreters such as PyPy, GeneratorExit can be raised in
1699 # a different thread than the one used to start the dispatch of the
1700 # parallel tasks. This can lead to hang when a thread attempts to
1701 # join itself. As workaround, we detach the execution of the
1702 # aborting code to a dedicated thread. We then need to make sure
1703 # the rest of the function does not call `_terminate_and_reset`
1704 # in finally.
1705 if dispatch_thread_id != threading.get_ident():
1706 warnings.warn(
1707 "A generator produced by joblib.Parallel has been "
1708 "gc'ed in an unexpected thread. This behavior should "
1709 "not cause major -issues but to make sure, please "
1710 "report this warning and your use case at "
1711 "https://github.com/joblib/joblib/issues so it can "
1712 "be investigated."
1713 )
1715 detach_generator_exit = True
1716 _parallel = self
1718 class _GeneratorExitThread(threading.Thread):
1719 def run(self):
1720 _parallel._abort()
1721 if _parallel.return_generator:
1722 _parallel._warn_exit_early()
1723 _parallel._terminate_and_reset()
1725 _GeneratorExitThread(name="GeneratorExitThread").start()
1726 return
1728 # Otherwise, we are in the thread that started the dispatch: we can
1729 # safely abort the execution and warn the user.
1730 self._abort()
1731 if self.return_generator:
1732 self._warn_exit_early()
1734 raise
1736 # Note: we catch any BaseException instead of just Exception instances
1737 # to also include KeyboardInterrupt
1738 except BaseException:
1739 self._exception = True
1740 self._abort()
1741 raise
1742 finally:
1743 # Store the unconsumed tasks and terminate the workers if necessary
1744 _remaining_outputs = [] if self._exception else self._jobs
1745 self._jobs = collections.deque()
1746 self._jobs_set = set()
1747 self._running = False
1748 if not detach_generator_exit:
1749 self._terminate_and_reset()
1751 while len(_remaining_outputs) > 0:
1752 batched_results = _remaining_outputs.popleft()
1753 batched_results = batched_results.get_result(self.timeout)
1754 for result in batched_results:
1755 yield result
1757 def _wait_retrieval(self):
1758 """Return True if we need to continue retrieving some tasks."""
1760 # If the input load is still being iterated over, it means that tasks
1761 # are still on the dispatch waitlist and their results will need to
1762 # be retrieved later on.
1763 if self._iterating:
1764 return True
1766 # If some of the dispatched tasks are still being processed by the
1767 # workers, wait for the compute to finish before starting retrieval
1768 if self.n_completed_tasks < self.n_dispatched_tasks:
1769 return True
1771 # For backends that does not support retrieving asynchronously the
1772 # result to the main process, all results must be carefully retrieved
1773 # in the _retrieve loop in the main thread while the backend is alive.
1774 # For other backends, the actual retrieval is done asynchronously in
1775 # the callback thread, and we can terminate the backend before the
1776 # `self._jobs` result list has been emptied. The remaining results
1777 # will be collected in the `finally` step of the generator.
1778 if not self._backend.supports_retrieve_callback:
1779 if len(self._jobs) > 0:
1780 return True
1782 return False
1784 def _retrieve(self):
1785 timeout_control_job = None
1786 while self._wait_retrieval():
1787 # If the callback thread of a worker has signaled that its task
1788 # triggered an exception, or if the retrieval loop has raised an
1789 # exception (e.g. `GeneratorExit`), exit the loop and surface the
1790 # worker traceback.
1791 if self._aborting:
1792 self._raise_error_fast()
1793 break
1795 nb_jobs = len(self._jobs)
1796 # Now wait for a job to be ready for retrieval.
1797 if self.return_ordered:
1798 # Case ordered: wait for completion (or error) of the next job
1799 # that have been dispatched and not retrieved yet. If no job
1800 # have been dispatched yet, wait for dispatch.
1801 # We assume that the time to wait for the next job to be
1802 # dispatched is always low, so that the timeout
1803 # control only have to be done on the amount of time the next
1804 # dispatched job is pending.
1805 if (nb_jobs == 0) or (
1806 self._jobs[0].get_status(timeout=self.timeout) == TASK_PENDING
1807 ):
1808 time.sleep(0.01)
1809 continue
1811 elif nb_jobs == 0:
1812 # Case unordered: jobs are added to the list of jobs to
1813 # retrieve `self._jobs` only once completed or in error, which
1814 # is too late to enable timeout control in the same way than in
1815 # the previous case.
1816 # Instead, if no job is ready to be retrieved yet, we
1817 # arbitrarily pick a dispatched job, and the timeout control is
1818 # done such that an error is raised if this control job
1819 # timeouts before any other dispatched job has completed and
1820 # been added to `self._jobs` to be retrieved.
1821 if timeout_control_job is None:
1822 try:
1823 timeout_control_job = self._jobs_set.pop()
1824 self._jobs_set.add(timeout_control_job)
1825 except KeyError:
1826 timeout_control_job = None
1828 # NB: it can be None if no job has been dispatched yet.
1829 if timeout_control_job is not None:
1830 timeout_control_job.get_status(timeout=self.timeout)
1832 time.sleep(0.01)
1833 continue
1835 elif timeout_control_job is not None:
1836 # Case unordered, when `nb_jobs > 0`:
1837 # It means that a job is ready to be retrieved, so no timeout
1838 # will occur during this iteration.
1839 # Before proceeding to retrieval of the next ready job, reset
1840 # the timeout control state to prepare the next iteration.
1841 timeout_control_job._completion_timeout_counter = None
1842 timeout_control_job = None
1844 # We need to be careful: the job list can be filling up as
1845 # we empty it and Python list are not thread-safe by
1846 # default hence the use of the lock
1847 with self._lock:
1848 batched_results = self._jobs.popleft()
1849 if not self.return_ordered:
1850 self._jobs_set.remove(batched_results)
1852 # Flatten the batched results to output one output at a time
1853 batched_results = batched_results.get_result(self.timeout)
1854 for result in batched_results:
1855 self._nb_consumed += 1
1856 yield result
1858 def _raise_error_fast(self):
1859 """If we are aborting, raise if a job caused an error."""
1861 # Find the first job whose status is TASK_ERROR if it exists.
1862 with self._lock:
1863 error_job = next(
1864 (job for job in self._jobs if job.status == TASK_ERROR), None
1865 )
1867 # If this error job exists, immediately raise the error by
1868 # calling get_result. This job might not exists if abort has been
1869 # called directly or if the generator is gc'ed.
1870 if error_job is not None:
1871 error_job.get_result(self.timeout)
1873 def _warn_exit_early(self):
1874 """Warn the user if the generator is gc'ed before being consumned."""
1875 ready_outputs = self.n_completed_tasks - self._nb_consumed
1876 is_completed = self._is_completed()
1877 msg = ""
1878 if ready_outputs:
1879 msg += (
1880 f"{ready_outputs} tasks have been successfully executed but not used."
1881 )
1882 if not is_completed:
1883 msg += " Additionally, "
1885 if not is_completed:
1886 msg += (
1887 f"{self.n_dispatched_tasks - self.n_completed_tasks} tasks "
1888 "which were still being processed by the workers have been "
1889 "cancelled."
1890 )
1892 if msg:
1893 msg += (
1894 " You could benefit from adjusting the input task "
1895 "iterator to limit unnecessary computation time."
1896 )
1898 warnings.warn(msg)
1900 def _get_sequential_output(self, iterable):
1901 """Separate loop for sequential output.
1903 This simplifies the traceback in case of errors and reduces the
1904 overhead of calling sequential tasks with `joblib`.
1905 """
1906 try:
1907 self._iterating = True
1908 self._original_iterator = iterable
1909 batch_size = self._get_batch_size()
1911 if batch_size != 1:
1912 it = iter(iterable)
1913 iterable_batched = iter(
1914 lambda: tuple(itertools.islice(it, batch_size)), ()
1915 )
1916 iterable = (task for batch in iterable_batched for task in batch)
1918 # first yield returns None, for internal use only. This ensures
1919 # that we enter the try/except block and setup the generator.
1920 yield None
1922 # Sequentially call the tasks and yield the results.
1923 for func, args, kwargs in iterable:
1924 self.n_dispatched_batches += 1
1925 self.n_dispatched_tasks += 1
1926 res = func(*args, **kwargs)
1927 self.n_completed_tasks += 1
1928 self.print_progress()
1929 yield res
1930 self._nb_consumed += 1
1931 except BaseException:
1932 self._exception = True
1933 self._aborting = True
1934 self._aborted = True
1935 raise
1936 finally:
1937 self._running = False
1938 self._iterating = False
1939 self._original_iterator = None
1940 self.print_progress()
1942 def _reset_run_tracking(self):
1943 """Reset the counters and flags used to track the execution."""
1945 # Makes sur the parallel instance was not previously running in a
1946 # thread-safe way.
1947 with getattr(self, "_lock", nullcontext()):
1948 if self._running:
1949 msg = "This Parallel instance is already running !"
1950 if self.return_generator is True:
1951 msg += (
1952 " Before submitting new tasks, you must wait for the "
1953 "completion of all the previous tasks, or clean all "
1954 "references to the output generator."
1955 )
1956 raise RuntimeError(msg)
1957 self._running = True
1959 # Counter to keep track of the task dispatched and completed.
1960 self.n_dispatched_batches = 0
1961 self.n_dispatched_tasks = 0
1962 self.n_completed_tasks = 0
1964 # Following count is incremented by one each time the user iterates
1965 # on the output generator, it is used to prepare an informative
1966 # warning message in case the generator is deleted before all the
1967 # dispatched tasks have been consumed.
1968 self._nb_consumed = 0
1970 # Following flags are used to synchronize the threads in case one of
1971 # the tasks error-out to ensure that all workers abort fast and that
1972 # the backend terminates properly.
1974 # Set to True as soon as a worker signals that a task errors-out
1975 self._exception = False
1976 # Set to True in case of early termination following an incident
1977 self._aborting = False
1978 # Set to True after abortion is complete
1979 self._aborted = False
1981 def __call__(self, iterable):
1982 """Main function to dispatch parallel tasks."""
1984 self._reset_run_tracking()
1985 self.n_tasks = len(iterable) if hasattr(iterable, "__len__") else None
1986 self._start_time = time.time()
1988 if not self._managed_backend:
1989 n_jobs = self._initialize_backend()
1990 else:
1991 n_jobs = self._effective_n_jobs()
1993 if n_jobs == 1:
1994 # If n_jobs==1, run the computation sequentially and return
1995 # immediately to avoid overheads.
1996 output = self._get_sequential_output(iterable)
1997 next(output)
1998 return output if self.return_generator else list(output)
2000 # Let's create an ID that uniquely identifies the current call. If the
2001 # call is interrupted early and that the same instance is immediately
2002 # reused, this id will be used to prevent workers that were
2003 # concurrently finalizing a task from the previous call to run the
2004 # callback.
2005 with self._lock:
2006 self._call_id = uuid4().hex
2008 # self._effective_n_jobs should be called in the Parallel.__call__
2009 # thread only -- store its value in an attribute for further queries.
2010 self._cached_effective_n_jobs = n_jobs
2012 if isinstance(self._backend, LokyBackend):
2013 # For the loky backend, we add a callback executed when reducing
2014 # BatchCalls, that makes the loky executor use a temporary folder
2015 # specific to this Parallel object when pickling temporary memmaps.
2016 # This callback is necessary to ensure that several Parallel
2017 # objects using the same reusable executor don't use the same
2018 # temporary resources.
2020 def _batched_calls_reducer_callback():
2021 # Relevant implementation detail: the following lines, called
2022 # when reducing BatchedCalls, are called in a thread-safe
2023 # situation, meaning that the context of the temporary folder
2024 # manager will not be changed in between the callback execution
2025 # and the end of the BatchedCalls pickling. The reason is that
2026 # pickling (the only place where set_current_context is used)
2027 # is done from a single thread (the queue_feeder_thread).
2028 self._backend._workers._temp_folder_manager.set_current_context( # noqa
2029 self._id
2030 )
2032 self._reducer_callback = _batched_calls_reducer_callback
2034 # self._effective_n_jobs should be called in the Parallel.__call__
2035 # thread only -- store its value in an attribute for further queries.
2036 self._cached_effective_n_jobs = n_jobs
2038 backend_name = self._backend.__class__.__name__
2039 if n_jobs == 0:
2040 raise RuntimeError("%s has no active worker." % backend_name)
2042 self._print(f"Using backend {backend_name} with {n_jobs} concurrent workers.")
2043 if hasattr(self._backend, "start_call"):
2044 self._backend.start_call()
2046 # Following flag prevents double calls to `backend.stop_call`.
2047 self._calling = True
2049 iterator = iter(iterable)
2050 pre_dispatch = self.pre_dispatch
2052 if pre_dispatch == "all":
2053 # prevent further dispatch via multiprocessing callback thread
2054 self._original_iterator = None
2055 self._pre_dispatch_amount = 0
2056 else:
2057 self._original_iterator = iterator
2058 if hasattr(pre_dispatch, "endswith"):
2059 pre_dispatch = eval_expr(pre_dispatch.replace("n_jobs", str(n_jobs)))
2060 self._pre_dispatch_amount = pre_dispatch = int(pre_dispatch)
2061 if pre_dispatch < 1:
2062 raise ValueError(
2063 "pre_dispatch must be 'all' or a positive number of "
2064 f"batches, got: {self.pre_dispatch!r}"
2065 )
2067 # The main thread will consume the first pre_dispatch items and
2068 # the remaining items will later be lazily dispatched by async
2069 # callbacks upon task completions.
2071 # TODO: this iterator should be batch_size * n_jobs
2072 iterator = itertools.islice(iterator, self._pre_dispatch_amount)
2074 # Use a caching dict for callables that are pickled with cloudpickle to
2075 # improve performances. This cache is used only in the case of
2076 # functions that are defined in the __main__ module, functions that
2077 # are defined locally (inside another function) and lambda expressions.
2078 self._pickle_cache = dict()
2080 output = self._get_outputs(iterator, pre_dispatch)
2081 self._call_ref = weakref.ref(output)
2083 # The first item from the output is blank, but it makes the interpreter
2084 # progress until it enters the Try/Except block of the generator and
2085 # reaches the first `yield` statement. This starts the asynchronous
2086 # dispatch of the tasks to the workers.
2087 next(output)
2089 return output if self.return_generator else list(output)
2091 def __repr__(self):
2092 return "%s(n_jobs=%s)" % (self.__class__.__name__, self.n_jobs)