[libvirt] Re: [PATCH 07/12] Domain Events - remote driver

[Ben Guthro <bguthro virtualiron com>]

[PATCH 07/12] Domain Events - remote driver
Deliver local callbacks in response to remote events

 remote_internal.c |  276 +++++++++++++++++++++++++++++++++++++++++++++++++++++-
 1 file changed, 271 insertions(+), 5 deletions(-)

diff --git a/src/remote_internal.c b/src/remote_internal.c
index 35b7b4b..8875674 100644
--- a/src/remote_internal.c
+++ b/src/remote_internal.c
@@ -34,6 +34,7 @@
 #include <signal.h>
 #include <sys/types.h>
 #include <sys/stat.h>
+#include <sys/poll.h>
 #include <fcntl.h>
 
 #ifdef HAVE_SYS_WAIT_H
@@ -73,6 +74,7 @@
 #include "remote_protocol.h"
 #include "memory.h"
 #include "util.h"
+#include "event.h"
 
 /* Per-connection private data. */
 #define MAGIC 999               /* private_data->magic if OK */
@@ -97,6 +99,13 @@ struct private_data {
     unsigned int saslDecodedLength;
     unsigned int saslDecodedOffset;
 #endif
+    /* The list of domain event callbacks */
+    virDomainEventCallbackListPtr callbackList;
+    /* The queue of domain events generated
+       during a call / response rpc          */
+    virDomainEventQueuePtr domainEvents;
+    /* Timer for flushing domainEvents queue */
+    int eventFlushTimer;
 };
 
 #define GET_PRIVATE(conn,retcode)                                       \
@@ -156,7 +165,10 @@ static void make_nonnull_domain (remote_nonnull_domain *dom_dst, virDomainPtr do
 static void make_nonnull_network (remote_nonnull_network *net_dst, virNetworkPtr net_src);
 static void make_nonnull_storage_pool (remote_nonnull_storage_pool *pool_dst, virStoragePoolPtr vol_src);
 static void make_nonnull_storage_vol (remote_nonnull_storage_vol *vol_dst, virStorageVolPtr vol_src);
-
+void remoteDomainEventFired(int fd, virEventHandleType event, void *data);
+static void remoteDomainProcessEvent(virConnectPtr conn, XDR *xdr);
+static void remoteDomainQueueEvent(virConnectPtr conn, XDR *xdr);
+void remoteDomainEventQueueFlush(int timer, void *opaque);
 /*----------------------------------------------------------------------*/
 
 /* Helper functions for remoteOpen. */
@@ -680,6 +692,36 @@ doRemoteOpen (virConnectPtr conn,
               (xdrproc_t) xdr_void, (char *) NULL) == -1)
         goto failed;
 
+    if(VIR_ALLOC(priv->callbackList)<0) {
+        error(conn, VIR_ERR_INVALID_ARG, _("Error allocating callbacks list"));
+        goto failed;
+    }
+
+    if(VIR_ALLOC(priv->domainEvents)<0) {
+        error(conn, VIR_ERR_INVALID_ARG, _("Error allocating domainEvents"));
+        goto failed;
+    }
+
+    DEBUG0("Adding Handler for remote events");
+    /* Set up a callback to listen on the socket data */
+    if (virEventAddHandle(priv->sock,
+                          VIR_EVENT_HANDLE_READABLE |
+                              VIR_EVENT_HANDLE_ERROR |
+                              VIR_EVENT_HANDLE_HANGUP,
+                          remoteDomainEventFired,
+                          conn) < 0) {
+        DEBUG0("virEventAddHandle failed: No addHandleImpl defined."
+               " continuing without events.");
+    } else {
+
+        DEBUG0("Adding Timeout for remote event queue flushing");
+        if ( (priv->eventFlushTimer = virEventAddTimeout(-1,
+                                               remoteDomainEventQueueFlush,
+                                               conn)) < 0) {
+            DEBUG0("virEventAddTimeout failed: No addTimeoutImpl defined. "
+                    "continuing without events.");
+        }
+    }
     /* Successful. */
     retcode = VIR_DRV_OPEN_SUCCESS;
 
@@ -1101,6 +1143,11 @@ doRemoteClose (virConnectPtr conn, struct private_data *priv)
               (xdrproc_t) xdr_void, (char *) NULL) == -1)
         return -1;
 
+    /* Remove handle for remote events */
+    virEventRemoveHandle(priv->sock);
+    /* Remove timout */
+    virEventRemoveTimeout(priv->eventFlushTimer);
+
     /* Close socket. */
     if (priv->uses_tls && priv->session) {
         gnutls_bye (priv->session, GNUTLS_SHUT_RDWR);
@@ -1132,6 +1179,12 @@ doRemoteClose (virConnectPtr conn, struct private_data *priv)
     /* Free private data. */
     priv->magic = DEAD;
 
+    /* Free callback list */
+    virDomainEventCallbackListFree(priv->callbackList);
+
+    /* Free queued events */
+    virDomainEventQueueFree(priv->domainEvents);
+
     return 0;
 }
 
