rpc: Do not use timer for sending keepalive responses

When a libvirt API is called from the main event loop (which seems to be
common in event-based glib apps), the client IO loop would properly
handle keepalive requests sent by a server but will not actually send
them because the main event loop is blocked with the API. This patch
gets rid of response timer and the thread which is processing keepalive
requests is also responsible for queueing responses for delivery.

1 files changed, 34 insertions, 1 deletions

diff --git a/src/rpc/virnetclient.c b/src/rpc/virnetclient.c
index b956f6e9b..48c6a5dc5 100644
--- a/src/rpc/virnetclient.c
+++ b/src/rpc/virnetclient.c
@@ -109,6 +109,8 @@ struct _virNetClient {
 
 static void virNetClientIOEventLoopPassTheBuck(virNetClientPtr client,
                                                virNetClientCallPtr thiscall);
+static int virNetClientQueueNonBlocking(virNetClientPtr client,
+                                        virNetMessagePtr msg);
 
 
 static void virNetClientLock(virNetClientPtr client)
@@ -937,14 +939,22 @@ static int virNetClientCallDispatchStream(virNetClientPtr client)
 static int
 virNetClientCallDispatch(virNetClientPtr client)
 {
+    virNetMessagePtr response = NULL;
+
     PROBE(RPC_CLIENT_MSG_RX,
           "client=%p len=%zu prog=%u vers=%u proc=%u type=%u status=%u serial=%u",
           client, client->msg.bufferLength,
           client->msg.header.prog, client->msg.header.vers, client->msg.header.proc,
           client->msg.header.type, client->msg.header.status, client->msg.header.serial);
 
-    if (virKeepAliveCheckMessage(client->keepalive, &client->msg))
+    if (virKeepAliveCheckMessage(client->keepalive, &client->msg, &response)) {
+        if (response &&
+            virNetClientQueueNonBlocking(client, response) < 0) {
+            VIR_WARN("Could not queue keepalive response");
+            virNetMessageFree(response);
+        }
         return 0;
+    }
 
     switch (client->msg.header.type) {
     case VIR_NET_REPLY: /* Normal RPC replies */
@@ -1637,6 +1647,8 @@ void virNetClientIncomingEvent(virNetSocketPtr sock,
     virNetClientCallRemovePredicate(&client->waitDispatch,
                                     virNetClientIOEventLoopRemoveDone,
                                     NULL);
+    virNetClientIOUpdateCallback(client, true);
+
 done:
     virNetClientUnlock(client);
 }
@@ -1696,6 +1708,27 @@ error:
 }
 
 
+static int
+virNetClientQueueNonBlocking(virNetClientPtr client,
+                             virNetMessagePtr msg)
+{
+    virNetClientCallPtr call;
+
+    PROBE(RPC_CLIENT_MSG_TX_QUEUE,
+          "client=%p len=%zu prog=%u vers=%u proc=%u"
+          " type=%u status=%u serial=%u",
+          client, msg->bufferLength,
+          msg->header.prog, msg->header.vers, msg->header.proc,
+          msg->header.type, msg->header.status, msg->header.serial);
+
+    if (!(call = virNetClientCallNew(msg, false, true)))
+        return -1;
+
+    virNetClientCallQueue(&client->waitDispatch, call);
+    return 0;
+}
+
+
 /*
  * Returns 1 if the call was queued and will be completed later (only
  * for nonBlock==true), 0 if the call was completed and -1 on error.