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","."]
code: def sync_up(self, force: bool = False, wait: bool = False) -> bool:
      syncer = self._storage.syncer
  
      if not syncer:
          return False
  
      # Always exclude checkpoints in the new persistence path.
      # TODO(justinvyu, krfricke): Ideally, this excludes all trial directories.
      # But for now, this is needed to upload driver artifacts that live in the
      # trial directory.
      exclude = _DRIVER_SYNC_EXCLUDE_PATTERNS
      experiment_local_path = self._storage.experiment_local_path
      experiment_fs_path = self._storage.experiment_fs_path
  
      if force:
          # Wait until previous sync command finished
          try:
              syncer.wait()
          except TimeoutError as e:
              logger.warning(
                  "The previous sync of the experiment directory to the cloud "
                  f"timed out with the error: {str(e)}\nSyncing will be retried. "
                  + _EXPERIMENT_SYNC_TIMEOUT_MESSAGE
              )
          except Exception as e:
              logger.warning(
                  "The previous sync of the experiment directory to the cloud "
                  f"failed with the error: {str(e)}\nSyncing will be retried."
              )
          synced = syncer.sync_up(
              local_dir=experiment_local_path,
              remote_dir=experiment_fs_path,
              exclude=exclude,
          )
      else:
          synced = syncer.sync_up_if_needed(
              local_dir=experiment_local_path,
              remote_dir=experiment_fs_path,
              exclude=exclude,
          )
  
      start_time = time.monotonic()
      if wait:
          try:
              syncer.wait()
          except Exception as e:
              raise RuntimeError(
                  "Uploading the experiment directory from the driver "
                  f"(local path: {experiment_local_path}) to the the cloud "
                  f"(remote path: {experiment_fs_path}) failed. "
                  "Please check the error message above."
              ) from e
  
      now = time.monotonic()
      sync_time_taken = now - start_time
  
      if sync_time_taken > self._slow_sync_threshold:
          try:
              import fsspec
          except Exception:
              fsspec = None
  
          fsspec_msg = ""
          if fsspec is None:
              fsspec_msg = (
                  "If your data is small, try installing fsspec "
                  "(`pip install fsspec`) for more efficient local file parsing. "
              )
  
          logger.warning(
              "Syncing the experiment checkpoint to cloud took a long time with "
              f"{sync_time_taken:.2f} seconds. This can be due to a large number "
              f"of trials, large logfiles, or throttling from the "
              f"remote storage provider for too frequent syncs. {fsspec_msg}"
              f"If your `CheckpointConfig.num_to_keep` is a low number, this can "
              f"trigger frequent syncing, in which case you should increase it. "
          )
  
      if not synced:
          return False
  
      self._should_force_cloud_sync = False
      self._trial_num_checkpoints_since_last_sync.clear()
  
      if now - self._last_sync_time < self._excessive_sync_threshold:
          logger.warning(
              "Experiment checkpoint syncing has been triggered multiple "
              f"times in the last {self._excessive_sync_threshold} seconds. "
              "A sync will be triggered whenever a trial has checkpointed "
              "more than `num_to_keep` times since last sync or if "
              f"{syncer.sync_period} seconds have passed since last "
              "sync. If you have set `num_to_keep` in your `CheckpointConfig`, "
              "consider increasing the checkpoint frequency or keeping more "
              "checkpoints. You can supress this warning by changing the "
              "`TUNE_WARN_EXCESSIVE_EXPERIMENT_CHECKPOINT_SYNC_THRESHOLD_S` "
              "environment variable."
          )
      self._last_sync_time = now
      return True
  
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 6 for calling the _ExperimentCheckpointManager obj's underlying member method code with example usage: obj.sync_up(force, wait) and returns: True, False, False