&`i)ddlZddlmZmZddlmZddlmZmZmZddl m Z ddl Z dZ e de DZdd hZed d e Zed d d e DZGdde ZGddZdZdZejddZGddeZdS)N) defaultdict namedtuple)datetime)AnyListOptional)Callback) ray_presubmitray_postsubmit ray_pretask ray_posttaskray_postsubmit_all ray_finishc# K|] }d|zV dS)_N.0fields k/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/util/dask/callbacks.py rs&//%#+//////r r RayCallback RayCallbackscg|]}|dzS)_cbsrrs r r s3T3T3TuEFN3T3T3TrceZdZdZeZfdZedZfdZ fdZ fdZ fdZ de efd Zd ejfd Zd eejfd ZdZd eejfdZdZxZS)RayDaskCallbacka Extends Dask's `Callback` class with Ray-specific hooks. When instantiating or subclassing this class, both the normal Dask hooks (e.g. pretask, posttask, etc.) and the Ray-specific hooks can be provided. See `dask.callbacks.Callback` for usage. Caveats: Any Dask-Ray scheduler must bring the Ray-specific callbacks into context using the `local_ray_callbacks` context manager, since the built-in `local_callbacks` context manager provided by Dask isn't aware of this class. c tD].}||d}|t|d|z|/tjdi|dS)Nrr)CBSpopsetattrsuper__init__)selfkwargscbcb_func __class__s rr&zRayDaskCallback.__init__4se 1 1BjjT**G"cBh000""6"""""rc8tfdtDS)Nc2g|]}t|dSN)getattr)rrr's rrz1RayDaskCallback._ray_callback..>s%OOOEWT5$77OOOr)r CB_FIELDSr's`r _ray_callbackzRayDaskCallback._ray_callback<s#OOOOYOOOPPrct||_|jt|Sr.)add_ray_callbacks_ray_cm __enter__r%r'r+s rr6zRayDaskCallback.__enter__@sA(..       rcRtj||jj|dSr.)r%__exit__r5)r'argsr+s rr9zRayDaskCallback.__exit__Fs/$ t$$$$rct|j|jt dSr.)type ray_activeaddr2r%registerr7s rr?zRayDaskCallback.registerJs? T !!$"4555 rct|j|jt dSr.)r<r=remover2r% unregisterr7s rrBzRayDaskCallback.unregisterNs? T $$T%7888 rreturncdS)atRun before submitting a Ray task. If this callback returns a non-`None` value, Ray does _not_ create a task and uses this value as the would-be task's result value. Args: task: A Dask task, where the first tuple item is the task function, and the remaining tuple items are the task arguments, which are either the actual argument values, or Dask keys into the deps dictionary whose corresponding values are the argument values. key: The Dask graph key for the given task. deps: The dependencies of this task. Returns: Either None, in which case Ray submits a task, or a non-None value, in which case Ray task doesn't submit a task and uses this return value as the would-be task result value. Nr)r'taskkeydepss r_ray_presubmitzRayDaskCallback._ray_presubmitRs * r object_refcdS)aRRun after submitting a Ray task. Args: task: A Dask task, where the first tuple item is the task function, and the remaining tuple items are the task arguments, which are either the actual argument values, or Dask keys into the deps dictionary whose corresponding values are the argument values. key: The Dask graph key for the given task. deps: The dependencies of this task. object_ref: The object reference for the return value of the Ray task. Nrr'rErFrGrIs r_ray_postsubmitzRayDaskCallback._ray_postsubmitis  r object_refscdS)a"Run before executing a Dask task within a Ray task. This method executes after Ray submits the task within a Ray worker. Ray passes the return value of this task to the _ray_posttask callback, if provided. Args: key: The Dask graph key for the Dask task. object_refs: The object references for the arguments of the Ray task. Returns: A value that Ray passes to the corresponding _ray_posttask callback, if the callback is defined. Nrr'rFrMs r _ray_pretaskzRayDaskCallback._ray_pretaskzs rcdS)aRun after executing a Dask task within a Ray task. This method executes within a Ray worker. This callback receives the return value of the _ray_pretask callback, if provided. Args: key: The Dask graph key for the Dask task. result: The task result value. pre_state: The return value of the corresponding _ray_pretask callback, if said callback is defined. Nrr'rFresult pre_states r _ray_posttaskzRayDaskCallback._ray_posttasks  rcdS)zRun after Ray submits all tasks. Args: object_refs: The object references for the output (leaf) Ray tasks of the task graph. dsk: The Dask graph. Nr)r'rMdsks r_ray_postsubmit_allz#RayDaskCallback._ray_postsubmit_alls  rcdS)a Run after Ray finishes executing all Ray tasks and returns the final result. Args: result: The final result (output) of the Dask computation, before any repackaging is done by Dask collection-specific post-compute callbacks. Nrr'rSs r _ray_finishzRayDaskCallback._ray_finishs  r)__name__ __module__ __qualname____doc__setr=r&propertyr2r6r9r?rBrrrHray ObjectRefrLrrPrUrXr[ __classcell__)r+s@rr r #so  J#####QQXQ %%%%% #    . 3=    " T#--@    $     tCM/B            rr c eZdZdZdZdZdS)r4cpd|D|_tj|jdS)Nc,g|]}t|Sr)normalize_ray_callback)rcs rrz.add_ray_callbacks.__init__..s!GGG033GGGr) callbacksr r=update)r'rjs rr&zadd_ray_callbacks.__init__s6GGYGGG"))$.99999rc|Sr.rr1s rr6zadd_ray_callbacks.__enter__s rcX|jD]!}tj|"dSr.)rjr r=discard)r'r:ris rr9zadd_ray_callbacks.__exit__s7 2 2A  & . .q 1 1 1 1 2 2rN)r\r]r^r&r6r9rrrr4r4sA:::22222rr4ct|tr|jSt|tr|St d)NzFCallbacks must be either 'RayDaskCallback' or 'RayCallback' namedtuple) isinstancer r2r TypeError)r)s rrhrhsI"o&&  B $ $   T   rc|r'tdtt|DStdgttzS)z>Take an iterable of callbacks, return a list of each callback.c3>K|]\}fd|DpdVdS)c@g|]}|sttv|Sr)r" CBS_DONT_DROP)rr)idxs rrz2unpack_ray_callbacks...s,FFFbFCH ,E,E,E,E,ErNr)rcbs_rvs @rrz'unpack_ray_callbacks..sUCGFFFdFFFN$rr)r enumerateziplenr")cbss runpack_ray_callbacksr|sX 0!*39!5!5  rdSXXo//rc#K|du}|r%tjtc}t_ |pdV|r|t_dSdS#|r |t_wxYw)z Allows Dask-Ray callbacks to work with nested schedulers. Callbacks will only be used by the first started scheduler they encounter. This means that only the outermost scheduler will use global callbacks. Nr)r r=r`)rjglobal_callbackss rlocal_ray_callbacksrs!D(T1@1KSUU- ?-3o2  3)2O & & & 3 3  3)2O & 2 2 2 2s AAc,eZdZdZdZdZdZdZdS)ProgressBarCallbackctjGdd} tjd|_tj|jjdS#t $ra|d|_tj|jjYdSwxYw)NcDeZdZdZdZdZdZdZdZdZ dZ d Z d S) 6ProgressBarCallback.__init__..ProgressBarActorc.|dSr._initr1s rr&z?ProgressBarCallback.__init__..ProgressBarActor.__init__ rc|D]"}|j||#||j|<|j||fdSr.)keysrGr> submittedsubmission_queueappend)r'rFrGnowdeps rsubmitz=ProgressBarCallback.__init__..ProgressBarActor.submitsc99;;,,CIcN&&s++++&)s#%,,c3Z88888rc||j|<dSr.) scheduledr'rFrs rtask_scheduledzEProgressBarCallback.__init__..ProgressBarActor.task_scheduleds&)s###rc||j|<dSr.)finishedrs rfinishz=ProgressBarCallback.__init__..ProgressBarActor.finishs%( c"""rcRt|jt|jfSr.)rzrrr1s rrSz=ProgressBarCallback.__init__..ProgressBarActor.results!4>**C ,>,>>>rc4tt}|jD]_\}}|j|}|j|}||z ||d<||z ||d<`|j|d<|S)Nexecution_timescheduling_timesubmission_order)rdictritemsrr total_secondsr)r'rSrFrrrs rreportz=ProgressBarCallback.__init__..ProgressBarActor.reports$T**%)]%8%8%:%:&&MC $s 3I $s 3I!9,#moo3K 01"I-#moo3K 122.2-B)* rcdSr.rr1s rreadyz.ProgressBarActor.readysrc.|dSr.rr1s rresetz.ProgressBarActor.resetrrcg|_td|_td|_td|_tt |_dSr.)rrrrrr`rGr1s rrz.ProgressBarActor._initsJ(*%!,T!2!2!,T!2!2 +D 1 1 ',, rN) r\r]r^r&rrrrSrrrrrrrProgressBarActorrs    9 9 9  * * * ) ) ) ? ? ?   (       - - - - -rr_dask_on_ray_pb)name) rbremote get_actorpbgetr ValueErroroptionsr)r'rs rr&zProgressBarCallback.__init__s 2 -2 -2 -2 -2 -2 -2 - 2 -h ,m$566DG GDGM((** + + + + + , , ,&..4E.FFMMOODG GDGM((** + + + + + + ,sA A,,A'CCcj|jj||tjdSr.)rrrrrrKs rrLz#ProgressBarCallback._ray_postsubmit&s* c488888rch|jj|tjdSr.)rrrrrrOs rrPz ProgressBarCallback._ray_pretask*s) %%c8<>>:::::rch|jj|tjdSr.)rrrrrrRs rrUz!ProgressBarCallback._ray_posttask-s( c8<>>22222rc$tddS)NzAll tasks are completed.)printrZs rr[zProgressBarCallback._ray_finish1s ()))))rN)r\r]r^r&rLrPrUr[rrrrrs`;,;,;,z999;;;333*****rrr.) contextlib collectionsrrrtypingrrrdask.callbacksr rbr"tupler0rujoinrrr r4rhr|contextmanagerrrrrrrs////////&&&&&&&&&&######  E//3/// / / / j 66 z.#((3T3TPS3T3T3T*U*UVV J J J J J hJ J J Z 2 2 2 2 2 2 2 2    0 0 0 3333"J*J*J*J*J*/J*J*J*J*J*r