daemon: Create priority workers pool

This patch annotates APIs with low or high priority.
In low set MUST be all APIs which might eventually access monitor
(and thus block indefinitely). Other APIs may be marked as high
priority. However, some must be (e.g. domainDestroy).

For high priority calls (HPC), there are some high priority workers
(HPW) created in the pool. HPW can execute only HPC, although normal
worker can process any call regardless priority. Therefore, only those
APIs which are guaranteed to end in reasonable small amount of time
can be marked as HPC.

The size of this HPC pool is static, because HPC are expected to end
quickly, therefore jobs assigned to this pool will be served quickly.
It can be configured in libvirtd.conf via prio_workers variable.
Default is set to 5.

To mark API with low or high priority, append priority:{low|high} to
it's comment in src/remote/remote_protocol.x. This is similar to
autogen|skipgen. If not marked, the generator assumes low as default.

1 files changed, 36 insertions, 32 deletions

diff --git a/src/rpc/virnetserver.c b/src/rpc/virnetserver.c
index fd8781d67..d71ed18d7 100644
--- a/src/rpc/virnetserver.c
+++ b/src/rpc/virnetserver.c
@@ -64,6 +64,7 @@ typedef virNetServerJob *virNetServerJobPtr;
 struct _virNetServerJob {
     virNetServerClientPtr client;
     virNetMessagePtr msg;
+    virNetServerProgramPtr prog;
 };
 
 struct _virNetServer {
@@ -128,61 +129,41 @@ static void virNetServerHandleJob(void *jobOpaque, void *opaque)
 {
     virNetServerPtr srv = opaque;
     virNetServerJobPtr job = jobOpaque;
-    virNetServerProgramPtr prog = NULL;
-    size_t i;
-
-    virNetServerLock(srv);
-    VIR_DEBUG("server=%p client=%p message=%p",
-              srv, job->client, job->msg);
 
-    for (i = 0 ; i < srv->nprograms ; i++) {
-        if (virNetServerProgramMatches(srv->programs[i], job->msg)) {
-            prog = srv->programs[i];
-            break;
-        }
-    }
+    VIR_DEBUG("server=%p client=%p message=%p prog=%p",
+              srv, job->client, job->msg, job->prog);
 
-    if (!prog) {
-        if (virNetServerProgramUnknownError(job->client,
-                                            job->msg,
-                                            &job->msg->header) < 0)
-            goto error;
-        else
-            goto cleanup;
-    }
-
-    virNetServerProgramRef(prog);
-    virNetServerUnlock(srv);
-
-    if (virNetServerProgramDispatch(prog,
+    if (virNetServerProgramDispatch(job->prog,
                                     srv,
                                     job->client,
                                     job->msg) < 0)
         goto error;
 
     virNetServerLock(srv);
-
-cleanup:
-    virNetServerProgramFree(prog);
+    virNetServerProgramFree(job->prog);
     virNetServerUnlock(srv);
     virNetServerClientFree(job->client);
     VIR_FREE(job);
     return;
 
 error:
+    virNetServerProgramFree(job->prog);
     virNetMessageFree(job->msg);
     virNetServerClientClose(job->client);
-    goto cleanup;
+    virNetServerClientFree(job->client);
+    VIR_FREE(job);
 }
 
-
 static int virNetServerDispatchNewMessage(virNetServerClientPtr client,
                                           virNetMessagePtr msg,
                                           void *opaque)
 {
     virNetServerPtr srv = opaque;
     virNetServerJobPtr job;
-    int ret;
+    virNetServerProgramPtr prog = NULL;
+    unsigned int priority = 0;
+    size_t i;
+    int ret = -1;
 
     VIR_DEBUG("server=%p client=%p message=%p",
               srv, client, msg);
@@ -196,8 +177,29 @@ static int virNetServerDispatchNewMessage(virNetServerClientPtr client,
     job->msg = msg;
 
     virNetServerLock(srv);
-    if ((ret = virThreadPoolSendJob(srv->workers, job)) < 0)
+    for (i = 0 ; i < srv->nprograms ; i++) {
+        if (virNetServerProgramMatches(srv->programs[i], job->msg)) {
+            prog = srv->programs[i];
+            break;
+        }
+    }
+
+    if (!prog) {
+        virNetServerProgramUnknownError(client, msg, &msg->header);
+        goto cleanup;
+    }
+
+    virNetServerProgramRef(prog);
+    job->prog = prog;
+    priority = virNetServerProgramGetPriority(prog, msg->header.proc);
+
+    ret = virThreadPoolSendJob(srv->workers, priority, job);
+
+cleanup:
+    if (ret < 0) {
         VIR_FREE(job);
+        virNetServerProgramFree(prog);
+    }
     virNetServerUnlock(srv);
 
     return ret;
@@ -274,6 +276,7 @@ static void virNetServerFatalSignal(int sig, siginfo_t *siginfo ATTRIBUTE_UNUSED
 
 virNetServerPtr virNetServerNew(size_t min_workers,
                                 size_t max_workers,
+                                size_t priority_workers,
                                 size_t max_clients,
                                 const char *mdnsGroupName,
                                 bool connectDBus ATTRIBUTE_UNUSED,
@@ -290,6 +293,7 @@ virNetServerPtr virNetServerNew(size_t min_workers,
     srv->refs = 1;
 
     if (!(srv->workers = virThreadPoolNew(min_workers, max_workers,
+                                          priority_workers,
                                           virNetServerHandleJob,
                                           srv)))
         goto error;