@@ -4288,6 +4341,52 @@ remoteAuthPolkit (virConnectPtr conn, struct private_data *priv, int in_open,
     return 0;
 }
 #endif /* HAVE_POLKIT */
+/*----------------------------------------------------------------------*/
+
+static int remoteDomainEventRegister (virConnectPtr conn,
+                               void *callback ATTRIBUTE_UNUSED,
+                               void *opaque ATTRIBUTE_UNUSED)
+{
+    struct private_data *priv = conn->privateData;
+
+    if (virDomainEventCallbackListAdd(conn, priv->callbackList,
+                                  callback, opaque) < 0) {
+         error (conn, VIR_ERR_RPC, _("adding cb to list"));
+         return -1;
+    }
+
+    if ( priv->callbackList->count == 1 ) {
+        /* Tell the server when we are the first callback deregistering */
+        if (call (conn, priv, 0, REMOTE_PROC_DOMAIN_EVENTS_REGISTER,
+                (xdrproc_t) xdr_void, (char *) NULL,
+                (xdrproc_t) xdr_void, (char *) NULL) == -1)
+            return -1;
+    }
+
+    return 0;
+}
+
+static int remoteDomainEventDeregister (virConnectPtr conn,
+                                 void *callback ATTRIBUTE_UNUSED)
+{
+    struct private_data *priv = conn->privateData;
+
+    if (virDomainEventCallbackListRemove(conn, priv->callbackList,
+                                  callback) < 0) {
+         error (conn, VIR_ERR_RPC, _("removing cb fron list"));
+         return -1;
+    }
+
+    if ( priv->callbackList->count == 0 ) {
+        /* Tell the server when we are the last callback deregistering */
+        if (call (conn, priv, 0, REMOTE_PROC_DOMAIN_EVENTS_DEREGISTER,
+                (xdrproc_t) xdr_void, (char *) NULL,
+                (xdrproc_t) xdr_void, (char *) NULL) == -1)
+            return -1;
+    }
+
+    return 0;
+}
 
 /*----------------------------------------------------------------------*/
 
@@ -4367,6 +4466,7 @@ call (virConnectPtr conn, struct private_data *priv,
         really_write (conn, priv, flags & REMOTE_CALL_IN_OPEN, buffer, len-4) == -1)
         return -1;
 
+retry_read:
     /* Read and deserialise length word. */
     if (really_read (conn, priv, flags & REMOTE_CALL_IN_OPEN, buffer2, sizeof buffer2) == -1)
         return -1;
@@ -4418,10 +4518,20 @@ call (virConnectPtr conn, struct private_data *priv,
         return -1;
     }
 
