&`i4ddlZddlmZddlmZmZmZmZmZddl Z ddl m Z ddl m Z ddlmZmZddlmZerddlZejeZGdd eZdS) N) ModuleType) TYPE_CHECKINGCallableListOptionalTuple)RayChannelError)AcceleratorContext) CommunicatorTorchTensorAllocator)ReduceOpc eZdZdZ d*dededeededded d ef d Z ded d fdZ d edfdZ de j jd efdZd eefdZd efdZddded d fdZeefdeeddded dfdZ d+d Z d,d!Zejfddddd"efd#Zejfddddd"efd$Zed%Zed&Zd-d'Zd e fd(Z!e"d e fd)Z#d S). _NcclGroupz Represents an actor's NCCL communicator. This is the default NCCL communicator to be used in Compiled Graph if a custom communicator is not provided. This class is not thread-safe. F world_sizecomm_idrank actor_handleszray.actor.ActorHandle cuda_streamztorch.cuda.Streamuse_communication_streamsc||_||_d|_||_||_|t js Jd| Jd|t jj }||ksJd|d|ddl m}||_|j ||||_ nd|_ d|_ d|_d|_|| Jd||_ |rpddl} t#jd} | j| |_| j| |_n|j |_|j |_d |_dS) a Initialize a NCCL communicator that can be used to communicate p2p with other GPU actors. This method blocks until the same call has been made on all other actors in the group, with the same arguments for world_size and comm_id. NOTE: A concurrent NCCL group can coexist with this one but using the two groups concurrently on different CUDA streams may cause deadlock. See https://docs.nvidia.com/deeplearning/nccl/user-guide/docs/usage/communicators.html #using-multiple-nccl-communicators-concurrently. If the user can guarantee that all involved actors execute the same ops in the same order, then the other NCCL group should use the given `cuda_stream`, and there will not be a concurrency issue. Otherwise, the other stream needs to synchronize with the given `cuda_stream` before and after it launches NCCL ops, e.g., at the beginning and end of a DAG task. Args: world_size: The number of participating actors/devices. comm_id: A unique communicator ID returned by cupy.cuda.nccl.get_unique_id(). rank: The rank of this actor. If None, then the caller is not a participant of the NCCL group. actor_handles: A list of actor handles, in rank order. cuda_stream: A raw CUDA stream to dispatch NCCL ops to. If rank is specified, then this must be specified too. use_communication_streams: Whether to use dedicated send and recv streams for communication. If True, communication and computation can be overlapped to improve performance. NzNCCL actor has no GPUs assignedz#NCCL actor must specify cuda_streamzNCCL actor's rank z does not match expected rank r) nccl_utilzNCCL actor has no rank assigned)deviceF) _world_size_rankr_actor_handles_use_communication_streamsray get_gpu_idsget_rankget_runtime_context current_actor$ray.util.collective.collective_groupNcclCommunicator_comm _cuda_stream _send_stream _recv_streamtorchr getget_accelerator_devicescudaStream_closed) selfrrrrrr expected_rankrr(rs w/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/experimental/channel/nccl_group.py__init__z_NcclGroup.__init__sV&$( /3+*C'  ?$$ G G&G G G$**,Q*** MM#*A*C*C*QRRM %%%WDWW WW&%% G F F F F F&DN88WdSSDJJDJ;?;?;?  "##%F### +D ( 6 ,/11IIKKAN$)J$5$5V$5$D$D!$)J$5$5V$5$D$D!!$($5!$($5! returnNcdSN)r.rs r0 initializez_NcclGroup.initializets r2c|jSr5)rr.s r0get_actor_handlesz_NcclGroup.get_actor_handlesxs ""r2actorcd|jD} ||j}n#t$rtdwxYw|S)z Return the given actor's rank in the NCCL communicator. Args: actor: The actor handle to look up. cg|] }|j Sr6) _ray_actor_id).0as r0 z'_NcclGroup.get_rank..sBBBQ_BBBr2zActor is not in the NCCL group.)rindexr> ValueError)r.r; actor_idsrs r0rz_NcclGroup.get_rank{sjCBd.ABBB  @??5#677DD @ @ @>?? ? @ s .Ac|jS)z+ Return this actor's rank. )rr9s r0 get_self_rankz_NcclGroup.get_self_ranks zr2c|jS)zF Return the number of ranks in the NCCL communicator. )rr9s r0get_world_sizez_NcclGroup.get_world_sizes r2buf torch.Tensor peer_rankcF|jrtd|jr|j|j|j|| |j |||jj dS)a Send a torch.Tensor to a peer. This returns when the send kernel has been queued, but the kernel may not have completed. Therefore, the caller should ensure that there are no concurrent writes to the sent `buf` until the send has finished. That is, either all writes should be submitted on the current stream (self._cuda_stream) or, if on a different stream, that stream should synchronize with the current stream. Args: buf: The torch.Tensor to send. It should already be on this actor's default device. peer_rank: The rank of the actor to send to. NCCL group has been destroyed.N) r-r rr& synchronizer$sendrget_tensor_ptrnumelget_nccl_tensor_dtyper)r.rIrKs r0rOz_NcclGroup.sends < D!"BCC C  * ,   ) ) + + +  N ) )# . . IIKK N 0 0 5 5    )      r2shapedtypez torch.dtypec|jrtd| Jd|||}|jr|j|j|j|| |j |||jj n|j|j|| |j |||jj |j |jrtd|S)a Receive a torch.Tensor from a peer and synchronize the current stream. After this call returns, the receive buffer is safe to read from from any stream. An RayChannelError will be raised if an error occurred (e.g., remote actor died), and the buffer is not safe to read. Args: buf: The torch.Tensor to receive into. This buffer is safe to read peer_rank: The rank of the actor to receive from. rMNz&NCCL group requires a tensor allocator) r-r rr'rNr$recvrrPrQrRrr%)r.rSrTrK allocatorrIs r0rVz_NcclGroup.recvsM$ < D!"BCC C$$&N$$$iu%%  * ,   ) ) + + + JOO--c22 44S99!-      JOO--c22 44S99!-       ) ) + + + < D!"BCC C r2send_bufrecv_buf operationCallable[..., None]c|jrtd|j|jks Jd|||j|jrtddS)NrMzRay Compiled Graph derived the dtype of recv_buf from send_buf, so send_buf and recv_buf must have the same dtype. If you see this error, please file an issue at Ray repository.zNCCL group has been destroyed during allreduce operation. There may be a dtype mismatch between input tensors from different ranks.)r-r rTr%rN)r.rXrYrZoperation_argss r0_exec_collectivez_NcclGroup._exec_collectives < D!"BCC C~/// M0//  >"" %%''' < !#   r2c|j||j|||j||jjg}|j|||jjg|RdSr5) rrPrQrRr%rr^r$ allGather)r.rXrYr]s r0 allgatherz_NcclGroup.allgathers N ) )( 3 3 N ) )( 3 3 NN   N 0 0 : :   )      J         r2opc|j||j|||j||j|jjg}|j|||jj g|RdSr5) rrPrQrRvaluer%rr^r$ allReducer.rXrYrbr]s r0 allreducez_NcclGroup.allreduce%s N ) )( 3 3 N ) )( 3 3 NN   N 0 0 : : H   )      J         r2c|j||j|||j||j|jjg}|j|||jj g|RdSr5) rrPrQrRrdr%rr^r$ reduceScatterrfs r0 reducescatterz_NcclGroup.reducescatter:s N ) )( 3 3 N ) )( 3 3 NN   N 0 0 : : H   )      J $        r2cHddl}|j|jSNr)r(r+ StreamContextr'r.r(s r0 recv_streamz_NcclGroup.recv_streamO$ z''(9:::r2cHddl}|j|jSrl)r(r+rmr&rns r0 send_streamz_NcclGroup.send_streamUrpr2c|jrdSd|_|jgtdt jj|j|jdSdS)z) Destroy the NCCL group. NTz!Destructing NCCL group on actor: ) r-r$loggerinforr r!abortdestroyr9s r0rwz_NcclGroup.destroy[s <  F : ! KK=*,,:==    J      J    " !r2cdS)N acceleratorr6r9s r0get_transport_namez_NcclGroup.get_transport_nameos}r2c6ddlm}|S)Nr)nccl) cupy.cudar| get_unique_id)clsr|s r0generate_communicator_idz#_NcclGroup.generate_communicator_idrs&""""""!!###r2)F)rXrJrYrJrZr[)rXrJrYrJ)r3N)$__name__ __module__ __qualname____doc__inttuplerrboolr1r7r:rr; ActorHandlerrFrHrOr rrVr^rar SUMrgrjpropertyrorrrwstrrz classmethodrr6r2r0rrs +0UUUUsm U 34 U 12 U$(UUUUn s t    #4(?#@#### ci3     x}      # # 3# 4# # # # T/0 77Sz77 7  7777r !) >   !    . |     !       2 |     !       *;;X; ;;X; !!!!(C$$$$[$$$r2r)loggingtypesrtypingrrrrrrray.exceptionsr ,ray.experimental.channel.accelerator_contextr %ray.experimental.channel.communicatorr r ray.experimental.util.typesr r( getLoggerrrtrr6r2r0rsAAAAAAAAAAAAAA ******KKKKKKTTTTTTTT000000LLL  8 $ $a$a$a$a$a$a$a$a$a$a$r2