&`iw_jddlZddlZddlZddlZddlmZmZddlmZddl m Z ddl m Z ddl m Z ddlmZddlZddlmZmZ dd lmZmZmZmZmZn$#e$rejd ejd YnwxYwdd lmZdd lm Z m!Z!ddl"Z"ddl#m$Z$m%Z%ddl&m'Z'ddl(m)Z)m*Z*ej+Z,da-ee.Z/ej0Z1dZ2 d%dee3dee4dej5j6fdZ7dZ8dZ9dZ:dZ;dZdZ?d&dZ@d ZAe Gd!d"ZBd#ZCd$ZDdS)'N) OrderedDict defaultdict)Mapping) dataclass) ThreadPool)pprint)Optional) ishashableistask)AliasDataNodeTaskTaskRefconvert_legacy_graphzEDask on Ray is available only on dask>=2024.11.0, you are on version .) CPU_COUNT)_thread_get_idpack_exception)local_ray_callbacksunpack_ray_callbacks)unpack_object_refs) apply_sync get_asynczUse ray_remote_args={"resources": {...}} instead of resources={...} to specify required Ray task resources; see https://docs.ray.io/en/master/ray-core/package-ref.html#ray-remote.tasksTshuffleuse_shuffle_optimizationreturncf|rddlm}nd}tjt ||S)a Enable Dask-on-Ray scheduler. This helper sets the Dask-on-Ray scheduler as the default Dask scheduler in the Dask config. By default, it will also cause the task-based shuffle to be used for any Dask shuffle operations (required for multi-node Ray clusters, not sharing a filesystem), and will enable a Ray-specific shuffle optimization. >>> enable_dask_on_ray() >>> ddf.compute() # <-- will use the Dask-on-Ray scheduler. If used as a context manager, the Dask-on-Ray scheduler will only be used within the context's scope. >>> with enable_dask_on_ray(): ... ddf.compute() # <-- will use the Dask-on-Ray scheduler. >>> ddf.compute() # <-- won't use the Dask-on-Ray scheduler. Args: shuffle: The shuffle method used by Dask, either "tasks" or "disk". This should be "tasks" if using a multi-node Ray cluster. Defaults to "tasks". use_shuffle_optimization: Enable our custom Ray-specific shuffle optimization. Defaults to True. Returns: The Dask config object, which can be used as a context manager to limit the scope of the Dask-on-Ray scheduler to the corresponding context. r)dataframe_optimizeN schedulerrr)ray.util.dask.optimizationsrdaskconfigset ray_dask_get)rrrs k/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/util/dask/scheduler.pyenable_dask_on_rayr(*sN> "BBBBBBB! ;??DV   cFtjdddS)zH Unsets the scheduler, shuffle method, and DataFrame optimizer. Nr )r#r$r%r)r'disable_dask_on_rayr,Vs ;??T4D? Q QQr)c $|dd}|dd}tj}|t5|J|turAt 2t tatj t j t }nd|tvr#|t|vrt||}n8t |}tj |j |t||<dddn #1swxYwY|dd}|dd}|dd}d|vrtt|d i} tj} d| vrttt!|t"s,t%|d r|j}n|}t+|| | } t-|5}t/|\} } }}}}t1|}t3t5|jt8| | ||| t;|j||ft>t@d |}||D]}|||~|r|}n)d}|rtCj"d }tG|| }||D] }||dddn #1swxYwYt5tItj%}|tur^tMtD]I}||vrCt|'D]}| Jdddn #1swxYwY|S)a A Dask-Ray scheduler. This scheduler will send top-level (non-inlined) Dask tasks to a Ray cluster for execution. The scheduler will wait for the tasks to finish executing, fetch the results, and repackage them into the appropriate Dask collections. This particular scheduler uses a threadpool to submit Ray tasks. This can be passed directly to `dask.compute()`, as the scheduler: >>> dask.compute(obj, scheduler=ray_dask_get) You can override the currently active global Dask-Ray callbacks (e.g. supplied via a context manager), the number of threads to use when submitting the Ray tasks, or the threadpool used to submit Ray tasks: >>> dask.compute( obj, scheduler=ray_dask_get, ray_callbacks=some_ray_dask_callbacks, num_workers=8, pool=some_cool_pool, ) Args: dsk: Dask graph, represented as a task DAG dictionary. keys (List[str]): List of Dask graph keys whose values we wish to compute and return. ray_callbacks (Optional[list[callable]]): Dask-Ray callbacks. num_workers (Optional[int]): The number of worker threads to use in the Ray task submission traversal of the Dask graph. pool (Optional[ThreadPool]): A multiprocessing threadpool to use to submit Ray tasks. Returns: Computed values corresponding to the provided keys. num_workersNpool ray_callbacks ray_persistF_ray_enable_progress_bar resourcesray_remote_args_optimized_dsk)get_idr_dask_on_ray_pb)progress_bar_actor)(pop threadingcurrent_thread pools_lock main_thread default_poolrratexitregisterclosepools ValueErrorTOP_LEVEL_RESOURCES_ERR_MSGr#get_annotations isinstancerhasattrr5__dask_graph__!_build_key_scoped_ray_remote_argsrrrr_apply_async_wrapper apply_async_rayify_task_wrapperlen_poolrrray get_actorray_get_unpackr% enumeratelistvalues)dskkeyskwargsr.r/threadr0persistenable_progress_barr4 annotationsscoped_ray_remote_argsray_presubmit_cbsray_postsubmit_cbsray_pretask_cbsray_posttask_cbsray_postsubmit_all_cbsray_finish_cbs object_refscbresultpb_actoractive_threadstps r'r&r&]sJ**]D11K ::fd # #D  % ' 'F |  2 2"v'<'<'#-i#8#8LOL$6777#5[E&M%A%AV}[1!+.. +++-1f k* 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2JJ55Mjj..G **%?FFf4555jj!2B77O&((Kk!!4555 c7 # #' 3( ) ) '$CC$$&&C> [/ ] + +0} ! / /     " #3''  $!" &    OO   ")    " " -, % %;$$$$   N FFH" <=):;;#KHMMMF  %$  6 a000000000000000f ""Y02233  $ $%[[ " "N**"YYq\\0022"" """"""""""""""" Ms9 B1DD  D  CKK#&K#1BNN  N c dfd }|S)a Wraps the given pool `apply_async` function, hotswapping `real_func` in as the function to be applied and adding `extra_args` and `extra_kwargs` to `real_func`'s call. Args: apply_async: The pool function to be wrapped. real_func: The real function that we wish the pool apply function to execute. *extra_args: Extra positional arguments to pass to the `real_func`. **extra_kwargs: Extra keyword arguments to pass to the `real_func`. Returns: A wrapper function that will ignore it's first `func` argument and pass `real_func` in its place. To be passed to `dask.local.get_async`. r+Nc F|si}|zt|fi|S)N)argskwdscallback)dict)funcrlrmrnrK extra_args extra_kwargs real_funcs r'wrapperz%_apply_async_wrapper..wrappersJ D{  "d++l++     r))r+NNr+)rKrsrqrrrts```` r'rJrJs:$          Nr)c  ||\} } t| || |||| | |i} |}|| |f} d}n%#t$r}|||} d}Yd}~nd}~wwxYw|| |fS)a The core Ray-Dask task execution wrapper, to be given to the thread pool's `apply_async` function. Exactly the same as `execute_task`, except that it calls `_rayify_task` on the task instead of `_execute_task`. Args: key: The Dask graph key whose corresponding task we wish to execute. task_info: The task to execute and its dependencies. dumps: A result serializing function. loads: A task_info deserializing function. get_id: An ID generating function. pack_exception: An exception serializing function. ray_presubmit_cbs: Pre-task submission callbacks. ray_postsubmit_cbs: Post-task submission callbacks. ray_pretask_cbs: Pre-task execution callbacks. ray_posttask_cbs: Post-task execution callbacks. scoped_ray_remote_args: Ray task options for each key. Returns: A 3-tuple of the task's key, a literal or a Ray object reference for a Ray task's result, and whether the Ray task submission failed. FTN) _rayify_taskget BaseException)key task_infodumpsloadsr6rr]r^r_r`r\taskdepsreidfailedes r'rLrL sHU9%% d        " & &sB / /   VXX|$$ 5))  sAA A6A11A6c ttrfdDStr fdD}|D]} | | cS ttr1j} t| t r | jS| Sttrj} ntdtzt\} } tj ddt| tsdn| jdj| g| R}D]}|||St#sSvrSS) a Rayifies the given task, submitting it as a Ray task to the Ray cluster. Args: task: A Dask graph value, being either a literal, dependency key, Dask task, or a list thereof. key: The Dask graph key for the given task. deps: The dependencies of this task. ray_presubmit_cbs: Pre-task submission callbacks. ray_postsubmit_cbs: Post-task submission callbacks. ray_pretask_cbs: Pre-task execution callbacks. ray_posttask_cbs: Post-task execution callbacks. ray_remote_args: Ray task options. See :func:`ray.remote` for details. Returns: A literal, a Ray object reference representing a submitted task, or a list thereof. c <g|]}t|Sr+)rv) .0rhr~ryr^r`r]r_r4s r' z _rayify_task..csM     !"       r)Nc*g|]}|Sr+r+)rrdr~ryr}s r'rz _rayify_task..ts' Q Q QD#t!4!4 Q Q Qr)zInvalid task type: %szdask:)name num_returnsr+)rFrSr r targetrryrrprCtyperdask_task_wrapperoptionsMultipleReturnFuncrremoter )r}ryr~r]r^r_r`r4alternate_returnsalternate_returnrrparg_object_refsrepackrcrds```````` r'rvrvDs;8$H                8  ( Q Q Q Q Q Q?P Q Q Q $5 , , $/++++0 dE " " C[F&'** $FJ''F|# d # # C9DD4tDzzABB B #5T":":'/  #D*<==S4CS                  )( 1 14dK0000    Dz r)c |fd|D}|\} fd  |}||}|%t||D]\} } | | || |S)ar A Ray remote function acting as a Dask task wrapper. This function will repackage the given `arg_object_refs` into its original `deps` using `repack`, and then pass it to the provided Dask Task object , `task`. Args: task: The Dask Task class object to execute. repack: A function that repackages the provided args into the original (possibly nested) Python objects. key: The Dask key for this task. ray_pretask_cbs: Pre-task execution callbacks. ray_posttask_cbs: Post-task execution callback. *arg_object_refs (ObjectRef): Ray object references representing the dependencies' results. Returns: The output of the Dask task. In the context of Ray, a dask_task_wrapper.remote() invocation will return a Ray object reference representing the Ray task's result. Nc0g|]}| |ndSNr+)rrdrrys r'rz%dask_task_wrapper..s>   )+BBsO $ $ $D   r)cPt|tr|j|_|St|tr fd|DSt|t rt fd|DSt|t jrt j|St|trPt|j t jr/t j|j }t|j |S|S|S)Nc.i|]\}}||Sr+r+)rkv_dereference_argss r' z@dask_task_wrapper.._dereference_args..s+BBB1A((++BBBr)c3.K|]}|VdSrr+)rxrs r' z?dask_task_wrapper.._dereference_args..s/99!**1--999999r))ryvalue) rFrrlritemstuplerO ObjectRefrwr rry)rrrs r'rz,dask_task_wrapper.._dereference_argss a   &&qv..AFH 7 # # BBBB BBB B 5 ! ! 9999q99999 9 3= ) ) 71::  8 $ $ !'3=11 8((AE7777HHr))zip) r}rryr_r`r pre_states repacked_depsrerd pre_staters ` ` @r'rrs0"     %   vo..]$  T " "D T- F# !1:>> + +MB ~3 *** Mr)c:ddlm}tj|j\}}d}||d}|dg}||krtj|j\}}|||z |}tj|dt|d\}} t|t|krntj d||k| tj|j\}}||krtdttj|jdS) Nr)tqdm)totalpositionF)timeoutr fetch_localg?z)Completed. There was state inconsistency.)rrOrwrerset_descriptionupdatewaitrMtimesleeprAprintrreport) trackerrcrrfinishedreported_finished_so_farpb_bar ready_refs submitted_s r'render_progress_barrsggn335566OE8  T * * *F 2J U  !ggn&;&;&=&=>> 8 h!99:::#+  K0@0@e   A z??c+.. . .  3 U   LLNNN''."7"7"9"9::IxH 9::: 377>((** + +,,,,,r)c fd}t|trt|}t|tr;td|Dr"t |\}}||}||S||S)am Unpacks object references, gets the object references, and repacks. Traverses arbitrary data structures. Args: object_refs: A (potentially nested) Python object containing Ray object references. Returns: The input Python object with all contained Ray object references resolved with their concrete values. cPrt|tj|Sr)rrOrw)rcr8s r' get_resultz"ray_get_unpack..get_results-  A  2K @ @ @w{###r)c3LK|]}t|tj V dSr)rFrOr)rrs r'rz!ray_get_unpack..sB---.Jq#- ( ((------r))rFrrSanyr)rcr8rrcomputed_results ` r'rQrQs$$$$$ +u%%(;'' +t$$ '--2=---** '1+> V$*[11vo&&&z+&&&r)c |dd}|dd}t|5}t|\}}}}} } t|}t t t t||||d||fi|} | | D]} | | |~|r| } nt| } | | D] } | | | cdddS#1swxYwYdS)a A synchronous Dask-Ray scheduler. This scheduler will send top-level (non-inlined) Dask tasks to a Ray cluster for execution. The scheduler will wait for the tasks to finish executing, fetch the results, and repackage them into the appropriate Dask collections. This particular scheduler submits Ray tasks synchronously, which can be useful for debugging. This can be passed directly to `dask.compute()`, as the scheduler: >>> dask.compute(obj, scheduler=ray_dask_get_sync) You can override the currently active global Dask-Ray callbacks (e.g. supplied via a context manager): >>> dask.compute( obj, scheduler=ray_dask_get_sync, ray_callbacks=some_ray_dask_callbacks, ) Args: dsk: Dask graph, represented as a task DAG dictionary. keys (List[str]): List of Dask graph keys whose values we wish to compute and return. Returns: Computed values corresponding to the provided keys. r0Nr1Fr) r9rrrrrJrrLrQ)rUrVrWr0rYr]r^r_r`rarbrcrdres r'ray_dask_get_syncr(s<JJ55Mjj..G ] + +,} ! / /     " #3'' $!"             " -, % %;$$$$   1 FF#K00F  %$  6 Y,,,,,,,,,,,,,,,,,,sB CCCc*eZdZUeed<eed<dZdS)rrprc|j|i|ttsttr(fdt t DS)Nc g|] }| Sr+r+)rrreturnss r'rz/MultipleReturnFunc.__call__..s???awqz???r))rprFrorrangerM)selfrlrWrs @r'__call__zMultipleReturnFunc.__call__}sj$)T,V,, gt $ $ @ 7K(H(H @????5W+>+>???Gr)N)__name__ __module__ __qualname__callable__annotations__intrr+r)r'rrxs: NNNr)rc||Srr+)multiple_returnsidxs r'multiple_return_getrs C  r)cttjjs4tjjt di}fdD}|D]\}}|j}||}nd|vrtt| D]Y}| } | || | |i| ||<Zi} |D]G\}}| } | | di| | |<H| S)Nr+) dependenciesc.g|]}|j|fSr+)layers)rrrUs r'rz5_build_key_scoped_ray_remote_args..s% J J J4tSZ%& J J Jr)r3r4)rFr#highlevelgraphHighLevelGraphfrom_collectionsr_toposort_layersr[rCrDget_output_keyscopyrrwr) rUr[r4scoped_annotationsrid_layerlayer_annotationsrylayer_annotations_for_keyr\layer_ray_remote_argss ` r'rIrIs c4.= > > !0AA sGGSrB   J J J J33G3G3I3I J J JF @ @ U!-  $ +   - - -899 9((** @ @C(3(8(8(:(: % % , ,-> ? ? ? % , ,-?-C-CC-L-L M M M&? s # #  @ .4466<<[ / 4 4 6 6 $$[__5F%K%KLLL&;s## !!r))rTr)Er?r:rwarnings collectionsrrcollections.abcr dataclassesrmultiprocessing.poolrrtypingr r# dask.corer r dask._task_specr r rrr ImportErrorwarn __version__ dask.systemr dask.threadedrrrOray.util.dask.callbacksrrray.util.dask.commonrray.util.dask.scheduler_utilsrrr;r=r>rorBLockr<rDstrboolr$r%r(r,r&rJrLrvrrrrQrrrrIr+r)r'rsK  00000000######!!!!!!++++++ ((((((((TTTTTTTTTTTTTTTHM 2". 2 2 2 "!!!!!88888888 MMMMMMMM333333????????&i&((   D Y^   J%/3)) c])&tn) [_))))XRRRLLL^>666rdddN88 8v---:!'!'!'!'HMMM`  !!!"""""sAA43A4