o 2wi @sddlZddlZddlZddlmZmZddlmZddlm Z m Z m Z m Z m Z mZddlmZdddd d e d e d ed edede f ddZGdddejZGdddZdS)N)ProcessPoolExecutorThreadPoolExecutor)partial)CallableDict GeneratorIterableOptionalType)tqdmiFnum_jobs queue_sizethreadsfn iterablesrrrreturncgst|g|R|||d}||j}|s|s>> for root in parallel_map(math.sqrt, range(1000), num_jobs=4): ... print(root) :param fn: function/callable to execute on each element. :param iterables: one of more iterables (one for each parameter of ``fn``). :param num_jobs: the number of parallel jobs. :param queue_size: max number of result items stored in memory. Decreasing this number might save more memory when the downstream processing is slower than the producer jobs. :param threads: whether to use threads instead of processes for producers (false by default). :return: a generator over results from ``fn`` applied to each item of ``iterables``. r Tg?)blocktimeoutN) SubmitterThreadstartqueueis_aliveemptygetresultEmptyjoin)rrrrrthreadqr!L/home/ubuntu/sommelier/.venv/lib/python3.10/site-packages/lhotse/parallel.py parallel_map s$ r#c sFeZdZdddddedededed d f fd d Zdd dZZS)rr i'Fr rrrrrNcs4t||_||_||_tj|d|_||_dS)N)maxsize)super__init__rrrrQueue use_threads)selfrrrrr __class__r!r"r&;s  zSubmitterThread.__init__cCsr|jrtnt}||j#}t|jD]}|j|jg|R}|jj |ddqWddS1s2wYdS)NT)r) r(rrrziprsubmitrrput)r)executorexargsfuturer!r!r"runJs "zSubmitterThread.run)rN) __name__ __module__ __qualname__rintboolr&r3 __classcell__r!r!r*r"r:s rc @sveZdZUdZiZeeeefe d<   ddegefdede d e fd d Z d d Z ddZdedefddZdS)ParallelExecutora A class which uses ProcessPoolExecutor to parallelize the execution of a callable class. The instances of the runner class are instantiated separately in each worker process. Example:: >>> class MyRunner: ... def __init__(self): ... pass ... def __call__(self, x): ... return f'processed: {x}' ... >>> runner = ParallelExecutor(MyRunner, num_jobs=4) >>> for item in runner(range(10)): ... print(item) If the __init__ method of the callable class accepts parameters except for `self`, use `functools.partial` or similar method to obtain a proper initialization function: >>> class MyRunner: ... def __init__(self, name): ... self.name = name ... def __call__(self, x): ... return f'{self.name}: {x}' ... >>> runner = ParallelExecutor(partial(MyRunner, name='my_name'), num_jobs=4) >>> for item in runner(range(10)): ... print(item) The initialization function will be called separately for each worker process. Steps like loading a PyTorch model instance to the selected device should be done inside the initialization function. _runnersr F Processinginit_fnrverbose descriptioncCs||_||_||_||_dS)a Instantiate a parallel executor. :param init_fn: A function which returns a runner object (e.g. a class) that will be instantiated in each worker process. :param num_jobs: The number of parallel jobs to run. Defaults to 1 (no parallelism). :param verbose: Whether to show a progress bar. :param description: The description to show in the progress bar. N) _make_runnerrr>r?)r)r=rr>r?r!r!r"r&ws zParallelExecutor.__init__cCstj}||j|<dSN)multiprocessingcurrent_processpidr@r;)r)rDr!r!r" _init_runners zParallelExecutor._init_runnercOs"tj}|j|}||i|SrA)rBrCrDr;)r)r1kwargsrDrunnerr!r!r"_processs  zParallelExecutor._processitemsrc ksJ|jdkr"|}t||j|j dD] }||fi|VqdSt|j|jtdd}|k}z[z"| t |j fi||}t||jt ||j dD]}|VqMWn1t yo}z|jdd|jritd|d}~wty}z |jddtd |d}~wwW|jn|jwWddS1swYdS) Nr )descdisablespawn) max_workers initializer mp_context)rJtotalrKF)waitzInterrupted by the user.z5Parallel processing failed. Please report this issue.)rr@r r?r>rrErB get_contextmaprrHlenKeyboardInterruptshutdownprint Exception RuntimeErrorr;clear) r)rIrFrGitempoolr/resexcr!r!r"__call__sP    "zParallelExecutor.__call__N)r Fr<)r4r5r6__doc__r;rr r7r__annotations__r8strr&rErHrrr_r!r!r!r"r:Rs$ "  r:)rBr threadingconcurrent.futuresrr functoolsrtypingrrrrr r tqdm.autor r7r8r#Threadrr:r!r!r!r"s2    /