Source/charon/thread_pool.c

   1 /**
   2  * @file thread_pool.c
   3  *
   4  * @brief Thread-pool with some threads processing the job_queue
   5  *
   6  */
   7
   8 /*
   9  * Copyright (C) 2005 Jan Hutter, Martin Willi
  10  * Hochschule fuer Technik Rapperswil
  11  *
  12  * This program is free software; you can redistribute it and/or modify it
  13  * under the terms of the GNU General Public License as published by the
  14  * Free Software Foundation; either version 2 of the License, or (at your
  15  * option) any later version.  See <http://www.fsf.org/copyleft/gpl.txt>.
  16  *
  17  * This program is distributed in the hope that it will be useful, but
  18  * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
  19  * or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
  20  * for more details.
  21  */
  22
  23 #include <stdlib.h>
  24 #include <pthread.h>
  25 #include <string.h>
  26 #include <errno.h>
  27
  28 #include "allocator.h"
  29 #include "logger.h"
  30 #include "thread_pool.h"
  31 #include "job_queue.h"
  32 #include "globals.h"
  33
  34 /**
  35  * structure with private members for thread_pool
  36  */
  37 typedef struct {
  38         /**
  39          * inclusion of public members
  40          */
  41         thread_pool_t public;
  42         /**
  43          * number of running threads
  44          */
  45          size_t pool_size;
  46         /**
  47          * array of thread ids
  48          */
  49         pthread_t *threads;
  50         /**
  51          * logger of the threadpool
  52          */
  53         logger_t *logger;
  54 } private_thread_pool_t;
  55
  56
  57 static void job_processing(private_thread_pool_t *this)
  58 {
  59         /* cancellation disabled by default */
  60         pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);
  61
  62         for (;;) {
  63                 job_t *job;
  64
  65                 global_job_queue->get(global_job_queue, &job);
  66
  67                 /* process them here */
  68
  69                 job->destroy(job);
  70         }
  71
  72 }
  73
  74 /**
  75  * Implementation of thread_pool_t.get_pool_size
  76  */
  77 static status_t get_pool_size(private_thread_pool_t *this, size_t *size)
  78 {
  79         *size = this->pool_size;
  80         return SUCCESS;
  81 }
  82
  83 /**
  84  * Implementation of thread_pool_t.destroy
  85  */
  86 static status_t destroy(private_thread_pool_t *this)
  87 {
  88         int current;
  89         /* flag thread for termination */
  90         for (current = 0; current < this->pool_size; current++) {
  91                 this->logger->log(this->logger, CONTROL, "cancelling thread %u", this->threads[current]);
  92                 pthread_cancel(this->threads[current]);
  93         }
  94
  95         /* wait for all threads */
  96         for (current = 0; current < this->pool_size; current++) {
  97                 pthread_join(this->threads[current], NULL);
  98                 this->logger->log(this->logger, CONTROL, "thread %u terminated", this->threads[current]);
  99         }
 100
 101         /* free mem */
 102         allocator_free(this->threads);
 103         allocator_free(this);
 104         return SUCCESS;
 105 }
 106
 107 /**
 108  * Implementation of default constructor for thread_pool_t
 109  */
 110 thread_pool_t *thread_pool_create(size_t pool_size)
 111 {
 112         int current;
 113
 114         private_thread_pool_t *this = allocator_alloc_thing(private_thread_pool_t);
 115
 116         /* fill in public fields */
 117         this->public.destroy = (status_t(*)(thread_pool_t*))destroy;
 118         this->public.get_pool_size = (status_t(*)(thread_pool_t*, size_t*))get_pool_size;
 119
 120         this->pool_size = pool_size;
 121         this->threads = allocator_alloc(sizeof(pthread_t) * pool_size);
 122         this->logger = logger_create("thread_pool", 0);
 123
 124         /* try to create as many threads as possible, up tu pool_size */
 125         for (current = 0; current < pool_size; current++) {
 126                 if (pthread_create(&(this->threads[current]), NULL, (void*(*)(void*))job_processing, this))
 127                 {
 128                         /* did we get any? */
 129                         if (current == 0)
 130                         {
 131                                 this->logger->log(this->logger, CONTROL, "could not create any thread: %s\n", strerror(errno));
 132                                 allocator_free(this->threads);
 133                                 allocator_free(this);
 134                                 return NULL;
 135                         }
 136                         /* not all threads could be created, but at least one :-/ */
 137                         this->logger->log(this->logger, CONTROL, "could only create %d from requested %d threads: %s\n", current, pool_size, strerror(errno));
 138
 139                         this->pool_size = current;
 140                         return (thread_pool_t*)this;
 141                 }
 142         }
 143
 144         return (thread_pool_t*)this;
 145 }