From: "Daniel P. Berrange" <berrange(a)redhat.com&gt; Remove the need for a virNetSocket object to be protected by locks from the object using it, by introducing its own native locking and reference counting * src/rpc/virnetsocket.c: Add locking & reference counting --- src/rpc/virnetsocket.c | 147 +++++++++++++++++++++++++++++++++++++++--------- 1 files changed, 120 insertions(+), 27 deletions(-) diff --git a/src/rpc/virnetsocket.c b/src/rpc/virnetsocket.c index 7ea1ab7..8dd4d3a 100644 --- a/src/rpc/virnetsocket.c +++ b/src/rpc/virnetsocket.c @@ -40,6 +40,7 @@ #include "logging.h" #include "files.h" #include "event.h" +#include "threads.h" #define VIR_FROM_THIS VIR_FROM_RPC @@ -49,6 +50,9 @@ struct _virNetSocket { + virMutex lock; + int refs; + int fd; int watch; pid_t pid; @@ -122,6 +126,14 @@ static virNetSocketPtr virNetSocketNew(virSocketAddrPtr localAddr, return NULL; } + if (virMutexInit(&sock->lock) < 0) { + virReportSystemError(errno, "%s", + _("Unable to initialize mutex")); + VIR_FREE(sock); + return NULL; + } + sock->refs = 1; + if (localAddr) sock->localAddr = *localAddr; if (remoteAddr) @@ -627,6 +639,13 @@ void virNetSocketFree(virNetSocketPtr sock) if (!sock) return; + virMutexLock(&sock->lock); + sock->refs--; + if (sock->refs > 0) { + virMutexUnlock(&sock->lock); + return; + } + VIR_DEBUG("sock=%p fd=%d", sock, sock->fd); if (sock->watch > 0) { virEventRemoveHandle(sock->watch); @@ -657,27 +676,41 @@ void virNetSocketFree(virNetSocketPtr sock) VIR_FREE(sock->localAddrStr); VIR_FREE(sock->remoteAddrStr); + virMutexUnlock(&sock->lock); + virMutexDestroy(&sock->lock); + VIR_FREE(sock); } int virNetSocketGetFD(virNetSocketPtr sock) { - return sock->fd; + int fd; + virMutexLock(&sock->lock); + fd = sock->fd; + virMutexUnlock(&sock->lock); + return fd; } bool virNetSocketIsLocal(virNetSocketPtr sock) { + bool isLocal = false; + virMutexLock(&sock->lock); if (sock->localAddr.data.sa.sa_family == AF_UNIX) - return true; - return false; + isLocal = true; + virMutexUnlock(&sock->lock); + return isLocal; } int virNetSocketGetPort(virNetSocketPtr sock) { - return virSocketGetPort(&sock->localAddr); + int port; + virMutexLock(&sock->lock); + port = virSocketGetPort(&sock->localAddr); + virMutexUnlock(&sock->lock); + return port; } @@ -688,15 +721,19 @@ int virNetSocketGetLocalIdentity(virNetSocketPtr sock, { struct ucred cr; unsigned int cr_len = sizeof (cr); + virMutexLock(&sock->lock); if (getsockopt(sock->fd, SOL_SOCKET, SO_PEERCRED, &cr, &cr_len) < 0) { virReportSystemError(errno, "%s", _("Failed to get client socket identity")); + virMutexUnlock(&sock->lock); return -1; } *pid = cr.pid; *uid = cr.uid; + + virMutexUnlock(&sock->lock); return 0; } #else @@ -715,7 +752,11 @@ int virNetSocketGetLocalIdentity(virNetSocketPtr sock ATTRIBUTE_UNUSED, int virNetSocketSetBlocking(virNetSocketPtr sock, bool blocking) { - return virSetBlocking(sock->fd, blocking); + int ret; + virMutexLock(&sock->lock); + ret = virSetBlocking(sock->fd, blocking); + virMutexUnlock(&sock->lock); + return ret; } @@ -751,6 +792,7 @@ static ssize_t virNetSocketTLSSessionRead(char *buf, void virNetSocketSetTLSSession(virNetSocketPtr sock, virNetTLSSessionPtr sess) { + virMutexLock(&sock->lock); virNetTLSSessionFree(sock->tlsSession); sock->tlsSession = sess; virNetTLSSessionSetIOCallbacks(sess, @@ -758,6 +800,7 @@ void virNetSocketSetTLSSession(virNetSocketPtr sock, virNetSocketTLSSessionRead, sock); virNetTLSSessionRef(sess); + virMutexUnlock(&sock->lock); } @@ -765,20 +808,25 @@ void virNetSocketSetTLSSession(virNetSocketPtr sock, void virNetSocketSetSASLSession(virNetSocketPtr sock, virNetSASLSessionPtr sess) { + virMutexLock(&sock->lock); virNetSASLSessionFree(sock->saslSession); sock->saslSession = sess; virNetSASLSessionRef(sess); + virMutexUnlock(&sock->lock); } #endif bool virNetSocketHasCachedData(virNetSocketPtr sock ATTRIBUTE_UNUSED) { + bool hasCached = false; + virMutexLock(&sock->lock); #if HAVE_SASL if (sock->saslDecoded) - return true; + hasCached = true; #endif - return false; + virMutexUnlock(&sock->lock); + return hasCached; } @@ -965,39 +1013,54 @@ static ssize_t virNetSocketWriteSASL(virNetSocketPtr sock, const char *buf, size ssize_t virNetSocketRead(virNetSocketPtr sock, char *buf, size_t len) { + ssize_t ret; + virMutexLock(&sock->lock); #if HAVE_SASL if (sock->saslSession) - return virNetSocketReadSASL(sock, buf, len); + ret = virNetSocketReadSASL(sock, buf, len); else #endif - return virNetSocketReadWire(sock, buf, len); + ret = virNetSocketReadWire(sock, buf, len); + virMutexUnlock(&sock->lock); + return ret; } ssize_t virNetSocketWrite(virNetSocketPtr sock, const char *buf, size_t len) { + ssize_t ret; + + virMutexLock(&sock->lock); #if HAVE_SASL if (sock->saslSession) - return virNetSocketWriteSASL(sock, buf, len); + ret = virNetSocketWriteSASL(sock, buf, len); else #endif - return virNetSocketWriteWire(sock, buf, len); + ret = virNetSocketWriteWire(sock, buf, len); + virMutexUnlock(&sock->lock); + return ret; } int virNetSocketListen(virNetSocketPtr sock) { + virMutexLock(&sock->lock); if (listen(sock->fd, 30) < 0) { virReportSystemError(errno, "%s", _("Unable to listen on socket")); + virMutexUnlock(&sock->lock); return -1; } + virMutexUnlock(&sock->lock); return 0; } int virNetSocketAccept(virNetSocketPtr sock, virNetSocketPtr *clientsock) { - int fd; + int fd = -1; virSocketAddr localAddr; virSocketAddr remoteAddr; + int ret = -1; + + virMutexLock(&sock->lock); *clientsock = NULL; @@ -1007,30 +1070,35 @@ int virNetSocketAccept(virNetSocketPtr sock, virNetSocketPtr *clientsock) remoteAddr.len = sizeof(remoteAddr.data.stor); if ((fd = accept(sock->fd, &remoteAddr.data.sa, &remoteAddr.len)) < 0) { if (errno == ECONNABORTED || - errno == EAGAIN) - return 0; + errno == EAGAIN) { + ret = 0; + goto cleanup; + } virReportSystemError(errno, "%s", _("Unable to accept client")); - return -1; + goto cleanup; } localAddr.len = sizeof(localAddr.data); if (getsockname(fd, &localAddr.data.sa, &localAddr.len) < 0) { virReportSystemError(errno, "%s", _("Unable to get local socket name")); - VIR_FORCE_CLOSE(fd); - return -1; + goto cleanup; } if (!(*clientsock = virNetSocketNew(&localAddr, &remoteAddr, true, - fd, -1, 0))) { - VIR_FORCE_CLOSE(fd); - return -1; - } + fd, -1, 0))) + goto cleanup; - return 0; + fd = -1; + ret = 0; + +cleanup: + VIR_FORCE_CLOSE(fd); + virMutexUnlock(&sock->lock); + return ret; } @@ -1040,52 +1108,77 @@ static void virNetSocketEventHandle(int watch ATTRIBUTE_UNUSED, void *opaque) { virNetSocketPtr sock = opaque; + virNetSocketIOFunc func; + void *eopaque; - sock->func(sock, events, sock->opaque); + virMutexLock(&sock->lock); + func = sock->func; + eopaque = sock->opaque; + virMutexUnlock(&sock->lock); + + if (func) + func(sock, events, eopaque); } + int virNetSocketAddIOCallback(virNetSocketPtr sock, int events, virNetSocketIOFunc func, void *opaque) { + int ret = -1; + + virMutexLock(&sock->lock); if (sock->watch > 0) { VIR_DEBUG("Watch already registered on socket %p", sock); - return -1; + goto cleanup; } + sock->refs++; if ((sock->watch = virEventAddHandle(sock->fd, events, virNetSocketEventHandle, sock, NULL)) < 0) { VIR_DEBUG("Failed to register watch on socket %p", sock); - return -1; + goto cleanup; } sock->func = func; sock->opaque = opaque; - return 0; + ret = 0; + +cleanup: + virMutexUnlock(&sock->lock); + return ret; } void virNetSocketUpdateIOCallback(virNetSocketPtr sock, int events) { + virMutexLock(&sock->lock); if (sock->watch <= 0) { VIR_DEBUG("Watch not registered on socket %p", sock); + virMutexUnlock(&sock->lock); return; } virEventUpdateHandle(sock->watch, events); + + virMutexUnlock(&sock->lock); } void virNetSocketRemoveIOCallback(virNetSocketPtr sock) { + virMutexLock(&sock->lock); + if (sock->watch <= 0) { VIR_DEBUG("Watch not registered on socket %p", sock); + virMutexUnlock(&sock->lock); return; } virEventRemoveHandle(sock->watch); - sock->watch = 0; + + virMutexUnlock(&sock->lock); } -- 1.7.6

