The keepalive program has three procedures: ADVERTISE, PING, and PONG. All are used only in asynchronous messages and the sender doesn't wait for any reply. However, the party which receives PING messages is supposed to react by sending PONG message the other party, but no explicit binding between PING and PONG messages is made. ADVERTISE is sent by a client to indicate it supports keepalive protocol. Server is not allowed to send any keepalive message until it sees ADVERTISE. --- .gitignore | 1 + src/Makefile.am | 12 +++++++++--- src/rpc/virkeepaliveprotocol.x | 8 ++++++++ 3 files changed, 18 insertions(+), 3 deletions(-) create mode 100644 src/rpc/virkeepaliveprotocol.x diff --git a/.gitignore b/.gitignore index 41fa50f..3859fab 100644 --- a/.gitignore +++ b/.gitignore @@ -65,6 +65,7 @@ /src/locking/qemu-sanlock.conf /src/remote/*_client_bodies.h /src/remote/*_protocol.[ch] +/src/rpc/virkeepaliveprotocol.[ch] /src/rpc/virnetprotocol.[ch] /src/util/virkeymaps.h /tests/*.log diff --git a/src/Makefile.am b/src/Makefile.am index 738ee91..ff890e1 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -266,7 +266,8 @@ PDWTAGS = \ PROTOCOL_STRUCTS = \ $(srcdir)/remote_protocol-structs \ $(srcdir)/qemu_protocol-structs \ - $(srcdir)/virnetprotocol-structs + $(srcdir)/virnetprotocol-structs \ + $(srcdir)/virkeepaliveprotocol-structs if WITH_REMOTE # The .o file that pdwtags parses is created as a side effect of running # libtool; but from make's perspective we depend on the .lo file. @@ -274,6 +275,7 @@ $(srcdir)/%_protocol-structs: libvirt_driver_remote_la-%_protocol.lo $(PDWTAGS) $(srcdir)/virnetprotocol-structs: libvirt_net_rpc_la-virnetprotocol.lo $(PDWTAGS) +$(srcdir)/virkeepaliveprotocol-structs: libvirt_net_rpc_la-virkeepaliveprotocol.lo else !WITH_REMOTE # These generated files must live in git, because they cannot be re-generated # when configured --without-remote. @@ -1307,12 +1309,15 @@ noinst_LTLIBRARIES += \ EXTRA_DIST += \ rpc/virnetprotocol.x \ + rpc/virkeepaliveprotocol.x \ rpc/gendispatch.pl \ rpc/genprotocol.pl VIR_NET_RPC_GENERATED = \ $(srcdir)/rpc/virnetprotocol.h \ - $(srcdir)/rpc/virnetprotocol.c + $(srcdir)/rpc/virnetprotocol.c \ + $(srcdir)/rpc/virkeepaliveprotocol.h \ + $(srcdir)/rpc/virkeepaliveprotocol.c BUILT_SOURCES += $(VIR_NET_RPC_GENERATED) @@ -1320,7 +1325,8 @@ libvirt_net_rpc_la_SOURCES = \ rpc/virnetmessage.h rpc/virnetmessage.c \ rpc/virnetprotocol.h rpc/virnetprotocol.c \ rpc/virnetsocket.h rpc/virnetsocket.c \ - rpc/virnettlscontext.h rpc/virnettlscontext.c + rpc/virnettlscontext.h rpc/virnettlscontext.c \ + rpc/virkeepaliveprotocol.h rpc/virkeepaliveprotocol.c if HAVE_SASL libvirt_net_rpc_la_SOURCES += \ rpc/virnetsaslcontext.h rpc/virnetsaslcontext.c diff --git a/src/rpc/virkeepaliveprotocol.x b/src/rpc/virkeepaliveprotocol.x new file mode 100644 index 0000000..326ea78 --- /dev/null +++ b/src/rpc/virkeepaliveprotocol.x @@ -0,0 +1,8 @@ +const KEEPALIVE_PROGRAM = 0x6b656570; +const KEEPALIVE_VERSION = 1; + +enum keepalive_procedure { + KEEPALIVE_PROC_ADVERTISE = 1, + KEEPALIVE_PROC_PING = 2, + KEEPALIVE_PROC_PONG = 3 +}; -- 1.7.6.1

