import asyncore
import asynchat
+import collections
+import grp
import os
+import pwd
import signal
import logging
import sched
from ganeti import utils
from ganeti import constants
from ganeti import errors
+from ganeti import netutils
+
+
+_DEFAULT_RUN_USER = "root"
+_DEFAULT_RUN_GROUP = "root"
class SchedulerBreakout(Exception):
if self.family == socket.AF_UNIX:
# override the client address, as for unix sockets nothing meaningful
# is passed in from accept anyway
- client_address = utils.GetSocketCredentials(connected_socket)
+ client_address = netutils.GetSocketCredentials(connected_socket)
logging.info("Accepted connection from %s",
FormatAddress(self.family, client_address))
self.handle_connection(connected_socket, client_address)
separator. For each complete message handle_message is called.
"""
- def __init__(self, connected_socket, peer_address, terminator, family):
+ def __init__(self, connected_socket, peer_address, terminator, family,
+ unhandled_limit):
"""AsyncTerminatedMessageStream constructor.
@type connected_socket: socket.socket
@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
self.family = family
self.peer_address = peer_address
self.terminator = terminator
+ self.unhandled_limit = unhandled_limit
self.set_terminator(terminator)
self.ibuffer = []
- self.next_incoming_message = 0
+ 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.next_incoming_message
- self.next_incoming_message += 1
- self.handle_message(message, message_id)
+ 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.
# 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",
FormatAddress(self.family, self.peer_address))
"""An improved asyncore udp socket.
"""
- def __init__(self):
+ def __init__(self, family):
"""Constructor for AsyncUDPSocket
"""
GanetiBaseAsyncoreDispatcher.__init__(self)
self._out_queue = []
- self.create_socket(socket.AF_INET, socket.SOCK_DGRAM)
+ self._family = family
+ self.create_socket(family, socket.SOCK_DGRAM)
# this method is overriding an asyncore.dispatcher method
def handle_connect(self):
constants.MAX_UDP_DATA_SIZE)
if recv_result is not None:
payload, address = recv_result
- ip, port = address
+ if self._family == socket.AF_INET6:
+ # we ignore 'flow info' and 'scope id' as we don't need them
+ ip, port, _, _ = address
+ else:
+ ip, port = address
+
self.handle_datagram(payload, ip, port)
def handle_datagram(self, payload, ip, port):
(self.in_socket, self.out_socket) = socket.socketpair(socket.AF_UNIX,
socket.SOCK_STREAM)
self.in_socket.setblocking(0)
+ self.in_socket.shutdown(socket.SHUT_WR)
+ self.out_socket.shutdown(socket.SHUT_RD)
self.set_socket(self.in_socket)
self.need_signal = True
self.signal_fn = signal_fn
def GenericMain(daemon_name, optionparser, dirs, check_fn, exec_fn,
multithreaded=False, console_logging=False,
- default_ssl_cert=None, default_ssl_key=None):
+ default_ssl_cert=None, default_ssl_key=None,
+ user=_DEFAULT_RUN_USER, group=_DEFAULT_RUN_GROUP):
"""Shared main function for daemons.
@type daemon_name: string
@param default_ssl_cert: Default SSL certificate path
@type default_ssl_key: string
@param default_ssl_key: Default SSL key path
+ @param user: Default user to run as
+ @type user: string
+ @param group: Default group to run as
+ @type group: string
"""
optionparser.add_option("-f", "--foreground", dest="fork",
choices=["no", "yes", "only"])
if daemon_name in constants.DAEMONS_PORTS:
- default_bind_address = "0.0.0.0"
- default_port = utils.GetDaemonPort(daemon_name)
+ default_bind_address = constants.IP4_ADDRESS_ANY
+ default_port = netutils.GetDaemonPort(daemon_name)
# For networked daemons we allow choosing the port and bind address
optionparser.add_option("-p", "--port", dest="port",
utils.EnsureDirs(dirs)
if options.fork:
+ try:
+ uid = pwd.getpwnam(user).pw_uid
+ gid = grp.getgrnam(group).gr_gid
+ except KeyError:
+ raise errors.ConfigurationError("User or group not existing on system:"
+ " %s:%s" % (user, group))
utils.CloseFDs()
- utils.Daemonize(logfile=constants.DAEMONS_LOGFILES[daemon_name])
+ utils.Daemonize(constants.DAEMONS_LOGFILES[daemon_name], uid, gid)
utils.WritePidFile(daemon_name)
try:
projects / ganeti-local / blobdiff