VpfHdZddlmZddlmZmZddlmZddlZddlm Z ddl Z ddl m Z ddlmZmZmZddlmZdd lmZdd lmZdd lmZdd lmZdd lmZddlmZddlmZddlmZddlmZddl m!Z"ddlm#Z#ddl$m%Z%ddl Z&dBdZ'dCdDdZ(dEdZ)dZ*d Z+dFdGd$Z,dHdId'Z-dJd*Z.ed+Z/ed,Z0ed-Z1dKd2Z2dLd5Z3ej4d6Z5e56e2d7Z7e58e7ej9e5d8d9Z:ee:d!ej;e5<ee:d!dej<e5<d:Z=ej>e5e=dKd;Z?dMd=Z@ej4d>ZAeA6e?d?ZBeA8eBej9eAd@ejCeAdAZDej>eAeDdS)NzDUtilities for synchronizing and communication across multiple hosts.) annotations)partial lru_cache)OptionalN)Any) tree_flattentree_maptree_unflatten)core)ad)batching)mlir)array)sharding_impls)pxla)xla)pjit) PartitionSpec) distributed)safe_zipxrreturncttjttjd|SNraxis)jaxtreemaprjnpsumrs `/var/www/html/nettyfy-visnx/env/lib/python3.11/site-packages/jax/experimental/multihost_utils.py_psumr$*s' gcgA... 2 22in_tree is_source bool | Nonec tjdkr*tjtj|Stjdkt jtj tjtj }tj |dtdfd}d}tj||}tjttj t|}tj||S) aBroadcast data from a source host (host 0 by default) to all other hosts. Args: in_tree: pytree of arrays - each array *must* have the same shape across the hosts. is_source: optional bool denoting whether the caller is the source. Only 'source host' will contribute the data for the broadcast. If None, then host 0 is used. Returns: A pytree matching in_tree where the leaves now all contain the data from the first host. Nr processes local_devicesr,cr|}ntj|}tj|d}t|Sr)np zeros_like expand_dims host_local_array_to_global_array)rinp global_meshr'pspecs r#pre_jitz%broadcast_one_to_all..pre_jitGsG cc M!  c .1 % % %C +Ce D DDr%cPtj|dS)Nr)r/asarrayaddressable_datar"s r#post_jitz&broadcast_one_to_all..post_jitOs :a((++ , ,,r% out_shardings)r process_countrrr/r8 process_indexrdevicesreshapelocal_device_countshardingMeshPjitr$ NamedSharding)r&r'r?r6r:out_treer4r5s ` @@r#broadcast_one_to_allrH.sY A 8<< G , ,,!##q(I kmmWS.00#2H2J2JKK  !!'+IJJ+ K..%EEEEEEE--- HLL' * *'SWU#,*D*D133++ ""( h ) ))r%namestrctjtj|}t |d|ddS)z+Creates a barrier across all hosts/devices.z$sync_global_devices name mismatch ('z')N)r/uint32zlibcrc32encode assert_equal)rIhs r#sync_global_devicesrRXsEi 4;;==))**!qAAAABBBBBr%c|SNr"s r# _identity_fnrV`s (r%c t|tjrQ|jsJtj|jj}tt||}ntj dkrtj|Stjtjtj tj}tj|d}t'd}tj||}tj| jdks|stj d t/j j j}t7j|t7j||} fdtjD} tj| j|| } |5ttd| }dddn #1swxYwYtj| dS)Nr;r*r+r,rrc:g|]}tj|SrU)r device_put).0d host_np_arrs r# z3_handle_array_process_allgather..|s% H H HqCN; * * H H Hr%)! isinstancer ArrayImplis_fully_addressabler GSPMDShardingget_replicatedrB_device_assignmentrrVrr=r/r8r?r@rArCrDrFndimr1r ShapedArrayshapedtypermesh_local_to_globalget_array_mappingr-$make_array_from_single_device_arraysr9) r3tiledrepsoutr?r4r5saval global_avalbufs global_arrr\s @r#_handle_array_process_allgatherrsdsCU_%%?c.F?  ' 6 6 ' ) )D 0$|4 0 0 0 5 5CC a Z__hs{}}%%--c.?.A.A.1.D.F.FHHG,##G-KLLK kNNE "";66A*S//K1EN;Q777k  K-{/@ A AD+T+E22D::K I H H HC4E4G4G H H HD;1d$$J ?? 2DT 2 2 2: > >c??????????????? C((++ , ,,s< H((H,/H,FrkboolcNfd}tj||S)aGather data from across processes. Args: in_tree: pytree of arrays - each array _must_ have the same shape across the hosts. tiled: Whether to stack or concat the output. Defaults to False i.e. stack into a new positional axis at index 0. Returns: Pytrees of numpy arrays. * If the input is a non-fully addressable jax.Array, then the data is fully replicated. * If the input is numpy array or fully addressable jax.Array, then the output shape is dependent on the `tiled` argument. If its False, then the output will be stacked else concatenated. * If the input is a scalar, then the output will be stacked. c$t|SrT)rs)r3rks r#_pjitz process_allgather.._pjits *3 6 66r%)rrr)r&rkrws ` r#process_allgatherrxs2&77777 eW % %%r% fail_messagect|}tjtjd||st |d|d|ddS)z9Verifies that all the hosts have the same tree of values.cBtjtj|SrT)r/allequalr"s r#zassert_equal..srx|(<(<r%z Expected: z; got: .N)rHr tree_utiltree_allr AssertionError)r&rzexpecteds r#rPrPs !' * *(    m< bool: # Should save an on-demand checkpoint for preemption if multihost_utils.reached_preemption_sync_point(step_id): return True # Should save a regular checkpoint return step_id - last_saved_checkpoint_step >= save_interval_steps ``` Preemption notice is provided by the cluster scheduler to notify the application in advance before it gets evicted. By default, we use SIGTERM as the signal for preemption notice. TODO(b/230630494): Add instructions for customized preemption notice. Returns: A boolean indicating whether all hosts have reached a synchronization step after some hosts are preempted. Raises: RuntimeError: if preemption sync manager has not been inititialized. NFz1Preemption sync manager has not been initialized.)r global_stateclientpreemption_sync_manager RuntimeErrorreached_sync_point)r sync_managers r#reached_preemption_sync_pointrsJL$, 5)A, J K KK  ( ( 1 11r%cBtj|||dS)NT) tupled_args)pjit_libflatten_axis_resources)rIr& pspecs_thunks r#_flatten_pspecsrs+  ( G\\^^ 7 7 77r%cRtj|tj||SrT)rrhri) local_avalmeshr5s r#_local_to_global_avalrs) "4)?)F)F#- / //r%cRtj|tj||SrT)rmesh_global_to_localri)rprr5s r#_global_to_local_avalrs) "4)?)F)F#. 0 00r%arrr4jax.sharding.Meshr5c|tdttjr jsSttjr8tjt jjrtjt j |j |}ttjr2j |j rdj D}nLtjfd|jD}t't)jjj||}t/j|t j |||t3|j jjS)N`None` is not a valid input to the pspecs argument. Please use jax.sharding.PartitionSpec() if you wanted to replicate your input.cg|] }|j SrU)data)rZrs r#r]z9host_local_array_to_global_array_impl..s 5 5 5af 5 5 5r%c&g|] \}}|SrUrUrZr[indexrs r#r]z9host_local_array_to_global_array_impl..7OOO Au E OOOr%) ValueErrorr^rr_r`rBr PmapShardingr/rF local_meshis_equivalent_tordaddressable_shardsrcanonicalize_dtypedevices_indices_maprfitemsrr rergrbatched_device_putlistr?flat)rr4r5local_shardingarraysrps` r#%host_local_array_to_global_array_implrs ]  N O OOU_%%c.F JU_%%* lCL-+/+/ (3--C<--k.DeLL. eo&&O l##NCH==O 5 5c4 5 5 5FF  % %COOOO&::39EEKKMMOOOF& sy#),,k5BB+  3<--k5AA d;)1677 9 99r% local_inputspspecsct|\}}td|tj|}fdt ||D}t ||S)a Converts a host local value to a globally sharded jax.Array. This function takes host-local data (which might be different across hosts), and populates a global array with this data, where each device on each host, get the appropriate slice of the data according to sharding defined by the global_mesh/pspects. For example: >>> global_mesh = jax.sharding.Mesh(jax.devices(), 'x') >>> pspecs = jax.sharding.PartitionSpec('x') >>> host_id = jax.process_index() >>> arr = host_local_array_to_global_array(np.arange(4) * host_id, mesh, pspecs) # NB: assumes jax.local_device_count() divides 4. # doctest: +SKIP The resulting array will have the shape (4 * num_processes) and will have distributed value of: (0, 1, 2, 3, 0, 2, 4, 6, 0, 3, 6, 9, ... ), where each slice np.arange(4) * host_id will be partitioned across the corresponding host's devices. Similarly: >>> mesh = jax.sharding.Mesh(np.array(jax.devices()).reshape(jax.process_count(), jax.local_device_count()), ['host', 'dev']) >>> pspecs = jax.sharding.PartitionSpec('host') >>> host_id = jax.process_index() >>> arr = host_local_array_to_global_array(np.arange(4) * host_id, mesh, pspecs) # doctest: +SKIP will create the same distributed value (0, 1, 2, 3, 0, 2, 4, 6, ...), however each slice np.arange(4) * i will be *replicated* across corresponding host devices. On the other hand, if pspecs = PartitionSpec(), which means replication across all axes, then this snippet: >>> pspecs = jax.sharding.PartitionSpec() >>> arr = host_local_array_to_global_array(np.arange(4), mesh, pspecs) # doctest: +SKIP will have the shape (4,) and the value (0, 1, 2, 3) will be replicated across all hosts and devices. It is an undefined behavior to have not identical local_inputs with pspec indicating data replication. You can use this function to transition to jax.Array. Using jax.Array with pjit has the same semantics of using GDA with pjit i.e. all jax.Array inputs to pjit should be globally shaped. If you are currently passing host local values to pjit, you can use this function to convert your host local values to global Arrays and then pass that to pjit. Example usage. >>> from jax.experimental import multihost_utils # doctest: +SKIP >>> >>> global_inputs = multihost_utils.host_local_array_to_global_array(host_local_inputs, global_mesh, in_pspecs) # doctest: +SKIP >>> >>> with mesh: # doctest: +SKIP >>> global_out = pjitted_fun(global_inputs) # doctest: +SKIP >>> >>> host_local_output = multihost_utils.global_array_to_host_local_array(global_out, mesh, out_pspecs) # doctest: +SKIP Please note ths function requires global mesh to be a continuous mesh, meaning that devices that belong to each host should form a subcube in this mesh. To move local data to global array with non-continuous mesh use jax.make_array_from_callback or jax.make_array_from_single_device_arrays instead. Args: local_inputs: A Pytree of host local values. global_mesh: A jax.sharding.Mesh object. The mesh must be a contiguous mesh, that is all hosts' devices must form a subcube in this mesh. pspecs: A Pytree of jax.sharding.PartitionSpec's. Returns: A pytree of global arrays. z input pspecscPg|]"\}}t||#Sr4r5"host_local_array_to_global_array_pbind)rZr3in_specr4s r#r]z4host_local_array_to_global_array..YsK #w)--c{4;.==r%rrrhashable_pytreerr )rr4r flat_inpsr& in_pspecsout_flats ` r#r2r2s^$L11)Wng&6v>>@@)#9i88(  * **r%r2c^ttj|j|j||SrT)rr rerfrgrr4r5s r#ltg_abstract_evalrc/  sy#),,k5 B BBr%c (tj|fi|fSrTrct_paramss r#rri 49"GGGG'Jr%c |\} |\} |dn|} t|} | | | t| } t| || } | | fS)Nr)rinsertrDrr) insert_axisspmd_axis_name axis_size axis_name main_typevals_indims_inr4r5rr[ new_parts new_pspecys r# ltg_batcherrlst"!"!$,dd.)5kk) 1i   m)(--[ .33! A+r%c|gSrTrUctxrr4r5s r# _ltg_loweringr} *r%c<|tdttjr jrSt j||}t j|j|}ttj j j ||}ttjr\j|jrj}nt j|}|j}tj|||dSt%jfd|j D}t-j|||t1|jjjS)NrT) committedc&g|] \}}|SrUrUrs r#r]z9global_array_to_host_local_array_impl..rr%)rr^rr_r`rrBrFrrr rerfrgrrd_arraysrYrrrrrrrr?r)rr4r5global_shardingrrrresharded_arrays` r#%global_array_to_host_local_array_implrs ]  N O OOU_%%#*B JL..{EBB/<--k.DeLL.$ sy#),,k5BB*U_%%3 |$$_ch??'{ffsO<>> from jax.experimental import multihost_utils # doctest: +SKIP >>> >>> global_inputs = multihost_utils.host_local_array_to_global_array(host_local_inputs, global_mesh, in_pspecs) # doctest: +SKIP >>> >>> with mesh: # doctest: +SKIP ... global_out = pjitted_fun(global_inputs) # doctest: +SKIP >>> >>> host_local_output = multihost_utils.global_array_to_host_local_array(global_out, mesh, out_pspecs) # doctest: +SKIP Args: global_inputs: A Pytree of global jax.Array's. global_mesh: A :class:`jax.sharding.Mesh` object. The mesh must be contiguous meaning all local devices of the host must form a subcube. pspecs: A Pytree of :class:`jax.sharding.PartitionSpec` objects. Returns: A Pytree of host local arrays. z output pspecscPg|]"\}}t||#Sr"global_array_to_host_local_array_pr)rZr3or4s r#r]z4global_array_to_host_local_array..sK #q)--c{45.77r%r)rr4rrrG out_pspecsrs ` r# global_array_to_host_local_arrayrsB%]33)X'7??AA*Y 33( ( + ++r%rc^ttj|j|j||SrT)rr rerfrgrs r#gtl_abstract_evalrrr%c (tj|fi|fSrTrrs r#rrrr%c|gSrTrUrs r# _gtl_loweringrrr%)rrrrrT)r&rr'r(rr)rIrJ)F)r&rrkrtrr)ry)rzrJ)rrrrt)rrr4rr5r)rrr4rrr)rrr4rrr)E__doc__ __future__r functoolsrrtypingrrMrr jax.numpynumpyr jax.tree_utilrr r jax._srcr jax._src.interpretersr r rrrrjax.interpretersrrrjax.experimental.pjit jax.shardingrrDr jax._src.utilrr/r$rHrRrVrsrxrPrrrrrr2 Primitiverdef_implrdef_abstract_eval deflinear2rspmd_axis_primitive_batchersaxis_primitive_batchersrregister_loweringrrrr defvectorizedrrUr%r#rsKJ""""""((((((((  @@@@@@@@@@$$$$$$******&&&&&&######&&&&&& %%%%%%&&&&&&++++++ """"""3333'*'*'*'*'*TCCCC   ---B&&&&&0AAAAA+2+2+2+2\ 77 7 // / 00 0 9 9 9 9FW+W+W+W+r&4T^4V%W%W""++,QRRRBBB#445FGGG  0JJKKK   MTGMM%&HIGNwHH !CD   9=III3333B),),),),V&4T^4V%W%W""++,QRRRBBB#445FGGG  0JJKKK9:::   9=IIIIIr%