[Date Prev][Date Next]   [Thread Prev][Thread Next]   [Thread Index] [Date Index] [Author Index]

[libvirt] [PATCH 1/4] Refactor incoming message handling to prepare for data stream support



* src/remote_internal.c: Rename processCallRecvMsg to
  processCallDispatch, and move code specific to method replies
  into processCallDispatchReply, and rename processCallAsyncEvent
  to processCallDispatchMessage

Signed-off-by: Daniel P. Berrange <berrange redhat com>
---
 src/remote_internal.c |  136 +++++++++++++++++++++++++++++--------------------
 1 files changed, 80 insertions(+), 56 deletions(-)

diff --git a/src/remote_internal.c b/src/remote_internal.c
index e661daa..4362521 100644
--- a/src/remote_internal.c
+++ b/src/remote_internal.c
@@ -6564,29 +6564,6 @@ processCallRecvSome(virConnectPtr conn, struct private_data *priv,
 }
 
 
-static void
-processCallAsyncEvent(virConnectPtr conn, struct private_data *priv,
-                      int in_open,
-                      remote_message_header *hdr,
-                      XDR *xdr) {
-    /* 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");
-
-    if (in_open) {
-        DEBUG("Ignoring bogus event %d received while in open", hdr->proc);
-        return;
-    }
-
-    if (hdr->proc == REMOTE_PROC_DOMAIN_EVENT) {
-        remoteDomainQueueEvent(conn, xdr);
-        virEventUpdateTimeout(priv->eventFlushTimer, 0);
-    } else {
-        DEBUG("Unexpected event proc %d", hdr->proc);
-    }
-}
-
 static int
 processCallRecvLen(virConnectPtr conn, struct private_data *priv,
                    int in_open) {
@@ -6625,12 +6602,25 @@ processCallRecvLen(virConnectPtr conn, struct private_data *priv,
 
 
 static int
-processCallRecvMsg(virConnectPtr conn, struct private_data *priv,
-                   int in_open) {
+processCallDispatchReply(virConnectPtr conn, struct private_data *priv,
+                         int in_open,
+                         remote_message_header *hdr,
+                         XDR *xdr);
+
+static int
+processCallDispatchMessage(virConnectPtr conn, struct private_data *priv,
+                           int in_open,
+                           remote_message_header *hdr,
+                           XDR *xdr);
+
+
+static int
+processCallDispatch(virConnectPtr conn, struct private_data *priv,
+                    int in_open) {
     XDR xdr;
     struct remote_message_header hdr;
     int len = priv->bufferLength - 4;
-    struct remote_thread_call *thecall;
+    int rv = -1;
 
     /* Deserialise reply header. */
     xdrmem_create (&xdr, priv->buffer + 4, len, XDR_DECODE);
@@ -6658,30 +6648,44 @@ processCallRecvMsg(virConnectPtr conn, struct private_data *priv,
         return -1;
     }
 
-    /* Async events from server need special handling */
-    if (hdr.type == REMOTE_MESSAGE) {
-        processCallAsyncEvent(conn, priv, in_open,
-                              &hdr, &xdr);
-        xdr_destroy(&xdr);
-        return 0;
-    }
+    switch (hdr.type) {
+    case REMOTE_REPLY: /* Normal RPC replies */
+        rv = processCallDispatchReply(conn, priv, in_open,
+                                      &hdr, &xdr);
+        break;
 
-    if (hdr.type != REMOTE_REPLY) {
-        virRaiseError (in_open ? NULL : conn,
-                       NULL, NULL, VIR_FROM_REMOTE,
-                       VIR_ERR_RPC, VIR_ERR_ERROR, NULL, NULL, NULL, 0, 0,
-                       _("got unexpected RPC call %d from server"),
-                       hdr.proc);
-        xdr_destroy(&xdr);
-        return -1;
+    case REMOTE_MESSAGE: /* Async notifications */
+        rv = processCallDispatchMessage(conn, priv, in_open,
+                                        &hdr, &xdr);
+        break;
+
+    default:
+         virRaiseError (in_open ? NULL : conn,
+                        NULL, NULL, VIR_FROM_REMOTE,
+                        VIR_ERR_RPC, VIR_ERR_ERROR, NULL, NULL, NULL, 0, 0,
+                        _("got unexpected RPC call %d from server"),
+                        hdr.proc);
+        rv = -1;
+        break;
     }
 
+    xdr_destroy(&xdr);
+    return rv;
+}
+
+
+static int
+processCallDispatchReply(virConnectPtr conn, struct private_data *priv,
+                         int in_open,
+                         remote_message_header *hdr,
+                         XDR *xdr) {
+    struct remote_thread_call *thecall;
+
     /* Ok, definitely got an RPC reply now find
        out who's been waiting for it */
-
     thecall = priv->waitDispatch;
     while (thecall &&
-           thecall->serial != hdr.serial)
+           thecall->serial != hdr->serial)
         thecall = thecall->next;
 
     if (!thecall) {
@@ -6689,18 +6693,16 @@ processCallRecvMsg(virConnectPtr conn, struct private_data *priv,
                        NULL, NULL, VIR_FROM_REMOTE,
                        VIR_ERR_RPC, VIR_ERR_ERROR, NULL, NULL, NULL, 0, 0,
                        _("no call waiting for reply with serial %d"),
-                       hdr.serial);
-        xdr_destroy(&xdr);
+                       hdr->serial);
         return -1;
     }
 