-    /* If we extend the server to actually send asynchronous messages, then
-     * we'll need to change this so that it can recognise an asynch
-     * message being received at this point.
-     */
+    if (hdr.proc == REMOTE_PROC_DOMAIN_EVENT &&
+        hdr.direction == REMOTE_MESSAGE) {
+        /* An async message has come in while we were waiting for the
+         * response. Process it to pull it off the wire, and try again
+         */
+        DEBUG0("Encountered an event while waiting for a response");
+
+        remoteDomainQueueEvent(conn, &xdr);
+        virEventUpdateTimeout(priv->eventFlushTimer, 0);
+
+        DEBUG0("Retrying read");
+        xdr_destroy (&xdr);
+        goto retry_read;
+    }
     if (hdr.proc != proc_nr) {
         __virRaiseError (flags & REMOTE_CALL_IN_OPEN ? NULL : conn,
                          NULL, NULL, VIR_FROM_REMOTE,
@@ -4872,6 +4982,8 @@ static virDriver driver = {
     .domainMemoryPeek = remoteDomainMemoryPeek,
     .nodeGetCellsFreeMemory = remoteNodeGetCellsFreeMemory,
     .getFreeMemory = remoteNodeGetFreeMemory,
+    .domainEventRegister = remoteDomainEventRegister,
+    .domainEventDeregister = remoteDomainEventDeregister,
 };
 
 static virNetworkDriver network_driver = {
@@ -4957,3 +5069,157 @@ remoteRegister (void)
 
     return 0;
 }
+
+/**
+ * remoteDomainReadEvent
+ *
+ * Read the event data off the wire
+ */
+static int
+remoteDomainReadEvent(virConnectPtr conn, XDR *xdr,
+                      virDomainPtr *dom, int *event)
+{
+    remote_domain_event_ret ret;
+    memset (&ret, 0, sizeof ret);
+
+    /* unmarshall parameters, and process it*/
+    if (! xdr_remote_domain_event_ret(xdr, &ret) ) {
+        error (conn, VIR_ERR_RPC,
+               _("remoteDomainProcessEvent: unmarshalling ret"));
+        return -1;
+    }
+
+    *dom = get_nonnull_domain(conn,ret.dom);
+    *event = ret.event;
+
+    return 0;
+}
+
+static void
+remoteDomainProcessEvent(virConnectPtr conn, XDR *xdr)
+{
+    virDomainPtr dom;
+    int event,i;
+    struct private_data *priv = conn->privateData;
+
+    if(!remoteDomainReadEvent(conn, xdr, &dom, &event)) {
+        DEBUG0("Calling domain event callbacks (no queue)");
+        for(i=0 ; i < priv->callbackList->count ; i++) {
+           if( priv->callbackList->callbacks[i] )
+               priv->callbackList->callbacks[i]->cb(conn, dom, event,
+                                     priv->callbackList->callbacks[i]->opaque);
+        }
+    }
+}
+
+static void
+remoteDomainQueueEvent(virConnectPtr conn, XDR *xdr)
+{
+    virDomainPtr dom;
+    int event;
+    struct private_data *priv = conn->privateData;
+
+    if(!remoteDomainReadEvent(conn, xdr, &dom, &event))
+    {
+        if( virDomainEventCallbackQueuePush(priv->domainEvents,
+                                            dom, event) < 0 ) {
+            DEBUG("%s", "Error adding event to queue");
+        }
+    }
+}
+
+/** remoteDomainEventFired:
+ *
+ * The callback for monitoring the remote socket
+ * for event data
+ */
+void
+remoteDomainEventFired(int fd ATTRIBUTE_UNUSED,
+                       virEventHandleType event,
+                       void *opaque)
+{
+    char buffer[REMOTE_MESSAGE_MAX];
+    char buffer2[4];
+    struct remote_message_header hdr;
+    XDR xdr;
+    int len;
+
+    virConnectPtr        conn = opaque;
+    struct private_data *priv = conn->privateData;
+
+    DEBUG("%s : Event fired %d %X", __FUNCTION__, event, event);
+
+    if (event & (VIR_EVENT_HANDLE_HANGUP | VIR_EVENT_HANDLE_ERROR)) {
+         DEBUG("%s : VIR_EVENT_HANDLE_HANGUP or "
+               "VIR_EVENT_HANDLE_ERROR encountered", __FUNCTION__);
+         virEventRemoveHandle(fd);
+         return;
+    }
+
+    /* Read and deserialise length word. */
+    if (really_read (conn, priv, 0, buffer2, sizeof buffer2) == -1)
+        return;
+
+    xdrmem_create (&xdr, buffer2, sizeof buffer2, XDR_DECODE);
+    if (!xdr_int (&xdr, &len)) {
+        error (conn, VIR_ERR_RPC, _("xdr_int (length word, reply)"));
+        return;
+    }
+    xdr_destroy (&xdr);
+
+    /* Length includes length word - adjust to real length to read. */
+    len -= 4;
+
+    if (len < 0 || len > REMOTE_MESSAGE_MAX) {
+        error (conn, VIR_ERR_RPC, _("packet received from server too large"));
+        return;
+    }
+
+    /* Read reply header and what follows (either a ret or an error). */
+    if (really_read (conn, priv, 0, buffer, len) == -1) {
+        error (conn, VIR_ERR_RPC, _("error reading buffer from memory"));
+        return;
+    }
+
+    /* Deserialise reply header. */
+    xdrmem_create (&xdr, buffer, len, XDR_DECODE);
+    if (!xdr_remote_message_header (&xdr, &hdr)) {
+        error (conn, VIR_ERR_RPC, _("invalid header in event firing"));
+        return;
+    }
+
+    if (hdr.proc == REMOTE_PROC_DOMAIN_EVENT &&
+        hdr.direction == REMOTE_MESSAGE) {
+        DEBUG0("Encountered an async event");
+        remoteDomainProcessEvent(conn, &xdr);
+    } else {
+        DEBUG0("invalid proc in event firing");
+        error (conn, VIR_ERR_RPC, _("invalid proc in event firing"));
+    }
+}
+
+void
+remoteDomainEventQueueFlush(int timer ATTRIBUTE_UNUSED, void *opaque)
+{
+    int i;
+    virDomainEventPtr domEvent;
+    void *user_data = NULL;
+    virConnectPtr conn = opaque;
+    struct private_data *priv = conn->privateData;
+
+    while( (domEvent = virDomainEventCallbackQueuePop(priv->domainEvents)) ) {
+        DEBUG("   Flushing %p", domEvent);
+        for (i=0 ; i < priv->callbackList->count ; i++) {
+           if( priv->callbackList->callbacks[i] ) {
+               user_data = priv->callbackList->callbacks[i]->opaque;
+               priv->callbackList->callbacks[i]->cb(domEvent->dom->conn,
+                                                    domEvent->dom,
+                                                    domEvent->event,
+                                                    user_data);
+           }
+        }
+        VIR_FREE(domEvent);
+    }
+
+    virEventUpdateTimeout(priv->eventFlushTimer, -1);
+}