From: "Daniel P. Berrange" <berrange(a)redhat.com&gt; * src/rpc/virnetclient.c: Add debugging of ref counts --- src/rpc/virnetclient.c | 3 +++ 1 files changed, 3 insertions(+), 0 deletions(-) diff --git a/src/rpc/virnetclient.c b/src/rpc/virnetclient.c index 9eaecfc..4a9eabc 100644 --- a/src/rpc/virnetclient.c +++ b/src/rpc/virnetclient.c @@ -146,6 +146,7 @@ static virNetClientPtr virNetClientNew(virNetSocketPtr sock, client) < 0) VIR_DEBUG("Failed to add event watch, disabling events"); + VIR_DEBUG("client=%p refs=%d", client, client->refs); return client; no_memory: @@ -214,6 +215,7 @@ void virNetClientRef(virNetClientPtr client) { virNetClientLock(client); client->refs++; + VIR_DEBUG("client=%p refs=%d", client, client->refs); virNetClientUnlock(client); } @@ -226,6 +228,7 @@ void virNetClientFree(virNetClientPtr client) return; virNetClientLock(client); + VIR_DEBUG("client=%p refs=%d", client, client->refs); client->refs--; if (client->refs > 0) { virNetClientUnlock(client); -- 1.7.6

From: "Daniel P. Berrange" <berrange(a)redhat.com&gt; When unregistering an I/O callback from a virNetSocket object, there is still a chance that an event may come in on the callback. In this case it is possible that the virNetSocket might have been freed already. Make use of a virFreeCallback when registering the I/O callbacks and hold a reference for the entire time the callback is set. * src/rpc/virnetsocket.c: Register a free function for the file handle watch * src/rpc/virnetsocket.h, src/rpc/virnetserverservice.c, src/rpc/virnetserverclient.c, src/rpc/virnetclient.c: Add a free function for the socket I/O watches --- src/rpc/virnetclient.c | 13 ++++++++++++- src/rpc/virnetserverclient.c | 13 ++++++++++++- src/rpc/virnetserverservice.c | 20 ++++++++++++++++++-- src/rpc/virnetsocket.c | 30 ++++++++++++++++++++++++++++-- src/rpc/virnetsocket.h | 3 ++- 5 files changed, 72 insertions(+), 7 deletions(-) diff --git a/src/rpc/virnetclient.c b/src/rpc/virnetclient.c index 4a9eabc..27542a5 100644 --- a/src/rpc/virnetclient.c +++ b/src/rpc/virnetclient.c @@ -110,6 +110,13 @@ static void virNetClientIncomingEvent(virNetSocketPtr sock, int events, void *opaque); +static void virNetClientEventFree(void *opaque) +{ + virNetClientPtr client = opaque; + + virNetClientFree(client); +} + static virNetClientPtr virNetClientNew(virNetSocketPtr sock, const char *hostname) { @@ -140,11 +147,15 @@ static virNetClientPtr virNetClientNew(virNetSocketPtr sock, goto no_memory; /* Set up a callback to listen on the socket data */ + client->refs++; if (virNetSocketAddIOCallback(client->sock, VIR_EVENT_HANDLE_READABLE, virNetClientIncomingEvent, - client) < 0) + client, + virNetClientEventFree) < 0) { + client->refs--; VIR_DEBUG("Failed to add event watch, disabling events"); + } VIR_DEBUG("client=%p refs=%d", client, client->refs); return client; diff --git a/src/rpc/virnetserverclient.c b/src/rpc/virnetserverclient.c index 341981f..317d59c 100644 --- a/src/rpc/virnetserverclient.c +++ b/src/rpc/virnetserverclient.c @@ -160,6 +160,13 @@ virNetServerClientCalculateHandleMode(virNetServerClientPtr client) { return mode; } +static void virNetServerClientEventFree(void *opaque) +{ + virNetServerClientPtr client = opaque; + + virNetServerClientFree(client); +} + /* * @server: a locked or unlocked server object * @client: a locked client object @@ -168,12 +175,16 @@ static int virNetServerClientRegisterEvent(virNetServerClientPtr client) { int mode = virNetServerClientCalculateHandleMode(client); + client->refs++; VIR_DEBUG("Registering client event callback %d", mode); if (virNetSocketAddIOCallback(client->sock, mode, virNetServerClientDispatchEvent, - client) < 0) + client, + virNetServerClientEventFree) < 0) { + client->refs--; return -1; + } return 0; } diff --git a/src/rpc/virnetserverservice.c b/src/rpc/virnetserverservice.c index 8c250e2..d5648dc 100644 --- a/src/rpc/virnetserverservice.c +++ b/src/rpc/virnetserverservice.c @@ -91,6 +91,14 @@ error: } +static void virNetServerServiceEventFree(void *opaque) +{ + virNetServerServicePtr svc = opaque; + + virNetServerServiceFree(svc); +} + + virNetServerServicePtr virNetServerServiceNewTCP(const char *nodename, const char *service, int auth, @@ -124,11 +132,15 @@ virNetServerServicePtr virNetServerServiceNewTCP(const char *nodename, /* IO callback is initially disabled, until we're ready * to deal with incoming clients */ + virNetServerServiceRef(svc); if (virNetSocketAddIOCallback(svc->socks[i], 0, virNetServerServiceAccept, - svc) < 0) + svc, + virNetServerServiceEventFree) < 0) { + virNetServerServiceFree(svc); goto error; + } } @@ -180,11 +192,15 @@ virNetServerServicePtr virNetServerServiceNewUNIX(const char *path, /* IO callback is initially disabled, until we're ready * to deal with incoming clients */ + virNetServerServiceRef(svc); if (virNetSocketAddIOCallback(svc->socks[i], 0, virNetServerServiceAccept, - svc) < 0) + svc, + virNetServerServiceEventFree) < 0) { + virNetServerServiceFree(svc); goto error; + } } diff --git a/src/rpc/virnetsocket.c b/src/rpc/virnetsocket.c index 8dd4d3a..43460d9 100644 --- a/src/rpc/virnetsocket.c +++ b/src/rpc/virnetsocket.c @@ -58,8 +58,12 @@ struct _virNetSocket { pid_t pid; int errfd; bool client; + + /* Event callback fields */ virNetSocketIOFunc func; void *opaque; + virFreeCallback ff; + virSocketAddr localAddr; virSocketAddr remoteAddr; char *localAddrStr; @@ -1121,10 +1125,31 @@ static void virNetSocketEventHandle(int watch ATTRIBUTE_UNUSED, } +static void virNetSocketEventFree(void *opaque) +{ + virNetSocketPtr sock = opaque; + virFreeCallback ff; + void *eopaque; + + virMutexLock(&sock->lock); + ff = sock->ff; + eopaque = sock->opaque; + sock->func = NULL; + sock->ff = NULL; + sock->opaque = NULL; + virMutexUnlock(&sock->lock); + + if (ff) + ff(eopaque); + + virNetSocketFree(sock); +} + int virNetSocketAddIOCallback(virNetSocketPtr sock, int events, virNetSocketIOFunc func, - void *opaque) + void *opaque, + virFreeCallback ff) { int ret = -1; @@ -1139,12 +1164,13 @@ int virNetSocketAddIOCallback(virNetSocketPtr sock, events, virNetSocketEventHandle, sock, - NULL)) < 0) { + virNetSocketEventFree)) < 0) { VIR_DEBUG("Failed to register watch on socket %p", sock); goto cleanup; } sock->func = func; sock->opaque = opaque; + sock->ff = ff; ret = 0; diff --git a/src/rpc/virnetsocket.h b/src/rpc/virnetsocket.h index 5f882ac..e13ab8f 100644 --- a/src/rpc/virnetsocket.h +++ b/src/rpc/virnetsocket.h @@ -109,7 +109,8 @@ int virNetSocketAccept(virNetSocketPtr sock, int virNetSocketAddIOCallback(virNetSocketPtr sock, int events, virNetSocketIOFunc func, - void *opaque); + void *opaque, + virFreeCallback ff); void virNetSocketUpdateIOCallback(virNetSocketPtr sock, int events); -- 1.7.6

