[Qemu-devel] Re: [RFC][PATCH v7 04/16] virtagent: bi-directional RPC handling logic

[Adam Litke]

On Mon, 2011-03-07 at 14:10 -0600, Michael Roth wrote:
> This implements the state machine/logic used to manage
> send/receive/execute phases of RPCs we send or receive. It does so using
> a set of abstract methods we implement with the application and
> transport level code which will follow.
> 
> Signed-off-by: Michael Roth <address@hidden>
> ---
>  virtagent-manager.c |  326 
> +++++++++++++++++++++++++++++++++++++++++++++++++++
>  virtagent-manager.h |  130 ++++++++++++++++++++
>  2 files changed, 456 insertions(+), 0 deletions(-)
>  create mode 100644 virtagent-manager.c
>  create mode 100644 virtagent-manager.h
> 
> diff --git a/virtagent-manager.c b/virtagent-manager.c
> new file mode 100644
> index 0000000..51d26a3
> --- /dev/null
> +++ b/virtagent-manager.c
> @@ -0,0 +1,326 @@
> +/*
> + * virtagent - job queue management
> + *
> + * Copyright IBM Corp. 2011
> + *
> + * Authors:
> + *  Michael Roth      <address@hidden>
> + *
> + * This work is licensed under the terms of the GNU GPL, version 2 or later.
> + * See the COPYING file in the top-level directory.
> + *
> + */
> +
> +#include "virtagent-common.h"
> +
> +typedef struct VAServerJob {
> +    char tag[64];
> +    void *opaque;
> +    VAServerJobOps ops;
> +    QTAILQ_ENTRY(VAServerJob) next;
> +    enum {
> +        VA_SERVER_JOB_STATE_NEW = 0,
> +        VA_SERVER_JOB_STATE_BUSY,
> +        VA_SERVER_JOB_STATE_EXECUTED,
> +        VA_SERVER_JOB_STATE_SENT,
> +        VA_SERVER_JOB_STATE_DONE,
> +    } state;
> +} VAServerJob;
> +
> +typedef struct VAClientJob {
> +    char tag[64];
> +    void *opaque;
> +    void *resp_opaque;
> +    VAClientJobOps ops;
> +    QTAILQ_ENTRY(VAClientJob) next;
> +    enum {
> +        VA_CLIENT_JOB_STATE_NEW = 0,
> +        VA_CLIENT_JOB_STATE_BUSY,
> +        VA_CLIENT_JOB_STATE_SENT,
> +        VA_CLIENT_JOB_STATE_READ,
> +        VA_CLIENT_JOB_STATE_DONE,
> +    } state;
> +} VAClientJob;
> +
> +#define SEND_COUNT_MAX 1
> +#define EXECUTE_COUNT_MAX 4

It's not immediately clear what the difference between SEND_COUNT_MAX
and EXECUTE_COUNT_MAX is.  Some comments would help.  Also, will the
code work if these numbers are changed?  If not, a note about what
someone needs to look at when changing these would seem appropriate.

