class_docstr: This DataStore sends policies over redis pubsub and get/set.
  
  Deployment
  ==========
  It assumes that a redis server is already available. We make this assumption because during
  multinode training at this time, redis is already used for communicating replay memories.
  
  Communication
  =============
  
  A redis pubsub channel is used by the training worker to signal to the rollout workers that a
  new policy is ready. When this occurs, a new policy is loaded from the redis key/value store
  where key is the same as the pubsub channel. Originally, just the pubsub was used, but that
  could result in a race condition where the master worker publishes the first policy and waits
  for the rollout workers to submit all rollouts, while a delayed rollout worker waits for the
  first policy since it subscribed to the channel after the initial policy was published.
class_docstr_tok: ["This","DataStore","sends","policies","over","redis","pubsub","and","get\/set",".","Deployment","==========","It","assumes","that","a","redis","server","is","already","available",".","We","make","this","assumption","because","during","multinode","training","at","this","time",",","redis","is","already","used","for","communicating","replay","memories",".","Communication","=============","A","redis","pubsub","channel","is","used","by","the","training","worker","to","signal","to","the","rollout","workers","that","a","new","policy","is","ready",".","When","this","occurs",",","a","new","policy","is","loaded","from","the","redis","key\/value","store","where","key","is","the","same","as","the","pubsub","channel",".","Originally",",","just","the","pubsub","was","used",",","but","that","could","result","in","a","race","condition","where","the","master","worker","publishes","the","first","policy","and","waits","for","the","rollout","workers","to","submit","all","rollouts",",","while","a","delayed","rollout","worker","waits","for","the","first","policy","since","it","subscribed","to","the","channel","after","the","initial","policy","was","published","."]
docstr: :param graph_manager: the graph_manager to load the policy into
  :param require_new_policy: if True, only load a policy if it hasn't been loaded in this
  process yet before.
  :param timeout: Will only try to load the policy once if timeout is None, otherwise will
  retry for timeout seconds
docstr_tok: [":","param","graph_manager",":","the","graph_manager","to","load","the","policy","into",":","param","require_new_policy",":","if","True",",","only","load","a","policy","if","it","has","n't","been","loaded","in","this","process","yet","before",".",":","param","timeout",":","Will","only","try","to","load","the","policy","once","if","timeout","is","None",",","otherwise","will","retry","for","timeout","seconds"]
code: def load_policy(
      self, graph_manager, require_new_policy=True, timeout=0
  ):
      """
      :param graph_manager: the graph_manager to load the policy into
      :param require_new_policy: if True, only load a policy if it hasn't been loaded in this
      process yet before.
      :param timeout: Will only try to load the policy once if timeout is None, otherwise will
      retry for timeout seconds
      """
      if self.saver is None:
          # the GlobalVariableSaver needs to be instantiated after the graph is created. For now,
          # it can be instantiated here, but it might be nicer to have a more explicit
          # on_graph_creation_end callback or similar to put it in
          self.saver = GlobalVariableSaver()
          self._connect()
  
      if not require_new_policy:
          # try just loading whatever policy is available most recently
          if self._load_policy(graph_manager):
              return
  
      message = "first"
      timeout_ends = time.time() + timeout
      while time.time() < timeout_ends or message == "first":
          message = self.pubsub.get_message()
  
          if message and message["type"] == "message":
              if message["data"] == b"end_of_policies":
                  self._end_of_policies = True
                  return
              elif message["data"] == b"new_policy":
                  if self._load_policy(graph_manager):
                      return
                  else:
                      raise ValueError(
                          "'new_policy' message was sent, but no policy was found."
                      )
  
          time.sleep(1.0)
  
      if require_new_policy:
          raise ValueError(
              "Waited for {timeout} seconds on channel {channel}, but no first policy was received.".format(
                  timeout=timeout, channel=self.params.redis_channel
              )
          )
  
file_path: coach/rl_coach/data_stores/redis_data_store.py
imports: import time
  import uuid
  import redis
  from rl_coach.architectures.tensorflow_components.savers import GlobalVariableSaver
  from rl_coach.data_stores.data_store import DataStore, DataStoreParameters
label_desc: Use image node_id 9 for calling the RedisDataStore obj's underlying member method code with example usage: obj.load_policy(graph_manager, require_new_policy, timeout) without return types