On 04/28/2011 09:40 PM, Sasha Levin wrote: > This patch adds a generic pool to create a common interface for working with > threads within the kvm tool. > Main idea here is using this threadpool for all I/O threads instead of having > every I/O module write it's own thread code. > > The process of working with the thread pool is supposed to be very simple. > During initialization, Each module which is interested in working with the > threadpool will call threadpool__add_jobtype with the callback function and a > void* parameter. For example, virtio modules will register every virt_queue > as a new job type. > During operation, When theres work to do for a specific job, the module will > signal it to the queue and would expect the callback to be called with proper > parameters. It is assured that the callback will be called once for every > signal action and each callback will be called only once at a time (i.e. > callback functions themselves don't need to handle threading). > > Signed-off-by: Sasha Levin <levinsasha...@gmail.com> > --- > tools/kvm/Makefile | 1 + > tools/kvm/include/kvm/threadpool.h | 16 ++++ > tools/kvm/kvm-run.c | 5 + > tools/kvm/threadpool.c | 171 > ++++++++++++++++++++++++++++++++++++ > 4 files changed, 193 insertions(+), 0 deletions(-) > create mode 100644 tools/kvm/include/kvm/threadpool.h > create mode 100644 tools/kvm/threadpool.c > > diff --git a/tools/kvm/Makefile b/tools/kvm/Makefile > index 1b0c76e..fbce14d 100644 > --- a/tools/kvm/Makefile > +++ b/tools/kvm/Makefile > @@ -36,6 +36,7 @@ OBJS += kvm-cmd.o > OBJS += kvm-run.o > OBJS += qcow.o > OBJS += mptable.o > +OBJS += threadpool.o > > DEPS := $(patsubst %.o,%.d,$(OBJS)) > > diff --git a/tools/kvm/include/kvm/threadpool.h > b/tools/kvm/include/kvm/threadpool.h > new file mode 100644 > index 0000000..25b5eb8 > --- /dev/null > +++ b/tools/kvm/include/kvm/threadpool.h > @@ -0,0 +1,16 @@ > +#ifndef KVM__THREADPOOL_H > +#define KVM__THREADPOOL_H > + > +#include <stdint.h> > + > +struct kvm; > + > +typedef void (*kvm_thread_callback_fn_t)(struct kvm *kvm, void *data); > + > +int thread_pool__init(unsigned long thread_count); > + > +void *thread_pool__add_jobtype(struct kvm *kvm, kvm_thread_callback_fn_t > callback, void *data); > + > +void thread_pool__signal_work(void *job); > + > +#endif > diff --git a/tools/kvm/kvm-run.c b/tools/kvm/kvm-run.c > index 071157a..97a17dd 100644 > --- a/tools/kvm/kvm-run.c > +++ b/tools/kvm/kvm-run.c > @@ -24,6 +24,7 @@ > #include <kvm/pci.h> > #include <kvm/term.h> > #include <kvm/ioport.h> > +#include <kvm/threadpool.h> > > /* header files for gitish interface */ > #include <kvm/kvm-run.h> > @@ -312,6 +313,7 @@ int kvm_cmd_run(int argc, const char **argv, const char > *prefix) > int i; > struct virtio_net_parameters net_params; > char *hi; > + unsigned int nr_online_cpus; > > signal(SIGALRM, handle_sigalrm); > signal(SIGQUIT, handle_sigquit); > @@ -457,6 +459,9 @@ int kvm_cmd_run(int argc, const char **argv, const char > *prefix) > > kvm__init_ram(kvm); > > + nr_online_cpus = sysconf(_SC_NPROCESSORS_ONLN); > + thread_pool__init(nr_online_cpus);
We may benefit from more threads than the number of hardware thread we have. Currently, virtio_console consumes two, virio_net consumes two, and virtio_blk consumes one. Can we adjust the thread pool size when devices register to use thread pool? > + > for (i = 0; i < nrcpus; i++) { > if (pthread_create(&kvm_cpus[i]->thread, NULL, kvm_cpu_thread, > kvm_cpus[i]) != 0) > die("unable to create KVM VCPU thread"); > diff --git a/tools/kvm/threadpool.c b/tools/kvm/threadpool.c > new file mode 100644 > index 0000000..e78db3a > --- /dev/null > +++ b/tools/kvm/threadpool.c > @@ -0,0 +1,171 @@ > +#include "kvm/threadpool.h" > +#include "kvm/mutex.h" > + > +#include <linux/kernel.h> > +#include <linux/list.h> > +#include <pthread.h> > +#include <stdbool.h> > + > +struct thread_pool__job_info { > + kvm_thread_callback_fn_t callback; > + struct kvm *kvm; > + void *data; > + > + int signalcount; > + pthread_mutex_t mutex; > + > + struct list_head queue; > +}; Does 'struct thread_pool__job' sound better? > +static pthread_mutex_t job_mutex = > PTHREAD_MUTEX_INITIALIZER; > +static pthread_mutex_t thread_mutex = PTHREAD_MUTEX_INITIALIZER; > +static pthread_cond_t job_cond = > PTHREAD_COND_INITIALIZER; These mutex and cond are global. As the number of thread/job grows, there may be a lot of contention. > + > +static LIST_HEAD(head); > + > +static pthread_t *threads; > +static long threadcount; > + > +static struct thread_pool__job_info *thread_pool__job_info_pop(void) > +{ > + struct thread_pool__job_info *job; > + > + if (list_empty(&head)) > + return NULL; > + > + job = list_first_entry(&head, struct thread_pool__job_info, queue); > + list_del(&job->queue); > + > + return job; > +} > + > +static void thread_pool__job_info_push(struct thread_pool__job_info *job) > +{ > + list_add_tail(&job->queue, &head); > +} > + > +static struct thread_pool__job_info *thread_pool__job_info_pop_locked(void) > +{ > + struct thread_pool__job_info *job; > + > + mutex_lock(&job_mutex); > + job = thread_pool__job_info_pop(); > + mutex_unlock(&job_mutex); > + return job; > +} > + > +static void thread_pool__job_info_push_locked(struct thread_pool__job_info > *job) > +{ > + mutex_lock(&job_mutex); > + thread_pool__job_info_push(job); > + mutex_unlock(&job_mutex); > +} > + > +static void thread_pool__handle_job(struct thread_pool__job_info *job) > +{ > + while (job) { > + job->callback(job->kvm, job->data); > + > + mutex_lock(&job->mutex); > + > + if (--job->signalcount > 0) > + /* If the job was signaled again while we were working > */ > + thread_pool__job_info_push_locked(job); > + > + mutex_unlock(&job->mutex); > + > + job = thread_pool__job_info_pop_locked(); > + } > +} > + > +static void thread_pool__threadfunc_cleanup(void *param) > +{ > + mutex_unlock(&job_mutex); > +} > + > +static void *thread_pool__threadfunc(void *param) > +{ > + pthread_cleanup_push(thread_pool__threadfunc_cleanup, NULL); > + > + for (;;) { > + struct thread_pool__job_info *curjob; > + > + mutex_lock(&job_mutex); > + pthread_cond_wait(&job_cond, &job_mutex); > + curjob = thread_pool__job_info_pop(); > + mutex_unlock(&job_mutex); > + > + if (curjob) > + thread_pool__handle_job(curjob); > + } > + > + pthread_cleanup_pop(0); > + > + return NULL; > +} > + > +static int thread_pool__addthread(void) > +{ > + int res; > + void *newthreads; > + > + mutex_lock(&thread_mutex); > + newthreads = realloc(threads, (threadcount + 1) * sizeof(pthread_t)); > + if (newthreads == NULL) { > + mutex_unlock(&thread_mutex); > + return -1; > + } > + > + threads = newthreads; > + > + res = pthread_create(threads + threadcount, NULL, > + > thread_pool__threadfunc, NULL); > + > + if (res == 0) > + threadcount++; > + mutex_unlock(&thread_mutex); > + > + return res; > +} > + > +int thread_pool__init(unsigned long thread_count) > +{ > + unsigned long i; > + > + for (i = 0 ; i < thread_count ; i++) > + if (thread_pool__addthread() < 0) > + return i; > + > + return i; > +} > + > +void *thread_pool__add_jobtype(struct kvm *kvm, > + > kvm_thread_callback_fn_t callback, > + void *data) Is thread_pool__add_job() better? > +{ > + struct thread_pool__job_info *job = calloc(1, sizeof(*job)); > + > + *job = (struct thread_pool__job_info) { > + .kvm = kvm, > + .data = data, > + .callback = callback, > + .mutex = PTHREAD_MUTEX_INITIALIZER > + }; > + > + return job; > +} > + > +void thread_pool__signal_work(void *job) I think thread_pool__signal_job() or thread_pool__do_job() would be more consistent. Consumer of this API can simply use it with: thread_pool_{add,do}_job(). > +{ > + struct thread_pool__job_info *jobinfo = job; > + > + if (jobinfo == NULL) > + return; > + > + mutex_lock(&jobinfo->mutex); > + if (jobinfo->signalcount++ == 0) > + thread_pool__job_info_push_locked(job); > + mutex_unlock(&jobinfo->mutex); > + > + pthread_cond_signal(&job_cond); > +} -- Best Regards, Asias He -- To unsubscribe from this list: send the line "unsubscribe kvm" in the body of a message to majord...@vger.kernel.org More majordomo info at http://vger.kernel.org/majordomo-info.html