-    if (hdr.proc != thecall->proc_nr) {
+    if (hdr->proc != thecall->proc_nr) {
         virRaiseError (in_open ? NULL : conn,
                        NULL, NULL, VIR_FROM_REMOTE,
                        VIR_ERR_RPC, VIR_ERR_ERROR, NULL, NULL, NULL, 0, 0,
                        _("unknown procedure (received %x, expected %x)"),
-                       hdr.proc, thecall->proc_nr);
-        xdr_destroy (&xdr);
+                       hdr->proc, thecall->proc_nr);
         return -1;
     }
 
@@ -6708,25 +6710,23 @@ processCallRecvMsg(virConnectPtr conn, struct private_data *priv,
      * structure), or REMOTE_ERROR (and what follows is a remote_error
      * structure).
      */
-    switch (hdr.status) {
+    switch (hdr->status) {
     case REMOTE_OK:
-        if (!(*thecall->ret_filter) (&xdr, thecall->ret)) {
+        if (!(*thecall->ret_filter) (xdr, thecall->ret)) {
             error (in_open ? NULL : conn, VIR_ERR_RPC,
                    _("unmarshalling ret"));
             return -1;
         }
         thecall->mode = REMOTE_MODE_COMPLETE;
-        xdr_destroy (&xdr);
         return 0;
 
     case REMOTE_ERROR:
         memset (&thecall->err, 0, sizeof thecall->err);
-        if (!xdr_remote_error (&xdr, &thecall->err)) {
+        if (!xdr_remote_error (xdr, &thecall->err)) {
             error (in_open ? NULL : conn,
                    VIR_ERR_RPC, _("unmarshalling remote_error"));
             return -1;
         }
-        xdr_destroy (&xdr);
         thecall->mode = REMOTE_MODE_ERROR;
         return 0;
 
@@ -6734,10 +6734,34 @@ processCallRecvMsg(virConnectPtr conn, struct private_data *priv,
         virRaiseError (in_open ? NULL : conn, NULL, NULL, VIR_FROM_REMOTE,
                        VIR_ERR_RPC, VIR_ERR_ERROR, NULL, NULL, NULL, 0, 0,
                        _("unknown status (received %x)"),
-                       hdr.status);
-        xdr_destroy (&xdr);
+                       hdr->status);
+        return -1;
+    }
+}
+
+static int
+processCallDispatchMessage(virConnectPtr conn, struct private_data *priv,
+                           int in_open,
+                           remote_message_header *hdr,
+                           XDR *xdr) {
+    /* 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");
+
+    if (in_open) {
+        DEBUG("Ignoring bogus event %d received while in open", hdr->proc);
         return -1;
     }
+
+    if (hdr->proc == REMOTE_PROC_DOMAIN_EVENT) {
+        remoteDomainQueueEvent(conn, xdr);
+        virEventUpdateTimeout(priv->eventFlushTimer, 0);
+    } else {
+        return -1;
+        DEBUG("Unexpected event proc %d", hdr->proc);
+    }
+    return 0;
 }
 
 
@@ -6770,7 +6794,7 @@ processCallRecv(virConnectPtr conn, struct private_data *priv,
                  * next iteration.
                  */
             } else {
-                ret = processCallRecvMsg(conn, priv, in_open);
+                ret = processCallDispatch(conn, priv, in_open);
                 priv->bufferOffset = priv->bufferLength = 0;
                 /*
                  * We've completed one call, so return even
-- 
1.6.2.5


-- 
|: Red Hat, Engineering, London   -o-   http://people.redhat.com/berrange/ :|
|: http://libvirt.org  -o-  http://virt-manager.org  -o-  http://ovirt.org :|
|: http://autobuild.org       -o-         http://search.cpan.org/~danberr/ :|
|: GnuPG: 7D3B9505  -o-  F3C9 553F A1DA 4AC2 5648 23C1 B3DF F742 7D3B 9505 :|


[Date Prev][Date Next]   [Thread Prev][Thread Next]   [Thread Index] [Date Index] [Author Index]