VpfBfdZddlmZddlZddlZddlmZmZmZddl m Z ddl Z ddl Z ddl Z ddlZddlZddlZddlmZmZddlZddlmZddlmZdd lmZdd lmZdd lmZdd lmZmZdd lmZddlmZddl m!Z"ddl#m$Z%ddl$Z&ddl'Z(e(j)dddiiZ*dZ+dZ,e j-Z.dZ/dZ0ddgZ1ddddddgZ2Gdde3Z4dZ5e j6e7Z8dbd%Z9d&Z:d'Z;dcd*ZGd3d4Z?dgd8Z@de*dddfdhd?ZAd@ZBdidEZCddde*d+fdjdNZD dkdldYZEdmd[ZFGd\d]ejGZHGd^d_ZIGd`daeIeHZJdS)nz(Array serialization and deserialization.) annotationsN) AwaitableCallableSequence)partial)AnyOptional)array)config) distributed)sharding)sharding_impls)LayoutDeviceLocalLayout)typing)util) xla_extensionfile_io_concurrencylimitz Value removedcheckpoint_write_successfileziPlease initialize the distributed system via `jax.distributed.initialize()` at the start of your program.gs://zs3://gcs)driver path_regexs3ceZdZdS)BarrierTimeoutExceptionN)__name__ __module__ __qualname__r/var/www/html/nettyfy-visnx/env/lib/python3.11/site-packages/jax/experimental/array_serialization/serialization.pyrr;s$r$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. global_shape array.Shape inp_shardingjax.sharding.Sharding data_callback9Callable[[array.Index, jax.Device], Awaitable[jax.Array]]cK|||j}fd|D}tj|d{V}t j|||S)Nc4g|]}||Sr#r#).0dr*device_to_index_maps r% z4create_async_array_from_callback..Ns;,,,!=!4Q!7;;,,,r$)devices_indices_map_addressable_device_assignmentasynciogatherr $make_array_from_single_device_arrays)r&r(r*addressable_da future_arraysdbsr0s ` @r% create_async_array_from_callbackr:Gs %88FF>.,,,,,*,,,-nm,,,,,,,#  3L# ' ''r$c|dj}ddi|jtjtjd|dS)Nridzstd) compressorshapechunks)addressable_datar@npr maximum)arr local_shapes r% _get_metadatarGUsL$$Q''-+6NyA{3344  r$ct|tsdSd|vp*td|DS)NFmetadatac3:K|]\}}t|VdSN)_spec_has_metadata)r._subtrees r% z%_spec_has_metadata..asH#B#B&0a!!#B#B#B#B#B#Br$) isinstancedictanyitems)trees r%rLrL^sd D$   5 t  Bs#B#B48JJLL#B#B#B B BBr$ ckpt_pathstrctjd|tj}|td||d}|d}d||dS)Nz^gs://([^/]*)/(.*)$zWThe ckpt_path should contain the bucket name and the file path inside the bucket. Got: r>r)rbucketpath)re fullmatchDOTALL ValueErrorgroup)rUm gcs_bucketpath_without_buckets r%_get_kvstore_for_gcsrcdswl()RY??!Y G;DGG H HHwwqzz* Z9L M MMr$Focdbtboolctj|dd}|d}did}|r|s1tj|st d|tj|}d|r|n td|tj |d|d <n"|rt||d <n t|d |d <|S) Nzgs:/rzarr)rkvstorez(Checkpoint path should be absolute. Got rdz://)rbaserZrh)rrZ) osrZnormpathreplace startswithisabsr^dirname_DEFAULT_DRIVERbasenamerc)rUrd is_gcs_pathspec base_paths r%get_tensorstore_specrumsgy))11&'BB)$$W--+r * *$ G Orw}}Y77O M)MM N NN **I(P .P.PY.P.P  ++DOO G,Y77d9oo#2IFFd9o +r$tspecdict[str, Any] | strreturncxt|tr5tjddt d|rdSdSdD]}||vrt ||cSd|vrKtD]C}|d|dkr/|ddStj|d|d rdSDdS) zDetect if user is using cloud storages. This can detect common defines and unable to detect some corner cases such as using gcsfuse. z^(|)TF)rirhrrNrZ)rPrVr[matchjoin_REMOTE_URL_PREFIXESis_remote_storage_REMOTE_DRIVER_VALIDATIONS)rvkeyrules r%rrs s x7chh344777?? T U ++c e|| uSz * ***** xDN * *   % 8D&f 6 6  r$c$eZdZdZdZdZdZdS)_LimitInFlightByteszDLimits in-flight bytes when reading/writing checkpoints per process.cx||_||_tjtj|_dS)N)lock) _max_bytes_available_bytesr4 ConditionLock_cv)self num_bytess r%__init__z_LimitInFlightBytes.__init__s0DO%D glnn555DHHHr$c@Kjkrtddjj4d{Vjfdd{Vxjzc_jdksJ dddd{VdS#1d{VswxYwYdS)Nz1Requested more bytes than we reserved space for: z > cjkSrK)r)requested_bytesrsr%z4_LimitInFlightBytes.wait_for_bytes..sd&;o&Mr$r)rr^rwait_forrrrs``r%wait_for_bytesz"_LimitInFlightBytes.wait_for_bytessv(( @)@@.2o@@ A AAx(((((((( H  MMMMM N NNNNNNNN .  "a ' ' ' ' '((((((((((((((((((((((((((((((sAB  BBcK|j4d{V|xj|z c_|j|jksJ|jdddd{VdS#1d{VswxYwYdSrK)rrr notify_allrs r% release_bytesz!_LimitInFlightBytes.release_bytess x .  "do 5 5 5 5 hs|r.sBLL"#af LLLLLLr$r) memory_kindrF)copy)datarRdeviceaddressable_memoriesr enable_memoriesvaluejaxr SingleDeviceSharding device_putcopy_to_host_asyncr4sleeprCr )rrhas_pinned_hostr s r%transfer_shard_to_hostrs $LL',|'H'H'J'JLLLLL/ !o|00!1##H >$ ) )DD a $U # # ##r$ primary_host int | None replica_idint transactionOptional[ts.Transaction]c& Kttjr=tjdkr&jrt d|dddt|st|d<d|vr!tj j j |d<|tj |krbtjtj|d d || }-tt sJ|n|d{Vtjtj|d d || d{V  fd }j} tj|| } t+j| d{VS) a[Serialize an array using TensorStore. Args: arr_inp: The array to serialize. tensorstore_spec: The tensorstore spec to use. commit_future: A list of futures that will be appended to. The futures can be awaited asynchronously. If None, the futures will be awaited synchronously by this method. context: ts.Context instance. primary_host: Primary host, which indicates the host that will be treated as the "leader". If None, all hosts are treated as the primary. DO NOT USE unless you are sure you know what you are doing. replica_id: Allows overriding the shard replica id that will be saved. DO NOT USE unless you are sure you know what you are doing. transaction: TensorStore transaction to use for opening and writing the array. If not specified, a non-transactional write will be used. r>zPassing fully addressable arrays to a multiprocess serialization is not allowed, as this may lead to a race condition between processes. Serialization have failed for the array with the path "rhrZz".rIdtypeNT)createopencontextr)rassume_metadatarrcbK|jkrt|d{V}|j|t t j}@t tsJ|j |j d{VdS|j d{VdSdS)N)&can_reference_source_data_indefinitely) rrindexwriterPr ArrayImpllistappendcommitr)rr write_futurearr_inp commit_futurerts r% _write_arrayz%async_serialize.._write_arrays :%%)%00 0 0 0 0 0 0du{^))  2<u2 2 *  l  "-.....\0111!!!!!!!!!!%&%r$)rPr rr process_countis_fully_addressabler^rLrGjnprname process_indextsrSpecrraddressable_shards tree_utiltree_mapr4r5) rtensorstore_specrrrrr open_futurer local_shardsfuture_write_staters ` ` ` @r%async_serializers4%/**>s/@/B/BQ/F/F "0G  =&i08 = = = > >> , - -:#0#9#9Z  $$$ # '- 8 8 =WS.00L@@'  !!  K  t , ,,,,;''''  Gg     !""""""""*+,}--lLII~12 2 2 2 2 2 22r$cNfd}tj|dS)Nc|Ktjt}t j|d{VSrK)rrrrr4r5) future_writerarraystensorstore_specss r%_run_serializerz*run_serialization.._run_serializer4s@M**?FDUVVM/ / / / / / //r$r4run)rrrs`` r%run_serializationr3sB000000 +oo     r$rts.TensorStoredomaints.IndexDomainc|j}|jjj}|jj}||j}|j}|j}|j}|j}t|D]} || j s |j |zcSt|D]G} || } || } || } | | z } | || z| z }| | z| z}| | z | z}|||z z}H|SrK) rankr numpy_dtypeitemsize chunk_layoutread_chunk_templateroriginr@rangefinitesize)rrrrchunk_templaterr@ chunk_origin chunk_shapei origin_valuechunk_origin_value chunk_sizelowerupper lower_aligned upper_aligneds r%estimate_read_memory_footprintr:s $g!*)>5. ^ XF =& ,%&,$+ ;;%%a !  #% [9 $$$$% ;;11a!9L%aQJ - -E 58 #&8 8EZ'*4Mf *+j8M --/0II r$user_in_shardingjax.sharding.Sharding | Layoutrts.Spec | dict[str, Any]Sequence[int] | None byte_limiter_LimitInFlightBytes | Nonerc Kt|tr|jn|}t|tjjst d|t|tr|jnd tj|d||d{V | j n| | t  d   fd }tt ||d{VS) Nzqsharding passed to deserialization should be specified, concrete and an instance of `jax.sharding.Sharding`. Got T)rrrr array.Indexr jax.Devicec Ktj |j} j|}t |}|d{Vt j jj }tj |tj ddj |j | |d{V | }|jt jkrt!j|}t'j|t+t&j|}|d{V|S)N) input_shape)r)rIndexTransformr intersectrrrCzerosrrr r/ translate_torrastyperint4asarrayrrrr rr) rrrequested_domainrestricted_domainroutresultrdllrnew_shard_shaper@rs r%cbzasync_deserialize..cbus(U;;;EBI**+;<<4Q8IJJO  ' ' 8 88888888 (?!'*= > > >C (3--QQQ,-=-DE F  5#4!5667777777  JJu  c  yCH K  c ^ VC::6BB C CEEF  & & 7 77777777 Mr$)rrrr) rPrr rShardingr^device_local_layoutrrr@ shard_shapetupler:) rrr&rrrr in_shardingrrrr@rs `` @@@@r%async_deserializer[sj/88O!**>N K!6 7 7G  F8C F F G GG ' 0 0 ;  - -6: G %    ! ")!''|%++E%LL99/%%%%%%%%%%%N0e k2NN N N N N N NNr$ shardings$Sequence[sharding.Sharding | Layout]rSequence[dict[str, Any]] global_shapesSequence[array.Shape] | Nonedtypes!Sequence[typing.DTypeLike] | None concurrent_gbc`|dzfd}tj|S)Niʚ;c Kt}tjt t |dgt zndgt zn}tj|d{VS)N)r) rrrrrrlenr4r5)rr8concurrent_bytesrrrrs r%_run_deserializerz.run_deserialization.._run_deserializers&'788LM**! ===$+8+@&''''m+1>&''''v GGM / / / / / / //r$r)rrrrr r%r$s```` @r%run_deserializationr&s] #U* 0 0 0 0 0 0 0 0 0 &&(( ) ))r$rc d|S)Ntensorstore_checkpoint_r#)rs r%_get_keyr)s (3 ( ((r$ceZdZdZejdZejdZejddZej dddZ dS) GlobalAsyncCheckpointManagerBasea&Interface for checkpointing GDAs asynchronously. This class manages the state of an ongoing asynchronous checkpoint. For example, say a checkpoint happens on every step. If you checkpoint on step 1 and after some computation the model is on checkpoint 2. But step 1's checkpoint hasn't finished committing to the storage layer yet. So until that is finished, checkpoint for step 2 will need to be blocked. Maintaining a class allows to maintain that state. Examples: Below is a simplified training loop: ``` # Call this at the start of your program. jax.distributed.initialize() manager = GlobalAsyncCheckpointManager() # Restore checkpoint if available or initialize the train_state from # init_fn(). train_state = manager.deserialize(...) while ...: if step % num_steps_between_checkpoints == 0: manager.serialize(train_state, temp_checkpoint_dir=..., final_checkpoint_dir=...) train_state = train_step(train_state, input) # This is a non-blocking call. manager.check_for_errors() manager.serialize(train_state, temp_checkpoint_dir=..., final_checkpoint_dir=...) # Wait before the end of the program for the checkpoint to finish. This is a # blocking call. manager.wait_until_finished() ``` cdS)zChecks if any errors have been raised in the child thread. This is a non-blocking call that can be called in the main thread. Nr#rs r%check_for_errorsz1GlobalAsyncCheckpointManagerBase.check_for_errorsr$cdS)z(Blocks until serialization has finished.Nr#r-s r%wait_until_finishedz4GlobalAsyncCheckpointManagerBase.wait_until_finishedr/r$on_commit_callbackCallable[[], None]cdS)zSerializes GDAs to TensorStore.Nr#)rrrr2s r% serializez*GlobalAsyncCheckpointManagerBase.serializer/r$NrSequence[sharding.Sharding]rrrrrrcdS)z#Deserializes GDAs from TensorStore.Nr#)rrrrrs r% deserializez,GlobalAsyncCheckpointManagerBase.deserializer/r$)r2r3)NN)rr6rrrrrr) r r!r"rabcabstractmethodr.r1r5r8r#r$r%r+r+s&&P 333****AE>B.......r$r+c<eZdZd dZdZdZdZdZdZd d Z d S) AsyncManager,c<||_|jdz|_d|_d|_d|_t jdkr%tjj ttt jdkrtjj |_ d|_ dS)Nir>) _timeout_secs_timeout_in_ms_commit_futures_thread _exceptionrrr global_stateclientr^_DISTRIBUTED_SYSTEM_MSG_client_count)r timeout_secss r%rzAsyncManager.__init__s%D,t3DDDLDO Q;#;#B#J . / // Q -4dlDKKKr$c|j5|jrtddSdSdS)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.)rBis_aliveloggerwarningr-s r%__del__zAsyncManager.__del__sV |DL$9$9$;$; nnKLLLLL r$c tj}tj}td|t j}|jD]}|td||dkrkt|j }td|||j ||j td||dkro| td|dkr;|j |ttd|tjd t j|z dS#t$$r}||_Yd}~dSd}~wwxYw) Nz/Starting commit to storage layer by process: %sz3Finished committing to storage layer by process: %sr>z)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)rrrrLinfotimerArr)rHrGwait_at_barrierr@_on_commit_callback key_value_set_CHECKPOINT_SUCCESS monitoringrecord_event_duration_secs ExceptionrC)rcurrent_processrthread_start_timefuturekey_for_barrieres r% _thread_funczAsyncManager._thread_funcs")++o'))m kkC"$$$)++(&  kkG"$$$   #4;// ?#_ 6 6 6 $$_d6IJJJ @# % % % A     """ :;;; 1   , $ $_6I J J J ++I% ' ' ' n// ; )++) )+++++ dooooooosF'F++ G5GGctt|_||_t j|j|_|jdS)N)target) next_module_unique_countrHrS threadingThreadr^rBstart)rr2s r%_start_async_commitz AsyncManager._start_async_commit4sL+,,DK1D#4+<===DLLr$c|jq|j}d|_t|tjrGdt |vr6t dt |tg|dS)Nz$DEADLINE_EXCEEDED: Barrier timed out )rCrPxeXlaRuntimeErrorrVrr}_BARRIER_TIMED_OUT_MSG)r exceptions r%r.zAsyncManager.check_for_errors;s| "/ido Y 2 3 3A 0C NN B B% IIs9~~'=> ? ?AA A o#"r$c|j:|jd|_td|tdt jdkrX|jSt|j}|j ||j td|dSdSdS)NzThread joined successfullyz!Error check finished successfullyr>z._run_serializer..s"O*% - - - r$)rrrr4r5)rrryrrs r%rz?GlobalAsyncCheckpointManager.serialize.._run_serializers`m,,        m>=1 1 1 1 1 1 11r$N)rLrPr1r4rrsrf)rrrr2rrrys `` ` @r%r5z&GlobalAsyncCheckpointManager.serialize^s@ KK?@@@&(N 2 2 2 2 2 2 2 2 K!!"""n%%% /00000r$rSequence[jax.Array]paths Sequence[str]ctjt|}|||||dS)N)r2r)rrTmaprur5)rrr{r2rtspecss r%serialize_with_pathsz1GlobalAsyncCheckpointManager.serialize_with_pathssKX\\. 6 6FNN- r$rrrrrrrrrr rcP|t|||||SrK)r1r&)rrrrrr s r%r8z(GlobalAsyncCheckpointManager.deserializes7   y*;,fm E EEr$r6c~tjt|}||||||SrK)rrTr~rur8)rrr{rrr rs r%deserialize_with_pathsz3GlobalAsyncCheckpointManager.deserialize_with_pathss= X\\. 6 6F   Iv}f) + ++r$)rr)rrzr{r|rrNNr rrrrrrrrr r) rr6r{r|rrrrr r)r r!r"rr5rr8rr#r$r%ruru[s99/3 818181818181@/3 $AE>B') EEEEE5926 +++++++r$ru)r&r'r(r)r*r+)rUrV)F)rUrVrdre)rvrwrxre)rrrxr)rrrrrr)rrrrrxr) rrrrr&rrrrrerr)rr)Kr __future__rr9r4collections.abcrrr functoolsr itertoolsloggingrjr[rcrQrrr rjax._srcr r r r rjax._src.layoutrrDLLr jax._src.librri jax.numpynumpyrrC tensorstorerContext TS_CONTEXT_REMOVED_VALUErUcountrbrprFr~rrXrrk getLoggerr rLr:rGrLrcrurrrrrrrr&r) StrictABCr+r<rur#r$r%rsd/."""""" 9999999999   ######<<<<<<<<,,,,,,RZ.#? @ @  0&y((C )D))4(( iM  8 $ $ ' ' ' 'BBB NNNN0>0$$$$,  ,0a3a3a3a3a3H!!!H*. /3 !AOAOAOAOAOLGKDH-/ *****())))>.>.>.>.>.t~>.>.>.B`#`#`#`#`#`#`#`#F^+^+^+^+^+<1Q^+^+^+^+^+r$