> +
> +struct VAManager {
> +    int send_count; /* sends in flight */
> +    int execute_count; /* number of jobs currently executing */
> +    QTAILQ_HEAD(, VAServerJob) server_jobs;
> +    QTAILQ_HEAD(, VAClientJob) client_jobs;
> +};
> +
> +/* server job operations/helpers */
> +
> +static VAServerJob *va_server_job_by_tag(VAManager *m, const char *tag)
> +{
> +    VAServerJob *j;
> +    QTAILQ_FOREACH(j, &m->server_jobs, next) {
> +        if (strcmp(j->tag, tag) == 0) {
> +            return j;
> +        }
> +    }
> +    return NULL;
> +}
> +
> +int va_server_job_add(VAManager *m, const char *tag, void *opaque,
> +                      VAServerJobOps ops)
> +{
> +    VAServerJob *j = qemu_mallocz(sizeof(VAServerJob));
> +    TRACE("called");

Qemu has a good tracing infrastructure.  If this is trace point is
useful enough to keep around, it should try to use that.  If it's not
that important, I'd remove it entirely.  I believe this has been flagged
in an earlier RFC too.

> +    j->state = VA_SERVER_JOB_STATE_NEW;
> +    j->ops = ops;
> +    j->opaque = opaque;
> +    memset(j->tag, 0, 64);
> +    pstrcpy(j->tag, 63, tag);

Magic numbers.  Should use something like #define TAG_LEN 64

> +    QTAILQ_INSERT_TAIL(&m->server_jobs, j, next);
> +    va_kick(m);
> +    return 0;
> +}
> +
> +static void va_server_job_execute(VAServerJob *j)
> +{
> +    TRACE("called");
> +    j->state = VA_SERVER_JOB_STATE_BUSY;
> +    j->ops.execute(j->opaque, j->tag);
> +}
> +
> +/* TODO: need a way to pass information back */
> +void va_server_job_execute_done(VAManager *m, const char *tag)
> +{
> +    VAServerJob *j = va_server_job_by_tag(m, tag);
> +    TRACE("called");
> +    if (!j) {
> +        LOG("server job with tag \"%s\" not found", tag);
> +        return;
> +    }
> +    j->state = VA_SERVER_JOB_STATE_EXECUTED;
> +    va_kick(m);
> +}
> +
> +static void va_server_job_send(VAServerJob *j)
> +{
> +    TRACE("called");
> +    j->state = VA_SERVER_JOB_STATE_BUSY;
> +    j->ops.send(j->opaque, j->tag);
> +}
> +
> +void va_server_job_send_done(VAManager *m, const char *tag)
> +{
> +    VAServerJob *j = va_server_job_by_tag(m, tag);
> +    TRACE("called");
> +    if (!j) {
> +        LOG("server job with tag \"%s\" not found", tag);
> +        return;
> +    }
> +    j->state = VA_SERVER_JOB_STATE_SENT;
> +    m->send_count--;
> +    va_kick(m);
> +}
> +
> +static void va_server_job_callback(VAServerJob *j)
> +{
> +    TRACE("called");
> +    j->state = VA_SERVER_JOB_STATE_BUSY;
> +    if (j->ops.callback) {
> +        j->ops.callback(j->opaque, j->tag);
> +    }
> +    j->state = VA_SERVER_JOB_STATE_DONE;
> +}
> +
> +void va_server_job_cancel(VAManager *m, const char *tag)
> +{
> +    VAServerJob *j = va_server_job_by_tag(m, tag);
> +    TRACE("called");
> +    if (!j) {
> +        LOG("server job with tag \"%s\" not found", tag);
> +        return;
> +    }
> +    /* TODO: need to decrement sends/execs in flight appropriately */
> +    /* make callback and move to done state, kick() will handle cleanup */
> +    va_server_job_callback(j);
> +    va_kick(m);
> +}
> +
> +/* client job operations */
> +
> +static VAClientJob *va_client_job_by_tag(VAManager *m, const char *tag)
> +{
> +    VAClientJob *j;
> +    QTAILQ_FOREACH(j, &m->client_jobs, next) {
> +        if (strcmp(j->tag, tag) == 0) {
> +            return j;
> +        }
> +    }
> +    return NULL;
> +}
> +
> +int va_client_job_add(VAManager *m, const char *tag, void *opaque,
> +                      VAClientJobOps ops)
> +{
> +    VAClientJob *j = qemu_mallocz(sizeof(VAClientJob));
> +    TRACE("called");
> +    j->ops = ops;
> +    j->opaque = opaque;
> +    memset(j->tag, 0, 64);
> +    pstrcpy(j->tag, 63, tag);
> +    QTAILQ_INSERT_TAIL(&m->client_jobs, j, next);
> +    va_kick(m);
> +    return 0;
> +}
> +
> +static void va_client_job_send(VAClientJob *j)
> +{
> +    TRACE("called");
> +    j->state = VA_CLIENT_JOB_STATE_BUSY;
> +    j->ops.send(j->opaque, j->tag);
> +}
> +
> +void va_client_job_send_done(VAManager *m, const char *tag)
> +{
> +    VAClientJob *j = va_client_job_by_tag(m, tag);
> +    TRACE("called");
> +    if (!j) {
> +        LOG("client job with tag \"%s\" not found", tag);
> +        return;
> +    }
> +    j->state = VA_CLIENT_JOB_STATE_SENT;
> +    m->send_count--;
> +    va_kick(m);
> +}
> +
> +void va_client_job_read_done(VAManager *m, const char *tag, void *resp)
> +{
> +    VAClientJob *j = va_client_job_by_tag(m, tag);
> +    TRACE("called");
> +    if (!j) {
> +        LOG("client job with tag \"%s\" not found", tag);
> +        return;
> +    }
> +    j->state = VA_CLIENT_JOB_STATE_READ;
> +    j->resp_opaque = resp;
> +    va_kick(m);
> +}
> +
> +static void va_client_job_callback(VAClientJob *j)
> +{
> +    TRACE("called");
> +    j->state = VA_CLIENT_JOB_STATE_BUSY;
> +    if (j->ops.callback) {
> +        j->ops.callback(j->opaque, j->resp_opaque, j->tag);
> +    }
> +    j->state = VA_CLIENT_JOB_STATE_DONE;
> +}
> +
> +void va_client_job_cancel(VAManager *m, const char *tag)
> +{
> +    VAClientJob *j = va_client_job_by_tag(m, tag);
> +    TRACE("called");
> +    if (!j) {
> +        LOG("client job with tag \"%s\" not found", tag);
> +        return;
> +    }
> +    /* TODO: need to decrement sends/execs in flight appropriately */
> +    /* make callback and move to done state, kick() will handle cleanup */
> +    va_client_job_callback(j);
> +    va_kick(m);
> +}
> +
> +/* general management functions */
> +
> +VAManager *va_manager_new(void)
> +{
> +    VAManager *m = qemu_mallocz(sizeof(VAManager));
> +    QTAILQ_INIT(&m->client_jobs);
> +    QTAILQ_INIT(&m->server_jobs);
> +    return m;
> +}
> +
> +static void va_process_server_job(VAManager *m, VAServerJob *sj)
> +{
> +    switch (sj->state) {
> +        case VA_SERVER_JOB_STATE_NEW:
> +            TRACE("marker");
> +            va_server_job_execute(sj);
> +            break;
> +        case VA_SERVER_JOB_STATE_EXECUTED:
> +            TRACE("marker");
> +            if (m->send_count < SEND_COUNT_MAX) {
> +                TRACE("marker");
> +                va_server_job_send(sj);
> +                m->send_count++;
> +            }
> +            break;
> +        case VA_SERVER_JOB_STATE_SENT:
> +            TRACE("marker");
> +            va_server_job_callback(sj);
> +            break;
> +        case VA_SERVER_JOB_STATE_BUSY:
> +            TRACE("marker, server job currently busy");
> +            break;
> +        case VA_SERVER_JOB_STATE_DONE:
> +            TRACE("marker");
> +            QTAILQ_REMOVE(&m->server_jobs, sj, next);
> +            break;
> +        default:
> +            LOG("error, unknown server job state");
> +            break;
> +    }
> +}
> +
> +static void va_process_client_job(VAManager *m, VAClientJob *cj)
> +{
> +    switch (cj->state) {
> +        case VA_CLIENT_JOB_STATE_NEW:
> +            TRACE("marker");
> +            if (m->send_count < SEND_COUNT_MAX) {
> +                TRACE("marker");
> +                va_client_job_send(cj);
> +                m->send_count++;
> +            }
> +            break;
> +        case VA_CLIENT_JOB_STATE_SENT:
> +            TRACE("marker");
> +            //nothing to do here, awaiting read_done()
> +            break;
> +        case VA_CLIENT_JOB_STATE_READ:
> +            TRACE("marker");
> +            va_client_job_callback(cj);
> +            break;
> +        case VA_CLIENT_JOB_STATE_DONE:
> +            TRACE("marker");
> +            QTAILQ_REMOVE(&m->client_jobs, cj, next);
> +            break;
> +        case VA_CLIENT_JOB_STATE_BUSY:
> +            TRACE("marker, client job currently busy");
> +            break;
> +        default:
> +            LOG("error, unknown client job state");
> +            break;
> +    }
> +}
> +
> +void va_kick(VAManager *m)
> +{
> +    VAServerJob *sj, *sj_tmp;
> +    VAClientJob *cj, *cj_tmp;
> +
> +    TRACE("called");
> +    TRACE("send_count: %u, execute_count: %u", m->send_count, 
> m->execute_count);
> +
> +    /* TODO: make sure there is no starvation of jobs/operations here */
> +
> +    /* look for any work to be done among pending server jobs */
> +    QTAILQ_FOREACH_SAFE(sj, &m->server_jobs, next, sj_tmp) {
> +        TRACE("marker, server tag: %s", sj->tag);
> +        va_process_server_job(m, sj);
> +    }
> +
> +    /* look for work to be done among pending client jobs */
> +    QTAILQ_FOREACH_SAFE(cj, &m->client_jobs, next, cj_tmp) {
> +        TRACE("marker, client tag: %s", cj->tag);
> +        va_process_client_job(m, cj);
> +    }
> +}
> diff --git a/virtagent-manager.h b/virtagent-manager.h
> new file mode 100644
> index 0000000..7b463fb
> --- /dev/null
> +++ b/virtagent-manager.h
> @@ -0,0 +1,130 @@
> +#ifndef VIRTAGENT_MANAGER_H
> +#define VIRTAGENT_MANAGER_H
> +
> +#include "qemu-common.h"
> +#include "qemu-queue.h"
> +
> +/*
> + * Protocol Overview:
> + *
> + * The virtagent protocol depends on a state machine to manage communication
> + * over a single connection stream, currently a virtio or isa serial channel.
> + * The basic characterization of the work being done is that clients
> + * send/handle client jobs locally, which are then read/handled remotely as
> + * server jobs. A client job consists of a request which is sent, and a
> + * response which is eventually recieved. A server job consists of a request
> + * which is recieved from the other end, and a response which is sent back.

"i before e, except after c ..." (I misspell receive all the time too).

> + * 
> + * Server jobs are given priority over client jobs, i.e. if we send a client
> + * job (our request) and recieve a server job (their request), rather than
> + * await a response to the client job, we immediately begin processing the
> + * server job and then send back the response. This prevents us from being
> + * deadlocked in a situation where both sides have sent a client job and are
> + * awaiting the response before handling the other side's client job.
> + *
> + * Multiple in-flight requests are supported, but high request rates can
> + * potentially starve out the other side's client jobs / requests, so we'll
> + * behaved participants should periodically backoff on high request rates, or
> + * limit themselves to 1 request at a time (anything more than 1 can still
> + * potentionally remove any window for the other end to service it's own
> + * client jobs, since we can begin sending the next request before it begins
> + * send the response for the 2nd).
> + * 
> + * On a related note, in the future, bidirectional user/session-level guest
> + * agents may also be supported via a forwarding service made available
> + * through the system-level guest agent. In this case it is up to the
> + * system-level agent to handle forwarding requests in such a way that we
> + * don't starve the host-side service out sheerly by having too many
> + * sessions/users trying to send RPCs at a constant rate. This would be
> + * supported through this job Manager via an additional "forwarder" job type.
> + *
> + * To encapsulate some of this logic, we define here a "Manager" class, which
> + * provides an abstract interface to a state machine which handles most of
> + * the above logic transparently to the transport/application-level code.
> + * This also makes it possible to utilize alternative
> + * transport/application-level protocols in the future.
> + *
> + */
> +
> +/*
> + * Two types of jobs are generated from various components of virtagent.
> + * Each job type has a priority, and a set of prioritized functions as well.
> + *
> + * The read handler generates new server jobs as it recieves requests from
> + * the channel. Server jobs make progress through the following operations.
> + *
> + * EXECUTE->EXECUTE_DONE->SEND->SEND_DONE
> + *
> + * EXECUTE (provided by user, manager calls)
> + * When server jobs are added, eventually (as execution slots become
> + * available) an execute() will be called to begin executing the job. An
> + * error value will be returned if there is no room in the queue for another
> + * server job.
> + *
> + * EXECUTE_DONE (provided by manager, user calls)
> + * As server jobs complete, execute_completed() is called to update execution
> + * status of that job (failure/success), inject the payload, and kick off the
> + * next operation.
> + *
> + * SEND (provided by user, manager calls)
> + * Eventually the send() operation is made. This will cause the send handler
> + * to begin sending the response.
> + *
> + * SEND_DONE (provided by manager, user calls)
> + * Upon completion of that send, the send_completed() operation will be
> + * called. This will free up the job, and kick off the next operation.
> + */

