Allow concurrent processing of RPC calls in daemon

ChangeLog

@@ -1,3 +1,12 @@
+Tue Jan 20 19:24:53 GMT 2009 Daniel P. Berrange <berrange@redhat.com>
+
+	* qemud/qemud.c, qemud/qemud.h, qemud/remote.c: Allow the
+	processing of multiple concurrent RPC calls per client
+	connection.
+	* qemud/libvirtd.conf, qemud/libvirtd.aug,
+	qemud/test_libvirtd.aug: Add config param for controlling
+	number of requests per client.
+
 Tue Jan 20 18:16:53 GMT 2009 Daniel P. Berrange <berrange@redhat.com>
 
 	* src/xm_internal.c: Fix 2 misleading comments & potential

qemud/libvirtd.aug

@@ -53,6 +53,8 @@ module Libvirtd =
    let processing_entry = int_entry "min_workers"
                         | int_entry "max_workers"
                         | int_entry "max_clients"
+                        | int_entry "max_requests"
+                        | int_entry "max_client_requests"
 
    let logging_entry = int_entry "log_level"
                      | str_entry "log_filters"

qemud/libvirtd.conf

@@ -247,6 +247,22 @@
 #min_workers = 5
 #max_workers = 20
 
+# Total global limit on concurrent RPC calls. Should be
+# at least as large as max_workers. Beyond this, RPC requests
+# will be read into memory and queued. This directly impact
+# memory usage, currently each request requires 256 KB of
+# memory. So by default upto 5 MB of memory is used
+#
+# XXX this isn't actually enforced yet, only the per-client
+# limit is used so far
+#max_requests = 20
+
+# Limit on concurrent requests from a single client
+# connection. To avoid one client monopolizing the server
+# this should be a small fraction of the global max_requests
+# and max_workers parameter
+#max_client_requests = 5
+
 #################################################################
 #
 # Logging controls

qemud/qemud.h

@@ -65,15 +65,6 @@
 
 #define qemudDebug DEBUG
 
