Allow sync IO and keepalives to be skipped in RPC client setup

Currently the virNetClientPtr constructor will always register
the async IO event handler and the keepalive objects. In the
case of the lock manager, there will be no event loop available
nor keepalive support required. Split this setup out of the
constructor and into separate methods.

The remote driver will enable async IO and keepalives, while
the LXC driver will only enable async IO

Signed-off-by: Daniel P. Berrange <berrange@redhat.com>

src/libvirt_private.syms

@@ -1322,6 +1322,8 @@ virNetClientNewExternal;
 virNetClientNewSSH;
 virNetClientNewTCP;
 virNetClientNewUNIX;
+virNetClientRegisterAsyncIO;
+virNetClientRegisterKeepAlive;
 virNetClientRemoteAddrString;
 virNetClientRemoveStream;
 virNetClientSendNoReply;

src/lxc/lxc_monitor.c

@@ -129,6 +129,8 @@ virLXCMonitorPtr virLXCMonitorNew(virDomainObjPtr vm,
     if (!(mon->client = virNetClientNewUNIX(sockpath, false, NULL)))
         goto error;
 
+    if (virNetClientRegisterAsyncIO(mon->client) < 0)
+        goto error;
 
     if (!(mon->program = virNetClientProgramNew(VIR_LXC_PROTOCOL_PROGRAM,
                                                 VIR_LXC_PROTOCOL_PROGRAM_VERSION,

src/remote/remote_driver.c

@@ -686,6 +686,15 @@ doRemoteOpen(virConnectPtr conn,
     } /* switch (transport) */
 
 
+    if (virNetClientRegisterAsyncIO(priv->client) < 0) {
+        VIR_DEBUG("Failed to add event watch, disabling events and support for"
+                  " keepalive messages");
+        virResetLastError();
+    } else {
+        if (virNetClientRegisterKeepAlive(priv->client) < 0)
+            goto failed;
+    }
+
     virNetClientSetCloseCallback(priv->client,
                                  remoteClientCloseFunc,
                                  conn, NULL);

src/rpc/virnetclient.c

@@ -68,6 +68,7 @@ struct _virNetClient {
     virMutex lock;
 
     virNetSocketPtr sock;
+    bool asyncIO;
 
     virNetTLSSessionPtr tls;
     char *hostname;
@@ -309,7 +310,6 @@ static virNetClientPtr virNetClientNew(virNetSocketPtr sock,
 {
     virNetClientPtr client = NULL;
     int wakeupFD[2] = { -1, -1 };
-    virKeepAlivePtr ka = NULL;
 
     if (virNetClientInitialize() < 0)
         return NULL;
@@ -337,29 +337,6 @@ static virNetClientPtr virNetClientNew(virNetSocketPtr sock,
         !(client->hostname = strdup(hostname)))
         goto no_memory;
 
-    /* Set up a callback to listen on the socket data */
-    virObjectRef(client);
-    if (virNetSocketAddIOCallback(client->sock,
-                                  VIR_EVENT_HANDLE_READABLE,
-                                  virNetClientIncomingEvent,
-                                  client,
-                                  virObjectFreeCallback) < 0) {
-        virObjectUnref(client);
-        VIR_DEBUG("Failed to add event watch, disabling events and support for"
-                  " keepalive messages");
-    } else {
-        /* Keepalive protocol consists of async messages so it can only be used
-         * if the client supports them */
-        if (!(ka = virKeepAliveNew(-1, 0, client,
-                                   virNetClientKeepAliveSendCB,
-                                   virNetClientKeepAliveDeadCB,
-                                   virObjectFreeCallback)))
-            goto error;
-        /* keepalive object has a reference to client */
-        virObjectRef(client);
-    }
-
-    client->keepalive = ka;
     PROBE(RPC_CLIENT_NEW,
           "client=%p sock=%p",
           client, client->sock);
@@ -370,10 +347,6 @@ no_memory:
 error:
     VIR_FORCE_CLOSE(wakeupFD[0]);
     VIR_FORCE_CLOSE(wakeupFD[1]);
-    if (ka) {
-        virKeepAliveStop(ka);
-        virObjectUnref(ka);
-    }
     virObjectUnref(client);
     return NULL;
 }
@@ -433,6 +406,58 @@ virNetClientPtr virNetClientNewExternal(const char **cmdargv)
 }
 
 
+int virNetClientRegisterAsyncIO(virNetClientPtr client)
+{
+    if (client->asyncIO)
+        return 0;
+
+    /* Set up a callback to listen on the socket data */
+    virObjectRef(client);
+    if (virNetSocketAddIOCallback(client->sock,
+                                  VIR_EVENT_HANDLE_READABLE,
+                                  virNetClientIncomingEvent,
+                                  client,
+                                  virObjectFreeCallback) < 0) {
+        virObjectUnref(client);
+        virReportError(VIR_ERR_INTERNAL_ERROR, "%s",
+                       _("Unable to register async IO callback"));
+        return -1;
+    }
+
+    client->asyncIO = true;
+    return 0;
+}
+
+
+int virNetClientRegisterKeepAlive(virNetClientPtr client)
+{
+    virKeepAlivePtr ka;
+
+    if (client->keepalive)
+        return 0;
+
+    if (!client->asyncIO) {
+        virReportError(VIR_ERR_OPERATION_INVALID, "%s",
+                       _("Unable to enable keepalives without async IO support"));
+        return -1;
+    }
+
+    /* Keepalive protocol consists of async messages so it can only be used
+     * if the client supports them */
+    if (!(ka = virKeepAliveNew(-1, 0, client,
+                               virNetClientKeepAliveSendCB,
+                               virNetClientKeepAliveDeadCB,
+                               virObjectFreeCallback)))
+        return -1;
+
+    /* keepalive object has a reference to client */
+    virObjectRef(client);
+
+    client->keepalive = ka;
+    return 0;
+}
+
+
 int virNetClientGetFD(virNetClientPtr client)
 {
     int fd;

src/rpc/virnetclient.h

@@ -52,6 +52,9 @@ virNetClientPtr virNetClientNewSSH(const char *nodename,
 
 virNetClientPtr virNetClientNewExternal(const char **cmdargv);
 
+int virNetClientRegisterAsyncIO(virNetClientPtr client);
+int virNetClientRegisterKeepAlive(virNetClientPtr client);
+
 typedef void (*virNetClientCloseFunc)(virNetClientPtr client,
                                       int reason,
                                       void *opaque);