BVh dZddlZddlZddlmZddlmZddlmZddlm Z ddlm Z dZ d Z d Z d Zd Zd Z d"dZdZdZdZd"dZdZdZd"dZd"dZdZdZd"dZdZdZd"dZd"dZ d"dZ!dZ" d"d Z# d"d!Z$y)#zIUtilities to construct a TF subgraph implementing distributed All-Reduce.N)device)ops) array_ops)math_ops)nccl_opsc|s td|dj}|D]}|j|j}|js tdt |dk7rOg}|D]F}t j |5|jtj|dgdddH|}||fS#1swYXxYw)aiCheck tensors for isomorphism and flatten. Args: tensors: list of `tf.Tensor` which must all have the same shape. Returns: tensors: a list of `tf.Tensor` which are flattened (1D) views of tensors shape: the original shape of each element of input tensors Raises: ValueError: tensors are empty or non-isomorphic or have unknown shape. tensors cannot be emptyrz)Tensors must have statically known shape.N) ValueErrorshape merge_withis_fully_definedlenr colocate_withappendrreshape)tensorsr tensorreshapedts Z/home/dcms/DCMS/lib/python3.12/site-packages/tensorflow/python/distribute/v1/all_reduce.py_flatten_tensorsrs  . // !*  %+f   V\\ *E+    ! @ AAZ1_H 4   Q 4 ))!bT23444G %44s 'B;;C cg}|D]E}tj|5|jtj||dddG|S#1swYSxYw)a=Reshape tensors flattened by _flatten_tensors. Args: tensors: list of `tf.Tensor` of identical length 1D tensors. shape: list of integers describing the desired shape. Product of the elements must equal the length of each tensor. Returns: list of `tf.Tensor` which are the reshaped inputs. N)rrrrr)rr rrs r_reshape_tensorsr8s`( 3a  1 3ooi''512333 /33s &AA c|j}dt|k7r td|jdj}t j |5||zdk7rld||zz}||kDr`||z }tj|tj|g|jgd}tj||}||fcdddS|dz |z|k\rc||z|z}tj|tj|g|jgd}tj||}||fcdddS||dz |zz }||z }t|dz D cgc]} |c} |gz} tj|| }tj|dtj|g|jgd|d<||fcdddStj||dfcdddScc} w#1swYyxYw)aLike split for 1D tensors but pads-out case where len % pieces != 0. Args: tensor: `tf.Tensor` that must be 1D. pieces: a positive integer specifying the number of pieces into which tensor should be split. Returns: list of `tf.Tensor` of length pieces, which hold the values of thin input tensor, in order. The final tensor may be zero-padded on the end to make its size equal to those of all of the other tensors. Raises: ValueError: The input tensor is not 1D. r input tensor must be 1Dr)dtypeNr ) r rr dimsvaluerrrconcatzerosrsplitrange) rpiecesr tensor_len chunk_sizepad_lenextended_wholepartslast_chunk_size_ piece_lenss r _padded_splitr.Js" ,,%#e*_ . //zz!}""*  0FazV++j * :%")) Y__gYfllC DaI7g~00 QJ* $ 2J&*4")) Y__gYfllC DaI7g~'00*% j'@@.*/ *;| | fS#1swYNxYwcc}w#1swYxYw)aConstruct a subgraph for the first (reduction) pass of ring all-reduce. Args: input_tensors: a list of `tf.Tensor` 1D input tensors of same shape and type. devices: array of device name strings num_subchunks: number of subchunks each device should process in one tick. pred_by_s_d: as produced by _ring_permutations rank_by_s_d: as produced by _ring_permutations red_op: a binary operator for elementwise reduction Raises: ValueError: tensors must all be one dimensional. Returns: list of list of `tf.Tensor` of (partially) reduced values where exactly num_subchunks chunks at each device are fully reduced. rr zinput tensors must be 1DNrL)rr r r$rrr.r)rPr=r:rHrIrQ num_devicesr num_chunks num_ticksrS split_pad_lenrGsplitstickr,new_partial_reductionsr@rank seg_indexpred_dev chunk_indexs rrMrM%sV(M"+A IA    %#e*_ / 00]**Ao)-- K #a GAJ #+M!,{(K a% LLL*  %%3## B--s7$F  F-4AF2 F* 2F; c g}|D]H}tj|d5|j|Dcgc] }|| c}dddJ|Scc}w#1swY[xYw)a&Apply a unary op to each tensor in chunks_by_dev, on same device. Args: f: a unary function over `tf.Tensor`. chunks_by_dev: list of lists of `tf.Tensor`. Returns: new list of lists of `tf.Tensor` with the same structure as chunks_by_dev containing the derived tensors. rN)rrr)frSoutputxrs rrNrNcsk & 'a  1Q4 ' mm1%aQqT%&''' -&''sAA AAA c^t|}t|d}d||zk7r tdt||z }|dz }td|D]}td|Dcgc]}d} }td|D]|} t j || d5td|D]F} || | } | |zd|zz |z} || | }| |z| z}t j|||| |<H ddd~td|D]<} td|D]+} || | } | |zd|zz |z} | |z| z}| ||| |<->g}|D]H}t j |d5|jt j|ddddJ|Scc}w#1swY,xYw#1swYhxYw)aConstruct subgraph for second (scatter) pass of ring all-reduce. Args: pred_by_s_d: as produced by _ring_permutations rank_by_s_d: as produced by _ring_permutations chunks_by_dev: list of list of `tf.Tensor` indexed by ints (device, chunk) Raises: ValueError: chunks_by_dev is not well-formed Returns: list of `tf.Tensor` which are the fully reduced tensors, one at each device corresponding to the outer dimension of chunks_by_dev. rzAExpect number of chunks per device to be divisible by num_devicesr N) rr r0r$rrridentityrr!)rHrIrSrWrXr:rYr\r, passed_valuesrGr@r^r_r`rardres rrOrOus+"M"+=#$**{ "" K MMj;./-Ao)Ay!Cd#(J#78aT8M8 1k "4   ]1-a0 14q-( 4AQ"$k)QX6+E) ^A&("]2a7+'0'9'9H%k2(4- $  44441k "CQ &C!1~a K'1t84 C  =0A5 (5k(B a% CCC$ & ,a  1Q4 , mmI$$Q*+,,, -+944",,s FAF&F#F #F, c|Dcgc]}|j}}t|\}}t|||}|r|Dcgc] }|| }}t||}t |dk7r t ||}|Scc}wcc}w)aConstruct a subgraph for recursive halving-doubling all-reduce. The recursive halving-doubling algorithm is described in (Thakur et al., 2015). The concept is to arrange the participating n devices in a linear sequence where devices exchange data pairwise with one other device in each round. During the gather phase there are lg(n) rounds where devices exchange increasingly smaller sub-tensors with another device at increasingly greater distances, until at the top each device has 1/n of the fully reduced values. During the scatter phase each device exchanges its fully reduced sub-tensor (which doubles in length at each round) with one other device at increasingly smaller distances until each device has all of the fully reduced values. Note: this preliminary version requires that len(input_tensors) be a power of 2. TODO(tucker): relax this restriction. Also, the number of elements in each tensor must be divisible by 2^h where h is the number of hops in each phase. This will also be relaxed in the future with edge-case specific logic. Args: input_tensors: list of `tf.Tensor` to be elementwise reduced. red_op: a binary elementwise reduction Op. un_op: an optional unary elementwise Op to apply to reduced values. Returns: list of `tf.Tensor` which are the fully reduced tensors, one at each device of input_tensors. Raises: ValueError: num_devices not a power of 2, or tensor len not divisible by 2 the proper number of times. References: Optimization of Collective Communication Operations in MPICH: [Thakur et al., 2005] (https://journals.sagepub.com/doi/abs/10.1177/1094342005051521) ([pdf](http://wwwi10.lrr.in.tum.de/~gerndt/home/Teaching/HPCSeminar/mpich_multi_coll.pdf)) r )rr_build_recursive_hd_gather_build_recursive_hd_scatterrr)rPrQrRrr=r reduced_shardsrTs rbuild_recursive_hd_all_reducermsV - -!QXX -' -)-8--mWfM. (671eAh7N7.~wG.Z1_%nez!}k!n= 1 > ::i E%jm[^D 1t8EE EF!" -'>>EEs D0D5 E5D> E cxt|}ttj|d}|d|zk(sJd|}t t d|D]}d|z}|dz}|Dcgc]}g} }t d|D]} | |z|dz k\r| } | |z} || } || }t j| 5tj|| || gd| | <dddt j|5tj|| || gd| | <ddd| }|Scc}w#1swYUxYw#1swYxYw)aRConstruct the scatter phase of recursive halving-doubling all-reduce. Args: input_tensors: list of `tf.Tensor` that are fully-reduced shards. devices: a list of strings naming the devices on which the reconstituted full tensors should be placed. Returns: list of `tf.Tensor` which are the fully reduced tensors. rLrorN) rr0rprqreversedr$rrrr!)rPr=rWrrrsrtrurvr,rwrGleft_idx right_idxrxrys rrkrksG + +q) *( h 'K)KK ' & E!X& 'a 6DJ%&"&J& 1k " I j.j1n -hd(i"h)$i ::h H(//1A17 1B1DEF H 8H ::i I ) 0 0&2B282C2EFG!I 9II IF%& -!'HHIIs" D."D$-"D0$D- 0D9 ct|\}}|Dcgc]}|j}}t||||}t||}t |dk7r t ||}|Scc}w)aConstruct a subgraph for shuffle all-reduce. Shuffle reduce is essentially the algorithm implemented when using parameter servers. Suppose tensor length is n, there are d devices and g gather shards. Each device sends a n/g length sub-tensor to each gather shard. The gather shards perform a reduction across d fragments, then broadcast the result back to each device. The devices then join the g fully reduced fragments they receive from the shards. The gather shards could perform d-1 pairwise reductions, or one d-way reduction. The first is better where reduction Op time is low compared to transmission time, the second better in the other case. Args: input_tensors: list of `tf.Tensor` values to be reduced. gather_devices: list of names of devices on which reduction shards should be placed. red_op: an n-array elementwise reduction Op un_op: optional elementwise unary Op to be applied to fully-reduced values. Returns: list of `tf.Tensor` which are the fully reduced tensors. r )rr_build_shuffle_gather_build_shuffle_scatterrr) rPgather_devicesrQrRr r dst_devicesrlrTs rbuild_shuffle_all_reducer*sk0*-8-#01a1+1()/8.).+F.Z1_%ne*   %Z1_ / 00 & '?a  =+ ,?  a(*< =????. & ''a N1% &'./!/f/.i )$ I& ''' ?? 0''s*(C<:D ? D %D <D D  D ct|}g}td|D]H}tj||5|j t j |ddddJ|S#1swYVxYw)aBuild the scatter phase of shuffle all-reduce. Args: reduced_shards: list of `tf.Tensor` fully reduced shards dst_devices: list of names of devices at which the fully-reduced value should be reconstituted. Returns: list of `tf.Tensor` scattered tensors. rN)rr$rrrrr!)rlrrW out_tensorsrGs rrrqsvK ++ K >a KN #>)).!<=>>> >>s &A''A0 czt|}|t|k7r tdtj}tj}t |D]}t j j||}t|dr |j Jd||z|jxsd|jxsd|jf}||vr g||<g||<||j||||j||t|jt|jfS)a]Partition devices and values by common task. Args: devices: list of device name strings values: list of `tf.Tensor` of same length as devices. Returns: (per_task_devices, per_task_values) where both values are lists of lists with isomorphic structure: the outer list is indexed by task, and the inner list has length of the number of values belonging to that task. per_task_devices contains the specific devices to which the values are local, and per_task_values contains the corresponding values. Raises: ValueError: devices must be same length as values. z#len(devices) must equal len(values)taskzfailed to parse device %s localhostr)rr collections OrderedDictr$ device_lib DeviceSpec from_stringhasattrrjobreplicarlistr)r=rrWper_task_devicesper_task_valuesrGd_specindexs r_split_by_taskrs/$G +CK : ;; ,,.++-/   -a  " " . .wqz :F 66 "fkk&9</'!*<GM ! !- 0/ ::mA&-- .< ))-*:; ! <(+A.jm// /!+. K !aK &q)!, -P((););N1 ::a=>9--m<=>>>!KN!&1B''BQABAB.BZ1_%ne>CsMG# G(#G:;G-G:(,H9%H4H-G7 2G::H H H ct|dkDr||S|s|Sg}|D]7}tj|5|j||ddd9|S#1swYExYw)z9If len(input_tensors) > 1, apply red_f, else apply un_op.r N)rrrr)rPred_frRrTrs r_reduce_non_singletonrsy!    N (   Q (eAh'((( ((s AA$ c>fdfd}t||S)z=Construct hybrid of NCCL within workers, Ring across workers.c8t|t|dgSNrrUr)yrQsubdivrRs r upper_builderz+build_nccl_then_ring..upper_builder s CFFQC GGrct|SNr)rerRrs rrz+build_nccl_then_ring..upper_level_f s M5 99rr)rPrrQrRrrs ``` @rbuild_nccl_then_ringrsH: M6= AArc,fd}t||S)zEConstruct hybrid of NCCL within workers, Recursive-HD across workers.ct|Sr)rm)rerQrRs rz.build_nccl_then_recursive_hd..s9!VUKrr)rPrQrRrs `` rbuild_nccl_then_recursive_hdrsK- M6= AArc0fd}t|||S)z@Construct hybrid of NCCL within workers, Shuffle across workers.c t|Srr)rershuffle_red_oprRs rrz.build_nccl_then_shuffle..upper_level_fs #A~~u MMrr)rPr nccl_red_oprrRrs ` `` rbuild_nccl_then_shufflersN M; FFrct|\}}|Dcgc]}|j}}t||\}}t|} g} t|| k7r t dt d| D]*} t || || g|} | j| d,|| } g}t d| D]} |t| | g|| z }t|dk7r t||}|Scc}w)aConstruct a subgraph for Shuffle hybrid all-reduce. Args: input_tensors: list of `tf.Tensor` of same-shape and type values to be reduced. gather_devices: list of device names on which to host gather shards. red_op: binary elementwise reduction operator. upper_level_f: function for reducing one value per worker, across workers. Returns: list of `tf.Tensor` of reduced values. Raises: ValueError: inputs not well-formed. zGFor shuffle hybrid, gather_devices must contain one device per worker. rr ) rrrrr r$rrrr)rPrrQrr rr=rrr9rrCrlrrTs r_build_shuffle_hybridrs#"*-8-, -!QXX -' -*8-*P''&'+)K' + ,, K (a*!~a016;N ^A&'( !+.. K 4a,  /244N4 Z1_%ne.upper_builderLs# #g,!' 00rct|SrrrrRrs rrz.build_shuffle_then_ring..upper_level_fO - ??rr)rPrrred_n_oprQrRrrs ` `` @rbuild_shuffle_then_ringrIs$0@ ^X} >>rc@fdfd}t|||S)zCConstruct hybrid of Shuffle within workers, Shuffle across workers.c t|Srr)rrQsecond_gather_devicesrRs rrz1build_shuffle_then_shuffle..upper_builderXs #G-B$*E 33rct|Srrrs rrz1build_shuffle_then_shuffle..upper_level_f[rrr)rPfirst_gather_devicesrrQrRrrs ``` @rbuild_shuffle_then_shufflerUs'3@ )6= BBrr)%__doc__rrptensorflow.python.frameworkrrrtensorflow.python.opsrrrrrr.r4r6rJrUrMrNrOrmrjrkrrrrrrrrrrrrrrrrsP <++**:$20j:-B:$|37(V;&|$.b3l&R"JD"J&"KJ:0f BB37G'V59 >EI Br