-enum qemud_mode {
-    QEMUD_MODE_RX_HEADER,       /* Receiving the fixed length RPC header data */
-    QEMUD_MODE_RX_PAYLOAD,      /* Receiving the variable length RPC payload data */
-    QEMUD_MODE_WAIT_DISPATCH,   /* Message received, waiting for worker to process */
-    QEMUD_MODE_IN_DISPATCH,     /* RPC call being processed */
-    QEMUD_MODE_TX_PACKET,       /* Transmitting reply to RPC call */
-    QEMUD_MODE_TLS_HANDSHAKE,   /* Performing TLS handshake */
-};
-
 /* Whether we're passing reads & writes through a sasl SSF */
 enum qemud_sasl_ssf {
     QEMUD_SASL_SSF_NONE = 0,
@@ -87,6 +78,16 @@ enum qemud_sock_type {
     QEMUD_SOCK_TYPE_TLS = 2,
 };
 
+struct qemud_client_message {
+    char buffer [REMOTE_MESSAGE_MAX + REMOTE_MESSAGE_HEADER_XDR_LEN];
+    unsigned int bufferLength;
+    unsigned int bufferOffset;
+
+    int async : 1;
+
+    struct qemud_client_message *next;
+};
+
 /* Stores the per-client connection state */
 struct qemud_client {
     virMutex lock;
@@ -97,7 +98,6 @@ struct qemud_client {
     int watch;
     int readonly:1;
     int closing:1;
-    enum qemud_mode mode;
 
     struct sockaddr_storage addr;
     socklen_t addrlen;
@@ -105,6 +105,7 @@ struct qemud_client {
     int type; /* qemud_sock_type */
     gnutls_session_t tlssession;
     int auth;
+    int handshake : 1; /* If we're in progress for TLS handshake */
 #if HAVE_SASL
     sasl_conn_t *saslconn;
     int saslSSF;
@@ -117,12 +118,20 @@ struct qemud_client {
     char *saslUsername;
 #endif
 
-    unsigned int incomingSerial;
+    /* Count of meages in 'dx' or 'tx' queue
-    unsigned int outgoingSerial;
+     * ie RPC calls in progress. Does not count
-
+     * async events which are not used for
-    char buffer [REMOTE_MESSAGE_MAX];
+     * throttling calculations */
-    unsigned int bufferLength;
+    int nrequests;
-    unsigned int bufferOffset;
+    /* Zero or one messages being received. Zero if
+     * nrequests >= max_clients and throttling */
+    struct qemud_client_message *rx;
+    /* Zero or many messages waiting for a worker
+     * to process them */
+    struct qemud_client_message *dx;
+    /* Zero or many messages waiting for transmit
+     * back to client, including async events */
+    struct qemud_client_message *tx;
 
     /* This is only valid if a remote open call has been made on this
      * connection, otherwise it will be NULL.  Also if remote close is
@@ -181,16 +190,20 @@ void qemudLog(int priority, const char *fmt, ...)
 int qemudSetCloseExec(int fd);
 int qemudSetNonBlock(int fd);
 
-unsigned int
+int
 remoteDispatchClientRequest (struct qemud_server *server,
-                             struct qemud_client *client);
+                             struct qemud_client *client,
+                             struct qemud_client_message *req);
 
-void qemudDispatchClientWrite(struct qemud_server *server,
+int qemudRegisterClientEvent(struct qemud_server *server,
-                             struct qemud_client *client);
+                             struct qemud_client *client,
+                             int update);
 
-#if HAVE_POLKIT
+void qemudDispatchClientFailure(struct qemud_client *client);
-int qemudGetSocketIdentity(int fd, uid_t *uid, pid_t *pid);
+
-#endif
+void
+qemudClientMessageQueuePush(struct qemud_client_message **queue,
+                            struct qemud_client_message *msg);
 
 int remoteRelayDomainEvent (virConnectPtr conn ATTRIBUTE_UNUSED,
                             virDomainPtr dom,
@@ -198,4 +211,9 @@ int remoteRelayDomainEvent (virConnectPtr conn ATTRIBUTE_UNUSED,
                             int detail,
                             void *opaque);
 
+
+#if HAVE_POLKIT
+int qemudGetSocketIdentity(int fd, uid_t *uid, pid_t *pid);
+#endif
+
 #endif

qemud/remote.c

@@ -111,6 +111,7 @@ static const dispatch_data const dispatch_table[] = {
 /* Prototypes */
 static void
 remoteDispatchDomainEventSend (struct qemud_client *client,
+                               struct qemud_client_message *msg,
                                virDomainPtr dom,
                                int event,
                                int detail);
@@ -219,9 +220,10 @@ remoteDispatchConnError (remote_error *rerr,
  * Server object is unlocked
  * Client object is locked
  */
-unsigned int
+int
 remoteDispatchClientRequest (struct qemud_server *server,
-                             struct qemud_client *client)
+                             struct qemud_client *client,
+                             struct qemud_client_message *msg)
 {
     XDR xdr;
     remote_message_header req, rep;
@@ -229,7 +231,8 @@ remoteDispatchClientRequest (struct qemud_server *server,
     dispatch_args args;
     dispatch_ret ret;
     const dispatch_data *data = NULL;
-    int rv = -1, len;
+    int rv = -1;
+    unsigned int len;
     virConnectPtr conn = NULL;
 
     memset(&args, 0, sizeof args);
@@ -237,7 +240,10 @@ remoteDispatchClientRequest (struct qemud_server *server,
     memset(&rerr, 0, sizeof rerr);
 
     /* Parse the header. */
-    xdrmem_create (&xdr, client->buffer, client->bufferLength, XDR_DECODE);
+    xdrmem_create (&xdr,
+                   msg->buffer + REMOTE_MESSAGE_HEADER_XDR_LEN,
+                   msg->bufferLength - REMOTE_MESSAGE_HEADER_XDR_LEN,
+                   XDR_DECODE);
 
     if (!xdr_remote_message_header (&xdr, &req))
         goto fatal_error;
@@ -333,10 +339,10 @@ rpc_error:
     rep.status = rv < 0 ? REMOTE_ERROR : REMOTE_OK;
 
     /* Serialise the return header. */
-    xdrmem_create (&xdr, client->buffer, sizeof client->buffer, XDR_ENCODE);
+    xdrmem_create (&xdr, msg->buffer, sizeof msg->buffer, XDR_ENCODE);
 
     len = 0; /* We'll come back and write this later. */
-    if (!xdr_int (&xdr, &len)) {
+    if (!xdr_u_int (&xdr, &len)) {
         if (rv == 0) xdr_free (data->ret_filter, (char*)&ret);
         goto fatal_error;
     }
@@ -364,17 +370,21 @@ rpc_error:
     if (xdr_setpos (&xdr, 0) == 0)
         goto fatal_error;
 
-    if (!xdr_int (&xdr, &len))
+    if (!xdr_u_int (&xdr, &len))
         goto fatal_error;
 
     xdr_destroy (&xdr);
-    return len;
+
+    msg->bufferLength = len;
+    msg->bufferOffset = 0;
+
+    return 0;
 
 fatal_error:
     /* Seriously bad stuff happened, so we'll kill off this client
        and not send back any RPC error */
     xdr_destroy (&xdr);
-    return 0;
+    return -1;
 }
 
 int remoteRelayDomainEvent (virConnectPtr conn ATTRIBUTE_UNUSED,
@@ -386,9 +396,20 @@ int remoteRelayDomainEvent (virConnectPtr conn ATTRIBUTE_UNUSED,
     struct qemud_client *client = opaque;
     REMOTE_DEBUG("Relaying domain event %d %d", event, detail);
 
-    if(client) {
+    if (client) {
-        remoteDispatchDomainEventSend (client, dom, event, detail);
+        struct qemud_client_message *ev;
-        qemudDispatchClientWrite(client->server,client);
+
+        if (VIR_ALLOC(ev) < 0)
+            return -1;
+
+        virMutexLock(&client->lock);
+
+        remoteDispatchDomainEventSend (client, ev, dom, event, detail);
+
+        if (qemudRegisterClientEvent(client->server, client, 1) < 0)
+            qemudDispatchClientFailure(client);
+
+        virMutexUnlock(&client->lock);
     }
     return 0;
 }
@@ -4202,13 +4223,14 @@ remoteDispatchDomainEventsDeregister (struct qemud_server *server ATTRIBUTE_UNUS
 
 static void
 remoteDispatchDomainEventSend (struct qemud_client *client,
+                               struct qemud_client_message *msg,
                                virDomainPtr dom,
                                int event,
                                int detail)
 {
     remote_message_header rep;
     XDR xdr;
-    int len;
+    unsigned int len;
     remote_domain_event_ret data;
 
     if (!client)
@@ -4222,11 +4244,11 @@ remoteDispatchDomainEventSend (struct qemud_client *client,
     rep.status = REMOTE_OK;
 
     /* Serialise the return header and event. */
-    xdrmem_create (&xdr, client->buffer, sizeof client->buffer, XDR_ENCODE);
+    xdrmem_create (&xdr, msg->buffer, sizeof msg->buffer, XDR_ENCODE);
 
     len = 0; /* We'll come back and write this later. */
-    if (!xdr_int (&xdr, &len)) {
+    if (!xdr_u_int (&xdr, &len)) {
-        /*remoteDispatchError (client, NULL, "%s", _("xdr_int failed (1)"));*/
+        /*remoteDispatchError (client, NULL, "%s", _("xdr_u_int failed (1)"));*/
         xdr_destroy (&xdr);
         return;
     }
@@ -4254,8 +4276,8 @@ remoteDispatchDomainEventSend (struct qemud_client *client,
         return;
     }
 
-    if (!xdr_int (&xdr, &len)) {
+    if (!xdr_u_int (&xdr, &len)) {
-        /*remoteDispatchError (client, NULL, "%s", _("xdr_int failed (2)"));*/
+        /*remoteDispatchError (client, NULL, "%s", _("xdr_u_int failed (2)"));*/
         xdr_destroy (&xdr);
         return;
     }
@@ -4263,9 +4285,10 @@ remoteDispatchDomainEventSend (struct qemud_client *client,
     xdr_destroy (&xdr);
 
     /* Send it. */
-    client->mode = QEMUD_MODE_TX_PACKET;
+    msg->async = 1;
-    client->bufferLength = len;
+    msg->bufferLength = len;
-    client->bufferOffset = 0;
+    msg->bufferOffset = 0;
+    qemudClientMessageQueuePush(&client->tx, msg);
 }
 
 /*----- Helpers. -----*/

qemud/test_libvirtd.aug

@@ -246,6 +246,19 @@ max_clients = 20
 # of clients allowed
 min_workers = 5
 max_workers = 20
+
+# Total global limit on concurrent RPC calls. Should be
+# at least as large as max_workers. Beyond this, RPC requests
+# will be read into memory and queued. This directly impact
+# memory usage, currently each request requires 256 KB of
+# memory. So by default upto 5 MB of memory is used
+max_requests = 20
+
+# Limit on concurrent requests from a single client
+# connection. To avoid one client monopolizing the server
+# this should be a small fraction of the global max_requests
+# and max_workers parameter
+max_client_requests = 5
 "
 
    test Libvirtd.lns get conf =
@@ -499,3 +512,16 @@ max_workers = 20
         { "#comment" = "of clients allowed"}
         { "min_workers" = "5" }
         { "max_workers" = "20" }
+	{ "#empty" }
+        { "#comment" = "Total global limit on concurrent RPC calls. Should be" }
+        { "#comment" = "at least as large as max_workers. Beyond this, RPC requests" }
+        { "#comment" = "will be read into memory and queued. This directly impact" }
+        { "#comment" = "memory usage, currently each request requires 256 KB of" }
+        { "#comment" = "memory. So by default upto 5 MB of memory is used" }
+        { "max_requests" = "20" }
+	{ "#empty" }
+        { "#comment" = "Limit on concurrent requests from a single client" }
+        { "#comment" = "connection. To avoid one client monopolizing the server" }
+        { "#comment" = "this should be a small fraction of the global max_requests" }
+        { "#comment" = "and max_workers parameter" }
+        { "max_client_requests" = "5" }

src/remote_internal.c

@@ -5663,13 +5663,13 @@ prepareCall(virConnectPtr conn,
     /* Length must include the length word itself (always encoded in
      * 4 bytes as per RFC 4506).
      */
-    rv->bufferLength += 4;
+    rv->bufferLength += REMOTE_MESSAGE_HEADER_XDR_LEN;
 
     /* Encode the length word. */
-    xdrmem_create (&xdr, rv->buffer, 4, XDR_ENCODE);
+    xdrmem_create (&xdr, rv->buffer, REMOTE_MESSAGE_HEADER_XDR_LEN, XDR_ENCODE);
-    if (!xdr_int (&xdr, (int *)&rv->bufferLength)) {
+    if (!xdr_u_int (&xdr, &rv->bufferLength)) {
         error (flags & REMOTE_CALL_IN_OPEN ? NULL : conn, VIR_ERR_RPC,
-               _("xdr_int (length word)"));
+               _("xdr_u_int (length word)"));
         goto error;
     }
     xdr_destroy (&xdr);
@@ -5965,20 +5965,26 @@ static int
 processCallRecvLen(virConnectPtr conn, struct private_data *priv,
                    int in_open) {
     XDR xdr;
-    int len;
+    unsigned int len;
 
     xdrmem_create (&xdr, priv->buffer, priv->bufferLength, XDR_DECODE);
-    if (!xdr_int (&xdr, &len)) {
+    if (!xdr_u_int (&xdr, &len)) {
         error (in_open ? NULL : conn,
-               VIR_ERR_RPC, _("xdr_int (length word, reply)"));
+               VIR_ERR_RPC, _("xdr_u_int (length word, reply)"));
         return -1;
     }
     xdr_destroy (&xdr);
 
-    /* Length includes length word - adjust to real length to read. */
+    if (len < REMOTE_MESSAGE_HEADER_XDR_LEN) {
-    len -= 4;
+        error (in_open ? NULL : conn,
+               VIR_ERR_RPC, _("packet received from server too small"));
+        return -1;
+    }
 
-    if (len < 0 || len > REMOTE_MESSAGE_MAX) {
+    /* Length includes length word - adjust to real length to read. */
+    len -= REMOTE_MESSAGE_HEADER_XDR_LEN;
+
+    if (len > REMOTE_MESSAGE_MAX) {
         error (in_open ? NULL : conn,
                VIR_ERR_RPC, _("packet received from server too large"));
         return -1;