uki4hdZddlmZddlZddlZddlmZmZddlZddl Z ddl Z ddl Z ddl Z ddl Z ddlmZddlZddlmZddlmZddlmZdd lmZdd lmZdd lmZdd lmZdd lmZddlmZm Z!m"Z"m#Z#m$Z%m&Z'm(Z(ddl)Z*ejVejXdZ-ejVej\ddZ.dZ/e j`Z1dZ2ddgZ3ddddddgZ4Gdde5Z6dZ7e jpe9Z:d%dZ;d&dZ<Gdd ejzZ>Gd!d"Z?Gd#d$e?e>Z@y)'z(Array serialization and deserialization.) annotationsN)CallableSequence)Any)array) distributed)sharding)typing)util)Format)_jax)tensorstore_impl)_run_deserialization_run_serializationasync_serializeasync_deserialize _TS_CONTEXT_DEFAULT_BASE_DRIVER_LimitInFlightByteszarr)driverF)rocdbtcheckpoint_write_successziPlease initialize the distributed system via `jax.distributed.initialize()` at the start of your program.zgs://zs3://gcs)r path_regexs3c eZdZy)BarrierTimeoutErrorN)__name__ __module__ __qualname__m/home/cdr/jupyterlab/.venv/lib/python3.12/site-packages/jax/experimental/array_serialization/serialization.pyrrCsr#rzSuggestions for possible fixes: * Check the logs to see if one or more processes failed. * Make sure the training and checkpointing endpoints are close geographically. * Try increasing the timeout you pass to GlobalAsyncCheckpointManager.c8t|tr/tjddj t d|ryydD]}||vst ||cSd|vrrArFrKr"r#r$r:r:ps&P 33*$6**AE>B.%=.!=.<..r#r:c:eZdZd dZdZdZdZdZdZd dZ y) AsyncManagerc0||_|jdz|_d|_d|_d|_t j dkDr)tjjtttjj|_ d|_ y)Ni) _timeout_secs_timeout_in_ms_commit_futures_thread _exceptionjax process_countr global_stateclient ValueError_DISTRIBUTED_SYSTEM_MSG_client_count)r= timeout_secss r$__init__zAsyncManager.__init__s{%D,,t3DDDLDO Q;#;#;#B#B#J . //++22DLDKr#c~|j1|jjrtjdyyy)NzPlease add `.wait_until_finished()` in the main thread before your program finishes because there is a possibility of losing errors raised if the this class is deleted before writing is completed.)rZis_aliveloggerwarningr<s r$__del__zAsyncManager.__del__s6 ||DLL$9$9$; nnKL%<r#c tj}tj}tj d|t j }|j D]}|jtj d|d}|dkDrv|jJt|j}tj d|||jj||jtj d||dk(rz|j%|jtj d|dkDrD|jJ|jj|ttj d|tj j#d t j |z y#t$$r}||_Yd}~yd}~wwxYw) Nz/Starting commit to storage layer by process: %sz3Finished committing to storage layer by process: %srVz)Key used for barrier is %s for process %sz*Finished waiting at barrier for process %srz$on_commit_callback successfully ran!z1Process 0 successfully set key %s in the kv storez//jax/checkpoint/write/async/thread_duration_sec)r\ process_indexr]rhinfotimerYresultrbr8rcwait_at_barrierrX_on_commit_callback key_value_set_CHECKPOINT_SUCCESS monitoringrecord_event_duration_secs Exceptionr[)r=current_processr]thread_start_timefuturekey_for_barrieres r$ _thread_funczAsyncManager._thread_funcs&))+o'')m kkC"$))+((&  kkG"$o  ||'''#4;;/ ?#_ 6 $$_d6I6IJ @# % A   # # /  " " $ ++< = 1 ) )) ,, $ $_6I J ++I% ' nn// ; ))+) )+ doosF:F== GGGctt|_||_t j |j |_|jjy)N)target) next_module_unique_countrcrq threadingThreadr|rZstart)r=rEs r$_start_async_commitz AsyncManager._start_async_commits?+,DK1D##4+<+<=DLLLr#c|je|j}d|_t|tjr6dt |vr)t dj t |tg|y)Nz$DEADLINE_EXCEEDED: Barrier timed out )r[r+r JaxRuntimeErrorr,rr/_BARRIER_TIMED_OUT_MSG)r= exceptions r$r>zAsyncManager.check_for_errorssg "//ido Y 4 4 5 0C N B! IIs9~'=> ?A A o#r#c|j6|jjd|_tjd|j tjdt j dkDrm|j`|jJt|j}|jj||jtjd|yyy)NzThread joined successfullyz!Error check finished successfullyrVzr\r]rcrbr8blocking_key_value_getrX)r=get_keys r$rAz AsyncManager.wait_until_finisheds || lldl kk./ KK34 Q4;;#: \\ %% %%g ll))'43F3FG kk ') $;r#c||_yN)rY)r=futuress r$ _add_futureszAsyncManager._add_futuress "Dr#N)i,)rzSequence[ts.Future]) rr r!rerjr|rr>rArr"r#r$rTrTs' L'R )$#r#rTceZdZdZddd ddZddd d dZ d d dZ d d dZy) GlobalAsyncCheckpointManagerz1Responsible for serializing GDAs via TensorStore.NrE transactionctjd|jgfd}tj||j |j |y)aSerializes Arrays or Arrays via TensorStore asynchronously. TensorStore writes to a storage layer in 2 steps: * Reading/copying from the source after which the source can be modified. * Returns a copy future. * Writing/committing to the storage layer. * Returns a commit future. In asynchronous mode, the serialization waits for the commit future to finish in a separate thread allowing other computation to proceed. Args: arrays: Arrays or Arrays that should be serialized. tensorstore_specs: TensorStore specs that are used to serialize GDAs or Arrays. on_commit_callback: This callback will be executed after all processes have finished writing their checkpoints to disk. Filesystems where atomic rename operations are supported, you can rename from the temporary directory to the final directory. On GCS, you write to the final directory directly and in `on_commit_callback` you write a success file indicating that the serialization was successful because GCS does not support atomic rename operations. transaction: Optional TensorStore transaction to use. z-Waiting for previous serialization to finish.cKtjjfd}tj|d{S7w)Nc6tj||S)N) commit_futurer)ts_implr)arr_inptensorstore_speccommit_futuresrs r$zQGlobalAsyncCheckpointManager.serialize.._run_serializer..BsG,C,C*% - r#)r\ tree_utiltree_mapasynciogather) future_writerrCrrDrs r$_run_serializerz?GlobalAsyncCheckpointManager.serialize.._run_serializer@s?mm,,   m>>=1 11 1s;AAAN)rhrmrArrunrr)r=rCrDrErrrs `` ` @r$rFz&GlobalAsyncCheckpointManager.serializesY@ KK?@+-N 2 KK!"n% /0r#cttjjt|}|j ||||S)Nr)r\treemapget_tensorstore_specrF)r=rCpathsrErtspecss r$serialize_with_pathsz1GlobalAsyncCheckpointManager.serialize_with_pathsTs;XX\\. 6F >>-  r#cT|jtj|||||Sr)rArr)r=rHrDrIrJ concurrent_gbs r$rKz(GlobalAsyncCheckpointManager.deserializeds1    ' '$mV] LLr#cttjjt|}|j |||||Sr)r\rrrrK)r=rHrrIrJrrs r$deserialize_with_pathsz3GlobalAsyncCheckpointManager.deserialize_with_pathsms7 XX\\. 6F   Iv}f) ++r#)rECallable[[], None] | Nonerts_impl.Transaction | None)rCzSequence[jax.Array]r Sequence[str]rErrr)NN ) rHz$Sequence[sharding.Sharding | Format]rDrMrIrNrJrOrint) rHrLrrrIrNrJrOrr)rr r!rPrFrrKrr"r#r$rrs97;04 71 4 71 . 71|7;04 ! 4  . $AE>B') L%=L!=L<L"% L5926 +2+ +2+0 +  +r#r)r3zdict[str, Any] | strreturnbool)r4r)ArP __future__rrQrcollections.abcrr functools itertoolsloggingr-rrnr rr\jax._srcrrr r jax._src.layoutr jax._src.libr $jax.experimental.array_serializationrr5jax.experimental.array_serialization.tensorstore_implrrun_deserializationrrun_serializationrrr TS_CONTEXTr_DEFAULT_DRIVERr tensorstoretspartial_get_tensorstore_metadata _get_metadatarrscountrrar0r2rvrr getLoggerrrhr1r8 StrictABCr:rTrr"r#r$rsH/" .    "LBBB " !!'"C"C)/1 (y(()E)E06eE 1&y(C )D)4( )M   8 $:)>.t~~>.Bd#d#N]+<1Q]+r#