Very helpful protocol overview.  Thanks for adding this.

> +typedef int (va_job_op)(void *opaque, const char *tag);
> +typedef struct VAServerJobOps {
> +    va_job_op *execute;
> +    va_job_op *send;
> +    va_job_op *callback;
> +} VAServerJobOps;
> +
> +/*
> + * The client component generates new client jobs as they're made by
> + * virtagent in response to monitored events or user-issued commands.
> + * Client jobs progress via the following operations.
> + *
> + * SEND->SEND_DONE->READ_DONE
> + * 
> + * SEND (provided by user, called by manager)
> + * After client jobs are added, send() will eventually be called to queue
> + * the job up for xmit over the channel.
> + *
> + * SEND_DONE (provided by manager, called by user)
> + * Upon completion of the send, send_completed() should be called with
> + * failure/success indication.
> + *
> + * READ_DONE (provided by manager, called by user)
> + * When a response for the request is read back via the transport layer,
> + * read_done() will be called by the user to indicate success/failure,
> + * inject the response, and make the associated callback.
> + */
> +typedef int (va_client_job_cb)(void *opaque, void *resp_opaque,
> +                               const char *tag);
> +typedef struct VAClientJobOps {
> +    va_job_op *send;
> +    va_client_job_cb *callback;
> +} VAClientJobOps;
> +
> +typedef struct VAManager VAManager;
> +
> +VAManager *va_manager_new(void);
> +void va_kick(VAManager *m);
> +
> +/* interfaces for server jobs */
> +int va_server_job_add(VAManager *m, const char *tag, void *opaque,
> +                      VAServerJobOps ops);
> +void va_server_job_execute_done(VAManager *m, const char *tag);
> +void va_server_job_send_done(VAManager *m, const char *tag);
> +void va_server_job_cancel(VAManager *m, const char *tag);
> +
> +/* interfaces for client jobs */
> +int va_client_job_add(VAManager *m, const char *tag, void *opaque,
> +                      VAClientJobOps ops);
> +void va_client_job_cancel(VAManager *m, const char *tag);
> +void va_client_job_send_done(VAManager *m, const char *tag);
> +void va_client_job_read_done(VAManager *m, const char *tag, void *resp);
> +
> +#endif /* VIRTAGENT_MANAGER_H */

-- 
Thanks,
Adam