Source Code for Module Products.ZenHub.notify

  1  ############################################################################## 
  2  #  
  3  # Copyright (C) Zenoss, Inc. 2011, all rights reserved. 
  4  #  
  5  # This content is made available according to terms specified in 
  6  # License.zenoss under the directory where your Zenoss product is installed. 
  7  #  
  8  ############################################################################## 
  9   
 10   
 11  __doc__ = """Provides a batch notifier to break up the expensive, blocking IO 
 12  involved with calls to DeviceOrganizer.getSubDevices which can call getObject 
 13  on the brains of every device in the system. Processes batches of 10 devices 
 14  giving way to the event loop between each batch. See ticket #26626.""" 
 15   
 16  import logging 
 17  import collections 
 18  from twisted.internet import reactor, defer 
 19   
 20  LOG = logging.getLogger("zen.hub.notify") 
 21   
 22 -class NotifyItem(object): 
 23      """These items are held in the BatchNotifier's queue. They contain all the 
 24      context needed to process the subdevices of a specific device class. This 
 25      context includes... 
 26       
 27          device class UID: e.g. /zport/dmd/Devices/Server/Linux) 
 28          subdevices: an iterator over the device classes subdevices) 
 29          notify_functions: a dictionary mapping Service UID to notifyAll 
 30                            function. An example Service UID is  
 31                                       ('CommandPerformanceConfig', 'localhost') 
 32          d: the deferred for this item. Always has the following callback  
 33             chain: 
 34               Slot          Callback                    Errback 
 35                 1    BatchNotifier._callback             None 
 36                 2             None                 BatchNotifier._errback 
 37      """ 
 38   
 39 -    def __init__(self, device_class_uid, subdevices): 
 40          self.device_class_uid = device_class_uid 
 41          self.subdevices = subdevices 
 42          # keys are service_uids eg ('CommandPerformanceConfig', 'localhost') 
 43          self.notify_functions = {} 
 44          self.d = None 
 45   
 46 -    def __repr__(self): 
 47          args = (self.device_class_uid, self.notify_functions.keys()) 
 48          return "<NotifyItem(device_class_uid=%s, notify_functions=%s)>" % args 
 49   
 50 -class BatchNotifier(object): 
 51      """Processes the expensive getSubDevices call in batches. A singleton 
 52      instance is registered as a utility in zcml. The queue contains NotifyItem 
 53      instances. If notify_subdevices is called and an item exists in the queue 
 54      for the same device class, then the new service UID and notify function 
 55      are appended to the existing item. Once an item is moved from the queue to 
 56      _current_item member, it is being processed and further notify_subdevices 
 57      calls for the same device class will append a new item to the queue. 
 58      """ 
 59   
 60      _BATCH_SIZE = 10 
 61      _DELAY = 0.05 
 62   
 63 -    def __init__(self): 
 64          self._empty = None 
 65          self._current_item = None 
 66          self._queue = collections.deque() 
 67          self._stopping = False 
 68   
 69 -    def notify_subdevices(self, device_class, service_uid, notify_function): 
 70          if not self._stopping: 
 71              LOG.debug("BatchNotifier.notify_subdevices: %r, %s" % (device_class, service_uid)) 
 72              item = self._find_or_create_item(device_class) 
 73              item.notify_functions[service_uid] = notify_function 
 74          else: 
 75              LOG.debug("notify_subdevices received a call while " 
 76                        "stopping: %r, %s" % (device_class, service_uid)) 
 77   
 78 -    def _find_or_create_item(self, device_class): 
 79          device_class_uid = device_class.getPrimaryId() 
 80          for item in self._queue: 
 81              if item.device_class_uid == device_class_uid: 
 82                  retval = item 
 83                  break 
 84          else: 
 85              subdevices = device_class.getSubDevicesGen() 
 86              retval = NotifyItem(device_class_uid, subdevices) 
 87              retval.d = self._create_deferred() 
 88              if not self._queue and self._current_item is None: 
 89                  self._call_later(retval.d) 
 90              self._queue.appendleft(retval) 
 91          return retval 
 92   
 93 -    def _create_deferred(self): 
 94          d = defer.Deferred() 
 95          d.addCallback(self._callback) 
 96          d.addErrback(self._errback) 
 97          return d 
 98   
 99 -    def _call_later(self, d): 
100          reactor.callLater(BATCH_NOTIFIER._DELAY, d.callback, None) 
101   
102 -    def _switch_to_next_item(self): 
103          self._current_item = self._queue.pop() if self._queue else None 
104   
105 -    def _call_notify_functions(self, device): 
106          for service_uid, notify_function in self._current_item.notify_functions.items(): 
107              try: 
108                  notify_function(device) 
109              except Exception, e: 
110                  args = (service_uid, device.getPrimaryId(), type(e).__name__, e) 
111                  LOG.error("%s failed to notify %s: %s: %s" % args) 
112   
113 -    def _callback(self, result): 
114          if self._current_item is None: 
115              self._current_item = self._queue.pop() 
116          batch_count = 0 
117          try: 
118              for device in self._current_item.subdevices: 
119                  self._call_notify_functions(device) 
120                  batch_count += 1 
121                  if batch_count == BatchNotifier._BATCH_SIZE: 
122                      self._current_item.d = self._create_deferred() 
123                      break 
124              else: 
125                  LOG.debug("BatchNotifier._callback: no more devices, %s in queue", len(self._queue)) 
126                  self._switch_to_next_item() 
127          except Exception, e: 
128              args = (self._current_item.device_class_uid, type(e).__name__, e) 
129              LOG.warn("Failed to get subdevice of %s: %s: %s" % args) 
130              self._switch_to_next_item() 
131          if self._current_item is not None: 
132              self._call_later(self._current_item.d) 
133          elif self._empty and not self._queue: 
134              empty, self._empty = (self._empty, None) 
135              empty.callback(None) 
136   
137   
138 -    def _errback(self, failure): 
139          LOG.error("Failure in batch notifier: %s: %s" % (failure.type.__name__, failure.value)) 
140          LOG.debug("BatchNotifier._errback: failure=%s" % failure) 
141   
142 -    def stop(self): 
143          self._stopping = True 
144          return defer.DeferredList([item.d for item in self._queue]) 
145   
146 -    def whenEmpty(self): 
147          isEmptyDeferred = self._empty 
148          def printEmpty(val): 
149              LOG.debug("Notifier is now empty") 
150          if not self._queue and not self._current_item: 
151              isEmptyDeferred = defer.Deferred() 
152              LOG.debug("Notifier is currently empty") 
153              isEmptyDeferred.callback(None) 
154          elif not self._empty: 
155              LOG.debug("Notifier is not currently empty, returning deferred") 
156              isEmptyDeferred = self._empty = defer.Deferred() 
157              isEmptyDeferred.addCallback(printEmpty) 
158          return isEmptyDeferred 
159   
160  BATCH_NOTIFIER = BatchNotifier() 
161