diff --git a/src/rpc/virnetserver.c b/src/rpc/virnetserver.c index 358666d2488b16e246a7f67881f3fafb560fa805..4a02aabc595fe2cce806b4f3409d4043939773b8 100644 --- a/src/rpc/virnetserver.c +++ b/src/rpc/virnetserver.c @@ -127,49 +127,64 @@ static void virNetServerUnlock(virNetServerPtr srv) } -static void virNetServerHandleJob(void *jobOpaque, void *opaque) +static int virNetServerProcessMsg(virNetServerPtr srv, + virNetServerClientPtr client, + virNetServerProgramPtr prog, + virNetMessagePtr msg) { - virNetServerPtr srv = opaque; - virNetServerJobPtr job = jobOpaque; - - VIR_DEBUG("server=%p client=%p message=%p prog=%p", - srv, job->client, job->msg, job->prog); - - if (!job->prog) { + int ret = -1; + if (!prog) { /* Only send back an error for type == CALL. Other * message types are not expecting replies, so we * must just log it & drop them */ - if (job->msg->header.type == VIR_NET_CALL || - job->msg->header.type == VIR_NET_CALL_WITH_FDS) { - if (virNetServerProgramUnknownError(job->client, - job->msg, - &job->msg->header) < 0) - goto error; + if (msg->header.type == VIR_NET_CALL || + msg->header.type == VIR_NET_CALL_WITH_FDS) { + if (virNetServerProgramUnknownError(client, + msg, + &msg->header) < 0) + goto cleanup; } else { VIR_INFO("Dropping client mesage, unknown program %d version %d type %d proc %d", - job->msg->header.prog, job->msg->header.vers, - job->msg->header.type, job->msg->header.proc); + msg->header.prog, msg->header.vers, + msg->header.type, msg->header.proc); /* Send a dummy reply to free up 'msg' & unblock client rx */ - virNetMessageClear(job->msg); - job->msg->header.type = VIR_NET_REPLY; - if (virNetServerClientSendMessage(job->client, job->msg) < 0) - goto error; + virNetMessageClear(msg); + msg->header.type = VIR_NET_REPLY; + if (virNetServerClientSendMessage(client, msg) < 0) + goto cleanup; } - goto cleanup; + goto done; } - if (virNetServerProgramDispatch(job->prog, + if (virNetServerProgramDispatch(prog, srv, - job->client, - job->msg) < 0) + client, + msg) < 0) + goto cleanup; + +done: + ret = 0; + +cleanup: + return ret; +} + +static void virNetServerHandleJob(void *jobOpaque, void *opaque) +{ + virNetServerPtr srv = opaque; + virNetServerJobPtr job = jobOpaque; + + VIR_DEBUG("server=%p client=%p message=%p prog=%p", + srv, job->client, job->msg, job->prog); + + if (virNetServerProcessMsg(srv, job->client, job->prog, job->msg) < 0) goto error; virNetServerLock(srv); virNetServerProgramFree(job->prog); virNetServerUnlock(srv); -cleanup: virNetServerClientFree(job->client); VIR_FREE(job); return; @@ -187,7 +202,6 @@ static int virNetServerDispatchNewMessage(virNetServerClientPtr client, void *opaque) { virNetServerPtr srv = opaque; - virNetServerJobPtr job; virNetServerProgramPtr prog = NULL; unsigned int priority = 0; size_t i; @@ -196,34 +210,42 @@ static int virNetServerDispatchNewMessage(virNetServerClientPtr client, VIR_DEBUG("server=%p client=%p message=%p", srv, client, msg); - if (VIR_ALLOC(job) < 0) { - virReportOOMError(); - return -1; - } - - job->client = client; - job->msg = msg; - virNetServerLock(srv); for (i = 0 ; i < srv->nprograms ; i++) { - if (virNetServerProgramMatches(srv->programs[i], job->msg)) { + if (virNetServerProgramMatches(srv->programs[i], msg)) { prog = srv->programs[i]; break; } } - if (prog) { - virNetServerProgramRef(prog); - job->prog = prog; - priority = virNetServerProgramGetPriority(prog, msg->header.proc); - } + if (srv->workers) { + virNetServerJobPtr job; + + if (VIR_ALLOC(job) < 0) { + virReportOOMError(); + goto cleanup; + } - ret = virThreadPoolSendJob(srv->workers, priority, job); + job->client = client; + job->msg = msg; + + if (prog) { + virNetServerProgramRef(prog); + job->prog = prog; + priority = virNetServerProgramGetPriority(prog, msg->header.proc); + } - if (ret < 0) { - VIR_FREE(job); - virNetServerProgramFree(prog); + ret = virThreadPoolSendJob(srv->workers, priority, job); + + if (ret < 0) { + VIR_FREE(job); + virNetServerProgramFree(prog); + } + } else { + ret = virNetServerProcessMsg(srv, client, prog, msg); } + +cleanup: virNetServerUnlock(srv); return ret; @@ -324,7 +346,8 @@ virNetServerPtr virNetServerNew(size_t min_workers, srv->refs = 1; - if (!(srv->workers = virThreadPoolNew(min_workers, max_workers, + if (max_workers && + !(srv->workers = virThreadPoolNew(min_workers, max_workers, priority_workers, virNetServerHandleJob, srv)))