vh~UdZddlmZddlmZmZddlZddlmZddl Z ddl m Z ddl mZmZddlmZddlmZdd lmZdd lmZdd lmZdd lmZdd lmZddlmZddlmZddlmZddlm Z ddl!m"Z"ddlm#Z#ddl$m%Z%ddl Z&d/dZ'd0d1dZ(d2dZ)dZ*dZ+d3d4dZ,d5d6dZ-d7dZ.edZ/edZ0ed Z1 d8d!Z2 d9d"Z3ejhd#Z5e5jme2d$Z7e5jqe7ejre5d%d&Z:ee:dejve5<d'Z<ejze5e< d8d(Z> d:d)Z?ejhd*Z@e@jme>d+ZAe@jqeAejre@d,eje@d-ZCejze@eCd;d.ZDy)z_psum..+sq Ajaxtreemap)xss r_psumr'*s A2 FFr!c tjdk(r.tjjtj |Stj dk(t jtjjtjtj}tjj|dtdfd}d}tjj||}tjttjj!t|}tjj||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. r processes local_devicesr+cr|}ntj|}tj|d}t|S)Nrr)np zeros_like expand_dims host_local_array_to_global_array)rinp global_mesh is_sourcepspecs rpre_jitz%broadcast_one_to_all..pre_jitGs8 c MM! c ..1 %C +Ce DDr!cJtj|jdS)Nr)r# device_getaddressable_datars rpost_jitz&broadcast_one_to_all..post_jitOs >>!,,Q/ 00r! out_shardings)r# process_countr$r%r/asarray process_indexr devicesreshapelocal_device_countshardingMeshPjitr' NamedSharding)in_treer5rAr7r;out_treer4r6s ` @@rbroadcast_one_to_allrK.s  A 88<< G ,,!!#q(I kkmWS..0#2H2H2JK  !!'+IJ+ K.%E1 HHLL' *'SWWU#,,*D*D13+ "( h ))r!ctjtj|j }t |d|dy)z+Creates a barrier across all hosts/devices.z$sync_global_devices name mismatch ('z')N)r/uint32zlibcrc32encode assert_equal)namehs rsync_global_devicesrTXs3ii 4;;=)*!q8bABr!c|SNrs r _identity_fnrX`s (r!c ,t|tjr|jst|jt j r%|jjt}nIt jj|jj|jj}tjt||}ntj dk(r0t#j$|}|st#j&|dS|St#jtj(j+tj tj,}tjj/|d}td}tjj ||}t#j$|}|j0dk(s|st#j&|d}t3j4|j6|j8} t;j<|t;j>|| } tj@D cgc]} tjB|| } } tjD| j6|| } tjttj |t| }t#j$|jGdScc} w) N)spec) memory_kindr<r)rr.r*r+)$ isinstancer ArrayImplis_fully_addressablerDrrHupdaterF GSPMDShardingget_replicated_device_assignmentr[r#rGrXr>r/r?r1rArBrCrEndimr ShapedArrayshaperrmesh_local_to_globalget_array_mappingr, device_put$make_array_from_single_device_arraysr:)r3tiledrepsoutrAr4r6s host_np_arraval global_avaldbufs global_arrs r_handle_array_process_allgatherrtds<U__%c.F.F#,, < <= \\ ac *d  ) ) 8 8 ,, ) )s||7O7O9Qd 3#'',d 3C 8C a JJsOc05R^^Ca (>3>hhs{{}%--c.?.?.A.1.D.D.FHG,,##G-KLK kNE "";6A**S/K1ENN;Q7k   K--{/@/@ AD++T++E2D:K584E4E4G HqCNN; * HD H;;1d$J E#'', # 1 1+qs C EEO QC C((+ ,, Is0LFcNfd}tjj||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. ct|SrV)rt)r3rjs r_pjitz process_allgather.._pjits *3 66r!r")rIrjrws ` rprocess_allgatherrxs&7 eW %%r!ct|}tjjtjj d||st |d|d|dy)z9Verifies that all the hosts have the same tree of values.cLtjtj|SrV)r/allequalrs rr zassert_equal..srxx|(<r!z Expected: z; got: .N)rKr# tree_utiltree_alltree_mapAssertionError)rI fail_messageexpecteds rrQrQsa !' *(    mm 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 initialized. Fz1Preemption sync manager has not been initialized.)r global_stateclientpreemption_sync_manager RuntimeErrorreached_sync_point)step_id sync_managers rreached_preemption_sync_pointrsNL$$, ))AA, J KK  ( ( 11r!c>tj|||dS)NT) tupled_args)pjit_libflatten_axis_resources)rRrI pspecs_thunks r_flatten_pspecsrs  ( ( G\^ 77r!cVtj|tj||SrV)rrfrg) local_avalmeshr6s r_local_to_global_avalrs' " "4)?)?)F#- //r!cVtj|tj||SrV)rmesh_global_to_localrg)rprr6s r_global_to_local_avalrs' " "4)?)?)F#. 00r!c"| tdt|tjr|js|St|tjrCt|j t j jrtj|}t j j|j|}t|tjrJ|j j||jr$|jDcgc]}|j}}nStj |}|j#|j$j'Dcgc] \}}|| }}}t)t+j,|j$|j.||}t1j2|t j j|||t5|jj6j8Scc}wcc}}w)N`None` is not a valid input to the pspecs argument. Please use jax.sharding.PartitionSpec() if you wanted to replicate your input.) ValueErrorr\r r]r^rDr# PmapShardingr/rH local_meshis_equivalent_torcaddressable_shardsdatarcanonicalize_dtypedevices_indices_mapreitemsrr rdrrbatched_device_putlistrAflat) arrr4r6local_shardingrarraysrqindexrps r%host_local_array_to_global_array_implrs ]  N OOU__%c.F.F JU__%* llCLL--+/ ((3-C<<--k.D.DeL. eoo& ll##NCHH=!44 5aff 5F 5  %C'::399EKKMO Au E OFO& syy#)),k5B+  3<<--k5A d;))11667 996Os H"H c t|\}}td|tj|}t ||Dcgc]\}}t j |||}}}t||Scc}}w)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 this 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 pspecsr4r6)rrrhashable_pytreer"host_local_array_to_global_array_pbindr) local_inputsr4pspecs flat_inpsrI in_pspecsr3in_specout_flats rr2r2 s^$L1)Wng&66v>@) #9i8 #w)--c{4;.=(  ** "A/r2clttj|j|j||SrV)rr rdrerrr4r6s rltg_abstract_evalrg-  syy#)),k5 BBr!c 0tj|fi|fSrV)rrct_paramss rr r m499"GG'Jr!c|\}|\}|jdn |j}t|} | j||t| } tj ||| } | |fS)Nr) spmd_namerinsertrFrr) insert_axis axis_datavals_indims_inr4r6rrq new_parts new_pspecys r ltg_batcherrpso"!"!))1dy7J7J)5k) 1i m)(--[ .3! A+r!c|gSrVrWctxrr4r6s r _ltg_loweringr} *r!c| tdt|tjr|jr|St j j||}t j j|j|}ttj|j|j||}t|tjrn|j j||jr |j }n"t j"||}|j }tj|||dSt%j&|}|j)|jj+D cgc] \}} ||  }}} t-j.|||t1|jj2j4Scc} }w)NrT) committed)rr\r r]r^r#rDrHrrr rdrerrrc_arraysrhrrrrrrrrAr) rr4r6global_shardingrrrresharded_arrayrqrs r%global_array_to_host_local_array_implrs ]  N OOU__%#*B*B JLL..{EB/<<--k.D.DeL.$ syy#)),k5B*U__% ||$$_chh?{{fsOGc t|\}}td|tj|}t ||Dcgc]\}}t j |||}}}t||Scc}}w)a Converts a global `jax.Array` to a host local `jax.Array`. 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 and the output from pjit will also be globally shaped jax.Array's You can use this function to convert the globally shaped `jax.Array` output from pjit to host local values again so that the transition to jax.Array can be a mechanical change. 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 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 pspecsr)rrrrr"global_array_to_host_local_array_prr) global_inputsr4rrrJ out_pspecsr3ors r global_array_to_host_local_arrayrsB%]3)X'77?A* Y 3 #q)--c{45.7( ( ++ rrclttj|j|j||SrV)rr rdrerrs rgtl_abstract_evalrrr!c 0tj|fi|fSrV)rrrs rr r rr!c|gSrVrWrs r _gtl_loweringrrr!ctjj}| td|s t d|Dchc]}|j }}t j |vr t dt|dk(r|S|jt|}|Dcgc]}|j |vs|c}Scc}wcc}w)ayReturns the subset of the provided devices that are live and healthy. This API is under active development and is not stable. `live_devices` is a low-level fault tolerance primitive that can be used to implement fault tolerant multi-process JAX programs. Barrier Semantics It's important that every process agrees on which devices are live to avoid the processes' behavior from diverging. For example, imagine a set of processes trying to run an AllGather, but they all disagree on which devices should be participating in the AllGather. This is buggy. To ensure that every process agrees on the set of live devices, the `live_devices` function has barrier-like semantics. Consider an invocation `live_devices(devices)` where `devices` includes devices across a set of processes P. The invocation acts as a barrier, waiting for every process in P to call `live_devices(devices)`. Afterwards, `live_devices` returns the same set of live devices `A` to all the processes in P. This ensures that every process agrees on the set of live devices. `live_devices` does not actually act as a barrier for *every* process in P because some processes in P might have failed. Instead, the `live_devices` function waits only for the processes with a device in the returned set of live devices A. An Example Imagine we have four processes, each with two devices: Process A: Devices 1 and 2 Process B: Devices 3 and 4 Process C: Devices 5 and 6 Process D: Devices 7 and 8 Further imagine that process D fails and that every process calls `live_devices(jax.devices())`. The invocation returns devices 1, 2, 3, 4, 5, and 6. Because these devices are hosted by processes A, B, and C, the call to `live_devices` acts as a barrier across processes A, B, and C. Process D, which failed, is ignored. Args: devices: A list of devices. The provided devices must include at least one local device. Returns: The subset of the provided devices that are live and healthy. Raises: RuntimeError: If the distributed runtime was not initialized. ValueError: If no local devices are provided. z Distributed JAX not initialized.zNo devices provided.z/Provided devices do not have any local devices.r)) rrrrrr@rlenget_live_nodesr)rArrq process_idslive_process_idss r live_devicesrsl  # # * *& ^ 9 ::  + ,,*12Q2+2{2 F GG N**4 +<= D3C C! DD3 EsB8B=1B=)r&rreturnrrV)rIrr5z bool | Nonerr)rRstr)F)rIrrjboolrr))rr)rintrr)rrr4jax.sharding.Meshr6r)rrr4rrr)rrr4rrr)rAlist[xla_client.Device]rr)E__doc__ __future__r functoolsrrrNtypingrr# jax.numpynumpyr jax.tree_utilrrjax._srcr jax._src.interpretersr r r r rrjax.interpretersrrr jax.shardingrrFr jax._src.utilrr jax._src.librr/r'rKrTrXrtrxrQrrrrrr2 Primitiverdef_implrdef_abstract_eval deflinear2rfancy_primitive_batchersrregister_loweringrrrr defvectorizedrrrWr!rrsFK"(  6$*&#& %+ "#G'*TC "-J&0A+2\ 7 7 / / 0 0 9  9/ 98; 9FW+W+$5W+?BW+r&4T^^4V%W""++,QRB#445FG  0JK IPI!!"DE 9=I3 3/38;3B),),%6),@C),V&4T^^4V%W""++,QRB#445FG  0JK9: 9=ILEr!