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 *pool_logger;
  63         /**
  64          * logger of the threadpool
  65          */
  66         logger_t *worker_logger;
  67 } ;
  68
  69
  70
  71 /**
  72  * implements private_thread_pool_t.function
  73  */
  74 static void job_processing(private_thread_pool_t *this)
  75 {
  76
  77         /* cancellation disabled by default */
  78         pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);
  79         this->worker_logger->log(this->worker_logger, CONTROL, "started working");
  80
  81         for (;;) {
  82                 job_t *job;
  83                 job_type_t job_type;
  84
  85                 global_job_queue->get(global_job_queue, &job);
  86                 job_type = job->get_type(job);
  87                 this->worker_logger->log(this->worker_logger, CONTROL|MORE, "got a job of type %s", mapping_find(job_type_m,job_type));
  88
  89                 /* process them here */
  90                 switch (job_type)
  91                 {
  92                         case INCOMING_PACKET:
  93                         {
  94                                 packet_t *packet;
  95                                 message_t *message;
  96                                 ike_sa_t *ike_sa;
  97                                 ike_sa_id_t *ike_sa_id;
  98                                 status_t status;
  99                                 incoming_packet_job_t *incoming_packet_job = (incoming_packet_job_t *)job;
 100
 101
 102                                 if (incoming_packet_job->get_packet(incoming_packet_job,&packet) != SUCCESS)
 103                                 {
 104                                         this->worker_logger->log(this->worker_logger, ERROR, "packet in job %s could not be retrieved!",
 105                                                                                 mapping_find(job_type_m,job_type));
 106                                         break;
 107                                 }
 108                                 message = message_create_from_packet(packet);
 109                                 if (message == NULL)
 110                                 {
 111                                         this->worker_logger->log(this->worker_logger, ERROR, "message could not be created from packet!",
 112                                                                                 mapping_find(job_type_m,job_type));
 113                                         packet->destroy(packet);
 114                                         break;
 115                                 }
 116
 117                                 status = message->parse_header(message);
 118                                 if (status != SUCCESS)
 119                                 {
 120                                         this->worker_logger->log(this->worker_logger, ERROR, "message header could not be verified!");
 121                                         message->destroy(message);
 122                                         break;
 123                                 }
 124
 125                                 this->worker_logger->log(this->worker_logger, CONTROL|MOST, "message is a %s %s",
 126                                                                         mapping_find(exchange_type_m, message->get_exchange_type(message)),
 127                                                                         message->get_request(message) ? "request" : "reply");
 128
 129                                 if ((message->get_major_version(message) != IKE_MAJOR_VERSION) ||
 130                                         (message->get_minor_version(message) != IKE_MINOR_VERSION))
 131                                 {
 132                                         this->worker_logger->log(this->worker_logger, ERROR, "IKE version %d.%d not supported",
 133                                                                                                 message->get_major_version(message),
 134                                                                                                 message->get_minor_version(message));
 135                                         /* Todo send notify */
 136                                 }
 137
 138                                 status = message->get_ike_sa_id(message, &ike_sa_id);
 139                                 if (status != SUCCESS)
 140                                 {
 141                                         this->worker_logger->log(this->worker_logger, ERROR, "IKE SA ID of message could not be created!");
 142                                         message->destroy(message);
 143                                         break;
 144                                 }
 145                                 /* we must switch the initiator flag here, because the sender
 146                                  * interprets this flag the other way round
 147                                  */
 148                                 ike_sa_id->switch_initiator(ike_sa_id);
 149
 150                                 this->worker_logger->log(this->worker_logger, CONTROL|MOST, "checking out IKE SA %lld:%lld, role %s",
 151                                                                         ike_sa_id->get_initiator_spi(ike_sa_id),
 152                                                                         ike_sa_id->get_responder_spi(ike_sa_id),
 153                                                                         ike_sa_id->is_initiator(ike_sa_id) ? "initiator" : "responder");
 154
 155                                 status = global_ike_sa_manager->checkout(global_ike_sa_manager,ike_sa_id, &ike_sa);
 156                                 if (status != SUCCESS)
 157                                 {
 158                                         this->worker_logger->log(this->worker_logger, ERROR, "IKE SA could not be checked out");
 159                                         ike_sa_id->destroy(ike_sa_id);
 160                                         message->destroy(message);
 161                                         break;
 162                                 }
 163
 164                                 status = ike_sa->process_message(ike_sa, message);
 165                                 if (status != SUCCESS)
 166                                 {
 167                                         this->worker_logger->log(this->worker_logger, ERROR, "message could not be processed by IKE SA");
 168                                 }
 169
 170                                 this->worker_logger->log(this->worker_logger, CONTROL|MOST, "checking in IKE SA %lld:%lld, role %s",
 171                                                                         ike_sa_id->get_initiator_spi(ike_sa_id),
 172                                                                         ike_sa_id->get_responder_spi(ike_sa_id),
 173                                                                         ike_sa_id->is_initiator(ike_sa_id) ? "initiator" : "responder");
 174
 175                                 status = global_ike_sa_manager->checkin(global_ike_sa_manager, ike_sa);
 176                                 if (status != SUCCESS)
 177                                 {
 178                                         this->worker_logger->log(this->worker_logger, ERROR, "checkin of IKE SA failed");
 179                                 }
 180                                 message->destroy(message);
 181                                 ike_sa_id->destroy(ike_sa_id);
 182
 183                                 break;
 184                         }
 185                         case INITIATE_IKE_SA:
 186                         {
 187                                 /*
 188                                  * Initiatie an IKE_SA:
 189                                  * - is defined by a name of a configuration
 190                                  * - create an empty IKE_SA via manager
 191                                  * - call initiate_connection on this sa
 192                                  */
 193                                 initiate_ike_sa_job_t *initiate_job;
 194                                 ike_sa_id_t *ike_sa_id;
 195                                 ike_sa_t *ike_sa;
 196                                 status_t status;
 197
 198                                 initiate_job = (initiate_ike_sa_job_t *)job;
 199
 200                                 ike_sa_id = ike_sa_id_create(0, 0, TRUE);
 201                                 if (ike_sa_id == NULL)
 202                                 {
 203                                         this->worker_logger->log(this->worker_logger, ERROR, "%s by creating ike_sa_id_t, job rejected.",
 204                                                                                 mapping_find(status_m, status));
 205                                         break;
 206                                 }
 207
 208                                 this->worker_logger->log(this->worker_logger, CONTROL|MOST, "checking out IKE SA %lld:%lld, role %s",
 209                                                                         ike_sa_id->get_initiator_spi(ike_sa_id),
 210                                                                         ike_sa_id->get_responder_spi(ike_sa_id),
 211                                                                         ike_sa_id->is_initiator(ike_sa_id) ? "initiator" : "responder");
 212
 213                                 status = global_ike_sa_manager->checkout(global_ike_sa_manager, ike_sa_id, &ike_sa);
 214                                 ike_sa_id->destroy(ike_sa_id);
 215                                 if (status != SUCCESS)
 216                                 {
 217                                         this->worker_logger->log(this->worker_logger, ERROR, "%s by checking out new IKE_SA, job rejected.",
 218                                                                                 mapping_find(status_m, status));
 219                                         break;
 220                                 }
 221
 222
 223                                 this->worker_logger->log(this->worker_logger, CONTROL|MOST, "initializing connection \"%s\"",
 224                                                                         initiate_job->get_configuration_name(initiate_job));
 225                                 status = ike_sa->initialize_connection(ike_sa, initiate_job->get_configuration_name(initiate_job));
 226                                 if (status != SUCCESS)
 227                                 {
 228                                         this->worker_logger->log(this->worker_logger, ERROR, "%s by initialize_conection, job and rejected, IKE_SA deleted.",
 229                                                                                 mapping_find(status_m, status));
 230                                         global_ike_sa_manager->checkin_and_delete(global_ike_sa_manager, ike_sa);
 231                                         break;
 232                                 }
 233
 234                                 this->worker_logger->log(this->worker_logger, CONTROL|MOST, "checking in IKE SA");
 235                                 status = global_ike_sa_manager->checkin(global_ike_sa_manager, ike_sa);
 236                                 if (status != SUCCESS)
 237                                 {
 238                                         this->worker_logger->log(this->worker_logger, ERROR, "%s could not checkin IKE_SA.",
 239                                                                                 mapping_find(status_m, status));
 240                                 }
 241                                 break;
 242                         }
 243                         case RETRANSMIT_REQUEST:
 244                         {
 245                                 this->worker_logger->log(this->worker_logger, ERROR, "job of type %s not supported!", mapping_find(job_type_m,job_type));
 246                                 break;
 247                         }
 248
 249                         case DELETE_IKE_SA:
 250                         {
 251                                 delete_ike_sa_job_t *delete_ike_sa_job = (delete_ike_sa_job_t*) job;
 252                                 ike_sa_id_t *ike_sa_id = delete_ike_sa_job->get_ike_sa_id(delete_ike_sa_job);
 253                                 status_t status;
 254
 255
 256                                 this->worker_logger->log(this->worker_logger, CONTROL|MOST, "deleting IKE SA %lld:%lld, role %s",
 257                                                                         ike_sa_id->get_initiator_spi(ike_sa_id),
 258                                                                         ike_sa_id->get_responder_spi(ike_sa_id),
 259                                                                         ike_sa_id->is_initiator(ike_sa_id) ? "initiator" : "responder");
 260
 261                                 status = global_ike_sa_manager->delete(global_ike_sa_manager, ike_sa_id);
 262                                 if (status != SUCCESS)
 263                                 {
 264                                         this->worker_logger->log(this->worker_logger, ERROR, "could not delete IKE_SA (%s)",
 265                                                                                 mapping_find(status_m, status));
 266                                 }
 267                                 break;
 268
 269                         }
 270                 }
 271                 job->destroy(job);
 272         }
 273
 274 }
 275
 276 /**
 277  * implementation of thread_pool_t.get_pool_size
 278  */
 279 static size_t get_pool_size(private_thread_pool_t *this)
 280 {
 281         return this->pool_size;
 282 }
 283
 284 /**
 285  * Implementation of thread_pool_t.destroy
 286  */
 287 static status_t destroy(private_thread_pool_t *this)
 288 {
 289         int current;
 290         /* flag thread for termination */
 291         for (current = 0; current < this->pool_size; current++) {
 292                 this->pool_logger->log(this->pool_logger, CONTROL, "cancelling thread %u", this->threads[current]);
 293                 pthread_cancel(this->threads[current]);
 294         }
 295
 296         /* wait for all threads */
 297         for (current = 0; current < this->pool_size; current++) {
 298                 pthread_join(this->threads[current], NULL);
 299                 this->pool_logger->log(this->pool_logger, CONTROL, "thread %u terminated", this->threads[current]);
 300         }
 301
 302         /* free mem */
 303         global_logger_manager->destroy_logger(global_logger_manager, this->pool_logger);
 304         global_logger_manager->destroy_logger(global_logger_manager, this->worker_logger);
 305         allocator_free(this->threads);
 306         allocator_free(this);
 307         return SUCCESS;
 308 }
 309
 310 #include <stdio.h>
 311
 312 /*
 313  * see header
 314  */
 315 thread_pool_t *thread_pool_create(size_t pool_size)
 316 {
 317         int current;
 318
 319         private_thread_pool_t *this = allocator_alloc_thing(private_thread_pool_t);
 320
 321         /* fill in public fields */
 322         this->public.destroy = (status_t(*)(thread_pool_t*))destroy;
 323         this->public.get_pool_size = (size_t(*)(thread_pool_t*))get_pool_size;
 324
 325         this->function = job_processing;
 326         this->pool_size = pool_size;
 327
 328         this->threads = allocator_alloc(sizeof(pthread_t) * pool_size);
 329         if (this->threads == NULL)
 330         {
 331                 allocator_free(this);
 332                 return NULL;
 333         }
 334         this->pool_logger = global_logger_manager->create_logger(global_logger_manager,THREAD_POOL,NULL);
 335         if (this->threads == NULL)
 336         {
 337                 allocator_free(this);
 338                 allocator_free(this->threads);
 339                 return NULL;
 340         }
 341         this->worker_logger = global_logger_manager->create_logger(global_logger_manager,WORKER,NULL);
 342         if (this->threads == NULL)
 343         {
 344                 global_logger_manager->destroy_logger(global_logger_manager, this->pool_logger);
 345                 allocator_free(this);
 346                 allocator_free(this->threads);
 347                 return NULL;
 348         }
 349
 350         /* try to create as many threads as possible, up tu pool_size */
 351         for (current = 0; current < pool_size; current++)
 352         {
 353                 if (pthread_create(&(this->threads[current]), NULL, (void*(*)(void*))this->function, this) == 0)
 354                 {
 355                         this->pool_logger->log(this->pool_logger, CONTROL, "thread %u created", this->threads[current]);
 356                 }
 357                 else
 358                 {
 359                         /* creation failed, is it the first one? */
 360                         if (current == 0)
 361                         {
 362                                 this->pool_logger->log(this->pool_logger, ERROR, "could not create any thread: %s\n", strerror(errno));
 363                                 global_logger_manager->destroy_logger(global_logger_manager, this->pool_logger);
 364                                 global_logger_manager->destroy_logger(global_logger_manager, this->worker_logger);
 365                                 allocator_free(this->threads);
 366                                 allocator_free(this);
 367                                 return NULL;
 368                         }
 369                         /* not all threads could be created, but at least one :-/ */
 370                         this->pool_logger->log(this->pool_logger, CONTROL, "could only create %d from requested %d threads: %s\n", current, pool_size, strerror(errno));
 371
 372                         this->pool_size = current;
 373                         return (thread_pool_t*)this;
 374                 }
 375         }
 376         return (thread_pool_t*)this;
 377 }