U Nv fA( @sddlZddlZddlZddlZddlmZmZddlm Z ddl m Z dgZ e ZdadadaddZdd dZGdddeZdS)N)ProcessPoolExecutorEXTRA_QUEUED_CALLS) cpu_count) get_contextget_reusable_executorc Cs,tt}td7a|W5QRSQRXdS)zEnsure that each successive executor instance has a unique, monotonic id. The purpose of this monotonic id is to help debug and test automated instance creation. rN)_executor_lock_next_executor_id) executor_idr K/tmp/pip-unpacked-wheel-dylwa62s/joblib/externals/loky/reusable_executor.py_get_next_executor_idsr Fautor c Cs&tj|||||||||| d \} } | S)aReturn the current ReusableExectutor instance. Start a new instance if it has not been started already or if the previous instance was left in a broken state. If the previous instance does not have the requested number of workers, the executor is dynamically resized to adjust the number of workers prior to returning. Reusing a singleton instance spares the overhead of starting new worker processes and importing common python packages each time. ``max_workers`` controls the maximum number of tasks that can be running in parallel in worker processes. By default this is set to the number of CPUs on the host. Setting ``timeout`` (in seconds) makes idle workers automatically shutdown so as to release system resources. New workers are respawn upon submission of new tasks so that ``max_workers`` are available to accept the newly submitted tasks. Setting ``timeout`` to around 100 times the time required to spawn new processes and import packages in them (on the order of 100ms) ensures that the overhead of spawning workers is negligible. Setting ``kill_workers=True`` makes it possible to forcibly interrupt previously spawned jobs to get a new instance of the reusable executor with new constructor argument values. The ``job_reducers`` and ``result_reducers`` are used to customize the pickling of tasks and results send to the executor. When provided, the ``initializer`` is run first in newly spawned processes with argument ``initargs``. The environment variable in the child process are a copy of the values in the main process. One can provide a dict ``{ENV: VAL}`` where ``ENV`` and ``VAL`` are string literals to overwrite the environment variable ``ENV`` in the child processes to value ``VAL``. The environment variables are set in the children before any module is loaded. This only works with the ``loky`` context. ) max_workerscontexttimeout kill_workersreuse job_reducersresult_reducers initializerinitargsenv)_ReusablePoolExecutorr) rrrrrrrrrr _executor_r r r r%s4 c sTeZdZdfdd Zedd d Zfd d Zd dZddZfddZ Z S)rNrr c s,tj||||||| | d||_||_dS)N)rrrrrrrr)super__init__r _submit_resize_lock) selfZsubmit_resize_lockrrrr rrrrr __class__r r ris  z_ReusablePoolExecutor.__init__rFrc  Cstt} |dkr4|dkr,| dk r,| j}qLt}n|dkrLtd|dt|tr^t|}|dk rz|dkrztdt |||||| | d} | dkrd} t j d |dt }| a|tf||d | a} n|d kr| tk}| jjs| jjs|st| jjr d }n| jjrd }nd}t j d|d|d| jd|dda} a|jfd|i| W5QRSt j d| jdd} | |W5QRX| | fS)NTrz(max_workers must be greater than 0, got .forkz4Cannot use reusable executor with the 'fork' context)rrrrrrrFz#Create a executor with max_workers=)rr rbrokenshutdownzarguments have changedz)Creating a new executor with max_workers=z, as the previous instance cannot be reused (z).)waitrrz+Reusing existing executor with max_workers=)rr _max_workersr ValueError isinstancestrrZget_start_methoddictmputildebugr _executor_kwargs_flagsr%r&r_resize)clsrrrrrrrrrrexecutorkwargsZ is_reusedr reasonr r r rs         z+_ReusablePoolExecutor.get_reusable_executorc s2|j"tj|f||W5QRSQRXdSN)rrsubmit)r fnargsr5r!r r r8sz_ReusablePoolExecutor.submitc Cs|j|dkrtdn||jkr4W5QRdS|jdkrR||_W5QRdS||jHt|j}t dd|D}||_t ||D]}|j dqW5QRXt |j|kr|jjstdq|t|j}tdd|DstdqW5QRXdS)Nz&Trying to resize with max_workers=Nonecss|]}|VqdSr7is_alive.0pr r r sz0_ReusablePoolExecutor._resize..MbP?css|]}|VqdSr7r;r=r r r r@s)rr)r(Z_executor_manager_thread_wait_job_completionZ_processes_management_locklistZ _processesvaluessumrangeZ _call_queueputlenr1r%timesleepZ_adjust_process_countall)r rZ processesZnb_children_aliverr r r r2s0      z_ReusablePoolExecutor._resizecCs>|jr(tdttjd|jd|jr:t dq(dS)z8Wait for the cache to be empty before resizing the pool.z\Trying to resize an executor with running jobs: waiting for jobs completion before resizing.z Executor z, waiting for jobs completion before resizingrAN) Z_pending_work_itemswarningswarn UserWarningr-r.r/r rIrJ)r r r r rBs z*_ReusablePoolExecutor._wait_job_completioncs$dtt}tj|||ddS)N) queue_size)rrr _setup_queues)r rrrPr!r r rQs z#_ReusablePoolExecutor._setup_queues) NNNrNNNr N) NNrFrNNNr N) __name__ __module__ __qualname__r classmethodrr8r2rBrQ __classcell__r r r!r rhs4 [ #r) NNrFrNNNr N)rIrL threadingmultiprocessingr-Zprocess_executorrrZbackend.contextrbackendr__all__RLockrr rr0r rrr r r r s0   C