class_docstr: Syncer class for synchronizing data between Ray nodes and remote (cloud) storage.
  
  This class handles data transfer for two cases:
  
  1. Synchronizing data such as experiment checkpoints from the driver to
     cloud storage.
  2. Synchronizing data such as trial checkpoints from remote trainables to
     cloud storage.
  
  Synchronizing tasks are usually asynchronous and can be awaited using ``wait()``.
  The base class implements a ``wait_or_retry()`` API that will retry a failed
  sync command.
  
  The base class also exposes an API to only kick off syncs every ``sync_period``
  seconds.
  
  Args:
      sync_period: The minimum time in seconds between sync operations, as
          used by ``sync_up/down_if_needed``.
      sync_timeout: The maximum time to wait for a sync process to finish before
          issuing a new sync operation. Ex: should be used by ``wait`` if launching
          asynchronous sync tasks.
class_docstr_tok: ["Syncer","class","for","synchronizing","data","between","Ray","nodes","and","remote","(","cloud",")","storage",".","This","class","handles","data","transfer","for","two","cases",":","1",".","Synchronizing","data","such","as","experiment","checkpoints","from","the","driver","to","cloud","storage",".","2",".","Synchronizing","data","such","as","trial","checkpoints","from","remote","trainables","to","cloud","storage",".","Synchronizing","tasks","are","usually","asynchronous","and","can","be","awaited","using","``","wait","(",")","``",".","The","base","class","implements","a","``","wait_or_retry","(",")","``","API","that","will","retry","a","failed","sync","command",".","The","base","class","also","exposes","an","API","to","only","kick","off","syncs","every","``","sync_period","``","seconds",".","Args",":","sync_period",":","The","minimum","time","in","seconds","between","sync","operations",",","as","used","by","``","sync_up\/down_if_needed","``",".","sync_timeout",":","The","maximum","time","to","wait","for","a","sync","process","to","finish","before","issuing","a","new","sync","operation",".","Ex",":","should","be","used","by","``","wait","``","if","launching","asynchronous","sync","tasks","."]
code: def wait_or_retry(self, max_retries: int = 2, backoff_s: int = 5):
      assert max_retries > 0
      last_error_traceback = None
      for i in range(max_retries + 1):
          try:
              self.wait()
          except Exception as e:
              attempts_remaining = max_retries - i
  
              # If we're out of retries, then save the full traceback of the last
              # error and show it when raising an exception.
              if attempts_remaining == 0:
                  last_error_traceback = traceback.format_exc()
                  break
  
              logger.error(
                  f"The latest sync operation failed with the following error: "
                  f"{repr(e)}\n"
                  f"Retrying {attempts_remaining} more time(s) after sleeping "
                  f"for {backoff_s} seconds..."
              )
              time.sleep(backoff_s)
              self.retry()
              continue
          # Succeeded!
          return
      raise RuntimeError(
          f"Failed sync even after {max_retries} retries. "
          f"The latest sync failed with the following error:\n{last_error_traceback}"
      )
  
imports: import abc
  import logging
  import threading
  import time
  import traceback
  import warnings
  from dataclasses import dataclass
  from typing import Any, Callable, Dict, List, Optional, Tuple, Union
  from ray._private.thirdparty.tabulate.tabulate import tabulate
  from ray.train.constants import _DEPRECATED_VALUE
  from ray.util import log_once
  from ray.util.annotations import PublicAPI
  from ray.widgets import Template
label_desc: Use image node_id 9 for calling the Syncer obj's underlying member method code with example usage: obj.wait_or_retry(max_retries, backoff_s) without return types