&`i/ddlZddlZddlZddlmZddlmZmZmZm Z m Z m Z m Z ddl Z ddlmZddlmZmZddlmZmZmZmZddlmZddlmZmZmZdd lmZdd l m!Z!dd l"m#Z#ej$e%Z&d e e!ed e e'e'e'ffdZ( d,dee!edede ede e e'ege'ffd eef dZ) d-deede ede e'de*de e'de e'de*d eefdZ+ d.deede e,de ed eefdZ- d.deede eegefde ed eefd Z. d.deed!eegefde ed eefd"Z/deed eefd#Z0d$Z1Gd%d&eZ2Gd'd(eZ3e j4d)Gd*d+Z5dS)/N) nullcontext)AnyCallableIteratorListOptionalTupleUnion) ActorHandle)BatcherShufflingBatcher)Batch BatchMetadataBlockPrefetcher CollatedBatch) DatasetStats)Block BlockAccessor DataBatch) DataContext) ObjectRef)NodeAffinitySchedulingStrategyrefsreturnctjtjj}|jrtj|}d| D}tfd|D}td|D}t||z |z }|||fSdS)aGGiven a list of object references, returns how many are already on the local node, how many require fetching from another node, and how many have unknown locations. If `DataContext.get_current().enable_get_object_locations_for_metrics` is False, this will return `(0, 0, 0)` as getting object locations is disabled.cg|] }|d S)node_ids).0locs z/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/data/_internal/block_batching/util.py z'_calculate_ref_hits..#s!K!K!Kc#j/!K!K!Kc3 K|]}|vV dSNr)rrcurrent_node_ids r! z&_calculate_ref_hits..$s(EE8?h.EEEEEEr#c3K|]}|dV dS)Nr)rrs r!r'z&_calculate_ref_hits..%s'??Xh?q??????r#)rrr) rayget_runtime_context get_node_iddatar get_current'enable_get_object_locations_for_metrics experimentalget_object_locationsvaluessumlen)rctxlocsnodeshitsunknownsmissesr&s @r!_calculate_ref_hitsr;s -//;;==O (  * * , ,C 2&44T::!K!KT[[]]!K!K!KEEEEuEEEEE???????Ud"X-VX%% 7r#block_ref_iterr5statsmax_get_batch_sizec#\ Kdd d dtffd }g dttf  fd }|D]C} |t |kr|D]}|VD|D]}|Vr_ _ _dSdS)a%Resolves the block references for each logical batch. Args: block_ref_iter: An iterator over block object references. ctx: The ``DataContext`` to use. stats: An optional stats object to recording block hits and misses. max_get_batch_size: Maximum number of block references to resolve in a single ``ray.get()`` call. This can be an integer override or a callable that returns the desired batch size dynamically. If ``None``, defaults to ``ctx.iter_get_block_batch_size``. rrcrtr }n}||nj}td|S)Nr))callableiter_get_block_batch_sizemax)override candidater5r>s r!_get_effective_batch_sizez5resolve_block_refs.._get_effective_batch_sizeAsJ & ' ' *))++HH)H ( 4HH#:W 1i   r#cnsgSt\}}}||cxkr |cxkrdkr nndxxndkr|z |z |z rjn t5t j}dddn #1swxYwY|S)N)r; iter_get_stimerrr*getclear) current_hit current_misscurrent_unknownblocksr8r:pendingr=r9s r!_resolve_pendingz,resolve_block_refs.._resolve_pendingMs' I5H5Q5Q2 \? , ? ? ? ?/ ? ? ? ?R ? ? ? ? ?') )D )6HH RZZ K D l "F  'H). AU  # # % % %KMM & &WW%%F & & & & & & & & & & & & & & &   s5BBBN)intrrappendr4iter_blocks_localiter_blocks_remoteiter_unknown_location) r<r5r=r>rFrR block_refblockr8r:rQr9s ``` @@@@r!resolve_block_refsrZ,sA" D FH!s!!!!!!!')Gd5k($ y!!! w<<4466 6 6))++   !!##  /"&#) &.###//r#F block_iter batch_size drop_lastshuffle_buffer_min_size shuffle_seed ensure_copyc#K|t|||}nt||}fd}d} |D]} || |rp|5|} dddn #1swxYwYt t | | V| dz } |p||rp|5|} dddn #1swxYwYt t | | V| dz } |p|sr|r`|5|} dddn #1swxYwYt t | | V| dz } dSdSdS) aGiven an iterator over blocks, returns an iterator over blocks of the appropriate bacth size. If the shuffling configurations are specified, then the output blocks contain shuffled data. Args: block_iter: An iterator over blocks. stats: Dataset stats object used to store block batching time. batch_size: Record batch size, or None to let the system pick. drop_last: Whether to drop the last batch if it's incomplete. shuffle_buffer_min_size: If non-None, the data will be randomly shuffled using a local in-memory shuffle buffer, and this value will serve as the minimum number of rows that must be in the local in-memory shuffle buffer in order to yield a batch. shuffle_seed: The seed to use for the local random shuffle. ensure_copy: Whether batches are always copied from the underlying base blocks (not zero-copy views). Returns: An iterator over blocks of the given size that are potentially shuffled. N)r\r^r_)r\r`cVrjn tSr%)iter_next_batch_srJr)r=sr!get_iter_next_batch_s_timerz6blocks_to_batches..get_iter_next_batch_s_timers&27Ju&,,...[]]Jr#r) batch_idxmetadatar-r)) r r add has_batch next_batchrr done_addinghas_any) r[r=r\r]r^r_r`batcherrdglobal_counterrYbatchs ` r!blocks_to_batchesrpps>*"!$;%    Z[IIIKKKKKN   E!! ,,.. - -**,, - - - - - - - - - - - - - - -!H!H!HuUUU U U U a N !!        ( ( * * ) )&&((E ) ) ) ) ) ) ) ) ) ) ) ) ) ) )]^DDD5QQQQQQ!      ** ( ( * * ) )&&((E ) ) ) ) ) ) ) ) ) ) ) ) ) ) )]^DDD5QQQQQQ! s6(B  B B D""D&)D&F((F,/F, batch_iter batch_formatc#K|D]}|r|jn t5tj|j|}dddn #1swxYwYtj||VdS)aCGiven an iterator of blocks, returns an iterator of formatted batches. Args: batch_iter: An iterator over batches. batch_format: The batch format to use. stats: An optional stats object to record formatting times. Returns: An iterator over batch index and the formatted batch. Nr-) iter_format_batch_srJrr for_blockr-to_batch_format dataclassesreplace)rqrrr=roformatted_batchs r!format_batchesr{s??27 JU & , , . . .[]]  +5ejAAQQO               !%o>>>>>>> ??s-A))A- 0A- collate_fnc#K|D]k}|r|jn t5||j}dddn #1swxYwYt |j|VldS)aReturns an iterator with the provided collate_fn applied to items of the batch iterator. Args: batch_iter: An iterator over formatted batches. collate_fn: A function to apply to each batch. stats: An optional stats object to record formatting times. Nrf)iter_collate_batch_srJrr-rrg)rqr|r=rocollated_batchs r!collatersJJ38 KU ' - - / / /kmm 4 4'Z 33N 4 4 4 4 4 4 4 4 4 4 4 4 4 4 4U^.IIIIIIIJJA  A A finalize_fnc#K|D]k}|r|jn t5||j}dddn #1swxYwYt j||VldS)aReturns an iterator with the provided finalize_fn applied to items of the batch iterator. This is the same as `collate` except the input batches can be of type Any. Args: batch_iter: An iterator over processed batches. finalize_fn: A function to apply to each batch. stats: An optional stats object to record formatting times. Returns: An iterator over batch index and the finalized batch. Nrt)iter_finalize_batch_srJrr-rxry)rqrr=rofinalized_batchs r!finalize_batchesrs$??49 LU ( . . 0 0 0{}} 6 6)k%*55O 6 6 6 6 6 6 6 6 6 6 6 6 6 6 6!%o>>>>>>>??rc#&K|D] }|jV dSr%rt)rqros r!extract_data_from_batchrs0jr#z ray.datasetcZeZdZdZdZdZdeeefdZ de fdZ dZ d Z d S) WaitBlockPrefetcherz Block prefetcher using ray.wait.cg|_d|_d|_tj|_tj|jdd|_|j dS)NFr PrefetcherT)targetnamedaemon) _blocks_stopped_last_prefetch_size threading Condition _conditionThread_run_threadstartselfs r!__init__zWaitBlockPrefetcher.__init__sh  #$ #-// '9    r#c|js |j5t|jdkr|j|jddgc}|_dddn #1swxYwYt|dkrt j|dddn*#t $rtdYnwxYw|jt ddS)Nrr)T) num_returnstimeout fetch_localzError in prefetcher thread.z&Exiting prefetcher's background thread) rrr4rwaitr* Exceptionlogger exceptiondebug)rblocks_to_fetchs r!rzWaitBlockPrefetcher._runse- @ @_HH4<((A--,,...48LOR1OT\ HHHHHHHHHHHHHHH''!++H'$%!"$(  @ @ @  !>????? @'- @,  =>>>>>s5BA A% B%A))B,A)-.B$CCrPc|j5|jrtd||_t ||_|jddddS#1swxYwYdS)NzPrefetcher is stopped.)rr RuntimeErrorrr4rnotifyrrPs r!prefetch_blocksz#WaitBlockPrefetcher.prefetch_blocks)s _ % %} ="#;<<<!DL'*6{{D $ O " " $ $ $  % % % % % % % % % % % % % % % % % %sA A  A$'A$rcR|j5|jcdddS#1swxYwYdSr%)rrrs r!num_prefetched_blocksz)WaitBlockPrefetcher.num_prefetched_blocks1st _ , ,+ , , , , , , , , , , , , , , , , , ,s   c|j5|jr ddddSd|_|jddddS#1swxYwYdS)NT)rrrrs r!stopzWaitBlockPrefetcher.stop5s _ % %}  % % % % % % % %!DM O " " $ $ $  % % % % % % % % % % % % % % % % % %s A  A  AAc.|dSr%)rrs r!__del__zWaitBlockPrefetcher.__del__<s r#N)__name__ __module__ __qualname____doc__rrrrrrrSrrrrr#r!rrs**   ???2%d9U+;&<%%%%,s,,,,%%%r#rc`eZdZdZdZed dZdeee fdZ de fdZ d S) ActorBlockPrefetcherz%Block prefetcher using a local actor.cF||_d|_dS)Nr)_get_or_create_actor_prefetcherprefetch_actorrrs r!rzActorBlockPrefetcher.__init__Cs$"BBDD#$   r#rr ctj}d|}tt |d|t dS)Nzdataset-block-prefetcher-F)softT)scheduling_strategyr namespace get_if_exists)r*r+r,_BlockPretcheroptionsrPREFETCHER_ACTOR_NAMESPACEremote)node_id actor_names r!rz4ActorBlockPrefetcher._get_or_create_actor_prefetcherGsj)++7799::: %% >wU S S S0 &  &((  r#rPcVt||_|jjj|dSr%)r4rrprefetchrrs r!rz$ActorBlockPrefetcher.prefetch_blocksRs,#&v;; +$+V4444r#c|jSr%)rrs r!rz*ActorBlockPrefetcher.num_prefetched_blocksVs ''r#N)rr ) rrrrr staticmethodrrrrrrSrrr#r!rr@s//%%%\5d9U+;&<5555(s((((((r#r)num_cpusceZdZdZddZdS)rz3Helper actor that prefetches blocks asynchronously.rNcdSr%rrs r!rz_BlockPretcher.prefetch^s r#)rN)rrrrrrr#r!rrZs.==      r#r)NN)NNFNNFr%)6rxloggingr contextlibrtypingrrrrrr r r* ray.actorr ray.data._internal.batcherr r ,ray.data._internal.block_batching.interfacesrrrrray.data._internal.statsrray.data.blockrrrray.data.contextr ray.typesrray.util.scheduling_strategiesr getLoggerrrrSr;rZboolrpstrr{rrrrrrrrrr#r!rs""""""HHHHHHHHHHHHHHHHHH !!!!!!@@@@@@@@ 211111::::::::::((((((IIIIII  8 $ $d9S>2uS#s]7K,%)BF A/A/Yu-.A/ A/ L !A/!sHRW,='=!>? A/ e_ A/A/A/A/L%) $-1"&DDD L !D D D &c] D 3- DDe_DDDDT%)???3-? L !?e_ ????4%)JJJ9+s"234J L !Jm JJJJ,%)??'?3%*%? L !?m ????0HSM +<<<<</<<<~(((((?(((4 Q          r#