o bi'@sJddlZddlZddlmZddlmZmZmZmZm Z m Z ddl Z ddl m Z ddlmZmZddlmZmZmZddlmZddlmZmZmZdd lmZdd lmZee Z!d eeed e e"e"e"ffd dZ# d0deeede ed eefddZ$      d1deede ede e"de%de e"de e"de%d eefddZ& d0deede e'de ed eefddZ( d0deed e eegefde ed eefd!d"Z) d0deed#eegefde ed eefd$d%Z*deed eefd&d'Z+d(Z,Gd)d*d*eZ-Gd+d,d,eZ.e j/dd-Gd.d/d/Z0dS)2N) nullcontext)AnyCallableIteratorListOptionalTuple) ActorHandle)BatcherShufflingBatcher)BatchBlockPrefetcher CollatedBatch) DatasetStats)Block BlockAccessor DataBatch) ObjectRef)NodeAffinitySchedulingStrategyrefsreturncsttjjj}|jr@tj |}dd| D}t fdd|D}t dd|D}t |||}|||fSdS)aJGiven 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 `(-1, -1, -1)` as getting object locations is disabled.cSsg|]}|dqS)node_ids).0locrrZ/home/ubuntu/.local/lib/python3.10/site-packages/ray/data/_internal/block_batching/util.py sz'_calculate_ref_hits..c3s|]}|vVqdSNrrrcurrent_node_idrr !z&_calculate_ref_hits..css|]}|sdVqdS)Nrrrrrr!"r")r$r$) rayget_runtime_context get_node_iddatacontext DataContext get_current'enable_get_object_locations_for_metrics experimentalget_object_locationsvaluessumlen)rctxlocsnodeshitsunknownsmissesrrr_calculate_ref_hitss   r8block_ref_iterstatsc csd}d}d}|D]8}t|g\}}}||7}||7}||7}|r&|jnt t|} Wdn1s9wY| Vq |rO||_||_||_dSdS)zResolves the block references for each logical batch. Args: block_ref_iter: An iterator over block object references. stats: An optional stats object to recording block hits and misses. rN) r8 iter_get_stimerrr%getiter_blocks_localiter_blocks_remoteiter_unknown_location) r9r:r5r7r6 block_ref current_hit current_misscurrent_unknownblockrrrresolve_block_refs)s$   rFF block_iter batch_size drop_lastshuffle_buffer_min_size shuffle_seed ensure_copyc #sP|dur t|||d}nt||d}fdd}d} |D]0} || |rM| |} Wdn1s:wYt| | V| d7} |s(q||r{| |} Wdn1shwYt| | V| d7} |sV|s|r| |} Wdn1swYt| | V| d7} 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)rHrJrK)rHrLcsrjStSr)iter_next_batch_sr<rrr:rrget_iter_next_batch_s_timerqsz6blocks_to_batches..get_iter_next_batch_s_timerrr#)r r add has_batch next_batchr done_addinghas_any) rGr:rHrIrJrKrLbatcherrOglobal_counterrEbatchrrNrblocks_to_batchesIsF           rX batch_formatc csd|D],}|r |jntt|j|}Wdn1s#wYt|j|VqdS)aBGiven an iterator of blocks, returns an iterator of formatted batches. Args: block_iter: An iterator over blocks. 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. N) iter_format_batch_sr<rr for_blockr(to_batch_formatr batch_idx)rGrYr:rWformatted_batchrrrformat_batchess r_ batch_iter collate_fnc c\|D](}|r |jnt ||j}Wdn1swYt|j|VqdS)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. N)iter_collate_batch_sr<rr(rr])r`rar:rWcollated_batchrrrcollates  re finalize_fnc crb)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. N)iter_finalize_batch_sr<rr(rr])r`rfr:rWfinalized_batchrrrfinalize_batchess riccs|D]}|jVqdSr)r()r`rWrrrextract_data_from_batchs rjz ray.datasetc@sFeZdZdZddZddZdeeefddZ d d Z d d Z d S)WaitBlockPrefetcherz Block prefetcher using ray.wait.cCs8g|_d|_t|_tj|jddd|_|jdS)NF PrefetcherT)targetnamedaemon) _blocks_stopped threading Condition _conditionThread_run_threadstartselfrrr__init__s zWaitBlockPrefetcher.__init__cCs zLg}|j/t|jdkr|jddg}|_n|jr( WdWdSg}|jWdn1s9wYt|dkrLtj|dddWnty[tdYnwq)NTrr#) num_returns fetch_localzError in prefetcher thread.) rtr1rprqwaitr% Exceptionlogger exception)rzblocks_to_waitrrrrvs(   zWaitBlockPrefetcher._runblockscCsJ|j|jr td||_|jWddS1swYdS)NzPrefetcher is stopped.)rtrq RuntimeErrorrpnotifyrzrrrrprefetch_blockss  "z#WaitBlockPrefetcher.prefetch_blockscCsT|j|jr WddSd|_|jWddS1s#wYdS)NT)rtrqrryrrrstops "zWaitBlockPrefetcher.stopcCs |dSr)rryrrr__del__s zWaitBlockPrefetcher.__del__N) __name__ __module__ __qualname____doc__r{rvrrrrrrrrrrrks  rkc@s<eZdZdZddZed ddZdeee fd d Z d S) ActorBlockPrefetcherz%Block prefetcher using a local actor.cCs||_dSr)_get_or_create_actor_prefetcherprefetch_actorryrrrr{szActorBlockPrefetcher.__init__rr cCs4t}d|}tjt|dd|tddS)Nzdataset-block-prefetcher-F)softT)scheduling_strategyrn namespace get_if_exists)r%r&r'_BlockPretcheroptionsrPREFETCHER_ACTOR_NAMESPACEremote)node_id actor_namerrrrs   z4ActorBlockPrefetcher._get_or_create_actor_prefetcherrcCs|jjj|dSr)rprefetchrrrrrrsz$ActorBlockPrefetcher.prefetch_blocksN)rr ) rrrrr{ staticmethodrrrrrrrrrr s   r)num_cpusc@seZdZdZdddZdS)rz3Helper actor that prefetches blocks asynchronously.rNcGsdSrrrrrrr$sz_BlockPretcher.prefetch)rN)rrrrrrrrrr srr)NNFNNF)1loggingrr contextlibrtypingrrrrrrr% ray.actorr ray.data._internal.batcherr r ,ray.data._internal.block_batching.interfacesr r rray.data._internal.statsrray.data.blockrrr ray.typesrray.util.scheduling_strategiesr getLoggerrrintr8rFboolrXstrr_rerirjrrkrrrrrrrs       $  " J    1