Re-factoring event loop

This commit is contained in:
Daniel P. Berrange 2007-06-26 19:11:00 +00:00
parent f2058815e7
commit 0cb54b45f1
6 changed files with 786 additions and 175 deletions

View File

@ -1,3 +1,11 @@
Tue Jun 26 14:52:00 EST 2007 Daniel P. Berrange <berrange@rdhat.com>
* qemud/event.c, qemud/event.h, qemud/Makefile.am: Generic
standalone event loop implementation for monitoring file
handles & timers.
* qemud/qemud.c, qemud/internal.h: Adapt to use the generic
event loop
Tue Jun 26 14:40:00 BST 2007 Richard W.M. Jones <rjones@redhat.com> Tue Jun 26 14:40:00 BST 2007 Richard W.M. Jones <rjones@redhat.com>
* src/remote_internal.c, python/Makefile.am: Python bindings * src/remote_internal.c, python/Makefile.am: Python bindings

View File

@ -16,7 +16,8 @@ libvirt_qemud_SOURCES = \
buf.c buf.h \ buf.c buf.h \
protocol.h protocol.c \ protocol.h protocol.c \
remote_protocol.h remote_protocol.c \ remote_protocol.h remote_protocol.c \
remote.c remote.c \
event.c event.h
#-D_XOPEN_SOURCE=600 -D_XOPEN_SOURCE_EXTENDED=1 -D_POSIX_C_SOURCE=199506L #-D_XOPEN_SOURCE=600 -D_XOPEN_SOURCE_EXTENDED=1 -D_POSIX_C_SOURCE=199506L
libvirt_qemud_CFLAGS = \ libvirt_qemud_CFLAGS = \
-I$(top_srcdir)/include -I$(top_builddir)/include $(LIBXML_CFLAGS) \ -I$(top_srcdir)/include -I$(top_builddir)/include $(LIBXML_CFLAGS) \

477
qemud/event.c Normal file
View File

