Pi ddlZddlZddlmZmZmZmZddlmZ ddl m Z ddl m Z mZddlmZmZddlmZde d eeje jfd ed ed ejd ejf dZdS)N)AnyDictOptionalUnion)BaseNode)ExceptionWrapperStartupExceptionWrapper)MonotonicIndex SnapshotStore) QUEUE_TIMEOUTsourceqsnapshot_storesnapshot_frequency semaphore stop_eventc  t ddtdtttt t ftff fd } t|tr|dks Jd| | n7#t$r*td } |YdSwxYwd}| s|dt s1 t!|} |d z }d} |dkr||zdkr| } || d | nN#t"$r}||d Yd}~dSd}~wt$r!t%d } || d YdSwxYw| dSdS)a_populate_queue calls `iter(source)` to get an iterator `it`, waits for semaphore.acquire, and puts its outputs onto q. It never releases the sempahore. It continues to put items on the q as long as it can acquire the sempahore, stop_event is not set, and StopIteration has not been thrown by the `it`. This function will always put tuples of (x, idx) on the q where idx starts from 0 and is monotonically increasing. x may be the output of next(it), StopIteration, or an ExceptionWrapper. If there is an exception raised during the call to `iter(source)`, this function does not wait to acquire sempahore before putting StartupExceptionWrapper on q. Note: this is only intended to be used by a single thread at once. Each instance creates its own iter for source so if this is called with multiple threads, you may get duplicates if source is not sharded properly. TNblocksnapshotc}|r||||f||rdnddS)N)rversiong?)rtimeout)getappendput)itemrr_idxidxrrs s/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/torchdata/nodes/_populate_queue.py_putz_populate_queue.._put1s_ wwyy  C  ! !8T ! B B B tTl%1G4HHHHHrz5snapshot_frequency must be non-negative integer! Got )rz%in _populate_queue startup for device)where)blockingrr F)rr)rzin _populate_queue)TN)r boolrrrstrrr isinstanceintappend_initial_snapshot state_dict Exceptionis_setacquirer next StopIterationr) rrrrrrr!eyieldedrrrs `` @r _populate_queuer2sr4   CMQIII5c3h1H!HIJIIIIIIII )3 / / X4F!4K4K4K WCU W W5L4K K..8I8I8K8K.LLLL  #*Q R R R...::: G!!  $  FF   <???D DU # # # # EE !!s2AB,,0C C AE F*) E<<*F*)F*)queue threadingtypingrrrrtorch.multiprocessingmultiprocessingmptorchdata.nodes.base_noder!torchdata.nodes.exception_wrapperrr torchdata.nodes.snapshot_storer r constantsr Queuer(BoundedSemaphoreEventr2r"r rAs ------------""""""......WWWWWWWWHHHHHHHH$$$$$$A A U["( "#A"A A ) A  AAAAAAr"