o iD @srdZddlZddlZddlZddlmZddlmZe e Z GdddeZGddde Z ejd d d ZdS) z\Common functionality for concurrent processing. The main entry point is :func:`create_pool`.N)deque)ThreadPoolExecutorc@s eZdZdZdddddZdS)rz+Subclass with a lazy consuming imap method.N)timeoutqueued_tasks_per_workerc 'st|j|d}}|j|j|j}}fdd} t|D]} |||g| Rt||kr8| Vq"|rC| V|s;dSdS)aOrdered imap that consumes iterables just-in-time. References: https://gist.github.com/ddelange/c98b05437f80e4b16bf4fc20fde9c999 Args: fn: Function to apply. iterables: One (or more) iterable(s) to pass to fn (using zip) as positional argument(s). timeout: Per-future result retrieval timeout in seconds. queued_tasks_per_worker: Amount of additional items per worker to fetch from iterables to fill the queue: this determines the total queue size. Setting 0 will result in a true just-in-time behaviour: when a worker finishes a task, it waits until a result is consumed from the imap generator, at which point next() is called on the input iterable(s) and a new task is submitted. Default 2 ensures there is always some work to pick up. Note that at imap startup, the queue will fill up before the first yield occurs. Example: long_generator = itertools.count() with ThreadPoolExecutor(42) as pool: result_generator = pool.imap(fn, long_generator) for result in result_generator: print(result) cs S)z8Block until the next task is done and return the result.)resultpopleftrr J/home/ubuntu/.local/lib/python3.10/site-packages/smart_open/concurrency.pyget3s z$ThreadPoolExecutor.imap..getN)r _max_workersr appendsubmitziplen) selffnrr iterablesfuturesmaxlenrrr argsr r r imaps   zThreadPoolExecutor.imap)__name__ __module__ __qualname____doc__rr r r r rsrc@s(eZdZdZddZddZddZdS) ConcurrentFuturesPoolz_A class that mimics multiprocessing.pool.Pool but uses concurrent futures instead of processes.cCst|d|_dS)N max_workers)rexecutor)rr r r r __init__FszConcurrentFuturesPool.__init__c#s6fdd|D}tj|D]}|VqdS)Ncsg|] }j|qSr )r!r).0itemfunctionrr r Jsz8ConcurrentFuturesPool.imap_unordered..) concurrentr as_completedr)rr&itemsrfuturer r%r imap_unorderedIs  z$ConcurrentFuturesPool.imap_unorderedcCs|jjdddS)NT)wait)r!shutdown)rr r r terminateNszConcurrentFuturesPool.terminateN)rrrrr"r,r/r r r r rDs  rrccs*td|t|d}|V|dS)Nz0creating concurrent futures pool with %i workersr)loggerinforr/) processespoolr r r create_poolRs    r4)r)rconcurrent.futuresr( contextliblogging collectionsrr_ThreadPoolExecutor getLoggerrr0objectrcontextmanagerr4r r r r s   0