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
projects / ganeti-local / commitdiff