Re: [Qemu-devel] [PATCH v4][ 1/7] guest agent: worker thread class

[Anthony Liguori <anthony@codemonkey.ws>]

On 06/03/2011 05:48 PM, Michael Roth wrote:
> Signed-off-by: Michael Roth<mdroth@linux.vnet.ibm.com>
> ---
>   qga/guest-agent-worker.c |  179 ++++++++++++++++++++++++++++++++++++++++++++++
>   1 files changed, 179 insertions(+), 0 deletions(-)
>   create mode 100644 qga/guest-agent-worker.c
>
> diff --git a/qga/guest-agent-worker.c b/qga/guest-agent-worker.c
> new file mode 100644
> index 0000000..e5fc845
> --- /dev/null
> +++ b/qga/guest-agent-worker.c
> @@ -0,0 +1,179 @@
> +/*
> + * QEMU Guest Agent worker thread interfaces
> + *
> + * Copyright IBM Corp. 2011
> + *
> + * Authors:
> + *  Michael Roth<mdroth@linux.vnet.ibm.com>
> + *
> + * 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<glib.h>
> +#include<stdlib.h>
> +#include<stdio.h>
> +#include<stdbool.h>
> +#include<pthread.h>
> +#include<errno.h>
> +#include<string.h>
> +#include "error.h"
> +#include "qga/guest-agent-core.h"
> +
> +struct GAWorker {
> +    pthread_t thread;
> +    ga_worker_func execute;
> +    pthread_mutex_t input_mutex;
> +    pthread_cond_t input_avail_cond;
> +    void *input;
> +    bool input_avail;
> +    pthread_mutex_t output_mutex;
> +    pthread_cond_t output_avail_cond;
> +    void *output;
> +    Error *output_error;
> +    bool output_avail;
> +};
> +
> +static void *worker_run(void *worker_p)
> +{
> +    GAWorker *worker = worker_p;
> +    Error *err = NULL;
> +    void *input = NULL, *output = NULL;
> +
> +    while (1) {
> +        /* wait for input */
> +        pthread_mutex_lock(&worker->input_mutex);

It's more typical to push the lock() and unlock to the outside of the 
loop.  That makes it safer in the event that someone adds a continue or 
break.

Regards,

Anthony Liguori

> +        while (!worker->input_avail) {
> +            pthread_cond_wait(&worker->input_avail_cond,&worker->input_mutex);
> +        }
> +        input = worker->input;
> +        worker->input = NULL;
> +        worker->input_avail = false;
> +        pthread_mutex_unlock(&worker->input_mutex);
> +
> +        /* process input. input points to shared data, so if we ever add
> +         * asynchronous dispatch, we'll need to copy the input instead
> +         */
> +        worker->execute(input,&output,&err);
> +
> +        /* signal waiters */
> +        pthread_mutex_lock(&worker->output_mutex);
> +        worker->output = output;
> +        worker->output_error = err;
> +        worker->output_avail = true;
> +        pthread_cond_signal(&worker->output_avail_cond);
> +        pthread_mutex_unlock(&worker->output_mutex);
> +    }
> +
> +    return NULL;
> +}
> +
> +static void ga_worker_set_input(GAWorker *worker, void *input)
> +{
> +    pthread_mutex_lock(&worker->input_mutex);
> +
> +    /* provide input for thread, and signal it */
> +    worker->input = input;
> +    worker->input_avail = true;
> +    pthread_cond_signal(&worker->input_avail_cond);
> +
> +    pthread_mutex_unlock(&worker->input_mutex);
> +}
> +
> +static bool ga_worker_get_output(GAWorker *worker, void **output, int timeout)
> +{
> +    struct timespec ts;
> +    GTimeVal tv;
> +    bool timed_out = false;
> +    int ret;
> +
> +    pthread_mutex_lock(&worker->output_mutex);
> +
> +    while (!worker->output_avail) {
> +        if (timeout>  0) {
> +            g_get_current_time(&tv);
> +            g_time_val_add(&tv, timeout * 1000);
> +            ts.tv_sec = tv.tv_sec;
> +            ts.tv_nsec = tv.tv_usec * 1000;
> +            ret = pthread_cond_timedwait(&worker->output_avail_cond,
> +&worker->output_mutex,&ts);
> +            if (ret == ETIMEDOUT) {
> +                timed_out = true;
> +                goto out;
> +            }
> +        } else {
> +            ret = pthread_cond_wait(&worker->output_avail_cond,
> +&worker->output_mutex);
> +        }
> +    }
> +
> +    /* handle output from thread */
> +    worker->output_avail = false;
> +    *output = worker->output;
> +    worker->output = NULL;
> +
> +out:
> +    pthread_mutex_unlock(&worker->output_mutex);
> +    return timed_out;
> +}
> +
> +bool ga_worker_dispatch(GAWorker *worker, void *input, void *output,
> +                        int timeout, Error **errp)
> +{
> +    ga_worker_set_input(worker, input);
> +    return ga_worker_get_output(worker, output, timeout);
> +}
> +
> +static void ga_worker_start(GAWorker *worker)
> +{
> +    int ret;
> +
> +    pthread_cond_init(&worker->input_avail_cond, NULL);
> +    pthread_cond_init(&worker->output_avail_cond, NULL);
> +    pthread_mutex_init(&worker->input_mutex, NULL);
> +    pthread_mutex_init(&worker->output_mutex, NULL);
> +    worker->output_avail = false;
> +    worker->input_avail = false;
> +
> +    ret = pthread_create(&worker->thread, NULL, worker_run, worker);
> +    if (ret == -1) {
> +        g_error("error: %s", strerror(errno));
> +    }
> +}
> +
> +static void ga_worker_stop(GAWorker *worker)
> +{
> +    int ret;
> +    void *status;
> +
> +    ret = pthread_cancel(worker->thread);
> +    if (ret == -1) {
> +        g_error("pthread_cancel() failed: %s", strerror(errno));
> +    }
> +
> +    ret = pthread_join(worker->thread,&status);
> +    if (ret) {
> +        g_error("pthread_join() failed: %s", strerror(ret));
> +    }
> +
> +    pthread_mutex_destroy(&worker->input_mutex);
> +    pthread_mutex_destroy(&worker->output_mutex);
> +    pthread_cond_destroy(&worker->input_avail_cond);
> +    pthread_cond_destroy(&worker->input_avail_cond);
> +}
> +
> +GAWorker *ga_worker_new(ga_worker_func func)
> +{
> +    GAWorker *worker = g_malloc0(sizeof(GAWorker));
> +
> +    g_assert(func);
> +    worker->execute = func;
> +    ga_worker_start(worker);
> +
> +    return worker;
> +}
> +
> +void ga_worker_cleanup(GAWorker *worker)
> +{
> +    ga_worker_stop(worker);
> +    g_free(worker);
> +}