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