o i 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_identityasync_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)start_time_secondsend_time_secondsrc/var/www/myenv/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) isinstanceweakrefref_maybe_initialize_trackableAttributeError_checkpointer_impl_checkpoint_items _checkpoint checkpointer_checkpoint_options _initialized_async_write_done_callback_original_nodes _object_map_tpu_embedding_objectsrcurrent_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_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)r:accelerator_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(r?rrr@rAr r rBrCr)_retrieve_variables)r:rDrE 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_trackabler@rA_handle_tpu_embedding_trackable_childrenvaluesadd)r: to_traversevisitedcurrent_trackablechildrrr_traverse_variabless"        z)AsyncCheckpointHelper._traverse_variablescCs$|jdur|jdi|j|_|jS)z3Gets or creates the underlying Checkpoint instance.Nr)r"r r!r:rrrr#s 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'rObjectIdentityDictionaryr(r) collectionsdequeObjectIdentitySetr!rQrr rrNr@rArOrMrRrWdirrYget_slotrKeyErrorr# save_counternumpyrinfo_saver threadingThread _async_saver/start) r:rSrTvrU slot_names slot_nameoriginal_variableoriginal_slot_variablercrrr_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)r6rerrorstr)r:errr_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.) r2putrrer/joinr0FullrqrtrXrrrr5s     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) enable_asyncenable_streaming_enqueuez0Starting async checkpoint save on the device: %s)optionswrite_done_callbackN api_label microsecondsz3Async save thread reached the end of the execution.)rexecutor_scoper new_executorr2getrrer+r9r r rasync_metrics_contextr.r#saver-r$_writer& Exceptionr6 task_doner AddAsyncCheckpointWriteDuration_ASYNC_CHECKPOINTrr7AddTrainingTimeSavedr8)r:async_save_start_timersasync_save_end_timerrrri-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. CPUr) device_type device_index) trainableshapedtypenameN) pydev DeviceSpec from_stringr replace to_stringr r rrr _shared_namer()r: original_var op_devicenew_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. )rN) _variablesrrMr(rr)r:r copied_varsrkrrr_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)rrrr rr)r:original_trackablerrrrNs  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)feature_config optimizer#pipeline_execution_with_tensor_core) r@rAcallablerrtype_feature_config _table_config$_pipeline_execution_with_tensor_corer(r)rM)r:rH new_embeddingrrrrOs&     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. )r#rcrXrrrrcs 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)rr: save_pathr{rrrwrites 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. FNTr})rpr9r2rwrIrtr async_waitr-r.copyr$$experimental_enable_async_checkpointr&rvr AddCheckpointWriteDurationrr)r:rr{r|write_start_timewrite_end_timerrrrs*  zAsyncCheckpointHelper._writecCs|t}|j|||jd}d ||}t ||_ d|_ |r5t|nd|_|jr?d|j_|jdt}tjtt||d|S)rrz{}-{}TNFr})rpr9r2rwrIrtr#rcrdformatrrr-r.rr$rrvr rrr)r:rr{save_start_timerc full_path save_end_timerrrrs(   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. )restorerrrrread@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$rr2rwr#rr%rF)r:rr{statusrrrrQs zAsyncCheckpointHelper.restorecCs|jtddS)z+Sync on any ongoing save or restore events.zSync on ongoing save/restore.N)r2rwrrerXrrrsyncqs zAsyncCheckpointHelper.sync)N)NN)__name__ __module__ __qualname____doc__r>rfunctionrFrIrWr#rprtr5rirrrNrOpropertyrcrrrrrrrrrrrFs2 K  6 9 !  6 =  r)'rr3r]rr0rgr9rabslrtensorflow.python.checkpointrtensorflow.python.distributer-tensorflow.python.distribute.sharded_variablertensorflow.python.eagerrrrtensorflow.python.frameworkr rr +tensorflow.python.ops.resource_variable_opsr tensorflow.python.ops.variablesr 0tensorflow.python.saved_model.pywrap_saved_modelr tensorflow.python.utilrr8Lockr7rrArrrrrrs6