* [Qemu-devel] [PATCH 01/12] Add aiocb_mutex and aiocb_completion.
2011-01-13 12:14 [Qemu-devel] [PATCH 00/12] Threadlets Infrastructure Arun R Bharadwaj
@ 2011-01-13 12:14 ` Arun R Bharadwaj
2011-01-13 12:14 ` [Qemu-devel] [PATCH 02/12] Introduce work concept in posix-aio-compat.c Arun R Bharadwaj
` (11 subsequent siblings)
12 siblings, 0 replies; 27+ messages in thread
From: Arun R Bharadwaj @ 2011-01-13 12:14 UTC (permalink / raw)
To: qemu-devel; +Cc: kwolf, aliguori, jvrao, aneesh.kumar
This patch adds the aiocb_mutex to protect aiocb.
This patch also removes the infinite loop present in paio_cancel.
Since there can only be one cancellation at a time, we need to
introduce a condition variable. For this, we need a
global aiocb_completion condition variable.
This patch also adds the Makefile entry to compile qemu-thread.c
when CONFIG_POSIX is set, instead of the unused CONFIG_THREAD.
Signed-off-by: Arun R Bharadwaj <arun@linux.vnet.ibm.com>
---
Makefile.objs | 2 +-
posix-aio-compat.c | 39 ++++++++++++++++++++++++++++-----------
2 files changed, 29 insertions(+), 12 deletions(-)
diff --git a/Makefile.objs b/Makefile.objs
index cd5a24b..3b7ec27 100644
--- a/Makefile.objs
+++ b/Makefile.objs
@@ -9,6 +9,7 @@ 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-$(CONFIG_POSIX) += qemu-thread.o
block-obj-$(CONFIG_POSIX) += posix-aio-compat.o
block-obj-$(CONFIG_LINUX_AIO) += linux-aio.o
@@ -124,7 +125,6 @@ endif
common-obj-y += $(addprefix ui/, $(ui-obj-y))
common-obj-y += iov.o acl.o
-common-obj-$(CONFIG_THREAD) += qemu-thread.o
common-obj-y += notify.o event_notifier.o
common-obj-y += qemu-timer.o
diff --git a/posix-aio-compat.c b/posix-aio-compat.c
index 7b862b5..82862ec 100644
--- a/posix-aio-compat.c
+++ b/posix-aio-compat.c
@@ -27,9 +27,12 @@
#include "qemu-common.h"
#include "trace.h"
#include "block_int.h"
+#include "qemu-thread.h"
#include "block/raw-posix-aio.h"
+static QemuMutex aiocb_mutex;
+static QemuCond aiocb_completion;
struct qemu_paiocb {
BlockDriverAIOCB common;
@@ -351,10 +354,14 @@ static void *aio_thread(void *unused)
}
mutex_lock(&lock);
- aiocb->ret = ret;
idle_threads++;
mutex_unlock(&lock);
+ qemu_mutex_lock(&aiocb_mutex);
+ aiocb->ret = ret;
+ qemu_cond_broadcast(&aiocb_completion);
+ qemu_mutex_unlock(&aiocb_mutex);
+
if (kill(pid, aiocb->ev_signo)) die("kill failed");
}
@@ -383,8 +390,11 @@ static void spawn_thread(void)
static void qemu_paio_submit(struct qemu_paiocb *aiocb)
{
+ qemu_mutex_lock(&aiocb_mutex);
aiocb->ret = -EINPROGRESS;
aiocb->active = 0;
+ qemu_mutex_unlock(&aiocb_mutex);
+
mutex_lock(&lock);
if (idle_threads == 0 && cur_threads < max_threads)
spawn_thread();
@@ -397,9 +407,9 @@ static ssize_t qemu_paio_return(struct qemu_paiocb *aiocb)
{
ssize_t ret;
- mutex_lock(&lock);
+ qemu_mutex_lock(&aiocb_mutex);
ret = aiocb->ret;
- mutex_unlock(&lock);
+ qemu_mutex_unlock(&aiocb_mutex);
return ret;
}
@@ -536,22 +546,26 @@ static void paio_cancel(BlockDriverAIOCB *blockacb)
struct qemu_paiocb *acb = (struct qemu_paiocb *)blockacb;
int active = 0;
- mutex_lock(&lock);
+ qemu_mutex_lock(&aiocb_mutex);
if (!acb->active) {
QTAILQ_REMOVE(&request_list, acb, node);
acb->ret = -ECANCELED;
} else if (acb->ret == -EINPROGRESS) {
active = 1;
}
- mutex_unlock(&lock);
- if (active) {
- /* fail safe: if the aio could not be canceled, we wait for
- it */
- while (qemu_paio_error(acb) == EINPROGRESS)
- ;
+ if (!active) {
+ acb->ret = -ECANCELED;
+ } else {
+ while (acb->ret == -EINPROGRESS) {
+ /*
+ * fail safe: if the aio could not be canceled,
+ * we wait for it
+ */
+ qemu_cond_wait(&aiocb_completion, &aiocb_mutex);
+ }
}
-
+ qemu_mutex_unlock(&aiocb_mutex);
paio_remove(acb);
}
@@ -623,6 +637,9 @@ int paio_init(void)
if (posix_aio_state)
return 0;
+ qemu_mutex_init(&aiocb_mutex);
+ qemu_cond_init(&aiocb_completion);
+
s = qemu_malloc(sizeof(PosixAioState));
sigfillset(&act.sa_mask);
^ permalink raw reply related [flat|nested] 27+ messages in thread* [Qemu-devel] [PATCH 02/12] Introduce work concept in posix-aio-compat.c
2011-01-13 12:14 [Qemu-devel] [PATCH 00/12] Threadlets Infrastructure Arun R Bharadwaj
2011-01-13 12:14 ` [Qemu-devel] [PATCH 01/12] Add aiocb_mutex and aiocb_completion Arun R Bharadwaj
@ 2011-01-13 12:14 ` Arun R Bharadwaj
2011-01-13 12:14 ` [Qemu-devel] [PATCH 03/12] Add callback function to ThreadletWork structure Arun R Bharadwaj
` (10 subsequent siblings)
12 siblings, 0 replies; 27+ messages in thread
From: Arun R Bharadwaj @ 2011-01-13 12:14 UTC (permalink / raw)
To: qemu-devel; +Cc: kwolf, aliguori, jvrao, aneesh.kumar
This patch introduces work concept by introducing the
ThreadletWork structure.
Signed-off-by: Arun R Bharadwaj <arun@linux.vnet.ibm.com>
---
posix-aio-compat.c | 20 ++++++++++++++------
1 files changed, 14 insertions(+), 6 deletions(-)
diff --git a/posix-aio-compat.c b/posix-aio-compat.c
index 82862ec..ddf42f5 100644
--- a/posix-aio-compat.c
+++ b/posix-aio-compat.c
@@ -33,6 +33,10 @@
static QemuMutex aiocb_mutex;
static QemuCond aiocb_completion;
+typedef struct ThreadletWork
+{
+ QTAILQ_ENTRY(ThreadletWork) node;
+} ThreadletWork;
struct qemu_paiocb {
BlockDriverAIOCB common;
@@ -47,13 +51,13 @@ struct qemu_paiocb {
int ev_signo;
off_t aio_offset;
- QTAILQ_ENTRY(qemu_paiocb) node;
int aio_type;
ssize_t ret;
int active;
struct qemu_paiocb *next;
int async_context_id;
+ ThreadletWork work;
};
typedef struct PosixAioState {
@@ -69,7 +73,7 @@ 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 QTAILQ_HEAD(, ThreadletWork) request_list;
#ifdef CONFIG_PREADV
static int preadv_present = 1;
@@ -307,6 +311,7 @@ static ssize_t handle_aiocb_rw(struct qemu_paiocb *aiocb)
static void *aio_thread(void *unused)
{
pid_t pid;
+ ThreadletWork *work;
pid = getpid();
@@ -330,8 +335,11 @@ static void *aio_thread(void *unused)
if (QTAILQ_EMPTY(&request_list))
break;
- aiocb = QTAILQ_FIRST(&request_list);
- QTAILQ_REMOVE(&request_list, aiocb, node);
+ work = QTAILQ_FIRST(&request_list);
+ QTAILQ_REMOVE(&request_list, work, node);
+
+ aiocb = container_of(work, struct qemu_paiocb, work);
+
aiocb->active = 1;
idle_threads--;
mutex_unlock(&lock);
@@ -398,7 +406,7 @@ static void qemu_paio_submit(struct qemu_paiocb *aiocb)
mutex_lock(&lock);
if (idle_threads == 0 && cur_threads < max_threads)
spawn_thread();
- QTAILQ_INSERT_TAIL(&request_list, aiocb, node);
+ QTAILQ_INSERT_TAIL(&request_list, &aiocb->work, node);
mutex_unlock(&lock);
cond_signal(&cond);
}
@@ -548,7 +556,7 @@ static void paio_cancel(BlockDriverAIOCB *blockacb)
qemu_mutex_lock(&aiocb_mutex);
if (!acb->active) {
- QTAILQ_REMOVE(&request_list, acb, node);
+ QTAILQ_REMOVE(&request_list, &acb->work, node);
acb->ret = -ECANCELED;
} else if (acb->ret == -EINPROGRESS) {
active = 1;
^ permalink raw reply related [flat|nested] 27+ messages in thread* [Qemu-devel] [PATCH 03/12] Add callback function to ThreadletWork structure.
2011-01-13 12:14 [Qemu-devel] [PATCH 00/12] Threadlets Infrastructure Arun R Bharadwaj
2011-01-13 12:14 ` [Qemu-devel] [PATCH 01/12] Add aiocb_mutex and aiocb_completion Arun R Bharadwaj
2011-01-13 12:14 ` [Qemu-devel] [PATCH 02/12] Introduce work concept in posix-aio-compat.c Arun R Bharadwaj
@ 2011-01-13 12:14 ` Arun R Bharadwaj
2011-01-13 12:14 ` [Qemu-devel] [PATCH 04/12] Add ThreadletQueue Arun R Bharadwaj
` (9 subsequent siblings)
12 siblings, 0 replies; 27+ messages in thread
From: Arun R Bharadwaj @ 2011-01-13 12:14 UTC (permalink / raw)
To: qemu-devel; +Cc: kwolf, aliguori, jvrao, aneesh.kumar
This patch adds the callback function to the ThreadletWork
structure and moves aio handler as a callback function.
Signed-off-by: Arun R Bharadwaj <arun@linux.vnet.ibm.com>
---
posix-aio-compat.c | 89 +++++++++++++++++++++++++++++-----------------------
1 files changed, 50 insertions(+), 39 deletions(-)
diff --git a/posix-aio-compat.c b/posix-aio-compat.c
index ddf42f5..4fa2c47 100644
--- a/posix-aio-compat.c
+++ b/posix-aio-compat.c
@@ -36,6 +36,7 @@ static QemuCond aiocb_completion;
typedef struct ThreadletWork
{
QTAILQ_ENTRY(ThreadletWork) node;
+ void (*func)(struct ThreadletWork *work);
} ThreadletWork;
struct qemu_paiocb {
@@ -308,18 +309,14 @@ static ssize_t handle_aiocb_rw(struct qemu_paiocb *aiocb)
return nbytes;
}
-static void *aio_thread(void *unused)
+static void *threadlet_worker(void *data)
{
- pid_t pid;
- ThreadletWork *work;
-
- pid = getpid();
while (1) {
- struct qemu_paiocb *aiocb;
ssize_t ret = 0;
qemu_timeval tv;
struct timespec ts;
+ ThreadletWork *work;
qemu_gettimeofday(&tv);
ts.tv_sec = tv.tv_sec + 10;
@@ -332,52 +329,64 @@ static void *aio_thread(void *unused)
ret = cond_timedwait(&cond, &lock, &ts);
}
- if (QTAILQ_EMPTY(&request_list))
+ if (QTAILQ_EMPTY(&request_list)) {
+ idle_threads--;
+ cur_threads--;
+ mutex_unlock(&lock);
break;
-
+ }
work = QTAILQ_FIRST(&request_list);
QTAILQ_REMOVE(&request_list, work, node);
-
- aiocb = container_of(work, struct qemu_paiocb, work);
-
- 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;
- }
-
+ work->func(work);
mutex_lock(&lock);
idle_threads++;
mutex_unlock(&lock);
- qemu_mutex_lock(&aiocb_mutex);
- aiocb->ret = ret;
- qemu_cond_broadcast(&aiocb_completion);
- qemu_mutex_unlock(&aiocb_mutex);
+ }
+
+ return NULL;
+}
+
+static void handle_work(ThreadletWork *work)
+{
+ pid_t pid;
+ ssize_t ret = 0;
+ struct qemu_paiocb *aiocb;
- if (kill(pid, aiocb->ev_signo)) die("kill failed");
+ pid = getpid();
+ qemu_mutex_lock(&aiocb_mutex);
+ aiocb = container_of(work, struct qemu_paiocb, work);
+ aiocb->active = 1;
+ qemu_mutex_unlock(&aiocb_mutex);
+
+ 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);
+ qemu_mutex_lock(&aiocb_mutex);
+ aiocb->ret = ret;
+ qemu_cond_broadcast(&aiocb_completion);
+ qemu_mutex_unlock(&aiocb_mutex);
- return NULL;
+ if (kill(pid, aiocb->ev_signo)) {
+ die("kill failed");
+ }
}
static void spawn_thread(void)
@@ -391,7 +400,7 @@ static void spawn_thread(void)
if (sigfillset(&set)) die("sigfillset");
if (sigprocmask(SIG_SETMASK, &set, &oldset)) die("sigprocmask");
- thread_create(&thread_id, &attr, aio_thread, NULL);
+ thread_create(&thread_id, &attr, threadlet_worker, NULL);
if (sigprocmask(SIG_SETMASK, &oldset, NULL)) die("sigprocmask restore");
}
@@ -406,6 +415,8 @@ static void qemu_paio_submit(struct qemu_paiocb *aiocb)
mutex_lock(&lock);
if (idle_threads == 0 && cur_threads < max_threads)
spawn_thread();
+
+ aiocb->work.func = handle_work;
QTAILQ_INSERT_TAIL(&request_list, &aiocb->work, node);
mutex_unlock(&lock);
cond_signal(&cond);
^ permalink raw reply related [flat|nested] 27+ messages in thread* [Qemu-devel] [PATCH 04/12] Add ThreadletQueue.
2011-01-13 12:14 [Qemu-devel] [PATCH 00/12] Threadlets Infrastructure Arun R Bharadwaj
` (2 preceding siblings ...)
2011-01-13 12:14 ` [Qemu-devel] [PATCH 03/12] Add callback function to ThreadletWork structure Arun R Bharadwaj
@ 2011-01-13 12:14 ` Arun R Bharadwaj
2011-01-13 12:14 ` [Qemu-devel] [PATCH 05/12] Threadlet: Add submit_work threadlet API Arun R Bharadwaj
` (8 subsequent siblings)
12 siblings, 0 replies; 27+ messages in thread
From: Arun R Bharadwaj @ 2011-01-13 12:14 UTC (permalink / raw)
To: qemu-devel; +Cc: kwolf, aliguori, jvrao, aneesh.kumar
This patch adds a global queue of type ThreadletQueue and
removes the earlier usage of request_list queue.
We want to create the thread on the first submit. Hence we
need to track whether the globalqueue is initialized or not.
Signed-off-by: Arun R Bharadwaj <arun@linux.vnet.ibm.com>
---
posix-aio-compat.c | 149 +++++++++++++++++++++++++++-------------------------
1 files changed, 76 insertions(+), 73 deletions(-)
diff --git a/posix-aio-compat.c b/posix-aio-compat.c
index 4fa2c47..b5d70c9 100644
--- a/posix-aio-compat.c
+++ b/posix-aio-compat.c
@@ -31,8 +31,23 @@
#include "block/raw-posix-aio.h"
+#define MAX_GLOBAL_THREADS 64
+#define MIN_GLOBAL_THREADS 8
+
static QemuMutex aiocb_mutex;
static QemuCond aiocb_completion;
+
+typedef struct ThreadletQueue
+{
+ QemuMutex lock;
+ QemuCond cond;
+ int max_threads;
+ int min_threads;
+ int cur_threads;
+ int idle_threads;
+ QTAILQ_HEAD(, ThreadletWork) request_list;
+} ThreadletQueue;
+
typedef struct ThreadletWork
{
QTAILQ_ENTRY(ThreadletWork) node;
@@ -66,15 +81,10 @@ 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;
+/* Default ThreadletQueue */
+static ThreadletQueue globalqueue;
+static int globalqueue_init;
static pthread_attr_t attr;
-static int max_threads = 64;
-static int cur_threads = 0;
-static int idle_threads = 0;
-static QTAILQ_HEAD(, ThreadletWork) request_list;
#ifdef CONFIG_PREADV
static int preadv_present = 1;
@@ -93,32 +103,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)
{
@@ -311,42 +295,45 @@ static ssize_t handle_aiocb_rw(struct qemu_paiocb *aiocb)
static void *threadlet_worker(void *data)
{
+ ThreadletQueue *queue = data;
+ qemu_mutex_lock(&queue->lock);
while (1) {
- ssize_t ret = 0;
- qemu_timeval tv;
- struct timespec ts;
ThreadletWork *work;
+ int ret = 0;
- 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);
+ while (QTAILQ_EMPTY(&queue->request_list) &&
+ (ret != ETIMEDOUT)) {
+ /* wait for cond to be signalled or broadcast for 1000s */
+ ret = qemu_cond_timedwait((&queue->cond),
+ &(queue->lock), 10*100000);
}
- if (QTAILQ_EMPTY(&request_list)) {
- idle_threads--;
- cur_threads--;
- mutex_unlock(&lock);
- break;
- }
- work = QTAILQ_FIRST(&request_list);
- QTAILQ_REMOVE(&request_list, work, node);
- idle_threads--;
- mutex_unlock(&lock);
+ assert(queue->idle_threads != 0);
+ if (QTAILQ_EMPTY(&queue->request_list)) {
+ if (queue->cur_threads > queue->min_threads) {
+ /* We retain the minimum number of threads */
+ break;
+ }
+ } else {
+ work = QTAILQ_FIRST(&queue->request_list);
+ QTAILQ_REMOVE(&queue->request_list, work, node);
+
+ queue->idle_threads--;
+ qemu_mutex_unlock(&queue->lock);
- work->func(work);
- mutex_lock(&lock);
- idle_threads++;
- mutex_unlock(&lock);
+ /* execute the work function */
+ work->func(work);
+ qemu_mutex_lock(&queue->lock);
+ queue->idle_threads++;
+ }
}
+ queue->idle_threads--;
+ queue->cur_threads--;
+ qemu_mutex_unlock(&queue->lock);
+
return NULL;
}
@@ -389,18 +376,19 @@ static void handle_work(ThreadletWork *work)
}
}
-static void spawn_thread(void)
+static void spawn_threadlet(ThreadletQueue *queue)
{
+ pthread_t thread_id;
sigset_t set, oldset;
- cur_threads++;
- idle_threads++;
+ queue->cur_threads++;
+ queue->idle_threads++;
/* block all signals */
if (sigfillset(&set)) die("sigfillset");
if (sigprocmask(SIG_SETMASK, &set, &oldset)) die("sigprocmask");
- thread_create(&thread_id, &attr, threadlet_worker, NULL);
+ thread_create(&thread_id, &attr, threadlet_worker, queue);
if (sigprocmask(SIG_SETMASK, &oldset, NULL)) die("sigprocmask restore");
}
@@ -412,14 +400,29 @@ static void qemu_paio_submit(struct qemu_paiocb *aiocb)
aiocb->active = 0;
qemu_mutex_unlock(&aiocb_mutex);
- mutex_lock(&lock);
- if (idle_threads == 0 && cur_threads < max_threads)
- spawn_thread();
+ qemu_mutex_lock(&globalqueue.lock);
+
+ if (!globalqueue_init) {
+ globalqueue.cur_threads = 0;
+ globalqueue.idle_threads = 0;
+ globalqueue.max_threads = MAX_GLOBAL_THREADS;
+ globalqueue.min_threads = MIN_GLOBAL_THREADS;
+ QTAILQ_INIT(&globalqueue.request_list);
+ qemu_mutex_init(&globalqueue.lock);
+ qemu_cond_init(&globalqueue.cond);
+
+ globalqueue_init = 1;
+ }
+
+ if (globalqueue.idle_threads == 0 &&
+ globalqueue.cur_threads < globalqueue.max_threads)
+ spawn_threadlet(&globalqueue);
aiocb->work.func = handle_work;
- QTAILQ_INSERT_TAIL(&request_list, &aiocb->work, node);
- mutex_unlock(&lock);
- cond_signal(&cond);
+
+ QTAILQ_INSERT_TAIL(&globalqueue.request_list, &aiocb->work, node);
+ qemu_cond_signal(&globalqueue.cond);
+ qemu_mutex_unlock(&globalqueue.lock);
}
static ssize_t qemu_paio_return(struct qemu_paiocb *aiocb)
@@ -566,12 +569,14 @@ static void paio_cancel(BlockDriverAIOCB *blockacb)
int active = 0;
qemu_mutex_lock(&aiocb_mutex);
+ qemu_mutex_lock(&globalqueue.lock);
if (!acb->active) {
- QTAILQ_REMOVE(&request_list, &acb->work, node);
+ QTAILQ_REMOVE(&globalqueue.request_list, &acb->work, node);
acb->ret = -ECANCELED;
} else if (acb->ret == -EINPROGRESS) {
active = 1;
}
+ qemu_mutex_unlock(&globalqueue.lock);
if (!active) {
acb->ret = -ECANCELED;
@@ -689,8 +694,6 @@ int paio_init(void)
if (ret)
die2(ret, "pthread_attr_setdetachstate");
- QTAILQ_INIT(&request_list);
-
posix_aio_state = s;
return 0;
}
^ permalink raw reply related [flat|nested] 27+ messages in thread* [Qemu-devel] [PATCH 05/12] Threadlet: Add submit_work threadlet API.
2011-01-13 12:14 [Qemu-devel] [PATCH 00/12] Threadlets Infrastructure Arun R Bharadwaj
` (3 preceding siblings ...)
2011-01-13 12:14 ` [Qemu-devel] [PATCH 04/12] Add ThreadletQueue Arun R Bharadwaj
@ 2011-01-13 12:14 ` Arun R Bharadwaj
2011-01-13 12:14 ` [Qemu-devel] [PATCH 06/12] Threadlet: Add dequeue_work threadlet API and remove active field Arun R Bharadwaj
` (7 subsequent siblings)
12 siblings, 0 replies; 27+ messages in thread
From: Arun R Bharadwaj @ 2011-01-13 12:14 UTC (permalink / raw)
To: qemu-devel; +Cc: kwolf, aliguori, jvrao, aneesh.kumar
This patch adds submit work threadlet API and shows how
the qemu_paio_submit changes to submit_work.
Signed-off-by: Arun R Bharadwaj <arun@linux.vnet.ibm.com>
---
posix-aio-compat.c | 33 ++++++++++++++++++++++-----------
1 files changed, 22 insertions(+), 11 deletions(-)
diff --git a/posix-aio-compat.c b/posix-aio-compat.c
index b5d70c9..2ab1109 100644
--- a/posix-aio-compat.c
+++ b/posix-aio-compat.c
@@ -393,13 +393,13 @@ static void spawn_threadlet(ThreadletQueue *queue)
if (sigprocmask(SIG_SETMASK, &oldset, NULL)) die("sigprocmask restore");
}
-static void qemu_paio_submit(struct qemu_paiocb *aiocb)
+/**
+ * submit_work: Submit to the global queue a new task to be executed
+ * asynchronously.
+ * @work: Contains information about the task that needs to be submitted.
+ */
+static void submit_work(ThreadletWork *work)
{
- qemu_mutex_lock(&aiocb_mutex);
- aiocb->ret = -EINPROGRESS;
- aiocb->active = 0;
- qemu_mutex_unlock(&aiocb_mutex);
-
qemu_mutex_lock(&globalqueue.lock);
if (!globalqueue_init) {
@@ -415,13 +415,13 @@ static void qemu_paio_submit(struct qemu_paiocb *aiocb)
}
if (globalqueue.idle_threads == 0 &&
- globalqueue.cur_threads < globalqueue.max_threads)
+ globalqueue.cur_threads < globalqueue.max_threads) {
spawn_threadlet(&globalqueue);
- aiocb->work.func = handle_work;
-
- QTAILQ_INSERT_TAIL(&globalqueue.request_list, &aiocb->work, node);
- qemu_cond_signal(&globalqueue.cond);
+ } else {
+ qemu_cond_signal(&globalqueue.cond);
+ }
+ QTAILQ_INSERT_TAIL(&globalqueue.request_list, work, node);
qemu_mutex_unlock(&globalqueue.lock);
}
@@ -448,6 +448,17 @@ static int qemu_paio_error(struct qemu_paiocb *aiocb)
return ret;
}
+static void qemu_paio_submit(struct qemu_paiocb *aiocb)
+{
+ qemu_mutex_lock(&aiocb_mutex);
+ aiocb->ret = -EINPROGRESS;
+ aiocb->active = 0;
+ qemu_mutex_unlock(&aiocb_mutex);
+
+ aiocb->work.func = handle_work;
+ submit_work(&aiocb->work);
+}
+
static int posix_aio_process_queue(void *opaque)
{
PosixAioState *s = opaque;
^ permalink raw reply related [flat|nested] 27+ messages in thread* [Qemu-devel] [PATCH 06/12] Threadlet: Add dequeue_work threadlet API and remove active field.
2011-01-13 12:14 [Qemu-devel] [PATCH 00/12] Threadlets Infrastructure Arun R Bharadwaj
` (4 preceding siblings ...)
2011-01-13 12:14 ` [Qemu-devel] [PATCH 05/12] Threadlet: Add submit_work threadlet API Arun R Bharadwaj
@ 2011-01-13 12:14 ` Arun R Bharadwaj
2011-01-13 12:14 ` [Qemu-devel] [PATCH 07/12] Remove thread_create routine Arun R Bharadwaj
` (6 subsequent siblings)
12 siblings, 0 replies; 27+ messages in thread
From: Arun R Bharadwaj @ 2011-01-13 12:14 UTC (permalink / raw)
To: qemu-devel; +Cc: kwolf, aliguori, jvrao, aneesh.kumar
This patch adds dequeue_work threadlet API and shows how
the paio_cancel changes to dequeue_work.
The active field in the qemu_aiocb structure is now useless.
Remove it.
Signed-off-by: Arun R Bharadwaj <arun@linux.vnet.ibm.com>
---
posix-aio-compat.c | 38 ++++++++++++++++----------------------
1 files changed, 16 insertions(+), 22 deletions(-)
diff --git a/posix-aio-compat.c b/posix-aio-compat.c
index 2ab1109..2d73846 100644
--- a/posix-aio-compat.c
+++ b/posix-aio-compat.c
@@ -69,7 +69,6 @@ struct qemu_paiocb {
int aio_type;
ssize_t ret;
- int active;
struct qemu_paiocb *next;
int async_context_id;
@@ -346,7 +345,6 @@ static void handle_work(ThreadletWork *work)
pid = getpid();
qemu_mutex_lock(&aiocb_mutex);
aiocb = container_of(work, struct qemu_paiocb, work);
- aiocb->active = 1;
qemu_mutex_unlock(&aiocb_mutex);
switch (aiocb->aio_type & QEMU_AIO_TYPE_MASK) {
@@ -452,7 +450,6 @@ static void qemu_paio_submit(struct qemu_paiocb *aiocb)
{
qemu_mutex_lock(&aiocb_mutex);
aiocb->ret = -EINPROGRESS;
- aiocb->active = 0;
qemu_mutex_unlock(&aiocb_mutex);
aiocb->work.func = handle_work;
@@ -574,33 +571,30 @@ static void paio_remove(struct qemu_paiocb *acb)
}
}
-static void paio_cancel(BlockDriverAIOCB *blockacb)
+/**
+ * dequeue_work: Cancel a task queued on the global queue.
+ * @work: Contains the information of the task that needs to be cancelled.
+ */
+static int dequeue_work(ThreadletWork *work)
{
- struct qemu_paiocb *acb = (struct qemu_paiocb *)blockacb;
- int active = 0;
-
- qemu_mutex_lock(&aiocb_mutex);
qemu_mutex_lock(&globalqueue.lock);
- if (!acb->active) {
- QTAILQ_REMOVE(&globalqueue.request_list, &acb->work, node);
- acb->ret = -ECANCELED;
- } else if (acb->ret == -EINPROGRESS) {
- active = 1;
- }
+ QTAILQ_REMOVE(&globalqueue.request_list, work, node);
qemu_mutex_unlock(&globalqueue.lock);
- if (!active) {
- acb->ret = -ECANCELED;
- } else {
+ return 0;
+}
+
+static void paio_cancel(BlockDriverAIOCB *blockacb)
+{
+ struct qemu_paiocb *acb = (struct qemu_paiocb *)blockacb;
+ if (dequeue_work(&acb->work) != 0) {
+ /* Wait for running work item to complete */
+ qemu_mutex_lock(&aiocb_mutex);
while (acb->ret == -EINPROGRESS) {
- /*
- * fail safe: if the aio could not be canceled,
- * we wait for it
- */
qemu_cond_wait(&aiocb_completion, &aiocb_mutex);
}
+ qemu_mutex_unlock(&aiocb_mutex);
}
- qemu_mutex_unlock(&aiocb_mutex);
paio_remove(acb);
}
^ permalink raw reply related [flat|nested] 27+ messages in thread* [Qemu-devel] [PATCH 07/12] Remove thread_create routine.
2011-01-13 12:14 [Qemu-devel] [PATCH 00/12] Threadlets Infrastructure Arun R Bharadwaj
` (5 preceding siblings ...)
2011-01-13 12:14 ` [Qemu-devel] [PATCH 06/12] Threadlet: Add dequeue_work threadlet API and remove active field Arun R Bharadwaj
@ 2011-01-13 12:14 ` Arun R Bharadwaj
2011-01-13 12:14 ` [Qemu-devel] [PATCH 08/12] Threadlet: Add aio_signal_handler threadlet API Arun R Bharadwaj
` (5 subsequent siblings)
12 siblings, 0 replies; 27+ messages in thread
From: Arun R Bharadwaj @ 2011-01-13 12:14 UTC (permalink / raw)
To: qemu-devel; +Cc: kwolf, aliguori, jvrao, aneesh.kumar
Remove thread_create and use qemu_thread_create instead.
Signed-off-by: Arun R Bharadwaj <arun@linux.vnet.ibm.com>
---
posix-aio-compat.c | 29 ++---------------------------
1 files changed, 2 insertions(+), 27 deletions(-)
diff --git a/posix-aio-compat.c b/posix-aio-compat.c
index 2d73846..8c5bb46 100644
--- a/posix-aio-compat.c
+++ b/posix-aio-compat.c
@@ -13,7 +13,6 @@
#include <sys/ioctl.h>
#include <sys/types.h>
-#include <pthread.h>
#include <unistd.h>
#include <errno.h>
#include <time.h>
@@ -83,7 +82,6 @@ typedef struct PosixAioState {
/* Default ThreadletQueue */
static ThreadletQueue globalqueue;
static int globalqueue_init;
-static pthread_attr_t attr;
#ifdef CONFIG_PREADV
static int preadv_present = 1;
@@ -102,13 +100,6 @@ static void die(const char *what)
die2(errno, what);
}
-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;
@@ -376,19 +367,12 @@ static void handle_work(ThreadletWork *work)
static void spawn_threadlet(ThreadletQueue *queue)
{
- pthread_t thread_id;
- sigset_t set, oldset;
+ QemuThread thread;
queue->cur_threads++;
queue->idle_threads++;
- /* block all signals */
- if (sigfillset(&set)) die("sigfillset");
- if (sigprocmask(SIG_SETMASK, &set, &oldset)) die("sigprocmask");
-
- thread_create(&thread_id, &attr, threadlet_worker, queue);
-
- if (sigprocmask(SIG_SETMASK, &oldset, NULL)) die("sigprocmask restore");
+ qemu_thread_create(&thread, threadlet_worker, queue);
}
/**
@@ -661,7 +645,6 @@ int paio_init(void)
struct sigaction act;
PosixAioState *s;
int fds[2];
- int ret;
if (posix_aio_state)
return 0;
@@ -691,14 +674,6 @@ 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");
-
posix_aio_state = s;
return 0;
}
^ permalink raw reply related [flat|nested] 27+ messages in thread* [Qemu-devel] [PATCH 08/12] Threadlet: Add aio_signal_handler threadlet API
2011-01-13 12:14 [Qemu-devel] [PATCH 00/12] Threadlets Infrastructure Arun R Bharadwaj
` (6 preceding siblings ...)
2011-01-13 12:14 ` [Qemu-devel] [PATCH 07/12] Remove thread_create routine Arun R Bharadwaj
@ 2011-01-13 12:14 ` Arun R Bharadwaj
2011-01-17 9:56 ` Stefan Hajnoczi
2011-01-13 12:15 ` [Qemu-devel] [PATCH 09/12] Remove all instances of CONFIG_THREAD Arun R Bharadwaj
` (4 subsequent siblings)
12 siblings, 1 reply; 27+ messages in thread
From: Arun R Bharadwaj @ 2011-01-13 12:14 UTC (permalink / raw)
To: qemu-devel; +Cc: kwolf, aliguori, jvrao, aneesh.kumar
This patch adds aio_signal_handler threadlet API. Earler
posix-aio-compat.c had its own signal handler code. Now
abstract this, in the later patch it is moved to a generic
code so that it can be used by other subsystems.
Signed-off-by: Arun R Bharadwaj <arun@linux.vnet.ibm.com>
---
posix-aio-compat.c | 49 +++++++++++++++++++++++++++----------------------
qemu-thread.h | 1 +
vl.c | 3 +++
3 files changed, 31 insertions(+), 22 deletions(-)
diff --git a/posix-aio-compat.c b/posix-aio-compat.c
index 8c5bb46..94dd007 100644
--- a/posix-aio-compat.c
+++ b/posix-aio-compat.c
@@ -327,11 +327,14 @@ static void *threadlet_worker(void *data)
return NULL;
}
+static PosixAioState *posix_aio_state;
+
static void handle_work(ThreadletWork *work)
{
pid_t pid;
ssize_t ret = 0;
struct qemu_paiocb *aiocb;
+ char byte = 0;
pid = getpid();
qemu_mutex_lock(&aiocb_mutex);
@@ -360,11 +363,35 @@ static void handle_work(ThreadletWork *work)
qemu_cond_broadcast(&aiocb_completion);
qemu_mutex_unlock(&aiocb_mutex);
+ ret = write(posix_aio_state->wfd, &byte, sizeof(byte));
+ if (ret < 0 && errno != EAGAIN) {
+ die("write()");
+ }
+
if (kill(pid, aiocb->ev_signo)) {
die("kill failed");
}
}
+static void threadlet_io_completion_signal_handler(int signum)
+{
+ qemu_service_io();
+}
+
+static void threadlet_register_signal_handler(void)
+{
+ struct sigaction act;
+ sigfillset(&act.sa_mask);
+ act.sa_flags = 0; /* do not restart syscalls to interrupt select() */
+ act.sa_handler = threadlet_io_completion_signal_handler;
+ sigaction(SIGUSR2, &act, NULL);
+}
+
+void threadlet_init(void)
+{
+ threadlet_register_signal_handler();
+}
+
static void spawn_threadlet(ThreadletQueue *queue)
{
QemuThread thread;
@@ -520,22 +547,6 @@ static int posix_aio_flush(void *opaque)
return !!s->first_aio;
}
-static PosixAioState *posix_aio_state;
-
-static void aio_signal_handler(int signum)
-{
- if (posix_aio_state) {
- char byte = 0;
- ssize_t ret;
-
- ret = write(posix_aio_state->wfd, &byte, sizeof(byte));
- if (ret < 0 && errno != EAGAIN)
- die("write()");
- }
-
- qemu_service_io();
-}
-
static void paio_remove(struct qemu_paiocb *acb)
{
struct qemu_paiocb **pacb;
@@ -642,7 +653,6 @@ BlockDriverAIOCB *paio_ioctl(BlockDriverState *bs, int fd,
int paio_init(void)
{
- struct sigaction act;
PosixAioState *s;
int fds[2];
@@ -654,11 +664,6 @@ int paio_init(void)
s = qemu_malloc(sizeof(PosixAioState));
- sigfillset(&act.sa_mask);
- act.sa_flags = 0; /* do not restart syscalls to interrupt select() */
- act.sa_handler = aio_signal_handler;
- sigaction(SIGUSR2, &act, NULL);
-
s->first_aio = NULL;
if (qemu_pipe(fds) == -1) {
fprintf(stderr, "failed to create pipe\n");
diff --git a/qemu-thread.h b/qemu-thread.h
index 19bb30c..c5b579c 100644
--- a/qemu-thread.h
+++ b/qemu-thread.h
@@ -40,5 +40,6 @@ 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_exit(void *retval);
+void threadlet_init(void);
#endif
diff --git a/vl.c b/vl.c
index df414ef..aba805f 100644
--- a/vl.c
+++ b/vl.c
@@ -148,6 +148,7 @@ int main(int argc, char **argv)
#include "qemu-config.h"
#include "qemu-objects.h"
#include "qemu-options.h"
+#include "qemu-thread.h"
#ifdef CONFIG_VIRTFS
#include "fsdev/qemu-fsdev.h"
#endif
@@ -2922,6 +2923,8 @@ int main(int argc, char **argv, char **envp)
exit(1);
}
+ threadlet_init();
+
/* init generic devices */
if (qemu_opts_foreach(qemu_find_opts("device"), device_init_func, NULL, 1) != 0)
exit(1);
^ permalink raw reply related [flat|nested] 27+ messages in thread* Re: [Qemu-devel] [PATCH 08/12] Threadlet: Add aio_signal_handler threadlet API
2011-01-13 12:14 ` [Qemu-devel] [PATCH 08/12] Threadlet: Add aio_signal_handler threadlet API Arun R Bharadwaj
@ 2011-01-17 9:56 ` Stefan Hajnoczi
2011-01-17 15:54 ` [Qemu-devel] " Paolo Bonzini
2011-01-18 4:43 ` [Qemu-devel] " Arun R Bharadwaj
0 siblings, 2 replies; 27+ messages in thread
From: Stefan Hajnoczi @ 2011-01-17 9:56 UTC (permalink / raw)
To: Arun R Bharadwaj; +Cc: kwolf, aliguori, jvrao, qemu-devel, aneesh.kumar
On Thu, Jan 13, 2011 at 12:14 PM, Arun R Bharadwaj
<arun@linux.vnet.ibm.com> wrote:
> +static void threadlet_io_completion_signal_handler(int signum)
> +{
> + qemu_service_io();
> +}
> +
> +static void threadlet_register_signal_handler(void)
> +{
> + struct sigaction act;
> + sigfillset(&act.sa_mask);
> + act.sa_flags = 0; /* do not restart syscalls to interrupt select() */
> + act.sa_handler = threadlet_io_completion_signal_handler;
> + sigaction(SIGUSR2, &act, NULL);
> +}
> +
> +void threadlet_init(void)
> +{
> + threadlet_register_signal_handler();
> +}
This would be the right place to create qemu-threadlet.c, instead of
adding the thread_init() prototype to qemu-thread.h and then including
that in vl.c.
Stefan
^ permalink raw reply [flat|nested] 27+ messages in thread* [Qemu-devel] Re: [PATCH 08/12] Threadlet: Add aio_signal_handler threadlet API
2011-01-17 9:56 ` Stefan Hajnoczi
@ 2011-01-17 15:54 ` Paolo Bonzini
2011-01-17 17:38 ` Stefan Hajnoczi
2011-01-18 4:43 ` [Qemu-devel] " Arun R Bharadwaj
1 sibling, 1 reply; 27+ messages in thread
From: Paolo Bonzini @ 2011-01-17 15:54 UTC (permalink / raw)
To: Stefan Hajnoczi
Cc: kwolf, qemu-devel, aliguori, aneesh.kumar, Arun R Bharadwaj, jvrao
On 01/17/2011 10:56 AM, Stefan Hajnoczi wrote:
> This would be the right place to create qemu-threadlet.c,
You mean .h?
Paolo
^ permalink raw reply [flat|nested] 27+ messages in thread
* [Qemu-devel] Re: [PATCH 08/12] Threadlet: Add aio_signal_handler threadlet API
2011-01-17 15:54 ` [Qemu-devel] " Paolo Bonzini
@ 2011-01-17 17:38 ` Stefan Hajnoczi
0 siblings, 0 replies; 27+ messages in thread
From: Stefan Hajnoczi @ 2011-01-17 17:38 UTC (permalink / raw)
To: Paolo Bonzini
Cc: kwolf, qemu-devel, aliguori, aneesh.kumar, Arun R Bharadwaj, jvrao
On Mon, Jan 17, 2011 at 3:54 PM, Paolo Bonzini <pbonzini@redhat.com> wrote:
> On 01/17/2011 10:56 AM, Stefan Hajnoczi wrote:
>>
>> This would be the right place to create qemu-threadlet.c,
>
> You mean .h?
We need both qemu-threadlet.c and qemu-threadlet.h.
Stefan
^ permalink raw reply [flat|nested] 27+ messages in thread
* Re: [Qemu-devel] [PATCH 08/12] Threadlet: Add aio_signal_handler threadlet API
2011-01-17 9:56 ` Stefan Hajnoczi
2011-01-17 15:54 ` [Qemu-devel] " Paolo Bonzini
@ 2011-01-18 4:43 ` Arun R Bharadwaj
2011-01-18 6:31 ` Stefan Hajnoczi
1 sibling, 1 reply; 27+ messages in thread
From: Arun R Bharadwaj @ 2011-01-18 4:43 UTC (permalink / raw)
To: Stefan Hajnoczi; +Cc: kwolf, aliguori, jvrao, qemu-devel, aneesh.kumar
* Stefan Hajnoczi <stefanha@gmail.com> [2011-01-17 09:56:58]:
> On Thu, Jan 13, 2011 at 12:14 PM, Arun R Bharadwaj
> <arun@linux.vnet.ibm.com> wrote:
> > +static void threadlet_io_completion_signal_handler(int signum)
> > +{
> > + qemu_service_io();
> > +}
> > +
> > +static void threadlet_register_signal_handler(void)
> > +{
> > + struct sigaction act;
> > + sigfillset(&act.sa_mask);
> > + act.sa_flags = 0; /* do not restart syscalls to interrupt select() */
> > + act.sa_handler = threadlet_io_completion_signal_handler;
> > + sigaction(SIGUSR2, &act, NULL);
> > +}
> > +
> > +void threadlet_init(void)
> > +{
> > + threadlet_register_signal_handler();
> > +}
>
> This would be the right place to create qemu-threadlet.c, instead of
> adding the thread_init() prototype to qemu-thread.h and then including
> that in vl.c.
>
> Stefan
I did not follow your comment here. How can we avoid including
threadler_init() in vl.c?
-arun
^ permalink raw reply [flat|nested] 27+ messages in thread* Re: [Qemu-devel] [PATCH 08/12] Threadlet: Add aio_signal_handler threadlet API
2011-01-18 4:43 ` [Qemu-devel] " Arun R Bharadwaj
@ 2011-01-18 6:31 ` Stefan Hajnoczi
2011-01-18 6:46 ` Arun R Bharadwaj
0 siblings, 1 reply; 27+ messages in thread
From: Stefan Hajnoczi @ 2011-01-18 6:31 UTC (permalink / raw)
To: arun; +Cc: kwolf, aliguori, jvrao, qemu-devel, aneesh.kumar
On Tue, Jan 18, 2011 at 4:43 AM, Arun R Bharadwaj
<arun@linux.vnet.ibm.com> wrote:
> * Stefan Hajnoczi <stefanha@gmail.com> [2011-01-17 09:56:58]:
>
>> On Thu, Jan 13, 2011 at 12:14 PM, Arun R Bharadwaj
>> <arun@linux.vnet.ibm.com> wrote:
>> > +static void threadlet_io_completion_signal_handler(int signum)
>> > +{
>> > + qemu_service_io();
>> > +}
>> > +
>> > +static void threadlet_register_signal_handler(void)
>> > +{
>> > + struct sigaction act;
>> > + sigfillset(&act.sa_mask);
>> > + act.sa_flags = 0; /* do not restart syscalls to interrupt select() */
>> > + act.sa_handler = threadlet_io_completion_signal_handler;
>> > + sigaction(SIGUSR2, &act, NULL);
>> > +}
>> > +
>> > +void threadlet_init(void)
>> > +{
>> > + threadlet_register_signal_handler();
>> > +}
>>
>> This would be the right place to create qemu-threadlet.c, instead of
>> adding the thread_init() prototype to qemu-thread.h and then including
>> that in vl.c.
>>
>> Stefan
>
> I did not follow your comment here. How can we avoid including
> threadler_init() in vl.c?
Instead of adding threadlet_init() and related functions to
posix-aio-compat.c and adding the prototype to qemu-thread.h, why not
just create qemu-threadlet.c/qemu-threadlet.h and put these functions
there instead?
Stefan
^ permalink raw reply [flat|nested] 27+ messages in thread* Re: [Qemu-devel] [PATCH 08/12] Threadlet: Add aio_signal_handler threadlet API
2011-01-18 6:31 ` Stefan Hajnoczi
@ 2011-01-18 6:46 ` Arun R Bharadwaj
2011-01-18 7:14 ` Stefan Hajnoczi
0 siblings, 1 reply; 27+ messages in thread
From: Arun R Bharadwaj @ 2011-01-18 6:46 UTC (permalink / raw)
To: Stefan Hajnoczi; +Cc: kwolf, aliguori, jvrao, qemu-devel, aneesh.kumar
* Stefan Hajnoczi <stefanha@gmail.com> [2011-01-18 06:31:34]:
> On Tue, Jan 18, 2011 at 4:43 AM, Arun R Bharadwaj
> <arun@linux.vnet.ibm.com> wrote:
> > * Stefan Hajnoczi <stefanha@gmail.com> [2011-01-17 09:56:58]:
> >
> >> On Thu, Jan 13, 2011 at 12:14 PM, Arun R Bharadwaj
> >> <arun@linux.vnet.ibm.com> wrote:
> >> > +static void threadlet_io_completion_signal_handler(int signum)
> >> > +{
> >> > + qemu_service_io();
> >> > +}
> >> > +
> >> > +static void threadlet_register_signal_handler(void)
> >> > +{
> >> > + struct sigaction act;
> >> > + sigfillset(&act.sa_mask);
> >> > + act.sa_flags = 0; /* do not restart syscalls to interrupt select() */
> >> > + act.sa_handler = threadlet_io_completion_signal_handler;
> >> > + sigaction(SIGUSR2, &act, NULL);
> >> > +}
> >> > +
> >> > +void threadlet_init(void)
> >> > +{
> >> > + threadlet_register_signal_handler();
> >> > +}
> >>
> >> This would be the right place to create qemu-threadlet.c, instead of
> >> adding the thread_init() prototype to qemu-thread.h and then including
> >> that in vl.c.
> >>
> >> Stefan
> >
> > I did not follow your comment here. How can we avoid including
> > threadler_init() in vl.c?
>
> Instead of adding threadlet_init() and related functions to
> posix-aio-compat.c and adding the prototype to qemu-thread.h, why not
> just create qemu-threadlet.c/qemu-threadlet.h and put these functions
> there instead?
>
> Stefan
Got it. So you mean I merge patch 8 and patch 10 into a single patch.
But wouldn't this mean we are moving code and adding new API in the
same patch? Anthony did not want this from what I recall. But I can do
it if you feel it makes things simple.
-arun
^ permalink raw reply [flat|nested] 27+ messages in thread* Re: [Qemu-devel] [PATCH 08/12] Threadlet: Add aio_signal_handler threadlet API
2011-01-18 6:46 ` Arun R Bharadwaj
@ 2011-01-18 7:14 ` Stefan Hajnoczi
2011-01-18 18:10 ` Venkateswararao Jujjuri (JV)
0 siblings, 1 reply; 27+ messages in thread
From: Stefan Hajnoczi @ 2011-01-18 7:14 UTC (permalink / raw)
To: arun; +Cc: kwolf, aliguori, jvrao, qemu-devel, aneesh.kumar
On Tue, Jan 18, 2011 at 6:46 AM, Arun R Bharadwaj
<arun@linux.vnet.ibm.com> wrote:
> * Stefan Hajnoczi <stefanha@gmail.com> [2011-01-18 06:31:34]:
>
>> On Tue, Jan 18, 2011 at 4:43 AM, Arun R Bharadwaj
>> <arun@linux.vnet.ibm.com> wrote:
>> > * Stefan Hajnoczi <stefanha@gmail.com> [2011-01-17 09:56:58]:
>> >
>> >> On Thu, Jan 13, 2011 at 12:14 PM, Arun R Bharadwaj
>> >> <arun@linux.vnet.ibm.com> wrote:
>> >> > +static void threadlet_io_completion_signal_handler(int signum)
>> >> > +{
>> >> > + qemu_service_io();
>> >> > +}
>> >> > +
>> >> > +static void threadlet_register_signal_handler(void)
>> >> > +{
>> >> > + struct sigaction act;
>> >> > + sigfillset(&act.sa_mask);
>> >> > + act.sa_flags = 0; /* do not restart syscalls to interrupt select() */
>> >> > + act.sa_handler = threadlet_io_completion_signal_handler;
>> >> > + sigaction(SIGUSR2, &act, NULL);
>> >> > +}
>> >> > +
>> >> > +void threadlet_init(void)
>> >> > +{
>> >> > + threadlet_register_signal_handler();
>> >> > +}
>> >>
>> >> This would be the right place to create qemu-threadlet.c, instead of
>> >> adding the thread_init() prototype to qemu-thread.h and then including
>> >> that in vl.c.
>> >>
>> >> Stefan
>> >
>> > I did not follow your comment here. How can we avoid including
>> > threadler_init() in vl.c?
>>
>> Instead of adding threadlet_init() and related functions to
>> posix-aio-compat.c and adding the prototype to qemu-thread.h, why not
>> just create qemu-threadlet.c/qemu-threadlet.h and put these functions
>> there instead?
>>
>> Stefan
>
> Got it. So you mean I merge patch 8 and patch 10 into a single patch.
> But wouldn't this mean we are moving code and adding new API in the
> same patch? Anthony did not want this from what I recall. But I can do
> it if you feel it makes things simple.
I don't think you need to merge the patches. It's odd to add
functions to posix-aio-compat.c but put the prototype in qemu-thread.h
(and then include and call from vl.c). So for these three functions
(threadlet_init, threadlet_register_signal_handler, and
threadlet_io_completion_signal_handler) only I think it makes sense to
move them to qemu-threadlet.[ch] straight away.
It's not the end of the world if you don't want to do that. It just
jumped out at me when reading and I had to remember that it'll get
fixed up later.
Stefan
^ permalink raw reply [flat|nested] 27+ messages in thread* Re: [Qemu-devel] [PATCH 08/12] Threadlet: Add aio_signal_handler threadlet API
2011-01-18 7:14 ` Stefan Hajnoczi
@ 2011-01-18 18:10 ` Venkateswararao Jujjuri (JV)
2011-01-19 7:54 ` Stefan Hajnoczi
0 siblings, 1 reply; 27+ messages in thread
From: Venkateswararao Jujjuri (JV) @ 2011-01-18 18:10 UTC (permalink / raw)
To: Stefan Hajnoczi; +Cc: kwolf, arun, qemu-devel, aneesh.kumar, aliguori
On 1/17/2011 11:14 PM, Stefan Hajnoczi wrote:
> On Tue, Jan 18, 2011 at 6:46 AM, Arun R Bharadwaj
> <arun@linux.vnet.ibm.com> wrote:
>> * Stefan Hajnoczi <stefanha@gmail.com> [2011-01-18 06:31:34]:
>>
>>> On Tue, Jan 18, 2011 at 4:43 AM, Arun R Bharadwaj
>>> <arun@linux.vnet.ibm.com> wrote:
>>>> * Stefan Hajnoczi <stefanha@gmail.com> [2011-01-17 09:56:58]:
>>>>
>>>>> On Thu, Jan 13, 2011 at 12:14 PM, Arun R Bharadwaj
>>>>> <arun@linux.vnet.ibm.com> wrote:
>>>>>> +static void threadlet_io_completion_signal_handler(int signum)
>>>>>> +{
>>>>>> + qemu_service_io();
>>>>>> +}
>>>>>> +
>>>>>> +static void threadlet_register_signal_handler(void)
>>>>>> +{
>>>>>> + struct sigaction act;
>>>>>> + sigfillset(&act.sa_mask);
>>>>>> + act.sa_flags = 0; /* do not restart syscalls to interrupt select() */
>>>>>> + act.sa_handler = threadlet_io_completion_signal_handler;
>>>>>> + sigaction(SIGUSR2, &act, NULL);
>>>>>> +}
>>>>>> +
>>>>>> +void threadlet_init(void)
>>>>>> +{
>>>>>> + threadlet_register_signal_handler();
>>>>>> +}
>>>>>
>>>>> This would be the right place to create qemu-threadlet.c, instead of
>>>>> adding the thread_init() prototype to qemu-thread.h and then including
>>>>> that in vl.c.
>>>>>
>>>>> Stefan
>>>>
>>>> I did not follow your comment here. How can we avoid including
>>>> threadler_init() in vl.c?
>>>
>>> Instead of adding threadlet_init() and related functions to
>>> posix-aio-compat.c and adding the prototype to qemu-thread.h, why not
>>> just create qemu-threadlet.c/qemu-threadlet.h and put these functions
>>> there instead?
>>>
>>> Stefan
>>
>> Got it. So you mean I merge patch 8 and patch 10 into a single patch.
>> But wouldn't this mean we are moving code and adding new API in the
>> same patch? Anthony did not want this from what I recall. But I can do
>> it if you feel it makes things simple.
>
> I don't think you need to merge the patches. It's odd to add
> functions to posix-aio-compat.c but put the prototype in qemu-thread.h
> (and then include and call from vl.c). So for these three functions
> (threadlet_init, threadlet_register_signal_handler, and
> threadlet_io_completion_signal_handler) only I think it makes sense to
> move them to qemu-threadlet.[ch] straight away.
So basically create the new file qemu-threadlet.[ch] with only these functions
and move the rest of the code in patch 10(as we do now).
>
> It's not the end of the world if you don't want to do that. It just
> jumped out at me when reading and I had to remember that it'll get
> fixed up later.
Yes; as things are settling down towards the end of this series; I guess it is
OK if you decide
not to alter the sequence of the patches and avoid merges.
Thanks,
JV
>
> Stefan
>
^ permalink raw reply [flat|nested] 27+ messages in thread* Re: [Qemu-devel] [PATCH 08/12] Threadlet: Add aio_signal_handler threadlet API
2011-01-18 18:10 ` Venkateswararao Jujjuri (JV)
@ 2011-01-19 7:54 ` Stefan Hajnoczi
0 siblings, 0 replies; 27+ messages in thread
From: Stefan Hajnoczi @ 2011-01-19 7:54 UTC (permalink / raw)
To: Venkateswararao Jujjuri (JV)
Cc: kwolf, arun, qemu-devel, aneesh.kumar, aliguori
On Tue, Jan 18, 2011 at 6:10 PM, Venkateswararao Jujjuri (JV)
<jvrao@linux.vnet.ibm.com> wrote:
> On 1/17/2011 11:14 PM, Stefan Hajnoczi wrote:
>> On Tue, Jan 18, 2011 at 6:46 AM, Arun R Bharadwaj
>> <arun@linux.vnet.ibm.com> wrote:
>>> * Stefan Hajnoczi <stefanha@gmail.com> [2011-01-18 06:31:34]:
>>>
>>>> On Tue, Jan 18, 2011 at 4:43 AM, Arun R Bharadwaj
>>>> <arun@linux.vnet.ibm.com> wrote:
>>>>> * Stefan Hajnoczi <stefanha@gmail.com> [2011-01-17 09:56:58]:
>>>>>
>>>>>> On Thu, Jan 13, 2011 at 12:14 PM, Arun R Bharadwaj
>>>>>> <arun@linux.vnet.ibm.com> wrote:
>>>>>>> +static void threadlet_io_completion_signal_handler(int signum)
>>>>>>> +{
>>>>>>> + qemu_service_io();
>>>>>>> +}
>>>>>>> +
>>>>>>> +static void threadlet_register_signal_handler(void)
>>>>>>> +{
>>>>>>> + struct sigaction act;
>>>>>>> + sigfillset(&act.sa_mask);
>>>>>>> + act.sa_flags = 0; /* do not restart syscalls to interrupt select() */
>>>>>>> + act.sa_handler = threadlet_io_completion_signal_handler;
>>>>>>> + sigaction(SIGUSR2, &act, NULL);
>>>>>>> +}
>>>>>>> +
>>>>>>> +void threadlet_init(void)
>>>>>>> +{
>>>>>>> + threadlet_register_signal_handler();
>>>>>>> +}
>>>>>>
>>>>>> This would be the right place to create qemu-threadlet.c, instead of
>>>>>> adding the thread_init() prototype to qemu-thread.h and then including
>>>>>> that in vl.c.
>>>>>>
>>>>>> Stefan
>>>>>
>>>>> I did not follow your comment here. How can we avoid including
>>>>> threadler_init() in vl.c?
>>>>
>>>> Instead of adding threadlet_init() and related functions to
>>>> posix-aio-compat.c and adding the prototype to qemu-thread.h, why not
>>>> just create qemu-threadlet.c/qemu-threadlet.h and put these functions
>>>> there instead?
>>>>
>>>> Stefan
>>>
>>> Got it. So you mean I merge patch 8 and patch 10 into a single patch.
>>> But wouldn't this mean we are moving code and adding new API in the
>>> same patch? Anthony did not want this from what I recall. But I can do
>>> it if you feel it makes things simple.
>>
>> I don't think you need to merge the patches. It's odd to add
>> functions to posix-aio-compat.c but put the prototype in qemu-thread.h
>> (and then include and call from vl.c). So for these three functions
>> (threadlet_init, threadlet_register_signal_handler, and
>> threadlet_io_completion_signal_handler) only I think it makes sense to
>> move them to qemu-threadlet.[ch] straight away.
>
> So basically create the new file qemu-threadlet.[ch] with only these functions
> and move the rest of the code in patch 10(as we do now).
Exactly.
Stefan
^ permalink raw reply [flat|nested] 27+ messages in thread
* [Qemu-devel] [PATCH 09/12] Remove all instances of CONFIG_THREAD
2011-01-13 12:14 [Qemu-devel] [PATCH 00/12] Threadlets Infrastructure Arun R Bharadwaj
` (7 preceding siblings ...)
2011-01-13 12:14 ` [Qemu-devel] [PATCH 08/12] Threadlet: Add aio_signal_handler threadlet API Arun R Bharadwaj
@ 2011-01-13 12:15 ` Arun R Bharadwaj
2011-01-13 12:15 ` [Qemu-devel] [PATCH 10/12] Move threadlet code to qemu-threadlets.c Arun R Bharadwaj
` (3 subsequent siblings)
12 siblings, 0 replies; 27+ messages in thread
From: Arun R Bharadwaj @ 2011-01-13 12:15 UTC (permalink / raw)
To: qemu-devel; +Cc: kwolf, aliguori, jvrao, aneesh.kumar
Remove all instances of CONFIG_THREAD, which is not used
anymore.
Signed-off-by: Arun R Bharadwaj <arun@linux.vnet.ibm.com>
---
configure | 2 --
1 files changed, 0 insertions(+), 2 deletions(-)
diff --git a/configure b/configure
index a079a49..addf733 100755
--- a/configure
+++ b/configure
@@ -2456,7 +2456,6 @@ if test "$vnc_png" != "no" ; then
fi
if test "$vnc_thread" != "no" ; then
echo "CONFIG_VNC_THREAD=y" >> $config_host_mak
- echo "CONFIG_THREAD=y" >> $config_host_mak
fi
if test "$fnmatch" = "yes" ; then
echo "CONFIG_FNMATCH=y" >> $config_host_mak
@@ -2534,7 +2533,6 @@ if test "$xen" = "yes" ; then
fi
if test "$io_thread" = "yes" ; then
echo "CONFIG_IOTHREAD=y" >> $config_host_mak
- echo "CONFIG_THREAD=y" >> $config_host_mak
fi
if test "$linux_aio" = "yes" ; then
echo "CONFIG_LINUX_AIO=y" >> $config_host_mak
^ permalink raw reply related [flat|nested] 27+ messages in thread* [Qemu-devel] [PATCH 10/12] Move threadlet code to qemu-threadlets.c
2011-01-13 12:14 [Qemu-devel] [PATCH 00/12] Threadlets Infrastructure Arun R Bharadwaj
` (8 preceding siblings ...)
2011-01-13 12:15 ` [Qemu-devel] [PATCH 09/12] Remove all instances of CONFIG_THREAD Arun R Bharadwaj
@ 2011-01-13 12:15 ` Arun R Bharadwaj
2011-01-17 9:57 ` Stefan Hajnoczi
2011-01-13 12:15 ` [Qemu-devel] [PATCH 11/12] Threadlets: Add functionality to create private queues Arun R Bharadwaj
` (2 subsequent siblings)
12 siblings, 1 reply; 27+ messages in thread
From: Arun R Bharadwaj @ 2011-01-13 12:15 UTC (permalink / raw)
To: qemu-devel; +Cc: kwolf, aliguori, jvrao, aneesh.kumar
This patch moves the threadlet queue API code to
qemu-threadlets.c where these APIs can be used by
other subsystems.
Signed-off-by: Arun R Bharadwaj <arun@linux.vnet.ibm.com>
---
Makefile.objs | 1
posix-aio-compat.c | 144 ----------------------------------------------------
qemu-thread.h | 1
qemu-threadlets.c | 142 +++++++++++++++++++++++++++++++++++++++++++++++++++
qemu-threadlets.h | 43 ++++++++++++++++
vl.c | 2 -
6 files changed, 188 insertions(+), 145 deletions(-)
create mode 100644 qemu-threadlets.c
create mode 100644 qemu-threadlets.h
diff --git a/Makefile.objs b/Makefile.objs
index 3b7ec27..2cf8aba 100644
--- a/Makefile.objs
+++ b/Makefile.objs
@@ -10,6 +10,7 @@ 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-$(CONFIG_POSIX) += qemu-thread.o
+block-obj-$(CONFIG_POSIX) += qemu-threadlets.o
block-obj-$(CONFIG_POSIX) += posix-aio-compat.o
block-obj-$(CONFIG_LINUX_AIO) += linux-aio.o
diff --git a/posix-aio-compat.c b/posix-aio-compat.c
index 94dd007..df55ce6 100644
--- a/posix-aio-compat.c
+++ b/posix-aio-compat.c
@@ -26,33 +26,13 @@
#include "qemu-common.h"
#include "trace.h"
#include "block_int.h"
-#include "qemu-thread.h"
+#include "qemu-threadlets.h"
#include "block/raw-posix-aio.h"
-#define MAX_GLOBAL_THREADS 64
-#define MIN_GLOBAL_THREADS 8
-
static QemuMutex aiocb_mutex;
static QemuCond aiocb_completion;
-typedef struct ThreadletQueue
-{
- QemuMutex lock;
- QemuCond cond;
- int max_threads;
- int min_threads;
- int cur_threads;
- int idle_threads;
- QTAILQ_HEAD(, ThreadletWork) request_list;
-} ThreadletQueue;
-
-typedef struct ThreadletWork
-{
- QTAILQ_ENTRY(ThreadletWork) node;
- void (*func)(struct ThreadletWork *work);
-} ThreadletWork;
-
struct qemu_paiocb {
BlockDriverAIOCB common;
int aio_fildes;
@@ -79,10 +59,6 @@ typedef struct PosixAioState {
struct qemu_paiocb *first_aio;
} PosixAioState;
-/* Default ThreadletQueue */
-static ThreadletQueue globalqueue;
-static int globalqueue_init;
-
#ifdef CONFIG_PREADV
static int preadv_present = 1;
#else
@@ -283,50 +259,6 @@ static ssize_t handle_aiocb_rw(struct qemu_paiocb *aiocb)
return nbytes;
}
-static void *threadlet_worker(void *data)
-{
- ThreadletQueue *queue = data;
-
- qemu_mutex_lock(&queue->lock);
- while (1) {
- ThreadletWork *work;
- int ret = 0;
-
- while (QTAILQ_EMPTY(&queue->request_list) &&
- (ret != ETIMEDOUT)) {
- /* wait for cond to be signalled or broadcast for 1000s */
- ret = qemu_cond_timedwait((&queue->cond),
- &(queue->lock), 10*100000);
- }
-
- assert(queue->idle_threads != 0);
- if (QTAILQ_EMPTY(&queue->request_list)) {
- if (queue->cur_threads > queue->min_threads) {
- /* We retain the minimum number of threads */
- break;
- }
- } else {
- 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);
-
- qemu_mutex_lock(&queue->lock);
- queue->idle_threads++;
- }
- }
-
- queue->idle_threads--;
- queue->cur_threads--;
- qemu_mutex_unlock(&queue->lock);
-
- return NULL;
-}
-
static PosixAioState *posix_aio_state;
static void handle_work(ThreadletWork *work)
@@ -373,67 +305,6 @@ static void handle_work(ThreadletWork *work)
}
}
-static void threadlet_io_completion_signal_handler(int signum)
-{
- qemu_service_io();
-}
-
-static void threadlet_register_signal_handler(void)
-{
- struct sigaction act;
- sigfillset(&act.sa_mask);
- act.sa_flags = 0; /* do not restart syscalls to interrupt select() */
- act.sa_handler = threadlet_io_completion_signal_handler;
- sigaction(SIGUSR2, &act, NULL);
-}
-
-void threadlet_init(void)
-{
- threadlet_register_signal_handler();
-}
-
-static void spawn_threadlet(ThreadletQueue *queue)
-{
- QemuThread thread;
-
- queue->cur_threads++;
- queue->idle_threads++;
-
- qemu_thread_create(&thread, threadlet_worker, queue);
-}
-
-/**
- * submit_work: Submit to the global queue a new task to be executed
- * asynchronously.
- * @work: Contains information about the task that needs to be submitted.
- */
-static void submit_work(ThreadletWork *work)
-{
- qemu_mutex_lock(&globalqueue.lock);
-
- if (!globalqueue_init) {
- globalqueue.cur_threads = 0;
- globalqueue.idle_threads = 0;
- globalqueue.max_threads = MAX_GLOBAL_THREADS;
- globalqueue.min_threads = MIN_GLOBAL_THREADS;
- QTAILQ_INIT(&globalqueue.request_list);
- qemu_mutex_init(&globalqueue.lock);
- qemu_cond_init(&globalqueue.cond);
-
- globalqueue_init = 1;
- }
-
- if (globalqueue.idle_threads == 0 &&
- globalqueue.cur_threads < globalqueue.max_threads) {
- spawn_threadlet(&globalqueue);
-
- } else {
- qemu_cond_signal(&globalqueue.cond);
- }
- QTAILQ_INSERT_TAIL(&globalqueue.request_list, work, node);
- qemu_mutex_unlock(&globalqueue.lock);
-}
-
static ssize_t qemu_paio_return(struct qemu_paiocb *aiocb)
{
ssize_t ret;
@@ -566,19 +437,6 @@ static void paio_remove(struct qemu_paiocb *acb)
}
}
-/**
- * dequeue_work: Cancel a task queued on the global queue.
- * @work: Contains the information of the task that needs to be cancelled.
- */
-static int dequeue_work(ThreadletWork *work)
-{
- qemu_mutex_lock(&globalqueue.lock);
- QTAILQ_REMOVE(&globalqueue.request_list, work, node);
- qemu_mutex_unlock(&globalqueue.lock);
-
- return 0;
-}
-
static void paio_cancel(BlockDriverAIOCB *blockacb)
{
struct qemu_paiocb *acb = (struct qemu_paiocb *)blockacb;
diff --git a/qemu-thread.h b/qemu-thread.h
index c5b579c..19bb30c 100644
--- a/qemu-thread.h
+++ b/qemu-thread.h
@@ -40,6 +40,5 @@ 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_exit(void *retval);
-void threadlet_init(void);
#endif
diff --git a/qemu-threadlets.c b/qemu-threadlets.c
new file mode 100644
index 0000000..42dd3d1
--- /dev/null
+++ b/qemu-threadlets.c
@@ -0,0 +1,142 @@
+/*
+ * Threadlet support for offloading tasks to be executed asynchronously
+ *
+ * Copyright IBM, Corp. 2008
+ * Copyright IBM, Corp. 2010
+ *
+ * Authors:
+ * Anthony Liguori <aliguori@us.ibm.com>
+ * Aneesh Kumar K.V <aneesh.kumar@linux.vnet.ibm.com>
+ * Gautham R Shenoy <gautham.shenoy@gmail.com>
+ * Arun R Bharadwaj <arun@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 <signal.h>
+
+#include "qemu-threadlets.h"
+#include "osdep.h"
+
+#define MAX_GLOBAL_THREADS 64
+#define MIN_GLOBAL_THREADS 8
+
+static ThreadletQueue globalqueue;
+static int globalqueue_init;
+
+static void threadlet_io_completion_signal_handler(int signum)
+{
+ qemu_service_io();
+}
+
+static void threadlet_register_signal_handler(void)
+{
+ struct sigaction act;
+ sigfillset(&act.sa_mask);
+ act.sa_flags = 0; /* do not restart syscalls to interrupt select() */
+ act.sa_handler = threadlet_io_completion_signal_handler;
+ sigaction(SIGUSR2, &act, NULL);
+}
+
+void threadlet_init(void)
+{
+ threadlet_register_signal_handler();
+}
+
+static void *threadlet_worker(void *data)
+{
+ ThreadletQueue *queue = data;
+
+ qemu_mutex_lock(&queue->lock);
+ while (1) {
+ ThreadletWork *work;
+ int ret = 0;
+
+ while (QTAILQ_EMPTY(&queue->request_list) &&
+ (ret != ETIMEDOUT)) {
+ /* wait for cond to be signalled or broadcast for 1000s */
+ ret = qemu_cond_timedwait((&queue->cond),
+ &(queue->lock), 10*100000);
+ }
+
+ assert(queue->idle_threads != 0);
+ if (QTAILQ_EMPTY(&queue->request_list)) {
+ if (queue->cur_threads > queue->min_threads) {
+ /* We retain the minimum number of threads */
+ break;
+ }
+ } else {
+ 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);
+
+ qemu_mutex_lock(&queue->lock);
+ queue->idle_threads++;
+ }
+ }
+
+ queue->idle_threads--;
+ queue->cur_threads--;
+ qemu_mutex_unlock(&queue->lock);
+
+ return NULL;
+}
+
+static void spawn_threadlet(ThreadletQueue *queue)
+{
+ QemuThread thread;
+
+ queue->cur_threads++;
+ queue->idle_threads++;
+
+ qemu_thread_create(&thread, threadlet_worker, queue);
+}
+
+/**
+ * submit_work: Submit to the global queue a new task to be executed
+ * asynchronously.
+ * @work: Contains information about the task that needs to be submitted.
+ */
+void submit_work(ThreadletWork *work)
+{
+ if (!globalqueue_init) {
+ globalqueue.cur_threads = 0;
+ globalqueue.idle_threads = 0;
+ globalqueue.max_threads = MAX_GLOBAL_THREADS;
+ globalqueue.min_threads = MIN_GLOBAL_THREADS;
+ QTAILQ_INIT(&globalqueue.request_list);
+ qemu_mutex_init(&globalqueue.lock);
+ qemu_cond_init(&globalqueue.cond);
+
+ globalqueue_init = 1;
+ }
+
+ qemu_mutex_lock(&globalqueue.lock);
+ if (globalqueue.idle_threads == 0 &&
+ globalqueue.cur_threads < globalqueue.max_threads) {
+ spawn_threadlet(&globalqueue);
+ } else {
+ qemu_cond_signal(&globalqueue.cond);
+ }
+ QTAILQ_INSERT_TAIL(&globalqueue.request_list, work, node);
+ qemu_mutex_unlock(&globalqueue.lock);
+}
+
+/**
+ * dequeue_work: Cancel a task queued on the global queue.
+ * @work: Contains the information of the task that needs to be cancelled.
+ */
+int dequeue_work(ThreadletWork *work)
+{
+ qemu_mutex_lock(&globalqueue.lock);
+ QTAILQ_REMOVE(&globalqueue.request_list, work, node);
+ qemu_mutex_unlock(&globalqueue.lock);
+
+ return 0;
+}
diff --git a/qemu-threadlets.h b/qemu-threadlets.h
new file mode 100644
index 0000000..03bb86b
--- /dev/null
+++ b/qemu-threadlets.h
@@ -0,0 +1,43 @@
+/*
+ * Threadlet support for offloading tasks to be executed asynchronously
+ *
+ * Copyright IBM, Corp. 2008
+ * Copyright IBM, Corp. 2010
+ *
+ * Authors:
+ * Anthony Liguori <aliguori@us.ibm.com>
+ * Aneesh Kumar K.V <aneesh.kumar@linux.vnet.ibm.com>
+ * Gautham R Shenoy <gautham.shenoy@gmail.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"
+
+typedef struct ThreadletQueue
+{
+ QemuMutex lock;
+ QemuCond cond;
+ int max_threads;
+ int min_threads;
+ int cur_threads;
+ int idle_threads;
+ QTAILQ_HEAD(, ThreadletWork) request_list;
+} ThreadletQueue;
+
+typedef struct ThreadletWork
+{
+ QTAILQ_ENTRY(ThreadletWork) node;
+ void (*func)(struct ThreadletWork *work);
+} ThreadletWork;
+
+void submit_work(ThreadletWork *work);
+int dequeue_work(ThreadletWork *work);
+void threadlet_init(void);
+#endif
diff --git a/vl.c b/vl.c
index aba805f..7b9a425 100644
--- a/vl.c
+++ b/vl.c
@@ -148,7 +148,7 @@ int main(int argc, char **argv)
#include "qemu-config.h"
#include "qemu-objects.h"
#include "qemu-options.h"
-#include "qemu-thread.h"
+#include "qemu-threadlets.h"
#ifdef CONFIG_VIRTFS
#include "fsdev/qemu-fsdev.h"
#endif
^ permalink raw reply related [flat|nested] 27+ messages in thread* Re: [Qemu-devel] [PATCH 10/12] Move threadlet code to qemu-threadlets.c
2011-01-13 12:15 ` [Qemu-devel] [PATCH 10/12] Move threadlet code to qemu-threadlets.c Arun R Bharadwaj
@ 2011-01-17 9:57 ` Stefan Hajnoczi
2011-01-18 4:37 ` Arun R Bharadwaj
0 siblings, 1 reply; 27+ messages in thread
From: Stefan Hajnoczi @ 2011-01-17 9:57 UTC (permalink / raw)
To: Arun R Bharadwaj; +Cc: kwolf, aliguori, jvrao, qemu-devel, aneesh.kumar
On Thu, Jan 13, 2011 at 12:15 PM, Arun R Bharadwaj
<arun@linux.vnet.ibm.com> wrote:
> This patch moves the threadlet queue API code to
> qemu-threadlets.c where these APIs can be used by
> other subsystems.
qemu-threadlet.c would be consistent with qemu-thread.c (not qemu-threads.c).
Stefan
^ permalink raw reply [flat|nested] 27+ messages in thread
* Re: [Qemu-devel] [PATCH 10/12] Move threadlet code to qemu-threadlets.c
2011-01-17 9:57 ` Stefan Hajnoczi
@ 2011-01-18 4:37 ` Arun R Bharadwaj
0 siblings, 0 replies; 27+ messages in thread
From: Arun R Bharadwaj @ 2011-01-18 4:37 UTC (permalink / raw)
To: Stefan Hajnoczi; +Cc: kwolf, aliguori, jvrao, qemu-devel, aneesh.kumar
* Stefan Hajnoczi <stefanha@gmail.com> [2011-01-17 09:57:45]:
> On Thu, Jan 13, 2011 at 12:15 PM, Arun R Bharadwaj
> <arun@linux.vnet.ibm.com> wrote:
> > This patch moves the threadlet queue API code to
> > qemu-threadlets.c where these APIs can be used by
> > other subsystems.
>
> qemu-threadlet.c would be consistent with qemu-thread.c (not qemu-threads.c).
>
> Stefan
Sure. Makes sense.
^ permalink raw reply [flat|nested] 27+ messages in thread
* [Qemu-devel] [PATCH 11/12] Threadlets: Add functionality to create private queues.
2011-01-13 12:14 [Qemu-devel] [PATCH 00/12] Threadlets Infrastructure Arun R Bharadwaj
` (9 preceding siblings ...)
2011-01-13 12:15 ` [Qemu-devel] [PATCH 10/12] Move threadlet code to qemu-threadlets.c Arun R Bharadwaj
@ 2011-01-13 12:15 ` Arun R Bharadwaj
2011-01-13 12:15 ` [Qemu-devel] [PATCH 12/12] Threadlets: Add documentation Arun R Bharadwaj
2011-01-17 10:32 ` [Qemu-devel] [PATCH 00/12] Threadlets Infrastructure Stefan Hajnoczi
12 siblings, 0 replies; 27+ messages in thread
From: Arun R Bharadwaj @ 2011-01-13 12:15 UTC (permalink / raw)
To: qemu-devel; +Cc: kwolf, aliguori, jvrao, aneesh.kumar
This patch allows subsystems to create their own private
queues with associated pools of threads.
Signed-off-by: Arun R Bharadwaj <arun@linux.vnet.ibm.com>
---
qemu-threadlets.c | 75 ++++++++++++++++++++++++++++++++++++++---------------
qemu-threadlets.h | 4 +++
2 files changed, 58 insertions(+), 21 deletions(-)
diff --git a/qemu-threadlets.c b/qemu-threadlets.c
index 42dd3d1..4df79b8 100644
--- a/qemu-threadlets.c
+++ b/qemu-threadlets.c
@@ -99,6 +99,25 @@ static void spawn_threadlet(ThreadletQueue *queue)
}
/**
+ * submit_work_to_queue: Submit a new task to a private queue to be
+ * executed asynchronously.
+ * @queue: Per-subsystem private queue to which the new task needs
+ * to be submitted.
+ * @work: Contains information about the task that needs to be submitted.
+ */
+void submit_work_to_queue(ThreadletQueue *queue, ThreadletWork *work)
+{
+ qemu_mutex_lock(&queue->lock);
+ if (queue->idle_threads == 0 && queue->cur_threads < queue->max_threads) {
+ spawn_threadlet(queue);
+ } else {
+ qemu_cond_signal(&queue->cond);
+ }
+ QTAILQ_INSERT_TAIL(&queue->request_list, work, node);
+ qemu_mutex_unlock(&queue->lock);
+}
+
+/**
* submit_work: Submit to the global queue a new task to be executed
* asynchronously.
* @work: Contains information about the task that needs to be submitted.
@@ -106,26 +125,26 @@ static void spawn_threadlet(ThreadletQueue *queue)
void submit_work(ThreadletWork *work)
{
if (!globalqueue_init) {
- globalqueue.cur_threads = 0;
- globalqueue.idle_threads = 0;
- globalqueue.max_threads = MAX_GLOBAL_THREADS;
- globalqueue.min_threads = MIN_GLOBAL_THREADS;
- QTAILQ_INIT(&globalqueue.request_list);
- qemu_mutex_init(&globalqueue.lock);
- qemu_cond_init(&globalqueue.cond);
-
+ threadlet_queue_init(&globalqueue, MAX_GLOBAL_THREADS,
+ MIN_GLOBAL_THREADS);
globalqueue_init = 1;
}
- qemu_mutex_lock(&globalqueue.lock);
- if (globalqueue.idle_threads == 0 &&
- globalqueue.cur_threads < globalqueue.max_threads) {
- spawn_threadlet(&globalqueue);
- } else {
- qemu_cond_signal(&globalqueue.cond);
- }
- QTAILQ_INSERT_TAIL(&globalqueue.request_list, work, node);
- qemu_mutex_unlock(&globalqueue.lock);
+ submit_work_to_queue(&globalqueue, work);
+}
+
+/**
+ * dequeue_work_on_queue: Cancel a task queued on a Queue.
+ * @queue: The queue containing the task to be cancelled.
+ * @work: Contains the information of the task that needs to be cancelled.
+ */
+int dequeue_work_on_queue(ThreadletQueue *queue, ThreadletWork *work)
+{
+ qemu_mutex_lock(&queue->lock);
+ QTAILQ_REMOVE(&queue->request_list, work, node);
+ qemu_mutex_unlock(&queue->lock);
+
+ return 0;
}
/**
@@ -134,9 +153,23 @@ void submit_work(ThreadletWork *work)
*/
int dequeue_work(ThreadletWork *work)
{
- qemu_mutex_lock(&globalqueue.lock);
- QTAILQ_REMOVE(&globalqueue.request_list, work, node);
- qemu_mutex_unlock(&globalqueue.lock);
+ return dequeue_work_on_queue(&globalqueue, work);
+}
- return 0;
+/**
+ * threadlet_queue_init: Initialize a threadlet queue.
+ * @queue: The threadlet queue to be initialized.
+ * @max_threads: Maximum number of threads processing the queue.
+ * @min_threads: Minimum number of threads processing the queue.
+ */
+void threadlet_queue_init(ThreadletQueue *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);
+ qemu_mutex_init(&queue->lock);
+ qemu_cond_init(&queue->cond);
}
diff --git a/qemu-threadlets.h b/qemu-threadlets.h
index 03bb86b..993d7ab 100644
--- a/qemu-threadlets.h
+++ b/qemu-threadlets.h
@@ -37,7 +37,11 @@ typedef struct ThreadletWork
void (*func)(struct ThreadletWork *work);
} ThreadletWork;
+void submit_work_to_queue(ThreadletQueue *queue, ThreadletWork *work);
void submit_work(ThreadletWork *work);
+int dequeue_work_on_queue(ThreadletQueue *queue, ThreadletWork *work);
int dequeue_work(ThreadletWork *work);
void threadlet_init(void);
+void threadlet_queue_init(ThreadletQueue *queue, int max_threads,
+ int min_threads);
#endif
^ permalink raw reply related [flat|nested] 27+ messages in thread* [Qemu-devel] [PATCH 12/12] Threadlets: Add documentation
2011-01-13 12:14 [Qemu-devel] [PATCH 00/12] Threadlets Infrastructure Arun R Bharadwaj
` (10 preceding siblings ...)
2011-01-13 12:15 ` [Qemu-devel] [PATCH 11/12] Threadlets: Add functionality to create private queues Arun R Bharadwaj
@ 2011-01-13 12:15 ` Arun R Bharadwaj
2011-01-17 10:32 ` [Qemu-devel] [PATCH 00/12] Threadlets Infrastructure Stefan Hajnoczi
12 siblings, 0 replies; 27+ messages in thread
From: Arun R Bharadwaj @ 2011-01-13 12:15 UTC (permalink / raw)
To: qemu-devel; +Cc: kwolf, aliguori, jvrao, aneesh.kumar
Signed-off-by: Arun R Bharadwaj <arun@linux.vnet.ibm.com>
Signed-off-by: Gautham R Shenoy <gautham.shenoy@gmail.com>
---
docs/async-support.txt | 141 ++++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 141 insertions(+), 0 deletions(-)
create mode 100644 docs/async-support.txt
diff --git a/docs/async-support.txt b/docs/async-support.txt
new file mode 100644
index 0000000..9f22b9a
--- /dev/null
+++ b/docs/async-support.txt
@@ -0,0 +1,141 @@
+== How to use the threadlets infrastructure supported in Qemu ==
+
+== Threadlets ==
+
+Q.1: What are threadlets ?
+A.1: Threadlets is an infrastructure within QEMU that allows other subsystems
+ to offload possibly blocking work to a queue to be processed by a pool
+ of threads asynchronously.
+
+Q.2: When would one want to use threadlets ?
+A.2: Threadlets are useful when there are operations that can be performed
+ outside the context of the VCPU/IO threads inorder to free these latter
+ to service any other guest requests.
+
+Q.3: I have some work that can be executed in an asynchronous context. How
+ should I go about it ?
+A.3: One could follow the steps listed below:
+
+ - Define a function which would do the asynchronous work.
+ static void my_threadlet_func(ThreadletWork *work)
+ {
+ }
+
+ - Declare an object of type ThreadletWork;
+ ThreadletWork work;
+
+
+ - Assign a value to the "func" member of ThreadletWork object.
+ work.func = my_threadlet_func;
+
+ - Submit the threadlet to the global queue.
+ submit_threadletwork(&work);
+
+ - Continue servicing some other guest operations.
+
+Q.4: I want to my_threadlet_func to access some non-global data. How do I do
+ that ?
+A.4: Suppose you want my_threadlet_func to access some non-global data-object
+ of type myPrivateData. In that case one could follow the following steps.
+
+ - Define a member of the type ThreadletWork within myPrivateData.
+ typedef struct MyPrivateData {
+ ...;
+ ...;
+ ...;
+ ThreadletWork work;
+ } MyPrivateData;
+
+ MyPrivateData my_data;
+
+ - Initialize myData.work as described in A.3
+ myData.work.func = my_threadlet_func;
+ submit_threadletwork(&myData.work);
+
+ - Access the myData object inside my_threadlet_func() using container_of
+ primitive
+ static void my_threadlet_func(ThreadletWork *work)
+ {
+ myPrivateData *mydata_ptr;
+ mydata_ptr = container_of(work, myPrivateData, work);
+
+ /* mydata_ptr now points to myData object */
+ }
+
+Q.5: Are there any precautions one must take while sharing data with the
+ Asynchronous thread-pool ?
+A.5: Yes, make sure that the helper function of the type my_threadlet_func()
+ does not access/modify data when it can be accessed or modified in the
+ context of VCPU thread or IO thread. This is because the asynchronous
+ threads in the pool can run in parallel with the VCPU/IOThreads as shown
+ in the figure.
+
+ A typical workflow is as follows:
+
+ VCPU/IOThread
+ |
+ | (1)
+ |
+ V
+ Offload work (2)
+ |-------> to threadlets -----------------------------> Helper thread
+ | | |
+ | | |
+ | | (3) | (4)
+ | | |
+ | Handle other Guest requests |
+ | | |
+ | | V
+ | | (3) Signal the I/O Thread
+ |(6) | |
+ | | /
+ | | /
+ | V /
+ | Do the post <---------------------------------/
+ | processing (5)
+ | |
+ | | (6)
+ | V
+ |-Yes------ More async work?
+ |
+ | (7)
+ No
+ |
+ |
+ .
+ .
+
+ Hence one needs to make sure that in the steps (3) and (4) which run in
+ parallel, any global data is accessed within only one context.
+
+Q.6: I have queued a threadlet which I want to cancel. How do I do that ?
+A.6: Threadlets framework provides the API cancel_threadlet:
+ - int cancel_threadletwork(ThreadletWork *work)
+
+ The API scans the ThreadletQueue to see if (work) is present. If it finds
+ work, it'll dequeue work and return 0.
+
+ On the other hand, if it does not find the (work) in the ThreadletQueue,
+ then it'll return 1. This can imply two things. Either the work is being
+ processed by one of the helper threads or it has been processed. The
+ threadlet infrastructure currently _does_not_ distinguish between these
+ two and the onus is on the caller to do that.
+
+Q.7: Apart from the global pool of threads, can I have my own private Queue ?
+A.7: Yes, the threadlets framework allows subsystems to create their own private
+ queues with associated pools of threads.
+
+ - Define a PrivateQueue
+ ThreadletQueue myQueue;
+
+ - Initialize it:
+ threadlet_queue_init(&myQueue, my_max_threads, my_min_threads);
+ where my_max_threads is the maximum number of threads that can be in the
+ thread pool and my_min_threads is the minimum number of active threads
+ that can be in the thread-pool.
+
+ - Submit work:
+ submit_threadletwork_to_queue(&myQueue, &my_work);
+
+ - Cancel work:
+ cancel_threadletwork_on_queue(&myQueue, &my_work);
^ permalink raw reply related [flat|nested] 27+ messages in thread* Re: [Qemu-devel] [PATCH 00/12] Threadlets Infrastructure.
2011-01-13 12:14 [Qemu-devel] [PATCH 00/12] Threadlets Infrastructure Arun R Bharadwaj
` (11 preceding siblings ...)
2011-01-13 12:15 ` [Qemu-devel] [PATCH 12/12] Threadlets: Add documentation Arun R Bharadwaj
@ 2011-01-17 10:32 ` Stefan Hajnoczi
2011-01-18 4:38 ` Arun R Bharadwaj
12 siblings, 1 reply; 27+ messages in thread
From: Stefan Hajnoczi @ 2011-01-17 10:32 UTC (permalink / raw)
To: Arun R Bharadwaj; +Cc: kwolf, aliguori, jvrao, qemu-devel, aneesh.kumar
This series needs a new pair of eyes for review. I'm probably missing
things here after seeing it many times.
posix-aio-compat.c:handle_work() doesn't need to take aiocb_mutex for
container_of(), which just calculates an address but doesn't actually
access the aiocb.
dequeue_work_on_queue() is incorrect because a threadlet_worker() may
remove the work item from the request_list just before we remove it
again. That double-remove is wrong and by not detecting this case we
return success even though the work function may be running
concurrently (and it's not safe to do anything with the work item at
this time!). Dequeue needs to lock the request_list, test if the item
is still
on the list, remove it if necessary, and return whether or not it was removed.
Stefan
^ permalink raw reply [flat|nested] 27+ messages in thread* Re: [Qemu-devel] [PATCH 00/12] Threadlets Infrastructure.
2011-01-17 10:32 ` [Qemu-devel] [PATCH 00/12] Threadlets Infrastructure Stefan Hajnoczi
@ 2011-01-18 4:38 ` Arun R Bharadwaj
0 siblings, 0 replies; 27+ messages in thread
From: Arun R Bharadwaj @ 2011-01-18 4:38 UTC (permalink / raw)
To: Stefan Hajnoczi; +Cc: kwolf, aliguori, jvrao, qemu-devel, aneesh.kumar
* Stefan Hajnoczi <stefanha@gmail.com> [2011-01-17 10:32:24]:
> This series needs a new pair of eyes for review. I'm probably missing
> things here after seeing it many times.
>
> posix-aio-compat.c:handle_work() doesn't need to take aiocb_mutex for
> container_of(), which just calculates an address but doesn't actually
> access the aiocb.
>
> dequeue_work_on_queue() is incorrect because a threadlet_worker() may
> remove the work item from the request_list just before we remove it
> again. That double-remove is wrong and by not detecting this case we
> return success even though the work function may be running
> concurrently (and it's not safe to do anything with the work item at
> this time!). Dequeue needs to lock the request_list, test if the item
> is still
> on the list, remove it if necessary, and return whether or not it was removed.
>
> Stefan
I'll make the above changes with respect to handle_work() and
dequeue_work_on_queue().
Hopefully the patch series gets more review now that I have broken
down the patchset so that review becomes easier.
-arun
^ permalink raw reply [flat|nested] 27+ messages in thread