o $im @sdZddlZddlZddlZddlZddlZddlmZmZddl Z ddl Z ddl m mZddlmZddlmZmZddlmZeeZz ddlmZd Zd aWn eyad Zd aYnwz dd lm Z d Z!Wn eywd Z!Ynwd d Z"ddZ#ddZ$dee%e&ffddZ'Gddde(Z)e)a*e+a,ddZ-ej.j/ddfde&de&de%de&fdd Z0ej.j/ddfde&d!ee&de%de&fd"d#Z1dfde%ddfd$d%Z2dfde%de&fd&d'Z3dfde%de&fd(d)Z4dej5j6fde%fd*d+Z7dej5j6fd,e8de%fd-d.Z9dfde%fd/d0Z:ddej5j6fd1e&de%fd2d3Z;dddej5j6fd,e8d1e&d4e&de%fd5d6Zdfd,e8de%fd=d>Z? dfd?e8d@e8de%fdAdBZ@dej5j6fd,e8de%fdCdDZAdej5j6fde%fdEdFZBdfd1e&de%fdGdHZC  did1e&dIe&de%dJe&fdKdLZDdfd7e&de%fdMdNZE  did7e&dOe&de%dJe&fdPdQZFdRe&fdSdTZGdfde%fdUdVZHdWdXZIdYej.fdZd[ZJd\d]ZKde&fd^d_ZLd`daZMdbdcZNdddeZOdS)jz5APIs exposed under the namespace ray.util.collective.N)ListTuple)types)find_free_portis_ipv6)get_master_address_metadata_key) NCCLGroupTF)TorchGLOOGroupcCstr tr tddatS)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_AVAILABLErr[/home/ubuntu/veenaModal/venv/lib/python3.10/site-packages/ray/util/collective/collective.pynccl_available*s rcCtSN_TORCH_DISTRIBUTED_AVAILABLErrrrgloo_available6srcCrrrrrrrtorch_distributed_available<srreturncCs*tj}tt|r tjntj}||fS)z4Returns the IP address and a free port on this node.)r utilget_node_ip_addressrrsocketAF_INET6AF_INET)addrportrrrget_address_and_port@s r"c@s8eZdZdZddZddZddZdd Zd d Zd S) 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. cCs i|_dSr_name_group_map)selfrrr__init__O zGroupManager.__init__c Cst|}|tjjkr_t|}|dkr$t\}}t||d|n+t|r-|dnd} t|} | dur;nt| krIt d|dt d q1t d |t||||} n |tjjkrxt|t d |t|||} ntd || |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_smetagrrrcreate_collective_groupRs<           z$GroupManager.create_collective_groupcCs ||jvSrr$r&r;rrris_group_exist~r(zGroupManager.is_group_existcCs(||std|dS|j|S)z,Get the collective group handle by its name.z"The group '{}' is not initialized.N)rCrrr4r%rBrrrget_group_by_names  zGroupManager.get_group_by_namecCsp||std|dS|j|}|j|=|d|}z t|}t|WdSt y7YdSw)zGroup destructor.zThe group '{}' does not exist.Ninfo_) rCrrr4r% destroy_groupr get_actorkill ValueError)r&r;r@namestorerrrdestroy_collective_groups    z%GroupManager.destroy_collective_groupN) __name__ __module__ __qualname____doc__r'rArCrDrLrrrrr#Gs, r#cCs2t t|WdS1swYdS)zDCheck if the group is initialized in this process by the group name.N)_group_mgr_lock _group_mgrrCr;rrris_group_initializeds$rTdefault0ur9r:r;r<cCstt|}t||std|t-t|r!t d|dks'J|dks-J||ks3Jt |||||WddS1sGwYdS)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_actorrr*r6rIr4rQrRrCr7rA)r9r:r8r;r<rrrinit_collective_groups      "rYranksc Cs6t|}t|d|}z t|tdtyYnwt|t|kr4tdt|t|t |t t t|krRtdt|d dd|D|dkr]td |t |dksgtd t ||ksqtd dd l m}d|}d d|D}|j|dd} t| 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 rErWzHEach actor should correspond to one rank. Got '{}' ranks but '{}' actorsz5Ranks must be a permutation from 0 to '{}'. Got '{}'.cSsg|]}t|qSr)str).0rrrr sz+create_collective_group..rz/World size must be greater than zero. Got '{}'.zRanks must be non-negative.z(Ranks cannot be greater than world_size.)InfocSsg|]}|jqSr) _ray_actor_id)r]arrrr_sdetached)rJlifetimeN)rr*r6r rGr7rIlenr4setrangejoinallray.util.collective.utilr`optionsremotegetset_info) actorsr9rZr8r;r<rJr` actors_idinforrrrAsB       rAcCs:ttt|WddS1swYdS)z0Destroy a collective group given its group name.N)rXrQrRrLrSrrrrLs "rLcCZttt|s WddSt|}|jWdS1s&wYdS)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)rXrQrRrCrDr:r;r@rrrget_ranks   $rucCrr)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. Nrs)rXrQrRrCrDr9rtrrrget_collective_group_size4s   $rvcCs.t|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)tensorr;opr@optsrrrr{Hs  r{ tensor_listcCs<tstdt|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_availabler7_check_tensor_list_inputrxryrzr{)rr;r}r@r~rrrallreduce_multigpuZs rcCst|}|dS)zBarrier all processes in the collective group. Args: group_name: the name of the group to barrier. Returns: None N)rxbarrierrtrrrrps rdst_rankcCsFt|t|}t||t}||_||_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) rwrx_check_rank_validr ReduceOptionsrz root_rank root_tensorreduce)r|rr;r}r@r~rrrr}s r dst_tensorcCsbtstdt|t|}t||tt||t}||_ ||_ ||_ | ||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) rrr7rrxr_check_root_tensor_validrerrzrrr)rrrr;r}r@r~rrrreduce_multigpus rsrc_rankcCs@t|t|}t||t}||_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)rwrxrrBroadcastOptionsrr broadcastr|rr;r@r~rrrrs  r src_tensorcCs\tstdt|t|}t||tt||t}||_ ||_ | ||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) rrr7rrxrrrerrrr)rrrr;r@r~rrrbroadcast_multigpus rcCsLt|t|t|}t||jkrtdt}||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) rwrrxrer9r7rAllGatherOptions allgather)rr|r;r@r~rrrrs routput_tensor_listsinput_tensor_listcCsBtstdt|t|t|}t}||||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)rrr7_check_tensor_lists_inputrrxrr)rrr;r@r~rrrallgather_multigpusrcCsRt|t|t|}t}||_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) rwrrxrReduceScatterOptionsrzrer9r7 reducescatter)r|rr;r}r@r~rrrrsrcCsHtstdt|t|t|}t}||_||||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) rrr7rrrxrrzr)output_tensor_listinput_tensor_listsr;r}r@r~rrrreducescatter_multigpu<srcCRt|t|}t||||jkrtd|t}||_| |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) rwrxrr:r7r4r SendOptionsrsend)r|rr;r@r~rrrrY   r dst_gpu_index n_elementscCtstdt|t|}t||||jkr!td||dkr,td|t}||_ ||_ ||_ | |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) rrr7rwrxrr:r4rrrrr)r|rrr;rr@r~rrr send_multigpuns"  rcCr)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) rwrxrr:r7r4r RecvOptionsrrecvrrrrrrr src_gpu_indexcCr)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) rrr7rwrxrr:r4rrrrr)r|rrr;rr@r~rrr recv_multigpus"  rgpu_idcCs,tstdddl}|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)rrr7cupycudaDevice synchronize)rcprrrrs rc Cs@ttt|sz3d|}tj|d}t|j\}}}}}tj j j }|j } ||| } t||| ||WnIty} z=dtjvrutjd|kruttjd}ttjd}tjd}tdd}t|||||ntd || WYd } ~ nd } ~ wwt|} | Wd S1swYd 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. rE)rJcollective_group_namecollective_rankcollective_world_sizecollective_backendcollective_gloo_timeoutrVzs  rcCs>t|tstdt||std|D]}t|qdS)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)rlistr7r4rrw)rtrrrrHs   rcCsDt|tstdt||std||D]}t|qdS)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)rrr7r4rr) tensor_listsrrrrrUs   rcCs2|dkr td|||krtd||dS)z9Check the root_tensor device is 0 <= root_tensor < lengthrzroot_tensor '{}' is negative.z9root_tensor '{}' is greater than the number of GPUs: '{}'N)rIr4)lengthrrrrrbsr)rU)rrU)rrrU)rUr)PrPloggingrr threadingr/typingrrnumpyrr ray.experimental.internal_kv experimental internal_kvr-r[rray._common.network_utilsrr@ray.util.collective.collective_group.torch_gloo_collective_grouprr, getLoggerrMr:ray.util.collective.collective_group.nccl_collective_groupr rr ImportErrorr rrrrr\rr"objectr#rRLockrQrTr*r5rYrArLrurvReduceOpSUMr{rrrrrrrrrrrrrrrrrxrwr6rXrrrrrrrrs0        W  , B     $   !  * (3