[Qemu-devel] [PATCH V3 0/3] qemu: Make AIO threading framework generic.

[Qemu-devel] [PATCH V3 0/3] qemu: Make AIO threading framework generic.

[Gautham R Shenoy]

Hi,

This is the v3 of the patch-series to have a generic asynchronous task
offloading framework within qemu.

V2 can be found here:
http://lists.gnu.org/archive/html/qemu-devel/2010-05/msg02227.html

Changes from V2:
=====================================================================
- Made use of the qemu-thread.c wrappers in place of pthread_() calls.
- Added a couple fo qemu-thread wrappers for handling pthread_attr_t type.
- Audited the error handling in the generic asynchronous task offloading
  framework code.

Description
=====================================================================
This patch series decouples the asynchrnous threading framework
implementation from posix-aio-compat.c to implement a generic asynchrnous
task offloading threading framework which can be used by other subsystems
within QEMU.

Currently within QEMU, the AIO subsystem (paio) creates a bunch of
asynchronous threads to offload any blocking operations so that
the vcpu threads and the IO thread can go back to servicing any
other guest requests.

This offloading framework can be used by subsystems such as virtio-9p,
Asynchronous encoding for vnc-server, so that the vcpu thread can offload
blocking operations on to the asynchronous threads and resume servicing
any other guest requests. The asynchronous threads, after
finishing the blocking operations can then transfer the control over
to the IO thread so that the latter can handle the post_blocking_operation().

The patch series passed fsstress test without any issues.

Could it be considered for inclusion ?

---

Aneesh Kumar K.V (1):
      qemu: Generic asynchronous threading framework to offload tasks

Gautham R Shenoy (2):
      qemu: Add qemu-wrappers for pthread_attr_t
      qemu: Convert AIO code to use the generic threading infrastructure.


 Makefile.objs      |    3 +
 async-work.c       |  136 ++++++++++++++++++++++++++++++++++++++++++++++
 async-work.h       |   85 +++++++++++++++++++++++++++++
 posix-aio-compat.c |  155 ++++++++++------------------------------------------
 qemu-thread.c      |   34 +++++++++++
 qemu-thread.h      |   11 ++++
 6 files changed, 297 insertions(+), 127 deletions(-)
 create mode 100644 async-work.c
 create mode 100644 async-work.h

-- 
Thanks and Regards
gautham.

[Qemu-devel] [PATCH V3 1/3] qemu: Add qemu-wrappers for pthread_attr_t

[Gautham R Shenoy]

Add qemu wrappers for pthread_attr_t handling.


Signed-off-by: Gautham R Shenoy <ego@in.ibm.com>
---
 qemu-thread.c |   34 ++++++++++++++++++++++++++++++++++
 qemu-thread.h |   11 +++++++++++
 2 files changed, 45 insertions(+), 0 deletions(-)

diff --git a/qemu-thread.c b/qemu-thread.c
index 3923db7..524860c 100644
--- a/qemu-thread.c
+++ b/qemu-thread.c
@@ -142,6 +142,40 @@ void qemu_thread_create(QemuThread *thread,
         error_exit(err, __func__);
 }
 
