BVhWfdZddlZddlZddlZddlZddlZddlZddlmZddl m Z ddl m Z ddl m Z ddlmZddlmZdd lmZdd lmZdd lmZdd lmZdd lmZddlmZdaej:ZdZdZ dZ!dZ"GddZ#y)z>Utilities for saving/loading Trackable objects asynchronously.N)logging)checkpoint_context)trackable_view) device_util)context) def_function)executor)ops) variables)metrics)base)object_identityasync_checkpoint!_create_copy_for_async_checkpointc0||kryt||z dzS)aCalculate the duration between start and end time. Args: start_time_seconds: The start time in seconds. end_time_seconds: The end time in seconds. Returns: The duration between the start and the end time. Return 0 if end_time_seconds < start_time_seconds. ri@B)round)start_time_secondsend_time_secondss d/home/dcms/DCMS/lib/python3.12/site-packages/tensorflow/python/checkpoint/async_checkpoint_helper.py_get_duration_microsecondsr3s'**  #55@ AAc:tj|j}d}|t|krE|rC|||vr&|j |||j |n|dz }|t|kr|rCd}|Dcgc] }||r|}}||fScc}w)aReturn the list of checkpointable trackables dependent on `root`. Args: root: The root trackable from where we get all its dependent trackables. exclude_set: An ObjectIdentitySet of Trackables to exclude before returning. Each element in `exclude_set` is a specific instance of a `Trackable` and appears precisely once in `TrackableView(root).descendants()`. Returns: saveable_trackables: All trackables that are saveable in `all_trackables` (see definition of "saveable" in `_trackable_needs_to_be_saved()`). A subset of `all_trackables`. all_trackables: All trackables returned by `TrackableView`'s `descendants()` after excluding `exclude_set`. A superset of `saveable_trackables`. )rootrc,t|dr+d|jvsd|jvsd|jvryt|jD]A}|tj urd|jvsd|jvsd|jvsAyy)zReturns whether a trackable needs to be saved. Returns a bool to indicate whether obj's class has `_serialize_to_tensors`, `gather_saveables_for_checkpoint`, or `_copy_trackable_to_cpu` defined. Args: obj: A Trackable object. __dict___serialize_to_tensors _gather_saveables_for_checkpoint_copy_trackable_to_cpuTF)hasattrrtypemror Trackable)objts r_trackable_needs_to_be_savedz9_get_all_trackables.._trackable_needs_to_be_savedlssJ !S\\ 1 /3<< ? % 5#Y]]_ dnn  #qzz 11QZZ?'1::5 r)r TrackableView descendantslendiscardpop)r exclude_setall_trackablestrackable_indexr&xsaveable_trackabless r_get_all_trackablesr1Ds "//T:FFH. /#n--+o&+5.9:)o #n--+$:%3:q5a8:: n ,,:sBceZdZdZddZej dZdZdZ dZ dZ d Z d Z ed Zdd Zdd ZddZddZddZdZy)AsyncCheckpointHelperz"Helper class for async checkpoint.Nc |r8t|tjr|n|}||d<|j| t d||_||_d|_|jd|_ d|_ d|_ d|_ d|_ d|_tj xsd|_tj$|j"|_d|_d|_d|_t-j.d|_t3j4|j6d|_t:5t<t?j>adddy#1swYyxYw)aInitialize AsyncCheckpoint. Args: checkpointer_impl: The Checkpoint class to power the AsyncCheckpoint. root: The root object to checkpoint. `root` may be a trackable object or `WeakRef` of a trackable object. **kwargs: The keyword arguments representing the checkpointed variables. Raises: AttributeError: when checkpointer_impl is None. rNz;checkpointer_impl cannot be None for AsyncCheckpointHelper.FzCPU:0r)maxsize) isinstanceweakrefref_maybe_initialize_trackableAttributeError_checkpointer_impl_checkpoint_items _checkpoint checkpointer_checkpoint_options _initialized_original_nodes _object_map_tpu_embedding_objects_saveable_trackablesrcurrent_default_device canonicalize_save_file_prefix_use_checkpoint_save_async_save_threadqueueQueue_queueatexitregister_join_async_save_thread _async_error"_END_TIME_OF_LAST_ASYNC_WRITE_LOCK_END_TIME_OF_LAST_ASYNC_WRITEtime)selfcheckpointer_implrkwargstrackable_roots r__init__zAsyncCheckpointHelper.__init__sI !+D'++!>tvDn%fVn002  G 0D#DD#DD DD"&D!%D&..0;GD&33D4H4HID!D %D"D++a(DK OOD001D ,4 & .(, %444s 4EE!c|jD]} |j|j!|j D]}|jy#t$r!}t j d||Yd}~id}~wwxYw)zCopy the checkpointed variables from the accelerator to the host CPU. TODO(chienchunh): Get the concrete function before firstly called to avoid hangining the accelerators idle during function tracing.  object_mapTrackable %s skipped due to: %sN)rDrrBNotImplementedErrorrwarningrC_retrieve_variables)rUr%e tpu_embeddings r _copy_to_cpuz"AsyncCheckpointHelper._copy_to_cpus & &AA   D,<,< =A 44* '')*!A91a@@AsA A;A66A;ct|j!|jdi|j|_|jS)z3Gets or creates the underlying Checkpoint instance.)r=r;r<rUs rr>z"AsyncCheckpointHelper.checkpointers8 000J43I3IJd   rcDtj|_g|_tj}|j |j |j |j jt|j |\|_ }|D]}tt|tr|j|dt|vs;|j}|D]m}|D]f}t!|t"j$s |j'||}t!|t,j.sL|jj1|ho|j jj3}t5j6d||j|j j8_|jD]} |j;|j!|jD]} | jAtCjD|jFd|_$|jHjKd|_&y#t(t*f$rY}wxYw#t<$r!} t5j>d|| Yd} ~ d} ~ wwxYw) z/Initialize the async checkpoint internal state.)rr,get_slot_namesz0Initializing async checkpoint's save_counter: %dr[r]NT)targetdaemon)'rObjectIdentityDictionaryrBrCObjectIdentitySetaddr> save_counterr1rDr r!_TPU_EMBEDDING_ATTR_handle_tpu_embeddingdirrhr6r Variableget_slotr:KeyErrorr r#appendnumpyrinfo_saverrr^r_r` threadingThread _async_saverJstartr@) rUr,r-r% slot_names slot_nameoriginal_variableoriginal_slot_variablernrarbs r_ensure_initializedz)AsyncCheckpointHelper._ensure_initializedsc'??AD"$D"335KOOD%%'(OOD%%'4450C    k1;-D~G a- . ""1% SV #%%' # GI#1 G/1C1CD '(zz2CY'O$0$..A''../EF G GG2$$&3399;L LLC.2-=-=D* & &AA   D,<,< =A 44* '')*(...D!!#DC#H-  ,!A91a@@As*II5I2 1I2 5 J>JJc|jr4|j}d|_tjdt||y)zMExpose the most recent error from the async saving thread to the caller. NzJPropagating the most recent error from the async thread before joining: %s)rQrerrorstr)rUras r_check_async_thread_errorz/AsyncCheckpointHelper._check_async_thread_error>sB    ad mm)*-a&2 g rc\ |jjddtjd|j|jj |jy#t j$rtjdY;wxYw#|jwxYw)aJoin the async save thread. The steps for terminating the async save thread: 1). Put will succeed when the last async save event is done. Putting a false triggers the async save thread's while loop to end. We use put instead of sync because sync does not have a timeout argument. 2). Join the async save thread. (The thread may finish before joining.) Fi,)timeoutzJoining the async save thread.NzrTimeout waiting for the async save thread; terminating the thread instead. The last checkpoint may be incomeplete.) rMputrrwrJjoinrKFullrrrfs rrPz-AsyncCheckpointHelper._join_async_save_threadHs ' kkooeSo) ll34 , $$&  $$& ::P mmOPP $$&s$AA++(BBBBB+c tjtjdd5|jj rt jd|jtj} tj|j5tj5|jr5|jj!|j"|j$n5|jj'|j"|j$dddddd|jj-tj}t/j0t2t5||t65t/j8t2t5t:||addd|jj rdddt jdy#1swYxYw#1swYxYw#t($r}||_Yd}~d}~wwxYw#|jj-wxYw#1swYxYw#1swYxYw)z2The thread function for the async checkpoint save.F) enable_asyncenable_streaming_enqueuez0Starting async checkpoint save on the device: %s)optionsN api_label microsecondsz3Async save thread reached the end of the execution.)rexecutor_scoper new_executorrMgetrrwrFrTr devicerasync_metrics_contextrIr>saverHr?_write ExceptionrQ task_doner AddAsyncCheckpointWriteDuration_ASYNC_CHECKPOINTrrRAddTrainingTimeSavedrS)rUasync_save_start_timeraasync_save_end_times rr{z!AsyncCheckpointHelper._async_save\s    @ A3@ KKOO  G)) +!%  "zz$../ #99; **!!#((**D,D,D!!#**** 44+   ++   !"iik//'34I4GI J0 @  & &)5/1FH I+@ '  @S KKOO  3@h LLFGE     $    ++   ! @ @]3@3@sAI<HH 0A7G>'H /H7AI,I<"I>H H  H H H0H+&H3+H00H33III II'ctt|trt|jst dt|z|j |j |jr|jdnd|j}||j|<||jvr|jj|yy)aHandle TPUEmbedding. This is the only place where we populate object map in the class of `AsyncCheckpointHelper`. For all other checkpointable trackables, we populate object map using the trackable's own `_copy_trackable_to_cpu()`. Args: tpu_embedding: TPUEmbedding object to be handled. Raises: AttributeError: if the input trackable is not TPUEmbedding type. z#Expecting TPUEmbedding type; got %srN)feature_config optimizer#pipeline_execution_with_tensor_core) r r!rocallablerr:_feature_config _table_config$_pipeline_execution_with_tensor_corerBrCru)rUrb new_embeddings rrpz+AsyncCheckpointHelper._handle_tpu_embeddings 4 &(; rnrfs rrnz"AsyncCheckpointHelper.save_counters     + ++rc&|j||S)Save the checkpointed variables. Args: save_path: The file prefix of the checkpoint file. options: Optional CheckpointOption instance. Returns: The full path of the checkpoint file. )rrU save_pathrs rwritezAsyncCheckpointHelper.writes ;;y' **rc>tj}|js|jn*|jj |j |j tj||_ d|_ |rtj|nd|_ |jrd|j_ |jjdtj}tj t"t%|||S)aSave the checkpointed variables. This method has exactly the same logic as save(), except it does not increment the underlying save_counter, which is done by the caller, e.g., CheckpointManager. Args: save_path: The file prefix of the checkpoint file. options: Optional CheckpointOption instance. Returns: The full path of the checkpoint file. FNTr)rTr@rrMrrcrr async_waitrHrIcopyr?$experimental_enable_async_checkpointrr AddCheckpointWriteDurationrr)rUrrwrite_start_timewrite_end_times rrzAsyncCheckpointHelper._writesyy{      kk   ""$ &D %D6=tyy1$D FKdCKKOODYY[N &&#/0@0>@A rctj}|js|jn*|jj |j |j |jjjdz}dj||}tj||_ d|_|rtj|nd|_|j rd|j _|jj%dtj}t'j(t*t-|||S)rrz{}-{}TNFr)rTr@rrMrrcrr>rnrvformatrrrHrIrr?rrr rrr)rUrrsave_start_timern full_path save_end_times rrzAsyncCheckpointHelper.savesiikO      kk   ""$ $$&3399;a?Ly,7I &D $D6=tyy1$D FKdCKKOODIIKM &&#/OQ rc&|j||S)aRestore the checkpointed variables. This method has exactly the same logic as restore(). This method is implemented only to fulfill the duty of subclassing tf.train.Checkpoint. Args: save_path: The full name of the checkpoint file to be restored. options: CheckpointOption instance. Returns: A load status object, which can be used to make assertions about the status of a checkpoint restoration. See tf.train.Checkpoint.restore() for more details. )restorers rreadzAsyncCheckpointHelper.readFs << 7 ++rc|rtj|n |j|_|jrd|j_|jj |j j ||j}|S)a`Restore the checkpointed variables. Args: save_path: The full name of the checkpoint file to be restored. options: CheckpointOption instance. Returns: A load status object, which can be used to make assertions about the status of a checkpoint restoration. See tf.train.Checkpoint.restore() for more details. F)rr?rrMrr>r)rUrrstatuss rrzAsyncCheckpointHelper.restoreWsq& '4+C+C  FKdC KK    ( (D4L4L MF Mrcb|jjtjdy)z+Sync on any ongoing save or restore events.zSync on ongoing save/restore.N)rMrrrwrfs rsynczAsyncCheckpointHelper.syncqsKK LL01r)N)__name__ __module__ __qualname____doc__rYrfunctionrcr>rrrPr{rppropertyrnrrrrrrrerrr3r3s}*G4R * * KZ'(6Hp#8J  ,  , +2h>@,"42rr3)$rrNrrKryrTr7abslrtensorflow.python.checkpointrrtensorflow.python.distributertensorflow.python.eagerrrr tensorflow.python.frameworkr tensorflow.python.opsr 0tensorflow.python.saved_model.pywrap_saved_modelr tensorflow.python.trackabler tensorflow.python.utilrrSLockrRrrorr1r3rerrrsE  ;74+0,++D,2!%%3Y^^%5"':B"H-Ve2e2r