code.grnet.gr Git - ganeti-local/blob - lib/daemon.py

   1 #
   2 #
   3
   4 # Copyright (C) 2006, 2007, 2008 Google Inc.
   5 #
   6 # This program is free software; you can redistribute it and/or modify
   7 # it under the terms of the GNU General Public License as published by
   8 # the Free Software Foundation; either version 2 of the License, or
   9 # (at your option) any later version.
  10 #
  11 # This program is distributed in the hope that it will be useful, but
  12 # WITHOUT ANY WARRANTY; without even the implied warranty of
  13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14 # General Public License for more details.
  15 #
  16 # You should have received a copy of the GNU General Public License
  17 # along with this program; if not, write to the Free Software
  18 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
  19 # 02110-1301, USA.
  20
  21
  22 """Module with helper classes and functions for daemons"""
  23
  24
  25 import select
  26 import signal
  27 import errno
  28 import time
  29
  30 from ganeti import utils
  31
  32
  33 class Timer(object):
  34   def __init__(self, owner, timer_id, start, interval, repeat):
  35     self.owner = owner
  36     self.timer_id = timer_id
  37     self.start = start
  38     self.interval = interval
  39     self.repeat = repeat
  40
  41
  42 class Mainloop(object):
  43   """Generic mainloop for daemons
  44
  45   """
  46   def __init__(self):
  47     """Constructs a new Mainloop instance.
  48
  49     """
  50     self._io_wait = {}
  51     self._io_wait_add = []
  52     self._io_wait_remove = []
  53     self._signal_wait = []
  54     self._timer_id_last = 0
  55     self._timer = {}
  56     self._timer_add = []
  57     self._timer_remove = []
  58
  59   def Run(self, handle_sigchld=True, handle_sigterm=True, stop_on_empty=False):
  60     """Runs the mainloop.
  61
  62     @type handle_sigchld: bool
  63     @param handle_sigchld: Whether to install handler for SIGCHLD
  64     @type handle_sigterm: bool
  65     @param handle_sigterm: Whether to install handler for SIGTERM
  66     @type stop_on_empty: bool
  67     @param stop_on_empty: Whether to stop mainloop once all I/O waiters
  68                           unregistered
  69
  70     """
  71     poller = select.poll()
  72
  73     # Setup signal handlers
  74     if handle_sigchld:
  75       sigchld_handler = utils.SignalHandler([signal.SIGCHLD])
  76     else:
  77       sigchld_handler = None
  78     try:
  79       if handle_sigterm:
  80         sigterm_handler = utils.SignalHandler([signal.SIGTERM])
  81       else:
  82         sigterm_handler = None
  83
  84       try:
  85         running = True
  86         timeout = None
  87         timeout_needs_update = True
  88
  89         # Start actual main loop
  90         while running:
  91           # Entries could be added again afterwards, hence removing first
  92           if self._io_wait_remove:
  93             for fd in self._io_wait_remove:
  94               try:
  95                 poller.unregister(fd)
  96               except KeyError:
  97                 pass
  98               try:
  99                 del self._io_wait[fd]
 100               except KeyError:
 101                 pass
 102             self._io_wait_remove = []
 103
 104           # Add new entries
 105           if self._io_wait_add:
 106             for (owner, fd, conditions) in self._io_wait_add:
 107               self._io_wait[fd] = owner
 108               poller.register(fd, conditions)
 109             self._io_wait_add = []
 110
 111           # Add new timers
 112           if self._timer_add:
 113             timeout_needs_update = True
 114             for timer in self._timer_add:
 115               self._timer[timer.timer_id] = timer
 116             del self._timer_add[:]
 117
 118           # Remove timers
 119           if self._timer_remove:
 120             timeout_needs_update = True
 121             for timer_id in self._timer_remove:
 122               try:
 123                 del self._timer[timer_id]
 124               except KeyError:
 125                 pass
 126             del self._timer_remove[:]
 127
 128           # Stop if nothing is listening anymore
 129           if stop_on_empty and not (self._io_wait or self._timer):
 130             break
 131
 132           # Calculate timeout again if required
 133           if timeout_needs_update:
 134             timeout = self._CalcTimeout(time.time())
 135
 136           # Wait for I/O events
 137           try:
 138             io_events = poller.poll(timeout)
 139           except select.error, err:
 140             # EINTR can happen when signals are sent
 141             if err.args and err.args[0] in (errno.EINTR,):
 142               io_events = None
 143             else:
 144               raise
 145
 146           after_poll = time.time()
 147
 148           if io_events:
 149             # Check for I/O events
 150             for (evfd, evcond) in io_events:
 151               owner = self._io_wait.get(evfd, None)
 152               if owner:
 153                 owner.OnIO(evfd, evcond)
 154
 155           if self._timer:
 156             self._CheckTimers(after_poll)
 157
 158           # Check whether signal was raised
 159           if sigchld_handler and sigchld_handler.called:
 160             self._CallSignalWaiters(signal.SIGCHLD)
 161             sigchld_handler.Clear()
 162
 163           if sigterm_handler and sigterm_handler.called:
 164             self._CallSignalWaiters(signal.SIGTERM)
 165             running = False
 166             sigterm_handler.Clear()
 167       finally:
 168         # Restore signal handlers
 169         if sigterm_handler:
 170           sigterm_handler.Reset()
 171     finally:
 172       if sigchld_handler:
 173         sigchld_handler.Reset()
 174
 175   def _CalcTimeout(self, now):
 176     if not self._timer:
 177       return None
 178
 179     timeout = None
 180
 181     # TODO: Repeating timers
 182
 183     min_timeout = 0.001
 184
 185     for timer in self._timer.itervalues():
 186       time_left = (timer.start + timer.interval) - now
 187       if timeout is None or time_left < timeout:
 188         timeout = time_left
 189       if timeout < 0:
 190         timeout = 0
 191         break
 192       elif timeout < min_timeout:
 193         timeout = min_timeout
 194         break
 195
 196     return timeout * 1000.0
 197
 198   def _CheckTimers(self, now):
 199     # TODO: Repeating timers
 200     for timer in self._timer.itervalues():
 201       if now < (timer.start + timer.interval):
 202         continue
 203
 204       timer.owner.OnTimer(timer.timer_id)
 205
 206       # TODO: Repeating timers should not be removed
 207       self._timer_remove.append(timer.timer_id)
 208
 209   def _CallSignalWaiters(self, signum):
 210     """Calls all signal waiters for a certain signal.
 211
 212     @type signum: int
 213     @param signum: Signal number
 214
 215     """
 216     for owner in self._signal_wait:
 217       owner.OnSignal(signal.SIGCHLD)
 218
 219   def RegisterIO(self, owner, fd, condition):
 220     """Registers a receiver for I/O notifications
 221
 222     The receiver must support a "OnIO(self, fd, conditions)" function.
 223
 224     @type owner: instance
 225     @param owner: Receiver
 226     @type fd: int
 227     @param fd: File descriptor
 228     @type condition: int
 229     @param condition: ORed field of conditions to be notified
 230                       (see select module)
 231
 232     """
 233     # select.Poller also supports file() like objects, but we don't.
 234     assert isinstance(fd, (int, long)), \
 235       "Only integers are supported for file descriptors"
 236
 237     self._io_wait_add.append((owner, fd, condition))
 238
 239   def UnregisterIO(self, fd):
 240     """Unregister a file descriptor.
 241
 242     It'll be unregistered the next time the mainloop checks for it.
 243
 244     @type fd: int
 245     @param fd: File descriptor
 246
 247     """
 248     # select.Poller also supports file() like objects, but we don't.
 249     assert isinstance(fd, (int, long)), \
 250       "Only integers are supported for file descriptors"
 251
 252     self._io_wait_remove.append(fd)
 253
 254   def RegisterSignal(self, owner):
 255     """Registers a receiver for signal notifications
 256
 257     The receiver must support a "OnSignal(self, signum)" function.
 258
 259     @type owner: instance
 260     @param owner: Receiver
 261
 262     """
 263     self._signal_wait.append(owner)
 264
 265   def AddTimer(self, owner, interval, repeat):
 266     """Add a new timer.
 267
 268     The receiver must support a "OnTimer(self, timer_id)" function.
 269
 270     @type owner: instance
 271     @param owner: Receiver
 272     @type interval: int or float
 273     @param interval: Timer interval in seconds
 274     @type repeat: bool
 275     @param repeat: Whether this is a repeating timer or one-off
 276
 277     """
 278     # TODO: Implement repeating timers
 279     assert not repeat, "Repeating timers are not yet supported"
 280
 281     # Get new ID
 282     self._timer_id_last += 1
 283
 284     timer_id = self._timer_id_last
 285
 286     self._timer_add.append(Timer(owner, timer_id, time.time(),
 287                                  float(interval), repeat))
 288
 289     return timer_id
 290
 291   def RemoveTimer(self, timer_id):
 292     """Removes a timer.
 293
 294     @type timer_id: int
 295     @param timer_id: Timer ID
 296
 297     """
 298     self._timer_remove.append(timer_id)