o ?eb@sdZddlZddlZddlZddlZddlZddlZddlZddlm Z ddl m Z ddl m Z ddlmZddlmZddlmZdd lmZdd lmZdd lmZdd lmZdd lmZddlmZddlmZda e!Z"dZ#dZ$ddZ%GdddZ&dS)z>Utilities for saving/loading Trackable objects asynchronously.N)logging)checkpoint_context) device_util)ShardedVariable)context) def_function)executor)device)ops)UninitializedVariable)Variable)metrics)object_identityZasync_checkpoint!_create_copy_for_async_checkpointcCs||krdSt||dS)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)Zstart_time_secondsZend_time_secondsru/home/www/facesmatcher.com/pyenv/lib/python3.10/site-packages/tensorflow/python/checkpoint/async_checkpoint_helper.py_get_duration_microseconds5s rc@seZdZdZd+ddZejddZejddZd d Z d d Z d dZ ddZ ddZ ddZddZddZddZddZeddZd+dd Zd,d!d"Zd+d#d$Zd+d%d&Zd+d'd(Zd)d*ZdS)-AsyncCheckpointHelperz"Helper class for async checkpoint.NcKs|rt|tjr |n|}||d<||durtd||_||_d|_|d|_ d|_ d|_ d|_ d|_ d|_tpAd|_t|j|_d|_d|_d|_tjdd|_t|jd|_ttdurvtaWddSWddS1swYdS)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. rootNz;checkpointer_impl cannot be None for AsyncCheckpointHelper.FzCPU:0)maxsize) isinstanceweakrefrefZ_maybe_initialize_trackableAttributeError_checkpointer_impl_checkpoint_items _checkpoint checkpointer_checkpoint_options _initialized_async_write_done_callback_original_nodes _object_map_tpu_embedding_objectsrcurrent_default_deviceZ 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)selfZcheckpointer_implrkwargsZtrackable_rootrrr__init__Is@  "zAsyncCheckpointHelper.__init__c Csh|jD],\}}t|tst|trqt|j|| Wdn1s,wYqdS)zCopy the checkpointed variables from the host CPU to the accelerator. TODO(chienchunh): Get the concrete function before firstly called to avoid hangining the accelerators idle during function tracing. N) r$itemsrrhasattr_TPU_EMBEDDING_ATTRr r assign read_value)r5accelerator_varcpu_varrrr_copy_from_cpus z$AsyncCheckpointHelper._copy_from_cpuc Cs||jD],\}}t|tst|trqt|j|| Wdn1s,wYq|j D]}| q5dS)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. N) r$r8rrr9r:r r r;r<r%Z_retrieve_variables)r5r=r> tpu_embeddingrrr _copy_to_cpus   z"AsyncCheckpointHelper._copy_to_cpucCs|r@|}|j|t|ttfr||t|tr"| ||j dd D]}||vr1q*| |||q*|sdSdS)aCreate the copied nodes and variables while traversing the nodes. This method performs a BFS to traverse the nodes while avoiding duplicated visits. Throughout the process, self._mapping, self._original_nodes, and self._var_pairs are populated. Args: to_traverse: A deque that stores the nodes to be traversed. visited: A list of nodes that have been visited. checkpoint) save_typeN) popleftr#appendrr r_copy_trackabler9r:_handle_tpu_embeddingZ_trackable_childrenvaluesadd)r5 to_traversevisitedcurrent_trackablechildrrr_traverse_variabless"        z)AsyncCheckpointHelper._traverse_variablescCs$|jdur|jdi|j|_|jS)z3Gets or creates the underlying Checkpoint instance.Nr)rrrr5rrrrs z"AsyncCheckpointHelper.checkpointerc Cs`|jrdSg|_t|_g|_tg}t}|j D]#}t |t t fr-||n t|tr7||||||q||||jD]=}dt|vr|}|D].}|jD](}t |t sfq^z|||}Wn ttfyyYq^wt |t t fr||q^qYqK|j} td| |j|j_t j!|j"dd|_#|j#$d|_dS)z/Initialize the async checkpoint internal state.Nget_slot_namesz0Initializing async checkpoint's save_counter: %dT)targetdaemon)%r!r#rZObjectIdentityDictionaryr$r% collectionsdequeZObjectIdentitySetrrHrr rrFr9r:rGrErIrNdirrPZget_slotrKeyErrorr save_counternumpyrinfoZ_saver threadingThread _async_saver*start) r5rJrKvrLZ slot_namesZ slot_nameZoriginal_variableZoriginal_slot_variablerWrrr_ensure_initializedsX                z)AsyncCheckpointHelper._ensure_initializedcCs*|jr|j}d|_tdt||dS)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)r1rerrorstr)r5errr_check_async_thread_errorsz/AsyncCheckpointHelper._check_async_thread_errorcCszz7z|jjdddtd|jdur|jWntjy)tdYnwW| dSW| dS| w)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.) r-putrrYr*joinr+Fullr`rcrOrrrr0s     z-AsyncCheckpointHelper._join_async_save_threadc Csttjddd|jrtd|jt }zdzGt |j7t #|jr7||j|jn |j|j|j|jdWdn1sNwYWdn1s]wYWntyx}z ||_WYd}~nd}~wwW|jn|jwt }tjtt||dttjttt|d|aWdn1swY|jsWdn1swYtddS)z2The thread function for the async checkpoint save.F)Z enable_asyncZenable_streaming_enqueuez0Starting async checkpoint save on the device: %s)optionswrite_done_callbackNZ api_label microsecondsz3Async save thread reached the end of the execution.)rZexecutor_scoperZ new_executorr-getrrYr'r4r r rZasync_metrics_contextr)rsaver(r _writer" Exceptionr1 task_doner ZAddAsyncCheckpointWriteDuration_ASYNC_CHECKPOINTrr2ZAddTrainingTimeSavedr3)r5Zasync_save_start_timerbZasync_save_end_timerrrr\-sj     5z!AsyncCheckpointHelper._async_savecCsltj|jjddd}t|t|j|j |j |j d}Wdn1s*wY||j |<dS)zvCreate a new instance for the input trackable. Args: original_var: Input Variable object to be copied. ZCPUr)Z device_typeZ device_index) trainableshapedtypenameN) pydevZ DeviceSpecZ from_stringr replaceZ to_stringr r rrrsrtZ _shared_namer$)r5 original_varZ op_deviceZnew_varrrr_copy_for_variablefs z(AsyncCheckpointHelper._copy_for_variablecCsBg}|jD]}||||j|qt||jd|j|<dS)zCreate a new instance for the input ShardedVariable. Args: original_var: Input ShardedVariable object to be copied. )ruN)Z _variablesryrEr$rru)r5rxZ copied_varsr^rrr_copy_for_sharded_variablevs  z0AsyncCheckpointHelper._copy_for_sharded_variablecCs8t|tr ||dSt|tr||dStd)zCreate a new instance for the input trackable. Args: original_trackable: The trackable instance to be copied. Raises: AttributeError: if the input trackable is not Variable or ShardedVariable. z/Only Variable or ShardedVariable can be copied.N)rrrzr ryr)r5Zoriginal_trackablerrrrFs  z%AsyncCheckpointHelper._copy_trackablecCspt|tr t|jstdt||j|j|jr|jdnd|jd}||j |<||j vr6|j |dSdS)zHandle TPUEmbedding. Args: tpu_embedding: TPUEmbedding object to be handled. Raises: AttributeError: if the input trackable is not TPUEmbedding type. z#Expecting TPUEmbedding type; got %srN)Zfeature_configZ optimizerZ#pipeline_execution_with_tensor_core) r9r:callablerrtypeZ_feature_configZ _table_configZ$_pipeline_execution_with_tensor_corer$r%rE)r5r@Z new_embeddingrrrrGs&     z+AsyncCheckpointHelper._handle_tpu_embeddingcCs |jS)aAn integer variable numbering the checkpoint events. This is maintained by the underlying tf.train.Checkpoing object employed by AsyncCheckpoint class. The number starts at 0 and gets incremented for each checkpoint event. Returns: The save counter variable. )rrWrOrrrrWs z"AsyncCheckpointHelper.save_countercCs|||dS)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. N)rnr5 save_pathrhrrrwrites zAsyncCheckpointHelper.writecCs|t}|j||t||_d|_ |r&t |nd|_ |j r0d|j _ ||_ |jdt}tjtt||d|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. write_done_callback: Optional callback function executed after the async write is done. Returns: The full path of the checkpoint file. FNTrj)r_r4r-rfrArcr async_waitr(r)copyr $experimental_enable_async_checkpointr"rer AddCheckpointWriteDurationrqr)r5rrhriZwrite_start_timeZwrite_end_timerrrrns*  zAsyncCheckpointHelper._writecCs|t}|j|||jd}d ||}t ||_ d|_ |r5t|nd|_|jr?d|j_|jdt}tjtt||d|S)r}rz{}-{}TNFrj)r_r4r-rfrArcrrWrXformatrrr(r)rr rrer rrqr)r5rrhZsave_start_timerW full_pathZ save_end_timerrrrms(   zAsyncCheckpointHelper.savecCs |||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. )restorer~rrrread@s zAsyncCheckpointHelper.readcCsR|rt|n|j|_|jrd|j_|j|||j}|jr'||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 rr-rfrrr!r?)r5rrhstatusrrrrQs zAsyncCheckpointHelper.restorecCs|jtddS)z+Sync on any ongoing save or restore events.zSync on ongoing save/restore.N)r-rfrrYrOrrrsyncqs zAsyncCheckpointHelper.sync)N)NN)__name__ __module__ __qualname____doc__r7rfunctionr?rArNrr_rcr0r\ryrzrFrGpropertyrWrrnrmrrrrrrrrFs2 K  6 9 !  6 =  r)'rr.rSrr+rZr4rZabslrZtensorflow.python.checkpointrZtensorflow.python.distributerZ-tensorflow.python.distribute.sharded_variablerZtensorflow.python.eagerrrrZtensorflow.python.frameworkr rvr Z+tensorflow.python.ops.resource_variable_opsr Ztensorflow.python.ops.variablesr Z0tensorflow.python.saved_model.pywrap_saved_modelr Ztensorflow.python.utilrr3Lockr2rqr:rrrrrrs6