&`ie.ddlmZmZmZmZmZmZmZerddlZddl Z ddl m Z m Z ddl mZmZddlmZddlmZmZddlmZmZmZmZddlmZGd d ZeGd d e ZdS) ) TYPE_CHECKINGAnyDictListOptionalTupleUnionN)ClassMethodNodeDAGNode)COLLECTIVE_OPERATION_KEYIS_CLASS_METHOD_OUTPUT_KEY)ChannelContext) CommunicatorTorchTensorType) AllGatherOp AllReduceOpReduceScatterOp _CollectiveOp) DeveloperAPIc eZdZdZ ddeeededeee e ffdZ de fdZ e ded fd Ze defd Zde fd Zd ddededffdZdS)_CollectiveOperationat Represent metadata for a collective communicator collective operation. Args: inputs: A list of lists of DAGNode. Each nested list inside of inputs should contain exactly one object per actor. If multiple nested lists are provided, then the order of actors should be the same for each nested list. op: The collective operation to perform. transport: The transport to use for the collective operation. Requirements: 1. Input nodes are unique. 2. Actor handles are unique. 3. Actor handles match the custom communicator group if specified. Ninputsop transportcg|_t|D]\}tdkr-t|dkrd|nd}td|dt dDs0t|dkrd|nd}td |d dtt tkr=fd D}t|dkrd|nd}td |d |gD]Y}|}|,t|dkrd|nd}td||Ztt tkr=fdD} t|dkrd|nd}td|d| |dkr} t|dt||kr?tdt|dd|dt||dt | t kr3tdt | d|dt dtt| D]-\} \} } | | krtd| d| d|d| d| d .ԉ|_||_ | tj }t|d|_ t|trFt |t |jkrtddSdS)Nrz at index zExpected non-empty input list.c3@K|]}t|tVdSN) isinstancer ).0nodes k/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/dag/collective_node.py z0_CollectiveOperation.__init__..9s,IITz$00IIIIIIz at list at index z&Expected all input nodes to be DAGNodez , but got cFg|]}|dk|Sr)count)r" input_node input_nodess r$ z1_CollectiveOperation.__init__..Ds<""((44q88888r&zExpected unique input nodesz, but found duplicates: z,Expected an actor handle from the input nodecjg|]/}|dk-|0Sr()r)_get_actor_handle)r"r*current_actor_handless r$r,z1_CollectiveOperation.__init__.._sI'''",22:3O3O3Q3QRRUVVVVVVr&zExpected unique actor handlesz6, but found duplicate actor handles from input nodes: zVExpected all input lists to have the same number of nodes. List at index 0 has length z, but list at index z has length z[Expected all input lists to have the same set of actor handles. List at index 0 has actors z has actors z\Expected all input lists to have the same order of actor handles. List at index 0 has actor z at position z has actor T)r_direct_returnz=Expected actor handles to match the custom communicator group)_actor_handles enumeratelen ValueErrorallsetr.appendzip_opr ACCELERATOR _type_hintr!rget_actor_handles)selfrrrinested_list_error_msg duplicatesr* actor_handleinvalid_input_nodesfirst_actor_handlesjfirstcurrentr/r+s @@r$__init__z_CollectiveOperation.__init__)s' >@'//Y Y NA{;1$$25fQi..>>>>,/q NN>>>&''3/D+E+EEE J256I2J2JJJJJ,/0E,F,FJJJ(1')>??((  ##E7G##$J5:JJIJJJ!"JJ/6JJEFJJJ$ 4  '3I)IdSSS i . . 9..0011S9L5M5MMM S  MMr&returnc8d|jd|jd|jdS)Nz#CollectiveOperation(_actor_handles=z, _op=z , _type_hint=))r1r9r;r=s r$__str__z_CollectiveOperation.__str__sA -"1 - -8 - -/ - - - r&zray.actor.ActorHandlec|jSr )r1rKs r$ actor_handlesz"_CollectiveOperation.actor_handles ""r&c|jSr )r;rKs r$ type_hintz_CollectiveOperation.type_hints r&c|jj+tj}|j|jj}nB|j|j}nt d|S)NzExpected a communicator group)r;communicator_idr get_current communicatorsget_custom_communicatorr4)r=ctx communicators r$get_communicatorz%_CollectiveOperation.get_communicatorsm ? * 6 ,..C,T_-LMLL _ 4 4 6 6 B?BBDDLL<== =r&send_buf torch.Tensor)r[.cFddltfdDstd|}t |jt rtdksJd}t|j}j |j d|zg|j ddR|j |j }| ||nt |jtrtdkr;d}j|}||||jjnUtfdDstdd Dd }jj}||||jj||}nt |jt*rtdksJd}t|j}|j d|zdkrtd |j |j d|zg|j ddR|j |j }||||jj|S) a, Call the collective operation on the input tensor(s). Output tensor(s) are allocated and returned. Args: *send_buf: A variable number of torch tensors to send to the collective operation. The tensors have the same order as the input nodes. Returns: A torch tensor or a tuple of torch tensors containing the results of the collective operation. The output tensors have the same length and order as the input node list of the actor of this operation. rNc3BK|]}t|jVdSr )r!Tensor)r"ttorchs r$r%z/_CollectiveOperation.execute..s/AA1:a..AAAAAAr&z+Expected a torch tensor for each input noder)dtypedevicec3DK|]}|jdjkVdS)rNra)r"r_rZs r$r%z/_CollectiveOperation.execute..s1JJA17hqk&77JJJJJJr&z;Expected all input tensors to have the same dtype, but got cg|] }|j Srd)r"r_s r$r,z0_CollectiveOperation.execute..s#>#>#>AG#>#>#>r&cg}d}|D]U}|}||||z|j}||||z }Vt |S)Nr)numelviewshaper7tuple)flat_bufbufsviewsoffsetr_rhs r$unflatten_fromz4_CollectiveOperation.execute..unflatten_fromsqEF!(( ! $Vfun%<=BB17KK Q% <<'r&zSExpected the first dimension of the input tensor to be divisible by the world size )r`r5r4rYr!r9rr3r1emptyrjrarb allgatherr empty_like allreducereduceOpnnutilsparameters_to_vectorr reducescatter) r=rZrXr_ world_sizerecv_bufrprlr`s ` @r$executez_CollectiveOperation.executes  AAAAAAAAA LJKK K,,.. dh , ,1 Gx==A%%%% AT011J"u{j(717122;77gxH  " "1h / / / / + . .' G8}}!!QK+5+A..&&q(DH4EFFFFJJJJJJJJJ$A#>#>X#>#>#>AA (((!8>>>xHH&&x48;LMMM)>(H== / 2 2 Gx==A%%%% AT011JwqzJ&!++ 6)366#u{z)8AGABBK88gxH  & &q(DH4E F F Fr&r )__name__ __module__ __qualname____doc__rr rrr strrrGrLpropertyrNrrQrYrr|rfr&r$rrs?*9= kkT']#k kE#|"345 kkkkZ     #t$;<###X#?X,H'H ~u%899 :HHHHHHr&rc eZdZdZdedeefdeeefdeeefdeeeff fd Z de ed eeefd eeefd eeeffd Z d Z e defdZxZS)CollectiveOutputNodezORepresent an output node from a communicator collective operation in a Ray DAG. method_name method_args method_kwargsmethod_optionsother_args_to_resolvec||_|td|_|td|_|j|jst dt|||||dS)NFzExpected a collective operation) _inputsgetr _collective_opr _is_class_method_outputr4superrG)r=rrrrr __class__s r$rGzCollectiveOutputNode.__init__s# 4I4M4M $d5 5 .C-F-F &. . $   &t/K &>?? ?      !      r&new_args new_kwargs new_optionsnew_other_args_to_resolvec4t|j||||S)N)r)r _method_name)r=rrrrs r$ _copy_implzCollectiveOutputNode._copy_impls-$     ";     r&c td)NzFCollectiveOutputNode is only supported with dag.experimental_compile())NotImplementedError)r=argskwargss r$ _execute_implz"CollectiveOutputNode._execute_impl,s! T   r&rHc|jSr )rrKs r$ collective_opz"CollectiveOutputNode.collective_op1rOr&)r}r~rrrrr rrrGrrrrrr __classcell__)rs@r$rrs"YY   H  CH~  S#X  $CH~      <  s)  cN  #s(^  $(S>        #3###X#####r&r)typingrrrrrrr r`rayray.dagr r ray.dag.constantsr r ray.experimental.channelr*ray.experimental.channel.torch_tensor_typerrray.experimental.util.typesrrrrray.util.annotationsrrrrfr&r$rsIIIIIIIIIIIIIIIIIILLL SRRRRRRR333333TTTTTTTT .-----aaaaaaaaH7#7#7#7#7#?7#7#7#7#7#r&