Source/charon/threads/thread_pool.c

   1 /**
   2  * @file thread_pool.c
   3  *
   4  * @brief Implementation of thread_pool_t.
   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 <daemon.h>
  31 #include <queues/job_queue.h>
  32 #include <queues/jobs/delete_half_open_ike_sa_job.h>
  33 #include <queues/jobs/delete_established_ike_sa_job.h>
  34 #include <queues/jobs/incoming_packet_job.h>
  35 #include <queues/jobs/initiate_ike_sa_job.h>
  36 #include <queues/jobs/retransmit_request_job.h>
  37 #include <encoding/payloads/notify_payload.h>
  38 #include <utils/logger.h>
  39
  40
  41 typedef struct private_thread_pool_t private_thread_pool_t;
  42
  43 /**
  44  * @brief Private data of thread_pool_t class.
  45  */
  46 struct private_thread_pool_t {
  47         /**
  48          * Public thread_pool_t interface.
  49          */
  50         thread_pool_t public;
  51
  52         /**
  53          * @brief Main processing function for worker threads.
  54          *
  55          * Gets a job from the job queue and calls corresponding
  56          * function for processing.
  57          *
  58          * @param this  calling object
  59          */
  60         void (*process_jobs) (private_thread_pool_t *this);
  61
  62         /**
  63          * @brief Process a INCOMING_PACKET job.
  64          *
  65          * @param this  calling object
  66          * @param job   incoming_packet_job_t object
  67          */
  68         void (*process_incoming_packet_job) (private_thread_pool_t *this, incoming_packet_job_t *job);
  69
  70         /**
  71          * @brief Process a INITIATE_IKE_SA job.
  72          *
  73          * @param this  calling object
  74          * @param job   initiate_ike_sa_job_t object
  75          */
  76         void (*process_initiate_ike_sa_job) (private_thread_pool_t *this, initiate_ike_sa_job_t *job);
  77
  78         /**
  79          * @brief Process a DELETE_HALF_OPEN_IKE_SA job.
  80          *
  81          * @param this  calling object
  82          * @param job   delete__half_open_ike_sa_job_t object
  83          */
  84         void (*process_delete_half_open_ike_sa_job) (private_thread_pool_t *this, delete_half_open_ike_sa_job_t *job);
  85
  86         /**
  87          * @brief Process a DELETE_ESTABLISHED_IKE_SA job.
  88          *
  89          * @param this  calling object
  90          * @param job   delete_established_ike_sa_job_t object
  91          */
  92         void (*process_delete_established_ike_sa_job) (private_thread_pool_t *this, delete_established_ike_sa_job_t *job);
  93
  94         /**
  95          * @brief Process a RETRANSMIT_REQUEST job.
  96          *
  97          * @param this  calling object
  98          * @param job   retransmit_request_job_t object
  99          */
 100         void (*process_retransmit_request_job) (private_thread_pool_t *this, retransmit_request_job_t *job);
 101
 102         /**
 103          * Creates a job of type DELETE_HALF_OPEN_IKE_SA.
 104          *
 105          * This job is used to delete IKE_SA's which are still in state INITIATOR_INIT,
 106          * RESPONDER_INIT, IKE_AUTH_REQUESTED, IKE_INIT_REQUESTED or IKE_INIT_RESPONDED.
 107          *
 108          * @param ike_sa_id             ID of IKE_SA to delete
 109          * @param delay                 Delay in ms after a half open IKE_SA gets deleted!
 110          */
 111         void (*create_delete_half_open_ike_sa_job) (private_thread_pool_t *this,ike_sa_id_t *ike_sa_id, u_int32_t delay);
 112
 113         /**
 114          * Number of running threads.
 115          */
 116         size_t pool_size;
 117
 118         /**
 119          * Array of thread ids.
 120          */
 121         pthread_t *threads;
 122
 123         /**
 124          * Logger of the thread pool.
 125          */
 126         logger_t *pool_logger;
 127
 128         /**
 129          * Logger of the worker threads.
 130          */
 131         logger_t *worker_logger;
 132 } ;
 133
 134 /**
 135  * Implementation of private_thread_pool_t.process_jobs.
 136  */
 137 static void process_jobs(private_thread_pool_t *this)
 138 {
 139         job_t *job;
 140         job_type_t job_type;
 141         timeval_t start_time;
 142         timeval_t end_time;
 143
 144         /* cancellation disabled by default */
 145         pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);
 146
 147         this->worker_logger->log(this->worker_logger, CONTROL, "Worker thread running, thread_id: %u", (int)pthread_self());
 148
 149         for (;;) {
 150
 151                 job = charon->job_queue->get(charon->job_queue);
 152                 job_type = job->get_type(job);
 153                 this->worker_logger->log(this->worker_logger, CONTROL|LEVEL2, "Process job of type %s",
 154                                                                  mapping_find(job_type_m,job_type));
 155                 gettimeofday(&start_time,NULL);
 156                 switch (job_type)
 157                 {
 158                         case INCOMING_PACKET:
 159                         {
 160                                 this->process_incoming_packet_job(this, (incoming_packet_job_t*)job);
 161                                 job->destroy(job);
 162                                 break;
 163                         }
 164                         case INITIATE_IKE_SA:
 165                         {
 166                                 this->process_initiate_ike_sa_job(this, (initiate_ike_sa_job_t*)job);
 167                                 job->destroy(job);
 168                                 break;
 169                         }
 170                         case DELETE_HALF_OPEN_IKE_SA:
 171                         {
 172                                 this->process_delete_half_open_ike_sa_job(this, (delete_half_open_ike_sa_job_t*)job);
 173                                 job->destroy(job);
 174                                 break;
 175                         }
 176                         case DELETE_ESTABLISHED_IKE_SA:
 177                         {
 178                                 this->process_delete_established_ike_sa_job(this, (delete_established_ike_sa_job_t*)job);
 179                                 job->destroy(job);
 180                                 break;
 181                         }
 182                         case RETRANSMIT_REQUEST:
 183                         {
 184                                 this->process_retransmit_request_job(this, (retransmit_request_job_t*)job);
 185                                 break;
 186                         }
 187                         default:
 188                         {
 189                                 this->worker_logger->log(this->worker_logger, ERROR, "Job of type %s not supported!",
 190                                                                                  mapping_find(job_type_m,job_type));
 191                                 job->destroy(job);
 192                                 break;
 193                         }
 194                 }
 195                 gettimeofday(&end_time,NULL);
 196
 197                 this->worker_logger->log(this->worker_logger, CONTROL | LEVEL2, "Processed job of type %s in %d us",
 198                                                                         mapping_find(job_type_m,job_type),
 199                                                                         (((end_time.tv_sec - start_time.tv_sec) * 1000000) + (end_time.tv_usec - start_time.tv_usec)));
 200
 201
 202         }
 203 }
 204
 205 /**
 206  * Implementation of private_thread_pool_t.process_incoming_packet_job.
 207  */
 208 static void process_incoming_packet_job(private_thread_pool_t *this, incoming_packet_job_t *job)
 209 {
 210         packet_t        *packet;
 211         message_t       *message;
 212         ike_sa_t        *ike_sa;
 213         ike_sa_id_t *ike_sa_id;
 214         status_t        status;
 215
 216
 217         packet = job->get_packet(job);
 218
 219         message = message_create_from_packet(packet);
 220
 221         status = message->parse_header(message);
 222         if (status != SUCCESS)
 223         {
 224                 this->worker_logger->log(this->worker_logger, ERROR, "Message header could not be verified!");
 225                 message->destroy(message);
 226                 return;
 227         }
 228
 229         this->worker_logger->log(this->worker_logger, CONTROL|LEVEL2, "Message is a %s %s",
 230                                                          mapping_find(exchange_type_m, message->get_exchange_type(message)),
 231                                                          message->get_request(message) ? "request" : "reply");
 232
 233         if ((message->get_major_version(message) != IKE_MAJOR_VERSION) ||
 234                         (message->get_minor_version(message) != IKE_MINOR_VERSION))
 235         {
 236                 this->worker_logger->log(this->worker_logger, ERROR | LEVEL2, "IKE version %d.%d not supported",
 237                                                                         message->get_major_version(message),
 238                                                                         message->get_minor_version(message));
 239                 /*
 240                  * This check is not handled in state_t object of IKE_SA to increase speed.
 241                  */
 242                 if ((message->get_exchange_type(message) == IKE_SA_INIT) && (message->get_request(message)))
 243                         {
 244                         message_t *response;
 245                         message->get_ike_sa_id(message, &ike_sa_id);
 246                         ike_sa_id->switch_initiator(ike_sa_id);
 247                         response = message_create_notify_reply(message->get_destination(message),
 248                                                                                                         message->get_source(message),
 249                                                                                                         IKE_SA_INIT,
 250                                                                                                         FALSE,ike_sa_id,INVALID_MAJOR_VERSION);
 251                         message->destroy(message);
 252                         ike_sa_id->destroy(ike_sa_id);
 253                         status = response->generate(response, NULL, NULL, &packet);
 254                         if (status != SUCCESS)
 255                         {
 256                                 this->worker_logger->log(this->worker_logger, ERROR, "Could not generate packet from message");
 257                                 response->destroy(response);
 258                                 return;
 259                         }
 260                         this->worker_logger->log(this->worker_logger, ERROR, "Send notify reply of type INVALID_MAJOR_VERSION");
 261                         charon->send_queue->add(charon->send_queue, packet);
 262                         response->destroy(response);
 263                         return;
 264                 }
 265                 message->destroy(message);
 266                 return;
 267         }
 268
 269         message->get_ike_sa_id(message, &ike_sa_id);
 270
 271         ike_sa_id->switch_initiator(ike_sa_id);
 272
 273         this->worker_logger->log(this->worker_logger, CONTROL|LEVEL3, "Checking out IKE SA %lld:%lld, role %s",
 274                                                          ike_sa_id->get_initiator_spi(ike_sa_id),
 275                                                          ike_sa_id->get_responder_spi(ike_sa_id),
 276                                                          ike_sa_id->is_initiator(ike_sa_id) ? "initiator" : "responder");
 277
 278         status = charon->ike_sa_manager->checkout(charon->ike_sa_manager,ike_sa_id, &ike_sa);
 279         if ((status != SUCCESS) && (status != CREATED))
 280         {
 281                 this->worker_logger->log(this->worker_logger, ERROR, "IKE SA could not be checked out");
 282                 ike_sa_id->destroy(ike_sa_id);
 283                 message->destroy(message);
 284
 285                 /*
 286                  * TODO send notify reply of type INVALID_IKE_SPI if SPI could not be found ?
 287                  */
 288
 289                 return;
 290         }
 291
 292         if (status == CREATED)
 293         {
 294                 this->worker_logger->log(this->worker_logger, CONTROL|LEVEL3,
 295                                                                  "Create Job to delete half open IKE_SA.");
 296                 this->create_delete_half_open_ike_sa_job(this,ike_sa_id,
 297                                 charon->configuration->get_half_open_ike_sa_timeout(charon->configuration));
 298         }
 299
 300         status = ike_sa->process_message(ike_sa, message);
 301
 302         this->worker_logger->log(this->worker_logger, CONTROL|LEVEL3, "%s IKE SA %lld:%lld, role %s",
 303                                                          (status == DELETE_ME) ? "Checkin and delete" : "Checkin",
 304                                                          ike_sa_id->get_initiator_spi(ike_sa_id),
 305                                                          ike_sa_id->get_responder_spi(ike_sa_id),
 306                                                          ike_sa_id->is_initiator(ike_sa_id) ? "initiator" : "responder");
 307         ike_sa_id->destroy(ike_sa_id);
 308
 309         if (status == DELETE_ME)
 310         {
 311                 status = charon->ike_sa_manager->checkin_and_delete(charon->ike_sa_manager, ike_sa);
 312         }
 313         else
 314         {
 315                 status = charon->ike_sa_manager->checkin(charon->ike_sa_manager, ike_sa);
 316         }
 317
 318         if (status != SUCCESS)
 319         {
 320                 this->worker_logger->log(this->worker_logger, ERROR, "Checkin of IKE SA failed!");
 321         }
 322         message->destroy(message);
 323 }
 324
 325 /**
 326  * Implementation of private_thread_pool_t.process_initiate_ike_sa_job.
 327  */
 328 static void process_initiate_ike_sa_job(private_thread_pool_t *this, initiate_ike_sa_job_t *job)
 329 {
 330         /*
 331          * Initiatie an IKE_SA:
 332          * - is defined by a name of a configuration
 333          * - create an empty IKE_SA via manager
 334          * - call initiate_connection on this sa
 335          */
 336         ike_sa_t *ike_sa;
 337         status_t status;
 338
 339
 340         this->worker_logger->log(this->worker_logger, CONTROL|LEVEL2, "Creating and checking out IKE SA");
 341         charon->ike_sa_manager->create_and_checkout(charon->ike_sa_manager, &ike_sa);
 342
 343         status = ike_sa->initiate_connection(ike_sa, job->get_connection(job));
 344         if (status != SUCCESS)
 345         {
 346                 this->worker_logger->log(this->worker_logger, ERROR, "Initiation returned %s, going to delete IKE_SA.",
 347                                                                  mapping_find(status_m, status));
 348                 charon->ike_sa_manager->checkin_and_delete(charon->ike_sa_manager, ike_sa);
 349                 return;
 350         }
 351
 352         this->worker_logger->log(this->worker_logger, CONTROL|LEVEL3, "Create Job to delete half open IKE_SA.");
 353         this->create_delete_half_open_ike_sa_job(this,ike_sa->get_id(ike_sa),
 354                                                                                          charon->configuration->get_half_open_ike_sa_timeout(charon->configuration));
 355
 356         this->worker_logger->log(this->worker_logger, CONTROL|LEVEL2, "Checking in IKE SA");
 357         status = charon->ike_sa_manager->checkin(charon->ike_sa_manager, ike_sa);
 358         if (status != SUCCESS)
 359         {
 360                 this->worker_logger->log(this->worker_logger, ERROR, "Could not checkin IKE_SA (%s)",
 361                                                                  mapping_find(status_m, status));
 362         }
 363 }
 364
 365 /**
 366  * Implementation of private_thread_pool_t.process_delete_ike_sa_job.
 367  */
 368 static void process_delete_half_open_ike_sa_job(private_thread_pool_t *this, delete_half_open_ike_sa_job_t *job)
 369 {
 370         ike_sa_id_t *ike_sa_id = job->get_ike_sa_id(job);
 371         ike_sa_t *ike_sa;
 372         status_t status;
 373         status = charon->ike_sa_manager->checkout(charon->ike_sa_manager,ike_sa_id, &ike_sa);
 374         if ((status != SUCCESS) && (status != CREATED))
 375         {
 376                 this->worker_logger->log(this->worker_logger, CONTROL | LEVEL3, "IKE SA seems to be allready deleted and so doesn't have to be deleted");
 377                 return;
 378         }
 379
 380
 381         switch (ike_sa->get_state(ike_sa))
 382         {
 383                 case INITIATOR_INIT:
 384                 case RESPONDER_INIT:
 385                 case IKE_SA_INIT_REQUESTED:
 386                 case IKE_SA_INIT_RESPONDED:
 387                 case IKE_AUTH_REQUESTED:
 388                 {
 389                         /* IKE_SA is half open and gets deleted! */
 390                         status = charon->ike_sa_manager->checkin_and_delete(charon->ike_sa_manager, ike_sa);
 391                         if (status != SUCCESS)
 392                         {
 393                                 this->worker_logger->log(this->worker_logger, ERROR, "Could not checkin and delete checked out IKE_SA!");
 394                         }
 395                         break;
 396                 }
 397                 default:
 398                 {
 399                         /* IKE_SA is established and so is not getting deleted! */
 400                         status = charon->ike_sa_manager->checkin(charon->ike_sa_manager, ike_sa);
 401                         if (status != SUCCESS)
 402                         {
 403                                 this->worker_logger->log(this->worker_logger, ERROR, "Could not checkin a checked out IKE_SA!");
 404                         }
 405                         break;
 406                 }
 407         }
 408 }
 409
 410 /**
 411  * Implementation of private_thread_pool_t.process_delete_established_ike_sa_job.
 412  */
 413 static void process_delete_established_ike_sa_job(private_thread_pool_t *this, delete_established_ike_sa_job_t *job)
 414 {
 415         ike_sa_id_t *ike_sa_id = job->get_ike_sa_id(job);
 416         ike_sa_t *ike_sa;
 417         status_t status;
 418         status = charon->ike_sa_manager->checkout(charon->ike_sa_manager,ike_sa_id, &ike_sa);
 419         if ((status != SUCCESS) && (status != CREATED))
 420         {
 421                 this->worker_logger->log(this->worker_logger, CONTROL | LEVEL3, "IKE SA seems to be allready deleted and so doesn't have to be deleted");
 422                 return;
 423         }
 424
 425         switch (ike_sa->get_state(ike_sa))
 426         {
 427                 case INITIATOR_INIT:
 428                 case RESPONDER_INIT:
 429                 case IKE_SA_INIT_REQUESTED:
 430                 case IKE_SA_INIT_RESPONDED:
 431                 case IKE_AUTH_REQUESTED:
 432                 {
 433                         break;
 434                 }
 435                 default:
 436                 {
 437                         this->worker_logger->log(this->worker_logger, CONTROL, "Send delete request for IKE_SA.");
 438                         ike_sa->send_delete_ike_sa_request(ike_sa);
 439                         break;
 440                 }
 441         }
 442         this->worker_logger->log(this->worker_logger, CONTROL, "Delete established IKE_SA.");
 443         status = charon->ike_sa_manager->checkin_and_delete(charon->ike_sa_manager, ike_sa);
 444         if (status != SUCCESS)
 445         {
 446                 this->worker_logger->log(this->worker_logger, ERROR, "Could not checkin and delete checked out IKE_SA!");
 447         }
 448 }
 449
 450
 451 /**
 452  * Implementation of private_thread_pool_t.process_retransmit_request_job.
 453  */
 454 static void process_retransmit_request_job(private_thread_pool_t *this, retransmit_request_job_t *job)
 455 {
 456
 457         ike_sa_id_t *ike_sa_id = job->get_ike_sa_id(job);
 458         u_int32_t message_id = job->get_message_id(job);
 459         bool stop_retransmitting = FALSE;
 460         u_int32_t timeout;
 461         ike_sa_t *ike_sa;
 462         status_t status;
 463
 464         this->worker_logger->log(this->worker_logger, CONTROL|LEVEL2, "Checking out IKE SA %lld:%lld, role %s",
 465                                                          ike_sa_id->get_initiator_spi(ike_sa_id),
 466                                                          ike_sa_id->get_responder_spi(ike_sa_id),
 467                                                          ike_sa_id->is_initiator(ike_sa_id) ? "initiator" : "responder");
 468
 469         status = charon->ike_sa_manager->checkout(charon->ike_sa_manager,ike_sa_id, &ike_sa);
 470         if ((status != SUCCESS) && (status != CREATED))
 471         {
 472                 job->destroy(job);
 473                 this->worker_logger->log(this->worker_logger, ERROR, "IKE SA could not be checked out. Allready deleted?");
 474                 return;
 475         }
 476
 477         status = ike_sa->retransmit_request(ike_sa, message_id);
 478
 479         if (status != SUCCESS)
 480         {
 481                 this->worker_logger->log(this->worker_logger, CONTROL | LEVEL3, "Message doesn't have to be retransmitted");
 482                 stop_retransmitting = TRUE;
 483         }
 484
 485         this->worker_logger->log(this->worker_logger, CONTROL|LEVEL2, "Checkin IKE SA %lld:%lld, role %s",
 486                                                          ike_sa_id->get_initiator_spi(ike_sa_id),
 487                                                          ike_sa_id->get_responder_spi(ike_sa_id),
 488                                                          ike_sa_id->is_initiator(ike_sa_id) ? "initiator" : "responder");
 489
 490         status = charon->ike_sa_manager->checkin(charon->ike_sa_manager, ike_sa);
 491         if (status != SUCCESS)
 492         {
 493                 this->worker_logger->log(this->worker_logger, ERROR, "Checkin of IKE SA failed!");
 494         }
 495
 496         if (stop_retransmitting)
 497         {
 498                 job->destroy(job);
 499                 return;
 500         }
 501
 502         job->increase_retransmit_count(job);
 503         status = charon->configuration->get_retransmit_timeout (charon->configuration,job->get_retransmit_count(job),&timeout);
 504         if (status != SUCCESS)
 505         {
 506                 this->worker_logger->log(this->worker_logger, CONTROL | LEVEL2, "Message will not be anymore retransmitted");
 507                 job->destroy(job);
 508                 /*
 509                  * TODO delete IKE_SA ?
 510                  */
 511                 return;
 512         }
 513         charon->event_queue->add_relative(charon->event_queue,(job_t *) job,timeout);
 514 }
 515
 516
 517
 518 /**
 519  * Implementation of private_thread_pool_t.create_delete_half_open_ike_sa_job.
 520  */
 521 static void create_delete_half_open_ike_sa_job(private_thread_pool_t *this,ike_sa_id_t *ike_sa_id, u_int32_t delay)
 522 {
 523         job_t *delete_job;
 524
 525         this->worker_logger->log(this->worker_logger, CONTROL | LEVEL2, "Going to create job to delete half open IKE_SA in %d ms", delay);
 526
 527         delete_job = (job_t *) delete_half_open_ike_sa_job_create(ike_sa_id);
 528         charon->event_queue->add_relative(charon->event_queue,delete_job, delay);
 529 }
 530
 531
 532 /**
 533  * Implementation of thread_pool_t.get_pool_size.
 534  */
 535 static size_t get_pool_size(private_thread_pool_t *this)
 536 {
 537         return this->pool_size;
 538 }
 539
 540 /**
 541  * Implementation of thread_pool_t.destroy.
 542  */
 543 static void destroy(private_thread_pool_t *this)
 544 {
 545         int current;
 546         /* flag thread for termination */
 547         for (current = 0; current < this->pool_size; current++) {
 548                 this->pool_logger->log(this->pool_logger, CONTROL, "cancelling worker thread #%d", current+1);
 549                 pthread_cancel(this->threads[current]);
 550         }
 551
 552         /* wait for all threads */
 553         for (current = 0; current < this->pool_size; current++) {
 554                 if (pthread_join(this->threads[current], NULL) == 0)
 555                 {
 556                         this->pool_logger->log(this->pool_logger, CONTROL, "worker thread #%d terminated", current+1);
 557                 }
 558                 else
 559                 {
 560                         this->pool_logger->log(this->pool_logger, ERROR, "could not terminate worker thread #%d", current+1);
 561                 }
 562         }
 563
 564         /* free mem */
 565         free(this->threads);
 566         free(this);
 567 }
 568
 569 /*
 570  * Described in header.
 571  */
 572 thread_pool_t *thread_pool_create(size_t pool_size)
 573 {
 574         int current;
 575
 576         private_thread_pool_t *this = malloc_thing(private_thread_pool_t);
 577
 578         /* fill in public fields */
 579         this->public.destroy = (void(*)(thread_pool_t*))destroy;
 580         this->public.get_pool_size = (size_t(*)(thread_pool_t*))get_pool_size;
 581
 582         this->process_jobs = process_jobs;
 583         this->process_initiate_ike_sa_job = process_initiate_ike_sa_job;
 584         this->process_delete_half_open_ike_sa_job = process_delete_half_open_ike_sa_job;
 585         this->process_delete_established_ike_sa_job = process_delete_established_ike_sa_job;
 586         this->process_incoming_packet_job = process_incoming_packet_job;
 587         this->process_retransmit_request_job = process_retransmit_request_job;
 588         this->create_delete_half_open_ike_sa_job = create_delete_half_open_ike_sa_job;
 589
 590         this->pool_size = pool_size;
 591
 592         this->threads = malloc(sizeof(pthread_t) * pool_size);
 593
 594         this->pool_logger = logger_manager->get_logger(logger_manager, THREAD_POOL);
 595
 596         this->worker_logger = logger_manager->get_logger(logger_manager, WORKER);
 597
 598         /* try to create as many threads as possible, up tu pool_size */
 599         for (current = 0; current < pool_size; current++)
 600         {
 601                 if (pthread_create(&(this->threads[current]), NULL, (void*(*)(void*))this->process_jobs, this) == 0)
 602                 {
 603                         this->pool_logger->log(this->pool_logger, CONTROL, "Created worker thread #%d", current+1);
 604                 }
 605                 else
 606                 {
 607                         /* creation failed, is it the first one? */
 608                         if (current == 0)
 609                         {
 610                                 this->pool_logger->log(this->pool_logger, ERROR, "Could not create any thread");
 611                                 free(this->threads);
 612                                 free(this);
 613                                 return NULL;
 614                         }
 615                         /* not all threads could be created, but at least one :-/ */
 616                         this->pool_logger->log(this->pool_logger, ERROR, "Could only create %d from requested %d threads!", current, pool_size);
 617
 618                         this->pool_size = current;
 619                         return (thread_pool_t*)this;
 620                 }
 621         }
 622         return (thread_pool_t*)this;
 623 }