[PATCH 1/2] rpc: client: fix race on stream error and stream creation

[Nikolay Shirokovskiy <nshirokovskiy@xxxxxxxxxxxxx>]

Message of API call that creates stream and stream itself have
same rpc serial. This can lead to issues. If stream got error
immediately after creation then notification can be delivered
before API call reply arrived. This is possible because the
reply and the error message are sent from different threads
(rpc worker thread and event loop thread respectively). As
we don't check for message type in virNetClientCallCompleteAllWaitingReply
we complete the API call which leads to API call error [1]
as there is no actual reply. Later when reply arrives connection
will be closed due to protocol error (see check in virNetClientCallDispatchReply).
Let's fix virNetClientCallCompleteAllWaitingReply.

Queue inspection on arriving VIR_NET_CONTINUE message in
virNetClientCallDispatchStream is safe because there we check for
status field also. Queue inspection on arriving VIR_NET_OK is safe
too as this message can not arrive before we call virFinishAbort(Finish)
which is not possible before API call that creates streams returns.
But just to be sure let's add checking message type in these places too.

[1] error: internal error: Unexpected message type 0

Signed-off-by: Nikolay Shirokovskiy <nshirokovskiy@xxxxxxxxxxxxx>
---
 src/rpc/virnetclient.c | 9 ++++++---
 1 file changed, 6 insertions(+), 3 deletions(-)

diff --git a/src/rpc/virnetclient.c b/src/rpc/virnetclient.c
index 64855fb..0393587 100644
--- a/src/rpc/virnetclient.c
+++ b/src/rpc/virnetclient.c
@@ -1163,7 +1163,8 @@ static void virNetClientCallCompleteAllWaitingReply(virNetClientPtr client)
     virNetClientCallPtr call;
 
     for (call = client->waitDispatch; call; call = call->next) {
-        if (call->msg->header.prog == client->msg.header.prog &&
+        if (call->msg->header.type == VIR_NET_STREAM &&
+            call->msg->header.prog == client->msg.header.prog &&
             call->msg->header.vers == client->msg.header.vers &&
             call->msg->header.serial == client->msg.header.serial &&
             call->expectReply)
@@ -1207,7 +1208,8 @@ static int virNetClientCallDispatchStream(virNetClientPtr client)
 
         /* Find oldest dummy message waiting for incoming data. */
         for (thecall = client->waitDispatch; thecall; thecall = thecall->next) {
-            if (thecall->msg->header.prog == client->msg.header.prog &&
+            if (thecall->msg->header.type == VIR_NET_STREAM &&
+                thecall->msg->header.prog == client->msg.header.prog &&
                 thecall->msg->header.vers == client->msg.header.vers &&
                 thecall->msg->header.serial == client->msg.header.serial &&
                 thecall->expectReply &&
@@ -1225,7 +1227,8 @@ static int virNetClientCallDispatchStream(virNetClientPtr client)
     case VIR_NET_OK:
         /* Find oldest abort/finish message. */
         for (thecall = client->waitDispatch; thecall; thecall = thecall->next) {
-            if (thecall->msg->header.prog == client->msg.header.prog &&
+            if (thecall->msg->header.type == VIR_NET_STREAM &&
+                thecall->msg->header.prog == client->msg.header.prog &&
                 thecall->msg->header.vers == client->msg.header.vers &&
                 thecall->msg->header.serial == client->msg.header.serial &&
                 thecall->expectReply &&
-- 
1.8.3.1