PiTddlZddlZddlZddlmZmZmZmZmZm Z m Z m Z m Z m Z mZmZddlmZddlmZmZddlmZmZddlmZmZddlmZmZddlmZdd l m Z dd l!m"Z"d Z#Gd d e Z$edZ%dee%dee%gefddfdZ&e e%Z'e eZ(Gddee%efZ)deej*ej*fdej*dej+fdZ,GddeeZ-GddeeZ.GddeeZ/Gd d!eeZ0eeej*ee1ej2ej+gdfZ3Gd"d#eeZ4dS)$N) AnyCallableDictGenericIteratorListLiteralOptionalProtocolSequenceTypeVarUnion)BaseNodeT)Batcher Unbatcher)ExceptionWrapperStartupExceptionWrapper)QueueSnapshotStore SnapshotStore) _apply_udf)_populate_queue) QUEUE_TIMEOUTi,c eZdZdZdZdZdS)_MultiprocessContextcdSNselfargskwargss g/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/torchdata/nodes/map.pyProcessz_MultiprocessContext.Process cdSrrr s r$Eventz_MultiprocessContext.Event r&r'cdSrrr s r$Queuez_MultiprocessContext.Queue#r&r'N)__name__ __module__ __qualname__r%r)r+rr'r$rrsA           r'rXsourcemap_fnreturnzParallelMapper[T]c&t||dS)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)r0r1 num_workers)ParallelMapper)r0r1s r$Mapperr6*s$    r'cTeZdZdeegeffdZdeedeefdZdS) MapOverBatchr1c||_dSrr1)r!r1s r$__init__zMapOverBatch.__init__=s  r'xlistr2c fd|DS)Nc:g|]}|Srr:).0xr!s r$ z)MapOverBatch.__call__..As#...1 A...r'r)r!r<s` r$__call__zMapOverBatch.__call__@s........r'N) r,r-r.rr/rr;r rBrr'r$r8r8<sdxQ//hqk/hqk//////r'r8in_qout_q stop_eventcli}d}|s |dt\}}n#tj$rYGwxYw||kr|||fd|dz }no||vrf t d|d|d |#t$rtd }YnwxYw|||fddS|||<||vr5|| ||fd|dz }||v5|dSdS) NrTblocktimeoutF)rHrzDuplicate index idx=z, buffer.keys()=z, item=zin _sort_worker)where) is_setgetrqueueEmptyput ValueErrorkeys Exceptionrpop)rCrDrEbuffercur_idxitemidxs r$ _sort_workerrXDsFG!! t]CCID##{    H  '>> IItWoUI 3 3 3 qLGGf}}E$%Y%Y%Y %Y%YRV%Y%YZZZ EEE+2CDDDDDDE 4+U 333F3K IIvzz'**G4EI B B B qLG%!!s!;A  A :*B$$CCc eZdZdZdZ d deedeegefde e e e ffdZ dZde e e ffd Zd ZdS) _InlineMapperIterz%Non-Parallel implementation of Mapperr0Nr1 initial_statec||_||_|'|j||jdS|jdSr)r0r1reset SOURCE_KEY)r!r0r1r[s r$r;z_InlineMapperIter.__init__csU    $ K  mDO< = = = = = K       r'cP|t|jSr)r1nextr0r!s r$__next__z_InlineMapperIter.__next__ps{{4 ,,---r'r2cB|j|jiSr)r^r0 state_dictras r$ get_statez_InlineMapperIter.get_statess!7!7!9!9::r'cdSrrras r$ _shutdownz_InlineMapperIter._shutdownvs r'r)r,r-r.__doc__r^rr/rrr rstrrr;rbrergrr'r$rZrZ^s//J 37      !a    S#X/     ...;4S>;;;;     r'rZceZdZdZdeedeegefdede de dde d e ed ed e e eeffd Zd eefdZd efdZd e eeffdZdefdZdZdZdS)_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. r0r1r4in_ordermethodthreadprocess mp_contextmax_concurrentsnapshot_frequencyr[c N||_||_||_||_||_||_||_|dkrtjn||_ |dkrtjn||_ | d|jzn||_ tj |j |_d|_tj|_||_d|_d} | 5| d|_| d} |j|jn d|_|jt-|_tjt2|j|j |j|j|j|jfd |_g|_t9|jD]} | |j |j |j|jdkr|jn|jf} |j|jdkrtjt<| d n|t<| d tj|_ tjtB|j |j |jfd |_"|j |_#|jr |j |_#|j$|jD]} | $|jr|j"$tKj&d |j'|jtP |_t9| D]6} tS|#tT$rtWd | d |dwxYwdS)NrovalueFrsnapshotsteps_since_snapshotTtargetr"daemong{Gz?rorITried to fast-forward / items during init but hit StopIteration after 4 items, this is likely a bug or malformed state_dict),r0r1r4rlrmrqrsrMr+_in_q _intermed_q _max_tasks threadingBoundedSemaphore_sem_doner)_stop_mp_stop_steps_since_snapshot _snapshotr]r_snapshot_storeThreadr _read_thread_workersrangeappendrr%_sort_qrX _sort_thread_out_qstarttimesleepget_initial_snapshot ACK_TIMEOUTr` StopIterationrP)r!r0r1r4rlrmrqrrrsr[ fast_forward worker_idr"tis r$r;z_ParallelMapperIter.__init__s  &   $"4DJhDVDV5;===\f\l\l\n\n JPT\J\J\blbrbrbtbt2@2H!d...n.T_EEE  _&& "((** %&"  $*:6DN()?@L K  dn - - - -!DN K     133%,"  $'      DF t/00  I   "kX55 4= D M ;(** dKKKK''zT'RR     %*KMM %,"DL$*=    & = ',DK !!!  A GGIIII = &   # # % % % 4-BB$J[epBqq|$$  A T      g\gg/0ggg   s 1N!N"r2c|Srrras r$__iter__z_ParallelMapperIter.__iter__ r'c |jrt|jrU|jj|jkr@|j|jt |j dt\}}n#tj $rYwxYwt|tr!d|_|jt|t rBt|t"s|j||xjdz c_|j|||SNTrGr)rrKrrr_valuersetrrrLrrMrN isinstancereleaserrreraiser_maybe_update_snapshotr!rVrWs r$rbz_ParallelMapperIter.__next__sp z  "" &#oo% & 0DO C C     !!####oo%  KOO$ ONN cc;    $ .. !  !!###D"233 !$(?@@(I%%'''   & &! + & & I       ' ' , , ,Ks$B++B=<B=c |j|jdSN)rxryrrras r$rez_ParallelMapperIter.get_state$($>   r'rWc^|j|x}||_d|_dSdSNrr pop_versionrrr!rWrxs r$rz*_ParallelMapperIter._maybe_update_snapshot <,88== =H J%DN)*D & & & K Jr'c.|dSrrgras r$__del__z_ParallelMapperIter.__del__ r'c6|j|jt|dr<|jr#|jtdzt|dr<|jr#|jtdzt|dr<|j D]6}|r|tdz5dSdS)NrrIrr) rrrhasattrris_alivejoinrrr)r!rs r$rgz_ParallelMapperIter._shutdowns   4 ( ( >T->-G-G-I-I >   " "=1+< " = = = 4 ( ( >T->-G-G-I-I >   " "=1+< " = = = 4 $ $ 6] 6 6::<<6FF=1#4F555 6 6 6 6r'N)r,r-r.rhrr/rrintboolr rr rrirr;rrrbrerrrgrr'r$rkrkzsT^ ^!a ^ ^  ^ +, ^)^! ^ ^ S#X/^^^^@(1+!: 4S>    +#++++  6 6 6 6 6r'rkc8eZdZdZ ddeedeegefded e d e d d e e d e edeffd Z dde ee efffd Zde ee effdZde ee effdZdefdZdee effdZxZS)_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. TroNrr0r1r4rlrmrnmultiprocessing_contextrrrsc t|dvsJ||_||_||_||_||_||_t|_ |jdkr%|jtj |j|_ |7|dkr1t|tr||krtd|d|d||_||_d|_dS)Nrnrprmax_concurrent= should be <= num_workers=!)superr;r0r1r4rlrmrmp _mp_context get_contextrrrPrrrs_it) r!r0r1r4rlrmrrrrs __class__s r$r;z_ParallelMapperImpl.__init__)s .....  &   '>$ " ;) # #(D(P!~d.JKKD   %+//.#.. T>K3O3O !RN!R!R;!R!R!RSSS,"4RVr'r[ct||j|`|jdkr|||_dS|||_dSr)rr]rr4_parallel_reset _inline_resetr!r[rs r$r]z_ParallelMapperImpl.resetGsf  m$$$ 8   a  ++M::DHHH))-88DHHHr'c:t|j|j|S)N)r0r1r[)rZr0r1r!r[s r$rz!_ParallelMapperImpl._inline_resetQs  DKWdeeeer'c t|j|j|j|j|j|j|j|j| S)N) r0r1r4rlrmrqrrrsr[) rkr0r1r4rlrmrrrrsrs r$rz#_ParallelMapperImpl._parallel_resetTsH";;(];'.#6'    r'r2c*t|jSrr`rras r$r`z_ParallelMapperImpl.nextaDH~~r'c4|jSr)rreras r$rez_ParallelMapperImpl.get_statedsx!!###r')TroNNrr)r,r-r.rhrr/rrrrr r rir;rrr]rrr`re __classcell__rs@r$rr s/715(,"#WW W!a W W  W +, W"*#W! W WWWWWW<998DcN#;999999f8DcN+Cffff  Xd38n-E     a$4S>$$$$$$$$r'rceZdZdZdZ ddeedeegefd e d e d e d d e e de e de de e ffd Zdde ee efffd ZdefdZdee effdZxZS)r5aParallelMapper 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_stateTroNrr0r1r4rlrmrnrrrrsprebatchc t|dvsJ||_||_||_||_|7|dkr1t |tr||krtd|d|d||_ ||_ | |_ | ||_ ||_ nE| dkrtd| dt||_ t|| d |_ t!|j |j |j|j|j|j|j |j } |j | |_dSt%| |_dS) Nrnrrrrz prebatch=z must be a positive integer!r:F) batch_size drop_last)r0r1r4rlrmrrrrs)rr;r4rlrmrrrrPrrrsrr1r0r8rrrr) r!r0r1r4rlrmrrrrsrrrs r$r;zParallelMapper.__init__sv .....&   '>$  %+//.#.. T>K3O3O !RN!R!R;!R!R!RSSS,"4    DK DKK1}} !KH!K!K!KLLL&f555DK!&XOOODK!;;(];$($@.#6     = DHHH ~~DHHHr'r[ct||'|j||jdS|jdSr)rr]r IT_STATE_KEYrs r$r]zParallelMapper.resetsY  m$$$  $ HNN=):; < < < < < HNN     r'r2c*t|jSrrras r$r`zParallelMapper.nextrr'cB|j|jiSr)rrrdras r$rezParallelMapper.get_states!48#6#6#8#899r')TroNNrNr)r,r-r.rhrrr/rrrrr r rir;rrr]r`rerrs@r$r5r5hsV2L/715(,"#"&/&/& /&!a /& /&  /& +, /&"*#/&! /& /&3-/&/&/&/&/&/&b8DcN#;a:4S>::::::::r'r5c eZdZdZdeededededee e e ff dZ de efd Zdefd Zde e e ffd Zd efd ZdZdZdS)_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. r0prefetch_factorworkerrsr[c ||_||_||_||_t j|_tj||_ tj |_ d|_ d|_ |:|d|_|d|_ |j|jn d|_|jt!|_tj|j|j|j|j|j|j |j fd|_|j|j|jt,|_t/|j D];} t1|#t2$rt5d|j d |d wxYwd|_ dS) NrvrrxryTrzr}r~rr)r0rrrsrMr+_qrrrr) _stop_eventr _fast_forwardrr]rrr_threadrrrrr`rrP)r!r0rrrsr[rs r$r;z_SingleThreadedMapper.__init__s . "4${}}._EEE $?,,%&"  $*:6DN!./E!FD  K  dn - - - -!DN K     133 '; $'        -BB$,`kBllt)**  A T      gT-?gg/0ggg  s F&F:r2c|Srrras r$rz_SingleThreadedMapper.__iter__!rr'c |jrt |jdt \}}n#t j$rY_wxYwt|tr4|j |j |t|tr\t|ts|j |j |n@|j |xjdz c_|||S^r)rrKrrrLrrMrNrrrrrrrrrrs r$rbz_SingleThreadedMapper.__next__$sR &&(( &#oo%  GKKdMKJJ cc;    $ ..  !!### $$&&& D"233 !$(?@@(I%%''' $$&&&  !!###**a/**++C000 - s$AA! A!c |j|jdSrrras r$rez_SingleThreadedMapper.get_state=rr'rWc^|j|x}||_d|_dSdSrrrs r$rz,_SingleThreadedMapper._maybe_update_snapshotCrr'c.|dSrrras r$rz_SingleThreadedMapper.__del__Hrr'c|jt|dr>|jr'|jt dzdSdSdS)Nrrr)rrrrrrrras r$rgz_SingleThreadedMapper._shutdownKsv  4 # # 9 (=(=(?(? 9 L  ma&7  8 8 8 8 8 9 9 9 9r'N)r,r-r.rhrrr _WorkerTyper rrirr;rrrbrerrrgrr'r$rrs 44 44 4  4  S#X/ 4444l(1+!2 4S>    +#++++ 99999r'r)5rMrrtypingrrrrrrr r r r r rtorch.multiprocessingmultiprocessingrtorchdata.nodes.base_noderrtorchdata.nodes.batchrr!torchdata.nodes.exception_wrapperrrtorchdata.nodes.snapshot_storerrrr constantsrrrr/r6XseqTseqr8r+r)rXrZrkrr5rrrrrr'r$rsd  vvvvvvvvvvvvvvvvvvvvvvvvvvvv""""""1111111144444444WWWWWWWWLLLLLLLL"""""",,,,,,$$$$$$      8    GCLL 8A; !a(8 =P    {{/////71a4=///uU["(23EKU^Ud4         8c6c6c6c6c6(1+c6c6c6LE$E$E$E$E$(1+E$E$E$PX:X:X:X:X:Xa[X:X:X:v  "     ~9~9~9~9~9HQK~9~9~9~9~9r'