Re: [Qemu-devel] [PATCH 3/3] Add helper functions to enable virtio-9p make use of the threadlets

["Venkateswararao Jujjuri (JV)" <jvrao@linux.vnet.ibm.com>]

On 11/9/2010 1:06 AM, Stefan Hajnoczi wrote:
> On Mon, Nov 8, 2010 at 10:47 AM, Arun R Bharadwaj
> <arun@linux.vnet.ibm.com> wrote:
>> diff --git a/hw/virtio-9p.c b/hw/virtio-9p.c
>> index 7c59988..61b65fd 100644
>> --- a/hw/virtio-9p.c
>> +++ b/hw/virtio-9p.c
>> @@ -18,6 +18,7 @@
>>  #include "fsdev/qemu-fsdev.h"
>>  #include "virtio-9p-debug.h"
>>  #include "virtio-9p-xattr.h"
>> +#include "qemu-threadlets.h"
>>
>>  int debug_9p_pdu;
>>
>> @@ -33,6 +34,146 @@ enum {
>>     Oappend = 0x80,
>>  };
>>
>> +struct v9fs_post_op {
>> +    QTAILQ_ENTRY(v9fs_post_op) node;
>> +    void (*func)(void *arg);
>> +    void *arg;
>> +};
> 
> Structs are usually typedefed:
> typedef struct V9fsPostOp {
>     ...
> } V9fsPostOp;
> 
> QEMU usually doesn't use struct explicitly unless a library interface
> requires it.
> 
>> +
>> +static struct {
>> +    int rfd;
>> +    int wfd;
>> +    QemuMutex lock;
>> +    QTAILQ_HEAD(, v9fs_post_op) post_op_list;
>> +} v9fs_async_struct;
>> +
>> +static void die2(int err, const char *what)
>> +{
>> +    fprintf(stderr, "%s failed: %s\n", what, strerror(err));
>> +    abort();
>> +}
>> +
>> +static void die(const char *what)
>> +{
>> +    die2(errno, what);
>> +}
>> +
>> +#define ASYNC_MAX_PROCESS   5
>> +
>> +/**
>> + * v9fs_process_post_ops: Process any pending v9fs_post_posix_operation
>> + * @arg: Not used.
>> + *
>> + * This function serves as a callback to the iothread to be called into whenever
>> + * the v9fs_async_struct.wfd is written into. This thread goes through the list
>> + * of v9fs_post_posix_operations() and executes them. In the process, it might
>> + * queue more job on the asynchronous thread pool.
>> + */
>> +static void v9fs_process_post_ops(void *arg)
>> +{
>> +    int count = 0;
>> +    struct v9fs_post_op *post_op;
>> +    int ret;
>> +    char byte;
>> +
>> +    qemu_mutex_lock(&v9fs_async_struct.lock);
>> +    do {
>> +        ret = read(v9fs_async_struct.rfd, &byte, sizeof(byte));
>> +    } while (ret >= 0 && errno != EAGAIN);
> 
> } while (ret > 1 || (ret == -1 && errno == EINTR)); ?

Not sure why we need to retry if ret > 1; all we need is

} while (ret == -1 && errno == EINTR);

