&`i[ PddlZddlZddlZddlZddlmZddlmZmZm Z m Z m Z m Z ddl Z ddlmZddlmZmZe dZe dZe d#d e ed ed ed dfdZe d#ded ed ed dfdZe d$de e ed ed dfdZe d%de dd dfdZeGdde eZeGdde eZeGddeZdZGdd e Z!Gd!d"eZ"dS)&N)contextmanager)AnyCallableGenericIterableListTypeVar) Deprecated)MetricsContext SharedMetricsTUFitems num_shardsrepeatreturnParallelIterator[T]cHdt|D}t|D]#\}}|||z|$d|rt |djpdt |||rdnd}t|||S)a:Create a parallel iterator from an existing set of objects. The objects will be divided round-robin among the number of shards. Args: items: The list of items to iterate over. num_shards: The number of worker actors to create. repeat: Whether to cycle over the items forever. cg|]}gSr).0_s a/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/util/iter.py zfrom_items..s , , ,Qb , , ,zfrom_items[{}, {}, shards={}{}]rNone , repeat=Truername)range enumerateappendformattype__name__lenfrom_iterators)rrrshardsiitemr!s r from_itemsr-s- ,% ++ , , ,FU##,,4q:~%%d++++ , 3 3 )$uQx..)3V E !)r   D &d ; ; ;;rnzParallelIterator[int]cg}||z}t|D]>}||z}||dz kr|}n|dz|z}|t||?d|d||rdndd}t|||S)a/Create a parallel iterator over the range 0..n. The range will be partitioned sequentially among the number of shards. Args: n: The max end of the range of numbers. num_shards: The number of worker actors to create. repeat: Whether to cycle over the range forever. z from_range[z , shards=rr]r )r"r$r)) r.rr generators shard_sizer+startendr!s r from_ranger6*sJjJ :  --J Q  CCq5J&C%s++,,,,VaVV*VV2S//QSVVV      rr2ctjtfd|D}|s'dt |rdnd}t ||S)a*Create a parallel iterator from a list of iterables. An iterable can be a conatiner (list, str, tuple, set, etc.), a generator, or a custom class that implements __iter__ or __getitem__. An actor will be created for each iterable. Examples: >>> # Create using a list of generators. >>> from_iterators([range(100), range(100)]) >>> # Certain generators are not serializable. >>> from_iterators([(x for x in range(100))]) ... TypeError: can't pickle generator objects >>> # So use lambda functions instead. >>> # Lambda functions are serializable. >>> from_iterators([lambda: (x for x in range(100))]) Args: generators: A list of Python iterables or lambda functions that produce an iterable when called. We allow lambda functions since certain generators might not be serializable, but a lambda that returns it can be. repeat: Whether to cycle over the iterators forever. name: Optional name to give the iterator. c<g|]}|Srremote)rgr worker_clss rrz"from_iterators..is) ? ? ?qj6** ? ? ?rzfrom_iterators[shards={}{}]rrr!)rayr:ParallelIteratorWorkerr%r( from_actors)r2rr!actorsr<s ` @rr)r)Jsx<233J ? ? ? ? ?J ? ? ?F  ,33  OO>__B   vD ) ) ))rrAray.actor.ActorHandlecn|sdt|d}tt|gg|gS)aCreate a parallel iterator from an existing set of actors. Each actor must subclass the ParallelIteratorWorker interface. Args: actors: List of actors that each implement ParallelIteratorWorker. name: Optional name to give the iterator. zfrom_actors[shards=r1parent_iterators)r(ParallelIterator _ActorSet)rAr!s rr@r@qsD 43S[[333 Yvr223TB O O OOrc eZdZdZdeddededfdZdZd Zd Z d Z d e e e ge efd dfdZ d3d e e gefd dfdZd e e gefd dfdZded dfdZd4dZd e e geefd dfdZ d5deded dfdZ d6ded ed dfd!Zd7d#Zd8d%Zd9d7d&Zded ee fd'Zd:defd)Zd;d+Zd,eed dfd-Zd efd.Z d ed"fd/Z! d9d0ed ed1ed d"fd2Z"dS)>> # Applying a function over items in parallel. >>> it = ray.util.iter.from_items([1, 2, 3], num_shards=2) ... <__main__.ParallelIterator object> >>> it = it.for_each(lambda x: x * 2).gather_sync() ... <__main__.LocalIterator object> >>> print(list(it)) ... [2, 4, 6] >>> # Creating from generators. >>> it = ray.util.iter.from_iterators([range(3), range(3)]) ... <__main__.ParallelIterator object> >>> print(list(it.gather_sync())) ... [0, 0, 1, 1, 2, 2] >>> # Accessing the individual shards of an iterator. >>> it = ray.util.iter.from_range(10, num_shards=2) ... <__main__.ParallelIterator object> >>> it0 = it.get_shard(0) ... <__main__.LocalIterator object> >>> print(list(it0)) ... [0, 1, 2, 3, 4] >>> it1 = it.get_shard(1) ... <__main__.LocalIterator object> >>> print(list(it1)) ... [5, 6, 7, 8, 9] >>> # Gathering results from actors synchronously in parallel. >>> it = ray.util.iter.from_actors(workers) ... <__main__.ParallelIterator object> >>> it = it.batch_across_shards() ... <__main__.LocalIterator object> >>> print(next(it)) ... [worker_1_result_1, worker_2_result_1] >>> print(next(it)) ... [worker_1_result_2, worker_2_result_2] actor_setsrGr!rEzParallelIterator[Any]c0||_||_||_dS)z:Create a parallel iterator (this is an internal function).N)rIr!rE)selfrIr!rEs r__init__zParallelIterator.__init__s"% !1rc td)NzeYou must use it.gather_sync() or it.gather_async() to iterate over the results of a ParallelIterator.) TypeErrorrKs r__iter__zParallelIterator.__iter__s >   rc t|SNreprrOs r__str__zParallelIterator.__str__Dzzrcd|jdS)NzParallelIterator[r1r=rOs r__repr__zParallelIterator.__repr__s/49////rcdtfd|jD|j|z|jS)z/Helper function to create new Parallel Iteratorc:g|]}|Sr)with_transform)ra local_it_fns rrz4ParallelIterator._with_transform..s' D D DqQ  k * * D D Dr)r!rE)rFrIr!rE)rKr]r!s ` r_with_transformz ParallelIterator._with_transformsC D D D DDO D D DT!!2    rfnrzParallelIterator[U]c6|fddS)a\Remotely transform the iterator. This is advanced version of for_each that allows you to apply arbitrary generator transformations over the iterator. Prefer to use .for_each() when possible for simplicity. Args: fn: function to use to transform the iterator. The function should pass through instances of _NextValueNotReady that appear in its input iterator. Note that this function is only called **once** over the input iterator. Returns: ParallelIterator[U]: a parallel iterator. Examples: >>> def f(it): ... for x in it: ... if x % 2 == 0: ... yield x >>> from_range(10, 1).transform(f).gather_sync().take(5) ... [0, 2, 4, 6, 8] c.|SrR) transformlocal_itr_s rz,ParallelIterator.transform..sX//33r .transform()r^rKr_s `rrbzParallelIterator.transforms+4## 3 3 3 3^   rr0Nc^dks Jd|fddS)aRemotely apply fn to each item in this iterator. If `max_concurrency` == 1 then `fn` will be executed serially by each shards `max_concurrency` should be used to achieve a high degree of parallelism without the overhead of increasing the number of shards (which are actor based). If `max_concurrency` is not 1, this function provides no semantic guarantees on the output order. Results will be returned as soon as they are ready. A performance note: When executing concurrently, this function maintains its own internal buffer. If `num_async` is `n` and max_concur is `k` then the total number of buffered objects could be up to `n + k - 1` Args: fn: function to apply to each item. max_concurrency: max number of concurrent calls to fn per shard. If 0, then apply all operations concurrently. resources: resources that the function requires to execute. This has the same default as `ray.remote` and is only used when `max_concurrency > 1`. Returns: ParallelIterator[U]: a parallel iterator whose elements have `fn` applied. Examples: >>> next(from_range(4).for_each( lambda x: x * 2, max_concur=2, resources={"num_cpus": 0.1}).gather_sync() ) ... [0, 2, 4, 8] rz%max_concurrency must be non-negative.c2|SrR)for_each)rdr_max_concurrency resourcess rrez+ParallelIterator.for_each.. sX..r?INNr .for_each()rg)rKr_rlrms ```rrkzParallelIterator.for_eachsOP!###%L##### N N N N N N    rrc6|fddS)aRemotely filter items from this iterator. Args: fn: returns False for items to drop from the iterator. Examples: >>> it = from_items([0, 1, 2]).filter(lambda x: x > 0) >>> next(it.gather_sync()) ... [1, 2] c.|SrR)filterrcs rrez)ParallelIterator.filter../sX__R5H5Hr .filter()rgrhs `rrqzParallelIterator.filter$s%##$H$H$H$H+VVVrr.zParallelIterator[List[T]]c>|fdddS)zRemotely batch together items in this iterator. Args: n: Number of items to batch together. Examples: >>> next(from_range(10, 1).batch(4).gather_sync()) ... [0, 1, 2, 3] c.|SrR)batch)rdr.s rrez(ParallelIterator.batch..;sX^^A5F5Fr.batch()rgrKr.s `rruzParallelIterator.batch1s.##$F$F$F$FRSWWWrParallelIterator[T[0]]c0|ddS)zFlatten batches of items into individual items. Examples: >>> next(from_range(10, 1).batch(4).flatten()) ... 0 c*|SrR)flatten)rds rrez*ParallelIterator.flatten..DsX5E5E5G5Gr .flatten()rgrOs rr|zParallelIterator.flatten=s##$G$GVVVrcr||}|jdz|_|S)z}Transform and then combine items horizontally. This is the equivalent of for_each(fn).flatten() (flat map). .combine()rkr|r!rKr_its rcombinezParallelIterator.combineFs4 ]]2   & & ( ()l* rshuffle_buffer_sizeseedc |fddtndS)aRemotely shuffle items of each shard independently Args: shuffle_buffer_size: The algorithm fills a buffer with shuffle_buffer_size elements and randomly samples elements from this buffer, replacing the selected elements with new elements. For perfect shuffling, this argument should be greater than or equal to the largest iterator size. seed: Seed to use for randomness. Default value is None. Returns: A ParallelIterator with a local shuffle applied on the base iterator Examples: >>> it = from_range(10, 1).local_shuffle(shuffle_buffer_size=2) >>> it = it.gather_sync() >>> next(it) 0 >>> next(it) 2 >>> next(it) 3 >>> next(it) 1 c0|SrR)shuffle)rdrrs rrez0ParallelIterator.local_shuffle..nsX--.A4HHrz/.local_shuffle(shuffle_buffer_size={}, seed={})Nr)r^r%str)rKrrs ``r local_shufflezParallelIterator.local_shuffleOsT<## H H H H H = D D#$2BSYYY     rrnum_partitionsbatch_msc g|jD]0}||j1d fd fd |jddz} fdt D}t jt fd|D}tt|gg||gS) aReturns a new ParallelIterator instance with num_partitions shards. The new iterator contains the same data in this instance except with num_partitions shards. The data is split in round-robin fashion for the new ParallelIterator. Args: num_partitions: The number of shards to use for the new ParallelIterator batch_ms: Batches items for batch_ms milliseconds on each shard before retrieving it. Increasing batch_ms increases latency but improves throughput. Returns: A ParallelIterator with num_partitions number of shards and the data of this ParallelIterator split round-robin among the new number of shards. Examples: >>> it = from_range(8, 2) >>> it = it.repartition(3) >>> list(it.get_shard(0)) [0, 4, 3, 7] >>> list(it.get_shard(1)) [1, 5] >>> list(it.get_shard(2)) [2, 6] Nc3PKi} D]"}|||j|| <#|rt|}|Ctj|t |d\}}|stj|d\}}n'tj|t ||\}}|D]e}||} tj|} | || j|| <| D]} | VV#t$rYbwxYw|tV|dSdS)N)stepr4rr num_returnstimeoutr0r) par_iter_slice_batchr:listr>waitr(popget StopIteration_NextValueNotReady)rpartition_indexrfuturesr\pendingreadyrobj_refactorrur, all_actorsrs r base_iteratorz3ParallelIterator.repartition..base_iteratorsG    *11+?X2  /w--?"xS\\STUUUHE1 D#&8G#C#C#Cq"xS\\7   HE1 %G#KK00E  # 0 0"  !6==%3&5)1>%*''D"&JJJJ'(&,.....; / / / / /s=D D  D cfdS)NcSrRr)rr+rsrrezBParallelIterator.repartition..make_gen_i..s==;;rr)r+rrs`r make_gen_iz0ParallelIterator.repartition..make_gen_is;;;;;; ;rz.repartition[num_partitions=r1c&g|] }|Srr)rsrs rrz0ParallelIterator.repartition..s!CCCjjmmCCCrc>g|]}|dS)F)rr9)rr;r<s rrz0ParallelIterator.repartition..s,III*##Ae#44IIIrrDrR) rI init_actorsextendrAr!r"r>r:r?rFrG) rKrr actor_setr!r2rArrrr<s `` @@@@r repartitionzParallelIterator.repartitiontsB  0 0I  ! ! # # #   i. / / / /% /% /% /% /% /% /% /N < < < < < <yK.KKKKCCCCU>-B-BCCC Z 677 IIIIjIII62!6!6 7QUPVWWWWrLocalIterator[T]cf|}|d|_|S)aReturns a local iterable for synchronous iteration. New items will be fetched from the shards on-demand as the iterator is stepped through. This is the equivalent of batch_across_shards().flatten(). Examples: >>> it = from_range(100, 1).gather_sync() >>> next(it) ... 0 >>> next(it) ... 1 >>> next(it) ... 2 z.gather_sync())batch_across_shardsr|r!)rKrs r gather_synczParallelIterator.gather_syncs6" % % ' ' / / 1 1))) rLocalIterator[List[T]]cVdfd }d}t|t|S)zIterate over the results of multiple shards in parallel. Examples: >>> it = from_iterators([range(3), range(3)]) >>> next(it.batch_across_shards()) ... [0, 0] Nc3rKg}jD]0}|||j1d|D}|r t j||Vd|D}|t Vn#t$rt VYnt$rg}tt||D]R\}} | t j|.#t$r| |YOwxYw|r|Vd|D}YnwxYw|dSdS)Nc@g|]}|jSr par_iter_nextr:rr\s rrzOParallelIterator.batch_across_shards..base_iterator..s&@@@Aq--//@@@rrc@g|]}|jSrrrs rrzOParallelIterator.batch_across_shards..base_iterator..&HHHAq5577HHHrc@g|]}|jSrrrs rrzOParallelIterator.batch_across_shards..base_iterator..rr) rIrrrAr>rr TimeoutErrorrziprr$remove)ractiverrresultsr\frKs rrz;ParallelIterator.batch_across_shards..base_iteratorsF!_ 0 0 %%''' i.////@@@@@G II''7;;;;;;HHHHHG*022222#///,.......$ I I I G #DLL' : :--1-#NN371::6666,---"MM!,,,,,-&% HHHHHGGG I I I I I Is< 6BD0 -D0'C65D06DD0DD0/D0z.batch_across_shards()r=rR) LocalIteratorr )rKrr!s` rrz$ParallelIterator.batch_across_shardssP I I I I I I6...]MOO$GGGGrcdkrtddkrtddd fd }d}t|t|S) a|Returns a local iterable for asynchronous iteration. New items will be fetched from the shards asynchronously as soon as the previous one is computed. Items arrive in non-deterministic order. Arguments: batch_ms: Batches items for batch_ms milliseconds on each shard before retrieving it. Increasing batch_ms increases latency but improves throughput. If this value is 0, then items are returned immediately. num_async: The max number of async requests in flight per actor. Increasing this improves the amount of pipeline parallelism in the iterator. Examples: >>> it = from_range(100, 1).gather_async() >>> next(it) ... 3 >>> next(it) ... 0 >>> next(it) ... 1 r0zqueue depth must be positiverbatch time must be positiveNc3Kg}jD]0}|||j1i}t D]$}|D]}|||j < %|rt|}|Ctj |t|d\}}|stj |d\}}n'tj |t||\}}|D]}| |} | j _tj |} | || j <| D]} | Vq#t$rY}wxYw|t!V|dSdS)Nrrr0r)rIrrrAr"par_iter_next_batchr:rr>rr(rshared_metricsr current_actorrr)rrrrrr\rrrrrur,r local_iter num_asyncrKs rrz4ParallelIterator.gather_async..base_iterator)sJ!_ 4 4 %%'''!!)"23333G9%% H H#HHAFGGA188BBCCH /w--?"xS\\STUUUHE1 D#&8G#C#C#Cq"xS\\7   HE1 %  G#KK00EHM 15577E # 0 0NS 9 @ @ J JK$)''D"&JJJJ'(&,.....1 / / / / /s AE%% E21E2z.gather_async()r=rR) ValueErrorrr )rKrrrr!rs``` @r gather_asynczParallelIterator.gather_asyncs2 q==;<< < a<<:;; ; ! /! /! /! /! /! /! /! /! /F'''"=-//MMM rcP||Sz2Return up to the first n items from this iterator.)rtakerxs rrzParallelIterator.takeP"!!&&q)))rcP||S)1Print up to the first n items from this iterator.)rshowrxs rrzParallelIterator.showTrrotherc"t|tstdt|g}||j||jt|d|d|d|j|jzS)z;Return an iterator that is the union of this and the other.z,other must be of type ParallelIterator, got zParallelUnion[, r1rD) isinstancerFrNr&rrIrE)rKrrIs runionzParallelIterator.unionXs%!122 LtE{{LL  $/***%*+++  -T - -U - - -!2U5KK    rshards_to_keepct|jdkrtdtdkrtd|jd}fdt|jD}t|tks Jdt ||j}t|g|dtd|j S) zReturn a child iterator that only iterates over given shards. It is the user's responsibility to ensure child iterators are operating over disjoint sub-sets of this iterator's shards. r0z,select_shards() is not allowed after union()rz#at least one shard must be selectedc"g|] \}}|v | Srr)rr+r\rs rrz2ParallelIterator.select_shards..ts.   1aqN?R?RA?R?R?RrzInvalid actor indexz.select_shards(z total)rD) r(rIrr#rArG transformsrFrE)rKr old_actor_set new_actors new_actor_sets ` r select_shardszParallelIterator.select_shardsis t  ! # #KLL L ~  ! # #BCC C*     %m&:;;   :#n"5"55557L555!*m.FGG  O @ @C$7$7 @ @ @!2    rc>td|jDS)z9Return the number of worker actors backing this iterator.c3>K|]}t|jVdSrR)r(rArs r z.ParallelIterator.num_shards..s*::Q3qx==::::::r)sumrIrOs rrzParallelIterator.num_shardss!::$/::::::rc^fdtDS)zReturn the list of all shards.c:g|]}|Sr) get_shard)rr+rKs rrz+ParallelIterator.shards..s%DDDaq!!DDDr)r"rrOs`rr*zParallelIterator.shardss/DDDD51B1B+C+CDDDDr shard_indexrc dkrtddkrtdd\ |}|jD]G}|t|jkr|j||j n|t|jz}H#td||d  fd }|jd |d z}t|t| S) aReturn a local iterator for the given shard. The iterator is guaranteed to be serializable and can be passed to remote tasks or actors. Arguments: shard_index: Index of the shard to gather. batch_ms: Batches items for batch_ms milliseconds before retrieving it. Increasing batch_ms increases latency but improves throughput. If this value is 0, then items are returned immediately. num_async: The max number of requests in flight. Increasing this improves the amount of pipeline parallelism in the iterator. r0znum async must be positiverr)NNNzShard index out of rangec3DKtj}tjjt D]/}|j0 tj| |}|j|D]}|V|tVn,#t$rtVYnt$rYdSwxYw)NTr) collectionsdequer>r par_iter_initr:r"r$rpopleftrrr) rqueuerrur,r\rrts rrz1ParallelIterator.get_shard..base_iteratorsA%''E GAO**1-- . . .9%% E E Q299(CCDDDD  GEMMOOWEEEELL!6!=!=h!G!GHHH %##" *022222#///,.......$EE sA0C55D DDz.shard[r1r=rR) rrIr(rArrr!rr ) rKrrrr+rrr!r\rs `` @@rrzParallelIterator.get_shards $ q==9:: : a<<:;; ;1  + +I3y'(((($Q'(S)*** 97dooFWFWXX X         &y3[3333]MOO$GGGGrr0N)rryrR)r)rr)rr)rr0r)rrrr)#r' __module__ __qualname____doc__rrrLrPrUrXr^rrr rrbrkboolrqintrur|rrrrrrrrrrrr*rrrrrFrFs..` 1% 1 167 1 1 1 1   000    HQK=(1+56     >BF, , A36", , , , , \ W1#t), W1F W W W W Xs X: X X X XWWWW(A3Q<05J59# # #&# .1# # # # # L45VXVX!VX-0VX VXVXVXVXp*%H%H%H%HNFFFFFP*c*d1g******c****    " DI :O    ,;C;;;;E/0EEEE EF6H6H6H*-6H>A6H 6H6H6H6H6H6HrrFc 0eZdZdZdZejZ d,dege e fde de ee ge fdefdZed efd Zd Zed Zd ZdZdZdZdee e ge efd dfdZ d-dee gefd dfdZdee gefd dfdZded dfdZd.dZ d/deded dfdZ!dee ge efd dfd Z"d!Z#ded e e fd"Z$d0defd$Z%d e dfd%Z&d&dd'd(dd)ed*e e'd dfd+Z(dS)1raWAn iterator over a single shard of data. It implements similar transformations as ParallelIterator[T], but the transforms will be applied locally and not remotely in parallel. This class is **serializable** and can be passed to other remote tasks and actors. However, it should be read from at most one process at a time._on_fetch_startNrrlocal_transformsrct|tsJ||_d|_|pg|_||_||_|pd|_dS)aCreate a local iterator (this is an internal function). Args: base_iterator: A function that produces the base iterator. This is a function so that we can ensure LocalIterator is serializable. shared_metrics: Existing metrics context or a new context. Should be the same for each chained iterator. local_transforms: A list of transformation functions to be applied on top of the base iterator. When iteration begins, we create the base iterator and apply these functions. This lazy creation ensures LocalIterator is serializable until you start iterating over it. timeout: Optional timeout in seconds for this iterator, after which _NextValueNotReady will be returned. This avoids blocking. name: Optional name for this iterator. Nunknown)rr rbuilt_iteratorrrrr!)rKrrrrr!s rrLzLocalIterator.__init__sV4.-88888*" 0 6B, %I rrcttjdrtjjt dtjjS)zaReturn the current metrics context. This can only be called within an iterator function.metricsNz*Cannot access context outside an iterator.)hasattrr thread_localrrrrr get_metricszLocalIterator.get_metricssA  2I>> K)19IJJ J)11rc|jEt||j}|jD] }||}||_dSdSrR)riterrrr)rKrr_s r _build_oncezLocalIterator._build_onces_   &d((6677B+  RVV"$D    ' &rc#XK|j|j_dVdSrR)rrrrrOs r_metrics_contextzLocalIterator._metrics_contexts-$($7$;$;$=$=! rc8||jSrR)rrrOs rrPzLocalIterator.__iter__ s ""rcR|t|jSrR)rnextrrOs r__next__zLocalIterator.__next__s% D'(((rc t|SrRrSrOs rrUzLocalIterator.__str__rVrcd|jdS)NzLocalIterator[r1r=rOs rrXzLocalIterator.__repr__s, ,,,,rr_zLocalIterator[U]cjfd}t|j|j|j|gz|jdzS)Nc30K|D]}|VdSrRr)rr,r_s rapply_transformz0LocalIterator.transform..apply_transforms32     rrfr=rrrrr!)rKr_rs ` rrbzLocalIterator.transformsY           !_$5 5^+     rr0cdkrfd}n ifd}ttjr |fd}|}tjjj|gzjdzS)Nr0c3K|D]o}t|tr|V 5|}dddn #1swxYwY|Vt|tsnRpdSrR)rrr)rr,resultr_rKs r apply_foreachz-LocalIterator.for_each..apply_foreach-s & &D!$(:;; &" &!%!6!6!8!822)+D222222222222222"(LLL#-f6H#I#I& % & !& & &s AA A c3Kg}tjjdi}|j}|D]}t|tr|VrDt |kr1tj|\}}tj|Ed{V||||r5tj|\}}tj|Ed{V|3dSdS)Nr) r>r:optionsrrr(rrr$) rcurr: remote_fnr,finishedr_rlrms rrz-LocalIterator.for_each..apply_foreach?s!/B/<<)<<"M 44D!$(:;;4" *9s3xx?/J/J,/HSMMMHc'*wx'8'88888888 99T??33331$'HSMMMHc"wx00000000011111rc3K|}d} |rB5dddn #1swxYwYd}t|}t|tsd}|Vo)NTF)rrr rr)rnew_itemr,r_rK unwrappeds radd_wait_hooksz.LocalIterator.for_each..add_wait_hooksRsYr]]  )!224411..000111111111111111#(88D%d,>??(#'JJJ sA  A A rnr=)rrON_FETCH_START_HOOK_NAMErrrr!)rKr_rlrmrrrs```` @rrkzLocalIterator.for_each(s a   & & & & & & &   1 1 1 1 1 1 1 2}= > > +%I       +M      !]O 3]*     rrcnfd}tjjj|gzjdzS)Nc3K|D]R}5t|ts |r|Vdddn #1swxYwYSdSrR)rrr)rr,r_rKs r apply_filterz*LocalIterator.filter..apply_filterks # #**,,##!$(:;;#rr$xx#" ############### # #s%A  A A rrr=r)rKr_r#s`` rrqzLocalIterator.filterjs^ # # # # # #       !\N 2[(     rr.rc rfd}t|j|j|j|gz|jddzS)Nc3Kg}|D]J}t|tr|V||t|kr|Vg}K|r|VdSdSrR)rrr$r()rrur,r.s r apply_batchz(LocalIterator.batch..apply_batchysE # #d$677#JJJJLL&&&5zzQ# "    rrvrwr=r)rKr.r&s ` rruzLocalIterator.batchxs`           ![M 1^q^^^+     rLocalIterator[T[0]]cdd}t|j|j|j|gz|jdzS)Nc3ZK|D]%}t|tr|V|D]}|V&dSrR)rr)rr,subitems r apply_flattenz,LocalIterator.flatten..apply_flattens\ & &d$677&JJJJ#'&&% &  & &rr}r=r)rKr+s rr|zLocalIterator.flattensM & & &      !]O 3\)     rrrc tj|fd}t|j|j|j|gz|jd|t|ndzS)a:Shuffle items of this iterator Args: shuffle_buffer_size: The algorithm fills a buffer with shuffle_buffer_size elements and randomly samples elements from this buffer, replacing the selected elements with new elements. For perfect shuffling, this argument should be greater than or equal to the largest iterator size. seed: Seed to use for randomness. Default value is None. Returns: A new LocalIterator with shuffling applied c 3Kg}|D]}t|tr|V||t|kr;|dt|dz Vt|dkrP|dt|dz Vt|dkNdSdS)Nrr0)rrr$r(rrandint)rbufferr,rshuffle_randoms r apply_shufflez,LocalIterator.shuffle..apply_shufflesF U Ud$677UJJJJMM$'''6{{&999$jj)?)?3v;;QR?)S)STTTTTf++//jj!7!73v;;?!K!KLLLLLf++//////rz).shuffle(shuffle_buffer_size={}, seed={})Nrr=) randomRandomrrrrr!r%r)rKrrr1r0s ` @rrzLocalIterator.shuffles t,, M M M M M M      !]O 39@@#$2BSYYY     rcr||}|jdz|_|S)Nrrrs rrzLocalIterator.combines2 ]]2   & & ( ()l* rcTd}||}|jdz|_|S)Ncrt}|jtd|j|fS)Nz'Could not identify source actor of item)rrrr)r,rs rzip_with_sourcez.zip_with_sources9#//11G$, !JKKK($. .rz.zip_with_source_actor())rkr!)rKr7rs rzip_with_source_actorz#LocalIterator.zip_with_source_actors8 / / / ]]? + +)88 rchg}|D],}||t||krn-|Sr)r$r()rKr.outr,s rrzLocalIterator.takesF  D JJt   3xx1}} rrcNd}|D]}t||dz }||krdS dS)rrr0N)print)rKr.r+r,s rrzLocalIterator.showsI   D $KKK FAAvv  rc j|dkrtdgt|D](}tj)fdfd}g}t|D]B}|t ||jgjd|dzC|S)aCopy this iterator `n` times, duplicating the data. The child iterators will be prioritized by how much of the parent stream they have consumed. That is, we will not allow children to fall behind, since that can cause infinite memory buildup in this operator. Returns: List[LocalIterator[T]]: child iterators that each have a copy of the data of this iterator. rzNumber of copies must be >= 2ch|_t}D]}||dSrR)rr r$)rr,qqueuesrKs r fill_nextz*LocalIterator.duplicate..fill_nexts?"DL::D    rcfd}|S)Nc3<K t}tdD}||krtVnStdkr |n#t$rYdSwxYwV)NTc34K|]}t|VdSrR)r()rr?s rrzJLocalIterator.duplicate..make_next..gen..s(!9!9Q#a&&!9!9!9!9!9!9rr)r(maxrrr)rmy_lenmax_lenrAr+r@s rgenz7LocalIterator.duplicate..make_next..gens 2 ^^F!!9!9&!9!9!999G''0222222vay>>Q..' ) ' 2 2 2 2#0''' &'$Qi//11111 2s$ A00 A>=A>r)r+rHrAr@s` r make_nextz*LocalIterator.duplicate..make_nexts. 2 2 2 2 2 2 2 Jrz .duplicate[r1r=)rr"r$rrrrr!)rKr.rrI iteratorsr+rAr@s` @@r duplicatezLocalIterator.duplicates q55<== =q / /A MM++-- . . . .             & q  A   IaLL'%71%7%7%77     rF) deterministicround_robin_weightsothersrLrMc |D]6}t|tstdt|7g |gt |z}t d|D}|rdnd}|r.9s/W/W/WQ0@/W/W/Wr)parentsNrzCLength of round robin weights must equal number of iterators total.c"g|] }|dkrdnd S)*rNr)rws rrz'LocalIterator.union..Bs$MMMAQ#XX4MMMrr0rc3BK tD]\}}|dkrd}nt|} t|D]2}t|}t |t r||Vn|V3a#t $r||fYwxYwsdS)NTrTd)r_randomized_int_castr"r rrrr)rweightrmax_pullrr,rs r build_unionz(LocalIterator.union..build_unionRs "&v,,44JFB}}#&#7#?#? 4!&x++A#'88D)$0BCC+#*#6*.JJJ %&* (444 vrl333334E% sAA55!BBz LocalUnion[rr1r=rR)rrrr&rr r(r#r$rrrjoinmapr) rKrLrMrNr parent_itersrrtimeoutsr+r[rs @rrzLocalIterator.unions( Y YBb-00 Y !WTRTXX!W!WXXX YvV , &/W/W,/W/W/WXXX'.$$Q  :&''3|+<+<<< 'NM9LMMMHHy3|#4#44H#$#L(9(9"9 |,,  EAr MM$"'$QK     c-v6677      *   EtEEtyyS&1A1A'B'BEEE     r)NNNr)rr'rRr))r'rrrr  threadinglocalrrrr r rrrrL staticmethodr rrrrrPr rUrXrrbrkrrqrur|rrr8rrrKfloatrrrrrrs? 1"9?$$L =A  & &HQK0 && &x C89 &  & & & &D 2 2 2 2\ 2%%%^###)))---  Hhqk]HQK%?@  EW     BF@ @ A36"@ @ @ @ @ D  1#t),  1C      s 7    *    % % 3% c% EW% % % % N(A3Q<05G   cd1gc7d#567777x$+/ M M M #M M "%[ M  M M M M M M rrc^eZdZdZdedefdZdZdZde fdZ d e d e fd Z d e d e de fd Z d S)r?zyWorker actor for a ParallelIterator. Actors that are passed to iter.from_actors() must subclass this interface. item_generatorrcfd|rfd}||_n|_g|_d|_d|_dS)aCreate an iterator worker. Subclasses must call this init function. Args: item_generator: A Python iterable or lambda function that produces a generator when called. We allow lambda functions since the generator itself might not be serializable, but a lambda that returns it can be. repeat: Whether to loop over the iterator forever. c:tr SSrR)callableresr make_iteratorz6ParallelIteratorWorker.__init__..make_iterators''' &%~'''%%rc3K t}|ur%tddz|D]}|VJ)NTzJCannot iterate over {0} multiple times.Please pass in the base iterable orzlambda: {0} instead.)rrr%)rr,rerjs rcyclez.ParallelIteratorWorker.__init__..cyclesz #mmoo..B^++(D4;;NKKL !###"  #rN)rerrdnext_ith_buffer)rKrerrlrjs ` @rrLzParallelIteratorWorker.__init__vs & & & & &  2 # # # # # ##(%''D  "/-//D  #rctfdt}|D]}||}| J|t|_dS)z(Implements ParallelIterator worker init.cjSrRri)rrKs rrez6ParallelIteratorWorker.par_iter_init..s 4+>rN)rr rrd)rKrrr_s` rrz$ParallelIteratorWorker.par_iter_inits^ >>>>  P P & &BBB>>2>>>>R rcL|j Jdt|jS)z.Implements ParallelIterator worker item fetch.Nmust call par_iter_init())rdr rOs rrz$ParallelIteratorWorker.par_iter_nexts)}((*E(((DM"""rrcg}|dkr)|||Stjd|zz}tj|krj ||n*#t$rt |dkrtYnwxYwtj|kj|S)zBatches par_iter_next.rMbP?)r$rtimerr()rKrrut_ends rrz*ParallelIteratorWorker.par_iter_next_batchs q== LL++-- . . .L ux/0ikkE!!  T//112222    u::??''D  ikkE!! s#'B $B21B2rr4c|j Jd|jtjt|_|j|}t |dkr|dSt|D]G} t|j}|j| |8#t$rYDwxYw|j|st|j|dS)z3Iterates in increments of step starting from start.Nrqr) rdrmr defaultdictrr(rr"r r$r)rKrr4 index_bufferjvals rpar_iter_slicez%ParallelIteratorWorker.par_iter_slices}((*E(((   '#.#:4#@#@D +E2 |  q ##A&& &4[[  t}--C(+2237777$D'. $###E*..q111s>4B33 C?Ccg}|dkr+|||||Stjd|zz}tj|krl ||||n*#t$rt |dkrtYnwxYwtj|kl|S)zBatches par_iter_slice.rrs)r$r{rtrr()rKrr4rrurus rrz+ParallelIteratorWorker.par_iter_slice_batchs q== LL,,T599 : : :L ux/0ikkE!!  T00u==>>>>    u::??''D  ikkE!! s%)B$B65B6N) r'rrrrrrLrrrrr{rrrrr?r?os '$s'$D'$'$'$'$R!!!### C"232s2222.SCrr?cft|}||z }tj|kr|dz }|S)Nr0)rr2) float_valuebase remainders rrXrXs8 {  Dd"I }""   KrceZdZdZdS)rzIndicates that a local iterator has no value currently available. This is used internally to implement the union() of multiple blocking local generators.N)r'rrrrrrrrs  DrrcXeZdZdZdeddeedgdffdZdZdZd S) rGz.s+NNNQ''88NNNr)r>rrArOs`rrz_ActorSet.init_actorss0 NNNN$+NNNOOOOOrc>t|j|j|gzSrR)rGrArrhs rr[z_ActorSet.with_transformsdo&<===rN) r'rrrrrrLrr[rrrrGrGsyFF%,-%?"3_"DEF%%%%PPP>>>>>rrG)rF)FNrR)#rr2r`rt contextlibrtypingrrrrrr r>ray.util.annotationsr ray.util.iter_metricsr r r rrrr-r6r)r@rFrobjectr?rX ExceptionrrGrrrrs  %%%%%%BBBBBBBBBBBBBBBB ++++++???????? GCLL GCLL 8=<< 7< #<15<<<< <0 05 )- > >B#*#*Xa[!#*+/#*#*#*#* #*L 04PP ( )PPPP P" yHyHyHyHyHwqzyHyH yHx k k k k k GAJk k  k \  sssssVss sl        >>>>>>>>>>r