&`im ZdZddlZddlZddlZddlZddlZddlmZmZddl Z ddl Z ddl m cmZddlmZddlmZmZddlmZejeZ ddlmZd Zd an#e$rd Zd aYnwxYw dd lm Z d Z!n #e$rd Z!YnwxYwd Z"d Z#dZ$dee%e&ffdZ'Gdde(Z)e)a*ej+a,dZ-ej.j/ddfde&de&de%de&fdZ0ej.j/ddfde&dee&de%de&fdZ1dDde%ddfdZ2dDde%de&fdZ3dDde%de&fdZ4dej5j6fde%fd Z7dej5j6fd!e8de%fd"Z9dDde%fd#Z:ddej5j6fd$e&de%fd%Z;dddej5j6fd!e8d$e&d&e&de%fd'ZdDd!e8de%fd,Z? dDd-e8d.e8de%fd/Z@dej5j6fd!e8de%fd0ZAdej5j6fde%fd1ZBdDd$e&de%fd2ZC dGd$e&d3e&de%d4e&fd5ZDdDd(e&de%fd6ZE dGd(e&d7e&de%d4e&fd8ZFd9e&fd:ZGdDde%fd;ZHd<ZId=ej.fd>ZJd?ZKde&fd@ZLdAZMdBZNdCZOdS)Hz5APIs exposed under the namespace ray.util.collective.N)ListTuple)types)find_free_portis_ipv6)get_master_address_metadata_key) NCCLGroupTF)TorchGLOOGroupc|tjr#trtddat S)NzqNCCL seems unavailable. Please install Cupy following the guide at: https://docs.cupy.dev/en/stable/install.html.F)ray get_gpu_ids_LOG_NCCL_WARNINGloggerwarning_NCCL_AVAILABLEr/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/util/collective/collective.pynccl_availabler*sC "." <   " rctSN_TORCH_DISTRIBUTED_AVAILABLErrrgloo_availabler6s  ('rctSrrrrrtorch_distributed_availabler<s ''rreturnctj}tt |r t jn t j}||fS)z4Returns the IP address and a free port on this node.)r utilget_node_ip_addressrrsocketAF_INET6AF_INET)addrports rget_address_and_portr'@s? 8 ' ' ) )D WT]]N&// O OD :rc0eZdZdZdZdZdZdZdZdS) GroupManageraUse this class to manage the collective groups we created so far. Each process will have an instance of `GroupManager`. Each process could belong to multiple collective groups. The membership information and other metadata are stored in the global `_group_mgr` object. ci|_dSr_name_group_map)selfs r__init__zGroupManager.__init__Os!rcVtj|}|tjjkrt|}|dkr,t \}}t j||d|nttj|r|dz ndz} t j|} | n?tj| krtd|dtj d Vt d |t||||} nu|tjjkrNt!|t d |t#|||} nt%d || |j|<|j|S) zThe entry to create new collective groups in the manager. Put the registration and the group information into the manager metadata as well. r:g@@g>@TNz:Timed out waiting for GLOO rendezvous metadata for group 'z'.g?z.Creating torch.distributed GLOO group: '{}'...zCreating NCCL group: '{}'...zUnexpected backend: )rBackendGLOO_get_master_addr_keyr' _internal_kv_internal_kv_puttime_internal_kv_get TimeoutErrorsleeprdebugformatr NCCL_check_backend_availabilityr RuntimeErrorr,) r-backend world_sizerank group_name gloo_timeout metadata_keyr%r& deadline_smetags rcreate_collective_groupz$GroupManager.create_collective_groupRs-(( em( ( (/ ;;Lqyy133 d-lt>zJJ K K K*dJ77AA?g??@@ @+,Z(#J//rc||jvSrr+r-rBs ris_group_existzGroupManager.is_group_exist~sT111rc||s/td|dS|j|S)z,Get the collective group handle by its name.z"The group '{}' is not initialized.N)rKrrr;r,rJs rget_group_by_namezGroupManager.get_group_by_namesL"":..  NN?FFzRR S S S4#J//rc^||s/td|dS|j|}|j|=|d|z} t j|}t j|dS#t$rYdSwxYw)zGroup destructor.zThe group '{}' does not exist.Ninfo_) rKrrr;r, destroy_groupr get_actorkill ValueError)r-rBrGnamestores rdestroy_collective_groupz%GroupManager.destroy_collective_groups"":..  NN;BB:NN O O O F   ,   , # M$''E HUOOOOO    DD s4(B B,+B,N) __name__ __module__ __qualname____doc__r.rHrKrMrVrrrr)r)Gsj"""*0*0*0X222000rr)cxt5t|cdddS#1swxYwYdS)zDCheck if the group is initialized in this process by the group name.N)_group_mgr_lock _group_mgrrKrBs ris_group_initializedr_s 55((44555555555555555555s /33default0ur@rArBrCcttj|}t||s"t d|t 5t|rtd|dksJ|dksJ||ksJt |||||ddddS#1swxYwYdS)a=Initialize a collective group inside an actor process. Args: world_size: the total number of processes in the group. rank: the rank of the current process. backend: the CCL backend to use, NCCL or GLOO. group_name: the name of the collective group. Returns: None z%group_name '{}' needs to be a string.#Trying to initialize a group twice.rN) _check_inside_actorrr1r=rSr;r\r]rKr>rH)r@rAr?rBrCs rinit_collective_groupres3$mG$$G((( U@GG SSTTT     $ $Z 0 0 FDEE EA~~~~qyyyyj    ** Zz<                      sA C  CCranksc @tj|}t|d|z} tj|t d#t $rYnwxYwt|t|kr=t dt|t|t|ttt|krMt dt|d d|D|dkr"t d|t|dkst d t||kst d dd l m}d|z}d |D}||d } tj| j|||||gdS)aDeclare a list of actors as a collective group. Note: This function should be called in a driver process. Args: actors: a list of actors to be set in a collective group. world_size: the total number of processes in the group. ranks (List[int]): the rank of each actor. backend: the CCL backend to use, NCCL or GLOO. group_name: the name of the collective group. Returns: None rOrczHEach actor should correspond to one rank. Got '{}' ranks but '{}' actorsz5Ranks must be a permutation from 0 to '{}'. Got '{}'.c,g|]}t|Sr)str).0rs r z+create_collective_group..s$;$;$;SVV$;$;$;rrz/World size must be greater than zero. Got '{}'.zRanks must be non-negative.z(Ranks cannot be greater than world_size.)Infocg|] }|j Sr) _ray_actor_id)rkas rrmz+create_collective_group..s111Q111rdetached)rTlifetimeN)rr1r=r rQr>rSlenr;setrangejoinallray.util.collective.utilrnoptionsremotegetset_info) actorsr@rfr?rBrCrTrn actors_idinfos rrHrHs!,mG$$G((( Z D  d@AAA       5zzS[[   $$*F3u::s6{{$C$C    5zzSs5zz**++++ C J JE BGG$;$;U$;$;$;<<     Q = D DZ P P    u::??8999 u:: " "EFFF.----- Z D11&111I < > @ @DGT] ! !)Z V V WXXXXXs#A AActt5t|ddddS#1swxYwYdS)z0Destroy a collective group given its group name.N)rdr\r]rVr^s rrVrVs 88++J777888888888888888888s>AActt5t|s ddddSt|}|jcdddS#1swxYwYdS)aReturn the rank of this process in the given group. Args: group_name: the name of the group to query Returns: the rank of this process in the named group, -1 if the group does not exist or the process does not belong to the group. N)rdr\r]rKrMrArBrGs rget_rankrs ((44   ( ( 4 4v A, A,,A03A0ctt5t|s ddddSt|}|jcdddS#1swxYwYdS)aReturn the size of the collective group with the given name. Args: group_name: the name of the group to query Returns: The world size of the collective group, -1 if the group does not exist or the process does not belong to the group. Nr)rdr\r]rKrMr@rs rget_collective_group_sizer4s ((44   ( ( 4 4| rct|t|}tj}||_||g|dS)aCollective allreduce the tensor across the group. Args: tensor: the tensor to be all-reduced on this process. group_name: the collective group name to perform allreduce. op: The reduce operation. Returns: None N)_check_single_tensor_inputget_group_handlerAllReduceOptionsreduceOp allreduce)tensorrBoprGoptss rrrHsJv&&&$$A  !DDMKK$r tensor_listctjstdt|t |}tj}||_|||dS)aCollective allreduce a list of tensors across the group. Args: tensor_list (List[tensor]): list of tensors to be allreduced, each on a GPU. group_name: the collective group name to perform allreduce. Returns: None &Multigpu calls requires NCCL and Cupy.N)rcupy_availabler>_check_tensor_list_inputrrrr)rrBrrGrs rallreduce_multigpurZsi   ! !ECDDD[)))$$A  !DDMKK T"""""rcLt|}|dS)zBarrier all processes in the collective group. Args: group_name: the name of the group to barrier. Returns: None N)rbarrierrs rrrps" $$AIIKKKKKrdst_rankct|t|}t||tj}||_||_d|_||g|dS)a:Reduce the tensor across the group to the destination rank. Args: tensor: the tensor to be reduced on this process. dst_rank: the rank of the destination process. group_name: the collective group name to perform reduce. op: The reduce operation. Returns: None rN) rr_check_rank_validr ReduceOptionsr root_rank root_tensorreduce)rrrBrrGrs rrr}sqv&&&$$Aa"""   DDMDNDHHfXtr dst_tensorc\tjstdt|t |}t ||t t||tj}||_ ||_ ||_ | ||dS)aReduce the tensor across the group to the destination rank and destination tensor. Args: tensor_list: the list of tensors to be reduced on this process; each tensor located on a GPU. dst_rank: the rank of the destination process. dst_tensor: the index of GPU at the destination. group_name: the collective group name to perform reduce. op: The reduce operation. Returns: None rN) rrr>rrr_check_root_tensor_validrtrrrrr)rrrrBrrGrs rreduce_multigpurs*   ! !ECDDD[)))$$Aa"""S--z:::   DDMDN!DHH[$rsrc_rankct|t|}t||tj}||_d|_||g|dS)a(Broadcast the tensor from a source process to all others. Args: tensor: the tensor to be broadcasted (src) or received (destination). src_rank: the rank of the source process. group_name: the collective group name to perform broadcast. Returns: None rN)rrrrBroadcastOptionsrr broadcastrrrBrGrs rrrsjv&&&$$Aa"""  ! # #DDNDKK$r src_tensorcNtjstdt|t |}t ||t t||tj}||_ ||_ | ||dS)agBroadcast the tensor from a source GPU to all other GPUs. Args: tensor_list: the tensors to broadcast (src) or receive (dst). src_rank: the rank of the source process. src_tensor: the index of the source GPU on the source process. group_name: the collective group name to perform broadcast. Returns: None rN) rrr>rrrrrtrrrr)rrrrBrGrs rbroadcast_multigpurs   ! !ECDDD[)))$$Aa"""S--z:::  ! # #DDN!DKK T"""""rct|t|t|}t||jkrt dt j}||g|g|dS)a Allgather tensors from each process of the group into a list. Args: tensor_list: the results, stored as a list of tensors. tensor: the tensor (to be gathered) in the current process group_name: the name of the collective group. Returns: None zPThe length of the tensor list operands to allgather must be equal to world_size.N) rrrrtr@r>rAllGatherOptions allgather)rrrBrGrs rrrsv&&&[)))$$A ;1<'' +     ! # #DKK x.....routput_tensor_listsinput_tensor_listctjstdt|t |t |}tj}||||dS)aAllgather tensors from each gpus of the group into lists. Args: output_tensor_lists (List[List[tensor]]): gathered results, with shape must be num_gpus * world_size * shape(tensor). input_tensor_list: (List[tensor]): a list of tensors, with shape num_gpus * shape(tensor). group_name: the name of the collective group. Returns: None rN)rrr>_check_tensor_lists_inputrrrr)rrrBrGrs rallgather_multigpurs|   ! !ECDDD1222.///$$A  ! # #DKK#%6=====rct|t|t|}tj}||_t ||jkrtd| |g|g|dS)aReducescatter a list of tensors across the group. Reduce the list of the tensors across each process in the group, then scatter the reduced list of tensors -- one tensor for each process. Args: tensor: the resulted tensor on this process. tensor_list: The list of tensors to be reduced and scattered. group_name: the name of the collective group. op: The reduce operation. Returns: None zXThe length of the tensor list operands to reducescatter must not be equal to world_size.N) rrrrReduceScatterOptionsrrtr@r> reducescatter)rrrBrrGrs rrrs"v&&&[)))$$A  % ' 'DDM ;1<'' /   OOVH{mT22222rctjstdt|t |t |}tj}||_||||dS)aReducescatter a list of tensors across all GPUs. Args: output_tensor_list: the resulted list of tensors, with shape: num_gpus * shape(tensor). input_tensor_lists: the original tensors, with shape: num_gpus * world_size * shape(tensor). group_name: the name of the collective group. op: The reduce operation. Returns: None. rN) rrr>rrrrrr)output_tensor_listinput_tensor_listsrBrrGrs rreducescatter_multigpur<s&   ! !ECDDD0111/000$$A  % ' 'DDMOO&(:DAAAAArct|t|}t||||jkr"t d|t j}||_| |g|dS)zSend a tensor to a remote process synchronously. Args: tensor: the tensor to send. dst_rank: the rank of the destination process. group_name: the name of the collective group. Returns: None "The destination rank '{}' is self.N) rrrrAr>r;r SendOptionsrsend)rrrBrGrs rrrYv&&&$$Aa"""16?FFxPPQQQ    DDMFFF8Tr dst_gpu_index n_elementsctjstdt|t |}t ||||jkr"td||dkr"td|tj}||_ ||_ ||_ | |g|dS)aSend a tensor to a remote GPU synchronously. The function assumes each process owns >1 GPUs, and the sender process and receiver process has equal number of GPUs. Args: tensor: the tensor to send, located on a GPU. dst_rank: the rank of the destination process. dst_gpu_index: the destination gpu index. group_name: the name of the collective group. n_elements: if specified, send the next n elements from the starting address of tensor. Returns: None z!send_multigpu call requires NCCL.GThe dst_rank '{}' is self. Considering doing GPU to GPU memcpy instead?rz The n_elements '{}' should >= 0.N) rrr>rrrrAr;rrrrr)rrrrBrrGrs r send_multigpurns.   ! !@>???v&&&$$Aa"""16 //5vh/?/?   A~~=DDZPPQQQ    DDM&D DOFFF8Trct|t|}t||||jkr"t d|t j}||_| |g|dS)zReceive a tensor from a remote process synchronously. Args: tensor: the received tensor. src_rank: the rank of the source process. group_name: the name of the collective group. Returns: None rN) rrrrAr>r;r RecvOptionsrrecvrs rrrrr src_gpu_indexctjstdt|t |}t ||||jkr"td||dkr"td|tj}||_ ||_ ||_ | |g|dS)aReceive a tensor from a remote GPU synchronously. The function asssume each process owns >1 GPUs, and the sender process and receiver process has equal nubmer of GPUs. Args: tensor: The received tensor, located on a GPU. src_rank: The rank of the source process. src_gpu_index: The index of the source GPU on the src process. group_name: The name of the collective group. Returns: None z!recv_multigpu call requires NCCL.rrz#The n_elements '{}' should be >= 0.N) rrr>rrrrAr;rrrrr)rrrrBrrGrs r recv_multigpurs*   ! !@>???v&&&$$Aa"""16 //5vh/?/?   A~~@GG SSTTT    DDM&D DOFFF8Trgpu_idctjstdddl}|j|dS)zSynchronize the current process to a give device. Args: gpu_id: the GPU device id to synchronize. Returns: None z(synchronize call requires CUDA and NCCL.rN)rrr>cupycudaDevice synchronize)rcps rrrsW   ! !GEFFFGNN6&&(((((rc tt5t|s d|z}t j|}t j|j\}}}}}tj j j }|j } ||| } t||| ||n#t $r} dt"jvrt"jd|krt't"jd}t't"jd}t"jd}t#jdd}t|||||n#t+d || Yd } ~ nd } ~ wwxYwt|} | cd d d S#1swxYwYd S) zCheck if the group is initialized and return the group handle. Args: group_name: the name of the collective group. Returns: The collective group handle. rO)rTcollective_group_namecollective_rankcollective_world_sizecollective_backendcollective_gloo_timeoutrazr;rM) rBrTmgridsr@rAr?rCrid_rlexcrGs rrrsT $$((44!  +m...?BwL''))@@<Zw ,:(55773(22ZJ    ,rz99 #:;zIIrz*;<==D!$RZ0G%H!I!IJ j)=>G#%9-F#N#NL66T:|'66r;type)rs rrrs&"*%%  feh. / /  F  feho . .  F  228&f2F2F  rr?c|tjjkrtst ddS|tjjkrt st ddSdS)z'Check whether the backend is available.z#torch.distributed is not available.zNCCL is not available.N)rr1r2rr>r<r)r?s rr=r=(s{%-$$$*,, FDEE E F F EM& & & 9788 8 ' & 9 9rcztjjj}|jtjkrdSt d)z1Check if currently it is inside a Ray actor/task.NzBThe collective APIs shall be only used inside a Ray actor or task.)r rrrmode WORKER_MODEr>)rs rrdrd3s8 \ .F {co%% P   rc|dkr"td|||jkr(td||jdS)z'Check the rank: 0 <= rank < world_size.rzrank '{}' is negative.z+rank '{}' must be less than world size '{}'N)rSr;r@)rGrAs rrr>sd axx188>>??? q| 9 @ @q| T T   rct|ts/tdt ||std|D]}t |dS)z7Check if the input is a list of supported tensor types.z.The input must be a list of tensors. Got '{}'.zGot an empty list of tensors.N)rlistr>r;rr)rts rrrHs k4 ( (  tK0011    <:;;; &&"1%%%%&&rct|ts/tdt ||std||D]}t |dS)z@Check if the input is a list of lists of supported tensor types.z7The input must be a list of lists of tensors. Got '{}'.zDid not receive tensors. Got: N)rrr>r;rr) tensor_listsrs rrrUs lD ) )  tL1122    LJLJJKKK $$ ####$$rc|dkr"td|||kr#td||dS)z9Check the root_tensor device is 0 <= root_tensor < lengthrzroot_tensor '{}' is negative.z9root_tensor '{}' is greater than the number of GPUs: '{}'N)rSr;)lengthrs rrrbsaQ8?? LLMMMf 6+v..   r)r`)rr`)rrr`)r`r)PrZloggingrr" threadingr6typingrrnumpyrr ray.experimental.internal_kv experimental internal_kvr4rhrray._common.network_utilsrr@ray.util.collective.collective_group.torch_gloo_collective_groupr r3 getLoggerrWr:ray.util.collective.collective_group.nccl_collective_groupr rr ImportErrorr rrrrrjrr'objectr)r]Lockr\r_r1r<rerHrVrrReduceOpSUMrrrrrrrrrrrrrrrrrrrr=rdrrrrrrrr s-;;   333333333========  8 $ $TTTTTTOO )$(  )))#(   )   ((( (((eCHoTTTTT6TTTn\^^  ).""555 M  % % % %  %  % % % % X M  >Y>Y>Y 9>Y  >Y  >Y>Y>Y>YD888T8888S,#c()2en6H  #    &*3u~7I####&####,      u~?Q+.8 ~ ! ! ! ! !  ! ! ! ! H   S    .LU###03#EH####8//4/S////4KT>>>26>DG>>>>22;u~?Q333+.3333B  ~ BBBBBB:3C2  '''' '  ''''T3C2  %%%% %  %%%%P ) ) ) ) ) 000000f    99999    s     & & & $ $ $     s$ A A,+A,0A99BB