Fix tracking of RPC messages wrt streams

Commit 2c85644b0b attempted to
fix a problem with tracking RPC messages from streams by doing

-            if (msg->header.type == VIR_NET_REPLY) {
+            if (msg->header.type == VIR_NET_REPLY ||
+                (msg->header.type == VIR_NET_STREAM &&
+                 msg->header.status != VIR_NET_CONTINUE)) {
                 client->nrequests--;

In other words any stream packet, with status NET_OK or NET_ERROR
would cause nrequests to be decremented. This is great if the
packet from from a synchronous virStreamFinish or virStreamAbort
API call, but wildly wrong if from a server initiated abort.
The latter resulted in 'nrequests' being decremented below zero.
This then causes all I/O for that client to be stopped.

Instead of trying to infer whether we need to decrement the
nrequests field, from the message type/status, introduce an
explicit 'bool tracked' field to mark whether the virNetMessagePtr
object is subject to tracking.

Also add a virNetMessageClear function to allow a message
contents to be cleared out, without adversely impacting the
'tracked' field as a naive memset() would do

* src/rpc/virnetmessage.c, src/rpc/virnetmessage.h: Add
  a 'bool tracked' field and virNetMessageClear() API
* daemon/remote.c, daemon/stream.c, src/rpc/virnetclientprogram.c,
  src/rpc/virnetclientstream.c, src/rpc/virnetserverclient.c,
  src/rpc/virnetserverprogram.c: Switch over to use
  virNetMessageClear() and pass in the 'bool tracked' value
  when creating messages.

daemon/remote.c

@@ -2495,7 +2495,7 @@ remoteDispatchDomainEventSend(virNetServerClientPtr client,
 {
     virNetMessagePtr msg;
 
-    if (!(msg = virNetMessageNew()))
+    if (!(msg = virNetMessageNew(false)))
         goto cleanup;
 
     msg->header.prog = virNetServerProgramGetID(program);

daemon/stream.c

@@ -207,7 +207,7 @@ daemonStreamEvent(virStreamPtr st, int events, void *opaque)
             virNetError(VIR_ERR_RPC,
                         "%s", _("stream had I/O failure"));
 
-        msg = virNetMessageNew();
+        msg = virNetMessageNew(false);
         if (!msg) {
             ret = -1;
         } else {
@@ -344,7 +344,7 @@ int daemonFreeClientStream(virNetServerClientPtr client,
         virNetMessagePtr tmp = msg->next;
         if (client) {
             /* Send a dummy reply to free up 'msg' & unblock client rx */
-            memset(msg, 0, sizeof(*msg));
+            virNetMessageClear(msg);
             msg->header.type = VIR_NET_REPLY;
             if (virNetServerClientSendMessage(client, msg) < 0) {
                 virNetServerClientImmediateClose(client);
@@ -653,7 +653,7 @@ daemonStreamHandleWrite(virNetServerClientPtr client,
          * its active request count / throttling
          */
         if (msg->header.status == VIR_NET_CONTINUE) {
-            memset(msg, 0, sizeof(*msg));
+            virNetMessageClear(msg);
             msg->header.type = VIR_NET_REPLY;
             if (virNetServerClientSendMessage(client, msg) < 0) {
                 virNetMessageFree(msg);
@@ -715,7 +715,7 @@ daemonStreamHandleRead(virNetServerClientPtr client,
 
         memset(&rerr, 0, sizeof(rerr));
 
-        if (!(msg = virNetMessageNew()))
+        if (!(msg = virNetMessageNew(false)))
             ret = -1;
         else
             ret = virNetServerProgramSendStreamError(remoteProgram,
@@ -729,7 +729,7 @@ daemonStreamHandleRead(virNetServerClientPtr client,
         stream->tx = 0;
         if (ret == 0)
             stream->recvEOF = 1;
-        if (!(msg = virNetMessageNew()))
+        if (!(msg = virNetMessageNew(false)))
             ret = -1;
 
         if (msg) {