Re: [libvirt] [PATCH 02/12] client rpc: Use event loop for writing

Normally, when every call has a thread associated with it, the thread may get the buck and be in charge of sending all calls until its own call is done. When we introduced non-blocking calls, we had to add special handling of new non-blocking calls. This patch uses event loop to send data if there is no thread to get the buck so that any non-blocking calls left in the queue are properly sent without having to handle them specially. It also avoids adding even more cruft to client IO loop in the following patches. With this change in, non-blocking calls may see unpredictable delays in delivery when the client has no event loop registered. However, the only non-blocking calls we have are keepalives and we already require event loop for them.

--- src/rpc/virnetclient.c | 35 +++++++++++++++++++++++++++++++---- 1 file changed, 31 insertions(+), 4 deletions(-) diff --git a/src/rpc/virnetclient.c b/src/rpc/virnetclient.c index c62e045..3e661d2 100644 --- a/src/rpc/virnetclient.c +++ b/src/rpc/virnetclient.c @@ -1470,12 +1470,30 @@ error: } +static bool +virNetClientIOUpdateEvents(virNetClientCallPtr call, + void *opaque) +{ + int *events = opaque; + + if (call->mode == VIR_NET_CLIENT_MODE_WAIT_TX) + *events |= VIR_EVENT_HANDLE_WRITABLE; + + return false; +} + + static void virNetClientIOUpdateCallback(virNetClientPtr client, bool enableCallback) { int events = 0; - if (enableCallback) + + if (enableCallback) { events |= VIR_EVENT_HANDLE_READABLE; + virNetClientCallMatchPredicate(client->waitDispatch, + virNetClientIOUpdateEvents, + &events); + } virNetSocketUpdateIOCallback(client->sock, events); } @@ -1670,11 +1688,20 @@ void virNetClientIncomingEvent(virNetSocketPtr sock, goto done; } - if (virNetClientIOHandleInput(client) < 0) { - VIR_WARN("Something went wrong during async message processing"); - virNetSocketRemoveIOCallback(sock); + if (events & VIR_EVENT_HANDLE_WRITABLE) { + if (virNetClientIOHandleOutput(client) < 0) + virNetSocketRemoveIOCallback(sock); + } + + if (events & VIR_EVENT_HANDLE_READABLE) { + if (virNetClientIOHandleInput(client) < 0) + virNetSocketRemoveIOCallback(sock); } + /* Remove completed calls or signal their threads. */ + virNetClientCallRemovePredicate(&client->waitDispatch, + virNetClientIOEventLoopRemoveDone, + NULL); done: virNetClientUnlock(client); }