@ -0,0 +1,477 @@
/*
* event.h: event loop for monitoring file handles
*
* Copyright (C) 2007 Daniel P. Berrange
* Copyright (C) 2007 Red Hat, Inc.
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 2.1 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public
* License along with this library; if not, write to the Free Software
* Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
*
* Author: Daniel P. Berrange <berrange@redhat.com>
*/
#include <stdlib.h>
#include <string.h>
#include <poll.h>
#include <sys/time.h>
#include <errno.h>
#include "internal.h"
#include "event.h"
/* State for a single file handle being monitored */
struct virEventHandle {
int fd;
int events;
virEventHandleCallback cb;
void *opaque;
int deleted;
};
/* State for a single timer being generated */
struct virEventTimeout {
int timer;
int timeout;
unsigned long long expiresAt;
virEventTimeoutCallback cb;
void *opaque;
int deleted;
};
/* Allocate extra slots for virEventHandle/virEventTimeout
records in this multiple */
#define EVENT_ALLOC_EXTENT 10
/* State for the main event loop */
struct virEventLoop {
int handlesCount;
int handlesAlloc;
struct virEventHandle *handles;
int timeoutsCount;
int timeoutsAlloc;
struct virEventTimeout *timeouts;
};
/* Only have one event loop */
static struct virEventLoop eventLoop;
/* Unique ID for the next timer to be registered */
static int nextTimer = 0;
/*
* Register a callback for monitoring file handle events.
* NB, it *must* be safe to call this from within a callback
* For this reason we only ever append to existing list.
*/
int virEventAddHandle(int fd, int events, virEventHandleCallback cb, void *opaque) {
qemudDebug("Add handle %d %d %p %p\n", fd, events, cb, opaque);
if (eventLoop.handlesCount == eventLoop.handlesAlloc) {
struct virEventHandle *tmp;
qemudDebug("Used %d handle slots, adding %d more\n",
eventLoop.handlesAlloc, EVENT_ALLOC_EXTENT);
tmp = realloc(eventLoop.handles,
sizeof(struct virEventHandle) *
(eventLoop.handlesAlloc + EVENT_ALLOC_EXTENT));
if (!tmp) {
return -1;
}
eventLoop.handles = tmp;
eventLoop.handlesAlloc += EVENT_ALLOC_EXTENT;
}
eventLoop.handles[eventLoop.handlesCount].fd = fd;
eventLoop.handles[eventLoop.handlesCount].events = events;
eventLoop.handles[eventLoop.handlesCount].cb = cb;
eventLoop.handles[eventLoop.handlesCount].opaque = opaque;
eventLoop.handles[eventLoop.handlesCount].deleted = 0;
eventLoop.handlesCount++;
return 0;
}
/*
* Unregister a callback from a file handle
* NB, it *must* be safe to call this from within a callback
* For this reason we only ever set a flag in the existing list.
* Actual deletion will be done out-of-band
*/
int virEventRemoveHandle(int fd) {
int i;
qemudDebug("Remove handle %d\n", fd);
for (i = 0 ; i < eventLoop.handlesCount ; i++) {
if (eventLoop.handles[i].deleted)
continue;
if (eventLoop.handles[i].fd == fd) {
qemudDebug("mark delete %d\n", i);
eventLoop.handles[i].deleted = 1;
return 0;
}
}
return -1;
}
/*
* Register a callback for a timer event
* NB, it *must* be safe to call this from within a callback
* For this reason we only ever append to existing list.
*/
int virEventAddTimeout(int timeout, virEventTimeoutCallback cb, void *opaque) {
struct timeval tv;
qemudDebug("Adding timeout with %d ms period", timeout);
if (gettimeofday(&tv, NULL) < 0) {
return -1;
}
if (eventLoop.timeoutsCount == eventLoop.timeoutsAlloc) {
struct virEventTimeout *tmp;
qemudDebug("Used %d timeout slots, adding %d more\n",
eventLoop.timeoutsAlloc, EVENT_ALLOC_EXTENT);
tmp = realloc(eventLoop.timeouts,
sizeof(struct virEventTimeout) *
(eventLoop.timeoutsAlloc + EVENT_ALLOC_EXTENT));
if (!tmp) {
return -1;
}
eventLoop.timeouts = tmp;
eventLoop.timeoutsAlloc += EVENT_ALLOC_EXTENT;
}
eventLoop.timeouts[eventLoop.timeoutsCount].timer = nextTimer++;
eventLoop.timeouts[eventLoop.timeoutsCount].timeout = timeout;
eventLoop.timeouts[eventLoop.timeoutsCount].cb = cb;
eventLoop.timeouts[eventLoop.timeoutsCount].opaque = opaque;
eventLoop.timeouts[eventLoop.timeoutsCount].deleted = 0;
eventLoop.timeouts[eventLoop.timeoutsCount].expiresAt =
timeout +
(((unsigned long long)tv.tv_sec)*1000) +
(((unsigned long long)tv.tv_usec)/1000);
eventLoop.timeoutsCount++;
return nextTimer-1;
}
/*
* Unregister a callback for a timer
* NB, it *must* be safe to call this from within a callback
* For this reason we only ever set a flag in the existing list.
* Actual deletion will be done out-of-band
*/
int virEventRemoveTimeout(int timer) {
int i;
for (i = 0 ; i < eventLoop.timeoutsCount ; i++) {
if (eventLoop.timeouts[i].deleted)
continue;
if (eventLoop.timeouts[i].timer == timer) {
eventLoop.timeouts[i].deleted = 1;
return 0;
}
}
return -1;
}
/* Iterates over all registered timeouts and determine which
* will be the first to expire.
* @timeout: filled with expiry time of soonest timer, or -1 if
* no timeout is pending
* returns: 0 on success, -1 on error
*/
static int virEventCalculateTimeout(int *timeout) {
unsigned long long then = 0;
int i;
/* Figure out if we need a timeout */
for (i = 0 ; i < eventLoop.timeoutsCount ; i++) {
if (eventLoop.timeouts[i].deleted)
continue;
qemudDebug("Got a timeout scheduled for %llu", eventLoop.timeouts[i].expiresAt);
if (then == 0 ||
eventLoop.timeouts[i].expiresAt < then)
then = eventLoop.timeouts[i].expiresAt;
}
/* Calculate how long we should wait for a timeout if needed */
if (then > 0) {
struct timeval tv;
if (gettimeofday(&tv, NULL) < 0) {
return -1;
}
*timeout = then -
((((unsigned long long)tv.tv_sec)*1000) +
(((unsigned long long)tv.tv_usec)/1000));
qemudDebug("Timeout at %llu due in %d ms", then, *timeout);
if (*timeout < 0)
*timeout = 1;
} else {
qemudDebug("No timeout due");
*timeout = -1;
}
return 0;
}
/*
* Allocate a pollfd array containing data for all registered
* file handles. The caller must free the returned data struct
* returns: the pollfd array, or NULL on error
*/
static int virEventMakePollFDs(struct pollfd **retfds) {
struct pollfd *fds;
int i, nfds = 0;
for (i = 0 ; i < eventLoop.handlesCount ; i++) {
if (eventLoop.handles[i].deleted)
continue;
nfds++;
}
*retfds = NULL;
/* Setup the poll file handle data structs */
if (!(fds = malloc(sizeof(struct pollfd) * nfds)))
return -1;
for (i = 0, nfds = 0 ; i < eventLoop.handlesCount ; i++) {
if (eventLoop.handles[i].deleted)
continue;
fds[nfds].fd = eventLoop.handles[i].fd;
fds[nfds].events = eventLoop.handles[i].events;
fds[nfds].revents = 0;
qemudDebug("Wait for %d %d\n", eventLoop.handles[i].fd, eventLoop.handles[i].events);
nfds++;
}
*retfds = fds;
return nfds;
}
/*
* Iterate over all timers and determine if any have expired.
* Invoke the user supplied callback for each timer whose
* expiry time is met, and schedule the next timeout. Does
* not try to 'catch up' on time if the actual expiry time
* was later than the requested time.
*
* This method must cope with new timers being registered
* by a callback, and must skip any timers marked as deleted.
*
* Returns 0 upon success, -1 if an error occurred
*/
static int virEventDispatchTimeouts(void) {
struct timeval tv;
unsigned long long now;
int i;
/* Save this now - it may be changed during dispatch */
int ntimeouts = eventLoop.timeoutsCount;
if (gettimeofday(&tv, NULL) < 0) {
return -1;
}
now = (((unsigned long long)tv.tv_sec)*1000) +
(((unsigned long long)tv.tv_usec)/1000);
for (i = 0 ; i < ntimeouts ; i++) {
if (eventLoop.timeouts[i].deleted)
continue;
if (eventLoop.timeouts[i].expiresAt <= now) {
(eventLoop.timeouts[i].cb)(eventLoop.timeouts[i].timer,
eventLoop.timeouts[i].opaque);
eventLoop.timeouts[i].expiresAt =
now + eventLoop.timeouts[i].timeout;
}
}
return 0;
}
/* Iterate over all file handles and dispatch any which
* have pending events listed in the poll() data. Invoke
* the user supplied callback for each handle which has
* pending events
*
* This method must cope with new handles being registered
* by a callback, and must skip any handles marked as deleted.
*
* Returns 0 upon success, -1 if an error occurred
*/
static int virEventDispatchHandles(struct pollfd *fds) {
int i;
/* Save this now - it may be changed during dispatch */
int nhandles = eventLoop.handlesCount;
for (i = 0 ; i < nhandles ; i++) {
if (eventLoop.handles[i].deleted) {
qemudDebug("Skip deleted %d\n", eventLoop.handles[i].fd);
continue;
}
if (fds[i].revents) {
qemudDebug("Dispatch %d %d %p\n", fds[i].fd, fds[i].revents, eventLoop.handles[i].opaque);
(eventLoop.handles[i].cb)(fds[i].fd, fds[i].revents,
eventLoop.handles[i].opaque);
}
}
return 0;
}
/* Used post dispatch to actually remove any timers that
* were previously marked as deleted. This asynchronous
* cleanup is needed to make dispatch re-entrant safe.
*/
static int virEventCleanupTimeouts(void) {
int i;
/* Remove deleted entries, shuffling down remaining
* entries as needed to form contigous series
*/
for (i = 0 ; i < eventLoop.timeoutsCount ; ) {
if (!eventLoop.timeouts[i].deleted) {
i++;
continue;
}
qemudDebug("Purging timeout %d with id %d", i, eventLoop.timeouts[i].timer);
if ((i+1) < eventLoop.timeoutsCount) {
memmove(eventLoop.timeouts+i,
eventLoop.timeouts+i+1,
sizeof(struct virEventTimeout)*(eventLoop.timeoutsCount-(i+1)));
}
eventLoop.timeoutsCount--;
}
/* Release some memory if we've got a big chunk free */
if ((eventLoop.timeoutsAlloc - EVENT_ALLOC_EXTENT) > eventLoop.timeoutsCount) {
struct virEventTimeout *tmp;
qemudDebug("Releasing %d out of %d timeout slots used, releasing %d\n",
eventLoop.timeoutsCount, eventLoop.timeoutsAlloc, EVENT_ALLOC_EXTENT);
tmp = realloc(eventLoop.timeouts,
sizeof(struct virEventTimeout) *
(eventLoop.timeoutsAlloc - EVENT_ALLOC_EXTENT));
if (!tmp) {
return -1;
}
eventLoop.timeouts = tmp;
eventLoop.timeoutsAlloc -= EVENT_ALLOC_EXTENT;
}
return 0;
}
/* Used post dispatch to actually remove any handles that
* were previously marked as deleted. This asynchronous
* cleanup is needed to make dispatch re-entrant safe.
*/
static int virEventCleanupHandles(void) {
int i;
/* Remove deleted entries, shuffling down remaining
* entries as needed to form contigous series
*/
for (i = 0 ; i < eventLoop.handlesCount ; ) {
if (!eventLoop.handles[i].deleted) {
i++;
continue;
}
if ((i+1) < eventLoop.handlesCount) {
memmove(eventLoop.handles+i,
eventLoop.handles+i+1,
sizeof(struct virEventHandle)*(eventLoop.handlesCount-(i+1)));
}
eventLoop.handlesCount--;
}
/* Release some memory if we've got a big chunk free */
if ((eventLoop.handlesAlloc - EVENT_ALLOC_EXTENT) > eventLoop.handlesCount) {
struct virEventHandle *tmp;
qemudDebug("Releasing %d out of %d handles slots used, releasing %d\n",
eventLoop.handlesCount, eventLoop.handlesAlloc, EVENT_ALLOC_EXTENT);
tmp = realloc(eventLoop.handles,
sizeof(struct virEventHandle) *
(eventLoop.handlesAlloc - EVENT_ALLOC_EXTENT));
if (!tmp) {
return -1;
}
eventLoop.handles = tmp;
eventLoop.handlesAlloc -= EVENT_ALLOC_EXTENT;
}
return 0;
}
/*
* Run a single iteration of the event loop, blocking until
* at least one file handle has an event, or a timer expires
*/
int virEventRunOnce(void) {
struct pollfd *fds;
int ret, timeout, nfds;
if ((nfds = virEventMakePollFDs(&fds)) < 0)
return -1;
if (virEventCalculateTimeout(&timeout) < 0) {
free(fds);
return -1;
}
retry:
qemudDebug("Poll on %d handles %p timeout %d\n", nfds, fds, timeout);
ret = poll(fds, nfds, timeout);
qemudDebug("Poll got %d event\n", ret);
if (ret < 0) {
if (errno == EINTR) {
goto retry;
}
free(fds);
return -1;
}
if (virEventDispatchTimeouts() < 0) {
free(fds);
return -1;
}
if (ret > 0 &&
virEventDispatchHandles(fds) < 0) {
free(fds);
return -1;
}
free(fds);
if (virEventCleanupTimeouts() < 0)
return -1;
if (virEventCleanupHandles() < 0)
return -1;
return 0;
}
/*
* Local variables:
* indent-tabs-mode: nil
* c-indent-level: 4
* c-basic-offset: 4
* tab-width: 4
* End:
*/

