o cig @s~dZddlZddlZddlmZddlZddlZddlm Z e e Z z ddl mZdZdaWn ey<dZdaYnwz ddlmZdZWn eyRdZYnwz dd lmZdZWn eyhdZYnwd d Zd d ZddZGdddeZeaddZe jj ddfde!de!de"de!fddZ#e jj ddfde!dee!de"de!fddZ$dbde"ddfd d!Z%dbde"de!fd"d#Z&dbde"de!fd$d%Z'de j(j)fde"fd&d'Z*de j(j)fd(e+de"fd)d*Z,dbde"fd+d,Z-dde j(j)fd-e!de"fd.d/Z.ddde j(j)fd(e+d-e!d0e!de"fd1d2Z/dcd3e!de"fd4d5Z0 ddd3e!d6e!de"fd7d8Z1dbd(e+de"fd9d:Z2 dbd;e+dZ3de j(j)fd(e+de"fd?d@Z4de j(j)fde"fdAdBZ5dbd-e!de"fdCdDZ6  ded-e!dEe!de"dFe!fdGdHZ7dbd3e!de"fdIdJZ8  ded3e!dKe!de"dFe!fdLdMZ9dNe!fdOdPZ:dbde"fdQdRZ;dSdTZde!fdZd[Z?d\d]Z@d^d_ZAd`daZBdS)fz5APIs exposed under the namespace ray.util.collective.N)List)types) NCCLGroupTF) GLOOGroup)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_AVAILABLEr r R/home/ubuntu/.local/lib/python3.10/site-packages/ray/util/collective/collective.pynccl_available(s rcCtSN)_GLOO_AVAILABLEr r r rgloo_available4rcCrr)_TORCH_DISTRIBUTED_AVAILABLEr r r rtorch_distributed_available8rrc@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. cCsi|_i|_dSr)_name_group_map_group_name_map)selfr r r__init__Ds zGroupManager.__init__cCst|}|tjjkrtd|tjjkr(td|t|||dd|d}n1|tjj kr=td|t |||}n|tjj krRtd|t |||}ntd|||j |<||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. zRay does not support MPI.zCreating GLOO group: '{}'...ray_internal_kvtcp) store_type device_type gloo_timeoutzCreating NCCL group: '{}'...z.Creating torch.distributed GLOO group: '{}'...zUnexpected backend: )rBackendMPI RuntimeErrorGLOOr debugformatrNCCLr TORCH_GLOOrrr)rbackend world_sizerank group_namer gr r rcreate_collective_groupHs2        z$GroupManager.create_collective_groupcCs ||jvSr)rrr,r r ris_group_existms zGroupManager.is_group_existcCs(||std|dS|j|S)z,Get the collective group handle by its name.z"The group '{}' is not initialized.N)r0r r r&rr/r r rget_group_by_nameps  zGroupManager.get_group_by_namecCsx||std|dS|j|}|j|=|j|=|d|}z t|}t |WdSt y;YdSw)zGroup destructor.zThe group '{}' does not exist.Ninfo_) r0r r r&rr destroy_groupr get_actorkill ValueError)rr,r-namestorer r rdestroy_collective_groupws    z%GroupManager.destroy_collective_groupN) __name__ __module__ __qualname____doc__rr.r0r1r9r r r rr<s% rcCs t|S)zDCheck if the group is initialized in this process by the group name.) _group_mgrr0r,r r ris_group_initializeds r@default0ur*r+r,r cCsvtt|}t||std|t|rtd|dks$J|dks*J||ks0Jt |||||dS)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!_check_backend_availabilityr6r&r>r0r#r.)r*r+r)r,r r r rinit_collective_groups      rFranksc 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 r2rCzHEach actor should correspond to one rank. Got '{}' ranks but '{}' actorsz5Ranks must be a permutation from 0 to '{}'. Got '{}'.cSsg|]}t|qSr )str).0rr r r 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)rJar r rrLsdetached)r7lifetimeN)rr!rErr4r#r6lenr&setrangejoinallray.util.collective.utilrMoptionsremotegetset_info) actorsr*rGr)r,r r7rM actors_idinfor r rr.sB       r.returncCstt|dS)z0Destroy a collective group given its group name.N)rDr>r9r?r r rr9sr9cC"tt|s dSt|}|jS)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. )rDr@r>r1r+r,r-r r rget_ranks   rccCr`)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. ra)rDr@r>r1r*rbr r rget_collective_group_sizes   rdcCs.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-optsr r rri's  ri 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_availabler#_check_tensor_list_inputrfrgrhri)rmr,rkr-rlr r rallreduce_multigpu9s rqcCst|}|dS)zBarrier all processes in the collective group. Args: group_name: the name of the group to barrier. Returns: None N)rfbarrierrbr r rrrOs rrdst_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) rerf_check_rank_validr ReduceOptionsrh root_rank root_tensorreduce)rjrsr,rkr-rlr r rrx\s rx 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 rnN) rror#rprfrt_check_root_tensor_validrRrurhrvrwrx)rmrsryr,rkr-rlr r rreduce_multigpuvs r{src_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)rerfrtrBroadcastOptionsrvrw broadcastrjr|r,r-rlr r rr~s  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 rnN) rror#rprfrtrzrRr}rvrwr~)rmr|rr,r-rlr r rbroadcast_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) rerprfrRr*r#rAllGatherOptions allgather)rmrjr,r-rlr r rrs 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 rnN)rror#_check_tensor_lists_inputrprfrr)rrr,r-rlr r rallgather_multigpusrcCsRt|t|t|}t||jkrtdt}||_| |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) rerprfrRr*r#rReduceScatterOptionsrh reducescatter)rjrmr,rkr-rlr r rrsrcCsHtstdt|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. rnN) rror#rrprfrrhr)output_tensor_listinput_tensor_listsr,rkr-rlr r rreducescatter_multigpusrcCRt|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) rerfrtr+r#r&r SendOptionsrssend)rjrsr,r-rlr r rr8   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) rror#rerfrtr+r&rrsrrr)rjrsrr,rr-rlr r r send_multigpuMs"  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) rerfrtr+r#r&r RecvOptionsr|recvrr r rrwrr 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) rror#rerfrtr+r&rr|rrr)rjr|rr,rr-rlr r r 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)rror#cupycudaDevice synchronize)rcpr r rrs rc Cstt|sz3d|}tj|d}t|j\}}}}}tjjj }|j } || | } t ||| ||WnIty} z=dtjvrqtjd|krqttjd}ttjd}tjd}tdd}t |||||ntd || WYd } ~ nd } ~ wwt |} | 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. r2)r7collective_group_namecollective_rankcollective_world_sizecollective_backendcollective_gloo_timeoutrBzr.r6osenvironintgetenvr#r&r1) r,r7mgridsr*r+r)r rid_rKexcr-r r rrfsF           rfcCsVt|tjrdStrt|tjjrdStr"t|tjjr"dSt d t |)z-Check if the tensor is with a supported type.Nz[Unrecognized tensor type '{}'. Supported types are: np.ndarray, torch.Tensor, cupy.ndarray.) isinstancenpndarrayrrortorch_availablethTensorr#r&type)rjr r rres  rer)cCs^|tjjkrts tddS|tjjkrtstddS|tjjkr+ts-tddSdS)z'Check whether the backend is available.zGLOO is not available.zNCCL is not available.z#torch.distributed is not available.N) rr!r$rr#r'rr(r)r)r r rrEs   rEcCs"tjjj}|jtjkr dStd)z1Check if currently it is inside a Ray actor/task.NzBThe collective APIs shall be only used inside a Ray actor or task.)rrrrmode WORKER_MODEr#)rr r rrDs  rDcCs6|dkr td|||jkrtd||jdS)z'Check the rank: 0 <= rank < world_size.rzrank '{}' is negative.z+rank '{}' must be less than world size '{}'N)r6r&r*)r-r+r r rrts  rtcCs>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)rlistr#r&rre)rmtr r rrp's   rpcCsDt|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)rrr#r&rrp) tensor_listsrr r rr4s   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)r6r&)lengthrwr r rrzAsrz)rA)rrA)rrrA)rAr)Cr=loggingrtypingrnumpyrrray.util.collectiver getLoggerr:r :ray.util.collective.collective_group.nccl_collective_grouprr r ImportError:ray.util.collective.collective_group.gloo_collective_grouprr@ray.util.collective.collective_group.torch_gloo_collective_grouprrrrrobjectrr>r@r!r'rrIrFr.r9rcrdReduceOpSUMrirrqrrrxr{r~rrrrrrrrrrrfrerErDrtrprrzr r r rs,          R ) B    $   !  * (1