src/ceph/src/pybind/mgr/dashboard/remote_view_cache.py

   1
   2
   3 from threading import Thread, Event, Lock
   4 import datetime
   5 import time
   6
   7
   8 class GetterThread(Thread):
   9     def __init__(self, view):
  10         super(GetterThread, self).__init__()
  11         self._view = view
  12         self.event = Event()
  13
  14     def run(self):
  15         try:
  16             t0 = time.time()
  17             val = self._view._get()
  18             t1 = time.time()
  19         except:
  20             self._view.log.exception("Error while calling _get:")
  21             # TODO: separate channel for passing back error
  22             self._view.value = None
  23             self._view.value_when = None
  24             self._view.getter_thread = None
  25         else:
  26             with self._view.lock:
  27                 self._view.latency = t1 - t0
  28                 self._view.value = val
  29                 self._view.value_when = datetime.datetime.now()
  30                 self._view.getter_thread = None
  31
  32         self.event.set()
  33
  34
  35 class RemoteViewCache(object):
  36     """
  37     A cached view of something we have to fetch remotely, e.g. the output
  38     of a 'tell' command to a daemon, or something we read from RADOS like
  39     and rbd image's status.
  40
  41     There are two reasons this wrapper exists:
  42      * To reduce load on remote entities when we frequently fetch something
  43        (i.e. 10 page loads in one second should not translate into 10 separate
  44         remote fetches)
  45      * To handle the blocking nature of some remote operations
  46        (i.e. if RADOS isn't responding, we need a predictable way to either
  47         return no data, or return stale data, when something from RADOS is
  48         requested)
  49
  50     Note that relying on timeouts in underlying libraries isn't wise here:
  51     if something is really slow, we would like to return a 'stale' state
  52     to a GUI quickly, but we should let our underlying request to the cluster
  53     run to completion so that we get some data even if it's slow, and so that
  54     a polling caller doesn't end up firing off a large number of requests to
  55     the cluster because each one timed out.
  56
  57     Subclasses may override _init and must override _get
  58     """
  59
  60     def __init__(self, module_inst):
  61         self.initialized = False
  62
  63         self.log = module_inst.log
  64
  65         self.getter_thread = None
  66
  67         # Consider data within 1s old to be sufficiently fresh
  68         self.stale_period = 1.0
  69
  70         # Return stale data if
  71         self.timeout = 5
  72
  73         self.event = Event()
  74         self.value_when = None
  75         self.value = None
  76         self.latency = 0
  77         self.lock = Lock()
  78
  79         self._module = module_inst
  80
  81     def init(self):
  82         self._init()
  83         self.initialized = True
  84
  85     VALUE_OK = 0
  86     VALUE_STALE = 1
  87     VALUE_NONE = 2
  88
  89     def get(self):
  90         """
  91         If data less than `stale_period` old is available, return it
  92         immediately.
  93         If an attempt to fetch data does not complete within `timeout`, then
  94         return the most recent data available, with a status to indicate that
  95         it is stale.
  96
  97         Initialization does not count towards the timeout, so the first call
  98         on one of these objects during the process lifetime may be slower
  99         than subsequent calls.
 100
 101         :return: 2-tuple of value status code, value
 102         """
 103         with self.lock:
 104             if not self.initialized:
 105                 self.init()
 106
 107             now = datetime.datetime.now()
 108             if self.value_when and now - self.value_when < datetime.timedelta(
 109                     seconds=self.stale_period):
 110                 return self.VALUE_OK, self.value
 111
 112             if self.getter_thread is None:
 113                 self.getter_thread = GetterThread(self)
 114                 self.getter_thread.start()
 115
 116             ev = self.getter_thread.event
 117
 118         success = ev.wait(timeout=self.timeout)
 119
 120         with self.lock:
 121             if success:
 122                 # We fetched the data within the timeout
 123                 return self.VALUE_OK, self.value
 124             elif self.value_when is not None:
 125                 # We have some data, but it doesn't meet freshness requirements
 126                 return self.VALUE_STALE, self.value
 127             else:
 128                 # We have no data, not even stale data
 129                 return self.VALUE_NONE, None
 130
 131     def _init(self):
 132         pass
 133
 134     def _get(self):
 135         raise NotImplementedError()