o ̳iT@sddlZddlZddlZddlmZmZmZmZmZm Z m Z m Z m Z m Z mZmZddlmZddlmZmZddlmZmZddlmZmZddlmZmZddlmZdd l m Z dd l!m"Z"d Z#Gd d d e Z$edZ%dee%dee%gefddfddZ&e e%Z'e eZ(Gdddee%efZ)deej*ej*fdej*dej+fddZ,GdddeeZ-GdddeeZ.Gd d!d!eeZ/Gd"d#d#eeZ0eeej*ee1ej2ej+gdfZ3Gd$d%d%eeZ4dS)&N) AnyCallableDictGenericIteratorListLiteralOptionalProtocolSequenceTypeVarUnion)BaseNodeT)Batcher Unbatcher)ExceptionWrapperStartupExceptionWrapper)QueueSnapshotStore SnapshotStore) _apply_udf)_populate_queue) QUEUE_TIMEOUTi,c@s$eZdZddZddZddZdS)_MultiprocessContextcOdSNselfargskwargsrrG/home/ubuntu/.local/lib/python3.10/site-packages/torchdata/nodes/map.pyProcessz_MultiprocessContext.ProcesscOrrrrrrr"Event r$z_MultiprocessContext.EventcOrrrrrrr"Queue#r$z_MultiprocessContext.QueueN)__name__ __module__ __qualname__r#r%r&rrrr"rs rXsourcemap_fnreturnzParallelMapper[T]cCst||ddS)aReturns a :class:`ParallelMapper` node with num_workers=0, which will execute map_fn in the current process/thread. Args: source (BaseNode[X]): The source node to map over. map_fn (Callable[[X], T]): The function to apply to each item from the source node. r)r+r, num_workers)ParallelMapper)r+r,rrr"Mapper*s r0c@s>eZdZdeegeffddZdeedeefddZdS) MapOverBatchr,cCs ||_dSrr,)rr,rrr"__init__= zMapOverBatch.__init__xlistr-csfdd|DS)Ncsg|]}|qSrr2).0xrrr" Asz)MapOverBatch.__call__..r)rr5rr8r"__call__@zMapOverBatch.__call__N) r'r(r)rr*rr3r r:rrrr"r1<sr1in_qout_q stop_eventcCsi}d}|sz |jdtd\}}Wn tjyYqw||kr0|j||fdd|d7}n1||vr]ztd|d|d |tyQt d d }Ynw|j||fdddS|||<||vry|j| ||fdd|d7}||vse|rdSdS) NrTblocktimeoutF)r@rzDuplicate index idx=z, buffer.keys()=z, item=zin _sort_worker)where) is_setgetrqueueEmptyput ValueErrorkeys Exceptionrpop)r<r=r>buffercur_idxitemidxrrr" _sort_workerDs2  rPc @sneZdZdZdZ ddeedeegefde e e e ffddZ dd Zd e e e ffd d Zd dZdS)_InlineMapperIterz%Non-Parallel implementation of Mapperr+Nr, initial_statecCs8||_||_|dur|j||jdS|jdSr)r+r,reset SOURCE_KEY)rr+r,rRrrr"r3cs z_InlineMapperIter.__init__cCs|t|jSr)r,nextr+r8rrr"__next__pz_InlineMapperIter.__next__r-cC|j|jiSr)rTr+ state_dictr8rrr" get_statesrWz_InlineMapperIter.get_statecCrrrr8rrr" _shutdownvr$z_InlineMapperIter._shutdownr)r'r(r)__doc__rTrr*rrr rstrrr3rVrZr[rrrr"rQ^s    rQc@seZdZdZdeedeegefdede de dde d e ed ed e e eeffd d ZdeefddZdefddZde eeffddZdefddZddZddZdS)_ParallelMapperItera_ParallelMapperIter will start at least two threads, one running _populate_queue, and one for _apply_udf. If in_order == True, a third thread will be started to read from _apply_udf's result q and block the output_q until the appropriate in_order element is available, buffering outputs as needed. A BoundedSemaphore with initial value max_concurrent will limit the number of items in flight, and in all of the queues. r+r,r.in_ordermethodthreadprocess mp_contextmax_concurrentsnapshot_frequencyrRc  CsT||_||_||_||_||_||_||_|dkrtn||_ |dkr*tn||_ |dur8d|jn||_ t j |j d|_d|_t |_||_d|_d} | durj| d|_| d} |j|jnd|_|jt|_t jt|j|j |j|j|j|jfdd |_g|_t|jD].} | |j |j |j|jdkr|jn|jf} |j|jdkrt jt| dd n|jt| dd qt|_ t jt!|j |j |jfdd |_"|j |_#|jr|j |_#|j$|jD]} | $q|jr|j"$t%&d |jj'|jt(d |_t| D]}zt)|Wq t*y't+d | d |dwdS)NrbvalueFrsnapshotsteps_since_snapshotTtargetr daemong{Gz?rbrATried to fast-forward / items during init but hit StopIteration after 4 items, this is likely a bug or malformed state_dict),r+r,r.r_r`rdrfrEr&_in_q _intermed_q _max_tasks threadingBoundedSemaphore_sem_doner%_stop_mp_stop_steps_since_snapshot _snapshotrSr_snapshot_storeThreadr _read_thread_workersrangeappendrr#_sort_qrP _sort_thread_out_qstarttimesleepget_initial_snapshot ACK_TIMEOUTrU StopIterationrH)rr+r,r.r_r`rdrerfrR fast_forward worker_idr tirrr"r3s              z_ParallelMapperIter.__init__r-cC|Srrr8rrr"__iter__r$z_ParallelMapperIter.__iter__cCs |jr t|jr |jj|jkr |j|jtz |j j dt d\}}Wn t j y6Yqwt|trEd|_|jqt|trXt|tsT|j||jd7_|j|||SNTr?r)rzrCrryrx_valuerusetr{rrDrrErF isinstancereleaserrreraiser|_maybe_update_snapshotrrNrOrrr"rVs0          z_ParallelMapperIter.__next__cC|j|jdSN)rjrkr}r|r8rrr"rZz_ParallelMapperIter.get_staterOcC(|j|}dur||_d|_dSdSNrr~ pop_versionr}r|rrOrjrrr"r  z*_ParallelMapperIter._maybe_update_snapshotcC |dSrr[r8rrr"__del__ z_ParallelMapperIter.__del__cCs|j|jt|dr|jr|jjtddt|dr0|jr0|jjtddt|drG|j D]}|rF|jtddq8dSdS)NrrArr) rzrr{hasattrris_alivejoinrrr)rrrrr"r[s    z_ParallelMapperIter._shutdownN)r'r(r)r\rr*rrintboolrrr rr]rr3rrrVrZrrr[rrrr"r^zs6    ` r^cseZdZdZ     ddeedeegefded e d e d d e e d e edeffdd Z dde ee efffdd Zde ee effddZde ee effddZdefddZdee effddZZS)_ParallelMapperImplaLThis class implements _ParallelMapperIter and _InlineMapperIter as a BaseNode, allowing them to be composed with other BaseNodes. TODO: In the future, this class may go away once we implement reset() on _ParallelMapperIter and _InlineMapperIter themselves so we don't need this additional level of abstraction. TrbNrr+r,r.r_r`ramultiprocessing_contextrerfc st|dvs J||_||_||_||_||_||_t|_ |jdkr1|jdur1t |j|_ |durM|dkrMt |t rM||krMt d|d|d||_||_d|_dS)Nrarcrmax_concurrent= should be <= num_workers=!)superr3r+r,r.r_r`rmp _mp_context get_contextrrrHrerf_it) rr+r,r.r_r`rrerf __class__rr"r3)s"  z_ParallelMapperImpl.__init__rRcsDt||jdur |`|jdkr|||_dS|||_dSr)rrSrr._parallel_reset _inline_resetrrRrrr"rSGs   z_ParallelMapperImpl.resetcCst|j|j|dS)N)r+r,rR)rQr+r,rrrr"rQr;z!_ParallelMapperImpl._inline_resetc Cs*t|j|j|j|j|j|j|j|j|d S)N) r+r,r.r_r`rdrerfrR) r^r+r,r.r_r`rrerfrrrr"rTsz#_ParallelMapperImpl._parallel_resetr-cC t|jSrrUrr8rrr"rUar4z_ParallelMapperImpl.nextcCs |jSr)rrZr8rrr"rZdr4z_ParallelMapperImpl.get_state)TrbNNrr)r'r(r)r\rr*rrrrrr r]r3rrrSrrrUrZ __classcell__rrrr"r s:      rcseZdZdZdZ      ddeedeegefd e d e d e d d e e de e de de e ffdd Zdde ee efffdd ZdefddZdee effddZZS)r/aParallelMapper executes map_fn in parallel either in num_workers threads or processes. For processes, multiprocessing_context can be spawn, forkserver, fork, or None (chooses OS default). At most max_concurrent items will be either processed or in the iterator's output queue, to limit CPU and Memory utilization. If None (default) the value will be 2 * num_workers. At most one iter() is created from source, and at most one thread will call next() on it at once. If in_order is true, the iterator will return items in the order from which they arrive from source's iterator, potentially blocking even if other items are available. Args: source (BaseNode[X]): The source node to map over. map_fn (Callable[[X], T]): The function to apply to each item from the source node. num_workers (int): The number of workers to use for parallel processing. in_order (bool): Whether to return items in the order from which they arrive from. Default is True. method (Literal["thread", "process"]): The method to use for parallel processing. Default is "thread". multiprocessing_context (Optional[str]): The multiprocessing context to use for parallel processing. Default is None. max_concurrent (Optional[int]): The maximum number of items to process at once. Default is None. snapshot_frequency (int): The frequency at which to snapshot the state of the source node. Default is 1. prebatch (Optional[int]): Optionally perform pre-batching of items from source before mapping. For small items, this may improve throughput at the expense of peak memory. it_stateTrbNrr+r,r.r_r`rarrerfprebatchc s t|dvs J||_||_||_||_|dur3|dkr3t|tr3||kr3td|d|d||_ ||_ | |_ | durG||_ ||_ n| dkrStd| dt|d|_ t|| d d |_ t|j |j |j|j|j|j|j |j d } |j dur| |_dSt| |_dS) Nrarrrrz prebatch=z must be a positive integer!r2F) batch_size drop_last)r+r,r.r_r`rrerf)rr3r.r_r`rrrrHrerfrr,r+r1rrrr) rr+r,r.r_r`rrerfrrrrr"r3s@   zParallelMapper.__init__rRcs8t||dur|j||jdS|jdSr)rrSr IT_STATE_KEYrrrr"rSs zParallelMapper.resetr-cCrrrr8rrr"rUr4zParallelMapper.nextcCrXr)rrrYr8rrr"rZrWzParallelMapper.get_state)TrbNNrNr)r'r(r)r\rrr*rrrrrr r]r3rrrSrUrZrrrrr"r/hs>    1r/c @seZdZdZdeededededee e e ff ddZ d e efd d Zd efd d Zd e e e ffddZdefddZddZddZdS)_SingleThreadedMapperaUtility Iterator for performing mapping with a single thread. Because only a single thread is used, we don't need an input queue to guard against multiple threads reading from the same iterator. This is used for Prefetcher and PinMemory. A thread is started on __init__ and stopped on __del__/_shutdown. The thread runs _populate_queue, which acquires a BoundedSemaphore with initial value of `prefetch_factor`. When next() is called on this iterator, it will block until an item is available on _q. Next will perform the following depending on what is pulled from the q: - StopIteration: raise StopIteration. Any subsequent next() calls will also raise StopIteration - ExceptionWrapper: call reraise() on the exception wraper - any other item: return the item A Bounded semaphore is used to limit concurrency and memory utilization. If N items have been pulled from the source, and M items have been yielded by this iterator, we maintain the invariant that semaphore.value + (N - M) == prefetch_factor (modulo non-atomicness of operations). _populate_queue calls semaphore.acquire. When we pull an item from the queue, we call semaphore.release (unless it's a StartupExceptionWrapper, because _populate_queue does not acquire sempahores in this case). All outstanding items are either being processed in _populate_queue, in the _q, or about to be returned by an in-flight next() call. r+prefetch_factorworkerrfrRc Cs ||_||_||_||_t|_tj|d|_ t |_ d|_ d|_ |dur9|d|_|d|_ |j|jnd|_|jt|_tj|j|j|j|j|j|j |j fdd|_|j|jj|jtd|_t|j D]}zt|Wqotytd|j d |d wd|_ dS) NrhrrjrkTrlrorprqrr)r+rrrfrEr&_qrvrwrxr% _stop_eventr| _fast_forwardr}rSrr~r_threadrrrrrUrrH)rr+rrrfrRrrrr"r3sP          z_SingleThreadedMapper.__init__r-cCrrrr8rrr"r!r$z_SingleThreadedMapper.__iter__cCs |jr tz |jjdtd\}}Wn tjyYqwt|tr1|j |j |t|t rJt|t s@|j |j |n|j |jd7_|||Sqr)rrCrrrDrrErFrrxrrrrrr|rrrrr"rV$s,           z_SingleThreadedMapper.__next__cCrrrr8rrr"rZ=rz_SingleThreadedMapper.get_staterOcCrrrrrrr"rCrz,_SingleThreadedMapper._maybe_update_snapshotcCrrrr8rrr"rHrz_SingleThreadedMapper.__del__cCs<|jt|dr|jr|jjtdddSdSdS)Nrrr)rrrrrrrr8rrr"r[Ks z_SingleThreadedMapper._shutdownN)r'r(r)r\rrr _WorkerTyper rr]rr3rrrVrZrrr[rrrr"rs& 6 r)5rErvrtypingrrrrrrrr r r r r torch.multiprocessingmultiprocessingrtorchdata.nodes.base_noderrtorchdata.nodes.batchrr!torchdata.nodes.exception_wrapperrrtorchdata.nodes.snapshot_storerrrr constantsrrrr*r0XseqTseqr1r&r%rPrQr^rr/rrwrrrrrr"sJ8     $&'H[