7:43 p.m.
On Wed, Dec 01, 2010 at 05:26:27PM +0000, Daniel P. Berrange wrote:
From: Hu Tao <hutao(a)cn.fujitsu.com>
* src/util/threadpool.c, src/util/threadpool.h: Thread pool
implementation
* src/Makefile.am: Build thread pool
---
src/Makefile.am | 1 +
src/util/threadpool.c | 178 +++++++++++++++++++++++++++++++++++++++++++++++++
src/util/threadpool.h | 23 ++++++
3 files changed, 202 insertions(+), 0 deletions(-)
create mode 100644 src/util/threadpool.c
create mode 100644 src/util/threadpool.h
diff --git a/src/Makefile.am b/src/Makefile.am
index a9a1986..d71c644 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -73,6 +73,7 @@ UTIL_SOURCES = \
util/threads.c util/threads.h \
util/threads-pthread.h \
util/threads-win32.h \
+ util/threadpool.c util/threadpool.h \
util/uuid.c util/uuid.h \
util/util.c util/util.h \
util/xml.c util/xml.h \
diff --git a/src/util/threadpool.c b/src/util/threadpool.c
new file mode 100644
index 0000000..cf998bf
--- /dev/null
+++ b/src/util/threadpool.c
@@ -0,0 +1,178 @@
+
+#include <config.h>
+
+#include "threadpool.h"
+#include "memory.h"
+#include "threads.h"
+#include "virterror_internal.h"
+
+#define VIR_FROM_THIS VIR_FROM_NONE
+
+typedef struct _virThreadPoolJob virThreadPoolJob;
+typedef virThreadPoolJob *virThreadPoolJobPtr;
+
+struct _virThreadPoolJob {
+ virThreadPoolJobPtr next;
+
+ void *data;
+};
+
+struct _virThreadPool {
+ int quit;
+
+ virThreadPoolJobFunc jobFunc;
+ void *jobOpaque;
+ virThreadPoolJobPtr jobList;
+
+ virMutex mutex;
+ virCond cond;
+ virCond quit_cond;
+
+ size_t maxWorkers;
+ size_t freeWorkers;
+ size_t nWorkers;
+ virThreadPtr workers;
+};
+
+static void virThreadPoolWorker(void *opaque)
+{
+ virThreadPoolPtr pool = opaque;
+
+ virMutexLock(&pool->mutex);
+
+ while (1) {
+ while (!pool->quit &&
+ !pool->jobList) {
+ pool->freeWorkers++;
+ if (virCondWait(&pool->cond, &pool->mutex) < 0) {
+ pool->freeWorkers--;
+ break;
+ }
+ pool->freeWorkers--;
+ }
+
+ if (pool->quit)
+ break;
+
+ virThreadPoolJobPtr job = pool->jobList;
+ pool->jobList = pool->jobList->next;
+ job->next = NULL;
+
+ virMutexUnlock(&pool->mutex);
+ (pool->jobFunc)(job->data, pool->jobOpaque);
This could race if jobFunc does something with jobOpaque unless jobFunc
is aware of this and provides a lock to protect jobOpaque.
+ VIR_FREE(job);
+ virMutexLock(&pool->mutex);
+ }
+
+ pool->nWorkers--;
+ if (pool->nWorkers == 0)
+ virCondSignal(&pool->quit_cond);
+ virMutexUnlock(&pool->mutex);
+}
+
+virThreadPoolPtr virThreadPoolNew(size_t minWorkers,
+ size_t maxWorkers,
+ virThreadPoolJobFunc func,
+ void *opaque)
+{
+ virThreadPoolPtr pool;
+ int i;
+
+ if (VIR_ALLOC(pool) < 0) {
+ virReportOOMError();
+ return NULL;
+ }
+
+ pool->jobFunc = func;
+ pool->jobOpaque = opaque;
+
+ if (virMutexInit(&pool->mutex) < 0)
+ goto error;
+ if (virCondInit(&pool->cond) < 0)
+ goto error;
+ if (virCondInit(&pool->quit_cond) < 0)
+ goto error;
+
+ if (VIR_ALLOC_N(pool->workers, minWorkers) < 0)
+ goto error;
+
+ pool->maxWorkers = maxWorkers;
+ for (i = 0; i < minWorkers; i++) {
+ if (virThreadCreate(&pool->workers[i],
+ true,
+ virThreadPoolWorker,
+ pool) < 0)
+ goto error;
+ pool->nWorkers++;
+ }
There will be more than maxWorkers threads created if
minWorkers > maxWorkers
+
+ return pool;
+
+error:
+ virThreadPoolFree(pool);
+ return NULL;
+}
+
+void virThreadPoolFree(virThreadPoolPtr pool)
+{
+ virMutexLock(&pool->mutex);
+ pool->quit = 1;
+ if (pool->nWorkers > 0) {
+ virCondBroadcast(&pool->cond);
+ if (virCondWait(&pool->quit_cond, &pool->mutex) < 0)
+ {}
+ }
+ VIR_FREE(pool->workers);
+ virMutexUnlock(&pool->mutex);
+ VIR_FREE(pool);
+}
+
+int virThreadPoolSendJob(virThreadPoolPtr pool,
+ void *jobData)
+{
+ virThreadPoolJobPtr job;
+ virThreadPoolJobPtr tmp;
+
+ virMutexLock(&pool->mutex);
+ if (pool->quit)
+ goto error;
+
+ if (pool->freeWorkers == 0 &&
+ pool->nWorkers < pool->maxWorkers) {
+ if (VIR_EXPAND_N(pool->workers, pool->nWorkers, 1) < 0) {
+ virReportOOMError();
+ goto error;
+ }
+
+ if (virThreadCreate(&pool->workers[pool->nWorkers],
+ true,
+ virThreadPoolWorker,
+ pool) < 0)
+ goto error;
+ pool->nWorkers++;
+ }
+
+ if (VIR_ALLOC(job) < 0) {
+ virReportOOMError();
+ goto error;
+ }
+
+ job->data = jobData;
+
+ tmp = pool->jobList;
+ while (tmp && tmp->next)
+ tmp = tmp->next;
+ if (tmp)
+ tmp->next = job;
+ else
+ pool->jobList = job;
+
+ virCondSignal(&pool->cond);
+ virMutexUnlock(&pool->mutex);
+
+ return 0;
+
+error:
+ virMutexUnlock(&pool->mutex);
+ return -1;
+}
diff --git a/src/util/threadpool.h b/src/util/threadpool.h
new file mode 100644
index 0000000..093786f
--- /dev/null
+++ b/src/util/threadpool.h
@@ -0,0 +1,23 @@
+#ifndef __THREADPOOL_H__
+#define __THREADPOOL_H__
+
+#include <pthread.h>
+
+typedef struct _virThreadPool virThreadPool;
+typedef virThreadPool *virThreadPoolPtr;
+
+typedef void (*virThreadPoolJobFunc)(void *jobdata, void *opaque);
+
+virThreadPoolPtr virThreadPoolNew(size_t minWorkers,
+ size_t maxWorkers,
+ virThreadPoolJobFunc func,
+ void *opaque);
+
+void virThreadPoolShutdown(virThreadPoolPtr pool);
+
+void virThreadPoolFree(virThreadPoolPtr pool);
+
+int virThreadPoolSendJob(virThreadPoolPtr pool,
+ void *jobdata);
+
+#endif
--
1.7.2.3