&`iT-TdZddlZddlZddlmZmZddlZddlmZ ddlm Z m Z n$#e $rej dej dYnwxYwddlmZmZdd lmZmZmZdd lmZejd krd Znd ZddZdZddZdZdefdZdZdZdZ ddZ!dZ"dede e!de"e"fdZ#ddZ$dS) zu The following is adapted from Dask release 2021.03.1: https://github.com/dask/dask/blob/2021.03.1/dask/local.py N)EmptyQueue)config)DataNodeDependenciesMappingzEDask on Ray is available only on dask>=2024.11.0, you are on version .)local_callbacksunpack_callbacks)flattenget_dependencies reverse_dict)orderntcV |ddS#t$rYnwxYw))NTg?)blocktimeout)getrqs q/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/util/dask/scheduler_utils.py queue_getrsG  uu4u555     s  ''c*|SN)rrs rrr&suuwwc |t|j}|tjdd}|t}t |D]<\}}t |tr"|||<|=| }| |t|}fd|D}t|}|D]6} || dD]} ||  | 7d|D} d|D} t| |d} d |D}|||| || t t t d }|S) aStart state from a dask Examples -------- >>> dsk = { 'x': 1, 'y': 2, 'z': (inc, 'x'), 'w': (add, 'z', 'y')} # doctest: +SKIP >>> from pprint import pprint # doctest: +SKIP >>> pprint(start_state_from_dask(dsk)) # doctest: +SKIP {'cache': {'x': 1, 'y': 2}, 'dependencies': {'w': {'z', 'y'}, 'x': set(), 'y': set(), 'z': {'x'}}, 'dependents': {'w': set(), 'x': {'z'}, 'y': {'w'}, 'z': {'w'}}, 'finished': set(), 'ready': ['z'], 'released': set(), 'running': set(), 'waiting': {'w': {'z'}}, 'waiting_data': {'x': {'z'}, 'y': {'w'}, 'z': {'w'}}} Ncachec>i|]\}}|v |t|S)set).0kv data_keyss r z)start_state_from_dask..Ps/PPPTQQi=O=Oq#a&&=O=O=OrrcBi|]\}}|||Sr)copyr r!r"s rr$z)start_state_from_dask..Vs+DDDDAq!DAqvvxxDDDrch|] \}}|| Srrr's r z(start_state_from_dask..Xs!888tq!a8888rTkeyreverseci|] \}}||| Srrr's rr$z)start_state_from_dask..Zs#555115q!555r) dependencies dependentswaiting waiting_datarreadyrunningfinishedreleased)rrrdictritems isinstanceraddr&updaterr removesorted)dskrsortkeyr!r"dsk2r.r0r/abr1 ready_setr2stater#s @rstart_state_from_daskrD*s***. } 7D)) }I 1 a " " qssE!H MM!    88::DKK&s++LPPPP\%7%7%9%9PPPGl++J !!2&& ! !A AJ  a  !DDJ,<,<,>,>DDDL88w}}888I 9'4 8 8 8E55 555G% $55EEEE  E Lrc ||\}}||}|} ||| f}d} n%#t$r} || |}d} Yd} ~ nd} ~ wwxYw||| fS)zx Compute task and handle all administration See Also -------- _execute_task : actually execute task FTN) BaseException) r+ task_infodumpsloadsget_idpack_exceptiontaskdataresultidfailedes r execute_taskrRksU9%% dd VXX|$$ 5))  s25 AAATc||dvr|d|rJ|d|=|d||r |d|=dSdS)zQRemove data from temporary storage See Also -------- finish_task r1r5rN)r9)r+rCdeletes r release_datarU~st  eN###(---- . !# & *#  'N3     rFc t|d||dD]K}|d|}|||s$|d|=|d|L|d|D]}||dvr|d|}|||sj||vrftrQdd lm} t d ||tt| |d  d z fz|||| |r||vr|||| |d ||d||S)zn Update execution state after a task finishes Mutates. This should run atomically (with a lock). r/Tr*r0r2r.r1r)nbytesz&Key: %s Dep: %s NBytes: %.2f Releaserg.A)rTr4r3) r<r;appendDEBUG chest.corerWprintsummapvaluesr9) r=r+rCresultsr>rTrUdepsrWs r finish_taskrbseL)#.GTJJJ'' ) S !   'i % 'N ! !# & & &^$S)44 %' ' 'n%c*A HHSMMM 8G++111111CSVU7^5J5J5L5L)M)MPS)S%T%TUV S%7777  47** LeF 3 3 3 3 *# )C   Lrctt|trtfd|DS|S)zGet nested index from collection Examples -------- >>> nested_get(1, 'abc') 'b' >>> nested_get([1, 0], 'abc') ('b', 'a') >>> nested_get([[1, 0], [0, 1]], 'abc') (('b', 'a'), ('a', 'b')) c38K|]}t|VdSr) nested_get)r icolls r znested_get..s-66QZ4((666666r)r8listtuple)indrgs `rreresD#t6666#666666CyrcdS)zDefault get_idNrrrrdefault_get_idrms 4rcrr)rQrHs rdefault_pack_exceptionros rcB|j|ur|||r) __traceback__with_traceback)exctbs rreraiserus) ""  $$$ Irc|S)z>> identity(3) 3 r)xs ridentityrxs Hrc R "#$t#t|trtt |} n|h} t| }t t | 5} t| \}}"}}g}d}i$ | D]0}|dr|d||1t}t||j $| D]\}}}}}|r |$|tj dd}$dr$dstd  "#$f d }$drDt$d |kr+|$drt$d |k+$ds$ds $d r t#\}}}|rL |\}}|r0$fd t!|D}|}||n ||| |\}}|$d |<t#|$||j |D]} | ||$|$drDt$d |kr+|$drt$d |k+$d$d$d  d }|D]\}}}}}!|!r|!$| n #|D]\}}}}}!|!r|!$| wxYw dddn #1swxYwYt%|$d S)aAsynchronous get function This is a general version of various asynchronous schedulers for dask. It takes a an apply_async function as found on Pool objects to form a more specific ``get`` method that walks through the dask array with parallel workers, avoiding repeat computation and minimizing memory use. Parameters ---------- apply_async : function Asynchronous apply function as found on Pool or ThreadPool num_workers : int The number of active tasks we should have at any one time dsk : dict A dask dictionary specifying a workflow result : key or list of keys Keys corresponding to desired data cache : dict-like, optional Temporary storage of results get_id : callable, optional Function to return the worker id, takes no arguments. Examples are `threading.current_thread` and `multiprocessing.current_process`. rerun_exceptions_locally : bool, optional Whether to rerun failing tasks in local process to enable debugging (False by default) pack_exception : callable, optional Function to take an exception and ``dumps`` method, and return a serialized tuple of ``(exception, traceback)`` to send back to the scheduler. Default is to just raise the exception. raise_exception : callable, optional Function that takes an exception and a traceback, and raises an error. dumps: callable, optional Function to serialize task data and results to communicate between worker and parent. Defaults to identity. loads: callable, optional Inverse function of `dumps`. Defaults to identity. callbacks : tuple or list of tuples, optional Callbacks are passed in as tuples of length 5. Multiple sets of callbacks may be passed in as a list of tuples. For more information, see the dask.diagnostics documentation. See Also -------- threaded.get Fr)rr>Nrerun_exceptions_locallyr0r2z Found no accessible jobs in daskc , d} d| D]}||  fdt|D}t|||ff jdS)z"Fire off a task to the thread poolr2r3c.i|]}|d|Srrr r`rCs rr$z0get_async..fire_task..Cs$WWWSU7^C0WWWr)argscallbackN)popr9r rRput) r+frM apply_asyncr=rHrJrIrK pretask_cbsqueuerCs r fire_taskzget_async..fire_task:sGn((**i $$S)))$''AAc3&&&&XWWW.\s4    # w!4   rrT)rr8rirr r6r r rXrrDrr ValueErrorlenrr rbre)%r num_workersr=rNrrJrzrKraise_exception callbacksrHrIkwargs result_flatr__ posttask_cbs started_cbs succeededcbkeyorder start_staterr+res_inforPrsrtrMrLres worker_idrfinishrrrCs%` ` ` ` `` @@@r get_asyncrsr GGE&$'&//** h +G s))C  # #S6y-=i-H-H*1k<  L 6 ' 'a5BqE#JJJ""2&&&&SzzH)#UHLQQQE+4 , ,';1a,KU+++'/+1:6PRW+X+X(Y Eg E !CDDD             2. Sy)9%:%:[%H%H . Sy)9%:%:[%H%H " eGn i8H (1%(8(8%Xv 1#eHooGC/1    '7S'A'A    #3xT 'R000!&xY&)gs#CeWhlCCC%77AAc3UI6666Gn U9-=)>)>)L)LIKKKGn U9-=)>)>)L)L' " eGn i8H ,I'2 6 6"1aF6F39}555 6k 6 6"1aF6F39}555 6 6cS6S6S6S6S6S6S6S6S6S6S6S6S6S6S6j feGn - --s+:LH(K?LK88LL L rc<|i}||i|}| ||dSdS)z*A naive synchronous version of apply_asyncNr)funcrkwdsrrs r apply_syncrws? | $   C r)NN)Tr)rNN)%__doc__oswarningsrrrdaskrdask._task_specrr ImportErrorwarn __version__dask.callbacksr r dask.corer r r dask.orderrnamerrDrRrUrYrbrermrorurxrrrrrrs?   =========HM 2". 2 2 2 =<<<<<<<==========7d?? >>>>B&     /3!!!!H"          !)  W.W.W.W.ts'AA