98
qemud/event.h Normal file
View File

@ -0,0 +1,98 @@
/*
* event.h: event loop for monitoring file handles
*
* Copyright (C) 2007 Daniel P. Berrange
* Copyright (C) 2007 Red Hat, Inc.
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 2.1 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public
* License along with this library; if not, write to the Free Software
* Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
*
* Author: Daniel P. Berrange <berrange@redhat.com>
*/
#ifndef __VIRTD_EVENT_H__
#define __VIRTD_EVENT_H__
/**
* virEventHandleCallback: callback for receiving file handle events
*
* @fd: file handle on which the event occured
* @events: bitset of events from POLLnnn constants
* @opaque: user data registered with handle
*/
typedef void (*virEventHandleCallback)(int fd, int events, void *opaque);
/**
* virEventAddHandle: register a callback for monitoring file handle events
*
* @fd: file handle to monitor for events
* @events: bitset of events to wach from POLLnnn constants
* @cb: callback to invoke when an event occurrs
* @opaque: user data to pass to callback
*
* returns -1 if the file handle cannot be registered, 0 upon success
*/
int virEventAddHandle(int fd, int events, virEventHandleCallback cb, void *opaque);
/**
* virEventRemoveHandle: unregister a callback from a file handle
*
* @fd: file handle to stop monitoring for events
*
* returns -1 if the file handle was not registered, 0 upon success
*/
int virEventRemoveHandle(int fd);
/**
* virEventTimeoutCallback: callback for receiving timer events
*
* @timer: timer id emitting the event
* @opaque: user data registered with handle
*/
typedef void (*virEventTimeoutCallback)(int timer, void *opaque);
/**
* virEventAddTimeout: register a callback for a timer event
*
* @timeout: timeout between events in milliseconds
* @cb: callback to invoke when an event occurrs
* @opaque: user data to pass to callback
*
* returns -1 if the file handle cannot be registered, a positive
* integer timer id upon success
*/
int virEventAddTimeout(int timeout, virEventTimeoutCallback cb, void *opaque);
/**
* virEventRemoveTimeout: unregister a callback for a timer
*
* @timer: the timer id to remove
*
* returns -1 if the timer was not registered, 0 upon success
*/
int virEventRemoveTimeout(int timer);
/**
* virEventRunOnce: run a single iteration of the event loop.
*
* Blocks the caller until at least one file handle has an
* event or the first timer expires.
*
* returns -1 if the event monitoring failed
*/
int virEventRunOnce(void);
#endif /* __VIRTD_EVENT_H__ */

