.PhS dZddlmZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl Z ddl Z ddl mZddlmZmZmZddlmZddlmZddlmZd d lmZd d lmZmZmZmZm Z m!Z!m"Z"d d l#m$Z$m%Z%d d l&m'Z'd dl(m)Z)m*Z*e"e!dZ+da,dZ-dZ.ddhZ/eee+d<d dl0m1Z1ee+d<da,dZ.e j2Z3dZ4de4iZ5e$de$de$de$de$de$de$de$ddZ6dZ7dZ8dZ9e6de6de6dfd Z:e6de6de6dfd!Z;Gd"d#Z<Gd$d%e<Z=dZ>e?ed&rBej@Ad'd(BpdZCeCejDeC)Z>Gd*d+eEZFd,ZGd-ZHd.ZId:d0ZJd1ZKd2ZLGd3d4eEZMd:d5ZNd;d7ZOGd8d9e)ZPdS)>>c""  r0r1r,cjt|||\}}ttd|d}||fS)!Return the active default backendr+)_get_active_backendr:r6)r0r1r,r*configr+s r%get_active_backendr@s: *&'7CCOGV 6x@&( S SF F?r;cttdt}ttd|d}t||d}t||d}t||d}|tvrt d|dt|t vrt d|d t |d kr|d krt d d }|ttd}d}|j }t|dd}t|dd}|d kr| p | o|dko| } | o|d ko|} | rjtt|} |dkr,|r*td| j j d|j j d|} d| d<| | fS| r1tt|} | |fS||fS)r=r?r*r0r1r,zprefer=z. is not a valid backend hint, expected one of zrequire=z4 is not a valid backend constraint, expected one of r2r4zJprefer == 'processes' and require == 'sharedmem' are inconsistent settingsTNr nesting_levelF uses_threadssupports_sharedmemr3 zUsing z as joblib backend instead of z8 as the latter does not provide shared memory semantics.r r+)getattr_backendr6r:VALID_BACKEND_HINTS ValueErrorVALID_BACKEND_CONSTRAINTSBACKENDSDEFAULT_BACKENDrCDEFAULT_THREAD_BACKENDprint __class____name__copyDEFAULT_PROCESS_BACKEND)r0r1r,backend_configr*explicit_backendrCrDrE force_threadsforce_processessharedmem_backend thread_configprocess_backends r%r>r>sXx1HIIN *NIGv~x @ @FCCGCCG ((( 5f 5 52 5 5   /// ;w ; ;8 ; ;   K!7!7 X   ?+!<<< )M7NE::L *>FF +F4F0F I9!4I\9I+*Uv/DUO0%%;<'    b==-= J*4=JJ-4->-GJJJ    '++-- "# h -//6##:;-XXX 3 3 5 555 N ""r;c eZdZdZedfededededededed d d d Zd ZdZdZdZ d S)parallel_configaSet the default backend or configuration for :class:`~joblib.Parallel`. This is an alternative to directly passing keyword arguments to the :class:`~joblib.Parallel` class constructor. It is particularly useful when calling into library code that uses joblib internally but does not expose the various parallel configuration arguments in its own API. Parameters ---------- backend: str or ParallelBackendBase instance, default=None If ``backend`` is a string it must match a previously registered implementation using the :func:`~register_parallel_backend` function. By default the following backends are available: - 'loky': single-host, process-based parallelism (used by default), - 'threading': single-host, thread-based parallelism, - 'multiprocessing': legacy single-host, process-based parallelism. 'loky' is recommended to run functions that manipulate Python objects. 'threading' is a low-overhead alternative that is most efficient for functions that release the Global Interpreter Lock: e.g. I/O-bound code or CPU-bound code in a few calls to native code that explicitly releases the GIL. Note that on some rare systems (such as pyodide), multiprocessing and loky may not be available, in which case joblib defaults to threading. In addition, if the ``dask`` and ``distributed`` Python packages are installed, it is possible to use the 'dask' backend for better scheduling of nested parallel calls without over-subscription and potentially distribute parallel calls over a networked cluster of several hosts. It is also possible to use the distributed 'ray' backend for distributing the workload to a cluster of nodes. See more details in the Examples section below. Alternatively the backend can be passed directly as an instance. n_jobs: int, default=None The maximum number of concurrently running jobs, such as the number of Python worker processes when ``backend="loky"`` or the size of the thread-pool when ``backend="threading"``. This argument is converted to an integer, rounded below for float. If -1 is given, `joblib` tries to use all CPUs. The number of CPUs ``n_cpus`` is obtained with :func:`~cpu_count`. For n_jobs below -1, (n_cpus + 1 + n_jobs) are used. For instance, using ``n_jobs=-2`` will result in all CPUs but one being used. This argument can also go above ``n_cpus``, which will cause oversubscription. In some cases, slight oversubscription can be beneficial, e.g., for tasks with large I/O operations. If 1 is given, no parallel computing code is used at all, and the behavior amounts to a simple python `for` loop. This mode is not compatible with `timeout`. None is a marker for 'unset' that will be interpreted as n_jobs=1 unless the call is performed under a :func:`~parallel_config` context manager that sets another value for ``n_jobs``. If n_jobs = 0 then a ValueError is raised. verbose: int, default=0 The verbosity level: if non zero, progress messages are printed. Above 50, the output is sent to stdout. The frequency of the messages increases with the verbosity level. If it more than 10, all iterations are reported. temp_folder: str or None, default=None Folder to be used by the pool for memmapping large arrays for sharing memory with worker processes. If None, this will try in order: - a folder pointed by the ``JOBLIB_TEMP_FOLDER`` environment variable, - ``/dev/shm`` if the folder exists and is writable: this is a RAM disk filesystem available by default on modern Linux distributions, - the default system temporary folder that can be overridden with ``TMP``, ``TMPDIR`` or ``TEMP`` environment variables, typically ``/tmp`` under Unix operating systems. max_nbytes: int, str, or None, optional, default='1M' Threshold on the size of arrays passed to the workers that triggers automated memory mapping in temp_folder. Can be an int in Bytes, or a human-readable string, e.g., '1M' for 1 megabyte. Use None to disable memmapping of large arrays. mmap_mode: {None, 'r+', 'r', 'w+', 'c'}, default='r' Memmapping mode for numpy arrays passed to workers. None will disable memmapping, other modes defined in the numpy.memmap doc: https://numpy.org/doc/stable/reference/generated/numpy.memmap.html Also, see 'max_nbytes' parameter documentation for more details. prefer: str in {'processes', 'threads'} or None, default=None Soft hint to choose the default backend. The default process-based backend is 'loky' and the default thread-based backend is 'threading'. Ignored if the ``backend`` parameter is specified. require: 'sharedmem' or None, default=None Hard constraint to select the backend. If set to 'sharedmem', the selected backend will be single-host and thread-based. inner_max_num_threads: int, default=None If not None, overwrites the limit set on the number of threads usable in some third-party library threadpools like OpenBLAS, MKL or OpenMP. This is only used with the ``loky`` backend. backend_params: dict Additional parameters to pass to the backend constructor when backend is a string. Notes ----- Joblib tries to limit the oversubscription by limiting the number of threads usable in some third-party library threadpools like OpenBLAS, MKL or OpenMP. The default limit in each worker is set to ``max(cpu_count() // effective_n_jobs, 1)`` but this limit can be overwritten with the ``inner_max_num_threads`` argument which will be used to set this limit in the child processes. .. versionadded:: 1.3 Examples -------- >>> from operator import neg >>> with parallel_config(backend='threading'): ... print(Parallel()(delayed(neg)(i + 1) for i in range(5))) ... [-1, -2, -3, -4, -5] To use the 'ray' joblib backend add the following lines: >>> from ray.util.joblib import register_ray # doctest: +SKIP >>> register_ray() # doctest: +SKIP >>> with parallel_config(backend="ray"): # doctest: +SKIP ... print(Parallel()(delayed(neg)(i + 1) for i in range(5))) [-1, -2, -3, -4, -5] r*r+r,r-r.r/r0r1N)r+r,r-r.r/r0r1inner_max_num_threadsc Xttdt|_|j|| fi| }||||||||d} |j|_|jd| Dttd|jdS)Nr?)r+r,r-r.r/r0r1r*cDi|]\}}t|t||S) isinstancer).0kvs r% z,parallel_config.__init__..s- Q Q Qda 1i8P8P QQ Q Q Qr;) rGrHr6old_parallel_config_check_backendrRr\updateitemssetattr) selfr*r+r,r-r.r/r0r1r]backend_params new_configs r%__init__zparallel_config.__init__ks$+8X?V#W#W %$%g/DWWWW&$"    $7<<>> ## Q Qj..00 Q Q Q    (D$899999r;c |tdur&|t|dkrtd|St|tr|t vr|t vrt |}|n|tvrCtj d|dtdtdt tt |<n9td |d tt t |d i|}n"t|dkrtd |'|jjd }|js J|||_|j-|jd}|tdurd}n|j}||_|S)Nr*rzrinner_max_num_threads and other constructor parameters backend_params are only supported when backend is not None.joblib backend '3' is not available on your system, falling back to . stacklevelInvalid backend: , expected one of zRConstructor parameters backend_params are only supported when backend is a string.z< does not accept setting the inner_max_num_threads argument.r`)r6lenrJrastrrLEXTERNAL_BACKENDSMAYBE_AVAILABLE_BACKENDSwarningswarnrM UserWarningsortedkeysrPrQsupports_inner_max_num_threadsr]rCrf)rkr*r]rlregisterr$parent_backendrCs r%rgzparallel_config._check_backends -i8 8 8$0C4G4G!4K4K 0 N gs # # h&&///09HHJJJJ 888MK7KK8GKKK##$  )1(AHW%%$5G55!(--//2255 w'99.99GG>""Q&& : ! ,$-222 9 > >3 > > >,AG )  (!5i@N!8!CCC ! . < $1G !r;c|jSN)r\rks r% __enter__zparallel_config.__enter__s ##r;c.|dSr) unregister)rktypevalue tracebacks r%__exit__zparallel_config.__exit__s r;c<ttd|jdS)Nr?)rjrHrfrs r%rzparallel_config.unregisters(D$<=====r;) rQ __module__ __qualname____doc__r6rnrgrrrr`r;r%r\r\sIIZ( 2":'x0' 2+M:*<8)+6&x0' 2"":":":":":H888t$$$>>>>>r;r\c,eZdZdZ dfd ZdZxZS)parallel_backenda Change the default backend used by Parallel inside a with block. .. warning:: It is advised to use the :class:`~joblib.parallel_config` context manager instead, which allows more fine-grained control over the backend configuration. If ``backend`` is a string it must match a previously registered implementation using the :func:`~register_parallel_backend` function. By default the following backends are available: - 'loky': single-host, process-based parallelism (used by default), - 'threading': single-host, thread-based parallelism, - 'multiprocessing': legacy single-host, process-based parallelism. 'loky' is recommended to run functions that manipulate Python objects. 'threading' is a low-overhead alternative that is most efficient for functions that release the Global Interpreter Lock: e.g. I/O-bound code or CPU-bound code in a few calls to native code that explicitly releases the GIL. Note that on some rare systems (such as Pyodide), multiprocessing and loky may not be available, in which case joblib defaults to threading. You can also use the `Dask `_ joblib backend to distribute work across machines. This works well with scikit-learn estimators with the ``n_jobs`` parameter, for example:: >>> import joblib # doctest: +SKIP >>> from sklearn.model_selection import GridSearchCV # doctest: +SKIP >>> from dask.distributed import Client, LocalCluster # doctest: +SKIP >>> # create a local Dask cluster >>> cluster = LocalCluster() # doctest: +SKIP >>> client = Client(cluster) # doctest: +SKIP >>> grid_search = GridSearchCV(estimator, param_grid, n_jobs=-1) ... # doctest: +SKIP >>> with joblib.parallel_backend("dask", scatter=[X, y]): # doctest: +SKIP ... grid_search.fit(X, y) It is also possible to use the distributed 'ray' backend for distributing the workload to a cluster of nodes. To use the 'ray' joblib backend add the following lines:: >>> from ray.util.joblib import register_ray # doctest: +SKIP >>> register_ray() # doctest: +SKIP >>> with parallel_backend("ray"): # doctest: +SKIP ... print(Parallel()(delayed(neg)(i + 1) for i in range(5))) [-1, -2, -3, -4, -5] Alternatively the backend can be passed directly as an instance. By default all available workers will be used (``n_jobs=-1``) unless the caller passes an explicit value for the ``n_jobs`` parameter. This is an alternative to passing a ``backend='backend_name'`` argument to the :class:`~Parallel` class constructor. It is particularly useful when calling into library code that uses joblib internally but does not expose the backend argument in its own API. >>> from operator import neg >>> with parallel_backend('threading'): ... print(Parallel()(delayed(neg)(i + 1) for i in range(5))) ... [-1, -2, -3, -4, -5] Joblib also tries to limit the oversubscription by limiting the number of threads usable in some third-party library threadpools like OpenBLAS, MKL or OpenMP. The default limit in each worker is set to ``max(cpu_count() // effective_n_jobs, 1)`` but this limit can be overwritten with the ``inner_max_num_threads`` argument which will be used to set this limit in the child processes. .. versionadded:: 0.10 See Also -------- joblib.parallel_config: context manager to change the backend configuration. Nc tjd|||d||jd|_n|jd|jdf|_|jd|jdf|_dS)N)r*r+r]r*r+r`)superrnrfold_backend_and_jobsr\new_backend_and_jobs)rkr*r+r]rlrPs r%rnzparallel_backend.__init__$s  "7        # +(,D % %(3(2)D %   +   *% !!!r;c|jSr)rrs r%rzparallel_backend.__enter__:s ((r;)rN)rQrrrrnr __classcell__rPs@r%rrs^NNb9=      ,)))))))r;r get_contextJOBLIB_START_METHOD)methodc.eZdZdZ ddZdZdZdZdS) BatchedCallszCWrap a sequence of (func, args, kwargs) tuples as a single callableNct||_t|j|_||_t |t r|\|_|_n|dc|_|_||ni|_ dSr) listrirx_size_reducer_callbackratuplerH_n_jobs _pickle_cache)rkiterator_slicebackend_and_jobsreducer_callback pickle_caches r%rnzBatchedCalls.__init__Msw.)) __ !1 & . . A*: 'DM4<<+;D 'DM4<-9-E\\2r;ct|j|j5d|jDcdddS#1swxYwYdS)N)r*r+c&g|]\}}}||i|Sr`r`)rbfuncargskwargss r% z)BatchedCalls.__call__.._s.OOO.@dD&DD$)&))OOOr;)r\rHrrirs r%__call__zBatchedCalls.__call__[sT]4< H H H P POODJOOO P P P P P P P P P P P P P P P P P Ps :>>c~|j|t|j|j|jfd|jffSr)rrrirHrrrs r% __reduce__zBatchedCalls.__reduce__asF  ! -  " " $ $ $  Z$-6d>P Q  r;c|jSr)rrs r%__len__zBatchedCalls.__len__js zr;)NN)rQrrrrnrrrr`r;r%rrJsiMMUY N N N NPPP    r;rDoneErrorPendingFc>tdStj|S)aReturn the number of CPUs. This delegates to loky.cpu_count that takes into account additional constraints such as Linux CFS scheduler quotas (typically set by container runtimes such as docker) and CPU affinity (for instance using the taskset command on Linux). Parameters ---------- only_physical_cores : boolean, default=False If True, does not take hyperthreading / SMT logical cores into account. Nr only_physical_cores)r r cpu_countrs r%rrws" zq >.A B B BBr;c|sdS|dkrdS|dkrdSdd|z dzz}t||z }t|dz|z }t|t|kS) zReturns False for indices increasingly apart, the distance depending on the value of verbose. We use a lag increasing as the square of index TrFFrg? rsr )rint)indexr,scale next_scales r%_verbosity_filterrs~ t 2u zzuR'\a''G  ! !EuqyG+,,J z??c%jj ((r;cpfd} tj|}n#t$rYnwxYw|S)z6Decorator used to capture the arguments of a function.c||fSrr`)rrfunctions r%delayed_functionz!delayed..delayed_functionsv%%r;) functoolswrapsAttributeError)rrs` r%delayedrsf&&&&&;49?8445EFF ;;;::; s & 33cHeZdZdZdZdZdZdZdZdZ dZ d Z d Z d S) BatchCompletionCallBackaCallback to keep track of completed results and schedule the next tasks. This callable is executed by the parent process whenever a worker process has completed a batch of tasks. It is used for progress reporting, to update estimate of the batch processing duration and to schedule the next batch of tasks to be processed. It is assumed that this callback will always be triggered by the backend right after the end of a task, in case of success as well as in case of failure. c||_||_||_|j|_d|_d|_|jjs d|_ dSt|_ dSr) dispatch_timestamp batch_sizeparallel_call_idparallel_call_id_completion_timeout_counterjobrHsupports_retrieve_callbackstatus TASK_PENDING)rkrrrs r%rnz BatchCompletionCallBack.__init__s\"4$  ( 1+/( ; 'DKKK'DKKKr;c||_dS)z)Register the object returned by `submit`.N)r)rkrs r% register_jobz$BatchCompletionCallBack.register_jobs r;cd|jj}|jr|S ||j|}t |t}n-#t$r }t |t}Yd}~nd}~wwxYw| ||S)aHReturns the raw result of the task that was submitted. If the task raised an exception rather than returning, this same exception will be raised instead. If the backend supports the retrieval callback, it is assumed that this method is only called after the result has been registered. It is ensured by checking that `self.status(timeout)` does not return TASK_PENDING. In this case, `get_result` directly returns the registered result (or raise the registered exception). For other backends, there are no such assumptions, but `get_result` still needs to synchronously retrieve the result before it can return it or raise. It will block at most `self.timeout` seconds waiting for retrieval to complete, after that it raises a TimeoutError. timeoutresultrN) rrHr_return_or_raiseretrieve_resultrdict TASK_DONE BaseException TASK_ERROR_register_outcome)rkrr*routcomer#s r% get_resultz"BatchCompletionCallBack.get_results$-(  - +((** * 8,,TXw,GGF&;;;GG 8 8 8!J777GGGGGG 8 w'''$$&&&s2A B&BBcR |jtkr|j|j|`S#|`wxYwr)rr_resultrs r%rz(BatchCompletionCallBack._return_or_raises9 {j((l"<     s"&c||jtkr|jStj}|j||_||jz |kr7t t t }|||jS)zGet the status of the task. This function also checks if the timeout has been reached and register the TimeoutError outcome when it is the case. Nr)rrtimerrrrr)rkrnowrs r% get_statusz"BatchCompletionCallBack.get_status s ?dk\99; ikk  + 3/2D , $2 2g = =,..DDDG  " "7 + + +{r;c`|jjjs|dS|jj5|jj|jkr ddddS|jjr ddddS|j|i|}dddn #1swxYwY|r|dSdS)z@Function called by the callback thread after a job is completed.N) rrHr _dispatch_new_lockrr _aborting_retrieve_result)rkrr job_succeededs r%rz BatchCompletionCallBack.__call__"sm }%@     F ]  C C }%)>>>  C C C C C C C C}&  C C C C C C C C2D14B6BBM C C C C C C C C C C C C C C C"  !     ! !sB  B 2 B  BBctj|jz }|jj|j||jj5|jxj|jz c_|j|jj |j ddddS#1swxYwYdS)z1Schedule the next batch of tasks to be processed.N) rrrrHbatch_completedrrn_completed_tasksprint_progress_original_iterator dispatch_next)rkthis_batch_durations r%rz%BatchCompletionCallBack._dispatch_newCs#ikkD,CC ..t@STTT]  . . M + +t > + + M ( ( * * *}/; ++---  . . . . . . . . . . . . . . . . . .s AB33B7:B7c$ |jj|}tt|}n4#t $r'}d|_t|t}Yd}~nd}~wwxYw|||dtkS)zFetch and register the outcome of a task. Return True if the task succeeded, False otherwise. This function is only called by backends that support retrieving the task result in the callback thread. )rrNrr) rrHretrieve_result_callbackrrr __traceback__rr)rkoutrrr#s r%rz(BatchCompletionCallBack._retrieve_resultQs 8]+DDSIIF)F;;;GG 8 8 8"AO!J777GGGGGG 8 w'''x J..s58 A)A$$A)c|jj5|jtdfvr ddddS|d|_dddn #1swxYwY|d|_d|_|jt krd|j_d|j_|jj rdS|jj5|jj |ddddS#1swxYwYdS)ztRegister the outcome of a task. This method can be called only once, future calls will be ignored. NrrT) rrrrrrr _exceptionrreturn_ordered_jobsappend)rkrs r%rz)BatchCompletionCallBack._register_outcomegs]  , ,{<"666 , , , , , , , ,"(+DK , , , , , , , , , , , , , , , x(  ;* $ $'+DM $&*DM # = '  F ]  - - M  & &t , , , - - - - - - - - - - - - - - - - - -s(A AA  A & CCCN) rQrrrrnrrrrrrrrr`r;r%rrs  "'''0"'"'"'H0!!!B . . .///,-----r;rc&|t|<|r|adSdS)aRegister a new Parallel backend factory. The new backend can then be selected by passing its name as the backend argument to the :class:`~Parallel` class. Moreover, the default backend can be overwritten globally by setting make_default=True. The factory can be any callable that takes no argument and return an instance of ``ParallelBackendBase``. Warning: this function is experimental and subject to change in a future version of joblib. .. versionadded:: 0.10 N)rLrM)namefactory make_defaults r%r!r!s(HTNr;rch|dkrdSt\}}||}||S)aeDetermine the number of jobs that can actually run in parallel n_jobs is the number of workers requested by the callers. Passing n_jobs=-1 means requesting all available workers for instance matching the number of CPU cores on the worker host(s). This method should return a guesstimate of the number of workers that can actually perform work concurrently with the currently enabled default backend. The primary use case is to make it possible for the caller to know in how many chunks to slice the work. In general working on larger data chunks is more efficient (less scheduling overhead and better use of CPU cache prefetching heuristics) as long as all the workers have enough work to do. Warning: this function is experimental and subject to change in a future version of joblib. .. versionadded:: 0.10 r N)r+)r@effective_n_jobs)r+r*backend_n_jobss r%r r sC*{{q022G^ ~  # #6 # 2 22r;c ,eZdZdZededdedddded ed ed ed ed f fd ZdZdZdZdZ dZ dZ dZ dZ dZdZdZdZdZdZdZdZdZd Zd!Zd"Zd#Zd$Zd%Zd&ZxZS)'Parallela/Helper class for readable parallel mapping. Read more in the :ref:`User Guide `. Parameters ---------- n_jobs: int, default=None The maximum number of concurrently running jobs, such as the number of Python worker processes when ``backend="loky"`` or the size of the thread-pool when ``backend="threading"``. This argument is converted to an integer, rounded below for float. If -1 is given, `joblib` tries to use all CPUs. The number of CPUs ``n_cpus`` is obtained with :func:`~cpu_count`. For n_jobs below -1, (n_cpus + 1 + n_jobs) are used. For instance, using ``n_jobs=-2`` will result in all CPUs but one being used. This argument can also go above ``n_cpus``, which will cause oversubscription. In some cases, slight oversubscription can be beneficial, e.g., for tasks with large I/O operations. If 1 is given, no parallel computing code is used at all, and the behavior amounts to a simple python `for` loop. This mode is not compatible with ``timeout``. None is a marker for 'unset' that will be interpreted as n_jobs=1 unless the call is performed under a :func:`~parallel_config` context manager that sets another value for ``n_jobs``. If n_jobs = 0 then a ValueError is raised. backend: str, ParallelBackendBase instance or None, default='loky' Specify the parallelization backend implementation. Supported backends are: - "loky" used by default, can induce some communication and memory overhead when exchanging input and output data with the worker Python processes. On some rare systems (such as Pyiodide), the loky backend may not be available. - "multiprocessing" previous process-based backend based on `multiprocessing.Pool`. Less robust than `loky`. - "threading" is a very low-overhead backend but it suffers from the Python Global Interpreter Lock if the called function relies a lot on Python objects. "threading" is mostly useful when the execution bottleneck is a compiled extension that explicitly releases the GIL (for instance a Cython loop wrapped in a "with nogil" block or an expensive call to a library such as NumPy). - finally, you can register backends by calling :func:`~register_parallel_backend`. This will allow you to implement a backend of your liking. It is not recommended to hard-code the backend name in a call to :class:`~Parallel` in a library. Instead it is recommended to set soft hints (prefer) or hard constraints (require) so as to make it possible for library users to change the backend from the outside using the :func:`~parallel_config` context manager. return_as: str in {'list', 'generator', 'generator_unordered'}, default='list' If 'list', calls to this instance will return a list, only when all results have been processed and retrieved. If 'generator', it will return a generator that yields the results as soon as they are available, in the order the tasks have been submitted with. If 'generator_unordered', the generator will immediately yield available results independently of the submission order. The output order is not deterministic in this case because it depends on the concurrency of the workers. prefer: str in {'processes', 'threads'} or None, default=None Soft hint to choose the default backend if no specific backend was selected with the :func:`~parallel_config` context manager. The default process-based backend is 'loky' and the default thread-based backend is 'threading'. Ignored if the ``backend`` parameter is specified. require: 'sharedmem' or None, default=None Hard constraint to select the backend. If set to 'sharedmem', the selected backend will be single-host and thread-based even if the user asked for a non-thread based backend with :func:`~joblib.parallel_config`. verbose: int, default=0 The verbosity level: if non zero, progress messages are printed. Above 50, the output is sent to stdout. The frequency of the messages increases with the verbosity level. If it more than 10, all iterations are reported. timeout: float or None, default=None Timeout limit for each task to complete. If any task takes longer a TimeOutError will be raised. Only applied when n_jobs != 1 pre_dispatch: {'all', integer, or expression, as in '3*n_jobs'}, default='2*n_jobs' The number of batches (of tasks) to be pre-dispatched. Default is '2*n_jobs'. When batch_size="auto" this is reasonable default and the workers should never starve. Note that only basic arithmetic are allowed here and no modules can be used in this expression. batch_size: int or 'auto', default='auto' The number of atomic tasks to dispatch at once to each worker. When individual evaluations are very fast, dispatching calls to workers can be slower than sequential computation because of the overhead. Batching fast computations together can mitigate this. The ``'auto'`` strategy keeps track of the time it takes for a batch to complete, and dynamically adjusts the batch size to keep the time on the order of half a second, using a heuristic. The initial batch size is 1. ``batch_size="auto"`` with ``backend="threading"`` will dispatch batches of a single task at a time as the threading backend has very little overhead and using larger batch size has not proved to bring any gain in that case. temp_folder: str or None, default=None Folder to be used by the pool for memmapping large arrays for sharing memory with worker processes. If None, this will try in order: - a folder pointed by the JOBLIB_TEMP_FOLDER environment variable, - /dev/shm if the folder exists and is writable: this is a RAM disk filesystem available by default on modern Linux distributions, - the default system temporary folder that can be overridden with TMP, TMPDIR or TEMP environment variables, typically /tmp under Unix operating systems. Only active when ``backend="loky"`` or ``"multiprocessing"``. max_nbytes int, str, or None, optional, default='1M' Threshold on the size of arrays passed to the workers that triggers automated memory mapping in temp_folder. Can be an int in Bytes, or a human-readable string, e.g., '1M' for 1 megabyte. Use None to disable memmapping of large arrays. Only active when ``backend="loky"`` or ``"multiprocessing"``. mmap_mode: {None, 'r+', 'r', 'w+', 'c'}, default='r' Memmapping mode for numpy arrays passed to workers. None will disable memmapping, other modes defined in the numpy.memmap doc: https://numpy.org/doc/stable/reference/generated/numpy.memmap.html Also, see 'max_nbytes' parameter documentation for more details. backend_kwargs: dict, optional Additional parameters to pass to the backend `configure` method. Notes ----- This object uses workers to compute in parallel the application of a function to many different arguments. The main functionality it brings in addition to using the raw multiprocessing or concurrent.futures API are (see examples for details): * More readable code, in particular since it avoids constructing list of arguments. * Easier debugging: - informative tracebacks even when the error happens on the client side - using 'n_jobs=1' enables to turn off parallel computing for debugging without changing the codepath - early capture of pickling errors * An optional progress meter. * Interruption of multiprocesses jobs with 'Ctrl-C' * Flexible pickling control for the communication to and from the worker processes. * Ability to use shared memory efficiently with worker processes for large numpy-based datastructures. Note that the intended usage is to run one call at a time. Multiple calls to the same Parallel object will result in a ``RuntimeError`` Examples -------- A simple example: >>> from math import sqrt >>> from joblib import Parallel, delayed >>> Parallel(n_jobs=1)(delayed(sqrt)(i**2) for i in range(10)) [0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0] Reshaping the output when the function has several return values: >>> from math import modf >>> from joblib import Parallel, delayed >>> r = Parallel(n_jobs=1)(delayed(modf)(i/2.) for i in range(10)) >>> res, i = zip(*r) >>> res (0.0, 0.5, 0.0, 0.5, 0.0, 0.5, 0.0, 0.5, 0.0, 0.5) >>> i (0.0, 0.0, 1.0, 1.0, 2.0, 2.0, 3.0, 3.0, 4.0, 4.0) The progress meter: the higher the value of `verbose`, the more messages: >>> from time import sleep >>> from joblib import Parallel, delayed >>> r = Parallel(n_jobs=2, verbose=10)( ... delayed(sleep)(.2) for _ in range(10)) #doctest: +SKIP [Parallel(n_jobs=2)]: Done 1 tasks | elapsed: 0.6s [Parallel(n_jobs=2)]: Done 4 tasks | elapsed: 0.8s [Parallel(n_jobs=2)]: Done 10 out of 10 | elapsed: 1.4s finished Traceback example, note how the line of the error is indicated as well as the values of the parameter passed to the function that triggered the exception, even though the traceback happens in the child process: >>> from heapq import nlargest >>> from joblib import Parallel, delayed >>> Parallel(n_jobs=2)( ... delayed(nlargest)(2, n) for n in (range(4), 'abcde', 3)) ... # doctest: +SKIP ----------------------------------------------------------------------- Sub-process traceback: ----------------------------------------------------------------------- TypeError Mon Nov 12 11:37:46 2012 PID: 12934 Python 2.7.3: /usr/bin/python ........................................................................ /usr/lib/python2.7/heapq.pyc in nlargest(n=2, iterable=3, key=None) 419 if n >= size: 420 return sorted(iterable, key=key, reverse=True)[:n] 421 422 # When key is none, use simpler decoration 423 if key is None: --> 424 it = izip(iterable, count(0,-1)) # decorate 425 result = _nlargest(n, it) 426 return map(itemgetter(0), result) # undecorate 427 428 # General case, slowest method TypeError: izip argument #1 must support iteration _______________________________________________________________________ Using pre_dispatch in a producer/consumer situation, where the data is generated on the fly. Note how the producer is first called 3 times before the parallel loop is initiated, and then called to generate new data on the fly: >>> from math import sqrt >>> from joblib import Parallel, delayed >>> def producer(): ... for i in range(6): ... print('Produced %s' % i) ... yield i >>> out = Parallel(n_jobs=2, verbose=100, pre_dispatch='1.5*n_jobs')( ... delayed(sqrt)(i) for i in producer()) #doctest: +SKIP Produced 0 Produced 1 Produced 2 [Parallel(n_jobs=2)]: Done 1 jobs | elapsed: 0.0s Produced 3 [Parallel(n_jobs=2)]: Done 2 jobs | elapsed: 0.0s Produced 4 [Parallel(n_jobs=2)]: Done 3 jobs | elapsed: 0.0s Produced 5 [Parallel(n_jobs=2)]: Done 4 jobs | elapsed: 0.0s [Parallel(n_jobs=2)]: Done 6 out of 6 | elapsed: 0.0s remaining: 0.0s [Parallel(n_jobs=2)]: Done 6 out of 6 | elapsed: 0.0s finished r+r*rr,Nz 2 * n_jobsautor-r.r/r0r1c t| td}t| | |\}|j}t |d|_||_||_|dvrtd|d||_ |dk|_ |dk|_ i| fd | d f|d f| d f| d f| df|dffD|_ t|j d tr"t!|j d |j d <t#d|j ddz |j d<t$t$|j d<n0t't(drt)j|j d<|tdus||}n7t|t,r|j||_nt'|dr+t'|dr||j d<t/|}n|t0vrg|t2vr^t5jd|dt8dt:dt0t8t0|<t1t8|}ng t0|}nL#t<$r?}td|dt?t0 |d}~wwxYw||}t |d}||j!} tE|}n#t$rtdwxYw||_#| dkr#tI|d d!std"|z|d#kst|tJr|dkr||_&ntd$|zt|tNs|j r*|j(s#td%)||tUj+|_,t[j.|_/ta|_1te|_3tij5|_6d|_7||_8d!|_9d!|_:twj<|_=d|_>dS)&Nr+)r0r1r,r,>r generatorgenerator_unorderedzlExpected `return_as` parameter to be a string equal to "list","generator" or "generator_unordered", but got z instead.rrc:i|]\}}|t||Sr`)r:)rbr7rcr8s r%rez%Parallel.__init__..s<   E1$UNA>>   r;r.r-r/r0r1r2contextrr*PoolLockrBrprqrrrsrtrvrwz$n_jobs could not be converted to intr4rEFz)Backend %s does not support shared memoryrz8batch_size must be 'auto' or a positive integer, got: %rz(Backend {} does not support return_as={})?rrnr6r>rCr:r,r pre_dispatchrJ return_asreturn_generatorr_backend_kwargsraryrmaxDEFAULT_MP_CONTEXThasattrr rrrrLr{r|r}rMr~KeyErrorrrdefault_n_jobsrr+rGr rrsupports_return_generatorformatrRLockr collectionsdequerset _jobs_setr_pending_outputsqueueQueue_ready_batchesrrH_running_managed_backendr hex_id _call_ref)rkr+r*rr,rrrr-r.r/r0r1backend_kwargsactive_backendrCbackend_factoryr#r8rPs @r%rnzParallel.__init__s%"  >,X6F)<7G* * * &'4 (.)LL  ( H H HBK  # )V 3'+@@       . -0 ,X&i(i( !     d*<8# > > 1@$\222D  .+.a1Ei1PSU1U*V*VY'  ).@D  + + R ' ' ?.0n.>.>D  + -i8 8 8GO$GG !4 5 5 C$,(5% Wf % % C''6*B*B C/6D  +,=IIIGG H $ $4L)L)L MC7CC0?CCC      !) 9HW /mLLLGG "*7"3    jwwx}} 7 7 79  &oMBBBG"6>8DD >+F E[[FF E E ECDD D E k ! !'';OQV*W*W !H7RSS S   :j(#C#C  UV(DOOJZW '#455 *$ W-N  >EE#**DJ$*,,DJ UUDN$(FFD !"'+--D %)D "   %77;s$$ I22 J;<:J66J;%K55LcJd|_d|_||S)NTF)r._calling_initialize_backendrs r%rzParallel.__enter__Ts( $    """ r;cd|_|jr|jr||dS)NF)r.rr6_abort_terminate_and_reset)rkexc_type exc_valuers r%rzParallel.__exit__ZsC %   T]  KKMMM !!#####r;cT |jjd|j|d|j}|jH|jjs> >qr;ct|jdr |jr|jd|_|js|jdSdS)N stop_callF)rrHr6rCr. terminaters r%r:zParallel._terminate_and_reset{si 4=+ . . &4= & M # # % % % $ & M # # % % % % % & &r;cJ|jrdSt|}|xj|z c_|xjdz c_t j}t |||}|||j||}| |dS)zQueue the batch for computing, with or without multiprocessing WARNING: this method is not thread-safe: it should be only called indirectly via dispatch_one_batch. Nr )callback) rrxn_dispatched_tasksn_dispatched_batchesrr_register_new_jobrHsubmitr)rkbatchrr batch_trackerrs r% _dispatchzParallel._dispatchs >  FZZ  :- !!Q&!!!Y[[/0BJPTUU  }---m""5="AA""3'''''r;c|jr|j|dS|j|dSr)rrrr(add)rkrLs r%rIzParallel._register_new_jobsE   . J  m , , , , , N  } - - - - -r;cZ||jsd|_d|_dSdS)aDispatch more data for parallel processing This method is meant to be called concurrently by the multiprocessing callback. We rely on the thread-safety of dispatch_one_batch to protect against concurrent consumption of the unprotected iterator. FN)dispatch_one_batchr _iteratingrs r%rzParallel.dispatch_nexts<&&t'>?? +#DO&*D # # # + +r;c |jrdS|}|j5 |jd}n#t j$r|j}||z} ttj ||}n#t$r}t|j t jrd|_td||}|||t%|t&Yd}~YddddSd}~wwxYwt)|dkrYddddS||jur7t)||kr$t-dt)|d|zz} n t-dt)||z} t/dt)|| D]Z} t1|| | | z|j|j|j}|j|[|jd}YnwxYwt)|dkr ddddS|| ddddS#1swxYwYdS) aTPrefetch the tasks for the next batch and dispatch them. The effective size of the batch is computed here. If there are no more jobs to dispatch, return False, else return True. The iterator consumption and dispatching is protected by the same lock so calling this function should be thread safe. F)blockNrrTr rF)r_get_batch_sizerr,getr*Empty_cached_effective_n_jobsr itertoolsislice Exceptionra __context__ __cause__rrIrrrrxrrrangerrHget_nested_backendrrputrM) rkiteratorrtasksr+big_batch_sizerZr#rLfinal_batch_sizeis r%rQzParallel.dispatch_one_batchs} > 5))++ ZD D 7 =+//e/<<;5 =5 =5 =6!+f!4 !)"28^"L"LMMFF    "!-==+ '+ $;Az4$P$PM**=999!33D*4U4U4UVVV4444GD D D D D D D D * v;;!## MD D D D D D D D P 777CKK.C4H52I<4H55I<I<<JJcX|jdkr|jS|jS)z-Returns the effective batch size for dispatchr)rrHcompute_batch_sizers r%rUzParallel._get_batch_sizes- ?f $ $=3355 5? "r;c|jsdS|jdkrtjj}ntjj}|d|d|ddS)z=Display the message on stout or stderr depending on verbosityNr[z]:  )r,sysstderrwritestdout)rkr$writers r%_printzParallel._prints`|  F <"  Z%FFZ%F#4##C###$$$$$r;c@|j|jko|jp|j S)z&Check if all tasks have been completed)rrGrRrrs r% _is_completedzParallel._is_completeds,%)@@ O -t~F  r;c  |jsdS|j0|jdkr%tt|jdz}nd}t j|jz }|r@|d|j|dd|j|ddt|d dS|j t|j |jrdSd t|}|j}|j-|d||dd|j|dd |dSd td |ztd z z}|d||dd||dS|j}|j}|dkr/||z dz|jz }||jzdz}|dz|k} | s||zrdS|t!|dz |j|z z} |d||dd||ddt|dt| dS)zsDisplay the process of the parallel execution only a fraction of time, controlled by self.verbose. Nrr zDone dz out of z | elapsed: z finishedz | elapsed:  zout of rbz tasks z remaining: )r,n_tasksrrr _start_timerrrprrrrrHrxrG_pre_dispatch_amountr) rkwidth elapsed_timefmt_timerpad total_taskscursor frequency is_last_itemremaining_times r%rzParallel.print_progress!s |  F < # q(8(8% --..2EEEy{{T%55     /  KK>.>AA>>)5>@@>>$\22>>>    F $ 0 !:DLII F%6|%D%DFFH*E|' VE5VVVVVDL%VVVVVHVVS^^e3c'llBC JE5JJJJJ3JJJKKKKK*E1Kzz%u,q043LL(DL8A= $qyK7 6I#5F*S]]:'%/N KK7u7GG77 U7GG77/ ==77$^4477     r;cd|_|j}|js-t|dr|j}||d|_dS)NTabort_everything) ensure_ready)rrH_abortedrr.r)rkr*rs r%r9zParallel._abort`s\ -} @2D!E!E @ 0L  $ $, $ ? ? ? r;cd|_||r|jdu|_||r |||dkr d|_dSdS)NFall)rRrQr)rkrars r%_startzParallel._startqs   " "8 , , B"5TADO%%h//  %%h//  5 $DOOO ! r;c# Ktj}d} |||dV|j5|Ed{Vdddn #1swxYwYn.#t $rd|_|tjkrtj dd}| G fddtj }|d Y|jrgn|j }tj|_ t|_d|_|s|dSdS||jr|t,$rd|_|wxYw|jrgn|j }tj|_ t|_d|_|s|n^#|jrgn|j }tj|_ t|_d|_|s|wwxYwt/|d krL|}||j}|D]}|Vt/|d kJdSdS) z?Iterator returning the tasks' output as soon as they are ready.FNTaA generator produced by joblib.Parallel has been gc'ed in an unexpected thread. This behavior should not cause major -issues but to make sure, please report this warning and your use case at https://github.com/joblib/joblib/issues so it can be investigated.ceZdZfdZdS)3Parallel._get_outputs.._GeneratorExitThreadcjrdSr)r9r_warn_exit_earlyr:)rk _parallels r%runz7Parallel._get_outputs.._GeneratorExitThread.runsJ!((***$59%66888!6688888r;N)rQrrr)rsr%_GeneratorExitThreadrs.9999999r;rGeneratorExitThread)rr)r get_identrrHretrieval_context _retrieve GeneratorExitrr|r}Threadstartrr%r&r'r(r-r:r9rrrrxpopleftrr) rkrardispatch_thread_iddetach_generator_exitr_remaining_outputsbatched_resultsrrs @r% _get_outputszParallel._get_outputssn&022 %C , KK, / / / EEE0022 , ,>>+++++++++ , , , , , , , , , , , , , , ,* * * #DO"Y%8%:%::: ')-% 99999999+;999%$*?@@@FFHHH$(,!FDJ $*,,DJ UUDN!DM( ,))+++++ , ,% KKMMM$ (%%'''    "DO KKMMM   (,!FDJ $*,,DJ UUDN!DM( ,))+++ (,!FDJ $*,,DJ UUDN!DM( ,))++++ ,$%%))088::O-88FFO)   $%%))))))sP3B A4( B4A88B;A8<B?HA8F,9HAF,,HAI#c|jrdS|j|jkrdS|jjst |jdkrdSdS)z9Return True if we need to continue retrieving some tasks.TrF)rRrrGrHrrxrrs r%_wait_retrievalzParallel._wait_retrievalsW ? 4  !D$; ; ;4}7 4:""tur;c#RKd}|r|jr|dSt|j}|jrK|dks/|jd|jtkrtj dng|dkrV|"tt|j d}|||jtj d| d|_d}|j5|j}|js|j |dddn #1swxYwY||j}|D]}|xjdz c_|V|dSdS)Nrrg{Gz?r )rr_raise_error_fastrxrrrrrrsleepnextiterr(rrrremover _nb_consumed)rktimeout_control_jobnb_jobsrrs r%rzParallel._retrieves&"""$$B  ~ &&((($*ooG") +qLLJqM,,T\,BBlRRJt$$$S A'.*.tDN/C/CT*J*J''2'224<2HHH 4   $0 CG#?&*#  ; ;"&*"4"4"6"6*;N))/::: ; ; ; ; ; ; ; ; ; ; ; ; ; ; ; .88FFO)  !!Q&!! E""$$B B B B B s ;EEEc|j5td|jDd}dddn #1swxYwY|||jdSdS)z3If we are aborting, raise if a job caused an error.c3:K|]}|jtk|VdSr)rr)rbrs r% z-Parallel._raise_error_fast..<s.GGcjJ.F.F.F.F.F.FGGr;N)rrrrr)rk error_jobs r%rzParallel._raise_error_fast6sZ  GG GGGI                   . . . . . ! s 488c|j|jz }|}d}|r||dz }|s|dz }|s||j|jz dz }|r|dz }t j|dSdS)z?Warn the user if the generator is gc'ed before being consumned.rz5 tasks have been successfully executed but not used.z Additionally, zK tasks which were still being processed by the workers have been cancelled.z` You could benefit from adjusting the input task iterator to limit unnecessary computation time.N)rrrrrGr|r})rk ready_outputs is_completedr$s r%rzParallel._warn_exit_earlyEs.1BB ))++   )  WWW C  )((  *T-CC C   B C M#        r;c#K d|_||_|dkr/t|tfdd}d|D}dV|D]f\}}}|xjdz c_|xjdz c_||i|}|xjdz c_||V|xjdz c_gn$#t$rd|_ d|_ d|_ wxYw d|_ d|_d|_|dS#d|_ d|_d|_|wxYw)zSeparate loop for sequential output. This simplifies the traceback in case of errors and reduces the overhead of calling sequential tasks with `joblib`. Tr cHttjSr)rrYrZ)ritsr%z1Parallel._get_sequential_output..nsE)"22z"B"BCCr;r`c3$K|] }|D]}|V dSrr`)rbrKtasks r%rz2Parallel._get_sequential_output..ps/QQU5QQ4DQQQQQQQr;NF)rRrrUrrHrGrrrrrrrr-) rkiterableiterable_batchedrrrresrrs @@r%_get_sequential_outputzParallel._get_sequential_output`s " ""DO&.D #--//JQ(^^#'CCCCCR$$ RQ.>QQQJJJ'/ ' '"dF))Q.))''1,''dD+F++&&!+&&##%%% !!Q&!!! '   "DO!DN DM    '"DM#DO&*D #    ! ! ! ! !"DM#DO&*D #    ! ! ! !sCC D !C,,D+Ec,t|dt5|jrd}|jdur|dz }t |d|_dddn #1swxYwYd|_d|_d|_d|_d|_ d|_ d|_ dS)z9Reset the counters and flags used to track the execution.rz+This Parallel instance is already running !Tz Before submitting new tasks, you must wait for the completion of all the previous tasks, or clean all references to the output generator.NrF) rGrr-r RuntimeErrorrHrGrrrrr)rkr$s r%_reset_run_trackingzParallel._reset_run_trackings T7KMM 2 2 ! !} (C(D00>C #3''' DM ! ! ! ! ! ! ! ! ! ! ! ! ! ! !%&!"#!"   s.AAAclt|drt|nd_t j_js}n}|dkr< |}t|j r|nt|Sj 5tj_dddn #1swxYwY|_t%jt(r fd}|_|_jjj}|dkrt1d|zd|d|d tjd rjd _t9|}j}|d krd_d_nw|_t|d r0tA|!dtE|}tG|x_}tIj%|j}tM_'(||}tSj*|_+t|j r|nt|S)z)Main function to dispatch parallel tasks.rNr cZjjjjdSr)rH_workers_temp_folder_managerset_current_contextr0rsr%_batched_calls_reducer_callbackz:Parallel.__call__.._batched_calls_reducer_callbacks4 &;OOHr;rz%s has no active worker.zUsing backend z with z concurrent workers. start_callTrendswithr+),rrrxrwrrxr.r7rArrrrrr r/rrXrarHrrrPrQrrprr6rrrryrreplaceryrrYrZrrrweakrefrefr1)rkrr+outputr backend_namerars` r%rzParallel.__call__s   """(/)(D(DNs8}}}$ 9;;$ .--//FF++--F Q;;00::F LLL!2D66V DZ ( (!GGKDM ( ( ( ( ( ( ( ( ( ( ( ( ( ( ( )/% dm[ 1 1 E     &ED ")/%}.7 Q;;9LHII I U\UUUUUVVV 4=, / / ' M $ $ & & & >>( 5 &*D #()D % %&.D #|Z00 V()=)=hF )T)TUU 7:<7H7H HD % !'$2KLLH "VV""8\:: V,, V .@vvDLL@s C22C69C6c0|jjd|jdS)Nz(n_jobs=))rPrQr+rs r%__repr__zParallel.__repr__s"&."9"9"94;;;GGr;)rQrrrr6rnrrr7rAr:rMrIrrQrUrprrrr9rrrrrrrrrrrrs@r%rrs${{~'x0' 2' 2!+M:*<8)+6&x0' 2VVVVVVp $$$ , &&&(((<... + + +TTTl### % % %   ===~"$$$*MMM^6DDDL / / /6("("("T%%%NgAgAgARHHHHHHHr;r)F)r)Qr __future__rr%rrYosr*rkrrr|r contextlibrmathrrrrrnumbersr uuidr _multiprocessing_helpersr _parallel_backendsr rrrrrr_utilsrrdiskrloggerrrrLrMrNrSr{ externalsrlocalrHr&rzr6rIrKr:r@r>r\rrrenvironrVstriprrobjectrrrrrrrrr!r rr`r;r%rs   """"""##########(((((((((((()(((((((!!!!!!--------"#  $%-v6>"8H "HVO$ 9?   & & &" Nyt,,,id+++yq)))94000)$///---id+++yt,,,  5/( #8 , #I . #I . #8 , #I . #I .O#O#O#O#dq>q>q>q>q>q>q>q>hh)h)h)h)h)h)h)h)` 72}; Z^^12 6 6 < < > > F$F +R^6:::!!!!!6!!!J    CCCC0)))&   V-V-V-V-V-fV-V-V-t*3333>]H]H]H]H]Hv]H]H]H]H]Hr;