src/charon/processing/processor.c

   1 /*
   2  * Copyright (C) 2005-2007 Martin Willi
   3  * Copyright (C) 2005 Jan Hutter
   4  * Hochschule fuer Technik Rapperswil
   5  *
   6  * This program is free software; you can redistribute it and/or modify it
   7  * under the terms of the GNU General Public License as published by the
   8  * Free Software Foundation; either version 2 of the License, or (at your
   9  * option) any later version.  See <http://www.fsf.org/copyleft/gpl.txt>.
  10  *
  11  * This program is distributed in the hope that it will be useful, but
  12  * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
  13  * or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
  14  * for more details.
  15  *
  16  * $Id$
  17  */
  18
  19 #include <stdlib.h>
  20 #include <pthread.h>
  21 #include <string.h>
  22 #include <errno.h>
  23
  24 #include "processor.h"
  25
  26 #include <daemon.h>
  27 #include <utils/linked_list.h>
  28
  29
  30 typedef struct private_processor_t private_processor_t;
  31
  32 /**
  33  * Private data of processor_t class.
  34  */
  35 struct private_processor_t {
  36         /**
  37          * Public processor_t interface.
  38          */
  39         processor_t public;
  40
  41         /**
  42          * Number of running threads
  43          */
  44         u_int total_threads;
  45
  46         /**
  47          * Desired number of threads
  48          */
  49         u_int desired_threads;
  50
  51         /**
  52          * Number of threads waiting for work
  53          */
  54         u_int idle_threads;
  55
  56         /**
  57          * The jobs are stored in a linked list
  58          */
  59         linked_list_t *list;
  60
  61         /**
  62          * access to linked_list is locked through this mutex
  63          */
  64         pthread_mutex_t mutex;
  65
  66         /**
  67          * Condvar to wait for new jobs
  68          */
  69         pthread_cond_t condvar;
  70 };
  71
  72 static void process_jobs(private_processor_t *this);
  73
  74 /**
  75  * restart a terminated thread
  76  */
  77 static void restart(private_processor_t *this)
  78 {
  79         pthread_t thread;
  80
  81         if (pthread_create(&thread, NULL, (void*)process_jobs, this) != 0)
  82         {
  83                 this->total_threads--;
  84         }
  85 }
  86
  87 /**
  88  * Process queued jobs, called by the worker threads
  89  */
  90 static void process_jobs(private_processor_t *this)
  91 {
  92         int oldstate;
  93
  94         pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &oldstate);
  95
  96         DBG2(DBG_JOB, "started worker thread, thread_ID: %06u", (int)pthread_self());
  97
  98         pthread_mutex_lock(&this->mutex);
  99         while (this->desired_threads >= this->total_threads)
 100         {
 101                 job_t *job;
 102
 103                 if (this->list->get_count(this->list) == 0)
 104                 {
 105                         this->idle_threads++;
 106                         pthread_cond_wait(&this->condvar, &this->mutex);
 107                         this->idle_threads--;
 108                         continue;
 109                 }
 110                 this->list->remove_first(this->list, (void**)&job);
 111                 pthread_mutex_unlock(&this->mutex);
 112                 /* terminated threads are restarted, so we have a constant pool */
 113                 pthread_cleanup_push((void*)restart, this);
 114                 job->execute(job);
 115                 pthread_cleanup_pop(0);
 116                 pthread_mutex_lock(&this->mutex);
 117         }
 118         this->total_threads--;
 119         pthread_cond_broadcast(&this->condvar);
 120         pthread_mutex_unlock(&this->mutex);
 121 }
 122
 123 /**
 124  * Implementation of processor_t.get_total_threads.
 125  */
 126 static u_int get_total_threads(private_processor_t *this)
 127 {
 128         return this->total_threads;
 129 }
 130
 131 /**
 132  * Implementation of processor_t.get_idle_threads.
 133  */
 134 static u_int get_idle_threads(private_processor_t *this)
 135 {
 136         return this->idle_threads;
 137 }
 138
 139 /**
 140  * implements processor_t.get_job_load
 141  */
 142 static u_int get_job_load(private_processor_t *this)
 143 {
 144         u_int load;
 145         pthread_mutex_lock(&this->mutex);
 146         load = this->list->get_count(this->list);
 147         pthread_mutex_unlock(&this->mutex);
 148         return load;
 149 }
 150
 151 /**
 152  * implements function processor_t.queue_job
 153  */
 154 static void queue_job(private_processor_t *this, job_t *job)
 155 {
 156         pthread_mutex_lock(&this->mutex);
 157         this->list->insert_last(this->list, job);
 158         pthread_mutex_unlock(&this->mutex);
 159         pthread_cond_signal(&this->condvar);
 160 }
 161
 162 /**
 163  * Implementation of processor_t.set_threads.
 164  */
 165 static void set_threads(private_processor_t *this, u_int count)
 166 {
 167         pthread_mutex_lock(&this->mutex);
 168         if (count > this->total_threads)
 169         {       /* increase thread count */
 170                 int i;
 171                 pthread_t current;
 172
 173                 this->desired_threads = count;
 174                 DBG1(DBG_JOB, "spawning %d worker threads", count - this->total_threads);
 175                 for (i = this->total_threads; i < count; i++)
 176                 {
 177                         if (pthread_create(&current, NULL, (void*)process_jobs, this) == 0)
 178                         {
 179                                 this->total_threads++;
 180                         }
 181                 }
 182         }
 183         else if (count < this->total_threads)
 184         {       /* decrease thread count */
 185                 this->desired_threads = count;
 186         }
 187         pthread_mutex_unlock(&this->mutex);
 188 }
 189
 190 /**
 191  * Implementation of processor_t.destroy.
 192  */
 193 static void destroy(private_processor_t *this)
 194 {
 195         set_threads(this, 0);
 196         pthread_mutex_lock(&this->mutex);
 197         while (this->total_threads > 0)
 198         {
 199                 pthread_cond_broadcast(&this->condvar);
 200                 pthread_cond_wait(&this->condvar, &this->mutex);
 201         }
 202         pthread_mutex_unlock(&this->mutex);
 203         this->list->destroy_offset(this->list, offsetof(job_t, destroy));
 204         free(this);
 205 }
 206
 207 /*
 208  * Described in header.
 209  */
 210 processor_t *processor_create(size_t pool_size)
 211 {
 212         private_processor_t *this = malloc_thing(private_processor_t);
 213
 214         this->public.get_total_threads = (u_int(*)(processor_t*))get_total_threads;
 215         this->public.get_idle_threads = (u_int(*)(processor_t*))get_idle_threads;
 216         this->public.get_job_load = (u_int(*)(processor_t*))get_job_load;
 217         this->public.queue_job = (void(*)(processor_t*, job_t*))queue_job;
 218         this->public.set_threads = (void(*)(processor_t*, u_int))set_threads;
 219         this->public.destroy = (void(*)(processor_t*))destroy;
 220
 221         this->list = linked_list_create();
 222         pthread_mutex_init(&this->mutex, NULL);
 223         pthread_cond_init(&this->condvar, NULL);
 224         this->total_threads = 0;
 225         this->desired_threads = 0;
 226         this->idle_threads = 0;
 227
 228         return &this->public;
 229 }
 230