View File

@ -338,7 +338,6 @@ struct qemud_server {
int nclients; int nclients;
struct qemud_client *clients; struct qemud_client *clients;
int sigread; int sigread;
int nvmfds;
int nactivevms; int nactivevms;
int ninactivevms; int ninactivevms;
struct qemud_vm *vms; struct qemud_vm *vms;

View File

@ -61,6 +61,7 @@
#include "driver.h" #include "driver.h"
#include "conf.h" #include "conf.h"
#include "iptables.h" #include "iptables.h"
#include "event.h"
static int godaemon = 0; /* -d: Be a daemon */ static int godaemon = 0; /* -d: Be a daemon */
static int verbose = 0; /* -v: Verbose mode */ static int verbose = 0; /* -v: Verbose mode */
@ -110,6 +111,13 @@ static void sig_handler(int sig) {
errno = origerrno; errno = origerrno;
} }
static void qemudDispatchVMEvent(int fd, int events, void *opaque);
static void qemudDispatchClientEvent(int fd, int events, void *opaque);
static void qemudDispatchServerEvent(int fd, int events, void *opaque);
static int qemudRegisterClientEvent(struct qemud_server *server,
struct qemud_client *client,
int remove);
static int static int
remoteInitializeGnuTLS (void) remoteInitializeGnuTLS (void)
{ {
@ -184,8 +192,10 @@ remoteInitializeGnuTLS (void)
return 0; return 0;
} }
static int qemudDispatchSignal(struct qemud_server *server) static void qemudDispatchSignalEvent(int fd ATTRIBUTE_UNUSED,
{ int events ATTRIBUTE_UNUSED,
void *opaque) {
struct qemud_server *server = (struct qemud_server *)opaque;
unsigned char sigc; unsigned char sigc;
struct qemud_vm *vm; struct qemud_vm *vm;
struct qemud_network *network; struct qemud_network *network;
@ -194,7 +204,7 @@ static int qemudDispatchSignal(struct qemud_server *server)
if (read(server->sigread, &sigc, 1) != 1) { if (read(server->sigread, &sigc, 1) != 1) {
qemudLog(QEMUD_ERR, "Failed to read from signal pipe: %s", qemudLog(QEMUD_ERR, "Failed to read from signal pipe: %s",
strerror(errno)); strerror(errno));
return -1; return;
} }
ret = 0; ret = 0;
@ -266,7 +276,8 @@ static int qemudDispatchSignal(struct qemud_server *server)
break; break;
} }
return ret; if (ret != 0)
server->shutdown = 1;
} }
static int qemudSetCloseExec(int fd) { static int qemudSetCloseExec(int fd) {
@ -474,19 +485,16 @@ static int qemudListenUnix(struct qemud_server *server,
} }
sock->readonly = readonly; sock->readonly = readonly;
sock->next = server->sockets;
server->sockets = sock;
server->nsockets++;
if ((sock->fd = socket(PF_UNIX, SOCK_STREAM, 0)) < 0) { if ((sock->fd = socket(PF_UNIX, SOCK_STREAM, 0)) < 0) {
qemudLog(QEMUD_ERR, "Failed to create socket: %s", qemudLog(QEMUD_ERR, "Failed to create socket: %s",
strerror(errno)); strerror(errno));
return -1; goto cleanup;
} }
if (qemudSetCloseExec(sock->fd) < 0 || if (qemudSetCloseExec(sock->fd) < 0 ||
qemudSetNonBlock(sock->fd) < 0) qemudSetNonBlock(sock->fd) < 0)
return -1; goto cleanup;
memset(&addr, 0, sizeof(addr)); memset(&addr, 0, sizeof(addr));
addr.sun_family = AF_UNIX; addr.sun_family = AF_UNIX;
@ -502,17 +510,35 @@ static int qemudListenUnix(struct qemud_server *server,
if (bind(sock->fd, (struct sockaddr *)&addr, sizeof(addr)) < 0) { if (bind(sock->fd, (struct sockaddr *)&addr, sizeof(addr)) < 0) {
qemudLog(QEMUD_ERR, "Failed to bind socket to '%s': %s", qemudLog(QEMUD_ERR, "Failed to bind socket to '%s': %s",
path, strerror(errno)); path, strerror(errno));
return -1; goto cleanup;
} }
umask(oldmask); umask(oldmask);
if (listen(sock->fd, 30) < 0) { if (listen(sock->fd, 30) < 0) {
qemudLog(QEMUD_ERR, "Failed to listen for connections on '%s': %s", qemudLog(QEMUD_ERR, "Failed to listen for connections on '%s': %s",
path, strerror(errno)); path, strerror(errno));
return -1; goto cleanup;
} }
if (virEventAddHandle(sock->fd,
POLLIN| POLLERR | POLLHUP,
qemudDispatchServerEvent,
server) < 0) {
qemudLog(QEMUD_ERR, "Failed to add server event callback");
goto cleanup;
}
sock->next = server->sockets;
server->sockets = sock;
server->nsockets++;
return 0; return 0;
cleanup:
if (sock->fd)
close(sock->fd);
free(sock);
return -1;
} }
// See: http://people.redhat.com/drepper/userapi-ipv6.html // See: http://people.redhat.com/drepper/userapi-ipv6.html
@ -606,6 +632,15 @@ remoteListenTCP (struct qemud_server *server,
"remoteListenTCP: listen: %s", strerror (errno)); "remoteListenTCP: listen: %s", strerror (errno));
return -1; return -1;
} }
if (virEventAddHandle(sock->fd,
POLLIN| POLLERR | POLLHUP,
qemudDispatchServerEvent,
server) < 0) {
qemudLog(QEMUD_ERR, "Failed to add server event callback");
return -1;
}
} }
return 0; return 0;
@ -1026,11 +1061,15 @@ static int qemudDispatchServer(struct qemud_server *server, struct qemud_socket
if (!client->tls) { if (!client->tls) {
client->mode = QEMUD_MODE_RX_HEADER; client->mode = QEMUD_MODE_RX_HEADER;
client->bufferLength = QEMUD_PKT_HEADER_XDR_LEN; client->bufferLength = QEMUD_PKT_HEADER_XDR_LEN;
if (qemudRegisterClientEvent (server, client, 0) < 0)
goto cleanup;
} else { } else {
int ret; int ret;
client->session = remoteInitializeTLSSession (); client->session = remoteInitializeTLSSession ();
if (client->session == NULL) goto tls_failed; if (client->session == NULL)
goto cleanup;
gnutls_transport_set_ptr (client->session, gnutls_transport_set_ptr (client->session,
(gnutls_transport_ptr_t) (long) fd); (gnutls_transport_ptr_t) (long) fd);
@ -1040,16 +1079,22 @@ static int qemudDispatchServer(struct qemud_server *server, struct qemud_socket
if (ret == 0) { if (ret == 0) {
/* Unlikely, but ... Next step is to check the certificate. */ /* Unlikely, but ... Next step is to check the certificate. */
if (remoteCheckAccess (client) == -1) if (remoteCheckAccess (client) == -1)
goto tls_failed; goto cleanup;
if (qemudRegisterClientEvent(server, client, 0) < 0)
goto cleanup;
} else if (ret == GNUTLS_E_INTERRUPTED || ret == GNUTLS_E_AGAIN) { } else if (ret == GNUTLS_E_INTERRUPTED || ret == GNUTLS_E_AGAIN) {
/* Most likely. */ /* Most likely. */
client->mode = QEMUD_MODE_TLS_HANDSHAKE; client->mode = QEMUD_MODE_TLS_HANDSHAKE;
client->bufferLength = -1; client->bufferLength = -1;
client->direction = gnutls_record_get_direction (client->session); client->direction = gnutls_record_get_direction (client->session);
if (qemudRegisterClientEvent (server, client, 0) < 0)
goto cleanup;
} else { } else {
qemudLog (QEMUD_ERR, "TLS handshake failed: %s", qemudLog (QEMUD_ERR, "TLS handshake failed: %s",
gnutls_strerror (ret)); gnutls_strerror (ret));
goto tls_failed; goto cleanup;
} }
} }
@ -1059,7 +1104,7 @@ static int qemudDispatchServer(struct qemud_server *server, struct qemud_socket
return 0; return 0;
tls_failed: cleanup:
if (client->session) gnutls_deinit (client->session); if (client->session) gnutls_deinit (client->session);
close (fd); close (fd);
free (client); free (client);
@ -1453,7 +1498,15 @@ int qemudStartVMDaemon(struct qemud_server *server,
server->ninactivevms--; server->ninactivevms--;
server->nactivevms++; server->nactivevms++;
server->nvmfds += 2;
virEventAddHandle(vm->stdout,
POLLIN | POLLERR | POLLHUP,
qemudDispatchVMEvent,
server);
virEventAddHandle(vm->stderr,
POLLIN | POLLERR | POLLHUP,
qemudDispatchVMEvent,
server);
ret = 0; ret = 0;
@ -1497,6 +1550,8 @@ static void qemudDispatchClientFailure(struct qemud_server *server, struct qemud
tmp = tmp->next; tmp = tmp->next;
} }
virEventRemoveHandle(client->fd);
if (client->tls && client->session) gnutls_deinit (client->session); if (client->tls && client->session) gnutls_deinit (client->session);
close(client->fd); close(client->fd);
free(client); free(client);
@ -1590,6 +1645,8 @@ static int qemudClientRead(struct qemud_server *server,
} else { } else {
ret = gnutls_record_recv (client->session, data, len); ret = gnutls_record_recv (client->session, data, len);
client->direction = gnutls_record_get_direction (client->session); client->direction = gnutls_record_get_direction (client->session);
if (qemudRegisterClientEvent (server, client, 1) < 0)
qemudDispatchClientFailure (server, client);
if (ret <= 0) { if (ret <= 0) {
if (ret == 0 || (ret != GNUTLS_E_AGAIN && if (ret == 0 || (ret != GNUTLS_E_AGAIN &&
ret != GNUTLS_E_INTERRUPTED)) { ret != GNUTLS_E_INTERRUPTED)) {
@ -1655,6 +1712,11 @@ static void qemudDispatchClientRead(struct qemud_server *server, struct qemud_cl
xdr_destroy (&x); xdr_destroy (&x);
if (qemudRegisterClientEvent(server, client, 1) < 0) {
qemudDispatchClientFailure(server, client);
return;
}
/* Fall through */ /* Fall through */
} }
@ -1679,6 +1741,8 @@ static void qemudDispatchClientRead(struct qemud_server *server, struct qemud_cl
if (remote && h.prog == REMOTE_PROGRAM) { if (remote && h.prog == REMOTE_PROGRAM) {
remoteDispatchClientRequest (server, client); remoteDispatchClientRequest (server, client);
if (qemudRegisterClientEvent(server, client, 1) < 0)
qemudDispatchClientFailure(server, client);
} else if (!remote && h.prog == QEMUD_PROGRAM) { } else if (!remote && h.prog == QEMUD_PROGRAM) {
qemud_packet_client p; qemud_packet_client p;
@ -1689,6 +1753,9 @@ static void qemudDispatchClientRead(struct qemud_server *server, struct qemud_cl
} }
qemudDispatchClientRequest(server, client, &p); qemudDispatchClientRequest(server, client, &p);
if (qemudRegisterClientEvent(server, client, 1) < 0)
qemudDispatchClientFailure(server, client);
} else { } else {
/* An internal error. */ /* An internal error. */
qemudDebug ("Not REMOTE_PROGRAM or QEMUD_PROGRAM"); qemudDebug ("Not REMOTE_PROGRAM or QEMUD_PROGRAM");
@ -1709,12 +1776,17 @@ static void qemudDispatchClientRead(struct qemud_server *server, struct qemud_cl
/* Finished. Next step is to check the certificate. */ /* Finished. Next step is to check the certificate. */
if (remoteCheckAccess (client) == -1) if (remoteCheckAccess (client) == -1)
qemudDispatchClientFailure (server, client); qemudDispatchClientFailure (server, client);
if (qemudRegisterClientEvent (server, client, 1) < 0)
qemudDispatchClientFailure (server, client);
} else if (ret != GNUTLS_E_AGAIN && ret != GNUTLS_E_INTERRUPTED) { } else if (ret != GNUTLS_E_AGAIN && ret != GNUTLS_E_INTERRUPTED) {
qemudLog (QEMUD_ERR, "TLS handshake failed: %s", qemudLog (QEMUD_ERR, "TLS handshake failed: %s",
gnutls_strerror (ret)); gnutls_strerror (ret));
qemudDispatchClientFailure (server, client); qemudDispatchClientFailure (server, client);
} else } else {
client->direction = gnutls_record_get_direction (client->session); client->direction = gnutls_record_get_direction (client->session);
if (qemudRegisterClientEvent (server ,client, 1) < 0)
qemudDispatchClientFailure (server, client);
}
break; break;
} }
@ -1745,6 +1817,8 @@ static int qemudClientWrite(struct qemud_server *server,
} else { } else {
ret = gnutls_record_send (client->session, data, len); ret = gnutls_record_send (client->session, data, len);
client->direction = gnutls_record_get_direction (client->session); client->direction = gnutls_record_get_direction (client->session);
if (qemudRegisterClientEvent (server, client, 1) < 0)
qemudDispatchClientFailure (server, client);
if (ret < 0) { if (ret < 0) {
if (ret != GNUTLS_E_INTERRUPTED && ret != GNUTLS_E_AGAIN) { if (ret != GNUTLS_E_INTERRUPTED && ret != GNUTLS_E_AGAIN) {
qemudLog (QEMUD_ERR, "gnutls_record_send: %s", qemudLog (QEMUD_ERR, "gnutls_record_send: %s",
@ -1772,6 +1846,9 @@ static void qemudDispatchClientWrite(struct qemud_server *server, struct qemud_c
client->bufferLength = QEMUD_PKT_HEADER_XDR_LEN; client->bufferLength = QEMUD_PKT_HEADER_XDR_LEN;
client->bufferOffset = 0; client->bufferOffset = 0;
if (client->tls) client->direction = QEMUD_TLS_DIRECTION_READ; if (client->tls) client->direction = QEMUD_TLS_DIRECTION_READ;
if (qemudRegisterClientEvent (server, client, 1) < 0)
qemudDispatchClientFailure (server, client);
} }
/* Still writing */ /* Still writing */
break; break;
@ -1786,12 +1863,18 @@ static void qemudDispatchClientWrite(struct qemud_server *server, struct qemud_c
/* Finished. Next step is to check the certificate. */ /* Finished. Next step is to check the certificate. */
if (remoteCheckAccess (client) == -1) if (remoteCheckAccess (client) == -1)
qemudDispatchClientFailure (server, client); qemudDispatchClientFailure (server, client);
if (qemudRegisterClientEvent (server, client, 1))
qemudDispatchClientFailure (server, client);
} else if (ret != GNUTLS_E_AGAIN && ret != GNUTLS_E_INTERRUPTED) { } else if (ret != GNUTLS_E_AGAIN && ret != GNUTLS_E_INTERRUPTED) {
qemudLog (QEMUD_ERR, "TLS handshake failed: %s", qemudLog (QEMUD_ERR, "TLS handshake failed: %s",
gnutls_strerror (ret)); gnutls_strerror (ret));
qemudDispatchClientFailure (server, client); qemudDispatchClientFailure (server, client);
} else } else {
client->direction = gnutls_record_get_direction (client->session); client->direction = gnutls_record_get_direction (client->session);
if (qemudRegisterClientEvent (server, client, 1))
qemudDispatchClientFailure (server, client);
}
break; break;
} }
@ -1842,6 +1925,10 @@ int qemudShutdownVMDaemon(struct qemud_server *server, struct qemud_vm *vm) {
qemudVMData(server, vm, vm->stdout); qemudVMData(server, vm, vm->stdout);
qemudVMData(server, vm, vm->stderr); qemudVMData(server, vm, vm->stderr);
virEventRemoveHandle(vm->stdout);
virEventRemoveHandle(vm->stderr);
if (close(vm->logfile) < 0) if (close(vm->logfile) < 0)
qemudLog(QEMUD_WARN, "Unable to close logfile %d: %s", errno, strerror(errno)); qemudLog(QEMUD_WARN, "Unable to close logfile %d: %s", errno, strerror(errno));
close(vm->stdout); close(vm->stdout);
@ -1852,7 +1939,6 @@ int qemudShutdownVMDaemon(struct qemud_server *server, struct qemud_vm *vm) {
vm->stdout = -1; vm->stdout = -1;
vm->stderr = -1; vm->stderr = -1;
vm->monitor = -1; vm->monitor = -1;
server->nvmfds -= 2;
if (waitpid(vm->pid, NULL, WNOHANG) != vm->pid) { if (waitpid(vm->pid, NULL, WNOHANG) != vm->pid) {
kill(vm->pid, SIGKILL); kill(vm->pid, SIGKILL);
@ -2340,92 +2426,102 @@ int qemudShutdownNetworkDaemon(struct qemud_server *server,
} }
static int qemudDispatchPoll(struct qemud_server *server, struct pollfd *fds) { static void qemudDispatchVMEvent(int fd, int events, void *opaque) {
struct qemud_socket *sock = server->sockets; struct qemud_server *server = (struct qemud_server *)opaque;
struct qemud_client *client = server->clients; struct qemud_vm *vm = server->vms;
struct qemud_vm *vm;
struct qemud_network *network;
int ret = 0;
int fd = 0;
if (fds[fd++].revents && qemudDispatchSignal(server) < 0)
return -1;
if (server->shutdown)
return 0;
vm = server->vms;
while (vm) { while (vm) {
struct qemud_vm *next = vm->next; if (qemudIsActiveVM(vm) &&
int failed = 0, (vm->stdout == fd ||
stdoutfd = vm->stdout, vm->stderr == fd))
stderrfd = vm->stderr; break;
if (!qemudIsActiveVM(vm)) { vm = vm->next;
vm = next;
continue;
}
if (stdoutfd != -1) {
if (fds[fd].revents) {
if (fds[fd].revents == POLLIN) {
if (qemudDispatchVMLog(server, vm, fds[fd].fd) < 0)
failed = 1;
} else {
if (qemudDispatchVMFailure(server, vm, fds[fd].fd) < 0)
failed = 1;
}
}
fd++;
}
if (stderrfd != -1) {
if (!failed) {
if (fds[fd].revents) {
if (fds[fd].revents == POLLIN) {
if (qemudDispatchVMLog(server, vm, fds[fd].fd) < 0)
failed = 1;
} else {
if (qemudDispatchVMFailure(server, vm, fds[fd].fd) < 0)
failed = 1;
}
}
}
fd++;
}
vm = next;
if (failed)
ret = -1; /* FIXME: the daemon shouldn't exit on failure here */
} }
if (!vm)
return;
if (events == POLLIN &&
qemudDispatchVMLog(server, vm, fd) == 0)
return;
qemudDispatchVMFailure(server, vm, fd);
}
static void qemudDispatchClientEvent(int fd, int events, void *opaque) {
struct qemud_server *server = (struct qemud_server *)opaque;
struct qemud_client *client = server->clients;
while (client) { while (client) {
struct qemud_client *next = client->next; if (client->fd == fd)
break;
assert (client->magic == QEMUD_CLIENT_MAGIC); client = client->next;
if (fds[fd].revents) {
qemudDebug("Poll data normal");
if (fds[fd].revents == POLLOUT)
qemudDispatchClientWrite(server, client);
else if (fds[fd].revents == POLLIN)
qemudDispatchClientRead(server, client);
else
qemudDispatchClientFailure(server, client);
}
fd++;
client = next;
} }
if (!client)
return;
if (events == POLLOUT)
qemudDispatchClientWrite(server, client);
else if (events == POLLIN)
qemudDispatchClientRead(server, client);
else
qemudDispatchClientFailure(server, client);
}
static int qemudRegisterClientEvent(struct qemud_server *server,
struct qemud_client *client,
int removeFirst) {
if (removeFirst)
if (virEventRemoveHandle(client->fd) < 0)
return -1;
if (client->tls) {
if (virEventAddHandle(client->fd,
(client->direction ?
POLLOUT : POLLIN) | POLLERR | POLLHUP,
qemudDispatchClientEvent,
server) < 0)
return -1;
} else {
if (virEventAddHandle(client->fd,
(client->mode == QEMUD_MODE_TX_PACKET ?
POLLOUT : POLLIN) | POLLERR | POLLHUP,
qemudDispatchClientEvent,
server) < 0)
return -1;
}
return 0;
}
static void qemudDispatchServerEvent(int fd, int events, void *opaque) {
struct qemud_server *server = (struct qemud_server *)opaque;
struct qemud_socket *sock = server->sockets;
while (sock) { while (sock) {
struct qemud_socket *next = sock->next; if (sock->fd == fd)
/* FIXME: the daemon shouldn't exit on error here */ break;
if (fds[fd].revents)
if (qemudDispatchServer(server, sock) < 0) sock = sock->next;
return -1;
fd++;
sock = next;
} }
if (!sock)
return;
if (events)
qemudDispatchServer(server, sock);
}
static void qemudCleanupInactive(struct qemud_server *server) {
struct qemud_vm *vm = server->vms;
struct qemud_network *network = server->networks;
/* Cleanup any VMs which shutdown & dont have an associated /* Cleanup any VMs which shutdown & dont have an associated
config file */ config file */
vm = server->vms;
while (vm) { while (vm) {
struct qemud_vm *next = vm->next; struct qemud_vm *next = vm->next;
@ -2436,7 +2532,6 @@ static int qemudDispatchPoll(struct qemud_server *server, struct pollfd *fds) {
} }
/* Cleanup any networks too */ /* Cleanup any networks too */
network = server->networks;
while (network) { while (network) {
struct qemud_network *next = network->next; struct qemud_network *next = network->next;
@ -2446,91 +2541,16 @@ static int qemudDispatchPoll(struct qemud_server *server, struct pollfd *fds) {
network = next; network = next;
} }
return ret; return;
}
static void qemudPreparePoll(struct qemud_server *server, struct pollfd *fds) {
int fd = 0;
struct qemud_socket *sock;
struct qemud_client *client;
struct qemud_vm *vm;
fds[fd].fd = server->sigread;
fds[fd].events = POLLIN;
fd++;
for (vm = server->vms ; vm ; vm = vm->next) {
if (!qemudIsActiveVM(vm))
continue;
if (vm->stdout != -1) {
fds[fd].fd = vm->stdout;
fds[fd].events = POLLIN | POLLERR | POLLHUP;
fd++;
}
if (vm->stderr != -1) {
fds[fd].fd = vm->stderr;
fds[fd].events = POLLIN | POLLERR | POLLHUP;
fd++;
}
}
for (client = server->clients ; client ; client = client->next) {
fds[fd].fd = client->fd;
if (!client->tls) {
/* Refuse to read more from client if tx is pending to
rate limit */
if (client->mode == QEMUD_MODE_TX_PACKET)
fds[fd].events = POLLOUT | POLLERR | POLLHUP;
else
fds[fd].events = POLLIN | POLLERR | POLLHUP;
} else {
qemudDebug ("direction = %s",
client->direction ? "WRITE" : "READ");
fds[fd].events = client->direction ? POLLOUT : POLLIN;
fds[fd].events |= POLLERR | POLLHUP;
}
fd++;
}
for (sock = server->sockets ; sock ; sock = sock->next) {
fds[fd].fd = sock->fd;
fds[fd].events = POLLIN;
fd++;
}
} }
static int qemudOneLoop(struct qemud_server *server) { static int qemudOneLoop(struct qemud_server *server) {
int nfds = server->nsockets + server->nclients + server->nvmfds + 1; /* server->sigread */
struct pollfd fds[nfds];
int thistimeout = -1;
int ret;
sig_atomic_t errors; sig_atomic_t errors;
/* If we have no clients or vms, then timeout after if (virEventRunOnce() < 0)
30 seconds, letting daemon exit */
if (timeout > 0 &&
!server->nclients &&
!server->nactivevms)
thistimeout = timeout;
qemudPreparePoll(server, fds);
retry:
if ((ret = poll(fds, nfds, thistimeout * 1000)) < 0) {
if (errno == EINTR) {
goto retry;
}
qemudLog(QEMUD_ERR, "Error polling on file descriptors: %s",
strerror(errno));
return -1; return -1;
}
/* Must have timed out */
if (ret == 0) {
qemudLog(QEMUD_INFO, "Timed out while polling on file descriptors");
return -1;
}
/* Check for any signal handling errors and log them. */ /* Check for any signal handling errors and log them. */
errors = sig_errors; errors = sig_errors;
@ -2542,8 +2562,7 @@ static int qemudOneLoop(struct qemud_server *server) {
return -1; return -1;
} }
if (qemudDispatchPoll(server, fds) < 0) qemudCleanupInactive(server);
return -1;
return 0; return 0;
} }
@ -2941,6 +2960,15 @@ int main(int argc, char **argv) {
goto error2; goto error2;
} }
if (virEventAddHandle(sigpipe[0],
POLLIN,
qemudDispatchSignalEvent,
server) < 0) {
qemudLog(QEMUD_ERR, "Failed to register callback for signal pipe");
ret = 3;
goto error2;
}
qemudRunLoop(server); qemudRunLoop(server);
qemudCleanup(server); qemudCleanup(server);