Re: [libvirt] [PATCH 6/6 (v2)] Allow non-blocking message sending on virNetClient

From: "Daniel P. Berrange" <berrange(a)redhat.com&gt; Split the existing virNetClientSend into two parts virNetClientSend and virNetClientSendNoReply, instead of having a 'bool expectReply' parameter.

Add a new virNetClientSendNonBlock which returns 2 on full send, 1 on partial send, 0 on no send, -1 on error If a partial send occurs, then a subsequent call to any of the virNetClientSend* APIs will finish any outstanding I/O. TODO: the virNetClientEvent event handler could be used to speed up completion of partial sends if an event loop is present. In v2: - Fix logic in virNetClientIOEventLoopRemoveNonBlocking * src/rpc/virnetclientprogram.c, src/rpc/virnetclientstream.c: Update for changed API

--- src/rpc/virnetclient.c | 200 +++++++++++++++++++++++++++++++++++++++++++----- src/rpc/virnetclient.h | 4 + src/rpc/virnetsocket.c | 13 +++ src/rpc/virnetsocket.h | 1 + 4 files changed, 200 insertions(+), 18 deletions(-) diff --git a/src/rpc/virnetclient.c b/src/rpc/virnetclient.c index 96d1886..17105fe 100644 --- a/src/rpc/virnetclient.c +++ b/src/rpc/virnetclient.c @@ -55,6 +55,9 @@ struct _virNetClientCall { virNetMessagePtr msg; bool expectReply; + bool nonBlock; + bool haveThread; + bool sentSomeData; virCond cond; @@ -86,7 +89,12 @@ struct _virNetClient { int wakeupSendFD; int wakeupReadFD; - /* List of threads currently waiting for dispatch */ + /* + * List of calls currently waiting for dispatch + * The calls should all have threads waiting for + * them, except possibly the first call in the list + * which might be a partially sent non-blocking call. + */ virNetClientCallPtr waitDispatch; /* True if a thread holds the buck */ bool haveTheBuck; @@ -648,7 +656,7 @@ virNetClientCallDispatchReply(virNetClientPtr client) virNetClientCallPtr thecall; /* Ok, definitely got an RPC reply now find - out who's been waiting for it */ + out which waiting call is associated with it */ thecall = client->waitDispatch; while (thecall && !(thecall->msg->header.prog == client->msg.header.prog && @@ -824,6 +832,8 @@ virNetClientIOWriteMessage(virNetClientPtr client, ret = virNetSocketWrite(client->sock, thecall->msg->buffer + thecall->msg->bufferOffset, thecall->msg->bufferLength - thecall->msg->bufferOffset); + if (ret || virNetSocketHasPendingData(client->sock)) + thecall->sentSomeData = true;

if (ret <= 0) return ret; @@ -1015,17 +1025,69 @@ static bool virNetClientIOEventLoopRemoveDone(virNetClientCallPtr call, return false; /* - * ...they won't actually wakeup until + * ...if the call being removed from the list + * still has a thread, then wake that thread up, + * otherwise free the call. The latter should + * only happen for calls without replies. + * + * ...the threads won't actually wakeup until * we release our mutex a short while * later... */ - VIR_DEBUG("Waking up sleeping call %p", call); - virCondSignal(&call->cond); + if (call->haveThread) { + VIR_DEBUG("Waking up sleep %p", call); + virCondSignal(&call->cond); + } else { + if (call->expectReply) + VIR_WARN("Got a call expecting a reply but without a waiting thread"); + ignore_value(virCondDestroy(&call->cond)); + VIR_FREE(call);

+ } return true; } +static bool virNetClientIOEventLoopRemoveNonBlocking(virNetClientCallPtr call, + void *opaque) +{ + virNetClientCallPtr thiscall = opaque; + + if (call == thiscall) + return false; + + if (!call->nonBlock) + return false; + + if (call->sentSomeData) { + /* + * If some data has been sent we must keep it in the list, + * but still wakeup any thread + */ + if (call->haveThread) { + VIR_DEBUG("Waking up sleep %p", call); + virCondSignal(&call->cond); + } + return false; + } else { + /* + * If no data has been sent, we can remove it from the list. + * Wakup any thread, otherwise free the caller ourselves + */ + if (call->haveThread) { + VIR_DEBUG("Waking up sleep %p", call); + virCondSignal(&call->cond); + } else { + if (call->expectReply) + VIR_WARN("Got a call expecting a reply but without a waiting thread"); + ignore_value(virCondDestroy(&call->cond)); + VIR_FREE(call);

+ } + return true; + } +} + + static void virNetClientIOEventLoopPassTheBuck(virNetClientPtr client, virNetClientCallPtr thiscall) { VIR_DEBUG("Giving up the buck %p", thiscall); @@ -1033,19 +1095,29 @@ static void virNetClientIOEventLoopPassTheBuck(virNetClientPtr client, virNetCli /* See if someone else is still waiting * and if so, then pass the buck ! */ while (tmp) { - if (tmp != thiscall) { + if (tmp != thiscall && tmp->haveThread) { VIR_DEBUG("Passing the buck to %p", tmp); virCondSignal(&tmp->cond); break; } tmp = tmp->next; } + VIR_DEBUG("No thread to pass the buck to"); +} + + +static bool virNetClientIOEventLoopWantNonBlock(virNetClientCallPtr call, void *opaque ATTRIBUTE_UNUSED) +{ + return call->nonBlock; } /* * Process all calls pending dispatch/receive until we * get a reply to our own call. Then quit and pass the buck * to someone else. + * + * Returns 2 if fully sent, 1 if partially sent (only for nonBlock==true), + * 0 if nothing sent (only for nonBlock==true) and -1 on error */ static int virNetClientIOEventLoop(virNetClientPtr client, virNetClientCallPtr thiscall) @@ -1068,6 +1140,14 @@ static int virNetClientIOEventLoop(virNetClientPtr client, if (virNetSocketHasCachedData(client->sock)) timeout = 0; + /* If there are any non-blocking calls in the queue, + * then we don't want to sleep in poll() + */ + if (virNetClientCallMatchPredicate(client->waitDispatch, + virNetClientIOEventLoopWantNonBlock, + NULL)) + timeout = 0; + fds[0].events = fds[0].revents = 0; fds[1].events = fds[1].revents = 0; @@ -1116,8 +1196,9 @@ static int virNetClientIOEventLoop(virNetClientPtr client, /* If we have existing SASL decoded data, pretend * the socket became readable so we consume it */ - if (virNetSocketHasCachedData(client->sock)) + if (virNetSocketHasCachedData(client->sock)) { fds[0].revents |= POLLIN; + } if (fds[1].revents) { VIR_DEBUG("Woken up from poll by other thread"); @@ -1129,6 +1210,7 @@ static int virNetClientIOEventLoop(virNetClientPtr client, } if (ret < 0) { + /* XXX what's this dubious errno check doing ? */ if (errno == EWOULDBLOCK) continue;

virReportSystemError(errno, @@ -1146,20 +1228,33 @@ static int virNetClientIOEventLoop(virNetClientPtr client, goto error; } - /* Iterate through waiting threads and if - * any are complete then tell 'em to wakeup + /* Iterate through waiting calls and if any are + * complete, remove them from the dispatch list.. */ virNetClientCallRemovePredicate(&client->waitDispatch, virNetClientIOEventLoopRemoveDone, thiscall); + /* Iterate through waiting calls and if any are + * non-blocking, remove them from the dispatch list... + */ + virNetClientCallRemovePredicate(&client->waitDispatch, + virNetClientIOEventLoopRemoveNonBlocking, + thiscall); + /* Now see if *we* are done */ if (thiscall->mode == VIR_NET_CLIENT_MODE_COMPLETE) { virNetClientCallRemove(&client->waitDispatch, thiscall); virNetClientIOEventLoopPassTheBuck(client, thiscall); - return 0; + return 2; } + /* We're not done, but we're non-blocking */ + if (thiscall->nonBlock) { + VIR_DEBUG("Giving up the buck %p %p", thiscall, client->waitDispatch);

+ virNetClientIOEventLoopPassTheBuck(client, thiscall); + return thiscall->sentSomeData ? 1 : 0; + } if (fds[0].revents & (POLLHUP | POLLERR)) { virNetError(VIR_ERR_INTERNAL_ERROR, "%s",