
    HWhc                     \   S r SSKrSSKrSSKrSSKrSSKrSSKJrJr  SSK	J
r
Jr  SSKJr  \b   SSKJr  SSKJr  SS	KJr  SS
KJrJr  SSKJr   " S S\S9r " S S\5      r " S S\5      r " S S\5      r " S S\\5      r " S S\\\5      r " S S\\5      r  " S S\!5      r"S r#g)z,
Backends for embarrassingly parallel code.
    N)ABCMetaabstractmethod   )_TracebackCapturingWrapper*_retrieve_traceback_capturing_wrapped_call)mp)MemmappingPool)
ThreadPool)get_memmapping_executor)process_executor	cpu_count)ShutdownExecutorErrorc                     ^  \ rS rSrSrSrSrSr\S 5       r	\S 5       r
SrSU 4S jjr/ S	QrS
r\S 5       r\SS j5       rS rSS jrS rS rS rS rS rS rSS jrS r\R:                  S 5       rS r\ S 5       r!Sr"U =r#$ )ParallelBackendBase   zEHelper abc which defines all methods a ParallelBackend must implementFr   c                     U R                   $ Nsupports_retrieve_callbackselfs    ;/usr/lib/python3/dist-packages/joblib/_parallel_backends.pysupports_return_generator-ParallelBackendBase.supports_return_generator$       ...    c                     U R                   $ r   r   r   s    r   supports_timeout$ParallelBackendBase.supports_timeout(   r   r   Nc                 >   > [         TU ]  " S0 UD6  Xl        X l        g N )super__init__nesting_levelinner_max_num_threads)r   r%   r&   kwargs	__class__s       r   r$   ParallelBackendBase.__init__.   s    "6"*%:"r   )OMP_NUM_THREADSOPENBLAS_NUM_THREADSMKL_NUM_THREADSBLIS_NUM_THREADSVECLIB_MAXIMUM_THREADSNUMBA_NUM_THREADSNUMEXPR_NUM_THREADS
ENABLE_IPCc                     g)a  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. 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.
Nr"   r   n_jobss     r   effective_n_jobs$ParallelBackendBase.effective_n_jobs<       r   c                     g)Schedule a func to be runNr"   r   funccallbacks      r   apply_asyncParallelBackendBase.apply_asyncN   r7   r   c                     g)zCalled within the callback function passed in apply_async.

The argument of this function is the argument given to a callback in
the considered backend. It is supposed to return the outcome of a task
if it succeeded or raise the exception if it failed.
Nr"   r   outs     r   retrieve_result_callback,ParallelBackendBase.retrieve_result_callbackR   r7   r   c                 0    X l         U R                  U5      $ )zReconfigure the backend and return the number of workers.

This makes it possible to reuse an existing backend instance for
successive independent calls to Parallel with different parameters.
)parallelr5   )r   r4   rE   preferrequirebackend_argss         r   	configureParallelBackendBase.configureZ   s     !$$V,,r   c                     g)z;Call-back method called at the beginning of a Parallel callNr"   r   s    r   
start_callParallelBackendBase.start_calld   r7   r   c                     g)z5Call-back method called at the end of a Parallel callNr"   r   s    r   	stop_callParallelBackendBase.stop_callg   r7   r   c                     g)z0Shutdown the workers and free the shared memory.Nr"   r   s    r   	terminateParallelBackendBase.terminatej   r7   r   c                     g) Determine the optimal batch sizer   r"   r   s    r   compute_batch_size&ParallelBackendBase.compute_batch_sizem   s    r   c                     g)1Callback indicate how long it took to run a batchNr"   )r   
batch_sizedurations      r   batch_completed#ParallelBackendBase.batch_completedq   r7   r   c                     / $ )z'List of exception types to be captured.r"   r   s    r   get_exceptions"ParallelBackendBase.get_exceptionst   s    	r   c                     g)a  Abort any running tasks

This is called when an exception has been raised when executing a task
and all the remaining tasks will be ignored and can therefore be
aborted to spare computation resources.

If ensure_ready is True, the backend should be left in an operating
state as future tasks might be re-submitted via that same backend
instance.

If ensure_ready is False, the implementer of this method can decide
to leave the backend in a closed / terminated state as no new task
are expected to be submitted to this backend.

