L i^p`dZddlZddlZddlZddlZddlZddlmZmZddl m Z ddl m Z m Z e ddlmZddlmZdd lmZmZdd lmZdd lmZGd d eZGddeZGddeZGddeZGddeeZGddeeeZGddeeZ Gdde!Z"dZ#y)z, Backends for embarrassingly parallel code. N)ABCMetaabstractmethod)mp)*_retrieve_traceback_capturing_wrapped_call_TracebackCapturingWrapper) ThreadPool)get_memmapping_executor) cpu_countprocess_executor)ShutdownExecutorError)MemmappingPoolceZdZdZdZdZdZedZedZ dZ dfd Z gdZ d Z ed Zdd Zdd Zd ZddZ ddZdZdZdZdZdZddZdZdZej<dZe dZ!xZ"S)ParallelBackendBasezEHelper abc which defines all methods a ParallelBackend must implementrFc|jSNsupports_retrieve_callbackselfs _/mnt/ssd/data/python-lab/Trading/venv/lib/python3.12/site-packages/joblib/_parallel_backends.pysupports_return_generatorz-ParallelBackendBase.supports_return_generator)...c|jSrrrs rsupports_timeoutz$ParallelBackendBase.supports_timeout-rrNc Lt|||_||_||_yr)super__init__ nesting_levelinner_max_num_threadsbackend_kwargs)rr r!r" __class__s rrzParallelBackendBase.__init__3s( *%:",r)OMP_NUM_THREADSOPENBLAS_NUM_THREADSMKL_NUM_THREADSBLIS_NUM_THREADSVECLIB_MAXIMUM_THREADSNUMBA_NUM_THREADSNUMEXPR_NUM_THREADS ENABLE_IPCcy)aDetermine 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. Nrn_jobss reffective_n_jobsz$ParallelBackendBase.effective_n_jobsGrctd)z'Deprecated: implement `submit` instead.zImplement `submit` instead.)NotImplementedErrorrfunccallbacks r apply_asynczParallelBackendBase.apply_asyncYs!"?@@rcZtjdt|j||S)a8Schedule a function to be run and return a future-like object. This method should return a future-like object that allow tracking the progress of the task. If ``supports_retrieve_callback`` is False, the return value of this method is passed to ``retrieve_result`` instead of calling ``retrieve_result_callback``. Parameters ---------- func: callable The function to be run in parallel. callback: callable A callable that will be called when the task is completed. This callable is a wrapper around ``retrieve_result_callback``. This should be added to the future-like object returned by this method, so that the callback is called when the task is completed. For future-like backends, this can be achieved with something like ``future.add_done_callback(callback)``. Returns ------- future: future-like A future-like object to track the execution of the submitted function. z@`apply_async` is deprecated, implement and use `submit` instead.)warningswarnDeprecationWarningr7r4s rsubmitzParallelBackendBase.submit]s+:  N  h//rcy)aCalled within the callback function passed to `submit`. This method can customise how the result of the function is retrieved from the future-like object. Parameters ---------- future: future-like The future-like object returned by the `submit` method. Returns ------- result: object The result of the function executed in parallel. Nr-routs rretrieve_result_callbackz,ParallelBackendBase.retrieve_result_callbackr1rc^|jr|j|S|jS)aHook to retrieve the result when support_retrieve_callback=False. The argument `out` is the result of the `submit` call. This method should return the result of the computation or raise an exception if the computation failed. )timeout)rget)rr?rBs rretrieve_resultz#ParallelBackendBase.retrieve_results*  7777+ +779 rc 2||_|j|S)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. )parallelr0)rr/rFpreferrequirer"s r configurezParallelBackendBase.configures! $$V,,rcy)z;Call-back method called at the beginning of a Parallel callNr-rs r start_callzParallelBackendBase.start_callr1rcy)z5Call-back method called at the end of a Parallel callNr-rs r stop_callzParallelBackendBase.stop_callr1rcy)z0Shutdown the workers and free the shared memory.Nr-rs r terminatezParallelBackendBase.terminater1rcy) Determine the optimal batch sizerr-rs rcompute_batch_sizez&ParallelBackendBase.compute_batch_sizesrcy)1Callback indicate how long it took to run a batchNr-)r batch_sizedurations rbatch_completedz#ParallelBackendBase.batch_completedr1rcy)auAbort 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 rabort_everythingz$ParallelBackendBase.abort_everythings, rcdt|dddz}|dkDrt|dfSt|dfS)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 rrr N)getattrSequentialBackendThreadingBackend)rr s rget_nested_backendz&ParallelBackendBase.get_nested_backends? oq9A= 1 $=A4G G#-@$F Frc.|j}tt|zd}i}|jD]5}|!tj j ||}n|}t|||<7|jtj vrd||j<|S)afReturn 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. r1) r!maxr MAX_NUM_THREADS_VARSosenvironrCstrTBB_ENABLE_IPC_VAR)rr/explicit_n_threadsdefault_n_threadsenvvar var_values r_prepare_worker_envz'ParallelBackendBase._prepare_worker_envs"77 v 5q9 ,, &C!)JJNN30AB . 9~CH  &  " ""** 4,/C'' ( rc#Kdyw)aContext 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-rs rretrieval_contextz%ParallelBackendBase.retrieval_contexts  scZttjtjSr) isinstance threadingcurrent_thread _MainThreadr-rrin_main_threadz"ParallelBackendBase.in_main_thread s)224i6K6KLLr)NNrrNNNT)#__name__ __module__ __qualname____doc__default_n_jobssupports_inner_max_num_threadsrpropertyrrr rrerirr0r7r<r@rDrIrKrMrOrRrWr[raro contextlibcontextmanagerrq staticmethodrw __classcell__r#s@rrrsON%*" "' ////M9=-&  "A!0F " =A -JD?@ 0 G: MMrr) metaclassc:eZdZdZdZdZdZdZdZd dZ dZ dZ y) r_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$|dk(r tdy?Determine the number of jobs which are going to run in parallelr&n_jobs == 0 in Parallel has no meaningr) ValueErrorr.s rr0z"SequentialBackend.effective_n_jobss Q;EF FrNctd)Schedule a func to be run-Should never be called for SequentialBackend. RuntimeErrorr4s rr<zSequentialBackend.submit"sJKKrctd)Nrrr>s rr@z*SequentialBackend.retrieve_result_callback&sJKKrcddlm}|S)Nr)get_active_backend)rFr)rrs rraz$SequentialBackend.get_nested_backend)s0"##rr) rzr{r|r} uses_threadsrrsupports_sharedmemr0r<r@rar-rrr_r_s6 L!& LL$rr_c<eZdZdZdZdZdZdZd dZdZ d dZ y) PoolManagerMixinz,A helper class for managing pool of workers.Ncv|dk(r tdt|y|dkrttdz|zd}|Sr)rrrdr r.s rr0z!PoolManagerMixin.effective_n_jobs7sF Q;EF F Z6> aZq6115F rc|j<|jj|jjd|_yyz#Shutdown the process or thread poolN)_poolcloserOrs rrOzPoolManagerMixin.terminateCs8 :: ! JJ    JJ "DJ "rc|jS)z;Used by `submit` to make it possible to implement lazy init)rrs r _get_poolzPoolManagerMixin._get_poolJs zzrcZ|jjt|d||S)rr-)r6error_callback)rr7rr4s rr<zPoolManagerMixin.submitNs4 ~~++ &t , # ,  rct|S)z=Mimic concurrent.futures results, raising an error if needed.)r)rresults rr@z)PoolManagerMixin.retrieve_result_callbackZs :&AArc|j|rH|jd|jj|jd|jjyy)z@Shutdown the pool and restart a new one with the same parametersr/rFNr-)rOrIrFr/_backend_kwargsrYs rr[z!PoolManagerMixin.abort_everythingbsO   DNN }}++ --//  rrry) rzr{r|r}rr0rOrr<r@r[r-rrrr2s)6 E   BrrcDeZdZdZdZdZdZdZfdZdZ dZ d Z xZ S) AutoBatchingMixinz/A helper class for automagically batching jobs.皙?rgc ht|di||j|_|j|_y)Nr-)rr_DEFAULT_EFFECTIVE_BATCH_SIZE_effective_batch_size _DEFAULT_SMOOTHED_BATCH_DURATION_smoothed_batch_duration)rkwargsr#s rrzAutoBatchingMixin.__init__~s/ "6"%)%G%G"(,(M(M%rc|j}|j}|dkDr||jkr~t||jz|z }|dz}t d|z|}t |d}||_|j jdk\r|j jd|d|dn||jkDrr|dk\rmt||jz|z }t d|zd}||_|j jdk\r%|j jd|d|dn|}||k7r|j|_|S) rQrrr zBatch computation too fast (zs.) Setting batch_size=.zBatch computation too slow () rrMIN_IDEAL_BATCH_DURATIONintminrdrFverbose_printMAX_IDEAL_BATCH_DURATIONr)rold_batch_sizebatch_durationideal_batch_sizerUs rrRz$AutoBatchingMixin.compute_batch_sizesw3366 A .43P3P"P #!>!>>O    ! Q/1ABJZ+J)3D &}}$$* $$2>2BC**4Q8d;; ;RS@S #!>!>>O  Q!1115J)3D &}}$$* $$2>2BC**4Q8 (J  ' -1,Q,QD )rc||jk(r1|j}||jk(r|}n d|zd|zz}||_yy)rTg?rN)rrr)rrUrV old_duration new_durations rrWz!AutoBatchingMixin.batch_completedsT 33 3 88LtDDD (  #\1C(NB ,8D ) 4rcH|j|_|j|_y)zgReset batch statistics to default values. This avoids interferences with future jobs. N)rrrrrs rreset_batch_statsz#AutoBatchingMixin.reset_batch_statss &*%G%G"(,(M(M%r) rzr{r|r}rrrrrrRrWrrrs@rrrms;9 # !%&!'*$N 9v9 Nrrc*eZdZdZdZdZdZddZdZy)r`aA 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 |j|}|dk(rtt|j||_||_|Sz?Build a process or thread pool and return the number of workersrr])r0FallbackToBackendr_r rF_n_jobs)rr/rFr"s rrIzThreadingBackend.configuresC&&v. Q;#$5DDVDV$WX X    rcf|jt|j|_|jS)zLazily initialize the thread pool The actual pool of worker threads is only initialized at the first call to apply_async. )rr rrs rrzThreadingBackend._get_pools( :: #DLL1DJzzr)rN) rzr{r|r}rrrrIrr-rrr`r`s$ "&Lrr`cDeZdZdZdZdZfdZ ddZfdZxZ S)MultiprocessingBackendaA 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. TFctytjjr,|dk7r&trd}nd}t j |dyt jdkDr|dk7rt j ddy|js,|jdk(s|dk7rt j ddytt|3|S) 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. rzInside 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 stacklevelrzTMultiprocessing-backed parallel loops cannot be nested, below loky, setting n_jobs=1zVMultiprocessing-backed parallel loops cannot be nested below threads, setting n_jobs=1) rcurrent_processdaemoninside_dask_workerr9r:r _CURRENT_DEPTHrwr rrr0)rr/msgr#s rr0z'MultiprocessingBackend.effective_n_jobs s :    & &{%'== ca0  * *Q .{ 4  %%'4+=+=+B{ 7  +TCFKKrc |j|}|dk(rtt|ji|j|}t j t|fi||_||_ |Sr) r0rr_r r"gccollectrrrF)rr/rFrGrHmemmapping_pool_kwargss rrIz MultiprocessingBackend.configure>sy&&v. Q;#$5DDVDV$WX X" !!" $"  #FE.DE    rcJtt| |jyr)rrrOr)rr#s rrOz MultiprocessingBackend.terminateVs $d57  rrx) rzr{r|r}rrr0rIrOrrs@rrrs:"& %2Ll 0!!rrcLeZdZdZdZdZ d dZdZd dZdZ dZ d d Z y) LokyBackendz>Managing pool of workers with loky instead of multiprocessing.TNc V|j|}|dk(rtt|ji|j|}d|vr t d||jj dd}t|f||j||jd||_ ||_ |S) z9Build a process executor and return the number of workersrr]rBztThe 'timeout' parameter is not supported by the LokyBackend. Please use the `idle_worker_timeout` parameter instead.idle_worker_timeouti,)r/)rBrl context_id) r0rr_r r"rrCr ro_id_workersrF)rr/rFrGrHrmemmapping_executor_kwargss rrIzLokyBackend.configurebs&&v. Q;#$5DDVDV$WX X& !!& (& " 2 2J   &"&"5"5"9"9:OQT"U /  '(((7||  )   !  rc|dk(r tdt|ytjjr,|dk7r&t rd}nd}t j |dy|js,|jdk(s|dk7rt j ddy|dkrttdz|zd}|S) rrrrzInside 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=1rrzKLoky-backed parallel loops cannot be nested below threads, setting n_jobs=1) rrrrrr9r:rwr rdr )rr/rs rr0zLokyBackend.effective_n_jobss Q;EF F Z6>    ! ( ({%'== ca0%%'4+=+=+B{ 0   aZq6115F rcb|jj|}||j||S)r)rr<add_done_callback)rr5r6futures rr<zLokyBackend.submits/%%d+    $ $X . rcT |jS#t$r tdwxYw)z;Retrieve the result, here out is the future given by submitaThe 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.)rr r)rrs rr@z$LokyBackend.retrieve_result_callbacks5 ==? "$ O  s'c|jB|jjj|jjdd|_|j y)NF)rforce)r_temp_folder_manager_clean_temporary_resourcesrFrrrs rrOzLokyBackend.terminatesP == $ MM . . I I==,,E J !DM  rc|jjdd|_|r2|j|jj|jyy)zCShutdown the workers and restart a new one with the same parametersT) kill_workersNr)rrOrIrFr/rYs rr[zLokyBackend.abort_everythingsD T2  NN$--"6"6N O r)rNNNNrry) rzr{r|r}rrrIr0r<r@rOr[r-rrrr\sCH!%%)" %N(T  !PrrceZdZdZdZy)rz *   s  ,,)$r}rrrfrtr9abcrr_multiprocessing_helpersr_utilsrrmultiprocessing.poolr executorr externals.lokyr r externals.loky.process_executorr poolrrr_objectrrr`rr Exceptionrrr-rrrs '( >/1<F$pMGpMf$+$D8v8vgNgNT%')<%PZ!-/@BUZ!z}P#%8}P@ r