These APIs are used by both client and server RPC layer to handle processing of keepalive messages. --- po/POTFILES.in | 1 + src/Makefile.am | 3 +- src/rpc/virkeepalive.c | 464 ++++++++++++++++++++++++++++++++++++++++++++++++ src/rpc/virkeepalive.h | 58 ++++++ 4 files changed, 525 insertions(+), 1 deletions(-) create mode 100644 src/rpc/virkeepalive.c create mode 100644 src/rpc/virkeepalive.h diff --git a/po/POTFILES.in b/po/POTFILES.in index 5ce35ae..71254dd 100644 --- a/po/POTFILES.in +++ b/po/POTFILES.in @@ -72,6 +72,7 @@ src/qemu/qemu_monitor_text.c src/qemu/qemu_process.c src/remote/remote_client_bodies.h src/remote/remote_driver.c +src/rpc/virkeepalive.c src/rpc/virnetclient.c src/rpc/virnetclientprogram.c src/rpc/virnetclientstream.c diff --git a/src/Makefile.am b/src/Makefile.am index ff890e1..d983d28 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -1326,7 +1326,8 @@ libvirt_net_rpc_la_SOURCES = \ rpc/virnetprotocol.h rpc/virnetprotocol.c \ rpc/virnetsocket.h rpc/virnetsocket.c \ rpc/virnettlscontext.h rpc/virnettlscontext.c \ - rpc/virkeepaliveprotocol.h rpc/virkeepaliveprotocol.c + rpc/virkeepaliveprotocol.h rpc/virkeepaliveprotocol.c \ + rpc/virkeepalive.h rpc/virkeepalive.c if HAVE_SASL libvirt_net_rpc_la_SOURCES += \ rpc/virnetsaslcontext.h rpc/virnetsaslcontext.c diff --git a/src/rpc/virkeepalive.c b/src/rpc/virkeepalive.c new file mode 100644 index 0000000..5536b61 --- /dev/null +++ b/src/rpc/virkeepalive.c @@ -0,0 +1,464 @@ +/* + * virkeepalive.c: keepalive handling + * + * Copyright (C) 2011 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: Jiri Denemark <jdenemar@redhat.com> + */ + +#include <config.h> + +#include "memory.h" +#include "threads.h" +#include "virfile.h" +#include "logging.h" +#include "util.h" +#include "virterror_internal.h" +#include "virnetsocket.h" +#include "virkeepaliveprotocol.h" +#include "virkeepalive.h" + +#define VIR_FROM_THIS VIR_FROM_RPC +#define virNetError(code, ...) \ + virReportErrorHelper(VIR_FROM_THIS, code, __FILE__, \ + __FUNCTION__, __LINE__, __VA_ARGS__) + +struct _virKeepAlive { + int refs; + virMutex lock; + + int interval; + unsigned int count; + unsigned int countToDeath; + time_t lastPacketReceived; + bool advertised; + int timer; + + virNetMessagePtr response; + int responseTimer; + + virKeepAliveSendFunc sendCB; + virKeepAliveDeadFunc deadCB; + virKeepAliveFreeFunc freeCB; + void *client; +}; + + +static void +virKeepAliveLock(virKeepAlivePtr ka) +{ + virMutexLock(&ka->lock); +} + +static void +virKeepAliveUnlock(virKeepAlivePtr ka) +{ + virMutexUnlock(&ka->lock); +} + + +static virNetMessagePtr +virKeepAliveMessage(int proc) +{ + virNetMessagePtr msg; + + if (!(msg = virNetMessageNew(false))) + return NULL; + + msg->header.prog = KEEPALIVE_PROGRAM; + msg->header.vers = KEEPALIVE_VERSION; + msg->header.type = VIR_NET_MESSAGE; + msg->header.proc = proc; + + if (virNetMessageEncodeHeader(msg) < 0 || + virNetMessageEncodePayloadEmpty(msg) < 0) { + virNetMessageFree(msg); + return NULL; + } + + return msg; +} + + +static int +virKeepAliveSend(virKeepAlivePtr ka, virNetMessagePtr msg) +{ + int ret; + const char *proc; + void *client = ka->client; + virKeepAliveSendFunc sendCB = ka->sendCB; + + switch (msg->header.proc) { + case KEEPALIVE_PROC_ADVERTISE: + proc = "advertisement"; + break; + case KEEPALIVE_PROC_PING: + proc = "request"; + break; + case KEEPALIVE_PROC_PONG: + proc = "response"; + break; + } + + VIR_DEBUG("Sending keepalive %s to client %p", proc, ka->client); + + ka->refs++; + virKeepAliveUnlock(ka); + + if ((ret = sendCB(client, msg)) < 0) { + VIR_WARN("Failed to send keepalive %s to client %p", proc, client); + virNetMessageFree(msg); + } + + virKeepAliveLock(ka); + ka->refs--; + + return ret; +} + + +int +virKeepAliveAdvertise(virKeepAlivePtr ka) +{ + virNetMessagePtr msg; + int ret = -1; + + virKeepAliveLock(ka); + + VIR_DEBUG("Advertising keepalive support to client %p", ka->client); + + if (!(msg = virKeepAliveMessage(KEEPALIVE_PROC_ADVERTISE))) { + VIR_WARN("Failed to generate keepalive advertisement"); + goto cleanup; + } + + if ((ret = virKeepAliveSend(ka, msg)) == 0) + ka->advertised = true; + +cleanup: + virKeepAliveUnlock(ka); + return ret; +} + + +static void +virKeepAliveScheduleResponse(virKeepAlivePtr ka) +{ + if (ka->responseTimer == -1) + return; + + VIR_DEBUG("Scheduling keepalive response to client %p", ka->client); + + if (!ka->response && + !(ka->response = virKeepAliveMessage(KEEPALIVE_PROC_PONG))) { + VIR_WARN("Failed to generate keepalive response"); + return; + } + + virEventUpdateTimeout(ka->responseTimer, 0); +} + + +static void +virKeepAliveTimer(int timer ATTRIBUTE_UNUSED, void *opaque) +{ + virKeepAlivePtr ka = opaque; + time_t now = time(NULL); + + virKeepAliveLock(ka); + + VIR_DEBUG("ka=%p, client=%p, countToDeath=%d, lastPacketReceived=%lds ago", + ka, ka->client, ka->countToDeath, now - ka->lastPacketReceived); + + if (now - ka->lastPacketReceived < ka->interval - 1) { + int timeout = ka->interval - (now - ka->lastPacketReceived); + virEventUpdateTimeout(ka->timer, timeout * 1000); + goto cleanup; + } + + if (ka->countToDeath == 0) { + virKeepAliveDeadFunc deadCB = ka->deadCB; + void *client = ka->client; + + VIR_WARN("No response from client %p after %d keepalive messages in" + " %d seconds", + ka->client, + ka->count, + (int) (now - ka->lastPacketReceived)); + ka->refs++; + virKeepAliveUnlock(ka); + deadCB(client); + virKeepAliveLock(ka); + ka->refs--; + } else { + virNetMessagePtr msg; + + ka->countToDeath--; + if (!(msg = virKeepAliveMessage(KEEPALIVE_PROC_PING))) + VIR_WARN("Failed to generate keepalive request"); + else + ignore_value(virKeepAliveSend(ka, msg)); + virEventUpdateTimeout(ka->timer, ka->interval * 1000); + } + +cleanup: + virKeepAliveUnlock(ka); +} + + +static void +virKeepAliveResponseTimer(int timer ATTRIBUTE_UNUSED, void *opaque) +{ + virKeepAlivePtr ka = opaque; + virNetMessagePtr msg; + + virKeepAliveLock(ka); + + VIR_DEBUG("ka=%p, client=%p, response=%p", + ka, ka->client, ka->response); + + if (ka->response) { + msg = ka->response; + ka->response = NULL; + ignore_value(virKeepAliveSend(ka, msg)); + } + + virEventUpdateTimeout(ka->responseTimer, ka->response ? 0 : -1); + + virKeepAliveUnlock(ka); +} + + +static void +virKeepAliveTimerFree(void *opaque) +{ + virKeepAliveFree(opaque); +} + + +virKeepAlivePtr +virKeepAliveNew(int interval, + unsigned int count, + void *client, + virKeepAliveSendFunc sendCB, + virKeepAliveDeadFunc deadCB, + virKeepAliveFreeFunc freeCB) +{ + virKeepAlivePtr ka; + + VIR_DEBUG("client=%p, interval=%d, count=%u", client, interval, count); + + if (VIR_ALLOC(ka) < 0) { + virReportOOMError(); + return NULL; + } + + if (virMutexInit(&ka->lock) < 0) { + VIR_FREE(ka); + return NULL; + } + + ka->refs = 1; + ka->interval = interval; + ka->count = count; + ka->countToDeath = count; + ka->timer = -1; + ka->client = client; + ka->sendCB = sendCB; + ka->deadCB = deadCB; + ka->freeCB = freeCB; + + ka->responseTimer = virEventAddTimeout(-1, virKeepAliveResponseTimer, + ka, virKeepAliveTimerFree); + if (ka->responseTimer < 0) { + virKeepAliveFree(ka); + return NULL; + } + /* the timer now has a reference to ka */ + ka->refs++; + + return ka; +} + + +void +virKeepAliveRef(virKeepAlivePtr ka) +{ + virKeepAliveLock(ka); + ka->refs++; + VIR_DEBUG("ka=%p, client=%p, refs=%d", ka, ka->client, ka->refs); + virKeepAliveUnlock(ka); +} + + +void +virKeepAliveFree(virKeepAlivePtr ka) +{ + if (!ka) + return; + + virKeepAliveLock(ka); + VIR_DEBUG("ka=%p, client=%p, refs=%d", ka, ka->client, ka->refs); + if (--ka->refs > 0) { + virKeepAliveUnlock(ka); + return; + } + + virMutexDestroy(&ka->lock); + ka->freeCB(ka->client); + VIR_FREE(ka); +} + + +int +virKeepAliveStart(virKeepAlivePtr ka, + int interval, + unsigned int count) +{ + int ret = -1; + time_t delay; + int timeout; + + VIR_DEBUG("ka=%p, client=%p, interval=%d, count=%u", + ka, ka->client, interval, count); + + virKeepAliveLock(ka); + + if (!ka->advertised) { + virNetError(VIR_ERR_INTERNAL_ERROR, "%s", + _("keepalive support was not advertised to remote party")); + goto cleanup; + } + + if (ka->timer >= 0) { + VIR_DEBUG("Keepalive messages already enabled"); + ret = 0; + goto cleanup; + } + + if (interval > 0) { + if (ka->interval > 0) { + virNetError(VIR_ERR_INTERNAL_ERROR, "%s", + _("keepalive interval already set")); + goto cleanup; + } + ka->interval = interval; + ka->count = count; + ka->countToDeath = count; + } + + if (ka->interval <= 0) { + VIR_DEBUG("Keepalive messages disabled by configuration"); + ret = 0; + goto cleanup; + } + + VIR_DEBUG("Enabling keepalive messages; interval=%d, count=%u", + ka->interval, ka->count); + + delay = time(NULL) - ka->lastPacketReceived; + if (delay > ka->interval) + timeout = 0; + else + timeout = ka->interval - delay; + ka->timer = virEventAddTimeout(timeout * 1000, virKeepAliveTimer, + ka, virKeepAliveTimerFree); + if (ka->timer < 0) + goto cleanup; + + /* the timer now has another reference to this object */ + ka->refs++; + ret = 0; + +cleanup: + virKeepAliveUnlock(ka); + return ret; +} + + +void +virKeepAliveStop(virKeepAlivePtr ka) +{ + VIR_DEBUG("ka=%p, client=%p", ka, ka->client); + + virKeepAliveLock(ka); + if (ka->timer > 0) { + virEventRemoveTimeout(ka->timer); + ka->timer = -1; + } + if (ka->responseTimer > 0) { + virEventRemoveTimeout(ka->responseTimer); + ka->responseTimer = -1; + } + virKeepAliveUnlock(ka); +} + + +bool +virKeepAliveCheckMessage(virKeepAlivePtr ka, + virNetMessagePtr msg) +{ + bool ret = false; + bool start = false; + + VIR_DEBUG("ka=%p, client=%p, msg=%p", + ka, ka ? ka->client : "(null)", msg); + + if (!ka) + return false; + + virKeepAliveLock(ka); + + ka->countToDeath = ka->count; + ka->lastPacketReceived = time(NULL); + + if (msg->header.prog == KEEPALIVE_PROGRAM && + msg->header.vers == KEEPALIVE_VERSION && + msg->header.type == VIR_NET_MESSAGE) { + ret = true; + switch (msg->header.proc) { + case KEEPALIVE_PROC_ADVERTISE: + VIR_DEBUG("Client %p advertises keepalive support", ka->client); + ka->advertised = true; + start = true; + break; + + case KEEPALIVE_PROC_PING: + VIR_DEBUG("Got keepalive request from client %p", ka->client); + virKeepAliveScheduleResponse(ka); + break; + + case KEEPALIVE_PROC_PONG: + VIR_DEBUG("Got keepalive response from client %p", ka->client); + break; + + default: + VIR_DEBUG("Ignoring unknown keepalive message %d from client %p", + msg->header.proc, ka->client); + } + } + + if (ka->timer >= 0) + virEventUpdateTimeout(ka->timer, ka->interval * 1000); + + virKeepAliveUnlock(ka); + + if (start && virKeepAliveStart(ka, 0, 0) < 0) + VIR_WARN("Failed to start keepalive protocol"); + + return ret; +} diff --git a/src/rpc/virkeepalive.h b/src/rpc/virkeepalive.h new file mode 100644 index 0000000..ac96859 --- /dev/null +++ b/src/rpc/virkeepalive.h @@ -0,0 +1,58 @@ +/* + * virkeepalive.h: keepalive handling + * + * Copyright (C) 2011 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: Jiri Denemark <jdenemar@redhat.com> + */ + +#ifndef __VIR_KEEPALIVE_H__ +# define __VIR_KEEPALIVE_H__ + +# include "virnetmessage.h" + +typedef int (*virKeepAliveSendFunc)(void *client, virNetMessagePtr msg); +typedef void (*virKeepAliveDeadFunc)(void *client); +typedef void (*virKeepAliveFreeFunc)(void *client); + +typedef struct _virKeepAlive virKeepAlive; +typedef virKeepAlive *virKeepAlivePtr; + + +virKeepAlivePtr virKeepAliveNew(int interval, + unsigned int count, + void *client, + virKeepAliveSendFunc sendCB, + virKeepAliveDeadFunc deadCB, + virKeepAliveFreeFunc freeCB) + ATTRIBUTE_NONNULL(3) ATTRIBUTE_NONNULL(4) + ATTRIBUTE_NONNULL(5) ATTRIBUTE_NONNULL(6); + +void virKeepAliveRef(virKeepAlivePtr ka); +void virKeepAliveFree(virKeepAlivePtr ka); + +int virKeepAliveAdvertise(virKeepAlivePtr ka); + +int virKeepAliveStart(virKeepAlivePtr ka, + int interval, + unsigned int count); +void virKeepAliveStop(virKeepAlivePtr ka); + +bool virKeepAliveCheckMessage(virKeepAlivePtr ka, + virNetMessagePtr msg); + +#endif /* __VIR_KEEPALIVE_H__ */ -- 1.7.6.1

