Update gitignore rules

[ganeti-local] / lib / daemon.py

diff --git a/lib/daemon.py b/lib/daemon.py
index 860e364..6fb2f66 100644 (file)
--- a/lib/daemon.py
+++ b/lib/daemon.py
@@ -24,12 +24,13 @@
 
 import asyncore
 import os
-import select
 import signal
 import errno
 import logging
 import sched
 import time
+import socket
+import sys
 
 from ganeti import utils
 from ganeti import constants
@@ -71,28 +72,107 @@ class AsyncoreScheduler(sched.scheduler):
     sched.scheduler.__init__(self, timefunc, AsyncoreDelayFunction)
 
 
+class AsyncUDPSocket(asyncore.dispatcher):
+  """An improved asyncore udp socket.
+
+  """
+  def __init__(self):
+    """Constructor for AsyncUDPSocket
+
+    """
+    asyncore.dispatcher.__init__(self)
+    self._out_queue = []
+    self.create_socket(socket.AF_INET, socket.SOCK_DGRAM)
+
+  # this method is overriding an asyncore.dispatcher method
+  def handle_connect(self):
+    # Python thinks that the first udp message from a source qualifies as a
+    # "connect" and further ones are part of the same connection. We beg to
+    # differ and treat all messages equally.
+    pass
+
+  # this method is overriding an asyncore.dispatcher method
+  def handle_read(self):
+    try:
+      try:
+        payload, address = self.recvfrom(constants.MAX_UDP_DATA_SIZE)
+      except socket.error, err:
+        if err.errno == errno.EINTR:
+          # we got a signal while trying to read. no need to do anything,
+          # handle_read will be called again if there is data on the socket.
+          return
+        else:
+          raise
+      ip, port = address
+      self.handle_datagram(payload, ip, port)
+    except:
+      # we need to catch any exception here, log it, but proceed, because even
+      # if we failed handling a single request, we still want to continue.
+      logging.error("Unexpected exception", exc_info=True)
+
+  def handle_datagram(self, payload, ip, port):
+    """Handle an already read udp datagram
+
+    """
+    raise NotImplementedError
+
+  # this method is overriding an asyncore.dispatcher method
+  def writable(self):
+    # We should check whether we can write to the socket only if we have
+    # something scheduled to be written
+    return bool(self._out_queue)
+
+  def handle_write(self):
+    try:
+      if not self._out_queue:
+        logging.error("handle_write called with empty output queue")
+        return
+      (ip, port, payload) = self._out_queue[0]
+      try:
+        self.sendto(payload, 0, (ip, port))
+      except socket.error, err:
+        if err.errno == errno.EINTR:
+          # we got a signal while trying to write. no need to do anything,
+          # handle_write will be called again because we haven't emptied the
+          # _out_queue, and we'll try again
+          return
+        else:
+          raise
+      self._out_queue.pop(0)
+    except:
+      # we need to catch any exception here, log it, but proceed, because even
+      # if we failed sending a single datagram we still want to continue.
+      logging.error("Unexpected exception", exc_info=True)
+
+  def enqueue_send(self, ip, port, payload):
+    """Enqueue a datagram to be sent when possible
+
+    """
+    if len(payload) > constants.MAX_UDP_DATA_SIZE:
+      raise errors.UdpDataSizeError('Packet too big: %s > %s' % (len(payload),
+                                    constants.MAX_UDP_DATA_SIZE))
+    self._out_queue.append((ip, port, payload))
+
+
 class Mainloop(object):
   """Generic mainloop for daemons
 
+  @ivar scheduler: A sched.scheduler object, which can be used to register
+    timed events
+
   """
   def __init__(self):
     """Constructs a new Mainloop instance.
 
-    @ivar scheduler: A L{sched.scheduler} object, which can be used to register
-    timed events
-
     """
     self._signal_wait = []
     self.scheduler = AsyncoreScheduler(time.time)
 
   @utils.SignalHandled([signal.SIGCHLD])
   @utils.SignalHandled([signal.SIGTERM])
-  def Run(self, stop_on_empty=False, signal_handlers=None):
+  def Run(self, signal_handlers=None):
     """Runs the mainloop.
 
-    @type stop_on_empty: bool
-    @param stop_on_empty: Whether to stop mainloop once all I/O waiters
-                          unregistered
     @type signal_handlers: dict
     @param signal_handlers: signal->L{utils.SignalHandler} passed by decorator
 
@@ -103,10 +183,6 @@ class Mainloop(object):
     running = True
     # Start actual main loop
     while running:
-      # Stop if nothing is listening anymore
-      if stop_on_empty and not (self._io_wait):
-        break
-
       if not self.scheduler.empty():
         try:
           self.scheduler.run()
@@ -131,7 +207,7 @@ class Mainloop(object):
 
     """
     for owner in self._signal_wait:
-      owner.OnSignal(signal.SIGCHLD)
+      owner.OnSignal(signum)
 
   def RegisterSignal(self, owner):
     """Registers a receiver for signal notifications
@@ -150,11 +226,9 @@ def GenericMain(daemon_name, optionparser, dirs, check_fn, exec_fn):
 
   @type daemon_name: string
   @param daemon_name: daemon name
-  @type optionparser: L{optparse.OptionParser}
+  @type optionparser: optparse.OptionParser
   @param optionparser: initialized optionparser with daemon-specific options
                        (common -f -d options will be handled by this module)
-  @type options: object @param options: OptionParser result, should contain at
-                 least the fork and the debug options
   @type dirs: list of strings
   @param dirs: list of directories that must exist for this daemon to work
   @type check_fn: function which accepts (options, args)
@@ -223,8 +297,7 @@ def GenericMain(daemon_name, optionparser, dirs, check_fn, exec_fn):
                        debug=options.debug,
                        stderr_logging=not options.fork,
                        multithreaded=multithread)
-    logging.info("%s daemon startup" % daemon_name)
+    logging.info("%s daemon startup", daemon_name)
     exec_fn(options, args)
   finally:
     utils.RemovePidFile(daemon_name)
-