/ - Diff - ganeti-local - Greek Research and Technology Network's projects

Revision 5f3269fc

     from ganeti.confd.server import ConfdProcessor
     class ConfdAsyncUDPServer(asyncore.dispatcher):
     class ConfdAsyncUDPServer(daemon.AsyncUDPSocket):
       """The confd udp server, suitable for use with asyncore.
       """
-...
         @param reader: ConfigReader to use to access the config
         """
         asyncore.dispatcher.__init__(self)
         daemon.AsyncUDPSocket.__init__(self)
         self.bind_address = bind_address
         self.port = port
         self.processor = processor
         self.out_queue = []
         self.create_socket(socket.AF_INET, socket.SOCK_DGRAM)
         self.bind((bind_address, port))
         logging.debug("listening on ('%s':%d)" % (bind_address, port))
       # 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_in, address = self.recvfrom(4096)
           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
           payload_out =  self.processor.ExecQuery(payload_in, ip, port)
           if payload_out is not None:
             self.out_queue.append((ip, port, payload_out))
         except:
           # we need to catch any exception here, log it, but proceed, because even
           # if we failed handling a single request, we still want the confd to
           # continue working.
           logging.error("Unexpected exception", exc_info=True)
       # this method is overriding an asyncore.dispatcher method
       def writable(self):
         # Only handle writes if we have something enqueued to write
         if self.out_queue:
           return True
         else:
           return False
       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 handling a single request, we still want the confd to
           # continue working.
           logging.error("Unexpected exception", exc_info=True)
       # this method is overriding a daemon.AsyncUDPSocket method
       def handle_datagram(self, payload_in, ip, port):
         payload_out =  self.processor.ExecQuery(payload_in, ip, port)
         if payload_out is not None:
           self.enqueue_send(ip, port, payload_out)
     class ConfdInotifyEventHandler(pyinotify.ProcessEvent):

     import logging
     import sched
     import time
     import socket
     from ganeti import utils
     from ganeti import constants
-...
         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(4096)
           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
         """
         self._out_queue.append((ip, port, payload))
     class Mainloop(object):
       """Generic mainloop for daemons

Also available in: Unified diff

Synnefo » snf-ganeti » ganeti-local

Revision 5f3269fc