> 
>> +
>> +    for (count = 0; count < ASYNC_MAX_PROCESS; count++) {
>> +        if (QTAILQ_EMPTY(&(v9fs_async_struct.post_op_list))) {
>> +           break;
>> +        }
>> +        post_op = QTAILQ_FIRST(&(v9fs_async_struct.post_op_list));
>> +        QTAILQ_REMOVE(&(v9fs_async_struct.post_op_list), post_op, node);
>> +
>> +        qemu_mutex_unlock(&v9fs_async_struct.lock);
>> +        post_op->func(post_op->arg);
>> +        qemu_free(post_op);
>> +        qemu_mutex_lock(&v9fs_async_struct.lock);
>> +    }
>> +    qemu_mutex_unlock(&v9fs_async_struct.lock);
>> +}
>> +
>> +/**
>> + * v9fs_async_signal: Inform the io-thread of completion of async job.
>> + *
>> + * This function is used to inform the iothread that a particular
>> + * async-operation pertaining to v9fs has been completed and that the io thread
>> + * can handle the v9fs_post_posix_operation.
>> + *
>> + * This is based on the aio_signal_handler
>> + */
>> +static inline void v9fs_async_signal(void)
>> +{
>> +    char byte = 0;
>> +    ssize_t ret;
>> +    int tries = 0;
>> +
>> +    qemu_mutex_lock(&v9fs_async_struct.lock);
>> +    do {
>> +        assert(tries != 100);
>> +       ret = write(v9fs_async_struct.wfd, &byte, sizeof(byte));
>> +        tries++;
>> +    } while (ret < 0 && errno == EAGAIN);
>> +    qemu_mutex_unlock(&v9fs_async_struct.lock);
>> +
>> +    if (ret < 0 && errno != EAGAIN)
>> +        die("write() in v9fs");
> 
> { } coding style
> 
>> +
>> +    if (kill(getpid(), SIGUSR2)) die("kill failed");
> 
> { } coding style
> 
>> +}
>> +
>> +/**
>> + * v9fs_async_helper_done: Marks the completion of the v9fs_async job
>> + * @func: v9fs_post_posix_func() for post-processing invoked in the context of
>> + *        the io-thread
>> + * @arg: Argument to func.
>> + *
>> + * This function is called from the context of one of the asynchronous threads
>> + * in the thread pool. This is called when the asynchronous thread has finished
>> + * executing a v9fs_posix_operation. It's purpose is to initiate the process of
>> + * informing the io-thread that the v9fs_posix_operation has completed.
>> + */
>> +static void v9fs_async_helper_done(void (*func)(void *arg), void *arg)
>> +{
>> +    struct v9fs_post_op *post_op;
>> +
>> +    post_op = qemu_mallocz(sizeof(*post_op));
>> +    post_op->func = func;
>> +    post_op->arg = arg;
>> +
>> +    qemu_mutex_lock(&v9fs_async_struct.lock);
>> +    QTAILQ_INSERT_TAIL(&(v9fs_async_struct.post_op_list), post_op, node);
>> +    qemu_mutex_unlock(&v9fs_async_struct.lock);
>> +
>> +    v9fs_async_signal();
>> +}
>> +
>> +/**
>> + * v9fs_do_async_posix: Offload v9fs_posix_operation onto async thread.
>> + * @vs: V9fsOPState variable for the OP operation.
>> + * @posix_fn: The posix function which has to be offloaded onto async thread.
>> + * @post_fn_ptr: Address of the location to hold the post_fn corresponding to
>> + * the posix_fn
>> + * @post_fn: The post processing function corresponding to the posix_fn.
>> + *
>> + * This function is a helper to offload posix_operation on to the asynchronous
>> + * thread pool. It sets up the associations with the post_function that needs to
>> + * be invoked by from the context of the iothread once the posix_fn has been
>> + * executed.
>> + */
>> +static void v9fs_do_async_posix(ThreadletWork *work ,
>> +                                void (*posix_fn)(ThreadletWork *work),
>> +                                void (**post_fn_ptr)(void *arg),
>> +                                void (*post_fn)(void *arg))
>> +{
>> +    *post_fn_ptr = post_fn;
>> +    work->func = posix_fn;
>> +    submit_work(work);
>> +}
>> +
>>  static int omode_to_uflags(int8_t mode)
>>  {
>>     int ret = 0;
>> @@ -3657,7 +3798,7 @@ VirtIODevice *virtio_9p_init(DeviceState *dev, V9fsConf *conf)
>>     int i, len;
>>     struct stat stat;
>>     FsTypeEntry *fse;
>> -
>> +    int fds[2];
>>
>>     s = (V9fsState *)virtio_common_init("virtio-9p",
>>                                     VIRTIO_ID_9P,
>> @@ -3740,5 +3881,27 @@ VirtIODevice *virtio_9p_init(DeviceState *dev, V9fsConf *conf)
>>                         s->tag_len;
>>     s->vdev.get_config = virtio_9p_get_config;
>>
>> +    if (qemu_pipe(fds) == -1) {
>> +        fprintf(stderr, "failed to create fd's for virtio-9p\n");
>> +        exit(1);
>> +    }
>> +
>> +    v9fs_async_struct.rfd = fds[0];
>> +    v9fs_async_struct.wfd = fds[1];
>> +
>> +    printf("v9fs: rfd: %d\n", v9fs_async_struct.rfd);
>> +    printf("v9fs: wfd: %d\n", v9fs_async_struct.wfd);
>> +
>> +    fcntl(fds[0], F_SETFL, O_NONBLOCK);
>> +    fcntl(fds[1], F_SETFL, O_NONBLOCK);
>> +
>> +    qemu_set_fd_handler(fds[0], v9fs_process_post_ops, NULL, NULL);
>> +    QTAILQ_INIT(&v9fs_async_struct.post_op_list);
>> +    qemu_mutex_init(&(v9fs_async_struct.lock));
>> +    /* Create async queue. */
> 
> Please delete this comment, it seems outdated.
> 
>> +
>> +    (void)v9fs_do_async_posix;
>> +    (void)v9fs_async_helper_done;
>> +
>>     return &s->vdev;
>>  }
>> diff --git a/posix-aio-compat.c b/posix-aio-compat.c
>> index 9e02d18..8019be7 100644
>> --- a/posix-aio-compat.c
>> +++ b/posix-aio-compat.c
>> @@ -260,6 +260,8 @@ static ssize_t handle_aiocb_rw(struct qemu_paiocb *aiocb)
>>     return nbytes;
>>  }
>>
>> +static PosixAioState */;
>> +
>>  static void aio_thread(ThreadletWork *work)
>>  {
>>     pid_t pid;
>> @@ -290,6 +292,15 @@ static void aio_thread(ThreadletWork *work)
>>     aiocb->ret = ret;
>>     qemu_mutex_unlock(&aiocb_mutex);
>>
>> +    if (posix_aio_state) {
> 
> This test is not needed because posix_aio_state must be non-NULL here.

I see similar check in the old code also..

- JV

> 
>> +        char byte = 0;
>> +        ssize_t ret;
>> +
>> +        ret = write(posix_aio_state->wfd, &byte, sizeof(byte));
>> +        if (ret < 0 && errno != EAGAIN)
> 
> Please always use if () { ... } curly braces.  QEMU coding style.
> 
> Stefan
>