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 "thread_pool.h"
  29
  30 #include "globals.h"
  31 #include "queues/job_queue.h"
  32 #include "utils/allocator.h"
  33 #include "utils/logger.h"
  34
  35 /**
  36  * @brief structure with private members for thread_pool_t
  37  */
  38 typedef struct private_thread_pool_s private_thread_pool_t;
  39
  40 struct private_thread_pool_s {
  41         /**
  42          * inclusion of public members
  43          */
  44         thread_pool_t public;
  45         /**
  46          * @brief Processing function of a worker thread
  47          *
  48          * @param this  private_thread_pool_t-Object
  49          */
  50         void (*function) (private_thread_pool_t *this);
  51         /**
  52          * number of running threads
  53          */
  54          size_t pool_size;
  55         /**
  56          * array of thread ids
  57          */
  58         pthread_t *threads;
  59         /**
  60          * logger of the threadpool
  61          */
  62         logger_t *logger;
  63 } ;
  64
  65
  66
  67 /**
  68  * implements private_thread_pool_t.function
  69  */
  70 static void job_processing(private_thread_pool_t *this)
  71 {
  72
  73         /* cancellation disabled by default */
  74         pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);
  75         this->logger->log(this->logger, CONTROL_MORE, "thread %u started working", pthread_self());
  76
  77         for (;;) {
  78                 job_t *job;
  79                 job_type_t job_type;
  80
  81                 global_job_queue->get(global_job_queue, &job);
  82                 job_type = job->get_type(job);
  83                 this->logger->log(this->logger, CONTROL_MORE, "thread %u got a job of type %s", pthread_self(),mapping_find(job_type_m,job_type));
  84
  85                 /* process them here */
  86                 switch (job_type)
  87                 {
  88                         case INCOMING_PACKET:
  89                         {
  90                                 packet_t *packet;
  91                                 message_t *message;
  92                                 ike_sa_t *ike_sa;
  93                                 ike_sa_id_t *ike_sa_id;
  94                                 status_t status;
  95                                 incoming_packet_job_t *incoming_packet_job = (incoming_packet_job_t *)job;
  96
  97                                 if (incoming_packet_job->get_packet(incoming_packet_job,&packet) != SUCCESS)
  98                                 {
  99                                         this->logger->log(this->logger, CONTROL_MORE, "thread %u: Packet in job of type %s could not be retrieved!", pthread_self(),mapping_find(job_type_m,job_type));
 100                                         break;
 101                                 }
 102                                 message = message_create_from_packet(packet);
 103                                 if (message == NULL)
 104                                 {
 105                                         this->logger->log(this->logger, CONTROL_MORE, "thread %u: Message could not be created from packet!", pthread_self(),mapping_find(job_type_m,job_type));
 106                                         packet->destroy(packet);
 107                                         break;
 108                                 }
 109
 110                                 status = message->parse_header(message);
 111                                 if (status != SUCCESS)
 112                                 {
 113                                         this->logger->log(this->logger, CONTROL_MORE, "thread %u: Message header could not be verified!", pthread_self());
 114                                         message->destroy(message);
 115                                         break;
 116                                 }
 117
 118                                 if ((message->get_major_version(message) != IKE_MAJOR_VERSION) || (message->get_minor_version(message) != IKE_MINOR_VERSION))
 119                                 {
 120                                         this->logger->log(this->logger, CONTROL_MORE, "thread %u: IKE Version %d.%d not supported", pthread_self(),message->get_major_version(message),message->get_minor_version(message));
 121                                         /* Todo send notify */
 122                                 }
 123
 124                                 status = message->get_ike_sa_id(message,&ike_sa_id);
 125                                 if (status != SUCCESS)
 126                                 {
 127                                         this->logger->log(this->logger, CONTROL_MORE, "thread %u: IKE SA ID of message could not be created!", pthread_self());
 128                                         message->destroy(message);
 129                                         break;
 130                                 }
 131
 132                                 status = global_ike_sa_manager->checkout(global_ike_sa_manager,ike_sa_id, &ike_sa);
 133                                 if (status != SUCCESS)
 134                                 {
 135                                         this->logger->log(this->logger, CONTROL_MORE, "thread %u: IKE SA could not be checked out", pthread_self());
 136                                         message->destroy(message);
 137                                         break;
 138                                 }
 139
 140                                 {
 141                                         /* only for logging */
 142                                         ike_sa_id_t *checked_out_ike_sa_id;
 143                                         checked_out_ike_sa_id = ike_sa->get_id(ike_sa);
 144                                         u_int64_t initiator;
 145                                         u_int64_t responder;
 146                                         bool is_initiator;
 147                                         checked_out_ike_sa_id->get_values(checked_out_ike_sa_id,&initiator,&responder,&is_initiator);
 148                                         this->logger->log(this->logger, CONTROL_MORE, "IKE SA with SPI's I:%d, R:%d checked out", initiator,responder);
 149                                 }
 150
 151                                 status = ike_sa->process_message (ike_sa,message);
 152                                 if (status != SUCCESS)
 153                                 {
 154                                         this->logger->log(this->logger, CONTROL_MORE, "thread %u: Message could not be processed by IKE SA", pthread_self());
 155                                 }
 156
 157                                 status = global_ike_sa_manager->checkin(global_ike_sa_manager,ike_sa);
 158                                 if (status != SUCCESS){
 159                                         this->logger->log(this->logger, CONTROL_MORE, "thread %u: Checkin of IKE SA return errors", pthread_self());
 160                                 }
 161                                 message->destroy(message);
 162                                 ike_sa_id->destroy(ike_sa_id);
 163
 164                                 break;
 165                         }
 166                         case INITIATE_IKE_SA:
 167                         {
 168                                 /*
 169                                  * Initiatie an IKE_SA:
 170                                  * - is defined by a name of a configuration
 171                                  * - create an empty IKE_SA via manager
 172                                  * - call initiate_connection on this sa
 173                                  */
 174                                 initiate_ike_sa_job_t *initiate_job;
 175                                 ike_sa_id_t *ike_sa_id;
 176                                 ike_sa_t *ike_sa;
 177                                 status_t status;
 178
 179                                 initiate_job = (initiate_ike_sa_job_t *)job;
 180                                 this->logger->log(this->logger, CONTROL, "thread %u: Initiating an IKE_SA for config \"%s\"",
 181                                                                         pthread_self(), initiate_job->get_configuration_name(initiate_job));
 182
 183                                 ike_sa_id = ike_sa_id_create(0, 0, TRUE);
 184                                 if (ike_sa_id == NULL)
 185                                 {
 186                                         this->logger->log(this->logger, ERROR, "thread %u: %s by creating ike_sa_id_t, job rejected.",
 187                                                                                 pthread_self(), mapping_find(status_m, status));
 188                                         break;
 189                                 }
 190
 191                                 status = global_ike_sa_manager->checkout(global_ike_sa_manager, ike_sa_id, &ike_sa);
 192                                 ike_sa_id->destroy(ike_sa_id);
 193                                 if (status != SUCCESS)
 194                                 {
 195                                         this->logger->log(this->logger, ERROR, "thread %u: %s by checking out new IKE_SA, job rejected.",
 196                                                                                 pthread_self(), mapping_find(status_m, status));
 197                                         break;
 198                                 }
 199
 200                                 status = ike_sa->initialize_connection(ike_sa, initiate_job->get_configuration_name(initiate_job));
 201                                 if (status != SUCCESS)
 202                                 {
 203                                         this->logger->log(this->logger, ERROR, "thread %u: %s by initialize_conection, job and rejected, IKE_SA deleted.",
 204                                                                                 pthread_self(), mapping_find(status_m, status));
 205                                         global_ike_sa_manager->checkin_and_delete(global_ike_sa_manager, ike_sa);
 206                                         break;
 207                                 }
 208
 209                                 status = global_ike_sa_manager->checkin(global_ike_sa_manager, ike_sa);
 210                                 if (status != SUCCESS)
 211                                 {
 212                                         this->logger->log(this->logger, ERROR, "thread %u: %s  could not checkin IKE_SA.",
 213                                                                                 pthread_self(), mapping_find(status_m, status));
 214                                 }
 215                                 break;
 216                         }
 217                         case RETRANSMIT_REQUEST:
 218                         {
 219                                 this->logger->log(this->logger, CONTROL_MORE, "thread %u: Job of type %s not supported!", pthread_self(),mapping_find(job_type_m,job_type));
 220                                 break;
 221                         }
 222                 }
 223                 job->destroy(job);
 224         }
 225
 226 }
 227
 228 /**
 229  * implementation of thread_pool_t.get_pool_size
 230  */
 231 static size_t get_pool_size(private_thread_pool_t *this)
 232 {
 233         return this->pool_size;
 234 }
 235
 236 /**
 237  * Implementation of thread_pool_t.destroy
 238  */
 239 static status_t destroy(private_thread_pool_t *this)
 240 {
 241         int current;
 242         /* flag thread for termination */
 243         for (current = 0; current < this->pool_size; current++) {
 244                 this->logger->log(this->logger, CONTROL, "cancelling thread %u", this->threads[current]);
 245                 pthread_cancel(this->threads[current]);
 246         }
 247
 248         /* wait for all threads */
 249         for (current = 0; current < this->pool_size; current++) {
 250                 pthread_join(this->threads[current], NULL);
 251                 this->logger->log(this->logger, CONTROL, "thread %u terminated", this->threads[current]);
 252         }
 253
 254         /* free mem */
 255         global_logger_manager->destroy_logger(global_logger_manager, this->logger);
 256         allocator_free(this->threads);
 257         allocator_free(this);
 258         return SUCCESS;
 259 }
 260
 261 #include <stdio.h>
 262
 263 /*
 264  * see header
 265  */
 266 thread_pool_t *thread_pool_create(size_t pool_size)
 267 {
 268         int current;
 269
 270         private_thread_pool_t *this = allocator_alloc_thing(private_thread_pool_t);
 271
 272         /* fill in public fields */
 273         this->public.destroy = (status_t(*)(thread_pool_t*))destroy;
 274         this->public.get_pool_size = (size_t(*)(thread_pool_t*))get_pool_size;
 275
 276         this->function = job_processing;
 277         this->pool_size = pool_size;
 278
 279         this->threads = allocator_alloc(sizeof(pthread_t) * pool_size);
 280         if (this->threads == NULL)
 281         {
 282                 allocator_free(this);
 283                 return NULL;
 284         }
 285         this->logger = global_logger_manager->create_logger(global_logger_manager,THREAD_POOL,NULL);
 286         if (this->threads == NULL)
 287         {
 288                 allocator_free(this);
 289                 allocator_free(this->threads);
 290                 return NULL;
 291         }
 292
 293         /* try to create as many threads as possible, up tu pool_size */
 294         for (current = 0; current < pool_size; current++)
 295         {
 296                 if (pthread_create(&(this->threads[current]), NULL, (void*(*)(void*))this->function, this) == 0)
 297                 {
 298                         this->logger->log(this->logger, CONTROL, "thread %u created", this->threads[current]);
 299                 }
 300                 else
 301                 {
 302                         /* creation failed, is it the first one? */
 303                         if (current == 0)
 304                         {
 305                                 this->logger->log(this->logger, CONTROL, "could not create any thread: %s\n", strerror(errno));
 306                                 allocator_free(this->threads);
 307                                 allocator_free(this->logger);
 308                                 allocator_free(this);
 309                                 return NULL;
 310                         }
 311                         /* not all threads could be created, but at least one :-/ */
 312                         this->logger->log(this->logger, CONTROL, "could only create %d from requested %d threads: %s\n", current, pool_size, strerror(errno));
 313
 314                         this->pool_size = current;
 315                         return (thread_pool_t*)this;
 316                 }
 317         }
 318         return (thread_pool_t*)this;
 319 }