Setting ensure_ready to False is an optimization that can be leveraged
when aborting tasks via killing processes from a local process pool
managed by the backend it-self: if we expect no new tasks, there is no
point in re-creating new workers.
Nr"   r   ensure_readys     r   abort_everything$ParallelBackendBase.abort_everythingx   s    , 	r   c                 Z    [        U SS5      S-   nUS:  a  [        US9S4$ [        US9S4$ )zBackend instance to be used by nested Parallel calls.

By default a thread-based backend is used for the first level of
nesting. Beyond, switch to sequential backend to avoid spawning too
many threads on the host.
r%   r   r   r%   N)getattrSequentialBackendThreadingBackend)r   r%   s     r   get_nested_backend&ParallelBackendBase.get_nested_backend   s?      oq9A=1$=A4GG#-@$FFr   c              #      #    Sv   g7f)aL  Context manager to manage an execution context.

Calls to Parallel.retrieve will be made inside this context.

By default, this does nothing. It may be useful for subclasses to
handle nested parallelism. In particular, it may be required to avoid
deadlocks if a backend manages a fixed number of workers, when those
workers may be asked to do nested Parallel calls. Without
'retrieval_context' this could lead to deadlock, as all the workers
managed by the backend may be "busy" waiting for the nested parallel
calls to finish, but the backend has no free workers to execute those
tasks.
Nr"   r   s    r   retrieval_context%ParallelBackendBase.retrieval_context   s
      	s   c                 2   U R                   n[        [        5       U-  S5      n0 nU R                   H5  nUc   [        R
                  R                  XS5      nOUn[        U5      XE'   M7     U R                  [        R
                  ;  a  SX@R                  '   U$ )a>  Return environment variables limiting threadpools in external libs.

This function return a dict containing environment variables to pass
when creating a pool of process. These environment variables limit the
number of threads to `n_threads` for OpenMP, MKL, Accelerated and
OpenBLAS libraries in the child processes.
r   1)	r&   maxr   MAX_NUM_THREADS_VARSosenvirongetstrTBB_ENABLE_IPC_VAR)r   r4   explicit_n_threadsdefault_n_threadsenvvar	var_values          r   _prepare_worker_env'ParallelBackendBase._prepare_worker_env   s     "77	v 5q9
 ,,C!)JJNN3B	.	9~CH - """**4 ,/C''(
r   c                  \    [        [        R                  " 5       [        R                  5      $ r   )
isinstance	threadingcurrent_thread_MainThreadr"   r   r   in_main_thread"ParallelBackendBase.in_main_thread   s    )224i6K6KLLr   )r&   r%   rE   )NNr   r   NNNT)$__name__
__module____qualname____firstlineno____doc__supports_inner_max_num_threadsr   default_n_jobspropertyr   r   r%   r$   rs   rx   r   r5   r=   rB   rI   rL   rO   rR   rV   r\   r_   rd   rk   
contextlibcontextmanagerrn   r~   staticmethodr   __static_attributes____classcell__r(   s   @r   r   r      s    O%*"!&N/ / / / M; & " ( (-JD?@0G   : M Mr   r   )	metaclassc                   D    \ rS rSrSrSrSrSrSrS r	SS jr
S rS	 rS
rg)ri      zA ParallelBackend which will execute all batches sequentially.

Does not use/create any threading objects, and hence has minimal
overhead. Used when n_jobs == 1.
TFc                 &    US:X  a  [        S5      eg?Determine the number of jobs which are going to run in parallelr   &n_jobs == 0 in Parallel has no meaningr   )
ValueErrorr3   s     r   r5   "SequentialBackend.effective_n_jobs   s    Q;EFFr   Nc                     [        S5      e)r9   -Should never be called for SequentialBackend.RuntimeErrorr:   s      r   r=   SequentialBackend.apply_async   s    JKKr   c                     [        S5      e)Nr   r   r@   s     r   rB   *SequentialBackend.retrieve_result_callback   s    JKKr   c                     SSK Jn  U" 5       $ )Nr   )get_active_backend)rE   r   )r   r   s     r   rk   $SequentialBackend.get_nested_backend   s    0 "##r   r"   r   )r   r   r   r   r   uses_threadsr   r   supports_sharedmemr5   r=   rB   rk   r   r"   r   r   ri   ri      s6     L!&LL$r   ri   c                   H    \ rS rSrSrSrS rS rS rSS jr	S r
SS	 jrS
rg)PoolManagerMixin   z,A helper class for managing pool of workers.Nc                 ~    US:X  a  [        S5      e[        b  Uc  gUS:  a  [        [        5       S-   U-   S5      nU$ r   )r   r   rr   r   r3   s     r   r5   !PoolManagerMixin.effective_n_jobs   sF    Q;EFFZ6> aZq6115Fr   c                     U R                   b<  U R                   R                  5         U R                   R                  5         SU l         ggz#Shutdown the process or thread poolN)_poolcloserR   r   s    r   rR   PoolManagerMixin.terminate  s8    ::!JJJJ  "DJ "r   c                     U R                   $ )z>Used by apply_async to make it possible to implement lazy initr   r   s    r   	_get_poolPoolManagerMixin._get_pool
  s    zzr   c                 R    U R                  5       R                  [        U5      SX"S9$ )r9   r"   )r<   error_callback)r   r=   r   r:   s      r   r=   PoolManagerMixin.apply_async  s0    
 ~~++&t,b , 
 	
r   c                     [        U5      $ )z=Mimic concurrent.futures results, raising an error if needed.)r   r@   s     r   rB   )PoolManagerMixin.retrieve_result_callback  s    9#>>r   c                     U R                  5         U(       aH  U R                  " SU R                  R                  U R                  S.U R                  R                  D6  gg)z@Shutdown the pool and restart a new one with the same parametersr4   rE   Nr"   )rR   rI   rE   r4   _backend_argsrb   s     r   rd   !PoolManagerMixin.abort_everything  sI    NN :$--"6"6 :!]]88: r   r   r   r   )r   r   r   r   r   r   r5   rR   r   r=   rB   rd   r   r"   r   r   r   r      s(    6E

?:r   r   c                   N   ^  \ rS rSrSrSrSrSrSrU 4S jr	S r
S	 rS
 rSrU =r$ )AutoBatchingMixini$  z/A helper class for automagically batching jobs.皙?   r   g        c                 j   > [         TU ]  " S0 UD6  U R                  U l        U R                  U l        g r!   )r#   r$   _DEFAULT_EFFECTIVE_BATCH_SIZE_effective_batch_size _DEFAULT_SMOOTHED_BATCH_DURATION_smoothed_batch_duration)r   r'   r(   s     r   r$   AutoBatchingMixin.__init__5  s/    "6"%)%G%G"(,(M(M%r   c                    U R                   nU R                  nUS:  a  X R                  :  a~  [        UU R                  -  U-  5      nUS-  n[	        SU-  U5      n[        US5      nX@l         U R                  R                  S:  a"  U R                  R                  SU SU S35        OX R                  :  ar  US:  al  [        XR                  -  U-  5      n[        SU-  S5      nX@l         U R                  R                  S:  a"  U R                  R                  SU SU S35        OUnXA:w  a  U R                  U l        U$ )	rU   r   r   r   
   zBatch computation too fast (zs.) Setting batch_size=.zBatch computation too slow ()r   r   MIN_IDEAL_BATCH_DURATIONintminrr   rE   verbose_printMAX_IDEAL_BATCH_DURATIONr   )r   old_batch_sizebatch_durationideal_batch_sizerZ   s        r   rV   $AutoBatchingMixin.compute_batch_size:  sy   3366Q!>!>>  #>#'#@#@$A#1$2  3 ! Q/1ABJZ+J)3&}}$$*$$2>2B C**4Q8 <<<!  #!>!>>O  Q!1115J)3&}}$$*$$2>2B C**4Q8 (J' 55 ) r   c                     XR                   :X  a0  U R                  nX0R                  :X  a  UnOSU-  SU-  -   nX@l        gg)rY   g?r   N)r   r   r   )r   rZ   r[   old_durationnew_durations        r   r\   !AutoBatchingMixin.batch_completedx  sN    333  88LDDD  (  #\1C(NB,8) 4r   c                 H    U R                   U l        U R                  U l        g)zWReset batch statistics to default values.

This avoids interferences with future jobs.
N)r   r   r   r   r   s    r   reset_batch_stats#AutoBatchingMixin.reset_batch_stats  s     
 &*%G%G"(,(M(M%r   )r   r   )r   r   r   r   r   r   r   r   r   r$   rV   r\   r   r   r   r   s   @r   r   r   $  sA    9  "  ! %&!'*$N
<|9 N Nr   r   c                   4    \ rS rSrSrSrSrSrSS jrS r	Sr
g)	rj   i  aj  A ParallelBackend which will use a thread pool to execute batches in.

This is a low-overhead backend but it suffers from the Python Global
Interpreter Lock if the called function relies a lot on Python objects.
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).

The actual thread pool is lazily initialized: the actual thread pool
construction is delayed to the first call to apply_async.

ThreadingBackend is used as the default backend for nested calls.
TNc                     U R                  U5      nUS:X  a  [        [        U R                  S95      eX l        Xl        U$ z?Build a process or thread pool and return the number of workersr   rg   )r5   FallbackToBackendri   r%   rE   _n_jobs)r   r4   rE   rH   s       r   rI   ThreadingBackend.configure  sE    &&v.Q;#!0B0BCE E r   c                 h    U R                   c  [        U R                  5      U l         U R                   $ )z{Lazily initialize the thread pool

The actual pool of worker threads is only initialized at the first
call to apply_async.
)r   r
   r   r   s    r   r   ThreadingBackend._get_pool  s(     ::#DLL1DJzzr   )r   r   rE   )r   N)r   r   r   r   r   r   r   r   rI   r   r   r"   r   r   rj   rj     s$     "&L	r   rj   c                   J   ^  \ rS rSrSrSrSrU 4S jrS	S jrU 4S jr	Sr
U =r$ )
MultiprocessingBackendi  a  A ParallelBackend which will use a multiprocessing.Pool.

Will introduce some communication and memory overhead when exchanging
input and output data with the with the worker Python processes.
However, does not suffer from the Python Global Interpreter Lock.
TFc                   > [         c  g[         R                  " 5       R                  (       a0  US:w  a)  [        5       (       a  SnOSn[        R
                  " USS9  g[        R                  S:  a  US:w  a  [        R
                  " SSS9  gU R                  5       (       d,  U R                  S:X  d  US:w  a  [        R
                  " SSS9  g[        [        U ]3  U5      $ )	zDetermine the number of jobs which are going to run in parallel.

This also checks if we are attempting to create a nested parallel
loop.
r   zInside a Dask worker with daemon=True, setting n_jobs=1.
Possible work-arounds:
- dask.config.set({'distributed.worker.daemon': False})- set the environment variable DASK_DISTRIBUTED__WORKER__DAEMON=False
before creating your Dask cluster.zHMultiprocessing-backed parallel loops cannot be nested, setting n_jobs=1   
stacklevelr   zTMultiprocessing-backed parallel loops cannot be nested, below loky, setting n_jobs=1zVMultiprocessing-backed parallel loops cannot be nested below threads, setting n_jobs=1)r   current_processdaemoninside_dask_workerwarningswarnr   _CURRENT_DEPTHr   r%   r#   r   r5   )r   r4   msgr(   s      r   r5   'MultiprocessingBackend.effective_n_jobs  s     :&&{%''= =  ca0**Q.{4 " %%''4+=+=+B{7 " +TCFKKr   c                     U R                  U5      nUS:X  a  [        [        U R                  S95      e[        R
                  " 5         [        U40 UD6U l        X l        U$ r   )	r5   r   ri   r%   gccollectr	   r   rE   )r   r4   rE   rF   rG   memmappingpool_argss         r   rI    MultiprocessingBackend.configure  s_     &&v.Q;#!0B0BCE E 	

#FB.AB
 r   c                 J   > [         [        U ]  5         U R                  5         gr   )r#   r   rR   r   )r   r(   s    r   rR    MultiprocessingBackend.terminate  s    $d57 r   )r   rE   r   )r   r   r   r   r   r   r   r5   rI   rR   r   r   r   s   @r   r   r     s,     "& %0Ld! !r   r   c                   T    \ rS rSrSrSrSr  SS jrS rSS jr	S r
S	 rSS
 jrSrg)LokyBackendi  z>Managing pool of workers with loky instead of multiprocessing.TNc                     U R                  U5      nUS:X  a  [        [        U R                  S95      e[	        U4UU R                  US9UR                  S.UD6U l        X l        U$ )z9Build a process executor and return the number of workersr   rg   )r4   )timeoutr{   
context_id)	r5   r   ri   r%   r   r~   _id_workersrE   )r   r4   rE   rF   rG   idle_worker_timeoutmemmappingexecutor_argss          r   rI   LokyBackend.configure  s     &&v.Q;#!0B0BCE E 0@/(((7||@ (?@ !r   c                    US:X  a  [        S5      e[        b  Uc  g[        R                  " 5       R                  (       a0  US:w  a)  [	        5       (       a  SnOSn[
        R                  " USS9  gU R                  5       (       d,  U R                  S:X  d  US:w  a  [
        R                  " SSS9  gUS:  a  [        [        5       S-   U-   S5      nU$ )	r   r   r   r   zInside a Dask worker with daemon=True, setting n_jobs=1.
Possible work-arounds:
- dask.config.set({'distributed.worker.daemon': False})
- set the environment variable DASK_DISTRIBUTED__WORKER__DAEMON=False
before creating your Dask cluster.zRLoky-backed parallel loops cannot be called in a multiprocessing, setting n_jobs=1r   r   zKLoky-backed parallel loops cannot be nested below threads, setting n_jobs=1)r   r   r   r   r   r   r   r   r%   rr   r   )r   r4   r   s      r   r5   LokyBackend.effective_n_jobs!  s    Q;EFFZ6> !(({%''= =  ca0%%''4+=+=+B{0 " aZq6115Fr   c                 d    U R                   R                  U5      nUb  UR                  U5        U$ )r9   )r  submitadd_done_callback)r   r;   r<   futures       r   r=   LokyBackend.apply_asyncJ  s/    %%d+$$X.r   c                 V     UR                  5       $ ! [         a    [        S5      ef = f)Na  The executor underlying Parallel has been shutdown. This is likely due to the garbage collection of a previous generator from a call to Parallel with return_as='generator'. Make sure the generator is not garbage collected when submitting a new job or that it is first properly exhausted.)resultr   r   r@   s     r   rB   $LokyBackend.retrieve_result_callbackQ  s5    		::<$ 	O 	s    (c                     U R                   b?  U R                   R                  R                  U R                  R                  SS9  S U l         U R                  5         g )NF)r  force)r  _temp_folder_manager_clean_temporary_resourcesrE   r  r   r   s    r   rR   LokyBackend.terminate]  sP    ==$ MM..II==,,E J  !DM r   c                     U R                   R                  SS9  SU l         U(       a/  U R                  U R                  R                  U R                  S9  gg)zLShutdown the workers and restart a new one with the same parameters
        T)kill_workersNr   )r  rR   rI   rE   r4   rb   s     r   rd   LokyBackend.abort_everythingi  sF     	T2NN$--"6"6NO r   )r  rE   )r   NNNi,  r   r   )r   r   r   r   r   r   r   rI   r5   r=   rB   rR   rd   r   r"   r   r   r   r     s7    H!%%)"FJ&)'R

!Pr   r   c                       \ rS rSrSrS rSrg)r   is  z<Raised when configuration should fallback to another backendc                     Xl         g r   backend)r   r  s     r   r$   FallbackToBackend.__init__v  s    r   r  N)r   r   r   r   r   r$   r   r"   r   r   r   r   s  s
    Fr   r   c                  b     SSK Jn    U " 5         g! [         a     gf = f! [         a     gf = f)zICheck whether the current function is executed inside a Dask worker.
    r   
get_workerFT)distributedr!  ImportErrorr   r   s    r   r   r   z  s>    *    s    ! 

..)$r   r   rt   r   r   r   abcr   r   _utilsr   r   _multiprocessing_helpersr   poolr	   multiprocessing.poolr
   executorr   externals.lokyr   r   externals.loky.process_executorr   r   ri   objectr   r   rj   r   r   	Exceptionr   r   r"   r   r   <module>r.     s    
 	    '
 )>$/1 <FpMG pMf$+ $D/:v /:djN jNZ&')< &RO!-/@0O!ddP#%8 dPN	 r   