6:29 p.m.
So far, we were dropping non-blocking calls whenever sending them would block.
In case a client is sending lots of stream calls (which are not supposed to
generate any reply), the assumption that having other calls in a queue is
sufficient to get a reply from the server doesn't work. I tried to fix this in
b1e374a7ac56927cfe62435179bf0bba1e08b372 but failed and reverted that commit.
While working on the proper fix, I discovered several other issues we had in
handling keepalive messages in client RPC code. See individual patches for more
details.
As a nice bonus, the fixed version is shorter by one line than the current
broken version :-)
Jiri Denemark (12):
client rpc: Improve debug messages in virNetClientIO
client rpc: Use event loop for writing
client rpc: Don't drop non-blocking calls
client rpc: Just queue non-blocking call if another thread has the
buck
client rpc: Drop unused return value of virNetClientSendNonBlock
rpc: Refactor keepalive timer code
rpc: Add APIs for direct triggering of keepalive timer
client rpc: Separate call creation from running IO loop
rpc: Do not use timer for sending keepalive responses
rpc: Remove unused parameter in virKeepAliveStopInternal
server rpc: Remove APIs for manipulating filters on locked client
client rpc: Send keepalive requests from IO event loop
src/libvirt_probes.d | 2 +-
src/rpc/virkeepalive.c | 233 ++++++++++++--------------
src/rpc/virkeepalive.h | 7 +-
src/rpc/virnetclient.c | 368 +++++++++++++++++++++++-------------------
src/rpc/virnetserverclient.c | 127 +++++++--------
5 files changed, 368 insertions(+), 369 deletions(-)
--
1.7.10.2
6:29 p.m.
New subject: [libvirt] [PATCH 01/12] client rpc: Improve debug messages in virNetClientIO
When analyzing our debug log, I'm always confused about what each of the
pointers mean. Let's be explicit.
---
src/rpc/virnetclient.c | 12 ++++++++----
1 file changed, 8 insertions(+), 4 deletions(-)
diff --git a/src/rpc/virnetclient.c b/src/rpc/virnetclient.c
index 14f806f..c62e045 100644
--- a/src/rpc/virnetclient.c
+++ b/src/rpc/virnetclient.c
@@ -1567,7 +1567,8 @@ static int virNetClientIO(virNetClientPtr client,
return -1;
}
- VIR_DEBUG("Going to sleep %p %p", client->waitDispatch, thiscall);
+ VIR_DEBUG("Going to sleep head=%p call=%p",
+ client->waitDispatch, thiscall);
/* Go to sleep while other thread is working... */
if (virCondWait(&thiscall->cond, &client->lock) < 0) {
virNetClientCallRemove(&client->waitDispatch, thiscall);
@@ -1576,7 +1577,8 @@ static int virNetClientIO(virNetClientPtr client,
return -1;
}
- VIR_DEBUG("Wokeup from sleep %p %p", client->waitDispatch,
thiscall);
+ VIR_DEBUG("Woken up from sleep head=%p call=%p",
+ client->waitDispatch, thiscall);
/* Three reasons we can be woken up
* 1. Other thread has got our reply ready for us
* 2. Other thread is all done, and it is our turn to
@@ -1608,7 +1610,8 @@ static int virNetClientIO(virNetClientPtr client,
client->haveTheBuck = true;
}
- VIR_DEBUG("We have the buck %p %p", client->waitDispatch, thiscall);
+ VIR_DEBUG("We have the buck head=%p call=%p",
+ client->waitDispatch, thiscall);
/*
* The buck stops here!
@@ -1637,7 +1640,8 @@ static int virNetClientIO(virNetClientPtr client,
rv = -1;
cleanup:
- VIR_DEBUG("All done with our call %p %p %d", client->waitDispatch,
thiscall, rv);
+ VIR_DEBUG("All done with our call head=%p call=%p rv=%d",
+ client->waitDispatch, thiscall, rv);
return rv;
}
--
1.7.10.2
6:37 p.m.
New subject: [libvirt] [PATCH 01/12] client rpc: Improve debug messages in virNetClientIO
On 06/12/2012 05:29 PM, Jiri Denemark wrote:
When analyzing our debug log, I'm always confused about what each
of the
pointers mean. Let's be explicit.
---
src/rpc/virnetclient.c | 12 ++++++++----
1 file changed, 8 insertions(+), 4 deletions(-)
ACK. No semantic change, and better debug can't hurt.
--
Eric Blake eblake(a)redhat.com +1-919-301-3266
Libvirt virtualization library http://libvirt.org
3:27 a.m.
New subject: [libvirt] [PATCH 01/12] client rpc: Improve debug messages in virNetClientIO
On Wed, Jun 13, 2012 at 01:29:19AM +0200, Jiri Denemark wrote:
When analyzing our debug log, I'm always confused about what each
of the
pointers mean. Let's be explicit.
---
src/rpc/virnetclient.c | 12 ++++++++----
1 file changed, 8 insertions(+), 4 deletions(-)
ACK
Daniel
--
|: http://berrange.com -o- http://www.flickr.com/photos/dberrange/ :|
|: http://libvirt.org -o- http://virt-manager.org :|
|: http://autobuild.org -o- http://search.cpan.org/~danberr/ :|
|: http://entangle-photo.org -o- http://live.gnome.org/gtk-vnc :|
6:29 p.m.
New subject: [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);
}
--
1.7.10.2
3:30 a.m.
New subject: [libvirt] [PATCH 02/12] client rpc: Use event loop for writing
On Wed, Jun 13, 2012 at 01:29:20AM +0200, Jiri Denemark wrote:
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.
Is that 'see unpredictable delays' part really correct. AFAIK, there
should be a pretty well defined "delay" - it'll be processed on the very
next iteration of the event - assuming the socket is writable. I don't
really thing this is a delay at all in fact.
---
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);
}
ACK, I always thought we should be doing this :-)
Daniel
--
|: http://berrange.com -o- http://www.flickr.com/photos/dberrange/ :|
|: http://libvirt.org -o- http://virt-manager.org :|
|: http://autobuild.org -o- http://search.cpan.org/~danberr/ :|
|: http://entangle-photo.org -o- http://live.gnome.org/gtk-vnc :|
3:40 a.m.
New subject: [libvirt] [PATCH 02/12] client rpc: Use event loop for writing
On Wed, Jun 13, 2012 at 09:30:30 +0100, Daniel P. Berrange wrote:
On Wed, Jun 13, 2012 at 01:29:20AM +0200, Jiri Denemark wrote:
> 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.
Is that 'see unpredictable delays' part really correct. AFAIK, there
should be a pretty well defined "delay" - it'll be processed on the very
next iteration of the event - assuming the socket is writable. I don't
really thing this is a delay at all in fact.
OK, it's unpredictable but in the case of keepalive calls the delay is at most
keepalive interval. The call may be processed earlier if a libvirt API is
called in the meantime. I'll reword it a bit.
Jirka
3:43 a.m.
New subject: [libvirt] [PATCH 02/12] client rpc: Use event loop for writing
On Wed, Jun 13, 2012 at 10:40:48AM +0200, Jiri Denemark wrote:
On Wed, Jun 13, 2012 at 09:30:30 +0100, Daniel P. Berrange wrote:
> On Wed, Jun 13, 2012 at 01:29:20AM +0200, Jiri Denemark wrote:
> > 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.
>
> Is that 'see unpredictable delays' part really correct. AFAIK, there
> should be a pretty well defined "delay" - it'll be processed on the
very
> next iteration of the event - assuming the socket is writable. I don't
> really thing this is a delay at all in fact.
OK, it's unpredictable but in the case of keepalive calls the delay is at most
keepalive interval. The call may be processed earlier if a libvirt API is
called in the meantime. I'll reword it a bit.
Doesn't the fact that we use the event loop for writing mean that we
are not needing to wait for a libvirt API to be called any more ?
Daniel
--
|: http://berrange.com -o- http://www.flickr.com/photos/dberrange/ :|
|: http://libvirt.org -o- http://virt-manager.org :|
|: http://autobuild.org -o- http://search.cpan.org/~danberr/ :|
|: http://entangle-photo.org -o- http://live.gnome.org/gtk-vnc :|
3:50 a.m.
New subject: [libvirt] [PATCH 02/12] client rpc: Use event loop for writing
On Wed, Jun 13, 2012 at 09:43:54 +0100, Daniel P. Berrange wrote:
On Wed, Jun 13, 2012 at 10:40:48AM +0200, Jiri Denemark wrote:
> On Wed, Jun 13, 2012 at 09:30:30 +0100, Daniel P. Berrange wrote:
> > On Wed, Jun 13, 2012 at 01:29:20AM +0200, Jiri Denemark wrote:
> > > 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.
> >
> > Is that 'see unpredictable delays' part really correct. AFAIK, there
> > should be a pretty well defined "delay" - it'll be processed on
the very
> > next iteration of the event - assuming the socket is writable. I don't
> > really thing this is a delay at all in fact.
>
> OK, it's unpredictable but in the case of keepalive calls the delay is at most
> keepalive interval. The call may be processed earlier if a libvirt API is
> called in the meantime. I'll reword it a bit.
Doesn't the fact that we use the event loop for writing mean that we
are not needing to wait for a libvirt API to be called any more ?
Yes, but the last paragraph was specifically talking about the case when the
client has no event loop registered, which we still need to support for
backward compatibility. However, as already said, we require event loop for
keepalive and keepalive calls are the only non-blocking calls we have. Thus it
was just a note for the future if someone ever introduces another type of
non-blocking calls.
Jirka
4:04 a.m.
New subject: [libvirt] [PATCH 02/12] client rpc: Use event loop for writing
On Wed, Jun 13, 2012 at 10:50:09AM +0200, Jiri Denemark wrote:
On Wed, Jun 13, 2012 at 09:43:54 +0100, Daniel P. Berrange wrote:
> On Wed, Jun 13, 2012 at 10:40:48AM +0200, Jiri Denemark wrote:
> > On Wed, Jun 13, 2012 at 09:30:30 +0100, Daniel P. Berrange wrote:
> > > On Wed, Jun 13, 2012 at 01:29:20AM +0200, Jiri Denemark wrote:
> > > > 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.
> > >
> > > Is that 'see unpredictable delays' part really correct. AFAIK,
there
> > > should be a pretty well defined "delay" - it'll be processed
on the very
> > > next iteration of the event - assuming the socket is writable. I
don't
> > > really thing this is a delay at all in fact.
> >
> > OK, it's unpredictable but in the case of keepalive calls the delay is at
most
> > keepalive interval. The call may be processed earlier if a libvirt API is
> > called in the meantime. I'll reword it a bit.
>
> Doesn't the fact that we use the event loop for writing mean that we
> are not needing to wait for a libvirt API to be called any more ?
Yes, but the last paragraph was specifically talking about the case when the
client has no event loop registered, which we still need to support for
backward compatibility. However, as already said, we require event loop for
keepalive and keepalive calls are the only non-blocking calls we have. Thus it
was just a note for the future if someone ever introduces another type of
non-blocking calls.
Ahhhh, I see. It wasn't clear that you were refering to the no-eventloop
case.
Daniel
--
|: http://berrange.com -o- http://www.flickr.com/photos/dberrange/ :|
|: http://libvirt.org -o- http://virt-manager.org :|
|: http://autobuild.org -o- http://search.cpan.org/~danberr/ :|
|: http://entangle-photo.org -o- http://live.gnome.org/gtk-vnc :|
6:29 p.m.
New subject: [libvirt] [PATCH 03/12] client rpc: Don't drop non-blocking calls
So far, we were dropping non-blocking calls whenever sending them would
block. In case a client is sending lots of stream calls (which are not
supposed to generate any reply), the assumption that having other calls
in a queue is sufficient to get a reply from the server doesn't work. I
tried to fix this in b1e374a7ac56927cfe62435179bf0bba1e08b372 but
failed and reverted that commit.
With this patch, non-blocking calls are never dropped (unless the
connection is being closed) and will always be sent.
---
src/rpc/virnetclient.c | 164 +++++++++++++++++++++---------------------------
1 file changed, 71 insertions(+), 93 deletions(-)
diff --git a/src/rpc/virnetclient.c b/src/rpc/virnetclient.c
index 3e661d2..614b469 100644
--- a/src/rpc/virnetclient.c
+++ b/src/rpc/virnetclient.c
@@ -58,7 +58,6 @@ struct _virNetClientCall {
bool expectReply;
bool nonBlock;
bool haveThread;
- bool sentSomeData;
virCond cond;
@@ -108,6 +107,10 @@ struct _virNetClient {
};
+static void virNetClientIOEventLoopPassTheBuck(virNetClientPtr client,
+ virNetClientCallPtr thiscall);
+
+
static void virNetClientLock(virNetClientPtr client)
{
virMutexLock(&client->lock);
@@ -525,19 +528,21 @@ void virNetClientClose(virNetClientPtr client)
virNetClientLock(client);
- /* If there is a thread polling for data on the socket, set wantClose flag
- * and wake the thread up or just immediately close the socket when no-one
- * is polling on it.
+ client->wantClose = true;
+
+ /* If there is a thread polling for data on the socket, wake the thread up
+ * otherwise try to pass the buck to a possibly waiting thread. If no
+ * thread is waiting, virNetClientIOEventLoopPassTheBuck will clean the
+ * queue and close the client because we set client->wantClose.
*/
- if (client->waitDispatch) {
+ if (client->haveTheBuck) {
char ignore = 1;
size_t len = sizeof(ignore);
- client->wantClose = true;
if (safewrite(client->wakeupSendFD, &ignore, len) != len)
VIR_ERROR(_("failed to wake up polling thread"));
} else {
- virNetClientCloseLocked(client);
+ virNetClientIOEventLoopPassTheBuck(client, NULL);
}
virNetClientUnlock(client);
@@ -972,8 +977,6 @@ virNetClientIOWriteMessage(virNetClientPtr client,
ret = virNetSocketWrite(client->sock,
thecall->msg->buffer +
thecall->msg->bufferOffset,
thecall->msg->bufferLength -
thecall->msg->bufferOffset);
- if (ret > 0 || virNetSocketHasPendingData(client->sock))
- thecall->sentSomeData = true;
if (ret <= 0)
return ret;
@@ -1197,71 +1200,43 @@ static bool virNetClientIOEventLoopRemoveDone(virNetClientCallPtr
call,
}
-static bool virNetClientIOEventLoopRemoveNonBlocking(virNetClientCallPtr call,
- void *opaque)
+static bool
+virNetClientIOEventLoopDetachNonBlocking(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);
- } else {
- VIR_DEBUG("Keeping unfinished call %p in the list", call);
- }
- 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 {
- VIR_DEBUG("Removing call %p", call);
- if (call->expectReply)
- VIR_WARN("Got a call expecting a reply but without a waiting
thread");
- ignore_value(virCondDestroy(&call->cond));
- VIR_FREE(call->msg);
- VIR_FREE(call);
- }
+ if (call != thiscall && call->nonBlock && call->haveThread) {
+ VIR_DEBUG("Waking up sleep %p", call);
+ call->haveThread = false;
+ virCondSignal(&call->cond);
return true;
}
+
+ return false;
}
-static void
-virNetClientIOEventLoopRemoveAll(virNetClientPtr client,
- virNetClientCallPtr thiscall)
+static bool
+virNetClientIOEventLoopRemoveAll(virNetClientCallPtr call,
+ void *opaque)
{
- if (!client->waitDispatch)
- return;
+ virNetClientCallPtr thiscall = opaque;
- if (client->waitDispatch == thiscall) {
- /* just pretend nothing was sent and the caller will free the call */
- thiscall->sentSomeData = false;
- } else {
- virNetClientCallPtr call = client->waitDispatch;
- virNetClientCallRemove(&client->waitDispatch, call);
- ignore_value(virCondDestroy(&call->cond));
- VIR_FREE(call->msg);
- VIR_FREE(call);
- }
+ if (call == thiscall)
+ return false;
+
+ VIR_DEBUG("Removing call %p", call);
+ ignore_value(virCondDestroy(&call->cond));
+ VIR_FREE(call->msg);
+ VIR_FREE(call);
+ return true;
}
-static void virNetClientIOEventLoopPassTheBuck(virNetClientPtr client,
virNetClientCallPtr thiscall)
+static void
+virNetClientIOEventLoopPassTheBuck(virNetClientPtr client,
+ virNetClientCallPtr thiscall)
{
VIR_DEBUG("Giving up the buck %p", thiscall);
virNetClientCallPtr tmp = client->waitDispatch;
@@ -1280,14 +1255,18 @@ static void virNetClientIOEventLoopPassTheBuck(virNetClientPtr
client, virNetCli
VIR_DEBUG("No thread to pass the buck to");
if (client->wantClose) {
virNetClientCloseLocked(client);
- virNetClientIOEventLoopRemoveAll(client, thiscall);
+ virNetClientCallRemovePredicate(&client->waitDispatch,
+ virNetClientIOEventLoopRemoveAll,
+ thiscall);
}
}
-static bool virNetClientIOEventLoopWantNonBlock(virNetClientCallPtr call, void *opaque
ATTRIBUTE_UNUSED)
+static bool
+virNetClientIOEventLoopWantNonBlock(virNetClientCallPtr call,
+ void *opaque ATTRIBUTE_UNUSED)
{
- return call->nonBlock;
+ return call->nonBlock && call->haveThread;
}
/*
@@ -1320,8 +1299,8 @@ static int virNetClientIOEventLoop(virNetClientPtr client,
if (virNetSocketHasCachedData(client->sock) || client->wantClose)
timeout = 0;
- /* If there are any non-blocking calls in the queue,
- * then we don't want to sleep in poll()
+ /* If there are any non-blocking calls with an associated thread
+ * in the queue, then we don't want to sleep in poll()
*/
if (virNetClientCallMatchPredicate(client->waitDispatch,
virNetClientIOEventLoopWantNonBlock,
@@ -1394,12 +1373,15 @@ static int virNetClientIOEventLoop(virNetClientPtr client,
}
/* If we were woken up because a new non-blocking call was queued,
- * we need to re-poll to check if we can send it.
+ * we need to re-poll to check if we can send it. To be precise, we
+ * will re-poll even if a blocking call arrived when unhandled
+ * non-blocking calls are still in the queue. But this can't hurt.
*/
if (virNetClientCallMatchPredicate(client->waitDispatch,
virNetClientIOEventLoopWantNonBlock,
NULL)) {
- VIR_DEBUG("New non-blocking call arrived; repolling");
+ VIR_DEBUG("The queue contains new non-blocking call(s);"
+ " repolling");
continue;
}
}
@@ -1424,18 +1406,18 @@ static int virNetClientIOEventLoop(virNetClientPtr client,
}
/* Iterate through waiting calls and if any are
- * complete, remove them from the dispatch list..
+ * 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...
+ /* Iterate through waiting calls and wake up and detach threads
+ * attached to non-blocking calls.
*/
- virNetClientCallRemovePredicate(&client->waitDispatch,
- virNetClientIOEventLoopRemoveNonBlocking,
- thiscall);
+ virNetClientCallMatchPredicate(client->waitDispatch,
+ virNetClientIOEventLoopDetachNonBlocking,
+ thiscall);
/* Now see if *we* are done */
if (thiscall->mode == VIR_NET_CLIENT_MODE_COMPLETE) {
@@ -1444,15 +1426,11 @@ static int virNetClientIOEventLoop(virNetClientPtr client,
return 2;
}
- /* We're not done, but we're non-blocking */
+ /* We're not done, but we're non-blocking; keep the call queued */
if (thiscall->nonBlock) {
+ thiscall->haveThread = false;
virNetClientIOEventLoopPassTheBuck(client, thiscall);
- if (thiscall->sentSomeData) {
- return 1;
- } else {
- virNetClientCallRemove(&client->waitDispatch, thiscall);
- return 0;
- }
+ return 1;
}
if (fds[0].revents & (POLLHUP | POLLERR)) {
@@ -1462,7 +1440,6 @@ static int virNetClientIOEventLoop(virNetClientPtr client,
}
}
-
error:
virNetClientCallRemove(&client->waitDispatch, thiscall);
virNetClientIOEventLoopPassTheBuck(client, thiscall);
@@ -1614,9 +1591,11 @@ static int virNetClientIO(virNetClientPtr client,
goto cleanup;
}
- /* If we're non-blocking, get outta here */
+ /* If we're non-blocking, we were either queued (and detached) or the
+ * call was not sent because of an error.
+ */
if (thiscall->nonBlock) {
- if (thiscall->sentSomeData)
+ if (!thiscall->haveThread)
rv = 1; /* In progress */
else
rv = 0; /* none at all */
@@ -1708,7 +1687,7 @@ done:
/*
- * Returns 2 if fully sent, 1 if partially sent (only for nonBlock==true),
+ * Returns 2 if fully sent, 1 if queued (only for nonBlock==true),
* 0 if nothing sent (only for nonBlock==true) and -1 on error
*/
static int virNetClientSendInternal(virNetClientPtr client,
@@ -1768,16 +1747,15 @@ static int virNetClientSendInternal(virNetClientPtr client,
ret = virNetClientIO(client, call);
- /* If partially sent, then the call is still on the dispatch queue */
- if (ret == 1) {
- call->haveThread = false;
- } else {
- ignore_value(virCondDestroy(&call->cond));
- }
+ /* If queued, the call will be finished and freed later by another thread;
+ * we're done. */
+ if (ret == 1)
+ return 1;
+
+ ignore_value(virCondDestroy(&call->cond));
cleanup:
- if (ret != 1)
- VIR_FREE(call);
+ VIR_FREE(call);
return ret;
}
--
1.7.10.2
4:47 a.m.
New subject: [libvirt] [PATCH 03/12] client rpc: Don't drop non-blocking calls
On Wed, Jun 13, 2012 at 01:29:21AM +0200, Jiri Denemark wrote:
So far, we were dropping non-blocking calls whenever sending them
would
block. In case a client is sending lots of stream calls (which are not
supposed to generate any reply), the assumption that having other calls
in a queue is sufficient to get a reply from the server doesn't work. I
tried to fix this in b1e374a7ac56927cfe62435179bf0bba1e08b372 but
failed and reverted that commit.
With this patch, non-blocking calls are never dropped (unless the
connection is being closed) and will always be sent.
---
src/rpc/virnetclient.c | 164 +++++++++++++++++++++---------------------------
1 file changed, 71 insertions(+), 93 deletions(-)
ACK, quite complex to follow but the principle is good - dropping calls
always felt slightly wrong.
Daniel
--
|: http://berrange.com -o- http://www.flickr.com/photos/dberrange/ :|
|: http://libvirt.org -o- http://virt-manager.org :|
|: http://autobuild.org -o- http://search.cpan.org/~danberr/ :|
|: http://entangle-photo.org -o- http://live.gnome.org/gtk-vnc :|
6:29 p.m.
New subject: [libvirt] [PATCH 04/12] client rpc: Just queue non-blocking call if another thread has the buck
As non-blocking calls are no longer dropped, we don't really need to
care that much about their faith and wait for the thread with the buck
to process them. If another thread has the buck, we can just push a
non-blocking call to the queue and be done with it.
---
src/rpc/virnetclient.c | 74 ++++++++++--------------------------------------
1 file changed, 15 insertions(+), 59 deletions(-)
diff --git a/src/rpc/virnetclient.c b/src/rpc/virnetclient.c
index 614b469..b3d8d14 100644
--- a/src/rpc/virnetclient.c
+++ b/src/rpc/virnetclient.c
@@ -1200,20 +1200,11 @@ static bool virNetClientIOEventLoopRemoveDone(virNetClientCallPtr
call,
}
-static bool
-virNetClientIOEventLoopDetachNonBlocking(virNetClientCallPtr call,
- void *opaque)
+static void
+virNetClientIODetachNonBlocking(virNetClientCallPtr call)
{
- virNetClientCallPtr thiscall = opaque;
-
- if (call != thiscall && call->nonBlock && call->haveThread) {
- VIR_DEBUG("Waking up sleep %p", call);
- call->haveThread = false;
- virCondSignal(&call->cond);
- return true;
- }
-
- return false;
+ VIR_DEBUG("Keeping unfinished non-blocking call %p in the queue", call);
+ call->haveThread = false;
}
@@ -1262,13 +1253,6 @@ virNetClientIOEventLoopPassTheBuck(virNetClientPtr client,
}
-static bool
-virNetClientIOEventLoopWantNonBlock(virNetClientCallPtr call,
- void *opaque ATTRIBUTE_UNUSED)
-{
- return call->nonBlock && call->haveThread;
-}
-
/*
* Process all calls pending dispatch/receive until we
* get a reply to our own call. Then quit and pass the buck
@@ -1299,12 +1283,8 @@ static int virNetClientIOEventLoop(virNetClientPtr client,
if (virNetSocketHasCachedData(client->sock) || client->wantClose)
timeout = 0;
- /* If there are any non-blocking calls with an associated thread
- * in the queue, then we don't want to sleep in poll()
- */
- if (virNetClientCallMatchPredicate(client->waitDispatch,
- virNetClientIOEventLoopWantNonBlock,
- NULL))
+ /* If we are non-blocking, then we don't want to sleep in poll() */
+ if (thiscall->nonBlock)
timeout = 0;
fds[0].events = fds[0].revents = 0;
@@ -1371,19 +1351,6 @@ static int virNetClientIOEventLoop(virNetClientPtr client,
_("read on wakeup fd failed"));
goto error;
}
-
- /* If we were woken up because a new non-blocking call was queued,
- * we need to re-poll to check if we can send it. To be precise, we
- * will re-poll even if a blocking call arrived when unhandled
- * non-blocking calls are still in the queue. But this can't hurt.
- */
- if (virNetClientCallMatchPredicate(client->waitDispatch,
- virNetClientIOEventLoopWantNonBlock,
- NULL)) {
- VIR_DEBUG("The queue contains new non-blocking call(s);"
- " repolling");
- continue;
- }
}
if (ret < 0) {
@@ -1412,13 +1379,6 @@ static int virNetClientIOEventLoop(virNetClientPtr client,
virNetClientIOEventLoopRemoveDone,
thiscall);
- /* Iterate through waiting calls and wake up and detach threads
- * attached to non-blocking calls.
- */
- virNetClientCallMatchPredicate(client->waitDispatch,
- virNetClientIOEventLoopDetachNonBlocking,
- thiscall);
-
/* Now see if *we* are done */
if (thiscall->mode == VIR_NET_CLIENT_MODE_COMPLETE) {
virNetClientCallRemove(&client->waitDispatch, thiscall);
@@ -1428,7 +1388,7 @@ static int virNetClientIOEventLoop(virNetClientPtr client,
/* We're not done, but we're non-blocking; keep the call queued */
if (thiscall->nonBlock) {
- thiscall->haveThread = false;
+ virNetClientIODetachNonBlocking(thiscall);
virNetClientIOEventLoopPassTheBuck(client, thiscall);
return 1;
}
@@ -1562,6 +1522,14 @@ static int virNetClientIO(virNetClientPtr client,
return -1;
}
+ /* If we are non-blocking, detach the thread and keep the call in the
+ * queue. */
+ if (thiscall->nonBlock) {
+ virNetClientIODetachNonBlocking(thiscall);
+ rv = 1;
+ goto cleanup;
+ }
+
VIR_DEBUG("Going to sleep head=%p call=%p",
client->waitDispatch, thiscall);
/* Go to sleep while other thread is working... */
@@ -1579,7 +1547,6 @@ static int virNetClientIO(virNetClientPtr client,
* 2. Other thread is all done, and it is our turn to
* be the dispatcher to finish waiting for
* our reply
- * 3. I/O was expected to block
*/
if (thiscall->mode == VIR_NET_CLIENT_MODE_COMPLETE) {
rv = 2;
@@ -1591,17 +1558,6 @@ static int virNetClientIO(virNetClientPtr client,
goto cleanup;
}
- /* If we're non-blocking, we were either queued (and detached) or the
- * call was not sent because of an error.
- */
- if (thiscall->nonBlock) {
- if (!thiscall->haveThread)
- rv = 1; /* In progress */
- else
- rv = 0; /* none at all */
- goto cleanup;
- }
-
/* Grr, someone passed the buck onto us ... */
} else {
client->haveTheBuck = true;
--
1.7.10.2
4:48 a.m.
New subject: [libvirt] [PATCH 04/12] client rpc: Just queue non-blocking call if another thread has the buck
On Wed, Jun 13, 2012 at 01:29:22AM +0200, Jiri Denemark wrote:
As non-blocking calls are no longer dropped, we don't really need
to
care that much about their faith and wait for the thread with the buck
ITYM to s/faith/fate/ ?
to process them. If another thread has the buck, we can just push a
non-blocking call to the queue and be done with it.
---
src/rpc/virnetclient.c | 74 ++++++++++--------------------------------------
1 file changed, 15 insertions(+), 59 deletions(-)
ACK
Daniel
--
|: http://berrange.com -o- http://www.flickr.com/photos/dberrange/ :|
|: http://libvirt.org -o- http://virt-manager.org :|
|: http://autobuild.org -o- http://search.cpan.org/~danberr/ :|
|: http://entangle-photo.org -o- http://live.gnome.org/gtk-vnc :|
6:29 p.m.
New subject: [libvirt] [PATCH 05/12] client rpc: Drop unused return value of virNetClientSendNonBlock
As we never drop non-blocking calls, the return value that used to
indicate a call was dropped is no longer needed.
---
src/rpc/virnetclient.c | 19 ++++++++++---------
1 file changed, 10 insertions(+), 9 deletions(-)
diff --git a/src/rpc/virnetclient.c b/src/rpc/virnetclient.c
index b3d8d14..e9898be 100644
--- a/src/rpc/virnetclient.c
+++ b/src/rpc/virnetclient.c
@@ -1258,8 +1258,8 @@ virNetClientIOEventLoopPassTheBuck(virNetClientPtr client,
* 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
+ * Returns 1 if the call was queued and will be completed later (only
+ * for nonBlock==true), 0 if the call was completed and -1 on error.
*/
static int virNetClientIOEventLoop(virNetClientPtr client,
virNetClientCallPtr thiscall)
@@ -1383,7 +1383,7 @@ static int virNetClientIOEventLoop(virNetClientPtr client,
if (thiscall->mode == VIR_NET_CLIENT_MODE_COMPLETE) {
virNetClientCallRemove(&client->waitDispatch, thiscall);
virNetClientIOEventLoopPassTheBuck(client, thiscall);
- return 2;
+ return 0;
}
/* We're not done, but we're non-blocking; keep the call queued */
@@ -1490,8 +1490,8 @@ static void virNetClientIOUpdateCallback(virNetClientPtr client,
*
* NB(7) Don't Panic!
*
- * 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
+ * Returns 1 if the call was queued and will be completed later (only
+ * for nonBlock==true), 0 if the call was completed and -1 on error.
*/
static int virNetClientIO(virNetClientPtr client,
virNetClientCallPtr thiscall)
@@ -1549,7 +1549,7 @@ static int virNetClientIO(virNetClientPtr client,
* our reply
*/
if (thiscall->mode == VIR_NET_CLIENT_MODE_COMPLETE) {
- rv = 2;
+ rv = 0;
/*
* We avoided catching the buck and our reply is ready !
* We've already had 'thiscall' removed from the list
@@ -1643,8 +1643,8 @@ done:
/*
- * Returns 2 if fully sent, 1 if queued (only for nonBlock==true),
- * 0 if nothing sent (only for nonBlock==true) and -1 on error
+ * Returns 1 if the call was queued and will be completed later (only
+ * for nonBlock==true), 0 if the call was completed and -1 on error.
*/
static int virNetClientSendInternal(virNetClientPtr client,
virNetMessagePtr msg,
@@ -1769,7 +1769,8 @@ int virNetClientSendNoReply(virNetClientPtr client,
* The caller is responsible for free'ing @msg, *except* if
* this method returns 1.
*
- * Returns 2 on full send, 1 on partial send, 0 on no send, -1 on error
+ * Returns 1 if the message was queued and will be completed later (only
+ * for nonBlock==true), 0 if the message was completed and -1 on error.
*/
int virNetClientSendNonBlock(virNetClientPtr client,
virNetMessagePtr msg)
--
1.7.10.2
4:48 a.m.
New subject: [libvirt] [PATCH 05/12] client rpc: Drop unused return value of virNetClientSendNonBlock
On Wed, Jun 13, 2012 at 01:29:23AM +0200, Jiri Denemark wrote:
As we never drop non-blocking calls, the return value that used to
indicate a call was dropped is no longer needed.
---
src/rpc/virnetclient.c | 19 ++++++++++---------
1 file changed, 10 insertions(+), 9 deletions(-)
ACK
Daniel
--
|: http://berrange.com -o- http://www.flickr.com/photos/dberrange/ :|
|: http://libvirt.org -o- http://virt-manager.org :|
|: http://autobuild.org -o- http://search.cpan.org/~danberr/ :|
|: http://entangle-photo.org -o- http://live.gnome.org/gtk-vnc :|
6:29 p.m.
New subject: [libvirt] [PATCH 06/12] rpc: Refactor keepalive timer code
The code that needs to be run every keepalive interval of inactivity was
only called from a timer and thus from the main event loop. We will need
to call the code directly from another place.
---
src/rpc/virkeepalive.c | 57 +++++++++++++++++++++++++++++-------------------
1 file changed, 35 insertions(+), 22 deletions(-)
diff --git a/src/rpc/virkeepalive.c b/src/rpc/virkeepalive.c
index a5c2b1a..b2e260e 100644
--- a/src/rpc/virkeepalive.c
+++ b/src/rpc/virkeepalive.c
@@ -152,51 +152,64 @@ virKeepAliveScheduleResponse(virKeepAlivePtr ka)
}
-static void
-virKeepAliveTimer(int timer ATTRIBUTE_UNUSED, void *opaque)
+static bool
+virKeepAliveTimerInternal(virKeepAlivePtr ka,
+ virNetMessagePtr *msg)
{
- virKeepAlivePtr ka = opaque;
time_t now = time(NULL);
- virKeepAliveLock(ka);
+ if (now - ka->lastPacketReceived < ka->interval - 1) {
+ int timeout = ka->interval - (now - ka->lastPacketReceived);
+ virEventUpdateTimeout(ka->timer, timeout * 1000);
+ return false;
+ }
PROBE(RPC_KEEPALIVE_TIMEOUT,
"ka=%p client=%p countToDeath=%d idle=%d",
ka, ka->client, ka->countToDeath,
(int) (now - ka->lastPacketReceived));
- if (now - ka->lastPacketReceived < ka->interval - 1) {
- int timeout = ka->interval - (now - ka->lastPacketReceived);
- virEventUpdateTimeout(ka->timer, timeout * 1000);
- goto cleanup;
- }
if (ka->countToDeath == 0) {
- virKeepAliveDeadFunc deadCB = ka->deadCB;
- void *client = ka->client;
-
VIR_WARN("No response from client %p after %d keepalive messages in"
" %d seconds",
ka->client,
ka->count,
(int) (now - ka->lastPacketReceived));
+ return true;
+ } else {
+ ka->countToDeath--;
+ *msg = virKeepAliveMessage(KEEPALIVE_PROC_PING);
+ virEventUpdateTimeout(ka->timer, ka->interval * 1000);
+ return false;
+ }
+}
+
+
+static void
+virKeepAliveTimer(int timer ATTRIBUTE_UNUSED, void *opaque)
+{
+ virKeepAlivePtr ka = opaque;
+ virNetMessagePtr msg = NULL;
+ bool dead;
+
+ virKeepAliveLock(ka);
+
+ dead = virKeepAliveTimerInternal(ka, &msg);
+
+ if (dead) {
+ virKeepAliveDeadFunc deadCB = ka->deadCB;
+ void *client = ka->client;
+
ka->refs++;
virKeepAliveUnlock(ka);
deadCB(client);
virKeepAliveLock(ka);
ka->refs--;
- } else {
- virNetMessagePtr msg;
-
- ka->countToDeath--;
- if (!(msg = virKeepAliveMessage(KEEPALIVE_PROC_PING)))
- VIR_WARN("Failed to generate keepalive request");
- else
- virKeepAliveSend(ka, msg);
- virEventUpdateTimeout(ka->timer, ka->interval * 1000);
+ } else if (msg) {
+ virKeepAliveSend(ka, msg);
}
-cleanup:
virKeepAliveUnlock(ka);
}
--
1.7.10.2
4:49 a.m.
New subject: [libvirt] [PATCH 06/12] rpc: Refactor keepalive timer code
On Wed, Jun 13, 2012 at 01:29:24AM +0200, Jiri Denemark wrote:
The code that needs to be run every keepalive interval of inactivity
was
only called from a timer and thus from the main event loop. We will need
to call the code directly from another place.
---
src/rpc/virkeepalive.c | 57 +++++++++++++++++++++++++++++-------------------
1 file changed, 35 insertions(+), 22 deletions(-)
ACK
Daniel
--
|: http://berrange.com -o- http://www.flickr.com/photos/dberrange/ :|
|: http://libvirt.org -o- http://virt-manager.org :|
|: http://autobuild.org -o- http://search.cpan.org/~danberr/ :|
|: http://entangle-photo.org -o- http://live.gnome.org/gtk-vnc :|
6:29 p.m.
New subject: [libvirt] [PATCH 07/12] rpc: Add APIs for direct triggering of keepalive timer
Add virKeepAliveTimeout and virKeepAliveTrigger APIs that can be used to
set poll timeouts and trigger keepalive timer. virKeepAliveTrigger
checks if it is called to early and does nothing in that case.
---
src/rpc/virkeepalive.c | 61 ++++++++++++++++++++++++++++++++++++++++++++----
src/rpc/virkeepalive.h | 3 +++
2 files changed, 60 insertions(+), 4 deletions(-)
diff --git a/src/rpc/virkeepalive.c b/src/rpc/virkeepalive.c
index b2e260e..035ac74 100644
--- a/src/rpc/virkeepalive.c
+++ b/src/rpc/virkeepalive.c
@@ -45,6 +45,7 @@ struct _virKeepAlive {
unsigned int count;
unsigned int countToDeath;
time_t lastPacketReceived;
+ time_t intervalStart;
int timer;
virNetMessagePtr response;
@@ -158,8 +159,11 @@ virKeepAliveTimerInternal(virKeepAlivePtr ka,
{
time_t now = time(NULL);
- if (now - ka->lastPacketReceived < ka->interval - 1) {
- int timeout = ka->interval - (now - ka->lastPacketReceived);
+ if (ka->interval <= 0 || ka->intervalStart == 0)
+ return false;
+
+ if (now - ka->intervalStart < ka->interval) {
+ int timeout = ka->interval - (now - ka->intervalStart);
virEventUpdateTimeout(ka->timer, timeout * 1000);
return false;
}
@@ -179,6 +183,7 @@ virKeepAliveTimerInternal(virKeepAlivePtr ka,
return true;
} else {
ka->countToDeath--;
+ ka->intervalStart = now;
*msg = virKeepAliveMessage(KEEPALIVE_PROC_PING);
virEventUpdateTimeout(ka->timer, ka->interval * 1000);
return false;
@@ -335,6 +340,7 @@ virKeepAliveStart(virKeepAlivePtr ka,
int ret = -1;
time_t delay;
int timeout;
+ time_t now;
virKeepAliveLock(ka);
@@ -365,11 +371,13 @@ virKeepAliveStart(virKeepAlivePtr ka,
"ka=%p client=%p interval=%d count=%u",
ka, ka->client, interval, count);
- delay = time(NULL) - ka->lastPacketReceived;
+ now = time(NULL);
+ delay = now - ka->lastPacketReceived;
if (delay > ka->interval)
timeout = 0;
else
timeout = ka->interval - delay;
+ ka->intervalStart = now - (ka->interval - timeout);
ka->timer = virEventAddTimeout(timeout * 1000, virKeepAliveTimer,
ka, virKeepAliveTimerFree);
if (ka->timer < 0)
@@ -427,6 +435,51 @@ virKeepAliveStopSending(virKeepAlivePtr ka)
}
+int
+virKeepAliveTimeout(virKeepAlivePtr ka)
+{
+ int timeout;
+
+ if (!ka)
+ return -1;
+
+ virKeepAliveLock(ka);
+
+ if (ka->interval <= 0 || ka->intervalStart == 0) {
+ timeout = -1;
+ } else {
+ timeout = ka->interval - (time(NULL) - ka->intervalStart);
+ if (timeout < 0)
+ timeout = 0;
+ }
+
+ virKeepAliveUnlock(ka);
+
+ if (timeout < 0)
+ return -1;
+ else
+ return timeout * 1000;
+}
+
+
+bool
+virKeepAliveTrigger(virKeepAlivePtr ka,
+ virNetMessagePtr *msg)
+{
+ bool dead;
+
+ *msg = NULL;
+ if (!ka)
+ return false;
+
+ virKeepAliveLock(ka);
+ dead = virKeepAliveTimerInternal(ka, msg);
+ virKeepAliveUnlock(ka);
+
+ return dead;
+}
+
+
bool
virKeepAliveCheckMessage(virKeepAlivePtr ka,
virNetMessagePtr msg)
@@ -442,7 +495,7 @@ virKeepAliveCheckMessage(virKeepAlivePtr ka,
virKeepAliveLock(ka);
ka->countToDeath = ka->count;
- ka->lastPacketReceived = time(NULL);
+ ka->lastPacketReceived = ka->intervalStart = time(NULL);
if (msg->header.prog == KEEPALIVE_PROGRAM &&
msg->header.vers == KEEPALIVE_PROTOCOL_VERSION &&
diff --git a/src/rpc/virkeepalive.h b/src/rpc/virkeepalive.h
index af9e722..09264a5 100644
--- a/src/rpc/virkeepalive.h
+++ b/src/rpc/virkeepalive.h
@@ -51,6 +51,9 @@ int virKeepAliveStart(virKeepAlivePtr ka,
void virKeepAliveStop(virKeepAlivePtr ka);
void virKeepAliveStopSending(virKeepAlivePtr ka);
+int virKeepAliveTimeout(virKeepAlivePtr ka);
+bool virKeepAliveTrigger(virKeepAlivePtr ka,
+ virNetMessagePtr *msg);
bool virKeepAliveCheckMessage(virKeepAlivePtr ka,
virNetMessagePtr msg);
--
1.7.10.2
4:49 a.m.
New subject: [libvirt] [PATCH 07/12] rpc: Add APIs for direct triggering of keepalive timer
On Wed, Jun 13, 2012 at 01:29:25AM +0200, Jiri Denemark wrote:
Add virKeepAliveTimeout and virKeepAliveTrigger APIs that can be used
to
set poll timeouts and trigger keepalive timer. virKeepAliveTrigger
checks if it is called to early and does nothing in that case.
---
src/rpc/virkeepalive.c | 61 ++++++++++++++++++++++++++++++++++++++++++++----
src/rpc/virkeepalive.h | 3 +++
2 files changed, 60 insertions(+), 4 deletions(-)
ACK
Daniel
--
|: http://berrange.com -o- http://www.flickr.com/photos/dberrange/ :|
|: http://libvirt.org -o- http://virt-manager.org :|
|: http://autobuild.org -o- http://search.cpan.org/~danberr/ :|
|: http://entangle-photo.org -o- http://live.gnome.org/gtk-vnc :|
6:29 p.m.
New subject: [libvirt] [PATCH 08/12] client rpc: Separate call creation from running IO loop
This makes it possible to create and queue new calls while we are
running IO loop.
---
src/rpc/virnetclient.c | 85 ++++++++++++++++++++++++++++++------------------
1 file changed, 54 insertions(+), 31 deletions(-)
diff --git a/src/rpc/virnetclient.c b/src/rpc/virnetclient.c
index e9898be..b956f6e 100644
--- a/src/rpc/virnetclient.c
+++ b/src/rpc/virnetclient.c
@@ -1642,53 +1642,38 @@ done:
}
-/*
- * Returns 1 if the call was queued and will be completed later (only
- * for nonBlock==true), 0 if the call was completed and -1 on error.
- */
-static int virNetClientSendInternal(virNetClientPtr client,
- virNetMessagePtr msg,
- bool expectReply,
- bool nonBlock)
+static virNetClientCallPtr
+virNetClientCallNew(virNetMessagePtr msg,
+ bool expectReply,
+ bool nonBlock)
{
- virNetClientCallPtr call;
- int ret = -1;
-
- PROBE(RPC_CLIENT_MSG_TX_QUEUE,
- "client=%p len=%zu prog=%u vers=%u proc=%u type=%u status=%u
serial=%u",
- client, msg->bufferLength,
- msg->header.prog, msg->header.vers, msg->header.proc,
- msg->header.type, msg->header.status, msg->header.serial);
+ virNetClientCallPtr call = NULL;
if (expectReply &&
(msg->bufferLength != 0) &&
(msg->header.status == VIR_NET_CONTINUE)) {
virNetError(VIR_ERR_INTERNAL_ERROR, "%s",
- _("Attempt to send an asynchronous message with a synchronous
reply"));
- return -1;
+ _("Attempt to send an asynchronous message with"
+ " a synchronous reply"));
+ goto error;
}
if (expectReply && nonBlock) {
virNetError(VIR_ERR_INTERNAL_ERROR, "%s",
- _("Attempt to send a non-blocking message with a synchronous
reply"));
- return -1;
+ _("Attempt to send a non-blocking message with"
+ " a synchronous reply"));
+ goto error;
}
if (VIR_ALLOC(call) < 0) {
virReportOOMError();
- return -1;
- }
-
- if (!client->sock || client->wantClose) {
- virNetError(VIR_ERR_INTERNAL_ERROR, "%s",
- _("client socket is closed"));
- goto cleanup;
+ goto error;
}
if (virCondInit(&call->cond) < 0) {
virNetError(VIR_ERR_INTERNAL_ERROR, "%s",
_("cannot initialize condition variable"));
- goto cleanup;
+ goto error;
}
msg->donefds = 0;
@@ -1699,8 +1684,48 @@ static int virNetClientSendInternal(virNetClientPtr client,
call->msg = msg;
call->expectReply = expectReply;
call->nonBlock = nonBlock;
- call->haveThread = true;
+ VIR_DEBUG("New call %p: msg=%p, expectReply=%d, nonBlock=%d",
+ call, msg, expectReply, nonBlock);
+
+ return call;
+
+error:
+ VIR_FREE(call);
+ return NULL;
+}
+
+
+/*
+ * Returns 1 if the call was queued and will be completed later (only
+ * for nonBlock==true), 0 if the call was completed and -1 on error.
+ */
+static int virNetClientSendInternal(virNetClientPtr client,
+ virNetMessagePtr msg,
+ bool expectReply,
+ bool nonBlock)
+{
+ virNetClientCallPtr call;
+ int ret = -1;
+
+ PROBE(RPC_CLIENT_MSG_TX_QUEUE,
+ "client=%p len=%zu prog=%u vers=%u proc=%u type=%u status=%u
serial=%u",
+ client, msg->bufferLength,
+ msg->header.prog, msg->header.vers, msg->header.proc,
+ msg->header.type, msg->header.status, msg->header.serial);
+
+ if (!client->sock || client->wantClose) {
+ virNetError(VIR_ERR_INTERNAL_ERROR, "%s",
+ _("client socket is closed"));
+ return -1;
+ }
+
+ if (!(call = virNetClientCallNew(msg, expectReply, nonBlock))) {
+ virReportOOMError();
+ return -1;
+ }
+
+ call->haveThread = true;
ret = virNetClientIO(client, call);
/* If queued, the call will be finished and freed later by another thread;
@@ -1709,8 +1734,6 @@ static int virNetClientSendInternal(virNetClientPtr client,
return 1;
ignore_value(virCondDestroy(&call->cond));
-
-cleanup:
VIR_FREE(call);
return ret;
}
--
1.7.10.2
4:50 a.m.
New subject: [libvirt] [PATCH 08/12] client rpc: Separate call creation from running IO loop
On Wed, Jun 13, 2012 at 01:29:26AM +0200, Jiri Denemark wrote:
This makes it possible to create and queue new calls while we are
running IO loop.
---
src/rpc/virnetclient.c | 85 ++++++++++++++++++++++++++++++------------------
1 file changed, 54 insertions(+), 31 deletions(-)
ACK
Daniel
--
|: http://berrange.com -o- http://www.flickr.com/photos/dberrange/ :|
|: http://libvirt.org -o- http://virt-manager.org :|
|: http://autobuild.org -o- http://search.cpan.org/~danberr/ :|
|: http://entangle-photo.org -o- http://live.gnome.org/gtk-vnc :|
6:29 p.m.
New subject: [libvirt] [PATCH 09/12] rpc: Do not use timer for sending keepalive responses
When a libvirt API is called from the main event loop (which seems to be
common in event-based glib apps), the client IO loop would properly
handle keepalive requests sent by a server but will not actually send
them because the main event loop is blocked with the API. This patch
gets rid of response timer and the thread which is processing keepalive
requests is also responsible for queueing responses for delivery.
---
src/rpc/virkeepalive.c | 155 +++++++++++-------------------------------
src/rpc/virkeepalive.h | 3 +-
src/rpc/virnetclient.c | 35 +++++++++-
src/rpc/virnetserverclient.c | 88 ++++++++++++------------
4 files changed, 120 insertions(+), 161 deletions(-)
diff --git a/src/rpc/virkeepalive.c b/src/rpc/virkeepalive.c
index 035ac74..7984ada 100644
--- a/src/rpc/virkeepalive.c
+++ b/src/rpc/virkeepalive.c
@@ -48,9 +48,6 @@ struct _virKeepAlive {
time_t intervalStart;
int timer;
- virNetMessagePtr response;
- int responseTimer;
-
virKeepAliveSendFunc sendCB;
virKeepAliveDeadFunc deadCB;
virKeepAliveFreeFunc freeCB;
@@ -72,12 +69,25 @@ virKeepAliveUnlock(virKeepAlivePtr ka)
static virNetMessagePtr
-virKeepAliveMessage(int proc)
+virKeepAliveMessage(virKeepAlivePtr ka, int proc)
{
virNetMessagePtr msg;
+ const char *procstr = NULL;
- if (!(msg = virNetMessageNew(false)))
+ switch (proc) {
+ case KEEPALIVE_PROC_PING:
+ procstr = "request";
+ break;
+ case KEEPALIVE_PROC_PONG:
+ procstr = "response";
+ break;
+ default:
+ VIR_WARN("Refusing to send unknown keepalive message: %d", proc);
return NULL;
+ }
+
+ if (!(msg = virNetMessageNew(false)))
+ goto error;
msg->header.prog = KEEPALIVE_PROGRAM;
msg->header.vers = KEEPALIVE_PROTOCOL_VERSION;
@@ -87,69 +97,20 @@ virKeepAliveMessage(int proc)
if (virNetMessageEncodeHeader(msg) < 0 ||
virNetMessageEncodePayloadEmpty(msg) < 0) {
virNetMessageFree(msg);
- return NULL;
- }
-
- return msg;
-}
-
-
-static void
-virKeepAliveSend(virKeepAlivePtr ka, virNetMessagePtr msg)
-{
- const char *proc = NULL;
- void *client = ka->client;
- virKeepAliveSendFunc sendCB = ka->sendCB;
-
- switch (msg->header.proc) {
- case KEEPALIVE_PROC_PING:
- proc = "request";
- break;
- case KEEPALIVE_PROC_PONG:
- proc = "response";
- break;
- }
-
- if (!proc) {
- VIR_WARN("Refusing to send unknown keepalive message: %d",
- msg->header.proc);
- virNetMessageFree(msg);
- return;
+ goto error;
}
- VIR_DEBUG("Sending keepalive %s to client %p", proc, ka->client);
+ VIR_DEBUG("Sending keepalive %s to client %p", procstr, ka->client);
PROBE(RPC_KEEPALIVE_SEND,
"ka=%p client=%p prog=%d vers=%d proc=%d",
ka, ka->client, msg->header.prog, msg->header.vers,
msg->header.proc);
- ka->refs++;
- virKeepAliveUnlock(ka);
-
- if (sendCB(client, msg) < 0) {
- VIR_WARN("Failed to send keepalive %s to client %p", proc, client);
- virNetMessageFree(msg);
- }
-
- virKeepAliveLock(ka);
- ka->refs--;
-}
-
-
-static void
-virKeepAliveScheduleResponse(virKeepAlivePtr ka)
-{
- if (ka->responseTimer == -1)
- return;
-
- VIR_DEBUG("Scheduling keepalive response to client %p", ka->client);
-
- if (!ka->response &&
- !(ka->response = virKeepAliveMessage(KEEPALIVE_PROC_PONG))) {
- VIR_WARN("Failed to generate keepalive response");
- return;
- }
+ return msg;
- virEventUpdateTimeout(ka->responseTimer, 0);
+error:
+ VIR_WARN("Failed to generate keepalive %s", procstr);
+ VIR_FREE(msg);
+ return NULL;
}
@@ -184,7 +145,7 @@ virKeepAliveTimerInternal(virKeepAlivePtr ka,
} else {
ka->countToDeath--;
ka->intervalStart = now;
- *msg = virKeepAliveMessage(KEEPALIVE_PROC_PING);
+ *msg = virKeepAliveMessage(ka, KEEPALIVE_PROC_PING);
virEventUpdateTimeout(ka->timer, ka->interval * 1000);
return false;
}
@@ -197,47 +158,30 @@ virKeepAliveTimer(int timer ATTRIBUTE_UNUSED, void *opaque)
virKeepAlivePtr ka = opaque;
virNetMessagePtr msg = NULL;
bool dead;
+ void *client;
virKeepAliveLock(ka);
+ client = ka->client;
dead = virKeepAliveTimerInternal(ka, &msg);
- if (dead) {
- virKeepAliveDeadFunc deadCB = ka->deadCB;
- void *client = ka->client;
-
- ka->refs++;
- virKeepAliveUnlock(ka);
- deadCB(client);
- virKeepAliveLock(ka);
- ka->refs--;
- } else if (msg) {
- virKeepAliveSend(ka, msg);
- }
+ if (!dead && !msg)
+ goto cleanup;
+ ka->refs++;
virKeepAliveUnlock(ka);
-}
-
-
-static void
-virKeepAliveResponseTimer(int timer ATTRIBUTE_UNUSED, void *opaque)
-{
- virKeepAlivePtr ka = opaque;
- virNetMessagePtr msg;
-
- virKeepAliveLock(ka);
- VIR_DEBUG("ka=%p, client=%p, response=%p",
- ka, ka->client, ka->response);
-
- if (ka->response) {
- msg = ka->response;
- ka->response = NULL;
- virKeepAliveSend(ka, msg);
+ if (dead) {
+ ka->deadCB(client);
+ } else if (ka->sendCB(client, msg) < 0) {
+ VIR_WARN("Failed to send keepalive request to client %p", client);
+ virNetMessageFree(msg);
}
- virEventUpdateTimeout(ka->responseTimer, ka->response ? 0 : -1);
+ virKeepAliveLock(ka);
+ ka->refs--;
+cleanup:
virKeepAliveUnlock(ka);
}
@@ -281,15 +225,6 @@ virKeepAliveNew(int interval,
ka->deadCB = deadCB;
ka->freeCB = freeCB;
- ka->responseTimer = virEventAddTimeout(-1, virKeepAliveResponseTimer,
- ka, virKeepAliveTimerFree);
- if (ka->responseTimer < 0) {
- virKeepAliveFree(ka);
- return NULL;
- }
- /* the timer now has a reference to ka */
- ka->refs++;
-
PROBE(RPC_KEEPALIVE_NEW,
"ka=%p client=%p refs=%d",
ka, ka->client, ka->refs);
@@ -394,7 +329,7 @@ cleanup:
static void
-virKeepAliveStopInternal(virKeepAlivePtr ka, bool all)
+virKeepAliveStopInternal(virKeepAlivePtr ka, bool all ATTRIBUTE_UNUSED)
{
virKeepAliveLock(ka);
@@ -407,16 +342,6 @@ virKeepAliveStopInternal(virKeepAlivePtr ka, bool all)
ka->timer = -1;
}
- if (all) {
- if (ka->responseTimer > 0) {
- virEventRemoveTimeout(ka->responseTimer);
- ka->responseTimer = -1;
- }
-
- virNetMessageFree(ka->response);
- ka->response = NULL;
- }
-
virKeepAliveUnlock(ka);
}
@@ -482,13 +407,15 @@ virKeepAliveTrigger(virKeepAlivePtr ka,
bool
virKeepAliveCheckMessage(virKeepAlivePtr ka,
- virNetMessagePtr msg)
+ virNetMessagePtr msg,
+ virNetMessagePtr *response)
{
bool ret = false;
VIR_DEBUG("ka=%p, client=%p, msg=%p",
ka, ka ? ka->client : "(null)", msg);
+ *response = NULL;
if (!ka)
return false;
@@ -508,7 +435,7 @@ virKeepAliveCheckMessage(virKeepAlivePtr ka,
switch (msg->header.proc) {
case KEEPALIVE_PROC_PING:
VIR_DEBUG("Got keepalive request from client %p", ka->client);
- virKeepAliveScheduleResponse(ka);
+ *response = virKeepAliveMessage(ka, KEEPALIVE_PROC_PONG);
break;
case KEEPALIVE_PROC_PONG:
diff --git a/src/rpc/virkeepalive.h b/src/rpc/virkeepalive.h
index 09264a5..62227d0 100644
--- a/src/rpc/virkeepalive.h
+++ b/src/rpc/virkeepalive.h
@@ -55,6 +55,7 @@ int virKeepAliveTimeout(virKeepAlivePtr ka);
bool virKeepAliveTrigger(virKeepAlivePtr ka,
virNetMessagePtr *msg);
bool virKeepAliveCheckMessage(virKeepAlivePtr ka,
- virNetMessagePtr msg);
+ virNetMessagePtr msg,
+ virNetMessagePtr *response);
#endif /* __VIR_KEEPALIVE_H__ */
diff --git a/src/rpc/virnetclient.c b/src/rpc/virnetclient.c
index b956f6e..48c6a5d 100644
--- a/src/rpc/virnetclient.c
+++ b/src/rpc/virnetclient.c
@@ -109,6 +109,8 @@ struct _virNetClient {
static void virNetClientIOEventLoopPassTheBuck(virNetClientPtr client,
virNetClientCallPtr thiscall);
+static int virNetClientQueueNonBlocking(virNetClientPtr client,
+ virNetMessagePtr msg);
static void virNetClientLock(virNetClientPtr client)
@@ -937,14 +939,22 @@ static int virNetClientCallDispatchStream(virNetClientPtr client)
static int
virNetClientCallDispatch(virNetClientPtr client)
{
+ virNetMessagePtr response = NULL;
+
PROBE(RPC_CLIENT_MSG_RX,
"client=%p len=%zu prog=%u vers=%u proc=%u type=%u status=%u
serial=%u",
client, client->msg.bufferLength,
client->msg.header.prog, client->msg.header.vers,
client->msg.header.proc,
client->msg.header.type, client->msg.header.status,
client->msg.header.serial);
- if (virKeepAliveCheckMessage(client->keepalive, &client->msg))
+ if (virKeepAliveCheckMessage(client->keepalive, &client->msg,
&response)) {
+ if (response &&
+ virNetClientQueueNonBlocking(client, response) < 0) {
+ VIR_WARN("Could not queue keepalive response");
+ virNetMessageFree(response);
+ }
return 0;
+ }
switch (client->msg.header.type) {
case VIR_NET_REPLY: /* Normal RPC replies */
@@ -1637,6 +1647,8 @@ void virNetClientIncomingEvent(virNetSocketPtr sock,
virNetClientCallRemovePredicate(&client->waitDispatch,
virNetClientIOEventLoopRemoveDone,
NULL);
+ virNetClientIOUpdateCallback(client, true);
+
done:
virNetClientUnlock(client);
}
@@ -1696,6 +1708,27 @@ error:
}
+static int
+virNetClientQueueNonBlocking(virNetClientPtr client,
+ virNetMessagePtr msg)
+{
+ virNetClientCallPtr call;
+
+ PROBE(RPC_CLIENT_MSG_TX_QUEUE,
+ "client=%p len=%zu prog=%u vers=%u proc=%u"
+ " type=%u status=%u serial=%u",
+ client, msg->bufferLength,
+ msg->header.prog, msg->header.vers, msg->header.proc,
+ msg->header.type, msg->header.status, msg->header.serial);
+
+ if (!(call = virNetClientCallNew(msg, false, true)))
+ return -1;
+
+ virNetClientCallQueue(&client->waitDispatch, call);
+ return 0;
+}
+
+
/*
* Returns 1 if the call was queued and will be completed later (only
* for nonBlock==true), 0 if the call was completed and -1 on error.
diff --git a/src/rpc/virnetserverclient.c b/src/rpc/virnetserverclient.c
index 6ae4e25..f3eb61a 100644
--- a/src/rpc/virnetserverclient.c
+++ b/src/rpc/virnetserverclient.c
@@ -103,13 +103,14 @@ struct _virNetServerClient
virNetServerClientCloseFunc privateDataCloseFunc;
virKeepAlivePtr keepalive;
- int keepaliveFilter;
};
static void virNetServerClientDispatchEvent(virNetSocketPtr sock, int events, void
*opaque);
static void virNetServerClientUpdateEvent(virNetServerClientPtr client);
static void virNetServerClientDispatchRead(virNetServerClientPtr client);
+static int virNetServerClientSendMessageLocked(virNetServerClientPtr client,
+ virNetMessagePtr msg);
static void virNetServerClientLock(virNetServerClientPtr client)
{
@@ -364,7 +365,6 @@ virNetServerClientPtr virNetServerClientNew(virNetSocketPtr sock,
client->readonly = readonly;
client->tlsCtxt = tls;
client->nrequests_max = nrequests_max;
- client->keepaliveFilter = -1;
client->sockTimer = virEventAddTimeout(-1, virNetServerClientSockTimerFunc,
client, NULL);
@@ -644,9 +644,6 @@ void virNetServerClientClose(virNetServerClientPtr client)
return;
}
- if (client->keepaliveFilter >= 0)
- virNetServerClientRemoveFilterLocked(client, client->keepaliveFilter);
-
if (client->keepalive) {
virKeepAliveStop(client->keepalive);
ka = client->keepalive;
@@ -844,6 +841,7 @@ readmore:
} else {
/* Grab the completed message */
virNetMessagePtr msg = client->rx;
+ virNetMessagePtr response = NULL;
virNetServerClientFilterPtr filter;
size_t i;
@@ -894,23 +892,35 @@ readmore:
msg->header.prog, msg->header.vers, msg->header.proc,
msg->header.type, msg->header.status, msg->header.serial);
+ if (virKeepAliveCheckMessage(client->keepalive, msg, &response)) {
+ virNetMessageFree(msg);
+ client->nrequests--;
+ msg = NULL;
+
+ if (response &&
+ virNetServerClientSendMessageLocked(client, response) < 0)
+ virNetMessageFree(response);
+ }
+
/* Maybe send off for queue against a filter */
- filter = client->filters;
- while (filter) {
- int ret = filter->func(client, msg, filter->opaque);
- if (ret < 0) {
- virNetMessageFree(msg);
- msg = NULL;
- if (ret < 0)
- client->wantClose = true;
- break;
- }
- if (ret > 0) {
- msg = NULL;
- break;
- }
+ if (msg) {
+ filter = client->filters;
+ while (filter) {
+ int ret = filter->func(client, msg, filter->opaque);
+ if (ret < 0) {
+ virNetMessageFree(msg);
+ msg = NULL;
+ if (ret < 0)
+ client->wantClose = true;
+ break;
+ }
+ if (ret > 0) {
+ msg = NULL;
+ break;
+ }
- filter = filter->next;
+ filter = filter->next;
+ }
}
/* Send off to for normal dispatch to workers */
@@ -1117,16 +1127,15 @@ virNetServerClientDispatchEvent(virNetSocketPtr sock, int events,
void *opaque)
}
-int virNetServerClientSendMessage(virNetServerClientPtr client,
- virNetMessagePtr msg)
+static int
+virNetServerClientSendMessageLocked(virNetServerClientPtr client,
+ virNetMessagePtr msg)
{
int ret = -1;
VIR_DEBUG("msg=%p proc=%d len=%zu offset=%zu",
msg, msg->header.proc,
msg->bufferLength, msg->bufferOffset);
- virNetServerClientLock(client);
-
msg->donefds = 0;
if (client->sock && !client->wantClose) {
PROBE(RPC_SERVER_CLIENT_MSG_TX_QUEUE,
@@ -1140,6 +1149,16 @@ int virNetServerClientSendMessage(virNetServerClientPtr client,
ret = 0;
}
+ return ret;
+}
+
+int virNetServerClientSendMessage(virNetServerClientPtr client,
+ virNetMessagePtr msg)
+{
+ int ret;
+
+ virNetServerClientLock(client);
+ ret = virNetServerClientSendMessageLocked(client, msg);
virNetServerClientUnlock(client);
return ret;
@@ -1176,20 +1195,6 @@ virNetServerClientFreeCB(void *opaque)
virNetServerClientFree(opaque);
}
-static int
-virNetServerClientKeepAliveFilter(virNetServerClientPtr client,
- virNetMessagePtr msg,
- void *opaque ATTRIBUTE_UNUSED)
-{
- if (virKeepAliveCheckMessage(client->keepalive, msg)) {
- virNetMessageFree(msg);
- client->nrequests--;
- return 1;
- }
-
- return 0;
-}
-
int
virNetServerClientInitKeepAlive(virNetServerClientPtr client,
int interval,
@@ -1208,13 +1213,6 @@ virNetServerClientInitKeepAlive(virNetServerClientPtr client,
/* keepalive object has a reference to client */
client->refs++;
- client->keepaliveFilter =
- virNetServerClientAddFilterLocked(client,
- virNetServerClientKeepAliveFilter,
- NULL);
- if (client->keepaliveFilter < 0)
- goto cleanup;
-
client->keepalive = ka;
ka = NULL;
--
1.7.10.2
4:50 a.m.
New subject: [libvirt] [PATCH 09/12] rpc: Do not use timer for sending keepalive responses
On Wed, Jun 13, 2012 at 01:29:27AM +0200, Jiri Denemark wrote:
When a libvirt API is called from the main event loop (which seems to
be
common in event-based glib apps), the client IO loop would properly
handle keepalive requests sent by a server but will not actually send
them because the main event loop is blocked with the API. This patch
gets rid of response timer and the thread which is processing keepalive
requests is also responsible for queueing responses for delivery.
---
src/rpc/virkeepalive.c | 155 +++++++++++-------------------------------
src/rpc/virkeepalive.h | 3 +-
src/rpc/virnetclient.c | 35 +++++++++-
src/rpc/virnetserverclient.c | 88 ++++++++++++------------
4 files changed, 120 insertions(+), 161 deletions(-)
ACK. Again quite hard to follow the patch, so I reviewed the result of
applying it instead.
Daniel
--
|: http://berrange.com -o- http://www.flickr.com/photos/dberrange/ :|
|: http://libvirt.org -o- http://virt-manager.org :|
|: http://autobuild.org -o- http://search.cpan.org/~danberr/ :|
|: http://entangle-photo.org -o- http://live.gnome.org/gtk-vnc :|
6:29 p.m.
New subject: [libvirt] [PATCH 10/12] rpc: Remove unused parameter in virKeepAliveStopInternal
The previous commit removed the only usage of ``all'' parameter in
virKeepAliveStopInternal, which was actually the only reason for having
virKeepAliveStopInternal. This effectively reverts most of commit
6446a9e20cc65561ce6061742baf35a3a63d5ba1.
---
src/libvirt_probes.d | 2 +-
src/rpc/virkeepalive.c | 22 ++++------------------
src/rpc/virkeepalive.h | 1 -
src/rpc/virnetclient.c | 2 +-
4 files changed, 6 insertions(+), 21 deletions(-)
diff --git a/src/libvirt_probes.d b/src/libvirt_probes.d
index ac6c546..0dac8f3 100644
--- a/src/libvirt_probes.d
+++ b/src/libvirt_probes.d
@@ -78,7 +78,7 @@ provider libvirt {
probe rpc_keepalive_ref(void *ka, void *client, int refs);
probe rpc_keepalive_free(void *ka, void *client, int refs);
probe rpc_keepalive_start(void *ka, void *client, int interval, int count);
- probe rpc_keepalive_stop(void *ka, void *client, bool all);
+ probe rpc_keepalive_stop(void *ka, void *client);
probe rpc_keepalive_send(void *ka, void *client, int prog, int vers, int proc);
probe rpc_keepalive_received(void *ka, void *client, int prog, int vers, int proc);
probe rpc_keepalive_timeout(void *ka, void *client, int coundToDeath, int idle);
diff --git a/src/rpc/virkeepalive.c b/src/rpc/virkeepalive.c
index 7984ada..70cf31e 100644
--- a/src/rpc/virkeepalive.c
+++ b/src/rpc/virkeepalive.c
@@ -328,14 +328,14 @@ cleanup:
}
-static void
-virKeepAliveStopInternal(virKeepAlivePtr ka, bool all ATTRIBUTE_UNUSED)
+void
+virKeepAliveStop(virKeepAlivePtr ka)
{
virKeepAliveLock(ka);
PROBE(RPC_KEEPALIVE_STOP,
- "ka=%p client=%p all=%d",
- ka, ka->client, all);
+ "ka=%p client=%p",
+ ka, ka->client);
if (ka->timer > 0) {
virEventRemoveTimeout(ka->timer);
@@ -346,20 +346,6 @@ virKeepAliveStopInternal(virKeepAlivePtr ka, bool all
ATTRIBUTE_UNUSED)
}
-void
-virKeepAliveStop(virKeepAlivePtr ka)
-{
- virKeepAliveStopInternal(ka, true);
-}
-
-
-void
-virKeepAliveStopSending(virKeepAlivePtr ka)
-{
- virKeepAliveStopInternal(ka, false);
-}
-
-
int
virKeepAliveTimeout(virKeepAlivePtr ka)
{
diff --git a/src/rpc/virkeepalive.h b/src/rpc/virkeepalive.h
index 62227d0..1e25214 100644
--- a/src/rpc/virkeepalive.h
+++ b/src/rpc/virkeepalive.h
@@ -49,7 +49,6 @@ int virKeepAliveStart(virKeepAlivePtr ka,
int interval,
unsigned int count);
void virKeepAliveStop(virKeepAlivePtr ka);
-void virKeepAliveStopSending(virKeepAlivePtr ka);
int virKeepAliveTimeout(virKeepAlivePtr ka);
bool virKeepAliveTrigger(virKeepAlivePtr ka,
diff --git a/src/rpc/virnetclient.c b/src/rpc/virnetclient.c
index 48c6a5d..25bafea 100644
--- a/src/rpc/virnetclient.c
+++ b/src/rpc/virnetclient.c
@@ -257,7 +257,7 @@ void
virNetClientKeepAliveStop(virNetClientPtr client)
{
virNetClientLock(client);
- virKeepAliveStopSending(client->keepalive);
+ virKeepAliveStop(client->keepalive);
virNetClientUnlock(client);
}
--
1.7.10.2
4:09 a.m.
New subject: [libvirt] [PATCH 10/12] rpc: Remove unused parameter in virKeepAliveStopInternal
On Wed, Jun 13, 2012 at 01:29:28AM +0200, Jiri Denemark wrote:
The previous commit removed the only usage of ``all''
parameter in
virKeepAliveStopInternal, which was actually the only reason for having
virKeepAliveStopInternal. This effectively reverts most of commit
6446a9e20cc65561ce6061742baf35a3a63d5ba1.
---
src/libvirt_probes.d | 2 +-
src/rpc/virkeepalive.c | 22 ++++------------------
src/rpc/virkeepalive.h | 1 -
src/rpc/virnetclient.c | 2 +-
4 files changed, 6 insertions(+), 21 deletions(-)
ACK
Daniel
--
|: http://berrange.com -o- http://www.flickr.com/photos/dberrange/ :|
|: http://libvirt.org -o- http://virt-manager.org :|
|: http://autobuild.org -o- http://search.cpan.org/~danberr/ :|
|: http://entangle-photo.org -o- http://live.gnome.org/gtk-vnc :|
6:29 p.m.
New subject: [libvirt] [PATCH 11/12] server rpc: Remove APIs for manipulating filters on locked client
We don't need to add or remove filters when client object is already
locked anymore. There's no reason to keep the *Locked variants of those
APIs.
---
src/rpc/virnetserverclient.c | 39 ++++++++++++---------------------------
1 file changed, 12 insertions(+), 27 deletions(-)
diff --git a/src/rpc/virnetserverclient.c b/src/rpc/virnetserverclient.c
index f3eb61a..a56031c 100644
--- a/src/rpc/virnetserverclient.c
+++ b/src/rpc/virnetserverclient.c
@@ -218,20 +218,21 @@ static void virNetServerClientUpdateEvent(virNetServerClientPtr
client)
}
-static int
-virNetServerClientAddFilterLocked(virNetServerClientPtr client,
- virNetServerClientFilterFunc func,
- void *opaque)
+int virNetServerClientAddFilter(virNetServerClientPtr client,
+ virNetServerClientFilterFunc func,
+ void *opaque)
{
virNetServerClientFilterPtr filter;
virNetServerClientFilterPtr *place;
- int ret = -1;
+ int ret;
if (VIR_ALLOC(filter) < 0) {
virReportOOMError();
- goto cleanup;
+ return -1;
}
+ virNetServerClientLock(client);
+
filter->id = client->nextFilterID++;
filter->func = func;
filter->opaque = opaque;
@@ -243,28 +244,18 @@ virNetServerClientAddFilterLocked(virNetServerClientPtr client,
ret = filter->id;
-cleanup:
- return ret;
-}
-
-int virNetServerClientAddFilter(virNetServerClientPtr client,
- virNetServerClientFilterFunc func,
- void *opaque)
-{
- int ret;
-
- virNetServerClientLock(client);
- ret = virNetServerClientAddFilterLocked(client, func, opaque);
virNetServerClientUnlock(client);
+
return ret;
}
-static void
-virNetServerClientRemoveFilterLocked(virNetServerClientPtr client,
- int filterID)
+void virNetServerClientRemoveFilter(virNetServerClientPtr client,
+ int filterID)
{
virNetServerClientFilterPtr tmp, prev;
+ virNetServerClientLock(client);
+
prev = NULL;
tmp = client->filters;
while (tmp) {
@@ -280,13 +271,7 @@ virNetServerClientRemoveFilterLocked(virNetServerClientPtr client,
prev = tmp;
tmp = tmp->next;
}
-}
-void virNetServerClientRemoveFilter(virNetServerClientPtr client,
- int filterID)
-{
- virNetServerClientLock(client);
- virNetServerClientRemoveFilterLocked(client, filterID);
virNetServerClientUnlock(client);
}
--
1.7.10.2
4:09 a.m.
New subject: [libvirt] [PATCH 11/12] server rpc: Remove APIs for manipulating filters on locked client
On Wed, Jun 13, 2012 at 01:29:29AM +0200, Jiri Denemark wrote:
We don't need to add or remove filters when client object is
already
locked anymore. There's no reason to keep the *Locked variants of those
APIs.
---
src/rpc/virnetserverclient.c | 39 ++++++++++++---------------------------
1 file changed, 12 insertions(+), 27 deletions(-)
ACK
Daniel
--
|: http://berrange.com -o- http://www.flickr.com/photos/dberrange/ :|
|: http://libvirt.org -o- http://virt-manager.org :|
|: http://autobuild.org -o- http://search.cpan.org/~danberr/ :|
|: http://entangle-photo.org -o- http://live.gnome.org/gtk-vnc :|
6:29 p.m.
New subject: [libvirt] [PATCH 12/12] client rpc: Send keepalive requests from IO event loop
In addition to keepalive responses, we also need to send keepalive
requests from client IO loop to properly detect dead connection in case
a libvirt API call is called from the main loop, which prevents any
timers to be called.
---
src/rpc/virnetclient.c | 12 ++++++++++++
1 file changed, 12 insertions(+)
diff --git a/src/rpc/virnetclient.c b/src/rpc/virnetclient.c
index 25bafea..033fda6 100644
--- a/src/rpc/virnetclient.c
+++ b/src/rpc/virnetclient.c
@@ -1284,6 +1284,7 @@ static int virNetClientIOEventLoop(virNetClientPtr client,
char ignore;
sigset_t oldmask, blockedsigs;
int timeout = -1;
+ virNetMessagePtr msg = NULL;
/* If we have existing SASL decoded data we don't want to sleep in
* the poll(), just check if any other FDs are also ready.
@@ -1297,6 +1298,10 @@ static int virNetClientIOEventLoop(virNetClientPtr client,
if (thiscall->nonBlock)
timeout = 0;
+ /* Limit timeout so that we can send keepalive request in time */
+ if (timeout == -1)
+ timeout = virKeepAliveTimeout(client->keepalive);
+
fds[0].events = fds[0].revents = 0;
fds[1].events = fds[1].revents = 0;
@@ -1342,6 +1347,13 @@ static int virNetClientIOEventLoop(virNetClientPtr client,
virNetClientLock(client);
+ if (virKeepAliveTrigger(client->keepalive, &msg)) {
+ client->wantClose = true;
+ } else if (msg && virNetClientQueueNonBlocking(client, msg) < 0) {
+ VIR_WARN("Could not queue keepalive request");
+ virNetMessageFree(msg);
+ }
+
/* If we have existing SASL decoded data, pretend
* the socket became readable so we consume it
*/
--
1.7.10.2
4:10 a.m.
New subject: [libvirt] [PATCH 12/12] client rpc: Send keepalive requests from IO event loop
On Wed, Jun 13, 2012 at 01:29:30AM +0200, Jiri Denemark wrote:
In addition to keepalive responses, we also need to send keepalive
requests from client IO loop to properly detect dead connection in case
a libvirt API call is called from the main loop, which prevents any
timers to be called.
---
src/rpc/virnetclient.c | 12 ++++++++++++
1 file changed, 12 insertions(+)
ACK
Daniel
--
|: http://berrange.com -o- http://www.flickr.com/photos/dberrange/ :|
|: http://libvirt.org -o- http://virt-manager.org :|
|: http://autobuild.org -o- http://search.cpan.org/~danberr/ :|
|: http://entangle-photo.org -o- http://live.gnome.org/gtk-vnc :|
4:54 a.m.
On Wed, Jun 13, 2012 at 01:29:18AM +0200, Jiri Denemark wrote:
So far, we were dropping non-blocking calls whenever sending them
would block.
In case a client is sending lots of stream calls (which are not supposed to
generate any reply), the assumption that having other calls in a queue is
sufficient to get a reply from the server doesn't work. I tried to fix this in
b1e374a7ac56927cfe62435179bf0bba1e08b372 but failed and reverted that commit.
While working on the proper fix, I discovered several other issues we had in
handling keepalive messages in client RPC code. See individual patches for more
details.
As a nice bonus, the fixed version is shorter by one line than the current
broken version :-)
Ok, this refactoring actually addresses many of the concerns I had
with the original design and so passes my totally subjective "feels
right" test. In particular the way we handle keepalives while in the
virNetClientIOEventLoop() method & use event handles while no, are the
key problem areas we're fixing with this.
On reviewing the code, I noticed the following (unrelated) issue we
should fix. First 'poll' can't return EWOULDBLOCK, and second, we're
checking errno so far away from the poll() call that we've probably
already trashed the original errno value.
Daniel
diff --git a/src/rpc/virnetclient.c b/src/rpc/virnetclient.c
index 033fda6..49d238e 100644
--- a/src/rpc/virnetclient.c
+++ b/src/rpc/virnetclient.c
@@ -1347,6 +1347,12 @@ static int virNetClientIOEventLoop(virNetClientPtr client,
virNetClientLock(client);
+ if (ret < 0) {
+ virReportSystemError(errno,
+ "%s", _("poll on socket failed"));
+ goto error;
+ }
+
if (virKeepAliveTrigger(client->keepalive, &msg)) {
client->wantClose = true;
} else if (msg && virNetClientQueueNonBlocking(client, msg) < 0) {
@@ -1375,15 +1381,6 @@ static int virNetClientIOEventLoop(virNetClientPtr client,
}
}
- if (ret < 0) {
- /* XXX what's this dubious errno check doing ? */
- if (errno == EWOULDBLOCK)
- continue;
- virReportSystemError(errno,
- "%s", _("poll on socket failed"));
- goto error;
- }
-
if (fds[0].revents & POLLOUT) {
if (virNetClientIOHandleOutput(client) < 0)
goto error;
--
|: http://berrange.com -o- http://www.flickr.com/photos/dberrange/ :|
|: http://libvirt.org -o- http://virt-manager.org :|
|: http://autobuild.org -o- http://search.cpan.org/~danberr/ :|
|: http://entangle-photo.org -o- http://live.gnome.org/gtk-vnc :|
8:21 a.m.
On Wed, Jun 13, 2012 at 10:54:02 +0100, Daniel P. Berrange wrote:
On Wed, Jun 13, 2012 at 01:29:18AM +0200, Jiri Denemark wrote:
> In case a client is sending lots of stream calls (which are not supposed to
> generate any reply), the assumption that having other calls in a queue is
> sufficient to get a reply from the server doesn't work. I tried to fix this in
> b1e374a7ac56927cfe62435179bf0bba1e08b372 but failed and reverted that commit.
> While working on the proper fix, I discovered several other issues we had in
> handling keepalive messages in client RPC code. See individual patches for more
> details.
>
> As a nice bonus, the fixed version is shorter by one line than the current
> broken version :-)
Ok, this refactoring actually addresses many of the concerns I had
with the original design and so passes my totally subjective "feels
right" test. In particular the way we handle keepalives while in the
virNetClientIOEventLoop() method & use event handles while no, are the
key problem areas we're fixing with this.
Thanks for the review, I tested scenarios covered by patches 2, 3, 4, 9, and
12 and all was working fine. I'll be pushing the series soon and I hope it
won't generate tons of bugs as I'll be on vacation for two weeks starting from
Friday :-P
On reviewing the code, I noticed the following (unrelated) issue we
should fix. First 'poll' can't return EWOULDBLOCK,
Oh yeah, the code I always wanted to remove but was afraid of doing so :-)
and second, we're checking errno so far away from the poll() call
that we've
probably already trashed the original errno value.
Daniel
diff --git a/src/rpc/virnetclient.c b/src/rpc/virnetclient.c
index 033fda6..49d238e 100644
--- a/src/rpc/virnetclient.c
+++ b/src/rpc/virnetclient.c
@@ -1347,6 +1347,12 @@ static int virNetClientIOEventLoop(virNetClientPtr client,
virNetClientLock(client);
+ if (ret < 0) {
+ virReportSystemError(errno,
+ "%s", _("poll on socket failed"));
+ goto error;
+ }
+
if (virKeepAliveTrigger(client->keepalive, &msg)) {
client->wantClose = true;
} else if (msg && virNetClientQueueNonBlocking(client, msg) < 0) {
@@ -1375,15 +1381,6 @@ static int virNetClientIOEventLoop(virNetClientPtr client,
}
}
- if (ret < 0) {
- /* XXX what's this dubious errno check doing ? */
- if (errno == EWOULDBLOCK)
- continue;
- virReportSystemError(errno,
- "%s", _("poll on socket failed"));
- goto error;
- }
-
if (fds[0].revents & POLLOUT) {
if (virNetClientIOHandleOutput(client) < 0)
goto error;
ACK, I'll push this as a 13th patch in this series.
Jirka
4546
days inactive
4547
days old
31 comments
3 participants
participants (3)
-
Daniel P. Berrange -
Eric Blake -
Jiri Denemark