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                                 /* we must switch the initiator flag here, because the sender
 132                                  * interprets this flag the other way round
 133                                  */
 134                                 ike_sa_id->switch_initiator(ike_sa_id);
 135
 136                                 status = global_ike_sa_manager->checkout(global_ike_sa_manager,ike_sa_id, &ike_sa);
 137                                 if (status != SUCCESS)
 138                                 {
 139                                         this->logger->log(this->logger, CONTROL|MORE, "thread %u: IKE SA could not be checked out", pthread_self());
 140                                         message->destroy(message);
 141                                         break;
 142                                 }
 143
 144                                 {
 145                                         /* only for logging */
 146                                         ike_sa_id_t *checked_out_ike_sa_id;
 147                                         checked_out_ike_sa_id = ike_sa->get_id(ike_sa);
 148                                         u_int64_t initiator = checked_out_ike_sa_id->get_initiator_spi(checked_out_ike_sa_id);
 149                                         u_int64_t responder = checked_out_ike_sa_id->get_responder_spi(checked_out_ike_sa_id);
 150                                         this->logger->log(this->logger, CONTROL|MORE, "IKE SA with SPI's I:%d, R:%d checked out", initiator,responder);
 151                                 }
 152
 153                                 status = ike_sa->process_message (ike_sa,message);
 154                                 if (status != SUCCESS)
 155                                 {
 156                                         this->logger->log(this->logger, CONTROL|MORE, "thread %u: Message could not be processed by IKE SA", pthread_self());
 157                                 }
 158
 159                                 status = global_ike_sa_manager->checkin(global_ike_sa_manager,ike_sa);
 160                                 if (status != SUCCESS){
 161                                         this->logger->log(this->logger, CONTROL|MORE, "thread %u: Checkin of IKE SA return errors", pthread_self());
 162                                 }
 163                                 message->destroy(message);
 164                                 ike_sa_id->destroy(ike_sa_id);
 165
 166                                 break;
 167                         }
 168                         case INITIATE_IKE_SA:
 169                         {
 170                                 /*
 171                                  * Initiatie an IKE_SA:
 172                                  * - is defined by a name of a configuration
 173                                  * - create an empty IKE_SA via manager
 174                                  * - call initiate_connection on this sa
 175                                  */
 176                                 initiate_ike_sa_job_t *initiate_job;
 177                                 ike_sa_id_t *ike_sa_id;
 178                                 ike_sa_t *ike_sa;
 179                                 status_t status;
 180
 181                                 initiate_job = (initiate_ike_sa_job_t *)job;
 182                                 this->logger->log(this->logger, CONTROL, "thread %u: Initiating an IKE_SA for config \"%s\"",
 183                                                                         pthread_self(), initiate_job->get_configuration_name(initiate_job));
 184
 185                                 ike_sa_id = ike_sa_id_create(0, 0, TRUE);
 186                                 if (ike_sa_id == NULL)
 187                                 {
 188                                         this->logger->log(this->logger, ERROR, "thread %u: %s by creating ike_sa_id_t, job rejected.",
 189                                                                                 pthread_self(), mapping_find(status_m, status));
 190                                         break;
 191                                 }
 192
 193
 194                                 status = global_ike_sa_manager->checkout(global_ike_sa_manager, ike_sa_id, &ike_sa);
 195                                 ike_sa_id->destroy(ike_sa_id);
 196                                 if (status != SUCCESS)
 197                                 {
 198                                         this->logger->log(this->logger, ERROR, "thread %u: %s by checking out new IKE_SA, job rejected.",
 199                                                                                 pthread_self(), mapping_find(status_m, status));
 200                                         break;
 201                                 }
 202
 203                                 status = ike_sa->initialize_connection(ike_sa, initiate_job->get_configuration_name(initiate_job));
 204                                 if (status != SUCCESS)
 205                                 {
 206                                         this->logger->log(this->logger, ERROR, "thread %u: %s by initialize_conection, job and rejected, IKE_SA deleted.",
 207                                                                                 pthread_self(), mapping_find(status_m, status));
 208                                         global_ike_sa_manager->checkin_and_delete(global_ike_sa_manager, ike_sa);
 209                                         break;
 210                                 }
 211
 212                                 status = global_ike_sa_manager->checkin(global_ike_sa_manager, ike_sa);
 213                                 if (status != SUCCESS)
 214                                 {
 215                                         this->logger->log(this->logger, ERROR, "thread %u: %s  could not checkin IKE_SA.",
 216                                                                                 pthread_self(), mapping_find(status_m, status));
 217                                 }
 218                                 break;
 219                         }
 220                         case RETRANSMIT_REQUEST:
 221                         {
 222                                 this->logger->log(this->logger, CONTROL|MORE, "thread %u: Job of type %s not supported!", pthread_self(),mapping_find(job_type_m,job_type));
 223                                 break;
 224                         }
 225
 226                         case DELETE_IKE_SA:
 227                         {
 228                                 delete_ike_sa_job_t *delete_ike_sa_job = (delete_ike_sa_job_t*) job;
 229                                 ike_sa_id_t *ike_sa_id = delete_ike_sa_job->get_ike_sa_id(delete_ike_sa_job);
 230                                 status_t status;
 231
 232                                 {
 233                                         /* only for logging */
 234                                         u_int64_t initiator = ike_sa_id->get_initiator_spi(ike_sa_id);
 235                                         u_int64_t responder = ike_sa_id->get_responder_spi(ike_sa_id);
 236                                         this->logger->log(this->logger, CONTROL|MORE, "thread %u: Going to delete IKE SA with SPI's I:%d, R:%d", pthread_self(),initiator,responder);
 237                                 }
 238                                 status = global_ike_sa_manager->delete(global_ike_sa_manager, ike_sa_id);
 239                                 if (status != SUCCESS)
 240                                 {
 241                                         this->logger->log(this->logger, ERROR, "thread %u: %s  could not delete IKE_SA.",
 242                                                                                 pthread_self(), mapping_find(status_m, status));
 243                                 }
 244                                 break;
 245
 246                         }
 247                 }
 248                 job->destroy(job);
 249         }
 250
 251 }
 252
 253 /**
 254  * implementation of thread_pool_t.get_pool_size
 255  */
 256 static size_t get_pool_size(private_thread_pool_t *this)
 257 {
 258         return this->pool_size;
 259 }
 260
 261 /**
 262  * Implementation of thread_pool_t.destroy
 263  */
 264 static status_t destroy(private_thread_pool_t *this)
 265 {
 266         int current;
 267         /* flag thread for termination */
 268         for (current = 0; current < this->pool_size; current++) {
 269                 this->logger->log(this->logger, CONTROL, "cancelling thread %u", this->threads[current]);
 270                 pthread_cancel(this->threads[current]);
 271         }
 272
 273         /* wait for all threads */
 274         for (current = 0; current < this->pool_size; current++) {
 275                 pthread_join(this->threads[current], NULL);
 276                 this->logger->log(this->logger, CONTROL, "thread %u terminated", this->threads[current]);
 277         }
 278
 279         /* free mem */
 280         global_logger_manager->destroy_logger(global_logger_manager, this->logger);
 281         allocator_free(this->threads);
 282         allocator_free(this);
 283         return SUCCESS;
 284 }
 285
 286 #include <stdio.h>
 287
 288 /*
 289  * see header
 290  */
 291 thread_pool_t *thread_pool_create(size_t pool_size)
 292 {
 293         int current;
 294
 295         private_thread_pool_t *this = allocator_alloc_thing(private_thread_pool_t);
 296
 297         /* fill in public fields */
 298         this->public.destroy = (status_t(*)(thread_pool_t*))destroy;
 299         this->public.get_pool_size = (size_t(*)(thread_pool_t*))get_pool_size;
 300
 301         this->function = job_processing;
 302         this->pool_size = pool_size;
 303
 304         this->threads = allocator_alloc(sizeof(pthread_t) * pool_size);
 305         if (this->threads == NULL)
 306         {
 307                 allocator_free(this);
 308                 return NULL;
 309         }
 310         this->logger = global_logger_manager->create_logger(global_logger_manager,THREAD_POOL,NULL);
 311         if (this->threads == NULL)
 312         {
 313                 allocator_free(this);
 314                 allocator_free(this->threads);
 315                 return NULL;
 316         }
 317
 318         /* try to create as many threads as possible, up tu pool_size */
 319         for (current = 0; current < pool_size; current++)
 320         {
 321                 if (pthread_create(&(this->threads[current]), NULL, (void*(*)(void*))this->function, this) == 0)
 322                 {
 323                         this->logger->log(this->logger, CONTROL, "thread %u created", this->threads[current]);
 324                 }
 325                 else
 326                 {
 327                         /* creation failed, is it the first one? */
 328                         if (current == 0)
 329                         {
 330                                 this->logger->log(this->logger, CONTROL, "could not create any thread: %s\n", strerror(errno));
 331                                 allocator_free(this->threads);
 332                                 allocator_free(this->logger);
 333                                 allocator_free(this);
 334                                 return NULL;
 335                         }
 336                         /* not all threads could be created, but at least one :-/ */
 337                         this->logger->log(this->logger, CONTROL, "could only create %d from requested %d threads: %s\n", current, pool_size, strerror(errno));
 338
 339                         this->pool_size = current;
 340                         return (thread_pool_t*)this;
 341                 }
 342         }
 343         return (thread_pool_t*)this;
 344 }