Re: [Qemu-devel] [0/25] Async threading for VirtFS using glib threads & coroutines.

[Anthony Liguori <anthony@codemonkey.ws>]

On 05/13/2011 03:55 AM, Stefan Hajnoczi wrote:
> On Thu, May 12, 2011 at 01:57:22PM -0700, Venkateswararao Jujjuri (JV) wrote:
>> VirtFS (fileserver base on 9P) performs many blocking system calls in the
>> vCPU context. This effort is to move the blocking calls out of vCPU/IO
>> thread context, into asynchronous threads.
>>
>> Anthony's " Add hard build dependency on glib" patch and
>> Kevin/Stefan's coroutine effort is a prerequisite.
>>
>> This patch set contains:
>>   - Converting all 9pfs calls into coroutines.
>>   - Each 9P operation will be modified for:
>>      - Remove post* functions. These are our call back functions which makes
>>        the code very hard to read. Now with coroutines, we can achieve the same
>>        state machine model with nice sequential code flow.
>>      - Move errno access near to the local_syscall()
>>      - Introduce asynchronous threading
>>
>> This series has the basic infrastructure and few routines like
>> mkdir,monod,unlink,readdir,xattr,lstat, etc converted.
>> Currently we are working on converting and testing other 9P operations also
>> into this model and those patches will follow shortly.
>>
>> Removing callback functions made some of the patches little lengthy.
>
> This long patch series adds temporary structs and marshalling code for
> each file system operation - I think none of this is necessary.  Instead
> we can exploit coroutines more:
>
> The point of coroutines is that you can suspend a thread of control (a
> call-stack, not an OS-level thread) and can re-enter it later.  We
> should make coroutines thread-safe (i.e. work outside of the global
> mutex) and then allow switching a coroutine from a QEMU thread to a
> worker thread and back again:
>
> int coroutine_fn v9fs_co_readdir(V9fsState *s, V9fsFidState *fidp,
>                                   struct dirent **dent)
> {
>      int ret = 0;
>
>      v9fs_co_run_in_worker({
>          errno = 0;
>          *dent = s->ops->readdir(&s->ctx, fidp->fs.dir);
>          if (!*dent&&  errno) {
>              ret = -errno;
>          }
>      });
>      return ret;
> }
>
> v9fs_co_readdir() can be called from a QEMU thread.  The block of code
> inside v9fs_co_run_in_worker() will be executed in a worker thread.

I'd prefer if we did:

threadlet_run({
    ret = readdir_r(dirp, dent);
    if (ret == -1) {
       ret = -errno;
    }
});

And not worry at all about re-entrancy.

> Notice that no marshalling variables is necessary at all; we can use the
> function arguments and local variables because this is still the same
> function!
>
> When control reaches the end of the v9fs_co_run_in_worker() block,
> execution is resumed in a QEMU thread and the function then returns ret.
> It would be incorrect to return inside the v9fs_co_run_in_worker() block
> because at that point we're still inside the worker thread.
>
> Here is how v9fs_co_run_in_worker() does its magic:
>
> #define v9fs_co_run_in_worker(block) \
> { \
>      BH *co_bh; \
> \
>      co_bh = qemu_bh_new(co_run_in_worker_bh, qemu_coroutine_self()); \
>      qemu_bh_schedule(co_bh); \
>      qemu_coroutine_yield(); /* re-entered in worker thread */ \
>      qemu_bh_delete(co_bh); \
> \
>      block; \
> \
>      qemu_coroutine_yield(); /* re-entered in QEMU thread */ \
> }
>
> void co_run_in_worker_bh(void *opaque)
> {
>      Coroutine *co = opaque;
>
>      g_thread_pool_push(pool, co, NULL);
> }
>
> void worker_thread_fn(gpointer data, gpointer user_data)
> {
>      Coroutine *co = user_data;
>      char byte = 0;
>      ssize_t len;
>
>      qemu_coroutine_enter(co, NULL);
>
>      g_async_queue_push(v9fs_pool.completed, co);
>      do {
>          len = write(v9fs_pool.wfd,&byte, sizeof(byte));
>      } while (len == -1&&  errno == EINTR);
> }
>
> void process_req_done(void *arg)
> {
>      Coroutine *co;
>      char byte;
>      ssize_t len;
>
>      do {
>          len = read(v9fs_pool.rfd,&byte, sizeof(byte));
>      } while (len == -1&&  errno == EINTR);
>
>      while ((co = g_async_queue_try_pop(v9fs_pool.completed)) != NULL) {
>          qemu_coroutine_enter(co, NULL);
>      }
> }
>
> I typed this code out in the email, it has not been compiled or tested.

I did this myself a while ago.  The concept is definitely sound.  I 
think it's the way to go.

I also think it's a candidate for the block layer too.

Regards,

Anthony Liguori