/usr/local/lib/python3.6/site-packages/joblib/__pycache__
Edit: /usr/local/lib/python3.6/site-packages/joblib/__pycache__/parallel.cpython-36.pyc (32416B)
3
Egb @ s d Z ddlmZ ddlZddlZddlmZ ddlZddlZddl Z ddl
Z
ddl m Z ddl
mZ ddlZddlZddlmZ dd lmZmZ dd
lmZ dd lmZmZmZm Z m Z dd l m Z m Z dd
l!m"Z" ddl#m$Z$ ddlm%Z% ddlm&Z& eee e dZ'da(dZ)dZ*e j+ Z,d3Z-d4Z.dd Z/de/iZ0d5ddZ1G d d d e2Z3dZ4e5ed rej6j7d d j8 prdZ9e9dk rej:e9d!Z4G d"d# d#e2Z;d6d%d&Z
G d+d, d,e2Z?d7d-d.Z@d9d/d0ZAG d1d2 d2eZBdS ):z+
Helpers for embarrassingly parallel code.
)divisionN)sqrt)uuid4)Integral )mp)Loggershort_format_time)memstr_to_bytes)FallbackToBackendMultiprocessingBackendThreadingBackendSequentialBackend LokyBackend)dumpsloads)loky) eval_expr)AutoBatchingMixin)ParallelBackendBase)multiprocessing threading
sequentialr r r processesthreads sharedmemc C sP yddl m} td| W n0 tk
rJ } zd}t||W Y dd}~X nX dS )z? Register Dask Backend if called with parallel_backend("dask") r )DaskDistributedBackenddaskzTo use the dask.distributed backend you must install both the `dask` and distributed modules.
See https://dask.pydata.org/en/latest/install.html for more information.N)Z_daskr register_parallel_backend ImportError)r emsg r" 9/usr/local/lib/python3.6/site-packages/joblib/parallel.py_register_dask; s r$ r c
C s | t krtd| t f |tkr0td|tf | dkrH|dkrHtdttdd}|dk r|\}}|j}t|dd }|dkr| rtt |d
}|d krtd |j j
|j j
f |t fS |S tt d
d
}t|dd }t|dd } |dkr| s| dkr| rtt d
d
}|t fS )z!Return the active default backendz9prefer=%r is not a valid backend hint, expected one of %rz@require=%r is not a valid backend constraint, expected one of %rr r zJprefer == 'processes' and require == 'sharedmem' are inconsistent settingsbackend_and_jobsNsupports_sharedmemF)
nesting_level
ziUsing %s as joblib.Parallel backend instead of %s as the latter does not provide shared memory semantics.r uses_threadsr )
VALID_BACKEND_HINTS
ValueErrorVALID_BACKEND_CONSTRAINTSgetattr_backendr' BACKENDSDEFAULT_THREAD_BACKENDprint __class____name__DEFAULT_N_JOBSDEFAULT_BACKEND)
preferrequireverboser% backendn_jobsr' r& Zsharedmem_backendr) r" r" r# get_active_backendM s: r; c @ s2 e Zd ZdZd
ddZdd Zdd Zd
d ZdS )parallel_backenda
Change the default backend used by Parallel inside a with block.
If ``backend`` is a string it must match a previously registered
implementation using the ``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.
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. 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 ``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]
Warning: this function is experimental and subject to change in a future
version of joblib.
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
r Nc K s t |tr6|tkr(|tkr(t| }| t| f |}|d k r`dj|jj}|jsZt|||_ t
t dd }|j d kr|d krd}n
|d j }||_ || _
||f| _||ft _d S )Nz>{} does not accept setting the inner_max_num_threads argument.r% r )
isinstancestrr/ EXTERNAL_BACKENDSformatr2 r3 Z supports_inner_max_num_threadsAssertionErrorinner_max_num_threadsr- r. r' old_backend_and_jobsnew_backend_and_jobsr% ) selfr9 r: rB Zbackend_paramsregisterr! Zcurrent_backend_and_jobsr' r" r" r# __init__ s&
zparallel_backend.__init__c C s | j S )N)rD )rE r" r" r# __enter__ s zparallel_backend.__enter__c C s | j d S )N)
unregister)rE typevalue tracebackr" r" r# __exit__ s zparallel_backend.__exit__c C s, | j d kr ttdd d k r(t`n| j t_d S )Nr% )rC r- r. r% )rE r" r" r# rI s
zparallel_backend.unregister)rN N)r3
__module__ __qualname____doc__rG rH rM rI r" r" r" r# r< { s
=
r< get_contextZJOBLIB_START_METHOD )methodc @ s2 e Zd ZdZd ddZdd Zdd Zd d
ZdS ) BatchedCallszCWrap a sequence of (func, args, kwargs) tuples as a single callableNc C sX t || _t| j| _|| _t|tr4|\| _| _n|d | _| _|d k rN|ni | _ d S )N)
listitemslen_size_reducer_callbackr= tupler. _n_jobs
_pickle_cache)rE Ziterator_slicer% Zreducer_callbackZ pickle_cacher" r" r# rG s
zBatchedCalls.__init__c C s, t | j| jd dd | jD S Q R X d S )N)r: c S s g | ]\}}}|||qS r" r" ).0funcargskwargsr" r" r#
s z)BatchedCalls.__call__..)r< r. r\ rW )rE r" r" r# __call__ s zBatchedCalls.__call__c C s. | j d k r| j t| j| j| jfd | jffS )N)rZ rU rW r. r\ r] )rE r" r" r#
__reduce__
s
zBatchedCalls.__reduce__c C s | j S )N)rY )rE r" r" r# __len__ s zBatchedCalls.__len__)NN)r3 rO rP rQ rG rc rd re r" r" r" r# rU s
rU Fc C s t dkr dS tj| dS )a Return 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).
If only_physical_cores is True, do not take hyperthreading / SMT logical
cores into account.
Nr )only_physical_cores)r r cpu_count)rf r" r" r# rg s rg c C s\ |sdS |dkrdS | dkr dS dd| d }t | | }t | d | }t|t|kS ) z Returns False for indices increasingly apart, the distance
depending on the value of verbose.
We use a lag increasing as the square of index
Tr( Fr g ? r )r int)indexr8 ZscaleZ
next_scaler" r" r# _verbosity_filter/ s rl c s8 fdd}yt j |}W n tk
r2 Y nX |S )z6Decorator used to capture the arguments of a function.c s
| |fS )Nr" )r` ra )functionr" r# delayed_functionE s z!delayed..delayed_function) functoolswrapsAttributeError)rm rn r" )rm r# delayedB s rr c @ s e Zd ZdZdd Zdd ZdS )BatchCompletionCallBacka_ Callback used by joblib.Parallel's multiprocessing backend.
This callable is executed by the parent process whenever a worker process
has returned the results of 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.
c C s || _ || _|| _d S )N)dispatch_timestamp
batch_sizeparallel)rE rt ru rv r" r" r# rG Z s z BatchCompletionCallBack.__init__c C sj | j j| j7 _tj | j }| j jj| j| | j j | j j | j j d k r\| j j
W d Q R X d S )N) rv n_completed_tasksru timert r. Zbatch_completedprint_progress_lock_original_iterator
dispatch_next)rE outZthis_batch_durationr" r" r# rc _ s
z BatchCompletionCallBack.__call__N)r3 rO rP rQ rG rc r" r" r" r# rs O s
rs c C s |t | < |r| adS )a Register a new Parallel backend factory.
The new backend can then be selected by passing its name as the backend
argument to the 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)r/ r5 )namefactoryZ make_defaultr" r" r# r l s r c C s" t \}}| dkr|} |j| dS )af Determine 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
N)r: )r; effective_n_jobs)r: r9 Zbackend_n_jobsr" r" r# r s
r c @ s e Zd ZdZd$dd Zd
d Zd d
Zdd Zdd Zdd Z dd Z
dd Z dd Z dd Z
d d Zd d Zd d! Zd"d# ZdS )%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="multiprocessing"
or the size of the thread-pool when backend="threading".
If -1 all CPUs are used. If 1 is given, no parallel computing code
is used at all, which is useful for debugging. For n_jobs below -1,
(n_cpus + 1 + n_jobs) are used. Thus for n_jobs = -2, all
CPUs but one are used.
None is a marker for 'unset' that will be interpreted as n_jobs=1
(sequential execution) unless the call is performed under a
parallel_backend context manager that sets another value for
n_jobs.
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.
- "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
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
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
parallel_backend context manager.
prefer: str in {'processes', 'threads'} or None, default: None
Soft hint to choose the default backend if no specific backend
was selected with the parallel_backend 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
parallel_backend.
verbose: int, optional
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, optional
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'}
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
arithmetics 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, optional
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, 1M by default
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.
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.
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
Nr
2 * n_jobsauto1Mrc C s t |
| |d\} }
| j}|d kr,|d kr,|
}|d kr8d}|| _|| _|| _|| _tj | _t j
| _ d | _ t
|tr|t|}t|| ||
| td| jd d| _td k rt| jd< nttdrtj | jd< |d kr| }nt
|tr|jd kr||_nt|dr"t|d r"|| jd< t|d
}nVy t| }W n> tk
rl } z td |t tj f |W Y d d }~X nX ||d
}| d krt |d
d rtd| |dkst
|t r|dkr|| _ n td| || _!d | _"t# | _$d| _%t&j' | _(d S )N)r6 r7 r8 r r 2 )
max_nbytes mmap_mode temp_folderr6 r7 r8 contextrR PoolLock)r' z'Invalid backend: %s, expected one of %rr r& Fz)Backend %s does not support shared memoryr z8batch_size must be 'auto' or a positive integer, got: %r))r; r' r: r8 timeout pre_dispatchqueueQueue_ready_batchesr hex_idrZ r= r> r
dictmax
_backend_argsDEFAULT_MP_CONTEXThasattrr rR r r r/ KeyErrorr+ sortedkeysr- r ru r. _outputrV _jobs_managed_backendr RLockrz )rE r: r9 r8 r r ru r r r r6 r7 Zactive_backendZcontext_n_jobsr' Zbackend_factoryr r" r" r# rG sp
(
zParallel.__init__c C s d| _ | j | S )NT)r _initialize_backend)rE r" r" r# rH s zParallel.__enter__c C s | j d| _d S )NF)_terminate_backendr )rE exc_type exc_valuerL r" r" r# rM s zParallel.__exit__c C s yN| j jf | j| d| j}| jdk rL| j j rLtjdj| j j j
| j W n2 t k
r } z|j | _ | j
}W Y dd}~X nX |S )z?Build a process or thread pool and return the number of workers)r: rv NzThe backend class {!r} does not support timeout. You have set 'timeout={}' in Parallel but the 'timeout' parameter will not be used.)r. configurer: r r supports_timeoutwarningswarnr@ r2 r3 r r9 r )rE r: r r" r" r# r s
z Parallel._initialize_backendc C s | j r| j j| jS dS )Nr )r. r r: )rE r" r" r# _effective_n_jobs s zParallel._effective_n_jobsc C s | j d k r| j j d S )N)r. terminate)rE r" r" r# r s
zParallel._terminate_backendc C s | j r
dS | jt|7 _| jd7 _tj }t|t|| }| j. t| j}| jj ||d}| jj
|| W dQ R X 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) _abortingn_dispatched_tasksrX n_dispatched_batchesrx rs rz r r. Z apply_asyncinsert)rE batchrt cbZjob_idxjobr" r" r# _dispatch s
zParallel._dispatchc C s | j | jsd| _d| _dS )a Dispatch 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{
_iterating)rE r" r" r# r| s zParallel.dispatch_nextc C sF | j dkr| jj }n| j }| j y| jjdd}W n tjk
r | j}|| }t t
j ||}t |dkrzdS || j
krt ||k rtdt |d| }ntdt || }xHtdt ||D ]4}t|||| | jj | j| j}| jj| qW | jjdd}Y nX t |dkr*dS | j| dS W dQ R X dS ) aT Prefetch 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.
r F)blockr r r( TN)ru r. Zcompute_batch_sizerz r getr Empty_cached_effective_n_jobsrV itertoolsislicerX r{ r rangerU Zget_nested_backendrZ r] putr ) rE iteratorru tasksr: Zbig_batch_sizer Zfinal_batch_sizeir" r" r# r ! s6
zParallel.dispatch_one_batchc C sB | j s
dS | j dk r tjj}ntjj}|| }|d| |f dS )z=Display the message on stout or stderr depending on verbosityNr z [%s]: %s
)r8 sysstderrwritestdout)rE r! Zmsg_argswriterr" r" r# _printc s
zParallel._printc C s | j s
dS tj | j }| jdk rLt| j| j r4dS | jd| jt|f n| j}| j }|dks|| d | j
}|| j d }|d |k}|s|| rdS || | j |d }| jd||t|t|f dS )zvDisplay the process of the parallel execution only a fraction
of time, controlled by self.verbose.
Nz!Done %3i tasks | elapsed: %sr r g ?z/Done %3i out of %3i | elapsed: %s remaining: %s) r8 rx _start_timer{ rl r r rw r r _pre_dispatch_amount)rE elapsed_timerk Z total_taskscursorZ frequencyZ is_last_itemZremaining_timer" r" r# ry p s2
zParallel.print_progressc C s t | _x| js t| jdkrt| jdkr8tjd q
| j | jjd}W d Q R X y:t | j
ddr~| jj |j | j
d n| jj |j W q
tk
r } z8d| _| j
}|d k rt|dr| j}|j|d W Y d d }~X q
X q
W d S ) Nr g{Gz?r F)r Tabort_everything) ensure_ready)rV r r rX r rx sleeprz popr- r. extendr r
BaseExceptionr r r r )rE r exceptionr9 r r" r" r# retrieve s&
zParallel.retrievec s j rtdd _ js$ j }n j }t jtrJ fdd}| _ | _
jj j }|dkrnt
d| jd||f t jdr jj t|} j}|d ks|d
krd _d _n@| _t|d rt|jd t|}t| _}tj| j}tj _ d _ d _ d _ t _!zd _" j#|rD jd k _"x j#|rVqFW |d ksl|d
krrd _" jj$ j% W d Q R X tj j } jd
t& j't& j't(|f W d t jdr܈ jj) js j* t+ _ d _!X j'}d _'|S )Nz)This Parallel instance is already runningFc s j jjj j d S )N)r. Z_workersZ_temp_folder_managerZset_current_contextr r" )rE r" r# _batched_calls_reducer_callback s
z:Parallel.__call__.._batched_calls_reducer_callbackr z%s has no active worker.z,Using backend %s with %d concurrent workers.
start_callallr endswithr: z*Done %3i out of %3i | elapsed: %s finished stop_call),r r+ r r r r r= r. r rZ r r2 r3 RuntimeErrorr r r iterr r{ r r replacer> rj r r rx r r r rw r r] r r Zretrieval_contextr rX r r r r rV ) rE iterabler: r backend_namer r r outputr" )rE r# rc sp
zParallel.__call__c C s d| j j| jf S )Nz
%s(n_jobs=%s))r2 r3 r: )rE r" r" r# __repr__6 s zParallel.__repr__) NNr Nr r Nr r NN)r3 rO rP rQ rG rH rM r r r r r| r r ry r rc r r" r" r" r# r s& c
Q B
+(sr )r r N)r N)NNr )F)FrN )rN )CrQ
__future__r osr mathr ro rx r r uuidr Znumbersr r r Z_multiprocessing_helpersr loggerr r Zdiskr
Z_parallel_backendsr r r
r r Zexternals.cloudpickler r Z externalsr _utilsr r r r/ r5 r4 r0 localr. r* r, r$ r? r; objectr< r r environr striprT rR rU rg rl rr rs r r r r" r" r" r# s`
.q
(