class_docstr: Helper class for managing experiment-level checkpoints.
  
  This class implements the ``checkpoint()`` method used to checkpoint
  experiment state. When called, this will serialize and write to disk
  the state of the trial runner, trial executor, and search algorithm, to
  a specified checkpoint file.
  
  The checkpoint period is automatically adjusted to
  ``max(10, time_per_checkpoint * 19)``. This means that at most 5% of the
  time (1/20) will be used for writing checkpoints, while 95% of the time
  (19/20) will be used to handle the rest of the training loop.
  
  If ``sync_every_n_trial_checkpoints`` is not None, syncing
  to cloud will be forced if any trial has checkpointed more times than
  ``sync_every_n_trial_checkpoints`` since last sync.
class_docstr_tok: ["Helper","class","for","managing","experiment-level","checkpoints",".","This","class","implements","the","``","checkpoint","(",")","``","method","used","to","checkpoint","experiment","state",".","When","called",",","this","will","serialize","and","write","to","disk","the","state","of","the","trial","runner",",","trial","executor",",","and","search","algorithm",",","to","a","specified","checkpoint","file",".","The","checkpoint","period","is","automatically","adjusted","to","``","max","(","10",",","time_per_checkpoint","*","19",")","``",".","This","means","that","at","most","5","%","of","the","time","(","1\/20",")","will","be","used","for","writing","checkpoints",",","while","95","%","of","the","time","(","19\/20",")","will","be","used","to","handle","the","rest","of","the","training","loop",".","If","``","sync_every_n_trial_checkpoints","``","is","not","None",",","syncing","to","cloud","will","be","forced","if","any","trial","has","checkpointed","more","times","than","``","sync_every_n_trial_checkpoints","``","since","last","sync","."]
docstr: Saves execution state to the local experiment directory.
  Overwrites the current session checkpoint, which starts when self
  is instantiated. Throttle depends on self._checkpoint_period.
  
  Also, automatically saves the search algorithm to the local
  checkpoint dir.
  
  Args:
      save_fn: Function to call to actually save data. Should expect
          one string argument specifying the directory to save to.
      force: Forces a checkpoint despite checkpoint_period.
      wait: Wait until sync to cloud has finished.
docstr_tok: ["Saves","execution","state","to","the","local","experiment","directory",".","Overwrites","the","current","session","checkpoint",",","which","starts","when","self","is","instantiated",".","Throttle","depends","on","self._checkpoint_period",".","Also",",","automatically","saves","the","search","algorithm","to","the","local","checkpoint","dir",".","Args",":","save_fn",":","Function","to","call","to","actually","save","data",".","Should","expect","one","string","argument","specifying","the","directory","to","save","to",".","force",":","Forces","a","checkpoint","despite","checkpoint_period",".","wait",":","Wait","until","sync","to","cloud","has","finished","."]
code: def checkpoint(
      self,
      save_fn: Callable[[], None],
      force: bool = False,
      wait: bool = False,
  ):
      """Saves execution state to the local experiment directory.
      Overwrites the current session checkpoint, which starts when self
      is instantiated. Throttle depends on self._checkpoint_period.
  
      Also, automatically saves the search algorithm to the local
      checkpoint dir.
  
      Args:
          save_fn: Function to call to actually save data. Should expect
              one string argument specifying the directory to save to.
          force: Forces a checkpoint despite checkpoint_period.
          wait: Wait until sync to cloud has finished.
  
      """
      experiment_local_path = self._storage.experiment_local_path
      if not experiment_local_path:
          return
  
      force = force or self._should_force_cloud_sync
  
      now = time.time()
      if (
          now - self._last_save_time < self._checkpoint_period
          and not force
      ):
          return
  
      # Checkpoint
      checkpoint_time_start = time.monotonic()
  
      # NOTE: This context manager is for Datasets captured in a trial config.
      # This is the case when *tuning over datasets*.
      # If the datasets have already been full executed, then serializing
      # block refs means that this checkpoint is not usable in a new Ray cluster.
      # This context will serialize the dataset execution plan instead, if available.
      with out_of_band_serialize_dataset():
          save_fn()
  
      # Sync to cloud
      self.sync_up(force=force, wait=wait)
  
      checkpoint_time_taken = time.monotonic() - checkpoint_time_start
  
      # Adjust dynamic checkpointing
      self._update_auto_checkpoint_time(
          time_taken=checkpoint_time_taken
      )
  
      # Finish
      self._last_save_time = time.time()
      return experiment_local_path
  
imports: from collections import Counter
  from dataclasses import dataclass
  from pathlib import Path
  from typing import Callable, Dict, Optional, Tuple, Union
  import click
  import logging
  import os
  import time
  import warnings
  from ray.train._internal.storage import StorageContext, get_fs_and_path, _download_from_fs_path, _list_at_fs_path
  from ray.tune.experiment import Trial
  from ray.tune.impl.out_of_band_serialize_dataset import out_of_band_serialize_dataset
label_desc: Use image node_id 5 for calling the _ExperimentCheckpointManager obj's underlying member method code with example usage: obj.checkpoint(save_fn, force, wait) and returns: experiment_local_path