&`i XddlZddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl m Z ddl m Z mZmZmZmZmZmZmZddlZddlmZddlmZ ddlmZddlmZmZn#e$r dZdZdZYnwxYwej e!Z"dZ#d e d eeej$fd ej$fd Z%d e d eee ej$fd ej$fdZ& d%d e d eeee ej$fd eeeej$fd ej$fdZ'd e d e fdZ(eGddeZ)ndZ)dZ*Gdde+Z,Gddej-Z.GddZ/GddZ0Gdde0Z1Gdde0Z2ej3d Gd!d"Z4Gd#d$Z5dS)&N) TimeoutError)AnyCallableDictHashableIterableListOptionalTuple) usage_lib)log_once) SafeFunction) BatchedCallsparallel_backend RAY_ADDRESSobjregistry_hashablereturncR||vrtj|}|||<n||}|SN)rayput)rrrets q/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/util/multiprocessing/pool.py_put_in_dict_registryr s: ###gcll!$#$ Jregistryc tfd|D}n;#t$r.tj}||fYnwxYw|S)Nc3*K|] \}}|u |VdSr).0orefrs r z(_put_in_list_registry../s+;;FAs!s((C((((;;r)next StopIterationrrappend)rrrs` r_put_in_list_registryr(+sw$;;;;h;;;<< $$$gcllc #####$ Js5AAct|tjstj|dkr|S|}|4 t ||}n4#t $r|t||}YnwxYw|t||}|S)zzray.put obj in object store if it's not an ObjRef and bigger than 100 bytes, with support for list and dict registriesd) isinstancer ObjectRefsys getsizeofr TypeErrorr()rrrrs rray_put_if_neededr06s#s}%%s););c)A)A C$ ;'->??CC ; ; ;#+C:: ;  #C22 JsA A*)A*cbt|tjrtj|S|S)z8If obj is an ObjectRef, do ray.get, otherwise return obj)r+rr,get)rs rray_get_if_neededr3Ks)#s}%%ws|| JrceZdZdZ ddeeeeej fdee e ej ffdZ dZ dZdS) RayBatchedCallszJoblib's BatchedCalls with basic Ray object store management This functionality is provided through the put_items_in_object_store, which uses external registries (list and dict) containing objects and their ObjectRefs.Nrrcg}|jD]N\}}}fd|D}fd|D}||||fO||_dS)aqPuts all applicable (kw)args in self.items in object store Takes two registries - list for unhashable objects and dict for hashable objects. The registries are a part of a Pool object. The method iterates through all entries in items list (usually, there will be only one, but the number depends on joblib Parallel settings) and puts all of the args and kwargs into the object store, updating the registries. If an arg or kwarg is already in a registry, it will not be put again, and instead, the cached object ref will be used.c2g|]}t|Sr r0)r!argrrs r z=RayBatchedCalls.put_items_in_object_store..ls4LO%c85FGGrc:i|]\}}|t|Sr r8)r!kvrrs r z=RayBatchedCalls.put_items_in_object_store..os=1(H6GHHrN)itemsr')selfrr new_itemsfuncargskwargss `` rput_items_in_object_storez)RayBatchedCalls.put_items_in_object_store[sI&*j 7 7"dFSW &   $f!56666"DJJJrct|j|j5d|jDcdddS#1swxYwYdS)N)n_jobsc rg|]4\}}}|d|Did|D5S)c,g|]}t|Sr r3)r!r9s rr:z7RayBatchedCalls.__call__...s!AAAS+C00AAArc4i|]\}}|t|Sr rJ)r!r<r=s rr>z7RayBatchedCalls.__call__...s'NNNtq!1/22NNNr)r?)r!rBrCrDs rr:z,RayBatchedCalls.__call__..sk +dF DAADAAANNv||~~NNNr)r_backend_n_jobsr?r@s r__call__zRayBatchedCalls.__call__vs"$- EEE   /3j                   s :>>c~|j|t|j|j|jfd|jffSr)_reducer_callbackr5r?rLrM _pickle_cacherNs r __reduce__zRayBatchedCalls.__reduce__sH%1&&((( dmT\:D$BTU rNN)__name__ __module__ __qualname____doc__r r r rrr,rrrErOrSr rrr5r5Ts ! !CGIM # #tE#s}*<$=>? # (Xs}-D(EF # # # #6   "     rr5c| |z Srr )abs r div_round_upr\sR1W:rceZdZdZdS) PoolTaskErrorc||_dSr) underlying)r@r`s r__init__zPoolTaskError.__init__s $rN)rUrVrWrar rrr^r^s#%%%%%rr^c teZdZdZdZ ddededededee f d Z d Z d Z d Z d ZdZdZddZdS) ResultThreadaThread that collects results from distributed actors. It winds down when either: - A pre-specified number of objects has been processed - When the END_SENTINEL (submitted through self.add_object_ref()) has been received and all objects received before that have been processed. Initialize the thread with total_object_refs = float('inf') to wait for the END_SENTINEL. Args: object_refs (List[RayActorObjectRefs]): ObjectRefs to Ray Actor calls. Thread tracks whether they are ready. More ObjectRefs may be added with add_object_ref (or _add_object_ref internally) until the object count reaches total_object_refs. single_result: Should be True if the thread is managing function with a single result (like apply_async). False if the thread is managing a function with a List of results. callback: called only once at the end of the thread if no results were errors. If single_result=True, and result is not an error, callback is invoked with the result as the only argument. If single_result=False, callback is invoked with a list of all the results as the only argument. error_callback: called only once on the first result that errors. Should take an Exception as the only argument. If no result errors, this callback is not called. total_object_refs: Number of ObjectRefs that this thread expects to be ready. May be more than len(object_refs) since more ObjectRefs can be submitted after the thread starts. If None, defaults to len(object_refs). If float("inf"), thread runs until END_SENTINEL (submitted through self.add_object_ref()) has been received and all objects received before that have been processed. NF object_refs single_resultcallbackerror_callbacktotal_object_refscxtj|dd|_g|_d|_g|_tj|_ ||_ ||_ ||_ |pt||_i|_tj|_|D]}||dS)NT)daemonFr) threadingThreadra _got_error _object_refs _num_ready_resultsqueueQueue_ready_index_queue_single_result _callback_error_callbacklen_total_object_refs_indices_new_object_refs_add_object_ref)r@rdrerfrgrh object_refs rrazResultThread.__init__s !!$t!444 "'+--+!-"3"Gs;7G7G !& % - -J   , , , , - -rct|j|j|<|j||jddSr)rwrnryr'rpr@r|s rr{zResultThread._add_object_refsL$'(9$:$: j!   ,,, T"""""rc:|j|dSr)rzrr~s radd_object_refzResultThread.add_object_refs !!*-----rcntj|j}g}|j|jkrd}| t |dk}|j|}||jurt |j|_n*||| |n#tj $rYnwxYwtj |dd\}}t |dkr|d}| tj|}n$#tjj$r }|g}Yd}~nd}~wwxYw|jsT|D]Q} t#| t$r%d|_|j|| n| | R|xjdz c_||j|j|<|j|j||j|jk|jsB|j=|js||dS||ddSdSdS)Nr)blockg? num_returnstimeoutT)copyrnrorxrwrzr2 END_SENTINELr{r'rqEmptyrwait exceptionsRayErrorrmr+ Exceptionrvrpryrsrrurt) r@unreadyaggregated_batch_resultsready_idrnew_object_refreadybatcheresults rrunzResultThread.runs)D-..#% o 777H" LLA-E%)%:%>%>U%>%K%KN%):::36d6G2H2H//,,^<<<~666{D"%'q#!N!N!Nwu::>>$QxH/"2 ))>*     ? @#@@F!&)44@*./; 00888077???? OOq OO5:DM$-1 2  # ' ' h(? @ @ @ao 777r <4>#=& <788888 7:;;;;; < <#=#=s*A;B..C?C;DD1$D,,D1c|jSr)rmrNs r got_errorzResultThread.got_error2s rc|j|Srrp)r@indexs rrzResultThread.result6s}U##rc|jSrrrNs rresultszResultThread.results:s }rcn |j|S#tj$rtwxYw)Nr)rsr2rqrrr@rs rnext_ready_indexzResultThread.next_ready_index>sD *..w.?? ?{     s4)FNNNr)rUrVrWrXrlistboolcallabler intrar{rrrrrrr rrrcrcs""HL $!#'+/ ---- - ! - $C= ----2### ...G<G<G0.zrTried to start a pool with {} processes on an existing ray cluster, but there are only {} CPUs in the ray cluster.r )ris_initializedRAY_ADDRESS_ENVosenviron_privateutilsread_ray_addressr2initrstatecluster_resourcesrformat)r@rr init_kwargsray_cpuss rr zPool._init_ray`sR!## -"2:--<%6688D :>>/22g==.7K +'';''''( ')).7K +<<< <<<<),,,,s|);;==eDEE   I >>@AA A i  ++16)X+F+F  rcd_fdt|D_tjdjDdS)Nc8g|]}Sr )_new_actor_entry)r!rr@s rr:z*Pool._start_actor_pool..s%NNND1133NNNrcFg|]\}}|jSr )rremote)r!actorrs rr:z*Pool._start_actor_pool..s*FFF""$$FFFr)r rrrr2)r@rs` rr zPool._start_actor_poolsWNNNNU9=M=MNNN FFT5EFFFGGGGGrct|jdkrdS|t|}tj|jt|j|\}}||_dS)Nrr)rwrrrr)r@rrdeletings r_wait_for_stopping_actorszPool._wait_for_stopping_actorsso t' ( (A - - F  GnnGh  $D455   8 $,   rc|d|j|jdS)Ngr)r'rr'__ray_terminate__r#)r@r$s r _stop_actorzPool._stop_actorsG &&s&333  ''(?(F(F(H(HIIIIIrc|jstjdi|j|_|j|j|jdfS)Nrr )r roptionsrr#rrrNs rr!zPool._new_actor_entrysO  J(0II43HIID  ''(94>JJANNrc|jt|jdz krd|_n|xjdz c_|jS)Nrr)rrwrrNs r_next_actor_indexzPool._next_actor_indexsH  #d&6"7"7!"; ; ;"#D     1 $  ""rc|j|\}}|j||}|dz }|jdks ||jksJ||jkr,|||\}}||f|j|<|S)Nrr)rrr#rr*r!)r@rrBrr$countr|s r _run_batchzPool._run_batchs' 4 u_++D%88   %++u8N/N/N/NN D* * *   U # # #0022LE5).%rrBrCrDcT||||S)a8Run the given function on a random actor process and return the result synchronously. Args: func: function to run. args: optional arguments to the function. kwargs: optional keyword arguments to the function. Returns: The result. ) apply_asyncr2)r@rBrCrDs rapplyz Pool.applys($dF3377999rrfrgc|||}|||||fg}t |g||dS)aRun the given function on a random actor process and return an asynchronous interface to the result. Args: func: function to run. args: optional arguments to the function. kwargs: optional keyword arguments to the function. callback: callback to be executed on the result once it is finished only if it succeeds. error_callback: callback to be executed the result once it is finished only if the task errors. The exception raised by the task will be passed as the only argument to the callback. Returns: AsyncResult containing the result. T)re)_check_running'_convert_to_ray_batched_calls_if_neededr1r.r)r@rBrCrDrfrgr|s rr3zPool.apply_asyncsk2 ;;DAA__T%;%;%=%=ttVnEUVV J<>QUVVVVrrct|S|}t|tr|j}t|trNt|j|j|jf|j|j }| |j |j n|}|S)aConvert joblib's BatchedCalls to RayBatchedCalls for ObjectRef caching. This converts joblib's BatchedCalls callable, which is a collection of functions with their args and kwargs to be ran sequentially in an Actor, to a RayBatchedCalls callable, which provides identical functionality in addition to a method which ensures that common args and kwargs are put into the object store just once, saving time and memory. That method is then ran. If func is not a BatchedCalls instance, it is returned without changes. The ObjectRefs are cached inside two registries (_registry and _registry_hashable), which are common for the entire Pool and are cleaned on close.) r5r+rrBrr?rLrMrQrRrErr)r@rB orginal_funcs rr7z,Pool._convert_to_ray_batched_calls_if_neededs  "K  dL ) ) 9D dL ) ) "  -&" D  * *4>4;R S S S SD rc~tt|t|jdz\}}|r|dz }|S)Nr)divmodrwr)r@rrextras rrzPool._calculate_chunksizesC!#h--T5E1F1F1JKK 5   NIrFc$g}t||krP t|}|s|f}||ifn#t$rYnwxYwt||kPt|dksJ||||Sr)rwr%r'r&r1)r@rBiteratorrr unpack_argschunkrCs rrzPool._submit_chunks%jj9$$ H~~"# 7D dBZ((((      %jj9$$5zzA~~~~{D%888s+A AAc t|dst|}|||}t|}g}t ||zt |krtt |t |jz}|||||||t ||zt |kt|S)N__len__)r@)hasattrrrrrwrr'r)r@rBrrr@r?rrs r_chunk_and_runzPool._chunk_and_run&sx++ &H~~H  11(;;I>>#$$y03x==@@/003t7G3H3HHK  $ $""(I{ #   #$$y03x==@@! rc~||||||}t|||S)Nrr@)r6rEr)r@rBrrr@rfrgrds r _map_asynczPool._map_async9sM )) (i[*  ;.AAArrrcX||||dS)aRun the given function on each element in the iterable round-robin on the actor processes and return the results synchronously. Args: func: function to run. iterable: iterable of objects to be passed as the sole argument to func. chunksize: number of tasks to submit as a batch to each actor process. If unspecified, a suitable chunksize will be chosen. Returns: A list of results. FrGrHr2r@rBrrs rmapzPool.mapHs1 (iU  #%% rc8||||d||S)aRun the given function on each element in the iterable round-robin on the actor processes and return an asynchronous interface to the results. Args: func: function to run. iterable: iterable of objects to be passed as the only argument to func. chunksize: number of tasks to submit as a batch to each actor process. If unspecified, a suitable chunksize will be chosen. callback: Will only be called if none of the results were errors, and will only be called once after all results are finished. A Python List of all the finished results will be passed as the only argument to the callback. error_callback: callback executed on the first errored result. The Exception raised by the task will be passed as the only argument to the callback. Returns: AsyncResult F)rr@rfrgrH)r@rBrrrfrgs r map_asynczPool.map_async[s1:  )    rcX||||dS)zSame as `map`, but unpacks each element of the iterable as the arguments to func like: [func(*args) for args in iterable]. TrGrJrKs rstarmapz Pool.starmaps1  (iT  #%% rc6|||d||S)zSame as `map_async`, but unpacks each element of the iterable as the arguments to func like: [func(*args) for args in iterable]. T)r@rfrgrN)r@rBrrfrgs r starmap_asynczPool.starmap_asyncs.  )    rrcP|t||||S)aSame as `map`, but only submits one batch of tasks to each actor process at a time. This can be useful if the iterable of arguments is very large or each task's arguments consumes a large amount of resources. The results are returned in the order corresponding to their arguments in the iterable. Returns: OrderedIMapIterator r)r6rrKs rimapz Pool.imaps- "4x9MMMMrcP|t||||S)agSame as `map`, but only submits one batch of tasks to each actor process at a time. This can be useful if the iterable of arguments is very large or each task's arguments consumes a large amount of resources. The results are returned in the order that they finish. Returns: UnorderedIMapIterator rU)r6rrKs rimap_unorderedzPool.imap_unordereds- $T4YOOOOrc2|jrtddS)NzPool not running)rrrNs rr6zPool._check_runnings% < 1/00 0 1 1rc.||Sr)r6rNs r __enter__zPool.__enter__s  rc.|dSr) terminate)r@exc_typeexc_valexc_tbs r__exit__z Pool.__exit__s rc|j|j|jD]\}}||d|_t jdS)zClose the pool. Prevents any more tasks from being submitted on the pool but allows outstanding work to finish. TN)rclearrrr*rgccollectr@r$rs rclosez Pool.closesp  %%'''( $ $HE1   U # # # #  rc~|js||jD]\}}tj|dS)z~Close the pool. Prevents any more tasks from being submitted on the pool and stops outstanding work. N)rrgrrkillrfs rr]zPool.terminatesJ|  JJLLL(  HE1 HUOOOO  rcZ|jstd|dS)zWait for the actors in a closed pool to exit. If the pool was closed using `close`, this will return once all outstanding work is completed. If the pool was closed using `terminate`, this will return quickly. zPool is still runningN)rrr'rNs rrz Pool.joins5| 6455 5 &&(((((r)NNNNNNNrTr)NNNN)F)NF)NFNN)NNN)r))rUrVrWrXr rrrrstrrrar r r'r*r!r.r1r r4rr3r7rrrErHrLr rOrQrSrVrXr6r[rargr]rr rrrr+sd*$(*.'+*.%)48**C=*h'*8$ * #3- *  *c]*"$sCx.1****B####JHHH , , , ,JJJOOO###   !%!% :::uo: ::::.!%!%*.6: WWWuoW W C5$;' W !)d!23 WWWW<"H"""""H 9999 !!!!. B B B BH#.$(+/6: $ $ $ $ C= $ D64<( $ !)d!23 $ $ $ $ L,06:     D64<(  !)d!23     &NNNXN(3-NNNN$NOPPP(0P=Ec]PPPP$111       ) ) ) ) )rrrT)6rrrdrloggingrrqr-rkrmultiprocessingrtypingrrrrrr r r rray._common.usager ray.utilr joblib._parallel_backendsrjoblib.parallelrr ImportError getLoggerrUr rr,rr(r0r3r5r\rr^rlrcrrrrr#rrr rrrusC   ((((((QQQQQQQQQQQQQQQQQQQQ ''''''666666>>>>>>>>>LLLL  8 $ $ !%h &=!>] U3 #567];?AE tE#s}"4567 Xs}%< => ] *33=====,====@O%%%%%I%%% bbbbb9#bbbJ=3=3=3=3=3=3=3=3@B"B"B"B"B"B"B"B"J&-&-&-&-&-,&-&-&-R-----L---: Q2E)E)E)E)E)E)E)E)E)E)sA## A10A1