Ensure client streams are closed when marking a client for close

Every active stream results in a reference being held on the
virNetServerClientPtr object. This meant that if a client quit
with any streams active, although all I/O was stopped the
virNetServerClientPtr object would leak. This causes libvirtd
to leak any file handles associated with open streams when a
client quit

To fix this, when we call virNetServerClientClose there is a
callback invoked which lets the daemon release the streams
and thus the extra references

* daemon/remote.c: Add a hook to close all streams
* daemon/stream.c, daemon/stream.h: Add API for releasing
  all streams
* src/rpc/virnetserverclient.c, src/rpc/virnetserverclient.h:
  Allow registration of a hook to trigger when closing client

daemon/remote.c

@@ -439,6 +439,13 @@ static void remoteClientFreeFunc(void *data)
 }
 
 
+static void remoteClientCloseFunc(virNetServerClientPtr client)
+{
+    struct daemonClientPrivate *priv = virNetServerClientGetPrivateData(client);
+
+    daemonRemoveAllClientStreams(priv->streams);
+}
+
 
 int remoteClientInitHook(virNetServerPtr srv ATTRIBUTE_UNUSED,
                          virNetServerClientPtr client)
@@ -460,7 +467,9 @@ int remoteClientInitHook(virNetServerPtr srv ATTRIBUTE_UNUSED,
     for (i = 0 ; i < VIR_DOMAIN_EVENT_ID_LAST ; i++)
         priv->domainEventCallbackID[i] = -1;
 
-    virNetServerClientSetPrivateData(client, priv, remoteClientFreeFunc);
+    virNetServerClientSetPrivateData(client, priv,
+                                     remoteClientFreeFunc);
+    virNetServerClientSetCloseHook(client, remoteClientCloseFunc);
     return 0;
 }
 

daemon/stream.c

@@ -334,6 +334,7 @@ int daemonFreeClientStream(virNetServerClientPtr client,
     msg = stream->rx;
     while (msg) {
         virNetMessagePtr tmp = msg->next;
+        if (client) {
             /* Send a dummy reply to free up 'msg' & unblock client rx */
             memset(msg, 0, sizeof(*msg));
             msg->header.type = VIR_NET_REPLY;
@@ -342,6 +343,9 @@ int daemonFreeClientStream(virNetServerClientPtr client,
                 virNetMessageFree(msg);
                 ret = -1;
             }
+        } else {
+            virNetMessageFree(msg);
+        }
         msg = tmp;
     }
 
@@ -441,6 +445,28 @@ daemonRemoveClientStream(virNetServerClientPtr client,
 }
 
 
+void
+daemonRemoveAllClientStreams(daemonClientStream *stream)
+{
+    daemonClientStream *tmp;
+
+    VIR_DEBUG("stream=%p", stream);
+
+    while (stream) {
+        tmp = stream->next;
+
+        if (!stream->closed) {
+            virStreamEventRemoveCallback(stream->st);
+            virStreamAbort(stream->st);
+        }
+
+        daemonFreeClientStream(NULL, stream);
+
+        VIR_DEBUG("next stream=%p", tmp);
+        stream = tmp;
+    }
+}
+
 /*
  * Returns:
  *   -1  if fatal error occurred

daemon/stream.h

@@ -45,4 +45,7 @@ int
 daemonRemoveClientStream(virNetServerClientPtr client,
                          daemonClientStream *stream);
 
+void
+daemonRemoveAllClientStreams(daemonClientStream *stream);
+
 #endif /* __LIBVIRTD_STREAM_H__ */

src/rpc/virnetserverclient.c

@@ -97,6 +97,7 @@ struct _virNetServerClient
 
     void *privateData;
     virNetServerClientFreeFunc privateDataFreeFunc;
+    virNetServerClientCloseFunc privateDataCloseFunc;
 };
 
 
@@ -492,6 +493,15 @@ void *virNetServerClientGetPrivateData(virNetServerClientPtr client)
 }
 
 
+void virNetServerClientSetCloseHook(virNetServerClientPtr client,
+                                    virNetServerClientCloseFunc cf)
+{
+    virNetServerClientLock(client);
+    client->privateDataCloseFunc = cf;
+    virNetServerClientUnlock(client);
+}
+
+
 void virNetServerClientSetDispatcher(virNetServerClientPtr client,
                                      virNetServerClientDispatchFunc func,
                                      void *opaque)
@@ -560,6 +570,8 @@ void virNetServerClientFree(virNetServerClientPtr client)
  */
 void virNetServerClientClose(virNetServerClientPtr client)
 {
+    virNetServerClientCloseFunc cf;
+
     virNetServerClientLock(client);
     VIR_DEBUG("client=%p refs=%d", client, client->refs);
     if (!client->sock) {
@@ -567,6 +579,15 @@ void virNetServerClientClose(virNetServerClientPtr client)
         return;
     }
 
+    if (client->privateDataCloseFunc) {
+        cf = client->privateDataCloseFunc;
+        client->refs++;
+        virNetServerClientUnlock(client);
+        (cf)(client);
+        virNetServerClientLock(client);
+        client->refs--;
+    }
+
     /* Do now, even though we don't close the socket
      * until end, to ensure we don't get invoked
      * again due to tls shutdown */

src/rpc/virnetserverclient.h

@@ -82,6 +82,11 @@ void virNetServerClientSetPrivateData(virNetServerClientPtr client,
                                       virNetServerClientFreeFunc ff);
 void *virNetServerClientGetPrivateData(virNetServerClientPtr client);
 
+typedef void (*virNetServerClientCloseFunc)(virNetServerClientPtr client);
+
+void virNetServerClientSetCloseHook(virNetServerClientPtr client,
+                                    virNetServerClientCloseFunc cf);
+
 void virNetServerClientSetDispatcher(virNetServerClientPtr client,
                                      virNetServerClientDispatchFunc func,
                                      void *opaque);