+void qemu_thread_create_attr(QemuThread *thread, QemuThreadAttr *attr1,
+                       void *(*start_routine)(void*),
+                       void *arg)
+{
+    int err;
+
+    err = pthread_create(&thread->thread, &(attr1->attr), start_routine,
+                            arg);
+    if (err)
+        error_exit(err, __func__);
+}
+
+void qemu_thread_attr_init(QemuThreadAttr *attr1)
+{
+    int err;
+
+    err = pthread_attr_init(&(attr1->attr));
+
+    if (err) {
+        error_exit(err, __func__);
+    }
+}
+
+void qemu_thread_attr_setdetachstate(QemuThreadAttr *attr1, int detachstate)
+{
+    int err;
+
+    err = pthread_attr_setdetachstate(&(attr1->attr), detachstate);
+
+    if (err) {
+        error_exit(err, __func__);
+    }
+}
+
 void qemu_thread_signal(QemuThread *thread, int sig)
 {
     int err;
diff --git a/qemu-thread.h b/qemu-thread.h
index 5ef4a3a..361ef19 100644
--- a/qemu-thread.h
+++ b/qemu-thread.h
@@ -15,9 +15,14 @@ struct QemuThread {
     pthread_t thread;
 };
 
+struct QemuThreadAttr {
+    pthread_attr_t attr;
+};
+
 typedef struct QemuMutex QemuMutex;
 typedef struct QemuCond QemuCond;
 typedef struct QemuThread QemuThread;
+typedef struct QemuThreadAttr QemuThreadAttr;
 
 void qemu_mutex_init(QemuMutex *mutex);
 void qemu_mutex_lock(QemuMutex *mutex);
@@ -34,7 +39,13 @@ int qemu_cond_timedwait(QemuCond *cond, QemuMutex *mutex, uint64_t msecs);
 void qemu_thread_create(QemuThread *thread,
                        void *(*start_routine)(void*),
                        void *arg);
+void qemu_thread_create_attr(QemuThread *thread, QemuThreadAttr *attr1,
+                       void *(*start_routine)(void*),
+                       void *arg);
 void qemu_thread_signal(QemuThread *thread, int sig);
 void qemu_thread_self(QemuThread *thread);
 int qemu_thread_equal(QemuThread *thread1, QemuThread *thread2);
+
+void qemu_thread_attr_init(QemuThreadAttr *attr1);
+void qemu_thread_attr_setdetachstate(QemuThreadAttr *attr1, int detachstate);
 #endif

[Qemu-devel] Re: [PATCH V3 1/3] qemu: Add qemu-wrappers for pthread_attr_t

[Paolo Bonzini]

On 06/03/2010 10:56 AM, Gautham R Shenoy wrote:
> Add qemu wrappers for pthread_attr_t handling.

The point of these wrappers AFAIU is not only to add error_exit, but 
also to be portable to Windows in the future.  Is it necessary to create 
the threads as detached?  If you set queue->min_threads to zero all 
threads should exit as soon as they finish their work (which is better 
than exiting immediately).

Paolo

Re: [Qemu-devel] Re: [PATCH V3 1/3] qemu: Add qemu-wrappers for pthread_attr_t

[Anthony Liguori]

On 06/03/2010 07:31 AM, Paolo Bonzini wrote:
> On 06/03/2010 10:56 AM, Gautham R Shenoy wrote:
>> Add qemu wrappers for pthread_attr_t handling.
>
> The point of these wrappers AFAIU is not only to add error_exit, but 
> also to be portable to Windows in the future.  Is it necessary to 
> create the threads as detached?  If you set queue->min_threads to zero 
> all threads should exit as soon as they finish their work (which is 
> better than exiting immediately).

This is historical because the code was largely inspired by glibc's 
implementation of posix-aio.  It doesn't need to be detached and since 
Corentin wants to be able to join a worker, it makes sense to just avoid 
detaching and pay the overhead of making the threads joinable.

Regards,

Anthony Liguori

> Paolo
>

Re: [Qemu-devel] Re: [PATCH V3 1/3] qemu: Add qemu-wrappers for pthread_attr_t

[Corentin Chary]

On Fri, Jun 4, 2010 at 3:07 PM, Anthony Liguori <anthony@codemonkey.ws> wrote:
> On 06/03/2010 07:31 AM, Paolo Bonzini wrote:
>>
>> On 06/03/2010 10:56 AM, Gautham R Shenoy wrote:
>>>
>>> Add qemu wrappers for pthread_attr_t handling.
>>
>> The point of these wrappers AFAIU is not only to add error_exit, but also
>> to be portable to Windows in the future.  Is it necessary to create the
>> threads as detached?  If you set queue->min_threads to zero all threads
>> should exit as soon as they finish their work (which is better than exiting
>> immediately).
>
> This is historical because the code was largely inspired by glibc's
> implementation of posix-aio.  It doesn't need to be detached and since
> Corentin wants to be able to join a worker, it makes sense to just avoid
> detaching and pay the overhead of making the threads joinable.
>
> Regards,
>
> Anthony Liguori

Actually, I want to know if the queue is empty and if no job are
currently being processed: all worker are idle or stopped. I don't
really need pthread_join() for that, since worker can be idle (we
don't want to always start and stop the thread :) ).



-- 
Corentin Chary
http://xf.iksaif.net

Re: [Qemu-devel] Re: [PATCH V3 1/3] qemu: Add qemu-wrappers for pthread_attr_t

[Paolo Bonzini]

On 06/04/2010 03:19 PM, Corentin Chary wrote:
>>> The point of these wrappers AFAIU is not only to add error_exit, but also
>>> to be portable to Windows in the future.
>>
>> This is historical because the code was largely inspired by glibc's
>> implementation of posix-aio.  It doesn't need to be detached and since
>> Corentin wants to be able to join a worker, it makes sense to just avoid
>> detaching and pay the overhead of making the threads joinable.
>
> Actually, I want to know if the queue is empty and if no job are
> currently being processed: all worker are idle or stopped. I don't
> really need pthread_join() for that, since worker can be idle (we
> don't want to always start and stop the thread :) ).

Then it's also fine to have all qemu_threads detached (like in my patch 
to create all qemu_threads with blocked signals).  I just want to avoid 
implementing pthreads one day for qemu-threads-win32.c.

Paolo

Re: [Qemu-devel] Re: [PATCH V3 1/3] qemu: Add qemu-wrappers for pthread_attr_t

[Gautham R Shenoy]

On Fri, Jun 04, 2010 at 03:19:34PM +0200, Corentin Chary wrote:
> On Fri, Jun 4, 2010 at 3:07 PM, Anthony Liguori <anthony@codemonkey.ws> wrote:
> > On 06/03/2010 07:31 AM, Paolo Bonzini wrote:
> >>
> >> On 06/03/2010 10:56 AM, Gautham R Shenoy wrote:
> >>>
> >>> Add qemu wrappers for pthread_attr_t handling.
> >>
> >> The point of these wrappers AFAIU is not only to add error_exit, but also
> >> to be portable to Windows in the future.  Is it necessary to create the
> >> threads as detached?  If you set queue->min_threads to zero all threads
> >> should exit as soon as they finish their work (which is better than exiting
> >> immediately).
> >
> > This is historical because the code was largely inspired by glibc's
> > implementation of posix-aio.  It doesn't need to be detached and since
> > Corentin wants to be able to join a worker, it makes sense to just avoid
> > detaching and pay the overhead of making the threads joinable.
> >
> > Regards,
> >
> > Anthony Liguori
>

Sorry for the late response. I had been away for the last few days.

> Actually, I want to know if the queue is empty and if no job are
> currently being processed: all worker are idle or stopped.

In other words, you just want to know if the already submitted jobs
have gotten over or not so that you could probably destroy the queue.

The subsystem using the queueing infrastructure is responsible for providing
the guarantee that no new job would be submitted while/after making a call
query the idleness of the queue.

If this understanding is correct, we might want to have a
qemu_async_barrier() implementation which waits for all the threads
finish processing the queued jobs, and destroy themselves.

>I don't really need pthread_join() for that, since worker can be idle (we
> don't want to always start and stop the thread :) ).

-- 
Thanks and Regards
gautham

[Qemu-devel] [PATCH V3 2/3] qemu: Generic asynchronous threading framework to offload tasks

[Gautham R Shenoy]

From: Aneesh Kumar K.V <aneesh.kumar@linux.vnet.ibm.com>

This patch creates a generic asynchronous-task-offloading infrastructure. It's
extracted out of the threading framework that is being used by paio.

The reason for extracting out this generic infrastructure of the
posix-aio-compat.c is so that other subsystems, such as virtio-9p could make use
of it for offloading tasks that could block.

[ego@in.ibm.com: work_item_pool, async_work_init, async_work_release,
async_cancel_work]

Signed-off-by: Aneesh Kumar K.V <aneesh.kumar@linux.vnet.ibm.com>
Signed-off-by: Gautham R Shenoy <ego@in.ibm.com>
---
 Makefile.objs |    3 +
 async-work.c  |  136 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++
 async-work.h  |   85 ++++++++++++++++++++++++++++++++++++
 3 files changed, 223 insertions(+), 1 deletions(-)
 create mode 100644 async-work.c
 create mode 100644 async-work.h

diff --git a/Makefile.objs b/Makefile.objs
index ecdd53e..fd5ea4d 100644
--- a/Makefile.objs
+++ b/Makefile.objs
@@ -9,6 +9,8 @@ qobject-obj-y += qerror.o
 
 block-obj-y = cutils.o cache-utils.o qemu-malloc.o qemu-option.o module.o
 block-obj-y += nbd.o block.o aio.o aes.o osdep.o qemu-config.o
+block-obj-y += qemu-thread.o
+block-obj-y += async-work.o
 block-obj-$(CONFIG_POSIX) += posix-aio-compat.o
 block-obj-$(CONFIG_LINUX_AIO) += linux-aio.o
 
@@ -108,7 +110,6 @@ common-obj-y += iov.o
 common-obj-$(CONFIG_VNC_TLS) += vnc-tls.o vnc-auth-vencrypt.o
 common-obj-$(CONFIG_VNC_SASL) += vnc-auth-sasl.o
 common-obj-$(CONFIG_COCOA) += cocoa.o
-common-obj-$(CONFIG_IOTHREAD) += qemu-thread.o
 common-obj-y += notify.o event_notifier.o
 common-obj-y += qemu-timer.o
 
diff --git a/async-work.c b/async-work.c
new file mode 100644
index 0000000..0675732
--- /dev/null
+++ b/async-work.c
@@ -0,0 +1,136 @@
+/*
+ * Async work support
+ *
+ * Copyright IBM, Corp. 2010
+ *
+ * Authors:
+ *  Aneesh Kumar K.V <aneesh.kumar@linux.vnet.ibm.com>
+ *
+ * This work is licensed under the terms of the GNU GPL, version 2.  See
+ * the COPYING file in the top-level directory.
+ *
+ */
+#include <stdio.h>
+#include <errno.h>
+#include <string.h>
+#include <stdlib.h>
+#include <signal.h>
+#include "async-work.h"
+#include "osdep.h"
+
+static void async_abort(int err, const char *what)
+{
+    fprintf(stderr, "%s failed: %s\n", what, strerror(err));
+    abort();
+}
+
+static void *async_worker_thread(void *data)
+{
+    struct async_queue *queue = data;
+
+    while (1) {
+        struct work_item *work;
+        int ret = 0;
+        qemu_mutex_lock(&(queue->lock));
+
+        while (QTAILQ_EMPTY(&(queue->request_list)) &&
+               (ret != ETIMEDOUT)) {
+            ret = qemu_cond_timedwait(&(queue->cond),
+					 &(queue->lock), 10*100000);
+        }
+
+        if (QTAILQ_EMPTY(&(queue->request_list)))
+            goto check_exit;
+
+        work = QTAILQ_FIRST(&(queue->request_list));
+        QTAILQ_REMOVE(&(queue->request_list), work, node);
+        queue->idle_threads--;
+        qemu_mutex_unlock(&(queue->lock));
+
+        /* execute the work function */
+        work->func(work);
+        async_work_release(queue, work);
+
+        qemu_mutex_lock(&(queue->lock));
+        queue->idle_threads++;
+
+check_exit:
+        if ((queue->idle_threads > 0) &&
+            (queue->cur_threads > queue->min_threads)) {
+            /* we retain minimum number of threads */
+            break;
+        }
+        qemu_mutex_unlock(&(queue->lock));
+    }
+
+    queue->idle_threads--;
+    queue->cur_threads--;
+    qemu_mutex_unlock(&(queue->lock));
+
+    return NULL;
+}
+
+static void spawn_async_thread(struct async_queue *queue)
+{
+    QemuThreadAttr attr;
+    QemuThread thread;
+    sigset_t set, oldset;
+
+    queue->cur_threads++;
+    queue->idle_threads++;
+
+    qemu_thread_attr_init(&attr);
+
+    /* create a detached thread so that we don't need to wait on it */
+    qemu_thread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
+
+    /* block all signals */
+    if (sigfillset(&set)) {
+        async_abort(errno, "sigfillset");
+    }
+
+    if (sigprocmask(SIG_SETMASK, &set, &oldset)) {
+        async_abort(errno, "sigprocmask");
+    }
+
+    qemu_thread_create_attr(&thread, &attr, async_worker_thread, queue);
+
+    if (sigprocmask(SIG_SETMASK, &oldset, NULL)) {
+        async_abort(errno, "sigprocmask restore");
+    }
+}
+
+void qemu_async_submit(struct async_queue *queue, struct work_item *work)
+{
+    qemu_mutex_lock(&(queue->lock));
+    if (queue->idle_threads == 0 && queue->cur_threads < queue->max_threads) {
+        spawn_async_thread(queue);
+    }
+    QTAILQ_INSERT_TAIL(&(queue->request_list), work, node);
+    qemu_mutex_unlock(&(queue->lock));
+    qemu_cond_signal(&(queue->cond));
+}
+
+int qemu_async_cancel_work(struct async_queue *queue, struct work_item *work)
+{
+    struct work_item *ret_work;
+    int found = 0;
+
+    qemu_mutex_lock(&(queue->lock));
+    QTAILQ_FOREACH(ret_work, &(queue->request_list), node) {
+        if (ret_work == work) {
+            QTAILQ_REMOVE(&(queue->request_list), ret_work, node);
+            found = 1;
+            break;
+        }
+    }
+    qemu_mutex_unlock(&(queue->lock));
+
+    if (found) {
+        async_work_release(queue, work);
+        return 0;
+    }
+
+    return 1;
+}
+
diff --git a/async-work.h b/async-work.h
new file mode 100644
index 0000000..8389f56
--- /dev/null
+++ b/async-work.h
@@ -0,0 +1,85 @@
+/*
+ * Async work support
+ *
+ * Copyright IBM, Corp. 2010
+ *
+ * Authors:
+ *  Aneesh Kumar K.V <aneesh.kumar@linux.vnet.ibm.com>
+ *
+ * This work is licensed under the terms of the GNU GPL, version 2.  See
+ * the COPYING file in the top-level directory.
+ *
+ */
+#ifndef QEMU_ASYNC_WORK_H
+#define QEMU_ASYNC_WORK_H
+
+#include "qemu-queue.h"
+#include "qemu-common.h"
+#include "qemu-thread.h"
+
+struct async_queue
+{
+    QemuMutex lock;
+    QemuCond cond;
+    int max_threads;
+    int min_threads;
+    int cur_threads;
+    int idle_threads;
+    QTAILQ_HEAD(, work_item) request_list;
+    QTAILQ_HEAD(, work_item) work_item_pool;
+};
+
+struct work_item
+{
+    QTAILQ_ENTRY(work_item) node;
+    void (*func)(struct work_item *work);
+    void *private;
+};
+
+static inline void async_queue_init(struct async_queue *queue,
+				    int max_threads, int min_threads)
+{
+    queue->cur_threads  = 0;
+    queue->idle_threads = 0;
+    queue->max_threads  = max_threads;
+    queue->min_threads  = min_threads;
+    QTAILQ_INIT(&(queue->request_list));
+    QTAILQ_INIT(&(queue->work_item_pool));
+    qemu_mutex_init(&(queue->lock));
+    qemu_cond_init(&(queue->cond));
+}
+
+static inline struct work_item *async_work_init(struct async_queue *queue,
+				   void (*func)(struct work_item *),
+				   void *data)
+{
+    struct work_item *work;
+    qemu_mutex_lock(&(queue->lock));
+    if (QTAILQ_EMPTY(&(queue->work_item_pool))) {
+        work = qemu_mallocz(sizeof(*work));
+    } else {
+        work = QTAILQ_FIRST(&(queue->work_item_pool));
+        QTAILQ_REMOVE(&(queue->work_item_pool), work, node);
+    }
+
+    work->func  = func;
+    work->private  = data;
+    qemu_mutex_unlock(&(queue->lock));
+
+    return work;
+}
+
+static inline void async_work_release(struct async_queue *queue,
+                                        struct work_item *work)
+{
+    qemu_mutex_lock(&(queue->lock));
+    QTAILQ_INSERT_TAIL(&(queue->work_item_pool), work, node);
+    qemu_mutex_unlock(&(queue->lock));
+}
+
+extern void qemu_async_submit(struct async_queue *queue,
+				  struct work_item *work);
+
+extern int qemu_async_cancel_work(struct async_queue *queue,
+                    struct work_item *work);
+#endif

[Qemu-devel] Re: [PATCH V3 2/3] qemu: Generic asynchronous threading framework to offload tasks

[Corentin Chary]

On Thu, Jun 3, 2010 at 10:56 AM, Gautham R Shenoy <ego@in.ibm.com> wrote:
> From: Aneesh Kumar K.V <aneesh.kumar@linux.vnet.ibm.com>
>
> This patch creates a generic asynchronous-task-offloading infrastructure. It's
> extracted out of the threading framework that is being used by paio.
>
> The reason for extracting out this generic infrastructure of the
> posix-aio-compat.c is so that other subsystems, such as virtio-9p could make use
> of it for offloading tasks that could block.
>
> [ego@in.ibm.com: work_item_pool, async_work_init, async_work_release,
> async_cancel_work]
>
> Signed-off-by: Aneesh Kumar K.V <aneesh.kumar@linux.vnet.ibm.com>
> Signed-off-by: Gautham R Shenoy <ego@in.ibm.com>
> ---
>  Makefile.objs |    3 +
>  async-work.c  |  136 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++
>  async-work.h  |   85 ++++++++++++++++++++++++++++++++++++
>  3 files changed, 223 insertions(+), 1 deletions(-)
>  create mode 100644 async-work.c
>  create mode 100644 async-work.h
>
> diff --git a/Makefile.objs b/Makefile.objs
> index ecdd53e..fd5ea4d 100644
> --- a/Makefile.objs
> +++ b/Makefile.objs
> @@ -9,6 +9,8 @@ qobject-obj-y += qerror.o
>
>  block-obj-y = cutils.o cache-utils.o qemu-malloc.o qemu-option.o module.o
>  block-obj-y += nbd.o block.o aio.o aes.o osdep.o qemu-config.o
> +block-obj-y += qemu-thread.o
> +block-obj-y += async-work.o
>  block-obj-$(CONFIG_POSIX) += posix-aio-compat.o
>  block-obj-$(CONFIG_LINUX_AIO) += linux-aio.o
>
> @@ -108,7 +110,6 @@ common-obj-y += iov.o
>  common-obj-$(CONFIG_VNC_TLS) += vnc-tls.o vnc-auth-vencrypt.o
>  common-obj-$(CONFIG_VNC_SASL) += vnc-auth-sasl.o
>  common-obj-$(CONFIG_COCOA) += cocoa.o
> -common-obj-$(CONFIG_IOTHREAD) += qemu-thread.o
>  common-obj-y += notify.o event_notifier.o
>  common-obj-y += qemu-timer.o
>
> diff --git a/async-work.c b/async-work.c
> new file mode 100644
> index 0000000..0675732
> --- /dev/null
> +++ b/async-work.c
> @@ -0,0 +1,136 @@
> +/*
> + * Async work support
> + *
> + * Copyright IBM, Corp. 2010
> + *
> + * Authors:
> + *  Aneesh Kumar K.V <aneesh.kumar@linux.vnet.ibm.com>
> + *
> + * This work is licensed under the terms of the GNU GPL, version 2.  See
> + * the COPYING file in the top-level directory.
> + *
> + */
> +#include <stdio.h>
> +#include <errno.h>
> +#include <string.h>
> +#include <stdlib.h>
> +#include <signal.h>
> +#include "async-work.h"
> +#include "osdep.h"
> +
> +static void async_abort(int err, const char *what)
> +{
> +    fprintf(stderr, "%s failed: %s\n", what, strerror(err));
> +    abort();
> +}
> +
> +static void *async_worker_thread(void *data)
> +{
> +    struct async_queue *queue = data;
> +
> +    while (1) {
> +        struct work_item *work;
> +        int ret = 0;
> +        qemu_mutex_lock(&(queue->lock));
> +
> +        while (QTAILQ_EMPTY(&(queue->request_list)) &&
> +               (ret != ETIMEDOUT)) {
> +            ret = qemu_cond_timedwait(&(queue->cond),
> +                                        &(queue->lock), 10*100000);
> +        }
> +
> +        if (QTAILQ_EMPTY(&(queue->request_list)))
> +            goto check_exit;
> +
> +        work = QTAILQ_FIRST(&(queue->request_list));
> +        QTAILQ_REMOVE(&(queue->request_list), work, node);
> +        queue->idle_threads--;
> +        qemu_mutex_unlock(&(queue->lock));
> +
> +        /* execute the work function */
> +        work->func(work);

Here the queue is empty, but there is a job running. In the VNC server
I need to be able to "join" jobs (before resize, deconnection, etc..).
What do you thing about adding something like qemu_async_join(queue,
work) (if work is null, join all job) ?

> +        async_work_release(queue, work);
> +
> +        qemu_mutex_lock(&(queue->lock));
> +        queue->idle_threads++;
> +
> +check_exit:
> +        if ((queue->idle_threads > 0) &&
> +            (queue->cur_threads > queue->min_threads)) {
> +            /* we retain minimum number of threads */
> +            break;
> +        }
> +        qemu_mutex_unlock(&(queue->lock));
> +    }
> +
> +    queue->idle_threads--;
> +    queue->cur_threads--;
> +    qemu_mutex_unlock(&(queue->lock));
> +
> +    return NULL;
> +}
> +
> +static void spawn_async_thread(struct async_queue *queue)
> +{
> +    QemuThreadAttr attr;
> +    QemuThread thread;
> +    sigset_t set, oldset;
> +
> +    queue->cur_threads++;
> +    queue->idle_threads++;
> +
> +    qemu_thread_attr_init(&attr);
> +
> +    /* create a detached thread so that we don't need to wait on it */
> +    qemu_thread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
> +
> +    /* block all signals */
> +    if (sigfillset(&set)) {
> +        async_abort(errno, "sigfillset");
> +    }
> +
> +    if (sigprocmask(SIG_SETMASK, &set, &oldset)) {
> +        async_abort(errno, "sigprocmask");
> +    }
> +
> +    qemu_thread_create_attr(&thread, &attr, async_worker_thread, queue);
> +
> +    if (sigprocmask(SIG_SETMASK, &oldset, NULL)) {
> +        async_abort(errno, "sigprocmask restore");
> +    }

Using PTHREAD_CREATE_DETACHED and signal stuff here doesn't looks
really portable. Can't we abstract that into qemu-thread (then, we
just need port qemu-thread to windows) ?

> +}
> +
> +void qemu_async_submit(struct async_queue *queue, struct work_item *work)
> +{
> +    qemu_mutex_lock(&(queue->lock));
> +    if (queue->idle_threads == 0 && queue->cur_threads < queue->max_threads) {
> +        spawn_async_thread(queue);
> +    }
> +    QTAILQ_INSERT_TAIL(&(queue->request_list), work, node);
> +    qemu_mutex_unlock(&(queue->lock));
> +    qemu_cond_signal(&(queue->cond));
> +}
> +
> +int qemu_async_cancel_work(struct async_queue *queue, struct work_item *work)
> +{
> +    struct work_item *ret_work;
> +    int found = 0;
> +
> +    qemu_mutex_lock(&(queue->lock));
> +    QTAILQ_FOREACH(ret_work, &(queue->request_list), node) {
> +        if (ret_work == work) {
> +            QTAILQ_REMOVE(&(queue->request_list), ret_work, node);
> +            found = 1;
> +            break;
> +        }
> +    }
> +    qemu_mutex_unlock(&(queue->lock));
> +
> +    if (found) {
> +        async_work_release(queue, work);
> +        return 0;
> +    }
> +
> +    return 1;
> +}
> +
> diff --git a/async-work.h b/async-work.h
> new file mode 100644
> index 0000000..8389f56
> --- /dev/null
> +++ b/async-work.h
> @@ -0,0 +1,85 @@
> +/*
> + * Async work support
> + *
> + * Copyright IBM, Corp. 2010
> + *
> + * Authors:
> + *  Aneesh Kumar K.V <aneesh.kumar@linux.vnet.ibm.com>
> + *
> + * This work is licensed under the terms of the GNU GPL, version 2.  See
> + * the COPYING file in the top-level directory.
> + *
> + */
> +#ifndef QEMU_ASYNC_WORK_H
> +#define QEMU_ASYNC_WORK_H
> +
> +#include "qemu-queue.h"
> +#include "qemu-common.h"
> +#include "qemu-thread.h"
> +
> +struct async_queue
> +{
> +    QemuMutex lock;
> +    QemuCond cond;
> +    int max_threads;
> +    int min_threads;
> +    int cur_threads;
> +    int idle_threads;
> +    QTAILQ_HEAD(, work_item) request_list;
> +    QTAILQ_HEAD(, work_item) work_item_pool;
> +};
> +
> +struct work_item
> +{
> +    QTAILQ_ENTRY(work_item) node;
> +    void (*func)(struct work_item *work);
> +    void *private;
> +};
> +
> +static inline void async_queue_init(struct async_queue *queue,
> +                                   int max_threads, int min_threads)
> +{
> +    queue->cur_threads  = 0;
> +    queue->idle_threads = 0;
> +    queue->max_threads  = max_threads;
> +    queue->min_threads  = min_threads;
> +    QTAILQ_INIT(&(queue->request_list));
> +    QTAILQ_INIT(&(queue->work_item_pool));
> +    qemu_mutex_init(&(queue->lock));
> +    qemu_cond_init(&(queue->cond));
> +}
> +
> +static inline struct work_item *async_work_init(struct async_queue *queue,
> +                                  void (*func)(struct work_item *),
> +                                  void *data)
> +{
> +    struct work_item *work;
> +    qemu_mutex_lock(&(queue->lock));
> +    if (QTAILQ_EMPTY(&(queue->work_item_pool))) {
> +        work = qemu_mallocz(sizeof(*work));
> +    } else {
> +        work = QTAILQ_FIRST(&(queue->work_item_pool));
> +        QTAILQ_REMOVE(&(queue->work_item_pool), work, node);
> +    }
> +
> +    work->func  = func;
> +    work->private  = data;
> +    qemu_mutex_unlock(&(queue->lock));
> +
> +    return work;
> +}
> +
> +static inline void async_work_release(struct async_queue *queue,
> +                                        struct work_item *work)
> +{
> +    qemu_mutex_lock(&(queue->lock));
> +    QTAILQ_INSERT_TAIL(&(queue->work_item_pool), work, node);
> +    qemu_mutex_unlock(&(queue->lock));
> +}
> +
> +extern void qemu_async_submit(struct async_queue *queue,
> +                                 struct work_item *work);
> +
> +extern int qemu_async_cancel_work(struct async_queue *queue,
> +                    struct work_item *work);
> +#endif
>
>



-- 
Corentin Chary
http://xf.iksaif.net

[Qemu-devel] Re: [PATCH V3 2/3] qemu: Generic asynchronous threading framework to offload tasks

[Paolo Bonzini]

On 06/03/2010 01:41 PM, Corentin Chary wrote:
>>  +    if (sigprocmask(SIG_SETMASK,&set,&oldset)) {
>>  +        async_abort(errno, "sigprocmask");
>>  +    }
>>  +
>>  +    qemu_thread_create_attr(&thread,&attr, async_worker_thread, queue);
>>  +
>>  +    if (sigprocmask(SIG_SETMASK,&oldset, NULL)) {
>>  +        async_abort(errno, "sigprocmask restore");
>>  +    }

I wonder if qemu_thread_create shouldn't block all signals by default. 
Then the cpu and iothreads can unblock whatever they want.

I'll send a patch shortly.

In any case, please use pthread_sigmask instead of sigprocmask.

Paolo

Re: [Qemu-devel] [PATCH V3 2/3] qemu: Generic asynchronous threading framework to offload tasks

[Anthony Liguori]

On 06/03/2010 03:56 AM, Gautham R Shenoy wrote:
> From: Aneesh Kumar K.V<aneesh.kumar@linux.vnet.ibm.com>
>
> This patch creates a generic asynchronous-task-offloading infrastructure. It's
> extracted out of the threading framework that is being used by paio.
>
> The reason for extracting out this generic infrastructure of the
> posix-aio-compat.c is so that other subsystems, such as virtio-9p could make use
> of it for offloading tasks that could block.
>
> [ego@in.ibm.com: work_item_pool, async_work_init, async_work_release,
> async_cancel_work]
>
> Signed-off-by: Aneesh Kumar K.V<aneesh.kumar@linux.vnet.ibm.com>
> Signed-off-by: Gautham R Shenoy<ego@in.ibm.com>
> ---
>   Makefile.objs |    3 +
>   async-work.c  |  136 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++
>   async-work.h  |   85 ++++++++++++++++++++++++++++++++++++
>   3 files changed, 223 insertions(+), 1 deletions(-)
>   create mode 100644 async-work.c
>   create mode 100644 async-work.h
>
> diff --git a/Makefile.objs b/Makefile.objs
> index ecdd53e..fd5ea4d 100644
> --- a/Makefile.objs
> +++ b/Makefile.objs
> @@ -9,6 +9,8 @@ qobject-obj-y += qerror.o
>
>   block-obj-y = cutils.o cache-utils.o qemu-malloc.o qemu-option.o module.o
>   block-obj-y += nbd.o block.o aio.o aes.o osdep.o qemu-config.o
> +block-obj-y += qemu-thread.o
> +block-obj-y += async-work.o
>   block-obj-$(CONFIG_POSIX) += posix-aio-compat.o
>   block-obj-$(CONFIG_LINUX_AIO) += linux-aio.o
>
> @@ -108,7 +110,6 @@ common-obj-y += iov.o
>   common-obj-$(CONFIG_VNC_TLS) += vnc-tls.o vnc-auth-vencrypt.o
>   common-obj-$(CONFIG_VNC_SASL) += vnc-auth-sasl.o
>   common-obj-$(CONFIG_COCOA) += cocoa.o
> -common-obj-$(CONFIG_IOTHREAD) += qemu-thread.o
>   common-obj-y += notify.o event_notifier.o
>   common-obj-y += qemu-timer.o
>
> diff --git a/async-work.c b/async-work.c
> new file mode 100644
> index 0000000..0675732
> --- /dev/null
> +++ b/async-work.c
> @@ -0,0 +1,136 @@
> +/*
> + * Async work support
> + *
> + * Copyright IBM, Corp. 2010
> + *
> + * Authors:
> + *  Aneesh Kumar K.V<aneesh.kumar@linux.vnet.ibm.com>
> + *
> + * This work is licensed under the terms of the GNU GPL, version 2.  See
> + * the COPYING file in the top-level directory.
>    

Please preserve the original copyright of the copied code.

> + */
> +#include<stdio.h>
> +#include<errno.h>
> +#include<string.h>
> +#include<stdlib.h>
> +#include<signal.h>
>    

qemu-common.h should have all of these.  Generally, you should avoid 
including system headers because qemu headers take care of portability.

> +#include "async-work.h"
> +#include "osdep.h"
> +
> +static void async_abort(int err, const char *what)
> +{
> +    fprintf(stderr, "%s failed: %s\n", what, strerror(err));
> +    abort();
> +}
> +
> +static void *async_worker_thread(void *data)
> +{
> +    struct async_queue *queue = data;
> +
> +    while (1) {
> +        struct work_item *work;
> +        int ret = 0;
> +        qemu_mutex_lock(&(queue->lock));
> +
> +        while (QTAILQ_EMPTY(&(queue->request_list))&&
> +               (ret != ETIMEDOUT)) {
> +            ret = qemu_cond_timedwait(&(queue->cond),
> +					&(queue->lock), 10*100000);
> +        }
> +
> +        if (QTAILQ_EMPTY(&(queue->request_list)))
> +            goto check_exit;
> +
> +        work = QTAILQ_FIRST(&(queue->request_list));
> +        QTAILQ_REMOVE(&(queue->request_list), work, node);
> +        queue->idle_threads--;
> +        qemu_mutex_unlock(&(queue->lock));
> +
> +        /* execute the work function */
> +        work->func(work);
> +        async_work_release(queue, work);
> +
> +        qemu_mutex_lock(&(queue->lock));
> +        queue->idle_threads++;
> +
> +check_exit:
> +        if ((queue->idle_threads>  0)&&
> +            (queue->cur_threads>  queue->min_threads)) {
> +            /* we retain minimum number of threads */
> +            break;
> +        }
> +        qemu_mutex_unlock(&(queue->lock));
> +    }
> +
> +    queue->idle_threads--;
> +    queue->cur_threads--;
> +    qemu_mutex_unlock(&(queue->lock));
> +
> +    return NULL;
> +}
> +
> +static void spawn_async_thread(struct async_queue *queue)
> +{
> +    QemuThreadAttr attr;
> +    QemuThread thread;
> +    sigset_t set, oldset;
> +
> +    queue->cur_threads++;
> +    queue->idle_threads++;
> +
> +    qemu_thread_attr_init(&attr);
> +
> +    /* create a detached thread so that we don't need to wait on it */
> +    qemu_thread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
> +
> +    /* block all signals */
> +    if (sigfillset(&set)) {
> +        async_abort(errno, "sigfillset");
> +    }
> +
> +    if (sigprocmask(SIG_SETMASK,&set,&oldset)) {
> +        async_abort(errno, "sigprocmask");
> +    }
> +
> +    qemu_thread_create_attr(&thread,&attr, async_worker_thread, queue);
> +
> +    if (sigprocmask(SIG_SETMASK,&oldset, NULL)) {
> +        async_abort(errno, "sigprocmask restore");
> +    }
> +}
> +
> +void qemu_async_submit(struct async_queue *queue, struct work_item *work)
> +{
> +    qemu_mutex_lock(&(queue->lock));
> +    if (queue->idle_threads == 0&&  queue->cur_threads<  queue->max_threads) {
> +        spawn_async_thread(queue);
> +    }
> +    QTAILQ_INSERT_TAIL(&(queue->request_list), work, node);
> +    qemu_mutex_unlock(&(queue->lock));
> +    qemu_cond_signal(&(queue->cond));
> +}
> +
> +int qemu_async_cancel_work(struct async_queue *queue, struct work_item *work)
> +{
> +    struct work_item *ret_work;
> +    int found = 0;
> +
> +    qemu_mutex_lock(&(queue->lock));
> +    QTAILQ_FOREACH(ret_work,&(queue->request_list), node) {
> +        if (ret_work == work) {
> +            QTAILQ_REMOVE(&(queue->request_list), ret_work, node);
> +            found = 1;
> +            break;
> +        }
> +    }
> +    qemu_mutex_unlock(&(queue->lock));
> +
> +    if (found) {
> +        async_work_release(queue, work);
> +        return 0;
> +    }
> +
> +    return 1;
> +}
> +
> diff --git a/async-work.h b/async-work.h
> new file mode 100644
> index 0000000..8389f56
> --- /dev/null
> +++ b/async-work.h
> @@ -0,0 +1,85 @@
> +/*
> + * Async work support
> + *
> + * Copyright IBM, Corp. 2010
> + *
> + * Authors:
> + *  Aneesh Kumar K.V<aneesh.kumar@linux.vnet.ibm.com>
> + *
> + * This work is licensed under the terms of the GNU GPL, version 2.  See
> + * the COPYING file in the top-level directory.
> + *
> + */
> +#ifndef QEMU_ASYNC_WORK_H
> +#define QEMU_ASYNC_WORK_H
> +
> +#include "qemu-queue.h"
> +#include "qemu-common.h"
> +#include "qemu-thread.h"
> +
> +struct async_queue
> +{
> +    QemuMutex lock;
> +    QemuCond cond;
> +    int max_threads;
> +    int min_threads;
> +    int cur_threads;
> +    int idle_threads;
> +    QTAILQ_HEAD(, work_item) request_list;
> +    QTAILQ_HEAD(, work_item) work_item_pool;
> +};
> +
> +struct work_item
> +{
> +    QTAILQ_ENTRY(work_item) node;
> +    void (*func)(struct work_item *work);
> +    void *private;
> +};
>    

Structs are not named in accordance to CODING_STYLE.

> +static inline void async_queue_init(struct async_queue *queue,
> +				    int max_threads, int min_threads)
> +{
> +    queue->cur_threads  = 0;
> +    queue->idle_threads = 0;
> +    queue->max_threads  = max_threads;
> +    queue->min_threads  = min_threads;
> +    QTAILQ_INIT(&(queue->request_list));
> +    QTAILQ_INIT(&(queue->work_item_pool));
> +    qemu_mutex_init(&(queue->lock));
> +    qemu_cond_init(&(queue->cond));
> +}
>    

I'd prefer there be a single queue that everything used verses multiple 
queues.  Otherwise, we'll end up having per device queues and my concern 
is that we'll end up with thousands and thousands of threads with no 
central place to tune the maximum thread number.

> +static inline struct work_item *async_work_init(struct async_queue *queue,
> +				   void (*func)(struct work_item *),
> +				   void *data)
>    

I'd suggest actually using a Notifier as the worker or at least 
something that looks exactly like it.  There's no need to pass a void * 
because more often than not, a caller just wants to pass a state 
structure anyway and they can embed the Notifier within the structure.  IOW:

async_work_submit(queue, &s->worker);

Then in the callback:

DeviceState *s = container_of(worker, DeviceState, worker);

I don't think the name makes the most sense either.  I think something like:

threadlet_submit()

Would work best.  It would be good for there to be a big comment warning 
that the routine does not run with the qemu_mutex and therefore cannot 
make use of any qemu functions without very special consideration.


There shouldn't need to be an explicit init vs. submit function either.

Regards,

Anthony Liguori

Re: [Qemu-devel] [PATCH V3 2/3] qemu: Generic asynchronous threading framework to offload tasks

[Corentin Chary]

On Fri, Jun 4, 2010 at 3:16 PM, Anthony Liguori <anthony@codemonkey.ws> wrote:
> On 06/03/2010 03:56 AM, Gautham R Shenoy wrote:
>>
>> From: Aneesh Kumar K.V<aneesh.kumar@linux.vnet.ibm.com>
>>
>> This patch creates a generic asynchronous-task-offloading infrastructure.
>> It's
>> extracted out of the threading framework that is being used by paio.
>>
>> The reason for extracting out this generic infrastructure of the
>> posix-aio-compat.c is so that other subsystems, such as virtio-9p could
>> make use
>> of it for offloading tasks that could block.
>>
>> [ego@in.ibm.com: work_item_pool, async_work_init, async_work_release,
>> async_cancel_work]
>>
>> Signed-off-by: Aneesh Kumar K.V<aneesh.kumar@linux.vnet.ibm.com>
>> Signed-off-by: Gautham R Shenoy<ego@in.ibm.com>
>> ---
>>  Makefile.objs |    3 +
>>  async-work.c  |  136
>> +++++++++++++++++++++++++++++++++++++++++++++++++++++++++
>>  async-work.h  |   85 ++++++++++++++++++++++++++++++++++++
>>  3 files changed, 223 insertions(+), 1 deletions(-)
>>  create mode 100644 async-work.c
>>  create mode 100644 async-work.h
>>
>> diff --git a/Makefile.objs b/Makefile.objs
>> index ecdd53e..fd5ea4d 100644
>> --- a/Makefile.objs
>> +++ b/Makefile.objs
>> @@ -9,6 +9,8 @@ qobject-obj-y += qerror.o
>>
>>  block-obj-y = cutils.o cache-utils.o qemu-malloc.o qemu-option.o module.o
>>  block-obj-y += nbd.o block.o aio.o aes.o osdep.o qemu-config.o
>> +block-obj-y += qemu-thread.o
>> +block-obj-y += async-work.o
>>  block-obj-$(CONFIG_POSIX) += posix-aio-compat.o
>>  block-obj-$(CONFIG_LINUX_AIO) += linux-aio.o
>>
>> @@ -108,7 +110,6 @@ common-obj-y += iov.o
>>  common-obj-$(CONFIG_VNC_TLS) += vnc-tls.o vnc-auth-vencrypt.o
>>  common-obj-$(CONFIG_VNC_SASL) += vnc-auth-sasl.o
>>  common-obj-$(CONFIG_COCOA) += cocoa.o
>> -common-obj-$(CONFIG_IOTHREAD) += qemu-thread.o
>>  common-obj-y += notify.o event_notifier.o
>>  common-obj-y += qemu-timer.o
>>
>> diff --git a/async-work.c b/async-work.c
>> new file mode 100644
>> index 0000000..0675732
>> --- /dev/null
>> +++ b/async-work.c
>> @@ -0,0 +1,136 @@
>> +/*
>> + * Async work support
>> + *
>> + * Copyright IBM, Corp. 2010
>> + *
>> + * Authors:
>> + *  Aneesh Kumar K.V<aneesh.kumar@linux.vnet.ibm.com>
>> + *
>> + * This work is licensed under the terms of the GNU GPL, version 2.  See
>> + * the COPYING file in the top-level directory.
>>
>
> Please preserve the original copyright of the copied code.
>
>> + */
>> +#include<stdio.h>
>> +#include<errno.h>
>> +#include<string.h>
>> +#include<stdlib.h>
>> +#include<signal.h>
>>
>
> qemu-common.h should have all of these.  Generally, you should avoid
> including system headers because qemu headers take care of portability.
>
>> +#include "async-work.h"
>> +#include "osdep.h"
>> +
>> +static void async_abort(int err, const char *what)
>> +{
>> +    fprintf(stderr, "%s failed: %s\n", what, strerror(err));
>> +    abort();
>> +}
>> +
>> +static void *async_worker_thread(void *data)
>> +{
>> +    struct async_queue *queue = data;
>> +
>> +    while (1) {
>> +        struct work_item *work;
>> +        int ret = 0;
>> +        qemu_mutex_lock(&(queue->lock));
>> +
>> +        while (QTAILQ_EMPTY(&(queue->request_list))&&
>> +               (ret != ETIMEDOUT)) {
>> +            ret = qemu_cond_timedwait(&(queue->cond),
>> +                                       &(queue->lock), 10*100000);
>> +        }
>> +
>> +        if (QTAILQ_EMPTY(&(queue->request_list)))
>> +            goto check_exit;
>> +
>> +        work = QTAILQ_FIRST(&(queue->request_list));
>> +        QTAILQ_REMOVE(&(queue->request_list), work, node);
>> +        queue->idle_threads--;
>> +        qemu_mutex_unlock(&(queue->lock));
>> +
>> +        /* execute the work function */
>> +        work->func(work);
>> +        async_work_release(queue, work);
>> +
>> +        qemu_mutex_lock(&(queue->lock));
>> +        queue->idle_threads++;
>> +
>> +check_exit:
>> +        if ((queue->idle_threads>  0)&&
>> +            (queue->cur_threads>  queue->min_threads)) {
>> +            /* we retain minimum number of threads */
>> +            break;
>> +        }
>> +        qemu_mutex_unlock(&(queue->lock));
>> +    }
>> +
>> +    queue->idle_threads--;
>> +    queue->cur_threads--;
>> +    qemu_mutex_unlock(&(queue->lock));
>> +
>> +    return NULL;
>> +}
>> +
>> +static void spawn_async_thread(struct async_queue *queue)
>> +{
>> +    QemuThreadAttr attr;
>> +    QemuThread thread;
>> +    sigset_t set, oldset;
>> +
>> +    queue->cur_threads++;
>> +    queue->idle_threads++;
>> +
>> +    qemu_thread_attr_init(&attr);
>> +
>> +    /* create a detached thread so that we don't need to wait on it */
>> +    qemu_thread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
>> +
>> +    /* block all signals */
>> +    if (sigfillset(&set)) {
>> +        async_abort(errno, "sigfillset");
>> +    }
>> +
>> +    if (sigprocmask(SIG_SETMASK,&set,&oldset)) {
>> +        async_abort(errno, "sigprocmask");
>> +    }
>> +
>> +    qemu_thread_create_attr(&thread,&attr, async_worker_thread, queue);
>> +
>> +    if (sigprocmask(SIG_SETMASK,&oldset, NULL)) {
>> +        async_abort(errno, "sigprocmask restore");
>> +    }
>> +}
>> +
>> +void qemu_async_submit(struct async_queue *queue, struct work_item *work)
>> +{
>> +    qemu_mutex_lock(&(queue->lock));
>> +    if (queue->idle_threads == 0&&  queue->cur_threads<
>>  queue->max_threads) {
>> +        spawn_async_thread(queue);
>> +    }
>> +    QTAILQ_INSERT_TAIL(&(queue->request_list), work, node);
>> +    qemu_mutex_unlock(&(queue->lock));
>> +    qemu_cond_signal(&(queue->cond));
>> +}
>> +
>> +int qemu_async_cancel_work(struct async_queue *queue, struct work_item
>> *work)
>> +{
>> +    struct work_item *ret_work;
>> +    int found = 0;
>> +
>> +    qemu_mutex_lock(&(queue->lock));
>> +    QTAILQ_FOREACH(ret_work,&(queue->request_list), node) {
>> +        if (ret_work == work) {
>> +            QTAILQ_REMOVE(&(queue->request_list), ret_work, node);
>> +            found = 1;
>> +            break;
>> +        }
>> +    }
>> +    qemu_mutex_unlock(&(queue->lock));
>> +
>> +    if (found) {
>> +        async_work_release(queue, work);
>> +        return 0;
>> +    }
>> +
>> +    return 1;
>> +}
>> +
>> diff --git a/async-work.h b/async-work.h
>> new file mode 100644
>> index 0000000..8389f56
>> --- /dev/null
>> +++ b/async-work.h
>> @@ -0,0 +1,85 @@
>> +/*
>> + * Async work support
>> + *
>> + * Copyright IBM, Corp. 2010
>> + *
>> + * Authors:
>> + *  Aneesh Kumar K.V<aneesh.kumar@linux.vnet.ibm.com>
>> + *
>> + * This work is licensed under the terms of the GNU GPL, version 2.  See
>> + * the COPYING file in the top-level directory.
>> + *
>> + */
>> +#ifndef QEMU_ASYNC_WORK_H
>> +#define QEMU_ASYNC_WORK_H
>> +
>> +#include "qemu-queue.h"
>> +#include "qemu-common.h"
>> +#include "qemu-thread.h"
>> +
>> +struct async_queue
>> +{
>> +    QemuMutex lock;
>> +    QemuCond cond;
>> +    int max_threads;
>> +    int min_threads;
>> +    int cur_threads;
>> +    int idle_threads;
>> +    QTAILQ_HEAD(, work_item) request_list;
>> +    QTAILQ_HEAD(, work_item) work_item_pool;
>> +};
>> +
>> +struct work_item
>> +{
>> +    QTAILQ_ENTRY(work_item) node;
>> +    void (*func)(struct work_item *work);
>> +    void *private;
>> +};
>>
>
> Structs are not named in accordance to CODING_STYLE.
>
>> +static inline void async_queue_init(struct async_queue *queue,
>> +                                   int max_threads, int min_threads)
>> +{
>> +    queue->cur_threads  = 0;
>> +    queue->idle_threads = 0;
>> +    queue->max_threads  = max_threads;
>> +    queue->min_threads  = min_threads;
>> +    QTAILQ_INIT(&(queue->request_list));
>> +    QTAILQ_INIT(&(queue->work_item_pool));
>> +    qemu_mutex_init(&(queue->lock));
>> +    qemu_cond_init(&(queue->cond));
>> +}
>>
>
> I'd prefer there be a single queue that everything used verses multiple
> queues.  Otherwise, we'll end up having per device queues and my concern is
> that we'll end up with thousands and thousands of threads with no central
> place to tune the maximum thread number.

If there a single queue, we'll need something to control how job are
processed. For example,
in the VNC server, they must be processed in order (in fact, in order
per VNC client, but I don't see how we could do that).

>> +static inline struct work_item *async_work_init(struct async_queue
>> *queue,
>> +                                  void (*func)(struct work_item *),
>> +                                  void *data)
>>
>
> I'd suggest actually using a Notifier as the worker or at least something
> that looks exactly like it.  There's no need to pass a void * because more
> often than not, a caller just wants to pass a state structure anyway and
> they can embed the Notifier within the structure.  IOW:
>
> async_work_submit(queue, &s->worker);
>
> Then in the callback:
>
> DeviceState *s = container_of(worker, DeviceState, worker);
>
> I don't think the name makes the most sense either.  I think something like:
>
> threadlet_submit()
>
> Would work best.  It would be good for there to be a big comment warning
> that the routine does not run with the qemu_mutex and therefore cannot make
> use of any qemu functions without very special consideration.
>
>
> There shouldn't need to be an explicit init vs. submit function either.
>
> Regards,
>
> Anthony Liguori
>



-- 
Corentin Chary
http://xf.iksaif.net

Re: [Qemu-devel] [PATCH V3 2/3] qemu: Generic asynchronous threading framework to offload tasks

[Gautham R Shenoy]

On Fri, Jun 04, 2010 at 08:16:19AM -0500, Anthony Liguori wrote:
>> --- /dev/null
>> +++ b/async-work.c
>> @@ -0,0 +1,136 @@
>> +/*
>> + * Async work support
>> + *
>> + * Copyright IBM, Corp. 2010
>> + *
>> + * Authors:
>> + *  Aneesh Kumar K.V<aneesh.kumar@linux.vnet.ibm.com>
>> + *
>> + * This work is licensed under the terms of the GNU GPL, version 2.  See
>> + * the COPYING file in the top-level directory.
>>    
>
> Please preserve the original copyright of the copied code.

Will update the comment containing the Copyright.
>
>> +
>> +struct work_item
>> +{
>> +    QTAILQ_ENTRY(work_item) node;
>> +    void (*func)(struct work_item *work);
>> +    void *private;
>> +};
>>    
>
> Structs are not named in accordance to CODING_STYLE.

Will fix this.

>
>> +static inline void async_queue_init(struct async_queue *queue,
>> +				    int max_threads, int min_threads)
>> +{
>> +    queue->cur_threads  = 0;
>> +    queue->idle_threads = 0;
>> +    queue->max_threads  = max_threads;
>> +    queue->min_threads  = min_threads;
>> +    QTAILQ_INIT(&(queue->request_list));
>> +    QTAILQ_INIT(&(queue->work_item_pool));
>> +    qemu_mutex_init(&(queue->lock));
>> +    qemu_cond_init(&(queue->cond));
>> +}
>>    
>
> I'd prefer there be a single queue that everything used verses multiple 
> queues.  Otherwise, we'll end up having per device queues and my concern is 
> that we'll end up with thousands and thousands of threads with no central 
> place to tune the maximum thread number.

Aah! So, the original idea was to have a single queue, but since we were
making it generic, we thought that the subsystems might like the
flexibility of having their own queue.

I suppose we are not looking to differentiate between the worker threads
belonging to different subsystems in terms of their relative
importance/priorities, right ?

>
>> +static inline struct work_item *async_work_init(struct async_queue *queue,
>> +				   void (*func)(struct work_item *),
>> +				   void *data)
>>    
>
> I'd suggest actually using a Notifier as the worker or at least something 
> that looks exactly like it.  There's no need to pass a void * because more 
> often than not, a caller just wants to pass a state structure anyway and 
> they can embed the Notifier within the structure.  IOW:
>
> async_work_submit(queue, &s->worker);
>
> Then in the callback:
>
> DeviceState *s = container_of(worker, DeviceState, worker);
>
> I don't think the name makes the most sense either.  I think something like:
>
> threadlet_submit()

Makes sense. Will implement this.
>
> Would work best.  It would be good for there to be a big comment warning 
> that the routine does not run with the qemu_mutex and therefore cannot make 
> use of any qemu functions without very special consideration.
>
>
> There shouldn't need to be an explicit init vs. submit function either.

Ok, will address these comments.
>
> Regards,
>
> Anthony Liguori

-- 
Thanks and Regards
gautham

[Qemu-devel] [PATCH V3 3/3] qemu: Convert AIO code to use the generic threading infrastructure.

[Gautham R Shenoy]

This patch makes the paio subsystem use the generic work offloading
infrastructure, there by 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>
---
 posix-aio-compat.c |  155 ++++++++++------------------------------------------
 1 files changed, 29 insertions(+), 126 deletions(-)

diff --git a/posix-aio-compat.c b/posix-aio-compat.c
index b43c531..f2e7c6a 100644
--- a/posix-aio-compat.c
+++ b/posix-aio-compat.c
@@ -28,6 +28,7 @@
 #include "block_int.h"
 
 #include "block/raw-posix-aio.h"
+#include "async-work.h"
 
 
 struct qemu_paiocb {
@@ -50,6 +51,7 @@ struct qemu_paiocb {
     struct qemu_paiocb *next;
 
     int async_context_id;
+    struct work_item *work;
 };
 
 typedef struct PosixAioState {
@@ -57,15 +59,8 @@ typedef struct PosixAioState {
     struct qemu_paiocb *first_aio;
 } 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;
+static struct async_queue aio_request_list;
 
 #ifdef CONFIG_PREADV
 static int preadv_present = 1;
@@ -84,39 +79,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;
@@ -300,47 +262,27 @@ static ssize_t handle_aiocb_rw(struct qemu_paiocb *aiocb)
     return nbytes;
 }
 
-static void *aio_thread(void *unused)
+static void aio_thread(struct work_item *work)
 {
-    pid_t pid;
 
-    pid = getpid();
-
-    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);
+    pid_t pid;
 
-        while (QTAILQ_EMPTY(&request_list) &&
-               !(ret == ETIMEDOUT)) {
-            ret = cond_timedwait(&cond, &lock, &ts);
-        }
+    struct qemu_paiocb *aiocb = (struct qemu_paiocb *) work->private;
+    ssize_t ret = 0;
 
-        if (QTAILQ_EMPTY(&request_list))
-            break;
+    pid = getpid();
 
-        aiocb = QTAILQ_FIRST(&request_list);
-        QTAILQ_REMOVE(&request_list, aiocb, node);
-        aiocb->active = 1;
-        idle_threads--;
-        mutex_unlock(&lock);
+    aiocb->active = 1;
 
-        switch (aiocb->aio_type & QEMU_AIO_TYPE_MASK) {
-        case QEMU_AIO_READ:
-        case QEMU_AIO_WRITE:
+    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:
+    case QEMU_AIO_FLUSH:
+        ret = handle_aiocb_flush(aiocb);
+        break;
+    case QEMU_AIO_IOCTL:
 		ret = handle_aiocb_ioctl(aiocb);
 		break;
 	default:
@@ -349,57 +291,28 @@ static void *aio_thread(void *unused)
 		break;
 	}
 
-        mutex_lock(&lock);
-        aiocb->ret = ret;
-        idle_threads++;
-        mutex_unlock(&lock);
-
-        if (kill(pid, aiocb->ev_signo)) die("kill failed");
-    }
-
-    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)
 {
+    struct work_item *work;
+
     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);
+
+    work = async_work_init(&aio_request_list, aio_thread, aiocb);
+    aiocb->work = work;
+    qemu_async_submit(&aio_request_list, 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;
 }
 
@@ -535,14 +448,14 @@ static void paio_cancel(BlockDriverAIOCB *blockacb)
     struct qemu_paiocb *acb = (struct qemu_paiocb *)blockacb;
     int active = 0;
 
-    mutex_lock(&lock);
     if (!acb->active) {
-        QTAILQ_REMOVE(&request_list, acb, node);
-        acb->ret = -ECANCELED;
+        if (!qemu_async_cancel_work(&aio_request_list, acb->work))
+            acb->ret = -ECANCELED;
+         else
+            active = 1;
     } else if (acb->ret == -EINPROGRESS) {
         active = 1;
     }
-    mutex_unlock(&lock);
 
     if (active) {
         /* fail safe: if the aio could not be canceled, we wait for
@@ -615,7 +528,6 @@ int paio_init(void)
     struct sigaction act;
     PosixAioState *s;
     int fds[2];
-    int ret;
 
     if (posix_aio_state)
         return 0;
@@ -642,16 +554,7 @@ int paio_init(void)
     qemu_aio_set_fd_handler(s->rfd, posix_aio_read, NULL, posix_aio_flush,
         posix_aio_process_queue, s);
 
-    ret = pthread_attr_init(&attr);
-    if (ret)
-        die2(ret, "pthread_attr_init");
-
-    ret = pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
-    if (ret)
-        die2(ret, "pthread_attr_setdetachstate");
-
-    QTAILQ_INIT(&request_list);
-
     posix_aio_state = s;
+    async_queue_init(&aio_request_list, max_threads, max_threads);
     return 0;
 }