From: "Daniel P. Berrange" <berrange(a)redhat.com&gt; Since the I/O callback registered against virNetSocket will hold a reference on the virNetClient, we can't rely on the virNetClientFree to be able to close the network connection. The last reference will only go away when the event callback fires (likely due to EOF from the server). This is sub-optimal and can potentially cause a leak of the virNetClient object if the server were to not explicitly close the socket itself * src/remote/remote_driver.c: Explicitly close the client object when disconnecting * src/rpc/virnetclient.c, src/rpc/virnetclient.h: Add a virNetClientClose method --- src/remote/remote_driver.c | 2 ++ src/rpc/virnetclient.c | 22 +++++++++++++++++++++- src/rpc/virnetclient.h | 1 + 3 files changed, 24 insertions(+), 1 deletions(-) diff --git a/src/remote/remote_driver.c b/src/remote/remote_driver.c index c2f8bbd..6bae353 100644 --- a/src/remote/remote_driver.c +++ b/src/remote/remote_driver.c @@ -684,6 +684,7 @@ doRemoteOpen (virConnectPtr conn, free_qparam_set (vars); failed: + virNetClientClose(priv->client); virNetClientFree(priv->client); priv->client = NULL; @@ -834,6 +835,7 @@ doRemoteClose (virConnectPtr conn, struct private_data *priv) virNetTLSContextFree(priv->tls); priv->tls = NULL; + virNetClientClose(priv->client); virNetClientFree(priv->client); priv->client = NULL; virNetClientProgramFree(priv->remoteProgram); diff --git a/src/rpc/virnetclient.c b/src/rpc/virnetclient.c index 27542a5..dfc4ed9 100644 --- a/src/rpc/virnetclient.c +++ b/src/rpc/virnetclient.c @@ -255,7 +255,8 @@ void virNetClientFree(virNetClientPtr client) VIR_FREE(client->hostname); - virNetSocketRemoveIOCallback(client->sock); + if (client->sock) + virNetSocketRemoveIOCallback(client->sock); virNetSocketFree(client->sock); virNetTLSSessionFree(client->tls); #if HAVE_SASL @@ -268,6 +269,22 @@ void virNetClientFree(virNetClientPtr client) } +void virNetClientClose(virNetClientPtr client) +{ + virNetClientLock(client); + virNetSocketRemoveIOCallback(client->sock); + virNetSocketFree(client->sock); + client->sock = NULL; + virNetTLSSessionFree(client->tls); + client->tls = NULL; +#if HAVE_SASL + virNetSASLSessionFree(client->sasl); + client->sasl = NULL; +#endif + virNetClientUnlock(client); +} + + #if HAVE_SASL void virNetClientSetSASLSession(virNetClientPtr client, virNetSASLSessionPtr sasl) @@ -1118,6 +1135,9 @@ void virNetClientIncomingEvent(virNetSocketPtr sock, virNetClientLock(client); + if (!client->sock) + goto done; + /* This should be impossible, but it doesn't hurt to check */ if (client->waitDispatch) goto done; diff --git a/src/rpc/virnetclient.h b/src/rpc/virnetclient.h index 6acdf50..a0983bc 100644 --- a/src/rpc/virnetclient.h +++ b/src/rpc/virnetclient.h @@ -81,5 +81,6 @@ const char *virNetClientRemoteAddrString(virNetClientPtr client); int virNetClientGetTLSKeySize(virNetClientPtr client); void virNetClientFree(virNetClientPtr client); +void virNetClientClose(virNetClientPtr client); #endif /* __VIR_NET_CLIENT_H__ */ -- 1.7.6