+ return AsyncoreDelayFunction(timeout)
+
+
+class GanetiBaseAsyncoreDispatcher(asyncore.dispatcher):
+ """Base Ganeti Asyncore Dispacher
+
+ """
+ # this method is overriding an asyncore.dispatcher method
+ def handle_error(self):
+ """Log an error in handling any request, and proceed.
+
+ """
+ logging.exception("Error while handling asyncore request")
+
+ # this method is overriding an asyncore.dispatcher method
+ def writable(self):
+ """Most of the time we don't want to check for writability.
+
+ """
+ return False
+
+
+class AsyncStreamServer(GanetiBaseAsyncoreDispatcher):
+ """A stream server to use with asyncore.
+
+ Each request is accepted, and then dispatched to a separate asyncore
+ dispatcher to handle.
+
+ """
+
+ _REQUEST_QUEUE_SIZE = 5
+
+ def __init__(self, family, address):
+ """Constructor for AsyncUnixStreamSocket
+
+ @type family: integer
+ @param family: socket family (one of socket.AF_*)
+ @type address: address family dependent
+ @param address: address to bind the socket to
+
+ """
+ GanetiBaseAsyncoreDispatcher.__init__(self)
+ self.family = family
+ self.create_socket(self.family, socket.SOCK_STREAM)
+ self.set_reuse_addr()
+ self.bind(address)
+ self.listen(self._REQUEST_QUEUE_SIZE)
+
+ # this method is overriding an asyncore.dispatcher method
+ def handle_accept(self):
+ """Accept a new client connection.
+
+ Creates a new instance of the handler class, which will use asyncore to
+ serve the client.
+
+ """
+ accept_result = utils.IgnoreSignals(self.accept)
+ if accept_result is not None:
+ connected_socket, client_address = accept_result
+ if self.family == socket.AF_UNIX:
+ # override the client address, as for unix sockets nothing meaningful
+ # is passed in from accept anyway
+ client_address = netutils.GetSocketCredentials(connected_socket)
+ logging.info("Accepted connection from %s",
+ netutils.FormatAddress(client_address, family=self.family))
+ self.handle_connection(connected_socket, client_address)
+
+ def handle_connection(self, connected_socket, client_address):
+ """Handle an already accepted connection.
+
+ """
+ raise NotImplementedError
+
+
+class AsyncTerminatedMessageStream(asynchat.async_chat):
+ """A terminator separated message stream asyncore module.
+
+ Handles a stream connection receiving messages terminated by a defined
+ separator. For each complete message handle_message is called.
+
+ """
+ def __init__(self, connected_socket, peer_address, terminator, family,
+ unhandled_limit):
+ """AsyncTerminatedMessageStream constructor.
+
+ @type connected_socket: socket.socket
+ @param connected_socket: connected stream socket to receive messages from
+ @param peer_address: family-specific peer address
+ @type terminator: string
+ @param terminator: terminator separating messages in the stream
+ @type family: integer
+ @param family: socket family
+ @type unhandled_limit: integer or None
+ @param unhandled_limit: maximum unanswered messages
+
+ """
+ # python 2.4/2.5 uses conn=... while 2.6 has sock=... we have to cheat by
+ # using a positional argument rather than a keyword one.
+ asynchat.async_chat.__init__(self, connected_socket)
+ self.connected_socket = connected_socket
+ # on python 2.4 there is no "family" attribute for the socket class
+ # FIXME: when we move to python 2.5 or above remove the family parameter
+ #self.family = self.connected_socket.family
+ self.family = family
+ self.peer_address = peer_address
+ self.terminator = terminator
+ self.unhandled_limit = unhandled_limit
+ self.set_terminator(terminator)
+ self.ibuffer = []
+ self.receive_count = 0
+ self.send_count = 0
+ self.oqueue = collections.deque()
+ self.iqueue = collections.deque()
+
+ # this method is overriding an asynchat.async_chat method
+ def collect_incoming_data(self, data):
+ self.ibuffer.append(data)
+
+ def _can_handle_message(self):
+ return (self.unhandled_limit is None or
+ (self.receive_count < self.send_count + self.unhandled_limit) and
+ not self.iqueue)
+
+ # this method is overriding an asynchat.async_chat method
+ def found_terminator(self):
+ message = "".join(self.ibuffer)
+ self.ibuffer = []
+ message_id = self.receive_count
+ # We need to increase the receive_count after checking if the message can
+ # be handled, but before calling handle_message
+ can_handle = self._can_handle_message()
+ self.receive_count += 1
+ if can_handle:
+ self.handle_message(message, message_id)
+ else:
+ self.iqueue.append((message, message_id))
+
+ def handle_message(self, message, message_id):
+ """Handle a terminated message.
+
+ @type message: string
+ @param message: message to handle
+ @type message_id: integer
+ @param message_id: stream's message sequence number
+
+ """
+ pass
+ # TODO: move this method to raise NotImplementedError
+ # raise NotImplementedError
+
+ def send_message(self, message):
+ """Send a message to the remote peer. This function is thread-safe.
+
+ @type message: string
+ @param message: message to send, without the terminator
+
+ @warning: If calling this function from a thread different than the one
+ performing the main asyncore loop, remember that you have to wake that one
+ up.
+
+ """
+ # If we just append the message we received to the output queue, this
+ # function can be safely called by multiple threads at the same time, and
+ # we don't need locking, since deques are thread safe. handle_write in the
+ # asyncore thread will handle the next input message if there are any
+ # enqueued.
+ self.oqueue.append(message)
+
+ # this method is overriding an asyncore.dispatcher method
+ def readable(self):
+ # read from the socket if we can handle the next requests
+ return self._can_handle_message() and asynchat.async_chat.readable(self)
+
+ # this method is overriding an asyncore.dispatcher method
+ def writable(self):
+ # the output queue may become full just after we called writable. This only
+ # works if we know we'll have something else waking us up from the select,
+ # in such case, anyway.
+ return asynchat.async_chat.writable(self) or self.oqueue
+
+ # this method is overriding an asyncore.dispatcher method
+ def handle_write(self):
+ if self.oqueue:
+ # if we have data in the output queue, then send_message was called.
+ # this means we can process one more message from the input queue, if
+ # there are any.
+ data = self.oqueue.popleft()
+ self.push(data + self.terminator)
+ self.send_count += 1
+ if self.iqueue:
+ self.handle_message(*self.iqueue.popleft())
+ self.initiate_send()
+
+ def close_log(self):
+ logging.info("Closing connection from %s",
+ netutils.FormatAddress(self.peer_address, family=self.family))
+ self.close()
+
+ # this method is overriding an asyncore.dispatcher method
+ def handle_expt(self):
+ self.close_log()
+
+ # this method is overriding an asyncore.dispatcher method
+ def handle_error(self):
+ """Log an error in handling any request, and proceed.
+
+ """
+ logging.exception("Error while handling asyncore request")
+ self.close_log()
+
+
+class AsyncUDPSocket(GanetiBaseAsyncoreDispatcher):