o 1 in @sdZddlZddlZddlZddlmZddlZddlZddl m m Z ddl mZddlmZddlmZeeZz ddlmZd Zd aWn eyUd Zd aYnwz dd lmZd ZWn eykd ZYnwz dd lm Z d Z!Wn eyd Z!Ynwd dZ"ddZ#ddZ$ddZ%Gddde&Z'e'a(ddZ)ej*j+ddfde,de,de-de,fdd Z.ej*j+ddfde,d!ee,de-de,fd"d#Z/dgde-d$dfd%d&Z0dgde-d$e,fd'd(Z1dgde-d$e,fd)d*Z2dej3j4fde-fd+d,Z5dej3j4fd-e6de-fd.d/Z7dgde-fd0d1Z8ddej3j4fd2e,de-fd3d4Z9dddej3j4fd-e6d2e,d5e,de-fd6d7Z:dhd8e,de-fd9d:Z; did8e,d;e,de-fdd?Z= dgd@e6dAe6de-fdBdCZ>dej3j4fd-e6de-fdDdEZ?dej3j4fde-fdFdGZ@dgd2e,de-fdHdIZA  djd2e,dJe,de-dKe,fdLdMZBdgd8e,de-fdNdOZC  djd8e,dPe,de-dKe,fdQdRZDdSe,fdTdUZEdgde-fdVdWZFdXdYZGdZej*fd[d\ZHd]d^ZIde,fd_d`ZJdadbZKdcddZLdedfZMdS)kz5APIs exposed under the namespace ray.util.collective.N)List)types)get_address_and_port)get_master_address_metadata_key) NCCLGroupTF)TorchGLOOGroup) NixlBackendcCstr tr tddatS)NzqNCCL seems unavailable. Please install Cupy following the guide at: https://docs.cupy.dev/en/stable/install.html.F)rayZ get_gpu_ids_LOG_NCCL_WARNINGloggerwarning_NCCL_AVAILABLErrj/home/app/PaddleOCR-VL-test/.venv_paddleocr/lib/python3.10/site-packages/ray/util/collective/collective.pynccl_available1s rcCtSN_TORCH_DISTRIBUTED_AVAILABLErrrrgloo_available=srcCrrrrrrrtorch_distributed_availableCrcCrr)_NIXL_AVAILABLErrrrnixl_availableGrrc@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)selfrrr__init__Ss zGroupManager.__init__c Csht|}|tjjkrtd|tjjks|tjjkrot|}|dkr4t\}}t ||d|n+t |r=|dnd} t |} | durKnt | krYt d|d t d qAtd |t||||} n6|tjjkrt|td |t|||} n|tjjkrt|td |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.r:g@@g>@TNz:Timed out waiting for GLOO rendezvous metadata for group 'z'.g?z.Creating torch.distributed GLOO group: '{}'...zCreating NCCL group: '{}'...zCreating NIXL Backend: '{}'...zUnexpected backend: )rBackendZMPI RuntimeErrorGLOO TORCH_GLOO_get_master_addr_key_get_address_and_port _internal_kvZ_internal_kv_puttimeZ_internal_kv_get TimeoutErrorsleepr debugformatrNCCL_check_backend_availabilityrNIXLr rr) rbackend world_sizerank group_name gloo_timeoutZ metadata_keyaddrportZ deadline_smetagrrrcreate_collective_groupWsJ             z$GroupManager.create_collective_groupcCs ||jvSr)rrr3rrris_group_exists zGroupManager.is_group_existcCs(||std|dS|j|S)z,Get the collective group handle by its name.z"The group '{}' is not initialized.N)r;r r r,rr:rrrget_group_by_names  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_) r;r r r,rrZ destroy_groupr get_actorkill ValueError)rr3r8namestorerrrdestroy_collective_groups    z%GroupManager.destroy_collective_groupN) __name__ __module__ __qualname____doc__rr9r;r<rCrrrrrKs4 rcCs t|S)zDCheck if the group is initialized in this process by the group name.) _group_mgrr;r3rrris_group_initializeds rJdefault0ur1r2r3r4cCsvtt|}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!r.r@r,rHr;r"r9)r1r2r0r3r4rrrinit_collective_groups      rOranksc 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 r=rMzHEach 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)Z _ray_actor_id)rSarrrrUsZdetached)rAZlifetimeN)rr!r.r r>r"r@lenr,setrangejoinallZray.util.collective.utilrVoptionsremotegetZset_info) Zactorsr1rPr0r3r4rArVZ actors_idinforrrr9sB       r9returncCstt|dS)z0Destroy a collective group given its group name.N)rNrHrCrIrrrrCsrCcC"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. )rNrJrHr<r2r3r8rrrget_rank"s   recCrb)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. rc)rNrJrHr<r1rdrrrget_collective_group_size4s   rfcCs.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)tensorr3opr8optsrrrrkEs  rk 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_inputrhrirjrk)ror3rmr8rnrrrallreduce_multigpuWs rscCst|}|dS)zBarrier all processes in the collective group. Args: group_name: the name of the group to barrier. Returns: None N)rhbarrierrdrrrrtms rtdst_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) rgrh_check_rank_validr ReduceOptionsrj root_rank root_tensorreduce)rlrur3rmr8rnrrrrzzs rz 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 rpN) rrqr"rrrhrv_check_root_tensor_validrXrwrjrxryrz)rorur{r3rmr8rnrrrreduce_multigpus 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)rgrhrvrBroadcastOptionsrxry broadcastrlr~r3r8rnrrrrs  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 rpN) rrqr"rrrhrvr|rXrrxryr)ror~rr3r8rnrrrbroadcast_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) rgrrrhrXr1r"rAllGatherOptions allgather)rorlr3r8rnrrrrs 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 rpN)rrqr"_check_tensor_lists_inputrrrhrr)rrr3r8rnrrrallgather_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) rgrrrhrReduceScatterOptionsrjrXr1r" reducescatter)rlror3rmr8rnrrrrsrcCsHtstdt|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. rpN) rrqr"rrrrhrrjr)Zoutput_tensor_listZinput_tensor_listsr3rmr8rnrrrreducescatter_multigpu9srcCRt|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) rgrhrvr2r"r,r SendOptionsrusend)rlrur3r8rnrrrrV   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) rrqr"rgrhrvr2r,rrurrr)rlrurr3rr8rnrrr send_multigpuks"  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) rgrhrvr2r"r,r RecvOptionsr~recvrrrrrrr 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) rrqr"rgrhrvr2r,rr~rrr)rlr~rr3rr8rnrrr 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)rrqr"ZcupycudaZDevice synchronize)rcprrrrs rc CsB|tjkrtt|szD|tjkrttjjdd|dn1d|}tj |d}t |j \}}}}}tj jj}|j} ||| } t||| ||WnIty} z=dtjvrtjd|krttjd}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. Nr=)rAZcollective_group_nameZcollective_rankZcollective_world_sizeZcollective_backendZcollective_gloo_timeoutrLzr_get_infor^_privateworker global_workerZ core_workerZ get_actor_idindexr@osenvironintgetenvr"r,r<) r3rAZmgridsr1r2r0r4rZid_rTexcr8rrrrhsR            rhcCsVt|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.) isinstancenpZndarrayrrqrZtorch_availablethZTensorr"r,type)rlrrrrgs  rgr0cCs||tjjkrts tddS|tjjkrtstddS|tjjkr-ts+tddS|tjjkr:t st|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,rrg)rotrrrrrOs   rrcCsDt|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,rrr)Z tensor_listsrrrrr\s   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)r@r,)lengthryrrrr|isr|)rK)rrK)rrrK)rKr)NrGloggingrr(typingrnumpyrr Zray.experimental.internal_kvZ experimentalZ internal_kvr'rQrZ ray.experimental.collective.utilrr&Z@ray.util.collective.collective_group.torch_gloo_collective_grouprr% getLoggerrDr Z:ray.util.collective.collective_group.nccl_collective_grouprrr ImportErrorrrZ1ray.util.collective.collective_group.nixl_backendr rrrrrobjectrrHrJr!r-rrRrOr9rCrerfZReduceOpZSUMrkrrsrtrzr}rrrrrrrrrrrrhrgr.rNrvrrrr|rrrrs6            a ) B    $   !  * (7