From 8cb0d0893fa902b8bd587c0ada663e2935544200 Mon Sep 17 00:00:00 2001
From: Jiri Denemark <jdenemar@redhat.com>
Date: Tue, 12 Jun 2012 08:41:59 +0200
Subject: [PATCH] client rpc: Use event loop for writing

Normally, when every call has a thread associated with it, the thread
may get the buck and be in charge of sending all calls until its own
call is done. When we introduced non-blocking calls, we had to add
special handling of new non-blocking calls. This patch uses event loop
to send data if there is no thread to get the buck so that any
non-blocking calls left in the queue are properly sent without having to
handle them specially. It also avoids adding even more cruft to client
IO loop in the following patches.

With this change in, non-blocking calls may see unpredictable delays in
delivery when the client has no event loop registered. However, the only
non-blocking calls we have are keepalives and we already require event
loop for them, which makes this a non-issue until someone introduces new
non-blocking calls.
(cherry picked from commit 9e747e5c5079a5aead2f248fc22ff658bd0180be)
---
 src/rpc/virnetclient.c | 35 +++++++++++++++++++++++++++++++----
 1 file changed, 31 insertions(+), 4 deletions(-)

diff --git a/src/rpc/virnetclient.c b/src/rpc/virnetclient.c
index 0943c11201..7f6437f2b6 100644
--- a/src/rpc/virnetclient.c
+++ b/src/rpc/virnetclient.c
@@ -1458,12 +1458,30 @@ error:
 }
 
 
+static bool
+virNetClientIOUpdateEvents(virNetClientCallPtr call,
+                           void *opaque)
+{
+    int *events = opaque;
+
+    if (call->mode == VIR_NET_CLIENT_MODE_WAIT_TX)
+        *events |= VIR_EVENT_HANDLE_WRITABLE;
+
+    return false;
+}
+
+
 static void virNetClientIOUpdateCallback(virNetClientPtr client,
                                          bool enableCallback)
 {
     int events = 0;
-    if (enableCallback)
+
+    if (enableCallback) {
         events |= VIR_EVENT_HANDLE_READABLE;
+        virNetClientCallMatchPredicate(client->waitDispatch,
+                                       virNetClientIOUpdateEvents,
+                                       &events);
+    }
 
     virNetSocketUpdateIOCallback(client->sock, events);
 }
@@ -1658,11 +1676,20 @@ void virNetClientIncomingEvent(virNetSocketPtr sock,
         goto done;
     }
 
-    if (virNetClientIOHandleInput(client) < 0) {
-        VIR_WARN("Something went wrong during async message processing");
-        virNetSocketRemoveIOCallback(sock);
+    if (events & VIR_EVENT_HANDLE_WRITABLE) {
+        if (virNetClientIOHandleOutput(client) < 0)
+            virNetSocketRemoveIOCallback(sock);
     }
 
+    if (events & VIR_EVENT_HANDLE_READABLE) {
+        if (virNetClientIOHandleInput(client) < 0)
+            virNetSocketRemoveIOCallback(sock);
+    }
+
+    /* Remove completed calls or signal their threads. */
+    virNetClientCallRemovePredicate(&client->waitDispatch,
+                                    virNetClientIOEventLoopRemoveDone,
+                                    NULL);
 done:
     virNetClientUnlock(client);
 }