On Thu, Oct 21, 2010 at 9:40 AM, Arun R Bharadwaj
<a...@linux.vnet.ibm.com> wrote:
> * Stefan Hajnoczi <stefa...@gmail.com> [2010-10-20 10:30:38]:
>
>> On Tue, Oct 19, 2010 at 6:43 PM, Arun R Bharadwaj
>> <a...@linux.vnet.ibm.com> wrote:
>> > From: Gautham R Shenoy <e...@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 <e...@in.ibm.com>
>> > Signed-off-by: Sripathi Kodi <sripat...@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.
I'd prefer for this to be addressed in this patch because the active
field served a function before but no longer works with threadlets.
You're right that the patch doesn't touch it, and QEMU still compiles
fine with it, but it's still broken as I described in the previous
email.
In other words, the patch breaks the active field, please fix it.
Stefan
