Re: [Qemu-devel] [PATCH 2/3] Make paio subsystem use threadlets

[Arun R Bharadwaj <arun@linux.vnet.ibm.com>]

* Stefan Hajnoczi <stefanha@gmail.com> [2010-10-20 10:30:38]:

> On Tue, Oct 19, 2010 at 6:43 PM, Arun R Bharadwaj
> <arun@linux.vnet.ibm.com> wrote:
> > From: Gautham R Shenoy <ego@in.ibm.com>
> >
> > This patch makes the paio subsystem use the threadlet framework thereby
> > decoupling asynchronous threading framework portion out of
> > posix-aio-compat.c
> >
> > The patch has been tested with fstress.
> >
> > Signed-off-by: Gautham R Shenoy <ego@in.ibm.com>
> > Signed-off-by: Sripathi Kodi <sripathik@in.ibm.com>
> > ---
> >  posix-aio-compat.c |  166 +++++++++-------------------------------------------
> >  1 files changed, 30 insertions(+), 136 deletions(-)
> >
> > diff --git a/posix-aio-compat.c b/posix-aio-compat.c
> > index 7b862b5..6977c18 100644
> > --- a/posix-aio-compat.c
> > +++ b/posix-aio-compat.c
> > @@ -29,6 +29,7 @@
> >  #include "block_int.h"
> >
> >  #include "block/raw-posix-aio.h"
> > +#include "qemu-threadlets.h"
> >
> >
> >  struct qemu_paiocb {
> > @@ -51,6 +52,7 @@ struct qemu_paiocb {
> >     struct qemu_paiocb *next;
> >
> >     int async_context_id;
> > +    ThreadletWork work;
> 
> The QTAILQ_ENTRY(qemu_paiocb) node field is no longer used, please remove it.
> 
> >  };
> >
> >  typedef struct PosixAioState {
> > @@ -59,15 +61,6 @@ typedef struct PosixAioState {
> >  } PosixAioState;
> >
> >
> > -static pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
> > -static pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
> > -static pthread_t thread_id;
> > -static pthread_attr_t attr;
> > -static int max_threads = 64;
> > -static int cur_threads = 0;
> > -static int idle_threads = 0;
> > -static QTAILQ_HEAD(, qemu_paiocb) request_list;
> > -
> >  #ifdef CONFIG_PREADV
> >  static int preadv_present = 1;
> >  #else
> > @@ -85,39 +78,6 @@ static void die(const char *what)
> >     die2(errno, what);
> >  }
> >
> > -static void mutex_lock(pthread_mutex_t *mutex)
> > -{
> > -    int ret = pthread_mutex_lock(mutex);
> > -    if (ret) die2(ret, "pthread_mutex_lock");
> > -}
> > -
> > -static void mutex_unlock(pthread_mutex_t *mutex)
> > -{
> > -    int ret = pthread_mutex_unlock(mutex);
> > -    if (ret) die2(ret, "pthread_mutex_unlock");
> > -}
> > -
> > -static int cond_timedwait(pthread_cond_t *cond, pthread_mutex_t *mutex,
> > -                           struct timespec *ts)
> > -{
> > -    int ret = pthread_cond_timedwait(cond, mutex, ts);
> > -    if (ret && ret != ETIMEDOUT) die2(ret, "pthread_cond_timedwait");
> > -    return ret;
> > -}
> > -
> > -static void cond_signal(pthread_cond_t *cond)
> > -{
> > -    int ret = pthread_cond_signal(cond);
> > -    if (ret) die2(ret, "pthread_cond_signal");
> > -}
> > -
> > -static void thread_create(pthread_t *thread, pthread_attr_t *attr,
> > -                          void *(*start_routine)(void*), void *arg)
> > -{
> > -    int ret = pthread_create(thread, attr, start_routine, arg);
> > -    if (ret) die2(ret, "pthread_create");
> > -}
> > -
> >  static ssize_t handle_aiocb_ioctl(struct qemu_paiocb *aiocb)
> >  {
> >     int ret;
> > @@ -301,106 +261,51 @@ static ssize_t handle_aiocb_rw(struct qemu_paiocb *aiocb)
> >     return nbytes;
> >  }
> >
> > -static void *aio_thread(void *unused)
> > +static void aio_thread(ThreadletWork *work)
> >  {
> >     pid_t pid;
> > +    struct qemu_paiocb *aiocb = container_of(work, struct qemu_paiocb, work);
> > +    ssize_t ret = 0;
> >
> >     pid = getpid();
> > +    aiocb->active = 1;
> >
> > -    while (1) {
> > -        struct qemu_paiocb *aiocb;
> > -        ssize_t ret = 0;
> > -        qemu_timeval tv;
> > -        struct timespec ts;
> > -
> > -        qemu_gettimeofday(&tv);
> > -        ts.tv_sec = tv.tv_sec + 10;
> > -        ts.tv_nsec = 0;
> > -
> > -        mutex_lock(&lock);
> > -
> > -        while (QTAILQ_EMPTY(&request_list) &&
> > -               !(ret == ETIMEDOUT)) {
> > -            ret = cond_timedwait(&cond, &lock, &ts);
> > -        }
> > -
> > -        if (QTAILQ_EMPTY(&request_list))
> > -            break;
> > -
> > -        aiocb = QTAILQ_FIRST(&request_list);
> > -        QTAILQ_REMOVE(&request_list, aiocb, node);
> > -        aiocb->active = 1;
> > -        idle_threads--;
> > -        mutex_unlock(&lock);
> > -
> > -        switch (aiocb->aio_type & QEMU_AIO_TYPE_MASK) {
> > -        case QEMU_AIO_READ:
> > -        case QEMU_AIO_WRITE:
> > -            ret = handle_aiocb_rw(aiocb);
> > -            break;
> > -        case QEMU_AIO_FLUSH:
> > -            ret = handle_aiocb_flush(aiocb);
> > -            break;
> > -        case QEMU_AIO_IOCTL:
> > -            ret = handle_aiocb_ioctl(aiocb);
> > -            break;
> > -        default:
> > -            fprintf(stderr, "invalid aio request (0x%x)\n", aiocb->aio_type);
> > -            ret = -EINVAL;
> > -            break;
> > -        }
> > -
> > -        mutex_lock(&lock);
> > -        aiocb->ret = ret;
> > -        idle_threads++;
> > -        mutex_unlock(&lock);
> > -
> > -        if (kill(pid, aiocb->ev_signo)) die("kill failed");
> > +    switch (aiocb->aio_type & QEMU_AIO_TYPE_MASK) {
> > +    case QEMU_AIO_READ:
> > +    case QEMU_AIO_WRITE:
> > +        ret = handle_aiocb_rw(aiocb);
> > +        break;
> > +    case QEMU_AIO_FLUSH:
> > +        ret = handle_aiocb_flush(aiocb);
> > +        break;
> > +    case QEMU_AIO_IOCTL:
> > +        ret = handle_aiocb_ioctl(aiocb);
> > +        break;
> > +    default:
> > +        fprintf(stderr, "invalid aio request (0x%x)\n", aiocb->aio_type);
> > +        ret = -EINVAL;
> > +        break;
> >     }
> >
> > -    idle_threads--;
> > -    cur_threads--;
> > -    mutex_unlock(&lock);
> > +    aiocb->ret = ret;
> >
> > -    return NULL;
> > -}
> > -
> > -static void spawn_thread(void)
> > -{
> > -    sigset_t set, oldset;
> > -
> > -    cur_threads++;
> > -    idle_threads++;
> > -
> > -    /* block all signals */
> > -    if (sigfillset(&set)) die("sigfillset");
> > -    if (sigprocmask(SIG_SETMASK, &set, &oldset)) die("sigprocmask");
> > -
> > -    thread_create(&thread_id, &attr, aio_thread, NULL);
> > -
> > -    if (sigprocmask(SIG_SETMASK, &oldset, NULL)) die("sigprocmask restore");
> > +    if (kill(pid, aiocb->ev_signo)) die("kill failed");
> >  }
> >
> >  static void qemu_paio_submit(struct qemu_paiocb *aiocb)
> >  {
> >     aiocb->ret = -EINPROGRESS;
> >     aiocb->active = 0;
> > -    mutex_lock(&lock);
> > -    if (idle_threads == 0 && cur_threads < max_threads)
> > -        spawn_thread();
> > -    QTAILQ_INSERT_TAIL(&request_list, aiocb, node);
> > -    mutex_unlock(&lock);
> > -    cond_signal(&cond);
> > +
> > +    aiocb->work.func = aio_thread;
> > +    submit_threadletwork(&aiocb->work);
> >  }
> >
> >  static ssize_t qemu_paio_return(struct qemu_paiocb *aiocb)
> >  {
> >     ssize_t ret;
> >
> > -    mutex_lock(&lock);
> >     ret = aiocb->ret;
> > -    mutex_unlock(&lock);
> > -
> >     return ret;
> >  }
> >
> > @@ -536,14 +441,14 @@ static void paio_cancel(BlockDriverAIOCB *blockacb)
> >     struct qemu_paiocb *acb = (struct qemu_paiocb *)blockacb;
> >     int active = 0;
> >
> > -    mutex_lock(&lock);
> >     if (!acb->active) {
> 
> I'm not sure the active field serves any purpose.  No memory barriers
> are used so the value of active is 0 before the work is executed and 0
> *or* 1 while the work is executed.
> 
> The cancel_threadletwork() function already indicates whether
> cancellation succeeded.  Why not just try to cancel instead of using
> the active field?
> 

This series does not touch the active field anywhere. So I feel we can
implement this as a separate patch instead of clubbing it with this.

-arun
> > -        QTAILQ_REMOVE(&request_list, acb, node);
> > -        acb->ret = -ECANCELED;
> > +        if (!cancel_threadletwork(&acb->work))
> > +            acb->ret = -ECANCELED;
> > +         else
> > +            active = 1;
> 
> The 0 and 1 return value from cancel_threadletwork() is inverted.  See
> also my comment on patch 1/3 in this series.
> 
> >     } else if (acb->ret == -EINPROGRESS) {
> >         active = 1;
> >     }
> > -    mutex_unlock(&lock);
> >
> >     if (active) {
> >         /* fail safe: if the aio could not be canceled, we wait for
> 
> while (qemu_paio_error(acb) == EINPROGRESS)
>     ;
> 
> Tight loop with no memory barrier reading a memory location that is
> updated by another thread.  We shouldn't communicate between threads
> without barriers.
> 
> Stefan
>