o %ݫi8%@sdZddlZddlZddlmZddlmZmZddlm Z ddl m Z m Z m Z mZddlmZdd ZGd d d ZGd d d Zeddddddifde e ge fde e dedededeededefddZdS)zuParallel processing tools to help speed up certain tasks like data preprocessing. Authors * Sylvain de Langen 2023 N)deque)ExecutorProcessPoolExecutor) Condition)AnyCallableIterableOptional)tqdmcCstt||SN)listmap)fnchunkrN/home/ubuntu/.local/lib/python3.10/site-packages/speechbrain/utils/parallel.py_chunk_process_wrappersrc@s(eZdZdZddZddZddZdS) CancelFuturesOnExitzContext manager that .cancel()s all elements of a list upon exit. This is used to abort futures faster when raising an exception.cCs ||_dSr ) future_list)selfrrrr__init__s zCancelFuturesOnExit.__init__cCsdSr r)rrrr __enter__szCancelFuturesOnExit.__enter__cCs|jD]}|qdSr )rcancel)r_type_value _tracebackfuturerrr__exit__ s  zCancelFuturesOnExit.__exit__N)__name__ __module__ __qualname____doc__rrrrrrrrs  rc@sleZdZdZdeegefdeededededee de d e fd d Z d d Z ddZddZddZdS)_ParallelMapperzEInternal class for `parallel_map`, arguments match the constructor's.rsource process_count chunk_size queue_sizeexecutor progress_barprogress_bar_kwargsc Cst|_t|_d|_ d|_ ||_||_||_||_ ||_ ||_ t |dr+t |nd|_t||_d|_|rLd|ji} | |tdi| |_dSd|_dS)Nr__len__Ftotalr)r future_chunksrcvjust_finished_countremote_exceptionrr#r$r%r&r'hasattrlen known_leniter source_itdepleted_sourceupdater pbar) rrr#r$r%r&r'r(r)tqdm_final_kwargsrrrr(s(     z_ParallelMapper.__init__c cst|jF|jdur|EdHn(t|jd}||_|EdHWdn1s/wYWddSWddSWddS1sOwYdS)z`Spins up an executor (if none were provided), then yields all processed chunks in order.N) max_workers)rr,r'_map_allrr$)rpoolrrrrunNs   "z_ParallelMapper.runcCs|rdS|}|j|dur||_|jd7_|jWdn1s+wY|durE|jdurG|jt| dSdSdS)aoNotifies the main thread of the finished job, bumping the number of jobs it should requeue. Updates the progress bar based on the returned chunk length. Arguments --------- future: concurrent.futures.Future A future holding a processed chunk (of type `list`). Returns ------- None N) cancelled exceptionr-r/r.notifyr7r6r1result)rrfuture_exceptionrrr_bump_processed_count^s  z%_ParallelMapper._bump_processed_countcCsXtt|j|j}t|dkrd|_dS|jt |j |}| |j |j |dS)a Pulls a chunk from the source iterable and submits it to the pool; must be run from the main thread. Returns ------- `True` if any job was submitted (that is, if there was any chunk left to process), `False` otherwise. rTF)r itertoolsislicer4r%r1r5r'submitrradd_done_callbackrCr,append)rrrrrr _enqueue_jobs   z_ParallelMapper._enqueue_jobccs(t|jD]}|snq|jrt|jdkr|j%|jdkr,|j|jdks"|j dur4|j |j}d|_Wdn1sDwYt|D]}|sUnqMt|jdkr||jd r||j EdHt|jdkr||jd sd|jrt|jdks|j dur|j dSdS)z~Performs all the parallel mapping logic. Yields ------ The items from source processed by fn rN)ranger&rIr5r1r,r-r.waitr/donepopleftrAr7close)r_to_queue_countrrrr:s4      z_ParallelMapper._map_allN)rrr r!rrrintr rbooldictrr<rCrIr:rrrrr"%s.  &# r"T smoothingg{Gz?rr#r$r%r&r'r(r)c cs*t||||||||}|EdHdS)aMaps iterable items with a function, processing chunks of items in parallel with multiple processes and displaying progress with tqdm. Processed elements will always be returned in the original, correct order. Unlike `ProcessPoolExecutor.map`, elements are produced AND consumed lazily. Arguments --------- fn: Callable The function that is called for every element in the source list. The output is an iterator over the source list after fn(elem) is called. source: Iterable Iterator whose elements are passed through the mapping function. process_count: int The number of processes to spawn. Ignored if a custom executor is provided. For CPU-bound tasks, it is generally not useful to exceed logical core count. For IO-bound tasks, it may make sense to as to limit the amount of time spent in iowait. chunk_size: int How many elements are fed to the worker processes at once. A value of 8 is generally fine. Low values may increase overhead and reduce CPU occupancy. queue_size: int Number of chunks to be waited for on the main process at a time. Low values increase the chance of the queue being starved, forcing workers to idle. Very high values may cause high memory usage, especially if the source iterable yields large objects. executor: Optional[Executor] Allows providing an existing executor (preferably a ProcessPoolExecutor). If None (the default), a process pool will be spawned for this mapping task and will be shut down after. progress_bar: bool Whether to show a tqdm progress bar. progress_bar_kwargs: dict A dict of keyword arguments that is forwarded to tqdm when `progress_bar == True`. Allows overriding the defaults or e.g. specifying `total` when it cannot be inferred from the source iterable. Yields ------ The items from source processed by fn N)r"r<) rr#r$r%r&r'r(r)mapperrrr parallel_maps> rX)r!rDmultiprocessing collectionsrconcurrent.futuresrr threadingrtypingrrrr tqdm.autor rrr" cpu_countrQrRrSrXrrrrsF   )