uki edZddlmZddlmZmZddlZddlZddlm Z ddl Z ddl m Z ddlmZmZddlmZddlmZdd lmZdd lmZdd lmZdd lmZdd lmZddlmZddlmZddlmZddlmZ ddlm!Z!ddl"m#Z#ddlm$Z$ddl%m&Z&ddl Z'd2dZ(d3d4dZ)dZ*dZ+d5d6dZ,d7dZ-d8d9dZ.d:dZ/edZ0ed Z1ed!Z2 d;d"Z3 dd/ZEGd0d1ZFeFZGy)?zDUtilities for synchronizing and communication across multiple hosts.) annotations)partial lru_cacheN)Any) tree_flattentree_unflatten)core)dtypes)ad)batching)mlir)array)sharding_impls)pxla)pjit)prng) PartitionSpec) distributed)safe_zip) xla_bridge) xla_clientcDtjjd|S)NcFtj||jdS)Nr)dtypeaxis)jnpsumrxs [/home/cdr/jupyterlab/.venv/lib/python3.12/site-packages/jax/experimental/multihost_utils.pyz_psum..-sq Ajaxtreemap)xss r _psumr(,s A2 FFr"c tjdk(r.tjjtj |Stj dk(t jtjjtjtj}tjj|dtdfd}d}tjj||}tj5tjt t|}dddtjj|S#1swY)xYw) 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|SNrr)np zeros_like expand_dims host_local_array_to_global_array)rinp global_mesh is_sourcepspecs r pre_jitz%broadcast_one_to_all..pre_jitIs8 c MM! c ..1 %C +Ce DDr"cJtj|jdS)Nr)r$ device_getaddressable_datars r post_jitz&broadcast_one_to_all..post_jitQs >>!,,Q/ 00r" out_shardings)r$ process_countr%r&r1asarray process_indexrdevicesreshapelocal_device_countshardingMeshPset_meshjitr()in_treer7rCr9r=out_treer6r8s ` @@r broadcast_one_to_allrM0s A 88<< G ,,!!#q(I kkmWS..0#2H2H2JK  !!'+IJ+ K.%E1 HHLL' *' ||K :0swwuAC09H: h ))::s '*E99Fc|SNrs r _identity_fnrQ]s (r"ct|tjr|js|s t dt|j t jr%|j jt}nIt jj|j j|j j}tjt ||}n1tj"dk(r0t%j&|}|st%j(|dS|St%jtj*j-tj"tj.}tj j1|d}td }tj j||}t%j&|}|j2dk(s|st%j(|d}t5j6|j8|j:} t j<|d }t?j@|t jB|| } tjDD cgc]} tjF|| } } tjH| j8|| } tjJ|5tjt t| }dddt%j&jMdScc} w#1swY2xYw) NzFGathering global non-fully-addressable arrays only supports tiled=True)spec) memory_kindr>r*rr0r+r,pspec to array_mapping)' isinstancer ArrayImplis_fully_addressable ValueErrorrFr NamedShardingupdaterH GSPMDShardingget_replicated_device_assignmentrTr$rJrQr@r1rAr3rCrDrErGndimr ShapedArrayshaperprepare_axis_resourcesrmesh_local_to_globalget_array_mappingr- device_put$make_array_from_single_device_arraysrIr<)r5tiledrepsoutrCr6r8s host_np_arraval global_avaldbufs global_arrs r _handle_array_process_allgatherrqasrU__%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  1 1%9Q RE++^55e(*MM FcNfd}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|SrO)rq)r5rgs r _pjitz process_allgather.._pjits *3 66r"r#)rKrgrts ` r process_allgatherrus&7 eW %%r"ctjtj|j }t |d|dy)z+Creates a barrier across all hosts/devices.z$sync_global_devices name mismatch ('z')N)r1uint32zlibcrc32encode assert_equal)namehs r sync_global_devicesr~s3ii 4;;=)*!q8bABr"c d}t|d}tjj||}tjj tjjd||st |d|d|dy) z9Verifies that all the hosts have the same tree of values.cbt|tjr0|js$t j |j dSt j |}|jdk(rt j|d}t j|gtjzSr/) rVrrWrXr1rAr<r_r3concatr$r@rs r concat_in_treez$assert_equal..concat_in_treesy!U__%a.D.D ZZ**1- .. **Q-a 1 NN11 % YYsS..00 11r"T)rgcLtjtj|SrO)r1allequalrs r r!zassert_equal..sbffRXXq\2r"z . Expected: z; got: .N)rur$r%r&rAssertionError)rK fail_messagerriexpected_in_trees r r{r{sz2 '.#XX\\.':  hhll24DcJ L  . SE ; == Lr"ctjjytjj}| t d|j |S)aDetermine whether all hosts have reached a preemption sync step. When any host receives a preemption notice, the notice is propagated to all hosts and triggers a synchronization protocol in the background. The synchronization protocol calculates the maximum step ids from all hosts, and uses the next step id (i.e., max + 1) as the safe step to save a checkpoint. All hosts should continue training more steps until this method returns True, indicating that the `step_id` is equal to the safe step and the hosts should start saving a checkpoint. To use this API, all hosts must start training from the same step and call it at every training step. Example usage: ``` def should_save(step_id: int) -> 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. FzrPreemption sync manager has not been initialized. Make sure the 'jax_enable_preemption_service' config is enabled.)r global_stateclientpreemption_sync_manager RuntimeErrorreached_sync_point)step_id sync_managers r reached_preemption_sync_pointrsVL$$, ))AA,  >   ( ( 11r"c>tj|||dS)NT) tupled_args)pjit_libflatten_axis_resources)r|rK pspecs_thunks r _flatten_pspecsrs  ( ( G\^ 77r"ctj|d}tj|tj||SNrU)rrbrrcrd) local_avalmeshr8s r _local_to_global_avalrs<  / /7O P% " " N , ,U 3Z AAr"ctj|d}tj|tj||Sr)rrbrmesh_global_to_localrd)rmrr8s r _global_to_local_avalrs<  / /7O P% " " N , ,U 3[ BBr"c| tdt|tjr|js|St|tjr.t|j t j js t|dstj|}|jtjk(r8tj|jtjt}|j}t|t j"x}r |j$}t j j'|j(|}t|tjrJ|j j+||j,r$|j.Dcgc]}|j0}}nNtj2|}|j5|jj7Dcgc]}|| }}t9t;j<|j|j||} t?j@| t j j'|||tC|j(jDjF} |r t!j"|jH| S| Scc}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.rar)%rYrVrrWrXrFr$ PmapShardinghasattrr1rr float0zerosraboolr PRNGKeyArray _base_arrayrZ local_meshis_equivalent_tor_addressable_shardsdatacanonicalize_valuedevices_indices_mapvaluesrr r`rbatched_device_putlistrCflat_impl) arrr6r8ris_prng_key_arraylocal_shardingrarraysirmris r %host_local_array_to_global_array_implrs ]  N OOU__%c.F.F Jeoo&: llCLL--,/7>sG7L ((3-CYY&-- ((399BHHTN 3C ))%$S$*;*;<<< //C<<--k.D.DeL. eoo& ll##NCHH=!44 5aff 5F 5  # #C (C&::399ELLNAF& syy#)),k5B+ 3<<--k5A d;))11667 9#   U[[# .. *6s J:6 J?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 pspecsr6r8)rrrhashable_pytreer"host_local_array_to_global_array_pbindr) local_inputsr6pspecs flat_inpsrK in_pspecsr5in_specout_flats r r4r4)s^$L1)Wng&66v>@) #9i8 #w)--c{4;.=(  ** "A/r4clttj|j|j||SrO)rr r`rarrr6r8s r ltg_abstract_evalr-  syy#)),k5 BBr"c 0tj|fi|fSrO)rrct_paramss r r!r!499"GG'Jr"c~|\}|\}|jdn |j}t|} || j||t| } tj ||| } | |fS)Nr) spmd_namerinsertrHrr) insert_axis axis_datavals_indims_inr6r8rrn new_parts new_pspecys r ltg_batcherrsw"!"!))1dy7J7J)5k)] Q "m)(--[ .3! A+r"c|gSrOrPctxrr6r8s r _ltg_loweringr *r"c:| tdt|tjr|jr|St |dst j|}|jtjk(r8t j|jt jt}|j}t|tjx}r |j}t j"j%||}t j"j%|j&|}t)t+j,|j|j||}t|tjr|j"j/||j0r |j2}n"t!j4||} | j2}tj|||d} |r tj|j6| S| Stj8|}|j;|jj=D cgc]} ||  }} t?j@|||tC|j&jDjFScc} w)NrrarT) committed)$rYrVrrWrXrr1rr rrrarrrrr$rFrZrrr r`rr__arraysrerrrrrrrrCr) rr6r8rrglobal_shardingrrrresharded_arrayrirs r %global_array_to_host_local_array_implrs ]  N OOU__%#*B*B J g  ((3-CYY&-- ((399BHHTN 3C ))%$S$*;*;<<< //CLL..{EB/<<--k.D.DeL.$ syy#)),k5B*U__% ||$$_chh?{{fsO>> 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_inputsr6rrrL out_pspecsr5ors r global_array_to_host_local_arrayrsB%]3)X'77?A* Y 3 #q)--c{45.7( ( ++ rrclttj|j|j||SrO)rr r`rarrs r gtl_abstract_evalrrr"c 0tj|fi|fSrO)rrrs r r!r!rr"c|gSrOrPrs r _gtl_loweringrrr"c|Dchc]}|j}}tj|vr td|jt |}|Dcic] }|j|vr|||j"c}Scc}wcc}w)zEReturns the subset of the provided devices that are live and healthy.z/Provided devices do not have any local devices.)rBrrYget_live_nodesr)rrCrn process_idslive_process_idss r _live_devicesr s*12Q2+2{2 F GG**4 +<=  , , !// ** 3 s B%Bc:eZdZdZdZej dZy) _LiveDevicesaA context manager for atomically running code on the set of live devices. THIS API IS UNDER ACTIVE DEVELOPMENT AND IS NOT STABLE. # Overview `live_devices` is a low-level primitive that can be used to make multi-controller JAX programs fault tolerant. A multi-controller JAX program runs across many devices, and the machines that host these devices might fail. `live_devices` is a context manager that yields the current set of healthy devices, allowing you to run JAX code on the healthy devices while ignoring the failed ones. Concretely, `live_devices` is a context manager. You provide it the set of devices you are interested in, and it yields the subset of these devices that are live. In the body of the `with` statement, you can execute arbitrary JAX code using the set of live devices. # Example Usage try: with jax.live_devices(jax.devices()) as devices: # Run JAX code here with devices. pass except: # A device died while executing the with statement above. pass else: # The with statement executed successfully. pass # 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` context manager has barrier-like semantics. Consider an invocation `with 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 `with 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 `with 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. # Atomicity `live_devices` also provides the following transaction-like atomicity property. When a process exits the body of a `with jax.live_devices(...) as devices:` block, there are two possibilities. 1. All processes in `devices` successfully executed all code in the block without any exceptions being raised. 2. All processes in `devices` did not successfully execute the code in the block, and all the processes will raise an exception. Consider the following code. try: with jax.live_devices(...) as devices: pass except: pass # A else: pass # B The atomicity property says that either every process with devices in `devices` will enter the except branch (A) or every process with devices in `devices` will enter the else branch (B). It is impossible for some processes to enter A and others to enter B. TODO: mwhittaker - Link to formal live devices semantics. 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. cd|_yrO)rC)selfs r __init__z_LiveDevices.__init__s DLr"c#Ktjj}| td|s t d|j t |||_d} t|j j}|jd||j }t ||}||_|r||j|jkst |d|y#t$r }|}Yd}~ed}~wwxYw#|j }t ||}||_|r||j|jkst |d|wxYww)Nz Distributed JAX not initialized.zNo devices provided.c|jSrO)id)rns r r!z'_LiveDevices.__call__..s qttr")keyz is not a subset of ) rrrrrYrCrrkeyssort Exceptionitems)rrCr exceptionalivee old_devices new_devicess r __call__z_LiveDevices.__call__sU  % % , ,F ~ ; <<  - .. ||"673dlI L4<<$$&'e jj^j$ kLLk!&'2k dl     K$5$5$7 7K=(<[MJKK8 iLLk!&'2k dl     K$5$5$7 7K=(<[MJKK8s>AE:C&AE& C;/C61C>6C;;C>>AEEN)__name__ __module__ __qualname____doc__r contextlibcontextmanagerr rPr"r rrs*hTLLr"r)r'rreturnrrO)rKrr7z bool | Nonerr)F)rKrrgrrr)r|str))rr)rintrr)rrr6jax.sharding.Meshr8r)rrr6rrr)rrr6rrr)rCzlist[xla_client.Device]rzdict[xla_client.Device, int])Hr __future__r functoolsrrrxrtypingrr$ jax.numpynumpyr jax.tree_utilrrjax._srcr r jax._src.interpretersr r r rrrrrr jax.shardingrrHr jax._src.utilrr jax._src.librr1r(rMrQrqrur~r{rrrrrr4 Primitiverdef_implrdef_abstract_eval deflinear2rfancy_primitive_batchersrregister_loweringrrrr defvectorizedrrr live_devicesrPr"r r)seK"(  6$*&#&%+ "#G(*Z &-R&0C =&.2b 7 7 A A  B B ( ( /( 8;( VW+W+$5W+?BW+r&4T^^4V%W""++,QRB#445FG  0JK IPI!!"DE 9=I(1 (1/(18;(1V),),%6),@C),V&4T^^4V%W""++,QRB#445FG  0JK9: 9=I  ILILV~ r"