class_docstr: Helper class for async checkpoint.
class_docstr_tok: ["Helper","class","for","async","checkpoint","."]
docstr: The thread function for the async checkpoint save.
docstr_tok: ["The","thread","function","for","the","async","checkpoint","save","."]
code: def _async_save(self):
      """The thread function for the async checkpoint save."""
      with context.executor_scope(
          executor.new_executor(
              enable_async=False, enable_streaming_enqueue=False
          )
      ):
          # The main thread inserts: a True to the queue when the user calls save,
          # triggering async save; and a False when we exit the Checkpoint instance.
          while self._queue.get():
              logging.info(
                  "Starting async checkpoint save on the device: %s",
                  self._default_device,
              )
  
              async_save_start_time = time.time()
  
              # Specify the ops placement on the worker if running with
              # coordinator-worker mode. This is required as launching a new thread
              # would clear the placement policy and make localhost the default
              # placement, while the main thread's default placement would be the
              # master worker's CPU:0.
              try:
                  with ops.device(self._default_device):
                      with checkpoint_context.async_metrics_context():
                          if self._use_checkpoint_save:
                              self.checkpointer().save(
                                  self._save_file_prefix,
                                  self._checkpoint_options,
                              )
                          else:
                              self.checkpointer()._write(  # pylint: disable=protected-access
                                  self._save_file_prefix,
                                  options=self._checkpoint_options,
                              )
              except Exception as e:  # # pylint: disable=broad-except
                  self._async_error = e
              finally:
                  self._queue.task_done()
  
              async_save_end_time = time.time()
              metrics.AddAsyncCheckpointWriteDuration(
                  api_label=_ASYNC_CHECKPOINT,
                  microseconds=_get_duration_microseconds(
                      async_save_start_time, async_save_end_time
                  ),
              )
  
              # Measure the elapsed time since the last checkpoint.
              # Due to the nature of async checkpoint, here it actually captures the
              # duration between the start_time of the previous checkpoint and the
              # start time of this checkpoint. As a result, the duration of the final
              # async checkpoint is excluded, which is fine since it does not take
              # much time.
              global _END_TIME_OF_LAST_ASYNC_WRITE
              with _END_TIME_OF_LAST_ASYNC_WRITE_LOCK:
                  metrics.AddTrainingTimeSaved(
                      api_label=_ASYNC_CHECKPOINT,
                      microseconds=_get_duration_microseconds(
                          _END_TIME_OF_LAST_ASYNC_WRITE,
                          async_save_start_time,
                      ),
                  )
                  _END_TIME_OF_LAST_ASYNC_WRITE = async_save_start_time
      logging.info(
          "Async save thread reached the end of the execution."
      )
  
imports: import atexit
  import copy
  import queue
  import threading
  import time
  import weakref
  from absl import logging
  from tensorflow.python.checkpoint import checkpoint_context
  from tensorflow.python.checkpoint import trackable_view
  from tensorflow.python.distribute import device_util
  from tensorflow.python.eager import context
  from tensorflow.python.eager import def_function
  from tensorflow.python.eager import executor
  from tensorflow.python.framework import ops
  from tensorflow.python.ops import variables
  from tensorflow.python.saved_model.pywrap_saved_model import metrics
  from tensorflow.python.trackable import base
  from tensorflow.python.util import object_identity
label_desc: Use image node_id 7 for calling the AsyncCheckpointHelper obj's underlying member method code with example usage: obj._async_save() without return types