[libvirt] PATCH: 17/25: Concurrent client dispatch in libvirtd
Jim Meyering
jim at meyering.net
Tue Jan 20 09:12:04 UTC 2009
"Daniel P. Berrange" <berrange at redhat.com> wrote:
> On Tue, Jan 13, 2009 at 05:45:43PM +0000, Daniel P. Berrange wrote:
>> Historically libvirtd was single threaded, serializing all
>> requests across clients. An recent patch allowed multiple
>> threads, so multiple clients could run in parallel. A single
>> client was still serialized.
>>
>> This patch removes that final restriction, allowing a single
>> client to have multiple in-flight RPC requests & replies.
>> Each client now has 3 variables
>>
>> - rx: zero or one. If ready for, or in process of reading
>> a message this will be non-null. If we're throttling the
>> client requests, it'll be NULL. Once completely read, moved
>> to the 'dx' queue.
>> - dx: zero or many. Requests read off wire currently waiting
>> to be picked up for processing by a worker thread. Once a
>> worker is available the message is removed from the 'dx'
>> queue for duration of processing. A reply is put on the
>> 'tx' queue once a call is finished
>> - tx: zero or many. Replies in process of, or ready to be,
>> sent back to a client. Also includes any asynchronous
>> event notifications to be sent.
...
Phew. That took nearly 2 hours.
And the review was relatively superficial in that sometimes
I skimmed past things in context that I'm not familiar with.
> diff --git a/qemud/qemud.c b/qemud/qemud.c
...
> -static int qemudClientReadBuf(struct qemud_server *server,
> - struct qemud_client *client,
> +/*
> + * Read data into buffer using wire decoding (plain or TLS)
> + */
> +static int qemudClientReadBuf(struct qemud_client *client,
> char *data, unsigned len) {
> int ret;
Probably doesn't affect correctness, but ret should be of type
ssize_t, to match types used by read and gnutls_record_recv.
Might as well declare "len" to be ssize_t, too, and then
assert that it's always positive. The assertion is nice
to catch a reversed difference in the caller -- nicer to
debug than a segfault.
> /*qemudDebug ("qemudClientRead: len = %d", len);*/
>
> if (!client->tlssession) {
> - if ((ret = read (client->fd, data, len)) <= 0) {
> - if (ret == 0 || errno != EAGAIN) {
> - if (ret != 0)
> - VIR_ERROR(_("read: %s"), strerror (errno));
> - qemudDispatchClientFailure(server, client);
> - }
> + ret = read (client->fd, data, len);
> + if (ret == -1 && (errno == EAGAIN ||
> + errno == EINTR))
> + return 0;
> + if (ret <= 0) {
> + if (ret != 0)
> + VIR_ERROR(_("read: %s"), strerror (errno));
> + qemudDispatchClientFailure(client);
> return -1;
> }
> } else {
> ret = gnutls_record_recv (client->tlssession, data, len);
> - if (qemudRegisterClientEvent (server, client, 1) < 0)
> - qemudDispatchClientFailure (server, client);
> - else if (ret <= 0) {
> - if (ret == 0 || (ret != GNUTLS_E_AGAIN &&
> - ret != GNUTLS_E_INTERRUPTED)) {
> - if (ret != 0)
> - VIR_ERROR(_("gnutls_record_recv: %s"),
> - gnutls_strerror (ret));
> - qemudDispatchClientFailure (server, client);
> - }
> +
> + if (ret == -1 && (ret == GNUTLS_E_AGAIN &&
> + ret == GNUTLS_E_INTERRUPTED))
> + return 0;
The above is dead code, since the condition can never be true.
It should be testing "ret < 0", not ret == -1.
Also, the 2nd "&&" should be "||".
if (ret < 0 && (ret == GNUTLS_E_AGAIN ||
ret == GNUTLS_E_INTERRUPTED))
return 0;
> + if (ret <= 0) {
> + if (ret != 0)
> + VIR_ERROR(_("gnutls_record_recv: %s"),
> + gnutls_strerror (ret));
> + qemudDispatchClientFailure(client);
> return -1;
> }
...
> @@ -1350,30 +1415,33 @@ static int qemudClientReadSASL(struct qe
>
> /* Need to read some more data off the wire */
> if (client->saslDecoded == NULL) {
> + int ret;
ssize_t, to match sasl_decode return type. no big deal
> char encoded[8192];
> int encodedLen = sizeof(encoded);
Likewise, "unsigned". also no big deal
> - encodedLen = qemudClientReadBuf(server, client, encoded, encodedLen);
> + encodedLen = qemudClientReadBuf(client, encoded, encodedLen);
>
> if (encodedLen < 0)
> return -1;
>
> - sasl_decode(client->saslconn, encoded, encodedLen,
> - &client->saslDecoded, &client->saslDecodedLength);
> + ret = sasl_decode(client->saslconn, encoded, encodedLen,
> + &client->saslDecoded, &client->saslDecodedLength);
> + if (ret != SASL_OK)
> + return -1;
If this ever fails, it's sure be nice to log why,
but I don't see a strerror analog for SASL_* values.
At least log the integer value.
...
> +/*
> + * Read data until we get a complete message to process
> + */
> +static void qemudDispatchClientRead(struct qemud_server *server,
> + struct qemud_client *client) {
> /*qemudDebug ("qemudDispatchClientRead: mode = %d", client->mode);*/
>
> - switch (client->mode) {
> - case QEMUD_MODE_RX_HEADER: {
> +readmore:
> + if (qemudClientRead(client) < 0)
> + return; /* Error, or blocking */
> +
> + if (client->rx->bufferOffset < client->rx->bufferLength)
> + return; /* Not read enough */
> +
> + /* Either done with length word header */
> + if (client->rx->bufferLength == REMOTE_MESSAGE_HEADER_XDR_LEN) {
> + int len;
> XDR x;
>
> - if (qemudClientRead(server, client) < 0)
> - return; /* Error, or blocking */
> + xdrmem_create(&x, client->rx->buffer, client->rx->bufferLength, XDR_DECODE);
>
> - if (client->bufferOffset < client->bufferLength)
> - return; /* Not read enough */
> -
> - xdrmem_create(&x, client->buffer, client->bufferLength, XDR_DECODE);
> -
> - if (!xdr_u_int(&x, &len)) {
> + if (!xdr_int(&x, &len)) {
> xdr_destroy (&x);
> DEBUG0("Failed to decode packet length");
> - qemudDispatchClientFailure(server, client);
> + qemudDispatchClientFailure(client);
> return;
> }
> xdr_destroy (&x);
>
> + /* Length includes the size of the length word itself */
> + len -= REMOTE_MESSAGE_HEADER_XDR_LEN;
> +
> if (len > REMOTE_MESSAGE_MAX) {
> DEBUG("Packet length %u too large", len);
> - qemudDispatchClientFailure(server, client);
> + qemudDispatchClientFailure(client);
> return;
> }
>
> /* Length include length of the length field itself, so
> * check minimum size requirements */
> - if (len <= REMOTE_MESSAGE_HEADER_XDR_LEN) {
> + if (len <= 0) {
> DEBUG("Packet length %u too small", len);
"len" may be negative here, so printing with %u will give a
misleading diagnostic.
Better use the original value, "len + REMOTE_MESSAGE_HEADER_XDR_LEN",
which is more likely to be non-negative. Might as well use %d,
in case even the original value is negative.
> - qemudDispatchClientFailure(server, client);
> + qemudDispatchClientFailure(client);
> return;
...
> static void
> qemudDispatchClientEvent(int watch, int fd, int events, void *opaque) {
> @@ -1642,59 +1734,66 @@ qemudDispatchClientEvent(int watch, int
> virMutexLock(&server->lock);
>
> for (i = 0 ; i < server->nclients ; i++) {
> + virMutexLock(&server->clients[i]->lock);
> if (server->clients[i]->watch == watch) {
> client = server->clients[i];
> break;
> }
> + virMutexUnlock(&server->clients[i]->lock);
> }
>
> + virMutexUnlock(&server->lock);
> +
> if (!client) {
> - virMutexUnlock(&server->lock);
> return;
> }
>
> - virMutexLock(&client->lock);
> - virMutexUnlock(&server->lock);
> + if (client->fd != fd) {
> + virMutexUnlock(&client->lock);
> + return;
> + }
>
> - if (client->fd != fd)
> - return;
> + if (events & (VIR_EVENT_HANDLE_WRITABLE |
> + VIR_EVENT_HANDLE_READABLE)) {
> + if (client->handshake) {
> + qemudDispatchClientHandshake(server, client);
> + } else {
> + if (events & VIR_EVENT_HANDLE_WRITABLE)
> + qemudDispatchClientWrite(server, client);
> + if (events == VIR_EVENT_HANDLE_READABLE)
> + qemudDispatchClientRead(server, client);
Why test with & to write, and then with "==" to read?
That makes it so we don't read when we've just written
(i.e., if both read and write bits were set).
Hmm... looking at the old code, it seems the intent is to
do only one of read/write operations at a time.
In that case, just use an "else":
if (events & VIR_EVENT_HANDLE_WRITABLE)
qemudDispatchClientWrite(server, client);
else
qemudDispatchClientRead(server, client);
...
> + /* NB, will get HANGUP + READABLE at same time upon
> + * disconnect */
Oh.
Now I suspect that the preceding "events == VIR_EVENT_HANDLE_READABLE"
test is to avoid reading upon HANGUP. But it also inhibits reading
when/if READABLE and WRITABLE are both set (probably doesn't matter).
Please add a comment.
> + if (events & (VIR_EVENT_HANDLE_ERROR |
> + VIR_EVENT_HANDLE_HANGUP))
> + qemudDispatchClientFailure(client);
> +
> virMutexUnlock(&client->lock);
> }
...
> diff --git a/qemud/qemud.h b/qemud/qemud.h
...
> +struct qemud_client_message {
...
> + int nrequests;
Logically, this should be an unsigned type.
But that means max_clients should be, too,
since they're compared, but max_clients comes
from the config file, which currently uses "int" (via GET_CONF_INT).
Maybe we need GET_CONF_UINT? I wonder if a bogus config "int"
value of 2^32 or 2^64 maps back to 0.
More information about the libvir-list
mailing list