When a client wants to send a keepalive message it needs to do so in a non-blocking way to avoid blocking its event loop. This patch adds dontBlock flag which says that the call should be processed without blocking. Such calls do not have a thread waiting for the result associated with them. This means, that sending such call fails if no thread is dispatching and writing to the socket would block. In case there is a thread waiting for its (normal) call to finish, sending non-blocking call just pushes it into the queue and lets the dispatching thread send it. The thread which has the buck tries to send all non-blocking calls in the queue in a best effort way---if sending them would block or there's an error on the socket, non-blocking calls are simply removed from the queue and discarded. --- src/rpc/virnetclient.c | 149 ++++++++++++++++++++++++++++++++++++------------ 1 files changed, 113 insertions(+), 36 deletions(-) diff --git a/src/rpc/virnetclient.c b/src/rpc/virnetclient.c index 055361d..7ea9a27 100644 --- a/src/rpc/virnetclient.c +++ b/src/rpc/virnetclient.c @@ -55,6 +55,7 @@ struct _virNetClientCall { virNetMessagePtr msg; bool expectReply; + bool dontBlock; virCond cond; @@ -94,6 +95,11 @@ struct _virNetClient { }; +static int virNetClientSendInternal(virNetClientPtr client, + virNetMessagePtr msg, + bool expectReply, + bool dontBlock); + static void virNetClientLock(virNetClientPtr client) { virMutexLock(&client->lock); @@ -848,6 +854,7 @@ static int virNetClientIOEventLoop(virNetClientPtr client, char ignore; sigset_t oldmask, blockedsigs; int timeout = -1; + bool discardNonBlocking; /* If we have existing SASL decoded data we * don't want to sleep in the poll(), just @@ -865,6 +872,11 @@ static int virNetClientIOEventLoop(virNetClientPtr client, fds[0].events |= POLLIN; if (tmp->mode == VIR_NET_CLIENT_MODE_WAIT_TX) fds[0].events |= POLLOUT; + /* We don't want to sleep in poll if any of the calls is + * non-blocking + */ + if (tmp->dontBlock) + timeout = 0; tmp = tmp->next; } @@ -937,35 +949,63 @@ static int virNetClientIOEventLoop(virNetClientPtr client, goto error; } - /* Iterate through waiting threads and if - * any are complete then tell 'em to wakeup + /* Iterate through waiting calls and + * - remove all completed nonblocking calls + * - remove all nonblocking calls in case poll() would block + * - remove all nonblocking calls if we got error from poll() + * - wake up threads waiting for calls that have been completed */ + discardNonBlocking = ret == 0 || + (fds[0].revents & POLLHUP) || + (fds[0].revents & POLLERR); tmp = client->waitDispatch; prev = NULL; while (tmp) { + virNetClientCallPtr next = tmp->next; + if (tmp != thiscall && - tmp->mode == VIR_NET_CLIENT_MODE_COMPLETE) { + (tmp->mode == VIR_NET_CLIENT_MODE_COMPLETE || + (discardNonBlocking && tmp->dontBlock))) { /* Take them out of the list */ if (prev) prev->next = tmp->next; else client->waitDispatch = tmp->next; - /* And wake them up.... - * ...they won't actually wakeup until - * we release our mutex a short while - * later... - */ - VIR_DEBUG("Waking up sleep %p %p", tmp, client->waitDispatch); - virCondSignal(&tmp->cond); + if (tmp->dontBlock) { + /* tmp is a non-blocking call, no-one is waiting for it so + * we just free it here + */ + if (tmp->mode != VIR_NET_CLIENT_MODE_COMPLETE) { + VIR_DEBUG("Can't finish nonblocking call %p without" + " blocking or error", tmp); + } + virNetMessageFree(tmp->msg); + VIR_FREE(tmp); + } else { + /* And wake them up.... + * ...they won't actually wakeup until + * we release our mutex a short while + * later... + */ + VIR_DEBUG("Waking up sleep %p %p", + tmp, client->waitDispatch); + virCondSignal(&tmp->cond); + } } else { prev = tmp; } - tmp = tmp->next; + tmp = next; } /* Now see if *we* are done */ if (thiscall->mode == VIR_NET_CLIENT_MODE_COMPLETE) { + /* If next call is non-blocking call, we need to process it + * before giving up the buck + */ + if (thiscall->next && thiscall->next->dontBlock) + continue; + /* We're at head of the list already, so * remove us */ @@ -980,14 +1020,18 @@ static int virNetClientIOEventLoop(virNetClientPtr client, return 0; } - if (fds[0].revents & (POLLHUP | POLLERR)) { virNetError(VIR_ERR_INTERNAL_ERROR, "%s", _("received hangup / error event on socket")); goto error; } - } + if (thiscall->dontBlock && discardNonBlocking) { + virNetError(VIR_ERR_INTERNAL_ERROR, "%s", + _("Can't finish nonblocking call without blocking")); + goto error; + } + } error: client->waitDispatch = thiscall->next; @@ -1040,38 +1084,41 @@ static int virNetClientIO(virNetClientPtr client, { int rv = -1; - VIR_DEBUG("Outgoing message prog=%u version=%u serial=%u proc=%d type=%d length=%zu dispatch=%p", + VIR_DEBUG("Outgoing message prog=%u version=%u serial=%u proc=%d type=%d" + " length=%zu dontBlock=%d dispatch=%p", thiscall->msg->header.prog, thiscall->msg->header.vers, thiscall->msg->header.serial, thiscall->msg->header.proc, thiscall->msg->header.type, thiscall->msg->bufferLength, + thiscall->dontBlock, client->waitDispatch); /* Check to see if another thread is dispatching */ if (client->waitDispatch) { - /* Stick ourselves on the end of the wait queue */ - virNetClientCallPtr tmp = client->waitDispatch; + virNetClientCallPtr tmp; char ignore = 1; - while (tmp && tmp->next) - tmp = tmp->next; - if (tmp) - tmp->next = thiscall; - else - client->waitDispatch = thiscall; /* Force other thread to wakeup from poll */ if (safewrite(client->wakeupSendFD, &ignore, sizeof(ignore)) != sizeof(ignore)) { - if (tmp) - tmp->next = NULL; - else - client->waitDispatch = NULL; virReportSystemError(errno, "%s", _("failed to wake up polling thread")); return -1; } + /* Stick ourselves on the end of the wait queue */ + tmp = client->waitDispatch; + while (tmp->next) + tmp = tmp->next; + tmp->next = thiscall; + + if (thiscall->dontBlock) { + VIR_DEBUG("Sending non-blocking call while another thread is" + " dispatching; it will send the call for us"); + return 0; + } + VIR_DEBUG("Going to sleep %p %p", client->waitDispatch, thiscall); /* Go to sleep while other thread is working... */ if (virCondWait(&thiscall->cond, &client->lock) < 0) { @@ -1091,7 +1138,7 @@ static int virNetClientIO(virNetClientPtr client, return -1; } - VIR_DEBUG("Wokeup from sleep %p %p", client->waitDispatch, thiscall); + VIR_DEBUG("Woken up from sleep %p %p", client->waitDispatch, thiscall); /* Two reasons we can be woken up * 1. Other thread has got our reply ready for us * 2. Other thread is all done, and it is our turn to @@ -1181,15 +1228,17 @@ done: } -int virNetClientSend(virNetClientPtr client, - virNetMessagePtr msg, - bool expectReply) +static int +virNetClientSendInternal(virNetClientPtr client, + virNetMessagePtr msg, + bool expectReply, + bool dontBlock) { virNetClientCallPtr call; int ret = -1; if (expectReply && - (msg->header.status == VIR_NET_CONTINUE)) { + (msg->header.status == VIR_NET_CONTINUE || dontBlock)) { virNetError(VIR_ERR_INTERNAL_ERROR, "%s", _("Attempt to send an asynchronous message with a synchronous reply")); return -1; @@ -1202,10 +1251,15 @@ int virNetClientSend(virNetClientPtr client, virNetClientLock(client); - if (virCondInit(&call->cond) < 0) { - virNetError(VIR_ERR_INTERNAL_ERROR, "%s", - _("cannot initialize condition variable")); - goto cleanup; + /* We don't need call->cond for non-blocking calls since there's no + * thread to be woken up anyway + */ + if (!dontBlock) { + if (virCondInit(&call->cond) < 0) { + virNetError(VIR_ERR_INTERNAL_ERROR, "%s", + _("cannot initialize condition variable")); + goto cleanup; + } } if (msg->bufferLength) @@ -1214,12 +1268,35 @@ int virNetClientSend(virNetClientPtr client, call->mode = VIR_NET_CLIENT_MODE_WAIT_RX; call->msg = msg; call->expectReply = expectReply; + call->dontBlock = dontBlock; ret = virNetClientIO(client, call); cleanup: ignore_value(virCondDestroy(&call->cond)); - VIR_FREE(call); + if (ret != 0) { + VIR_FREE(call); + } else if (dontBlock) { + /* Only free the call if it was completed since otherwise it was just + * queued up and will be processed later. + */ + if (call->mode == VIR_NET_CLIENT_MODE_COMPLETE) { + /* We need to free the message as well since no-one is waiting for + * it. + */ + virNetMessageFree(msg); + VIR_FREE(call); + } + } else { + VIR_FREE(call); + } virNetClientUnlock(client); return ret; } + +int virNetClientSend(virNetClientPtr client, + virNetMessagePtr msg, + bool expectReply) +{ + return virNetClientSendInternal(client, msg, expectReply, false); +} -- 1.7.6.1

--- src/remote/remote_driver.c | 30 +++++++++++++ src/rpc/virnetclient.c | 102 ++++++++++++++++++++++++++++++++++++++++++- src/rpc/virnetclient.h | 5 ++ 3 files changed, 134 insertions(+), 3 deletions(-) diff --git a/src/remote/remote_driver.c b/src/remote/remote_driver.c index 1217d94..e218c40 100644 --- a/src/remote/remote_driver.c +++ b/src/remote/remote_driver.c @@ -4082,6 +4082,34 @@ done: } +static int +remoteAllowKeepAlive(virConnectPtr conn) +{ + struct private_data *priv = conn->privateData; + int ret = -1; + + remoteDriverLock(priv); + ret = virNetClientKeepAliveAdvertise(priv->client); + remoteDriverUnlock(priv); + + return ret; +} + + +static int +remoteStartKeepAlive(virConnectPtr conn, int interval, unsigned int count) +{ + struct private_data *priv = conn->privateData; + int ret = -1; + + remoteDriverLock(priv); + ret = virNetClientKeepAliveStart(priv->client, interval, count); + remoteDriverUnlock(priv); + + return ret; +} + + #include "remote_client_bodies.h" #include "qemu_client_bodies.h" @@ -4430,6 +4458,8 @@ static virDriver remote_driver = { .domainGetBlockJobInfo = remoteDomainGetBlockJobInfo, /* 0.9.4 */ .domainBlockJobSetSpeed = remoteDomainBlockJobSetSpeed, /* 0.9.4 */ .domainBlockPull = remoteDomainBlockPull, /* 0.9.4 */ + .allowKeepAlive = remoteAllowKeepAlive, /* 0.9.7 */ + .startKeepAlive = remoteStartKeepAlive, /* 0.9.7 */ }; static virNetworkDriver network_driver = { diff --git a/src/rpc/virnetclient.c b/src/rpc/virnetclient.c index 45b0edb..dee5059 100644 --- a/src/rpc/virnetclient.c +++ b/src/rpc/virnetclient.c @@ -29,6 +29,7 @@ #include "virnetclient.h" #include "virnetsocket.h" +#include "virkeepalive.h" #include "memory.h" #include "threads.h" #include "virfile.h" @@ -93,11 +94,12 @@ struct _virNetClient { size_t nstreams; virNetClientStreamPtr *streams; + virKeepAlivePtr keepalive; bool wantClose; }; -void virNetClientRequestClose(virNetClientPtr client); +static void virNetClientRequestClose(virNetClientPtr client); static int virNetClientSendInternal(virNetClientPtr client, virNetMessagePtr msg, @@ -127,11 +129,71 @@ static void virNetClientEventFree(void *opaque) virNetClientFree(client); } +int +virNetClientKeepAliveAdvertise(virNetClientPtr client) +{ + virKeepAlivePtr ka; + int ret = -1; + + virNetClientLock(client); + if ((ka = client->keepalive)) + virKeepAliveRef(ka); + virNetClientUnlock(client); + + if (ka) { + ret = virKeepAliveAdvertise(ka); + virKeepAliveFree(ka); + } else { + virNetError(VIR_ERR_INTERNAL_ERROR, "%s", + _("the caller doesn't support keepalive protocol;" + " perhaps it's missing event loop implementation")); + } + + return ret; +} + +int +virNetClientKeepAliveStart(virNetClientPtr client, + int interval, + unsigned int count) +{ + int ret = -1; + + virNetClientLock(client); + + if (!client->keepalive) { + virNetError(VIR_ERR_INTERNAL_ERROR, "%s", + _("the caller doesn't support keepalive protocol;" + " perhaps it's missing event loop implementation")); + goto cleanup; + } + + ret = virKeepAliveStart(client->keepalive, interval, count); + +cleanup: + virNetClientUnlock(client); + return ret; +} + +static void +virNetClientKeepAliveDeadCB(void *opaque) +{ + virNetClientRequestClose(opaque); +} + +static int +virNetClientKeepAliveSendCB(void *opaque, + virNetMessagePtr msg) +{ + return virNetClientSendInternal(opaque, msg, false, true); +} + static virNetClientPtr virNetClientNew(virNetSocketPtr sock, const char *hostname) { virNetClientPtr client = NULL; int wakeupFD[2] = { -1, -1 }; + virKeepAlivePtr ka = NULL; if (pipe2(wakeupFD, O_CLOEXEC) < 0) { virReportSystemError(errno, "%s", @@ -164,9 +226,21 @@ static virNetClientPtr virNetClientNew(virNetSocketPtr sock, client, virNetClientEventFree) < 0) { client->refs--; - VIR_DEBUG("Failed to add event watch, disabling events"); + VIR_DEBUG("Failed to add event watch, disabling events and support for" + " keepalive messages"); + } else { + /* Keepalive protocol consists of async messages so it can only be used + * if the client supports them */ + if (!(ka = virKeepAliveNew(-1, 0, client, + virNetClientKeepAliveSendCB, + virNetClientKeepAliveDeadCB, + virNetClientEventFree))) + goto error; + /* keepalive object has a reference to client */ + client->refs++; } + client->keepalive = ka; VIR_DEBUG("client=%p refs=%d", client, client->refs); return client; @@ -175,6 +249,10 @@ no_memory: error: VIR_FORCE_CLOSE(wakeupFD[0]); VIR_FORCE_CLOSE(wakeupFD[1]); + if (ka) { + virKeepAliveStop(ka); + virKeepAliveFree(ka); + } virNetClientFree(client); return NULL; } @@ -304,6 +382,8 @@ void virNetClientFree(virNetClientPtr client) static void virNetClientCloseLocked(virNetClientPtr client) { + virKeepAlivePtr ka; + VIR_DEBUG("client=%p, sock=%p", client, client->sock); if (!client->sock) @@ -318,7 +398,20 @@ virNetClientCloseLocked(virNetClientPtr client) virNetSASLSessionFree(client->sasl); client->sasl = NULL; #endif + ka = client->keepalive; + client->keepalive = NULL; client->wantClose = false; + + if (ka) { + client->refs++; + virNetClientUnlock(client); + + virKeepAliveStop(ka); + virKeepAliveFree(ka); + + virNetClientLock(client); + client->refs--; + } } void virNetClientClose(virNetClientPtr client) @@ -331,7 +424,7 @@ void virNetClientClose(virNetClientPtr client) virNetClientUnlock(client); } -void +static void virNetClientRequestClose(virNetClientPtr client) { VIR_DEBUG("client=%p", client); @@ -723,6 +816,9 @@ virNetClientCallDispatch(virNetClientPtr client) client->msg.header.proc, client->msg.header.type, client->msg.header.status, client->msg.header.serial); + if (virKeepAliveCheckMessage(client->keepalive, &client->msg)) + return 0; + switch (client->msg.header.type) { case VIR_NET_REPLY: /* Normal RPC replies */ return virNetClientCallDispatchReply(client); diff --git a/src/rpc/virnetclient.h b/src/rpc/virnetclient.h index 1fabcfd..e337ee0 100644 --- a/src/rpc/virnetclient.h +++ b/src/rpc/virnetclient.h @@ -87,4 +87,9 @@ int virNetClientGetTLSKeySize(virNetClientPtr client); void virNetClientFree(virNetClientPtr client); void virNetClientClose(virNetClientPtr client); +int virNetClientKeepAliveAdvertise(virNetClientPtr client); +int virNetClientKeepAliveStart(virNetClientPtr client, + int interval, + unsigned int count); + #endif /* __VIR_NET_CLIENT_H__ */ -- 1.7.6.1

--- src/esx/esx_driver.c | 18 ++++++++++++++++++ src/hyperv/hyperv_driver.c | 18 ++++++++++++++++++ src/libxl/libxl_driver.c | 8 ++++++++ src/lxc/lxc_driver.c | 7 +++++++ src/openvz/openvz_driver.c | 7 +++++++ src/phyp/phyp_driver.c | 18 ++++++++++++++++++ src/qemu/qemu_driver.c | 6 ++++++ src/remote/remote_driver.c | 18 ++++++++++++++++++ src/rpc/virnetclient.c | 14 ++++++++++++++ src/rpc/virnetclient.h | 1 + src/test/test_driver.c | 6 ++++++ src/uml/uml_driver.c | 7 +++++++ src/vbox/vbox_tmpl.c | 6 ++++++ src/vmware/vmware_driver.c | 7 +++++++ src/xen/xen_driver.c | 8 ++++++++ src/xenapi/xenapi_driver.c | 12 ++++++++++++ 16 files changed, 161 insertions(+), 0 deletions(-) diff --git a/src/esx/esx_driver.c b/src/esx/esx_driver.c index f1102ea..c71f4f8 100644 --- a/src/esx/esx_driver.c +++ b/src/esx/esx_driver.c @@ -4145,6 +4145,23 @@ esxIsSecure(virConnectPtr conn) static int +esxIsAlive(virConnectPtr conn) +{ + esxPrivate *priv = conn->privateData; + + /* XXX we should be able to do something better than this is simple, safe, + * and good enough for now. In worst case, the function will return true + * even though the connection is not alive. + */ + if (priv->host) + return 1; + else + return 0; +} + + + +static int esxDomainIsActive(virDomainPtr domain) { int result = -1; @@ -4813,6 +4830,7 @@ static virDriver esxDriver = { .domainSnapshotCurrent = esxDomainSnapshotCurrent, /* 0.8.0 */ .domainRevertToSnapshot = esxDomainRevertToSnapshot, /* 0.8.0 */ .domainSnapshotDelete = esxDomainSnapshotDelete, /* 0.8.0 */ + .isAlive = esxIsAlive, /* 0.9.7 */ }; diff --git a/src/hyperv/hyperv_driver.c b/src/hyperv/hyperv_driver.c index b022fee..bf8575c 100644 --- a/src/hyperv/hyperv_driver.c +++ b/src/hyperv/hyperv_driver.c @@ -1100,6 +1100,23 @@ hypervIsSecure(virConnectPtr conn) static int +hypervIsAlive(virConnectPtr conn) +{ + hypervPrivate *priv = conn->privateData; + + /* XXX we should be able to do something better than this is simple, safe, + * and good enough for now. In worst case, the function will return true + * even though the connection is not alive. + */ + if (priv->client) + return 1; + else + return 0; +} + + + +static int hypervDomainIsActive(virDomainPtr domain) { int result = -1; @@ -1257,6 +1274,7 @@ static virDriver hypervDriver = { .domainManagedSave = hypervDomainManagedSave, /* 0.9.5 */ .domainHasManagedSaveImage = hypervDomainHasManagedSaveImage, /* 0.9.5 */ .domainManagedSaveRemove = hypervDomainManagedSaveRemove, /* 0.9.5 */ + .isAlive = hypervIsAlive, /* 0.9.7 */ }; diff --git a/src/libxl/libxl_driver.c b/src/libxl/libxl_driver.c index d324632..42503d6 100644 --- a/src/libxl/libxl_driver.c +++ b/src/libxl/libxl_driver.c @@ -3875,6 +3875,13 @@ libxlDomainEventDeregisterAny(virConnectPtr conn, int callbackID) } +static int +libxlIsAlive(virConnectPtr conn ATTRIBUTE_UNUSED) +{ + return 1; +} + + static virDriver libxlDriver = { .no = VIR_DRV_LIBXL, .name = "xenlight", @@ -3948,6 +3955,7 @@ static virDriver libxlDriver = { .domainIsUpdated = libxlDomainIsUpdated, /* 0.9.0 */ .domainEventRegisterAny = libxlDomainEventRegisterAny, /* 0.9.0 */ .domainEventDeregisterAny = libxlDomainEventDeregisterAny, /* 0.9.0 */ + .isAlive = libxlIsAlive, /* 0.9.7 */ }; static virStateDriver libxlStateDriver = { diff --git a/src/lxc/lxc_driver.c b/src/lxc/lxc_driver.c index 6cf7203..28a7f9c 100644 --- a/src/lxc/lxc_driver.c +++ b/src/lxc/lxc_driver.c @@ -185,6 +185,12 @@ static int lxcIsEncrypted(virConnectPtr conn ATTRIBUTE_UNUSED) } +static int lxcIsAlive(virConnectPtr conn ATTRIBUTE_UNUSED) +{ + return 1; +} + + static char *lxcGetCapabilities(virConnectPtr conn) { lxc_driver_t *driver = conn->privateData; char *xml; @@ -2986,6 +2992,7 @@ static virDriver lxcDriver = { .domainEventRegisterAny = lxcDomainEventRegisterAny, /* 0.8.0 */ .domainEventDeregisterAny = lxcDomainEventDeregisterAny, /* 0.8.0 */ .domainOpenConsole = lxcDomainOpenConsole, /* 0.8.6 */ + .isAlive = lxcIsAlive, /* 0.9.7 */ }; static virStateDriver lxcStateDriver = { diff --git a/src/openvz/openvz_driver.c b/src/openvz/openvz_driver.c index 69ff444..8a4e6cf 100644 --- a/src/openvz/openvz_driver.c +++ b/src/openvz/openvz_driver.c @@ -1426,6 +1426,12 @@ static int openvzIsSecure(virConnectPtr conn ATTRIBUTE_UNUSED) { return 1; } +static int +openvzIsAlive(virConnectPtr conn ATTRIBUTE_UNUSED) +{ + return 1; +} + static char *openvzGetCapabilities(virConnectPtr conn) { struct openvz_driver *driver = conn->privateData; char *ret; @@ -1714,6 +1720,7 @@ static virDriver openvzDriver = { .domainIsActive = openvzDomainIsActive, /* 0.7.3 */ .domainIsPersistent = openvzDomainIsPersistent, /* 0.7.3 */ .domainIsUpdated = openvzDomainIsUpdated, /* 0.8.6 */ + .isAlive = openvzIsAlive, /* 0.9.7 */ }; int openvzRegister(void) { diff --git a/src/phyp/phyp_driver.c b/src/phyp/phyp_driver.c index ff16aae..458c9c9 100644 --- a/src/phyp/phyp_driver.c +++ b/src/phyp/phyp_driver.c @@ -1284,6 +1284,23 @@ phypIsSecure(virConnectPtr conn ATTRIBUTE_UNUSED) return 1; } + +static int +phypIsAlive(virConnectPtr conn) +{ + ConnectionData *connection_data = conn->networkPrivateData; + + /* XXX we should be able to do something better but this is simple, safe, + * and good enough for now. In worst case, the function will return true + * even though the connection is not alive. + */ + if (connection_data && connection_data->session) + return 1; + else + return 0; +} + + static int phypIsUpdated(virDomainPtr conn ATTRIBUTE_UNUSED) { @@ -3786,6 +3803,7 @@ static virDriver phypDriver = { .isEncrypted = phypIsEncrypted, /* 0.7.3 */ .isSecure = phypIsSecure, /* 0.7.3 */ .domainIsUpdated = phypIsUpdated, /* 0.8.6 */ + .isAlive = phypIsAlive, /* 0.9.7 */ }; static virStorageDriver phypStorageDriver = { diff --git a/src/qemu/qemu_driver.c b/src/qemu/qemu_driver.c index 0d0bea2..c5fd9eb 100644 --- a/src/qemu/qemu_driver.c +++ b/src/qemu/qemu_driver.c @@ -938,6 +938,11 @@ static int qemuIsEncrypted(virConnectPtr conn ATTRIBUTE_UNUSED) return 0; } +static int qemuIsAlive(virConnectPtr conn ATTRIBUTE_UNUSED) +{ + return 1; +} + static int kvmGetMaxVCPUs(void) { int maxvcpus = 1; @@ -10450,6 +10455,7 @@ static virDriver qemuDriver = { .domainGetBlockJobInfo = qemuDomainGetBlockJobInfo, /* 0.9.4 */ .domainBlockJobSetSpeed = qemuDomainBlockJobSetSpeed, /* 0.9.4 */ .domainBlockPull = qemuDomainBlockPull, /* 0.9.4 */ + .isAlive = qemuIsAlive, /* 0.9.7 */ }; diff --git a/src/remote/remote_driver.c b/src/remote/remote_driver.c index e218c40..9bae3fc 100644 --- a/src/remote/remote_driver.c +++ b/src/remote/remote_driver.c @@ -4110,6 +4110,23 @@ remoteStartKeepAlive(virConnectPtr conn, int interval, unsigned int count) } +static int +remoteIsAlive(virConnectPtr conn) +{ + struct private_data *priv = conn->privateData; + bool ret; + + remoteDriverLock(priv); + ret = virNetClientIsOpen(priv->client); + remoteDriverUnlock(priv); + + if (ret) + return 1; + else + return 0; +} + + #include "remote_client_bodies.h" #include "qemu_client_bodies.h" @@ -4460,6 +4477,7 @@ static virDriver remote_driver = { .domainBlockPull = remoteDomainBlockPull, /* 0.9.4 */ .allowKeepAlive = remoteAllowKeepAlive, /* 0.9.7 */ .startKeepAlive = remoteStartKeepAlive, /* 0.9.7 */ + .isAlive = remoteIsAlive, /* 0.9.7 */ }; static virNetworkDriver network_driver = { diff --git a/src/rpc/virnetclient.c b/src/rpc/virnetclient.c index dee5059..9ccdf08 100644 --- a/src/rpc/virnetclient.c +++ b/src/rpc/virnetclient.c @@ -581,6 +581,20 @@ bool virNetClientIsEncrypted(virNetClientPtr client) } +bool virNetClientIsOpen(virNetClientPtr client) +{ + bool ret; + + if (!client) + return false; + + virNetClientLock(client); + ret = client->sock && !client->wantClose; + virNetClientUnlock(client); + return ret; +} + + int virNetClientAddProgram(virNetClientPtr client, virNetClientProgramPtr prog) { diff --git a/src/rpc/virnetclient.h b/src/rpc/virnetclient.h index e337ee0..5787a7c 100644 --- a/src/rpc/virnetclient.h +++ b/src/rpc/virnetclient.h @@ -78,6 +78,7 @@ int virNetClientSetTLSSession(virNetClientPtr client, virNetTLSContextPtr tls); bool virNetClientIsEncrypted(virNetClientPtr client); +bool virNetClientIsOpen(virNetClientPtr client); const char *virNetClientLocalAddrString(virNetClientPtr client); const char *virNetClientRemoteAddrString(virNetClientPtr client); diff --git a/src/test/test_driver.c b/src/test/test_driver.c index b3e24b4..7f1f8b6 100644 --- a/src/test/test_driver.c +++ b/src/test/test_driver.c @@ -1184,6 +1184,11 @@ static int testIsEncrypted(virConnectPtr conn ATTRIBUTE_UNUSED) return 0; } +static int testIsAlive(virConnectPtr conn ATTRIBUTE_UNUSED) +{ + return 1; +} + static int testGetMaxVCPUs(virConnectPtr conn ATTRIBUTE_UNUSED, const char *type ATTRIBUTE_UNUSED) { @@ -5624,6 +5629,7 @@ static virDriver testDriver = { .domainIsUpdated = testDomainIsUpdated, /* 0.8.6 */ .domainEventRegisterAny = testDomainEventRegisterAny, /* 0.8.0 */ .domainEventDeregisterAny = testDomainEventDeregisterAny, /* 0.8.0 */ + .isAlive = testIsAlive, /* 0.9.7 */ }; static virNetworkDriver testNetworkDriver = { diff --git a/src/uml/uml_driver.c b/src/uml/uml_driver.c index 2b7219a..2ad3084 100644 --- a/src/uml/uml_driver.c +++ b/src/uml/uml_driver.c @@ -1098,6 +1098,12 @@ static int umlIsEncrypted(virConnectPtr conn ATTRIBUTE_UNUSED) } +static int umlIsAlive(virConnectPtr conn ATTRIBUTE_UNUSED) +{ + return 1; +} + + static char *umlGetCapabilities(virConnectPtr conn) { struct uml_driver *driver = (struct uml_driver *)conn->privateData; char *xml; @@ -2462,6 +2468,7 @@ static virDriver umlDriver = { .domainEventRegisterAny = umlDomainEventRegisterAny, /* 0.9.4 */ .domainEventDeregisterAny = umlDomainEventDeregisterAny, /* 0.9.4 */ .domainOpenConsole = umlDomainOpenConsole, /* 0.8.6 */ + .isAlive = umlIsAlive, /* 0.9.7 */ }; static int diff --git a/src/vbox/vbox_tmpl.c b/src/vbox/vbox_tmpl.c index 0e4364f..93ab930 100644 --- a/src/vbox/vbox_tmpl.c +++ b/src/vbox/vbox_tmpl.c @@ -1070,6 +1070,11 @@ static int vboxIsEncrypted(virConnectPtr conn ATTRIBUTE_UNUSED) { return 0; } +static int vboxIsAlive(virConnectPtr conn ATTRIBUTE_UNUSED) +{ + return 1; +} + static int vboxGetMaxVcpus(virConnectPtr conn, const char *type ATTRIBUTE_UNUSED) { VBOX_OBJECT_CHECK(conn, int, -1); PRUint32 maxCPUCount = 0; @@ -8875,6 +8880,7 @@ virDriver NAME(Driver) = { .domainSnapshotCurrent = vboxDomainSnapshotCurrent, /* 0.8.0 */ .domainRevertToSnapshot = vboxDomainRevertToSnapshot, /* 0.8.0 */ .domainSnapshotDelete = vboxDomainSnapshotDelete, /* 0.8.0 */ + .isAlive = vboxIsAlive, /* 0.9.7 */ }; virNetworkDriver NAME(NetworkDriver) = { diff --git a/src/vmware/vmware_driver.c b/src/vmware/vmware_driver.c index b2cfdce..987a7a8 100644 --- a/src/vmware/vmware_driver.c +++ b/src/vmware/vmware_driver.c @@ -958,6 +958,12 @@ vmwareDomainGetState(virDomainPtr dom, return ret; } +static int +vmwareIsAlive(virConnectPtr conn ATTRIBUTE_UNUSED) +{ + return 1; +} + static virDriver vmwareDriver = { .no = VIR_DRV_VMWARE, .name = "VMWARE", @@ -990,6 +996,7 @@ static virDriver vmwareDriver = { .domainUndefineFlags = vmwareDomainUndefineFlags, /* 0.9.4 */ .domainIsActive = vmwareDomainIsActive, /* 0.8.7 */ .domainIsPersistent = vmwareDomainIsPersistent, /* 0.8.7 */ + .isAlive = vmwareIsAlive, /* 0.9.7 */ }; int diff --git a/src/xen/xen_driver.c b/src/xen/xen_driver.c index 9c96fca..52f8601 100644 --- a/src/xen/xen_driver.c +++ b/src/xen/xen_driver.c @@ -516,6 +516,13 @@ xenUnifiedIsSecure(virConnectPtr conn) return ret; } +static int +xenUnifiedIsAlive(virConnectPtr conn ATTRIBUTE_UNUSED) +{ + /* XenD reconnects for each request */ + return 1; +} + int xenUnifiedGetMaxVcpus (virConnectPtr conn, const char *type) { @@ -2259,6 +2266,7 @@ static virDriver xenUnifiedDriver = { .domainEventRegisterAny = xenUnifiedDomainEventRegisterAny, /* 0.8.0 */ .domainEventDeregisterAny = xenUnifiedDomainEventDeregisterAny, /* 0.8.0 */ .domainOpenConsole = xenUnifiedDomainOpenConsole, /* 0.8.6 */ + .isAlive = xenUnifiedIsAlive, /* 0.9.7 */ }; /** diff --git a/src/xenapi/xenapi_driver.c b/src/xenapi/xenapi_driver.c index 80a706a..a5ab98a 100644 --- a/src/xenapi/xenapi_driver.c +++ b/src/xenapi/xenapi_driver.c @@ -1887,6 +1887,17 @@ xenapiNodeGetCellsFreeMemory (virConnectPtr conn, unsigned long long *freeMems, } } +static int +xenapiIsAlive(virConnectPtr conn) +{ + struct _xenapiPrivate *priv = conn->privateData; + + if (priv->session && priv->session->ok) + return 1; + else + return 0; +} + /* The interface which we export upwards to libvirt.c. */ static virDriver xenapiDriver = { .no = VIR_DRV_XENAPI, @@ -1937,6 +1948,7 @@ static virDriver xenapiDriver = { .nodeGetCellsFreeMemory = xenapiNodeGetCellsFreeMemory, /* 0.8.0 */ .nodeGetFreeMemory = xenapiNodeGetFreeMemory, /* 0.8.0 */ .domainIsUpdated = xenapiDomainIsUpdated, /* 0.8.6 */ + .isAlive = xenapiIsAlive, /* 0.9.7 */ }; /** -- 1.7.6.1

--- src/qemu/libvirtd_qemu.aug | 2 ++ src/qemu/qemu.conf | 16 ++++++++++++++++ src/qemu/qemu_conf.c | 11 +++++++++++ src/qemu/qemu_conf.h | 3 +++ src/qemu/qemu_migration.c | 10 ++++++++++ src/qemu/test_libvirtd_qemu.aug | 6 ++++++ 6 files changed, 48 insertions(+), 0 deletions(-) diff --git a/src/qemu/libvirtd_qemu.aug b/src/qemu/libvirtd_qemu.aug index 6c145c7..ad34e42 100644 --- a/src/qemu/libvirtd_qemu.aug +++ b/src/qemu/libvirtd_qemu.aug @@ -52,6 +52,8 @@ module Libvirtd_qemu = | int_entry "max_processes" | str_entry "lock_manager" | int_entry "max_queued" + | int_entry "keepalive_interval" + | int_entry "keepalive_count" (* Each enty in the config is one of the following three ... *) let entry = vnc_entry diff --git a/src/qemu/qemu.conf b/src/qemu/qemu.conf index 4da5d5a..e623bc3 100644 --- a/src/qemu/qemu.conf +++ b/src/qemu/qemu.conf @@ -316,3 +316,19 @@ # Note, that job lock is per domain. # # max_queued = 0 + +################################################################### +# Keepalive protocol: +# This allows qemu driver to detect broken connections to remote +# libvirtd during peer-to-peer migration. A keepalive message is +# sent to the deamon after keepalive_interval seconds of inactivity +# to check if the deamon is still responding; keepalive_count is a +# maximum number of keepalive messages that are allowed to be sent +# to the deamon without getting any response before the connection +# is considered broken. In other words, the connection is +# automatically closed approximately after +# keepalive_interval * (keepalive_count + 1) seconds since the last +# message received from the deamon. +# +#keepalive_interval = 5 +#keepalive_count = 5 diff --git a/src/qemu/qemu_conf.c b/src/qemu/qemu_conf.c index d1bf075..19f24b1 100644 --- a/src/qemu/qemu_conf.c +++ b/src/qemu/qemu_conf.c @@ -118,6 +118,9 @@ int qemudLoadDriverConfig(struct qemud_driver *driver, virLockManagerPluginNew("nop", NULL, 0))) return -1; + driver->keepAliveInterval = 5; + driver->keepAliveCount = 5; + /* Just check the file is readable before opening it, otherwise * libvirt emits an error. */ @@ -462,6 +465,14 @@ int qemudLoadDriverConfig(struct qemud_driver *driver, CHECK_TYPE("max_queued", VIR_CONF_LONG); if (p) driver->max_queued = p->l; + p = virConfGetValue(conf, "keepalive_interval"); + CHECK_TYPE("keepalive_interval", VIR_CONF_LONG); + if (p) driver->keepAliveInterval = p->l; + + p = virConfGetValue(conf, "keepalive_count"); + CHECK_TYPE("keepalive_count", VIR_CONF_LONG); + if (p) driver->keepAliveCount = p->l; + virConfFree (conf); return 0; } diff --git a/src/qemu/qemu_conf.h b/src/qemu/qemu_conf.h index e8b92a4..70c3fda 100644 --- a/src/qemu/qemu_conf.h +++ b/src/qemu/qemu_conf.h @@ -138,6 +138,9 @@ struct qemud_driver { * of guests which will be automatically killed * when the virConnectPtr is closed*/ virHashTablePtr autodestroy; + + int keepAliveInterval; + unsigned int keepAliveCount; }; typedef struct _qemuDomainCmdlineDef qemuDomainCmdlineDef; diff --git a/src/qemu/qemu_migration.c b/src/qemu/qemu_migration.c index 0a5a13d..301d8ba 100644 --- a/src/qemu/qemu_migration.c +++ b/src/qemu/qemu_migration.c @@ -2173,6 +2173,7 @@ static int doPeer2PeerMigrate(struct qemud_driver *driver, virConnectPtr dconn = NULL; bool p2p; virErrorPtr orig_err = NULL; + int rc; VIR_DEBUG("driver=%p, sconn=%p, vm=%p, xmlin=%s, dconnuri=%s, " "uri=%s, flags=%lx, dname=%s, resource=%lu", @@ -2193,6 +2194,15 @@ static int doPeer2PeerMigrate(struct qemud_driver *driver, } qemuDomainObjEnterRemoteWithDriver(driver, vm); + if ((rc = virConnectAllowKeepAlive(dconn)) == 0) { + rc = virConnectStartKeepAlive(dconn, driver->keepAliveInterval, + driver->keepAliveCount); + } + qemuDomainObjExitRemoteWithDriver(driver, vm); + if (rc < 0) + goto cleanup; + + qemuDomainObjEnterRemoteWithDriver(driver, vm); p2p = VIR_DRV_SUPPORTS_FEATURE(dconn->driver, dconn, VIR_DRV_FEATURE_MIGRATION_P2P); /* v3proto reflects whether the caller used Perform3, but with diff --git a/src/qemu/test_libvirtd_qemu.aug b/src/qemu/test_libvirtd_qemu.aug index b1f9114..f7476ae 100644 --- a/src/qemu/test_libvirtd_qemu.aug +++ b/src/qemu/test_libvirtd_qemu.aug @@ -115,6 +115,9 @@ vnc_auto_unix_socket = 1 max_processes = 12345 lock_manager = \"fcntl\" + +keepalive_interval = 1 +keepalive_count = 42 " test Libvirtd_qemu.lns get conf = @@ -240,3 +243,6 @@ lock_manager = \"fcntl\" { "max_processes" = "12345" } { "#empty" } { "lock_manager" = "fcntl" } +{ "#empty" } +{ "keepalive_interval" = "1" } +{ "keepalive_count" = "42" } -- 1.7.6.1