ukiwTdZddlmZddlmZddlZddlZddlmZddlm Z m Z ddl Z ddl Z ddl Z ddlmZddlZddlZddlZddlmZdd lmZdd lmZdd lmZdd lmZddlmcm cm!Z"dd lm#Z$ddl%Z&ejNe(Z)e jTZ+ddgZ,dZ-dZ.dZ/dZ0dZ1dZ2dZ3dZ4dZ5eZ6gdZ7d4dZ8d5dZ9d6d7dZ:dZ;d8dZd;d!Z?dd$ZB d?d%ZCd@d&ZD dAd'ZEd(fd)ZFee4*ZGddd+ dBd,ZHddd+ dBd-ZI dCd.ZJdDd/ZKddd0 dEd1ZLddd+ dFd2ZMddd0 dGd3ZNy)HzR Serializations routines for pytrees including array and non-array serialization. ) annotations)PathLikeN)Any)uuid4UUID)ThreadPoolExecutor) distributed) flatten_axes)Format)multihost_utils)tensorstore_implpathzgs://zs3://zpytreedef.jsonz archive.zipTi array_storezArray({dtype}[{shape}])z&Array\(([a-zA-Z0-9_]+)\[([0-9, ]*)\]\) )saveloadload_pytreedefnonblocking_loadnonblocking_savec<tjdk(rytjt j t jtj}tt j|j}dt|S)zHGenerate a thread-local key for ensuring all host finish (de)serializingNdtype)bytes jax_sync_key_) jax process_countr broadcast_one_to_allnp frombufferrrint32rarraytobytesstr)sync_key_unique sync_key_ids t/home/cdr/jupyterlab/.venv/lib/python3.12/site-packages/jax/experimental/array_serialization/pytree_serialization.py_get_unique_sync_keyr*AsmA #88mmEGMM24/288O4<<>?+ [)* ++c tjdkryt|jd}t |t kDrt dt dtjtj|tjtjt t |z tjg}tj|}ttj|d|ddk(S) zEAll-gather the location of the checkpoint and check if it's the same.rFzutf-8zPath exceeds maximum length of z in multiprocess case.rrN)rrr&encodelen_MAX_PATH_LENGTH ValueErrorr! concatenater"uint8zerosr process_allgatherboolall)rpath_b path_arrays r)_is_str_same_on_all_hostsr9LsA  t9  G $&[## 67G6HI++ ,,~~mmF"((+RXX S[ (.:;<*00<* bffZ]jn45 66r+c|y|r|d|n|}tjjx}|j|tdzyy)N-i) timeout_in_ms)r global_stateclientwait_at_barrier _TIMEOUT_SEC)key extra_tagfull_keyr>s r) _sync_on_keyrD[sR[ %.uAi[ !C(((///f< 8<$3FG=r+cVt|tjtjfSN) isinstancerArrayr!ndarrayxs r)_is_array_likerLcs A 2::. //r+c |yt|rRtj|jjdj t t|jSt|jS)Nnullz, )rshape) rL_ARRAY_TYPE_FORMATformatrnamejoinmapr&rOtype__name__)leafs r) _leaf_to_descrXgs_ \ d  $ $jjooTYYs3 /C%D % FF :  r+c|ytjt|s|Stjt|}|J|j\}}|j dj j dDcgc]7}t |j dkDrt|j 9}}tj|tjj|Scc}w)N],r) rematch_ARRAY_TYPE_REGEXgroupsstripsplitr.intrShapeDtypeStructnumpyr) leaf_descshape_dtype_match dtype_str shape_strrKrOs r) _desc_to_leafriqs  #Y / hh0)<  && &*113)Y#,??3#7#=#=#?#E#Ec#J #a1779~! qwwy> #% #  eSYY__Y%? @@ #sz"_is_remote_path..s. 1Y ! !&"+ . 1s'*)any_REMOTE_URL_PREFIXESrs`r)_is_remote_pathrs~s  1/ 1 11r+ct|rtj|Stj|jj SrF)rspathlibPath expanduserresolvers r) _norm_pathrys:T <<  d  & & ( 0 0 22r+cft|r|jytj|yrF)rsrmtreeshutil)roots r)_rm_dirr~sTKKM MM$r+cnt|}|r|jrtt|j dkDrt|j Dcgc]"}|j t ttfvs!|$}}t|dk7rtd|d|dtjdk(s|r|jr t|t|d|S|jr7tt|j dkDrtd|d|d|Scc}w) zOInspect the destination, set it up for writing, potentially read existing data.rzWRefusing to work on a directory that is not a previous checkpoint. Unrecognized paths: z6. Remove them manually if you're sure you want to use z as the checkpoint directory. overwritezFiles already exist at path: `z `, but you specified `overwrite=`)ryexistsr.listiterdirrR_PYTREEDEF_FILE _ARCHIVE_NAME_ARRAY_STORE_DIRNAME RuntimeErrorr process_indexr~rDr0)r}r pytree_reprdistinct_locationssync_keyrextra_member_pathss r)_set_up_destinationrs5 D $ {{}T$,,.12Q6  /B499 M+? @4A$BB  A % $$6#78//3f5       "&8dkkm ;'  #d4<<>23a7 7v>1&/\4 55 'Bs "D2D2ct|}t|||||}t|sY|stjdk(r@|j d|j r|jstd|t|d|S)zPrepare the directory: check destination, potentially read existing data and overwrite. Raises: RuntimeError: If the destination directory cannot be created. rT)exist_okz*Could not create destination directory at mkdir) ryrrsrrrris_dirrrD)r}rpytreedef_reprrrs r)_prepare_directoryrs D $& I~'98E.  $6'*'8'8':a'?JJJ ;;= EdVL MMx! r+c r|Dcgc]}|t|z }}d}|s+tjdkDrtj}t |D cgc]"\}} t j |t|| $} }} t | D cgc]\} } t j| | c} } tdkDr@dddd} t D]#\} } t j| | | td%fd }tj|ycc}wcc} }wcc} } w) Nr)ocdbt process_idxarrrkvstorebaserT)rcheck_metadatac KtjtDcgc]\}}tj||dc}}d{ycc}}w7 w)N) primary_host)asynciogatherzipts_implasync_serialize)rts_specarrsts_specss r)_serialize_arraysz(_write_arrays.._serialize_arrayssW ..!$13 S' W4@3 44434sA"A A A A) r&rrrrrget_tensorstore_spec _USE_OCDBTmerge_nested_ts_specsr.verify_tensorstore_specrrun)array_store_pathr arr_leaf_idsrrleaf_idpathsrrrdefault_ts_specsdefault_ts_specr expected_pathrs ` ` r) _write_arraysrsD;G Gw c'l * G% G+  1 1 3a 7##%K*-UD)9;&4224z?J7:<;; 144Dh0OQ,?G++OWEQ( ]QQK *626:MHd+M  %%gsM,6tMM4  ++!"/ H;QsD('D- D3c&|s.tjdk(stjdk7rytjj |d}t j|ddd}ttjDcgc]}t j|d|d}}|jd}|Dcgc]}|jd}}tjt j||ycc}wcc}w) zWhen multiple processes are writing, they must write to a per-process location followed by combining them via no-copy links to the final location. rrN dummy_keyTrrrr) rrrosrrSrrrangepoprrcombine_kvstores) kvstore_pathrdummy_key_pathcombined_kvstoreichildren_kvstores_rs r)_finalize_array_storers 3,,.!3s7H7H7Ja7O 77<< k:.11Dd44=?S&&()+ 33Da11:<++6"!*;rs AIr+cntjj||\}}tt |tjj |j }tdtjj|||}tjjfd|S)zBroadcast the prefix tree `a` to the full tree `b` Uses `flatten_axes` for better error messages on mismatched arity but allowing for custom is_leaf in the `a` and `b` trees. is_leaftree_broadcastc|SrFr)ra_idx2leaf_maps r)rz!_tree_broadcast.. s q 1r+) rtreeflattendict enumerate unflattenkeysr structurerT)abra_leavesa_structa_idxa_idx_broadcastrs @r)_tree_broadcastrs xx''7';(H (+,. ((  X~':':'< =% !1!$!3!3Aw!3!GP/ 1? CCr+) max_workersrrcXt5t||||cdddS#1swYyxYw)aSaves the given data structure to the provided directory path. This function provides functionality to serialize and save a data structure comprising JAX arrays, along with its structure to a given directory. It leverages `PyTree` for flattening and reconstructing the data structure. This is a simple experimental array serialization API, for anything more complex and for all checkpointing prefer: https://github.com/google/orbax Args: data: The data structure to be saved. Arbitrary composition of JAX arrays, including nested structures. directory: The directory path where the data will be saved. A local path or a remote URL (e.g., gs://, s3://). For remote URLs, `etils` is required. overwrite: If True, any existing directory with the same name will be overwritten. ts_specs: Optional tensorstore specs to use for serialization. If None, defaults to using the default tensorstore specs. Example: >>> data = {"a": jnp.array([1, 2]), "b": None} >>> save(data, directory) rN)_THREADING_SAVE_LOCK_save)datarrrs r)rrs.2J yI IJJJs )c &t}t|rtjs t dt ||t j}tjj|d\}}td|Ds tdt| }tjdkDr |r tdt|}t!j"|} tjj%t&|| t j(<t+||| ||} g} | j-t.j1t2|| ||t4z } |Dcgc]}t7|s|} }t9|D cgc]\} }t7|s| }} }tjj;|t j}|D cgc]} ||  }} | j-t.j1t<| | |||| Dcgc]}|j?}}tA|dtC|d kDr/tDr)t.j1tF| |j?tA|d ycc}wcc}} wcc} wcc}w) Nz}For saving to remote URLs (e.g., gs, s3) you need the `etils` module installed. You can install it using `pip install etils`.rc |duSrFrrJs r)rz_save..8s !t)r+c3>K|]}|duxs t|ywrF)rL)rnrKs r)rpz_save..9s! ?Q$Y +.+ + ?szLFor serialization, all leaves must be either None or jax.Array-like objects.rzSaving to different locations on different hosts is not supported, because it is extremely fragile. Consider using a single location.array_serializationrend)$r*rsruepath_installedrrris_tensorstore_spec_leafrrrr6r0r9rryutilsserialize_pytreedefrTrX _LEAF_IDS_KEYrappend_serialization_executorsubmitrrrLrleavesrresultrDr.rr)rrrrr data_flat pytreedefrr}rfuturesrrrr ts_specs_flatfutrs r)rr.s\ ! #(Y(?(? / 00Xt%,%E%EG())$8K)L)Y ?Y ? ? 0 114Y??1!3  N OO I $,,Y7.(+ ]I(N.$$%% I~'98E. ' ..(//n.@BC00$ =4t(<$ =$ =#,Y#7P4>$;O!P,P((//(*1*J*J"L--9:=#:-: ..(//%t\= ''szz|'!'x./ z""/1CEEKVXx% >P; (s$I>.I>JJ J Jc @ t|}|Dcgc]}|t|z }}tjd |Dcgc]}tj|t d }}t |D cgc]\} } tj| | c} } tdkDr3dddd} D] } tj| d| t d" fd } tt |tj| Scc}wcc}wcc} } w) Nl rrrrrF)rrrrc KtjtDcgc]\}}tj||c}}d{Scc}}w7 w)N) byte_limiter)rrrrasync_deserialize)shardingrr shardingsrs r)_deserialize_arraysz)_read_arrays.._deserialize_arraysysW#&y(#;"= Xw !!(G,O"=> >>"= >sA"A A A A) ryr&r_LimitInFlightBytesrrrrr.rrrr)rrrrarr_store_pathr arr_pathsrrrrrrrs `` @r) _read_arraysrds../. c, ,?,A BC DD-J -QsD#D7 DcXt|rtjsJdt|tz j }t j|}tt|tj}tjjtj||S)aLLoads a pytree from the given directory. This is a simple experimental array serialization API, for anything more complex and for all checkpointing prefer: https://github.com/google/orbax Args: directory: Directory path to load from. Returns: The loaded pytree with arrays represented as jax.ShapeDtypeStruct's. For checkpointing using remote URLs (e.g., gs, s3) you need `etils` module installed. You can install it using `pip install etils`.)rsrurryr read_textrloadsrTrirrrrrdeserialize_pytreedef)r json_contentraw_treers r)rrsY '7+B+BHGH BY'/9DDF, ZZ %( }hu':':; <&   E77A6 JJr+maskrc\t|rtjsJdt|}|j s Jd|dd}t |}t ||}t ||tj}|$d}tjj|||}tjj||}tjj||} tjj||} td| Dr td tjj|tj} t!| D cgc] \} } | |  }} } |D cgc]} | |  } } |D cgc]} | |  } } t"j%t&|t(z || | }|j+}t!| D cgc]\} }|j-| d}} }tjj/||Scc} } wcc} wcc} wcc}} w) aLoads and reconstructs a data structure from a directory. This is a simple experimental array serialization API, for anything more complex and for all checkpointing prefer: https://github.com/google/orbax Args: directory: Directory path where the data is stored. shardings: Sharding strategy for array objects, either a Sharding or a ShapeDtypeStruct with a Sharding/Format. mask: boolean prefix tree for partial loading, will return None for False leaves. ts_specs: Optional tensorstore specs to use for deserialization. If None, defaults to using the default tensorstore specs. Returns: Reconstructed data. Example: >>> save(data, directory) >>> restored_data = load(directory, SingleDeviceSharding(jax.devices()[0])) rzCheckpoint directory z does not existc |duSrFrrJs r)rzload..s a4ir+rNcL|s!tjjd|S|S)NcyrFr)rs r)rz(load....sr+)rrrT)mrKs r)rzload..sq ^Q ?ar+c3<K|]}t|tywrF)rGr )rnrs r)rpzload..sG9Iv &GszDeserialization with `Format` instead of `Sharding` is not currently supported. Pass ShapeDtypeStruct(shape, dtype, sharding=format) instead.)rsrurryrrrrrrrrTrrrqNotImplementedErrorrrrrrrgetr)rrrrr}rpytree _prefix_maskr leaf_ids_flatshardings_flatrrrrarrs_futrr filled_valuess r)rrs0Y '7+B+BHGH B I $ E/v_EE ' ) $&i0) Xv%,%E%EG( OL XX\\,f 5Fhh   9)((//&'/:-88??9g?>.GGG   ((//(*1*J*J"L-'0 &>* 7(*,*/;>> fut = nonblocking_save(data, directory) >>> print(fut.pytree) # a pytree of jax.ShapeDtypeStruct's >>> print(fut.result()) # None, blocking until the serialization is done rcpt|r*tj|j|jS|SrF)rLrrcrOrrJs r)rz"nonblocking_save..s.-a0'*&:&:177AGG&L67r+) r PyTreeFuturerrrrrrTr)rrrrrs r)rrsW 299 D)y8:E F#xx||89=?#* *r+c tjtjt||||}t ||_|S)aNonblocking alias of load, return an awaitable future with a pytree stub. This is a simple experimental array serialization API, for anything more complex and for all checkpointing prefer: https://github.com/google/orbax Examples: >>> fut = nonblocking_load(directory) >>> print(fut.pytree) # a pytree of jax.ShapeDtypeStruct >>> print(fut.result()) # the fully populated pytree r )rrrrrrr)rrrrrs r)rrsE 299 Iyth:@ A#i(#* *r+)return str | None)rstr | PathLike[str]r!r5))rAr"rBr&r!None)r!r&)rer"r!z!str | None | jax.ShapeDtypeStruct)rr#)rr#r!r)r}rr!r%) r}r#rr5rdict[str, Any]rr5rr"r!r&) r}r#rr5rr&rr5rr") rrr list[Any]r list[int]rzlist[Any | None]rr5)rr5)rrrr&rr5) rPyTreeTrr#rr5rPyTreeT | Noner!r%)rr#rr(rr'rr')rr#r!r)) rr#rr)rr*rr*r!r)) rr)rr#rr5rr*r!utils.PyTreeFuture) rr#rr)rr*rr*r!r+)O__doc__ __future__rrrr\typingruuidrrrr threadingconcurrent.futuresrr|loggingrjax._srcr jax._src.api_utilr jax._src.layoutr jax.experimentalr $jax.experimental.array_serializationr r?jax.experimental.array_serialization.pytree_serialization_utils experimentalrpytree_serialization_utilsrrrurdr! getLoggerrVloggerLockrrrrrrr/rrPr^_MAX_CONCURRENCYr@r)__all__r*r9rDrLrXrirsryr~rrrrrrrrrrrrrrrr+r)r@sQ#  1  *",LOO$   8 $%y~~')"  $.=  3, 7H0 A13 %3IM",1?>'5KO!+.# )#5E#&*#:M$I)-I#6 D-9IJ DJJ+9JEIJ<!T3 3 ,:3 FJ3 lE$E1:E:K*!%@6@60>@6@6H(, $ 7E , .-104 ) - 9K r+