&`i1=ddlZddlZddlZddlZddlmZddlmZddlm Z m Z m Z m Z m Z mZmZmZddlZddlmZddlmZmZddlmZddlmZdd lmZed ZejeZ Gd d Z!eGd dZ"eGddZ#dde edefdZ$de"fdZ%GddeZ&dS)N) defaultdict) dataclass)CallableDictListOptionalTupleTypeTypeVarUnion) ActorHandle)exception_causeskip_exceptions)BaseWorkerGroup) ObjectRef)PlacementGroupTc2eZdZdZdedefdefdZdS)RayTrainWorkerz@A class to execute arbitrary functions. Does not hold any state.func.returncv ||i|S#t$r$}t|}|t|d}~wwxYw)zExecutes the input function and returns the output. Args: func: The function to execute. args, kwargs: The arguments to pass into func. N) Exceptionrr)selfrargskwargseskippeds t/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/train/_internal/worker_group.py __executezRayTrainWorker.__executesX 84((( ( 8 8 8%a((Gw77 7 8s 838N)__name__ __module__ __qualname____doc__rr_RayTrainWorker__executerrrsEJJ 8hsAv. 8A 8 8 8 8 8 8r'rcbeZdZUdZeed<eed<eed<eeeefed<eed<dS)WorkerMetadataaMetadata for each worker/actor. This information is expected to stay the same throughout the lifetime of actor. Args: node_id: ID of the node this worker is on. node_ip: IP address of the node this worker is on. hostname: Hostname that this worker is on. resource_ids: Map of accelerator resources ("GPU", "neuron_cores", ..) to their IDs. pid: Process ID of this worker. node_idnode_iphostname resource_idspidN) r!r"r#r$str__annotations__rrintr&r'rr)r)&sY  LLL LLLMMMsDI~&&&& HHHHHr'r)c(eZdZUdZeed<eed<dS)WorkerzClass representing a Worker.actormetadataN)r!r"r#r$r r0r)r&r'rr3r3=s0&& r'r3executable_clsrcn|stSt|tr|SGdd|t}|S)z5Create the executable class to use as the Ray actors.ceZdZfdZxZS)3create_executable_class.._WrappedExecutablec:tj|i|dSN)super__init__)rrr __class__s rr=z._WrappedExecutable.__init__Ns%  $1&11111r')r!r"r#r= __classcell__)r>s@r_WrappedExecutabler9Ms8 2 2 2 2 2 2 2 2 2r'r@)r issubclass)r6r@s rcreate_executable_classrBEs\  " NN 3 3" 2 2 2 2 2 2 2 2"!r'cFtj}tj}t j}tj}tj }t|||||S)z]Creates metadata for this worker. This function is expected to be run on the actor. )r*r+r,r-r.) rayget_runtime_context get_node_idutilget_node_ip_addresssocket gethostnameget_accelerator_idsosgetpidr))r*r+r,accelerator_idsr.s rconstruct_metadatarOTs %''3355Gh**,,G!##H-//CCEEO )++C $     r'ceZdZdZ ddedeeeefde dee d eed e e eff d Z d Zd defdZdedefdeefdZdedefdeefdZdededefdefdZdededefdefdZdeefdZdefdZd!deefdZdZdefdZdS)" WorkerGroupa:Group of Ray Actors that can execute arbitrary functions. ``WorkerGroup`` launches Ray actors according to the given specification. It can then execute arbitrary Python functions in each of these workers. If not enough resources are available to launch the actors, the Ray cluster will automatically scale up if autoscaling is enabled. Args: num_workers: The number of workers (Ray actors) to launch. Defaults to 1. resources_per_worker (Optional[Dict[str, float]]): Dictionary specifying the resources that will be requested for each worker. Defaults to {"CPU": 1}. actor_cls (Optional[Type]): If specified use this class as the remote actors. remote_cls_args, remote_cls_kwargs: If ``remote_cls`` is provided, these args will be used for the worker initialization. placement_group (PlacementGroup|str): The placement group that workers should be created in. Defaults to "default" which will inherit the parent placement group (if child tasks should be captured). Example: .. code_block:: python worker_group = WorkerGroup(num_workers=2) output = worker_group.execute(lambda: 1) assert len(output) == 2 assert all(o == 1 for o in output) Ndefault num_workersresources_per_worker actor_clsactor_cls_argsactor_cls_kwargsplacement_groupc6|ddi}n|}|dkrtd|dtd|Drtd|d|s|r|std ||_||_tj|}|dd|_|d d|_ |d d|_ g|_ t||_ t|j tsJ|pg|_|pi|_||_t'j|j|j |j | |j |_|dS) NCPUrRrzHThe provided `num_workers` must be greater than 0. Received num_workers=z instead.c3"K|] }|dkV dS)rNr&).0vs r z'WorkerGroup.__init__..s&<<q1u<<<<<#> !  0;  <<299;;<<< < < I1EIII   .   =  '$8! $ .B C C#8#<#z(WorkerGroup.shutdown..s)RRR299;;RRRr')timeoutz8Graceful termination failed. Falling back to force kill.zShutdown successful.N)r|r}rzrorDkillr4wait)rrworker done_refsdonenot_dones rshutdownzWorkerGroup.shutdowns  Bc$,&7&7BBBCCC ??, ' '&&&& 'SRT\RRRI XiDDDND( + WXXX"l++FHV\**** +,,, r'r.rct|jdkrtdfd|jDS)aExecute ``func`` on each worker and return the futures. Args: func: A function to call on each worker. args, kwargs: Passed directly into func. Returns: (List[ObjectRef]) A list of ``ObjectRef`` representing the output of ``func`` from each worker. The order is the same as ``self.workers``. rzThere are no active workers. This worker group has most likely been shut down. Pleasecreate a new WorkerGroup or restart this one.c~g|]9}|jjdjjgRi:S)_RayTrainWorker__execute.name)r4r%optionsr!rt)r]rrrrs rrz-WorkerGroup.execute_async..s}    AG , 4 4@@@ 5  T ,  , , ,$* , ,   r')rzror{rrrrs ```r execute_asynczWorkerGroup.execute_asyncsm t|   ! !@        \     r'cHtj|j|g|Ri|S)adExecute ``func`` on each worker and return the outputs of ``func``. Args: func: A function to call on each worker. args, kwargs: Passed directly into func. Returns: (List[T]) A list containing the output of ``func`` from each worker. The order is the same as ``self.workers``. )rDgetrrs rexecutezWorkerGroup.execute s2w)t)$@@@@@@AAAr' worker_indexc|t|jkrtd|d|jd|j|jjd|jj|g|Ri|S)aWExecute ``func`` on worker ``worker_index`` and return futures. Args: worker_index: The index to execute func on. func: A function to call on the first worker. args, kwargs: Passed directly into func. Returns: (ObjectRef) An ObjectRef representing the output of func. zThe provided worker_index z is not valid for ryrr) rzrorgrTr4r%rr!rtrrrrrs rexecute_single_asyncz WorkerGroup.execute_single_asyncs 3t|,, , ,=\==!%!1===   DL & +GG@@@-4--D  +  + + +$*  + + r'cJtj|j||g|Ri|S)a-Execute ``func`` on worker with index ``worker_index``. Args: worker_index: The index to execute func on. func: A function to call on the first worker. args, kwargs: Passed directly into func. Returns: (T) The output of func. )rDrrrs rexecute_singlezWorkerGroup.execute_single4s4w0t0tUdUUUfUUVVVr'worker_indexescg}tt|jD]&}||vr ||j|'||_dS)zRemoves the workers with the specified indexes. The removed workers will go out of scope and their actor processes will be terminated. Args: worker_indexes (List[int]): The indexes of the workers to remove. N)rangerzroappend)rr new_workersis rremove_workerszWorkerGroup.remove_workersEs[ s4<(()) 4 4A&&""4<?333" r'c"g}g}t|D]}|j|jj|ji|j}||||jdttj |}tt|D]7}|j t||||8dS)auAdds ``num_workers`` to this WorkerGroup. Note: Adding workers when the cluster/placement group is at capacity may lead to undefined hanging behavior. If you are attempting to replace existing workers in the WorkerGroup, remove_workers() should be called first. Args: num_workers: The number of workers to add. )rYz+_RayTrainWorker__execute.construct_metadatar)r4r5N)rrurrsrtrqrrrr%rOrDrrzror3)rrT new_actorsnew_actor_metadata_r4r5rs rr~zWorkerGroup.add_workersTs7 {##  AD$,, $ 5-d*F.2.DFFE   e $ $ $  % %.66F7&+,,    7-..s:'' S SA L  Z]Xa[ Q Q Q R R R R S Sr'_first_node_idc\tt}|g||<|jD]'}||jj|(dt fd}|D]}|||g}|D]}| |||_dS)aReorder the workers by their node id and the lowest GPU id. This is useful for collocating workers on the same node. Example: Given workers with the following attributes: worker_0: node_id=1, gpu_ids=[1] worker_1: node_id=0, gpu_ids=[0] worker_2: node_id=1, gpu_ids=[0] worker_3: node_id=0, gpu_ids=[1] The function will perform the following steps: 1. Group by node ID: node_id=0: worker_1, worker_3 node_id=1: worker_0, worker_2 2. Sort each group by GPU ID: node_id=0: worker_1 (gpu_id=0), worker_3 (gpu_id=1) node_id=1: worker_2 (gpu_id=0), worker_0 (gpu_id=1) Resulting in the order: [worker_1, worker_3, worker_2, worker_0] Args: _first_node_id: The first ID to group by. Set this to the node ID of the trainer coordinator to ensure that the rank 0 worker is on the same node, allowing additional resources to be specified for rank 0 workers via `ScalingConfig(trainer_resources=)`. Nrc|jjdg}|sdS td|DS#t$rt|cYSwxYw)Nrarc34K|]}t|VdSr;)r1)r]gpu_ids rr_z\WorkerGroup.sort_workers_by_node_id_and_gpu_id..get_lowest_gpu_id..s(==63v;;======r')r5r-rminrg)rgpu_idss rget_lowest_gpu_idzIWorkerGroup.sort_workers_by_node_id_and_gpu_id..get_lowest_gpu_idsxo266ubAAG q $==W====== $ $ $7||### $s?AA)key) rlistror5r*rr1sortriextend)rrnode_id_to_workersrrr*sorted_workersros r"sort_workers_by_node_id_and_gpu_idz.WorkerGroup.sort_workers_by_node_id_and_gpu_idrs<)..  %13 ~ .l G GF v6 7 > >v F F F F $ $ $ $ $* D DG w ' , ,1B , C C C C)0022 + +G  ! !' * * * *% r'c*t|jSr;)rzrors r__len__zWorkerGroup.__len__s4<   r'c4tj|jS)z'Get the resources allocated per worker.)rfrjrUrs rget_resources_per_workerz$WorkerGroup.get_resources_per_workers}T6777r')rRNNNNrS)rr;)r!r"r#r$r1rrr/floatr r r rr=rvrrrrrrrrrrr~rrdictrr&r'rrQrQhs<  H;?*.+/6?999'tCJ'789 9 ! 9 #4. 9~s239999v P P P56 (36"2 Y    6BHS!V,B$q'BBBB   '/Q'7     6WW'/Q'7W WWWW" #T#Y # # # #SsSSSS<<&<&#<&<&<&<&|!!!8$888888r'rQr;)'rfloggingrLrI collectionsr dataclassesrtypingrrrrr r r r rD ray.actorr ray.air._internal.utilrr%ray.train._internal.base_worker_groupr ray.typesrray.util.placement_grouprr getLoggerr!r|rr)r3rBrOrQr&r'rrs%  ######!!!!!!NNNNNNNNNNNNNNNNNNNN !!!!!!CCCCCCCCAAAAAA333333 GCLL  8 $ $88888888"          ,   " "HTN "d " " " "N(M8M8M8M8M8/M8M8M8M8M8r'