Aluode/PerceptionLabPortable
0
1"""2Helpers for embarrassingly parallel code.3"""4# Author: Gael Varoquaux < gael dot varoquaux at normalesup dot org >5# Copyright: 2010, Gael Varoquaux6# License: BSD 3 clause7 8from __future__ import division9 10import collections11import functools12import itertools13import os14import queue15import sys16import threading17import time18import warnings19import weakref20from contextlib import nullcontext21from math import floor, log10, sqrt22from multiprocessing import TimeoutError23from numbers import Integral24from uuid import uuid425 26from ._multiprocessing_helpers import mp27 28# Make sure that those two classes are part of the public joblib.parallel API29# so that 3rd party backend implementers can import them from here.30from ._parallel_backends import (31 AutoBatchingMixin, # noqa32 FallbackToBackend,33 LokyBackend,34 MultiprocessingBackend,35 ParallelBackendBase, # noqa36 SequentialBackend,37 ThreadingBackend,38)39from ._utils import _Sentinel, eval_expr40from .disk import memstr_to_bytes41from .logger import Logger, short_format_time42 43BACKENDS = {44 "threading": ThreadingBackend,45 "sequential": SequentialBackend,46}47# name of the backend used by default by Parallel outside of any context48# managed by ``parallel_config`` or ``parallel_backend``.49 50# threading is the only backend that is always everywhere51DEFAULT_BACKEND = "threading"52DEFAULT_THREAD_BACKEND = "threading"53DEFAULT_PROCESS_BACKEND = "threading"54 55MAYBE_AVAILABLE_BACKENDS = {"multiprocessing", "loky"}56 57# if multiprocessing is available, so is loky, we set it as the default58# backend59if mp is not None:60 BACKENDS["multiprocessing"] = MultiprocessingBackend61 from .externals import loky62 63 BACKENDS["loky"] = LokyBackend64 DEFAULT_BACKEND = "loky"65 DEFAULT_PROCESS_BACKEND = "loky"66 67# Thread local value that can be overridden by the ``parallel_config`` context68# manager69_backend = threading.local()70 71 72def _register_dask():73 """Register Dask Backend if called with parallel_config(backend="dask")"""74 try:75 from ._dask import DaskDistributedBackend76 77 register_parallel_backend("dask", DaskDistributedBackend)78 except ImportError as e:79 msg = (80 "To use the dask.distributed backend you must install both "81 "the `dask` and distributed modules.\n\n"82 "See https://dask.pydata.org/en/latest/install.html for more "83 "information."84 )85 raise ImportError(msg) from e86 87 88EXTERNAL_BACKENDS = {89 "dask": _register_dask,90}91 92 93# Sentinels for the default values of the Parallel constructor and94# the parallel_config and parallel_backend context managers95default_parallel_config = {96 "backend": _Sentinel(default_value=None),97 "n_jobs": _Sentinel(default_value=None),98 "verbose": _Sentinel(default_value=0),99 "temp_folder": _Sentinel(default_value=None),100 "max_nbytes": _Sentinel(default_value="1M"),101 "mmap_mode": _Sentinel(default_value="r"),102 "prefer": _Sentinel(default_value=None),103 "require": _Sentinel(default_value=None),104}105 106 107VALID_BACKEND_HINTS = ("processes", "threads", None)108VALID_BACKEND_CONSTRAINTS = ("sharedmem", None)109 110 111def _get_config_param(param, context_config, key):112 """Return the value of a parallel config parameter113 114 Explicitly setting it in Parallel has priority over setting in a115 parallel_(config/backend) context manager.116 """117 if param is not default_parallel_config[key]:118 # param is explicitly set, return it119 return param120 121 if context_config[key] is not default_parallel_config[key]:122 # there's a context manager and the key is set, return it123 return context_config[key]124 125 # Otherwise, we are in the default_parallel_config,126 # return the default value127 return param.default_value128 129 130def get_active_backend(131 prefer=default_parallel_config["prefer"],132 require=default_parallel_config["require"],133 verbose=default_parallel_config["verbose"],134):135 """Return the active default backend"""136 backend, config = _get_active_backend(prefer, require, verbose)137 n_jobs = _get_config_param(default_parallel_config["n_jobs"], config, "n_jobs")138 return backend, n_jobs139 140 141def _get_active_backend(142 prefer=default_parallel_config["prefer"],143 require=default_parallel_config["require"],144 verbose=default_parallel_config["verbose"],145):146 """Return the active default backend"""147 148 backend_config = getattr(_backend, "config", default_parallel_config)149 150 backend = _get_config_param(151 default_parallel_config["backend"], backend_config, "backend"152 )153 154 prefer = _get_config_param(prefer, backend_config, "prefer")155 require = _get_config_param(require, backend_config, "require")156 verbose = _get_config_param(verbose, backend_config, "verbose")157 158 if prefer not in VALID_BACKEND_HINTS:159 raise ValueError(160 f"prefer={prefer} is not a valid backend hint, "161 f"expected one of {VALID_BACKEND_HINTS}"162 )163 if require not in VALID_BACKEND_CONSTRAINTS:164 raise ValueError(165 f"require={require} is not a valid backend constraint, "166 f"expected one of {VALID_BACKEND_CONSTRAINTS}"167 )168 if prefer == "processes" and require == "sharedmem":169 raise ValueError(170 "prefer == 'processes' and require == 'sharedmem' are inconsistent settings"171 )172 173 explicit_backend = True174 if backend is None:175 # We are either outside of the scope of any parallel_(config/backend)176 # context manager or the context manager did not set a backend.177 # create the default backend instance now.178 backend = BACKENDS[DEFAULT_BACKEND](nesting_level=0)179 explicit_backend = False180 181 # Try to use the backend set by the user with the context manager.182 183 nesting_level = backend.nesting_level184 uses_threads = getattr(backend, "uses_threads", False)185 supports_sharedmem = getattr(backend, "supports_sharedmem", False)186 # Force to use thread-based backend if the provided backend does not187 # match the shared memory constraint or if the backend is not explicitly188 # given and threads are preferred.189 force_threads = (require == "sharedmem" and not supports_sharedmem) or (190 not explicit_backend and prefer == "threads" and not uses_threads191 )192 force_processes = not explicit_backend and prefer == "processes" and uses_threads193 194 if force_threads:195 # This backend does not match the shared memory constraint:196 # fallback to the default thead-based backend.197 sharedmem_backend = BACKENDS[DEFAULT_THREAD_BACKEND](198 nesting_level=nesting_level199 )200 # Warn the user if we forced the backend to thread-based, while the201 # user explicitly specified a non-thread-based backend.202 if verbose >= 10 and explicit_backend:203 print(204 f"Using {sharedmem_backend.__class__.__name__} as "205 f"joblib backend instead of {backend.__class__.__name__} "206 "as the latter does not provide shared memory semantics."207 )208 # Force to n_jobs=1 by default209 thread_config = backend_config.copy()210 thread_config["n_jobs"] = 1211 return sharedmem_backend, thread_config212 213 if force_processes:214 # This backend does not match the prefer="processes" constraint:215 # fallback to the default process-based backend.216 process_backend = BACKENDS[DEFAULT_PROCESS_BACKEND](nesting_level=nesting_level)217 218 return process_backend, backend_config.copy()219 220 return backend, backend_config221 222 223class parallel_config:224 """Set the default backend or configuration for :class:`~joblib.Parallel`.225 226 This is an alternative to directly passing keyword arguments to the227 :class:`~joblib.Parallel` class constructor. It is particularly useful when228 calling into library code that uses joblib internally but does not expose229 the various parallel configuration arguments in its own API.230 231 Parameters232 ----------233 backend: str or ParallelBackendBase instance, default=None234 If ``backend`` is a string it must match a previously registered235 implementation using the :func:`~register_parallel_backend` function.236 237 By default the following backends are available:238 239 - 'loky': single-host, process-based parallelism (used by default),240 - 'threading': single-host, thread-based parallelism,241 - 'multiprocessing': legacy single-host, process-based parallelism.242 243 'loky' is recommended to run functions that manipulate Python objects.244 'threading' is a low-overhead alternative that is most efficient for245 functions that release the Global Interpreter Lock: e.g. I/O-bound246 code or CPU-bound code in a few calls to native code that explicitly247 releases the GIL. Note that on some rare systems (such as pyodide),248 multiprocessing and loky may not be available, in which case joblib249 defaults to threading.250 251 In addition, if the ``dask`` and ``distributed`` Python packages are252 installed, it is possible to use the 'dask' backend for better253 scheduling of nested parallel calls without over-subscription and254 potentially distribute parallel calls over a networked cluster of255 several hosts.256 257 It is also possible to use the distributed 'ray' backend for258 distributing the workload to a cluster of nodes. See more details259 in the Examples section below.260 261 Alternatively the backend can be passed directly as an instance.262 263 n_jobs: int, default=None264 The maximum number of concurrently running jobs, such as the number265 of Python worker processes when ``backend="loky"`` or the size of the266 thread-pool when ``backend="threading"``.267 This argument is converted to an integer, rounded below for float.268 If -1 is given, `joblib` tries to use all CPUs. The number of CPUs269 ``n_cpus`` is obtained with :func:`~cpu_count`.270 For n_jobs below -1, (n_cpus + 1 + n_jobs) are used. For instance,271 using ``n_jobs=-2`` will result in all CPUs but one being used.272 This argument can also go above ``n_cpus``, which will cause273 oversubscription. In some cases, slight oversubscription can be274 beneficial, e.g., for tasks with large I/O operations.275 If 1 is given, no parallel computing code is used at all, and the276 behavior amounts to a simple python `for` loop. This mode is not277 compatible with `timeout`.278 None is a marker for 'unset' that will be interpreted as n_jobs=1279 unless the call is performed under a :func:`~parallel_config`280 context manager that sets another value for ``n_jobs``.281 If n_jobs = 0 then a ValueError is raised.282 283 verbose: int, default=0284 The verbosity level: if non zero, progress messages are285 printed. Above 50, the output is sent to stdout.286 The frequency of the messages increases with the verbosity level.287 If it more than 10, all iterations are reported.288 289 temp_folder: str or None, default=None290 Folder to be used by the pool for memmapping large arrays291 for sharing memory with worker processes. If None, this will try in292 order:293 294 - a folder pointed by the ``JOBLIB_TEMP_FOLDER`` environment295 variable,296 - ``/dev/shm`` if the folder exists and is writable: this is a297 RAM disk filesystem available by default on modern Linux298 distributions,299 - the default system temporary folder that can be300 overridden with ``TMP``, ``TMPDIR`` or ``TEMP`` environment301 variables, typically ``/tmp`` under Unix operating systems.302 303 max_nbytes: int, str, or None, optional, default='1M'304 Threshold on the size of arrays passed to the workers that305 triggers automated memory mapping in temp_folder. Can be an int306 in Bytes, or a human-readable string, e.g., '1M' for 1 megabyte.307 Use None to disable memmapping of large arrays.308 309 mmap_mode: {None, 'r+', 'r', 'w+', 'c'}, default='r'310 Memmapping mode for numpy arrays passed to workers. None will311 disable memmapping, other modes defined in the numpy.memmap doc:312 https://numpy.org/doc/stable/reference/generated/numpy.memmap.html313 Also, see 'max_nbytes' parameter documentation for more details.314 315 prefer: str in {'processes', 'threads'} or None, default=None316 Soft hint to choose the default backend.317 The default process-based backend is 'loky' and the default318 thread-based backend is 'threading'. Ignored if the ``backend``319 parameter is specified.320 321 require: 'sharedmem' or None, default=None322 Hard constraint to select the backend. If set to 'sharedmem',323 the selected backend will be single-host and thread-based.324 325 inner_max_num_threads: int, default=None326 If not None, overwrites the limit set on the number of threads327 usable in some third-party library threadpools like OpenBLAS,328 MKL or OpenMP. This is only used with the ``loky`` backend.329 330 backend_params: dict331 Additional parameters to pass to the backend constructor when332 backend is a string.333 334 Notes335 -----336 Joblib tries to limit the oversubscription by limiting the number of337 threads usable in some third-party library threadpools like OpenBLAS, MKL338 or OpenMP. The default limit in each worker is set to339 ``max(cpu_count() // effective_n_jobs, 1)`` but this limit can be340 overwritten with the ``inner_max_num_threads`` argument which will be used341 to set this limit in the child processes.342 343 .. versionadded:: 1.3344 345 Examples346 --------347 >>> from operator import neg348 >>> with parallel_config(backend='threading'):349 ... print(Parallel()(delayed(neg)(i + 1) for i in range(5)))350 ...351 [-1, -2, -3, -4, -5]352 353 To use the 'ray' joblib backend add the following lines:354 355 >>> from ray.util.joblib import register_ray # doctest: +SKIP356 >>> register_ray() # doctest: +SKIP357 >>> with parallel_config(backend="ray"): # doctest: +SKIP358 ... print(Parallel()(delayed(neg)(i + 1) for i in range(5)))359 [-1, -2, -3, -4, -5]360 361 """362 363 def __init__(364 self,365 backend=default_parallel_config["backend"],366 *,367 n_jobs=default_parallel_config["n_jobs"],368 verbose=default_parallel_config["verbose"],369 temp_folder=default_parallel_config["temp_folder"],370 max_nbytes=default_parallel_config["max_nbytes"],371 mmap_mode=default_parallel_config["mmap_mode"],372 prefer=default_parallel_config["prefer"],373 require=default_parallel_config["require"],374 inner_max_num_threads=None,375 **backend_params,376 ):377 # Save the parallel info and set the active parallel config378 self.old_parallel_config = getattr(_backend, "config", default_parallel_config)379 380 backend = self._check_backend(backend, inner_max_num_threads, **backend_params)381 382 new_config = {383 "n_jobs": n_jobs,384 "verbose": verbose,385 "temp_folder": temp_folder,386 "max_nbytes": max_nbytes,387 "mmap_mode": mmap_mode,388 "prefer": prefer,389 "require": require,390 "backend": backend,391 }392 self.parallel_config = self.old_parallel_config.copy()393 self.parallel_config.update(394 {k: v for k, v in new_config.items() if not isinstance(v, _Sentinel)}395 )396 397 setattr(_backend, "config", self.parallel_config)398 399 def _check_backend(self, backend, inner_max_num_threads, **backend_params):400 if backend is default_parallel_config["backend"]:401 if inner_max_num_threads is not None or len(backend_params) > 0:402 raise ValueError(403 "inner_max_num_threads and other constructor "404 "parameters backend_params are only supported "405 "when backend is not None."406 )407 return backend408 409 if isinstance(backend, str):410 # Handle non-registered or missing backends411 if backend not in BACKENDS:412 if backend in EXTERNAL_BACKENDS:413 register = EXTERNAL_BACKENDS[backend]414 register()415 elif backend in MAYBE_AVAILABLE_BACKENDS:416 warnings.warn(417 f"joblib backend '{backend}' is not available on "418 f"your system, falling back to {DEFAULT_BACKEND}.",419 UserWarning,420 stacklevel=2,421 )422 BACKENDS[backend] = BACKENDS[DEFAULT_BACKEND]423 else:424 raise ValueError(425 f"Invalid backend: {backend}, expected one of "426 f"{sorted(BACKENDS.keys())}"427 )428 429 backend = BACKENDS[backend](**backend_params)430 else:431 if len(backend_params) > 0:432 raise ValueError(433 "Constructor parameters backend_params are only "434 "supported when backend is a string."435 )436 437 if inner_max_num_threads is not None:438 msg = (439 f"{backend.__class__.__name__} does not accept setting the "440 "inner_max_num_threads argument."441 )442 assert backend.supports_inner_max_num_threads, msg443 backend.inner_max_num_threads = inner_max_num_threads444 445 # If the nesting_level of the backend is not set previously, use the446 # nesting level from the previous active_backend to set it447 if backend.nesting_level is None:448 parent_backend = self.old_parallel_config["backend"]449 if parent_backend is default_parallel_config["backend"]:450 nesting_level = 0451 else:452 nesting_level = parent_backend.nesting_level453 backend.nesting_level = nesting_level454 455 return backend456 457 def __enter__(self):458 return self.parallel_config459 460 def __exit__(self, type, value, traceback):461 self.unregister()462 463 def unregister(self):464 setattr(_backend, "config", self.old_parallel_config)465 466 467class parallel_backend(parallel_config):468 """Change the default backend used by Parallel inside a with block.469 470 .. warning::471 It is advised to use the :class:`~joblib.parallel_config` context472 manager instead, which allows more fine-grained control over the473 backend configuration.474 475 If ``backend`` is a string it must match a previously registered476 implementation using the :func:`~register_parallel_backend` function.477 478 By default the following backends are available:479 480 - 'loky': single-host, process-based parallelism (used by default),481 - 'threading': single-host, thread-based parallelism,482 - 'multiprocessing': legacy single-host, process-based parallelism.483 484 'loky' is recommended to run functions that manipulate Python objects.485 'threading' is a low-overhead alternative that is most efficient for486 functions that release the Global Interpreter Lock: e.g. I/O-bound code or487 CPU-bound code in a few calls to native code that explicitly releases the488 GIL. Note that on some rare systems (such as Pyodide),489 multiprocessing and loky may not be available, in which case joblib490 defaults to threading.491 492 You can also use the `Dask <https://docs.dask.org/en/stable/>`_ joblib493 backend to distribute work across machines. This works well with494 scikit-learn estimators with the ``n_jobs`` parameter, for example::495 496 >>> import joblib # doctest: +SKIP497 >>> from sklearn.model_selection import GridSearchCV # doctest: +SKIP498 >>> from dask.distributed import Client, LocalCluster # doctest: +SKIP499 500 >>> # create a local Dask cluster501 >>> cluster = LocalCluster() # doctest: +SKIP502 >>> client = Client(cluster) # doctest: +SKIP503 >>> grid_search = GridSearchCV(estimator, param_grid, n_jobs=-1)504 ... # doctest: +SKIP505 >>> with joblib.parallel_backend("dask", scatter=[X, y]): # doctest: +SKIP506 ... grid_search.fit(X, y)507 508 It is also possible to use the distributed 'ray' backend for distributing509 the workload to a cluster of nodes. To use the 'ray' joblib backend add510 the following lines::511 512 >>> from ray.util.joblib import register_ray # doctest: +SKIP513 >>> register_ray() # doctest: +SKIP514 >>> with parallel_backend("ray"): # doctest: +SKIP515 ... print(Parallel()(delayed(neg)(i + 1) for i in range(5)))516 [-1, -2, -3, -4, -5]517 518 Alternatively the backend can be passed directly as an instance.519 520 By default all available workers will be used (``n_jobs=-1``) unless the521 caller passes an explicit value for the ``n_jobs`` parameter.522 523 This is an alternative to passing a ``backend='backend_name'`` argument to524 the :class:`~Parallel` class constructor. It is particularly useful when525 calling into library code that uses joblib internally but does not expose526 the backend argument in its own API.527 528 >>> from operator import neg529 >>> with parallel_backend('threading'):530 ... print(Parallel()(delayed(neg)(i + 1) for i in range(5)))531 ...532 [-1, -2, -3, -4, -5]533 534 Joblib also tries to limit the oversubscription by limiting the number of535 threads usable in some third-party library threadpools like OpenBLAS, MKL536 or OpenMP. The default limit in each worker is set to537 ``max(cpu_count() // effective_n_jobs, 1)`` but this limit can be538 overwritten with the ``inner_max_num_threads`` argument which will be used539 to set this limit in the child processes.540 541 .. versionadded:: 0.10542 543 See Also544 --------545 joblib.parallel_config: context manager to change the backend configuration.546 """547 548 def __init__(549 self, backend, n_jobs=-1, inner_max_num_threads=None, **backend_params550 ):551 super().__init__(552 backend=backend,553 n_jobs=n_jobs,554 inner_max_num_threads=inner_max_num_threads,555 **backend_params,556 )557 558 if self.old_parallel_config is None:559 self.old_backend_and_jobs = None560 else:561 self.old_backend_and_jobs = (562 self.old_parallel_config["backend"],563 self.old_parallel_config["n_jobs"],564 )565 self.new_backend_and_jobs = (566 self.parallel_config["backend"],567 self.parallel_config["n_jobs"],568 )569 570 def __enter__(self):571 return self.new_backend_and_jobs572 573 574# Under Linux or OS X the default start method of multiprocessing575# can cause third party libraries to crash. Under Python 3.4+ it is possible576# to set an environment variable to switch the default start method from577# 'fork' to 'forkserver' or 'spawn' to avoid this issue albeit at the cost578# of causing semantic changes and some additional pool instantiation overhead.579DEFAULT_MP_CONTEXT = None580if hasattr(mp, "get_context"):581 method = os.environ.get("JOBLIB_START_METHOD", "").strip() or None582 if method is not None:583 DEFAULT_MP_CONTEXT = mp.get_context(method=method)584 585 586class BatchedCalls(object):587 """Wrap a sequence of (func, args, kwargs) tuples as a single callable"""588 589 def __init__(590 self, iterator_slice, backend_and_jobs, reducer_callback=None, pickle_cache=None591 ):592 self.items = list(iterator_slice)593 self._size = len(self.items)594 self._reducer_callback = reducer_callback595 if isinstance(backend_and_jobs, tuple):596 self._backend, self._n_jobs = backend_and_jobs597 else:598 # this is for backward compatibility purposes. Before 0.12.6,599 # nested backends were returned without n_jobs indications.600 self._backend, self._n_jobs = backend_and_jobs, None601 self._pickle_cache = pickle_cache if pickle_cache is not None else {}602 603 def __call__(self):604 # Set the default nested backend to self._backend but do not set the605 # change the default number of processes to -1606 with parallel_config(backend=self._backend, n_jobs=self._n_jobs):607 return [func(*args, **kwargs) for func, args, kwargs in self.items]608 609 def __reduce__(self):610 if self._reducer_callback is not None:611 self._reducer_callback()612 # no need to pickle the callback.613 return (614 BatchedCalls,615 (self.items, (self._backend, self._n_jobs), None, self._pickle_cache),616 )617 618 def __len__(self):619 return self._size620 621 622# Possible exit status for a task623TASK_DONE = "Done"624TASK_ERROR = "Error"625TASK_PENDING = "Pending"626 627 628###############################################################################629# CPU count that works also when multiprocessing has been disabled via630# the JOBLIB_MULTIPROCESSING environment variable631def cpu_count(only_physical_cores=False):632 """Return the number of CPUs.633 634 This delegates to loky.cpu_count that takes into account additional635 constraints such as Linux CFS scheduler quotas (typically set by container636 runtimes such as docker) and CPU affinity (for instance using the taskset637 command on Linux).638 639 Parameters640 ----------641 only_physical_cores : boolean, default=False642 If True, does not take hyperthreading / SMT logical cores into account.643 644 """645 if mp is None:646 return 1647 648 return loky.cpu_count(only_physical_cores=only_physical_cores)649 650 651###############################################################################652# For verbosity653 654 655def _verbosity_filter(index, verbose):656 """Returns False for indices increasingly apart, the distance657 depending on the value of verbose.658 659 We use a lag increasing as the square of index660 """661 if not verbose:662 return True663 elif verbose > 10:664 return False665 if index == 0:666 return False667 verbose = 0.5 * (11 - verbose) ** 2668 scale = sqrt(index / verbose)669 next_scale = sqrt((index + 1) / verbose)670 return int(next_scale) == int(scale)671 672 673###############################################################################674def delayed(function):675 """Decorator used to capture the arguments of a function."""676 677 def delayed_function(*args, **kwargs):678 return function, args, kwargs679 680 try:681 delayed_function = functools.wraps(function)(delayed_function)682 except AttributeError:683 " functools.wraps fails on some callable objects "684 return delayed_function685 686 687###############################################################################688class BatchCompletionCallBack(object):689 """Callback to keep track of completed results and schedule the next tasks.690 691 This callable is executed by the parent process whenever a worker process692 has completed a batch of tasks.693 694 It is used for progress reporting, to update estimate of the batch695 processing duration and to schedule the next batch of tasks to be696 processed.697 698 It is assumed that this callback will always be triggered by the backend699 right after the end of a task, in case of success as well as in case of700 failure.701 """702 703 ##########################################################################704 # METHODS CALLED BY THE MAIN THREAD #705 ##########################################################################706 def __init__(self, dispatch_timestamp, batch_size, parallel):707 self.dispatch_timestamp = dispatch_timestamp708 self.batch_size = batch_size709 self.parallel = parallel710 self.parallel_call_id = parallel._call_id711 self._completion_timeout_counter = None712 713 # Internals to keep track of the status and outcome of the task.714 715 # Used to hold a reference to the future-like object returned by the716 # backend after launching this task717 # This will be set later when calling `register_job`, as it is only718 # created once the task has been submitted.719 self.job = None720 721 if not parallel._backend.supports_retrieve_callback:722 # The status is only used for asynchronous result retrieval in the723 # callback.724 self.status = None725 else:726 # The initial status for the job is TASK_PENDING.727 # Once it is done, it will be either TASK_DONE, or TASK_ERROR.728 self.status = TASK_PENDING729 730 def register_job(self, job):731 """Register the object returned by `submit`."""732 self.job = job733 734 def get_result(self, timeout):735 """Returns the raw result of the task that was submitted.736 737 If the task raised an exception rather than returning, this same738 exception will be raised instead.739 740 If the backend supports the retrieval callback, it is assumed that this741 method is only called after the result has been registered. It is742 ensured by checking that `self.status(timeout)` does not return743 TASK_PENDING. In this case, `get_result` directly returns the744 registered result (or raise the registered exception).745 746 For other backends, there are no such assumptions, but `get_result`747 still needs to synchronously retrieve the result before it can748 return it or raise. It will block at most `self.timeout` seconds749 waiting for retrieval to complete, after that it raises a TimeoutError.750 """751 752 backend = self.parallel._backend753 754 if backend.supports_retrieve_callback:755 # We assume that the result has already been retrieved by the756 # callback thread, and is stored internally. It's just waiting to757 # be returned.758 return self._return_or_raise()759 760 # For other backends, the main thread needs to run the retrieval step.761 try:762 result = backend.retrieve_result(self.job, timeout=timeout)763 outcome = dict(result=result, status=TASK_DONE)764 except BaseException as e:765 outcome = dict(result=e, status=TASK_ERROR)766 self._register_outcome(outcome)767 768 return self._return_or_raise()769 770 def _return_or_raise(self):771 try:772 if self.status == TASK_ERROR:773 raise self._result774 return self._result775 finally:776 del self._result777 778 def get_status(self, timeout):779 """Get the status of the task.780 781 This function also checks if the timeout has been reached and register782 the TimeoutError outcome when it is the case.783 """784 if timeout is None or self.status != TASK_PENDING:785 return self.status786 787 # The computation are running and the status is pending.788 # Check that we did not wait for this jobs more than `timeout`.789 now = time.time()790 if self._completion_timeout_counter is None:791 self._completion_timeout_counter = now792 793 if (now - self._completion_timeout_counter) > timeout:794 outcome = dict(result=TimeoutError(), status=TASK_ERROR)795 self._register_outcome(outcome)796 797 return self.status798 799 ##########################################################################800 # METHODS CALLED BY CALLBACK THREADS #801 ##########################################################################802 def __call__(self, *args, **kwargs):803 """Function called by the callback thread after a job is completed."""804 805 # If the backend doesn't support callback retrievals, the next batch of806 # tasks is dispatched regardless. The result will be retrieved by the807 # main thread when calling `get_result`.808 if not self.parallel._backend.supports_retrieve_callback:809 self._dispatch_new()810 return811 812 # If the backend supports retrieving the result in the callback, it813 # registers the task outcome (TASK_ERROR or TASK_DONE), and schedules814 # the next batch if needed.815 with self.parallel._lock:816 # Edge case where while the task was processing, the `parallel`817 # instance has been reset and a new call has been issued, but the818 # worker managed to complete the task and trigger this callback819 # call just before being aborted by the reset.820 if self.parallel._call_id != self.parallel_call_id:821 return822 823 # When aborting, stop as fast as possible and do not retrieve the824 # result as it won't be returned by the Parallel call.825 if self.parallel._aborting:826 return827 828 # Retrieves the result of the task in the main process and dispatch829 # a new batch if needed.830 job_succeeded = self._retrieve_result(*args, **kwargs)831 832 if job_succeeded:833 self._dispatch_new()834 835 def _dispatch_new(self):836 """Schedule the next batch of tasks to be processed."""837 838 # This steps ensure that auto-batching works as expected.839 this_batch_duration = time.time() - self.dispatch_timestamp840 self.parallel._backend.batch_completed(self.batch_size, this_batch_duration)841 842 # Schedule the next batch of tasks.843 with self.parallel._lock:844 self.parallel.n_completed_tasks += self.batch_size845 self.parallel.print_progress()846 if self.parallel._original_iterator is not None:847 self.parallel.dispatch_next()848 849 def _retrieve_result(self, out):850 """Fetch and register the outcome of a task.851 852 Return True if the task succeeded, False otherwise.853 This function is only called by backends that support retrieving854 the task result in the callback thread.855 """856 try:857 result = self.parallel._backend.retrieve_result_callback(out)858 outcome = dict(status=TASK_DONE, result=result)859 except BaseException as e:860 # Avoid keeping references to parallel in the error.861 e.__traceback__ = None862 outcome = dict(result=e, status=TASK_ERROR)863 864 self._register_outcome(outcome)865 return outcome["status"] != TASK_ERROR866 867 ##########################################################################868 # This method can be called either in the main thread #869 # or in the callback thread. #870 ##########################################################################871 def _register_outcome(self, outcome):872 """Register the outcome of a task.873 874 This method can be called only once, future calls will be ignored.875 """876 # Covers the edge case where the main thread tries to register a877 # `TimeoutError` while the callback thread tries to register a result878 # at the same time.879 with self.parallel._lock:880 if self.status not in (TASK_PENDING, None):881 return882 self.status = outcome["status"]883 884 self._result = outcome["result"]885 886 # Once the result and the status are extracted, the last reference to887 # the job can be deleted.888 self.job = None889 890 # As soon as an error as been spotted, early stopping flags are sent to891 # the `parallel` instance.892 if self.status == TASK_ERROR:893 self.parallel._exception = True894 self.parallel._aborting = True895 896 if self.parallel.return_ordered:897 return898 899 with self.parallel._lock:900 # For `return_as=generator_unordered`, append the job to the queue901 # in the order of completion instead of submission.902 self.parallel._jobs.append(self)903 904 905###############################################################################906def register_parallel_backend(name, factory, make_default=False):907 """Register a new Parallel backend factory.908 909 The new backend can then be selected by passing its name as the backend910 argument to the :class:`~Parallel` class. Moreover, the default backend can911 be overwritten globally by setting make_default=True.912 913 The factory can be any callable that takes no argument and return an914 instance of ``ParallelBackendBase``.915 916 Warning: this function is experimental and subject to change in a future917 version of joblib.918 919 .. versionadded:: 0.10920 """921 BACKENDS[name] = factory922 if make_default:923 global DEFAULT_BACKEND924 DEFAULT_BACKEND = name925 926 927def effective_n_jobs(n_jobs=-1):928 """Determine the number of jobs that can actually run in parallel929 930 n_jobs is the number of workers requested by the callers. Passing n_jobs=-1931 means requesting all available workers for instance matching the number of932 CPU cores on the worker host(s).933 934 This method should return a guesstimate of the number of workers that can935 actually perform work concurrently with the currently enabled default936 backend. The primary use case is to make it possible for the caller to know937 in how many chunks to slice the work.938 939 In general working on larger data chunks is more efficient (less scheduling940 overhead and better use of CPU cache prefetching heuristics) as long as all941 the workers have enough work to do.942 943 Warning: this function is experimental and subject to change in a future944 version of joblib.945 946 .. versionadded:: 0.10947 """948 if n_jobs == 1:949 return 1950 951 backend, backend_n_jobs = get_active_backend()952 if n_jobs is None:953 n_jobs = backend_n_jobs954 return backend.effective_n_jobs(n_jobs=n_jobs)955 956 957###############################################################################958class Parallel(Logger):959 """Helper class for readable parallel mapping.960 961 Read more in the :ref:`User Guide <parallel>`.962 963 Parameters964 ----------965 n_jobs: int, default=None966 The maximum number of concurrently running jobs, such as the number967 of Python worker processes when ``backend="loky"`` or the size of968 the thread-pool when ``backend="threading"``.969 This argument is converted to an integer, rounded below for float.970 If -1 is given, `joblib` tries to use all CPUs. The number of CPUs971 ``n_cpus`` is obtained with :func:`~cpu_count`.972 For n_jobs below -1, (n_cpus + 1 + n_jobs) are used. For instance,973 using ``n_jobs=-2`` will result in all CPUs but one being used.974 This argument can also go above ``n_cpus``, which will cause975 oversubscription. In some cases, slight oversubscription can be976 beneficial, e.g., for tasks with large I/O operations.977 If 1 is given, no parallel computing code is used at all, and the978 behavior amounts to a simple python `for` loop. This mode is not979 compatible with ``timeout``.980 None is a marker for 'unset' that will be interpreted as n_jobs=1981 unless the call is performed under a :func:`~parallel_config`982 context manager that sets another value for ``n_jobs``.983 If n_jobs = 0 then a ValueError is raised.984 backend: str, ParallelBackendBase instance or None, default='loky'985 Specify the parallelization backend implementation.986 Supported backends are:987 988 - "loky" used by default, can induce some989 communication and memory overhead when exchanging input and990 output data with the worker Python processes. On some rare991 systems (such as Pyiodide), the loky backend may not be992 available.993 - "multiprocessing" previous process-based backend based on994 `multiprocessing.Pool`. Less robust than `loky`.995 - "threading" is a very low-overhead backend but it suffers996 from the Python Global Interpreter Lock if the called function997 relies a lot on Python objects. "threading" is mostly useful998 when the execution bottleneck is a compiled extension that999 explicitly releases the GIL (for instance a Cython loop wrapped1000 in a "with nogil" block or an expensive call to a library such1001 as NumPy).1002 - finally, you can register backends by calling1003 :func:`~register_parallel_backend`. This will allow you to1004 implement a backend of your liking.1005 1006 It is not recommended to hard-code the backend name in a call to1007 :class:`~Parallel` in a library. Instead it is recommended to set1008 soft hints (prefer) or hard constraints (require) so as to make it1009 possible for library users to change the backend from the outside1010 using the :func:`~parallel_config` context manager.1011 return_as: str in {'list', 'generator', 'generator_unordered'}, default='list'1012 If 'list', calls to this instance will return a list, only when1013 all results have been processed and retrieved.1014 If 'generator', it will return a generator that yields the results1015 as soon as they are available, in the order the tasks have been1016 submitted with.1017 If 'generator_unordered', the generator will immediately yield1018 available results independently of the submission order. The output1019 order is not deterministic in this case because it depends on the1020 concurrency of the workers.1021 prefer: str in {'processes', 'threads'} or None, default=None1022 Soft hint to choose the default backend if no specific backend1023 was selected with the :func:`~parallel_config` context manager.1024 The default process-based backend is 'loky' and the default1025 thread-based backend is 'threading'. Ignored if the ``backend``1026 parameter is specified.1027 require: 'sharedmem' or None, default=None1028 Hard constraint to select the backend. If set to 'sharedmem',1029 the selected backend will be single-host and thread-based even1030 if the user asked for a non-thread based backend with1031 :func:`~joblib.parallel_config`.1032 verbose: int, default=01033 The verbosity level: if non zero, progress messages are1034 printed. Above 50, the output is sent to stdout.1035 The frequency of the messages increases with the verbosity level.1036 If it more than 10, all iterations are reported.1037 timeout: float or None, default=None1038 Timeout limit for each task to complete. If any task takes longer1039 a TimeOutError will be raised. Only applied when n_jobs != 11040 pre_dispatch: {'all', integer, or expression, as in '3*n_jobs'}, default='2*n_jobs'1041 The number of batches (of tasks) to be pre-dispatched.1042 Default is '2*n_jobs'. When batch_size="auto" this is reasonable1043 default and the workers should never starve. Note that only basic1044 arithmetic are allowed here and no modules can be used in this1045 expression.1046 batch_size: int or 'auto', default='auto'1047 The number of atomic tasks to dispatch at once to each1048 worker. When individual evaluations are very fast, dispatching1049 calls to workers can be slower than sequential computation because1050 of the overhead. Batching fast computations together can mitigate1051 this.1052 The ``'auto'`` strategy keeps track of the time it takes for a1053 batch to complete, and dynamically adjusts the batch size to keep1054 the time on the order of half a second, using a heuristic. The1055 initial batch size is 1.1056 ``batch_size="auto"`` with ``backend="threading"`` will dispatch1057 batches of a single task at a time as the threading backend has1058 very little overhead and using larger batch size has not proved to1059 bring any gain in that case.1060 temp_folder: str or None, default=None1061 Folder to be used by the pool for memmapping large arrays1062 for sharing memory with worker processes. If None, this will try in1063 order:1064 1065 - a folder pointed by the JOBLIB_TEMP_FOLDER environment1066 variable,1067 - /dev/shm if the folder exists and is writable: this is a1068 RAM disk filesystem available by default on modern Linux1069 distributions,1070 - the default system temporary folder that can be1071 overridden with TMP, TMPDIR or TEMP environment1072 variables, typically /tmp under Unix operating systems.1073 1074 Only active when ``backend="loky"`` or ``"multiprocessing"``.1075 max_nbytes int, str, or None, optional, default='1M'1076 Threshold on the size of arrays passed to the workers that1077 triggers automated memory mapping in temp_folder. Can be an int1078 in Bytes, or a human-readable string, e.g., '1M' for 1 megabyte.1079 Use None to disable memmapping of large arrays.1080 Only active when ``backend="loky"`` or ``"multiprocessing"``.1081 mmap_mode: {None, 'r+', 'r', 'w+', 'c'}, default='r'1082 Memmapping mode for numpy arrays passed to workers. None will1083 disable memmapping, other modes defined in the numpy.memmap doc:1084 https://numpy.org/doc/stable/reference/generated/numpy.memmap.html1085 Also, see 'max_nbytes' parameter documentation for more details.1086 backend_kwargs: dict, optional1087 Additional parameters to pass to the backend `configure` method.1088 1089 Notes1090 -----1091 1092 This object uses workers to compute in parallel the application of a1093 function to many different arguments. The main functionality it brings1094 in addition to using the raw multiprocessing or concurrent.futures API1095 are (see examples for details):1096 1097 * More readable code, in particular since it avoids1098 constructing list of arguments.1099 1100 * Easier debugging:1101 - informative tracebacks even when the error happens on1102 the client side1103 - using 'n_jobs=1' enables to turn off parallel computing1104 for debugging without changing the codepath1105 - early capture of pickling errors1106 1107 * An optional progress meter.1108 1109 * Interruption of multiprocesses jobs with 'Ctrl-C'1110 1111 * Flexible pickling control for the communication to and from1112 the worker processes.1113 1114 * Ability to use shared memory efficiently with worker1115 processes for large numpy-based datastructures.1116 1117 Note that the intended usage is to run one call at a time. Multiple1118 calls to the same Parallel object will result in a ``RuntimeError``1119 1120 Examples1121 --------1122 1123 A simple example:1124 1125 >>> from math import sqrt1126 >>> from joblib import Parallel, delayed1127 >>> Parallel(n_jobs=1)(delayed(sqrt)(i**2) for i in range(10))1128 [0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0]1129 1130 Reshaping the output when the function has several return1131 values:1132 1133 >>> from math import modf1134 >>> from joblib import Parallel, delayed1135 >>> r = Parallel(n_jobs=1)(delayed(modf)(i/2.) for i in range(10))1136 >>> res, i = zip(*r)1137 >>> res1138 (0.0, 0.5, 0.0, 0.5, 0.0, 0.5, 0.0, 0.5, 0.0, 0.5)1139 >>> i1140 (0.0, 0.0, 1.0, 1.0, 2.0, 2.0, 3.0, 3.0, 4.0, 4.0)1141 1142 The progress meter: the higher the value of `verbose`, the more1143 messages:1144 1145 >>> from time import sleep1146 >>> from joblib import Parallel, delayed1147 >>> r = Parallel(n_jobs=2, verbose=10)(1148 ... delayed(sleep)(.2) for _ in range(10)) #doctest: +SKIP1149 [Parallel(n_jobs=2)]: Done 1 tasks | elapsed: 0.6s1150 [Parallel(n_jobs=2)]: Done 4 tasks | elapsed: 0.8s1151 [Parallel(n_jobs=2)]: Done 10 out of 10 | elapsed: 1.4s finished1152 1153 Traceback example, note how the line of the error is indicated1154 as well as the values of the parameter passed to the function that1155 triggered the exception, even though the traceback happens in the1156 child process:1157 1158 >>> from heapq import nlargest1159 >>> from joblib import Parallel, delayed1160 >>> Parallel(n_jobs=2)(1161 ... delayed(nlargest)(2, n) for n in (range(4), 'abcde', 3))1162 ... # doctest: +SKIP1163 -----------------------------------------------------------------------1164 Sub-process traceback:1165 -----------------------------------------------------------------------1166 TypeError Mon Nov 12 11:37:46 20121167 PID: 12934 Python 2.7.3: /usr/bin/python1168 ........................................................................1169 /usr/lib/python2.7/heapq.pyc in nlargest(n=2, iterable=3, key=None)1170 419 if n >= size:1171 420 return sorted(iterable, key=key, reverse=True)[:n]1172 4211173 422 # When key is none, use simpler decoration1174 423 if key is None:1175 --> 424 it = izip(iterable, count(0,-1)) # decorate1176 425 result = _nlargest(n, it)1177 426 return map(itemgetter(0), result) # undecorate1178 4271179 428 # General case, slowest method1180 TypeError: izip argument #1 must support iteration1181 _______________________________________________________________________1182 1183 1184 Using pre_dispatch in a producer/consumer situation, where the1185 data is generated on the fly. Note how the producer is first1186 called 3 times before the parallel loop is initiated, and then1187 called to generate new data on the fly:1188 1189 >>> from math import sqrt1190 >>> from joblib import Parallel, delayed1191 >>> def producer():1192 ... for i in range(6):1193 ... print('Produced %s' % i)1194 ... yield i1195 >>> out = Parallel(n_jobs=2, verbose=100, pre_dispatch='1.5*n_jobs')(1196 ... delayed(sqrt)(i) for i in producer()) #doctest: +SKIP1197 Produced 01198 Produced 11199 Produced 21200 [Parallel(n_jobs=2)]: Done 1 jobs | elapsed: 0.0s