4 * @brief Thread pool with some threads processing the job_queue.
9 * Copyright (C) 2005 Jan Hutter, Martin Willi
10 * Hochschule fuer Technik Rapperswil
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>.
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
28 #include "thread_pool.h"
31 #include "queues/job_queue.h"
32 #include "utils/allocator.h"
33 #include "utils/logger.h"
36 * @brief structure with private members for thread_pool_t
38 typedef struct private_thread_pool_s private_thread_pool_t
;
40 struct private_thread_pool_s
{
42 * inclusion of public members
46 * @brief Processing function of a worker thread
48 * @param this private_thread_pool_t-Object
50 void (*function
) (private_thread_pool_t
*this);
52 * number of running threads
60 * logger of the threadpool
68 * implements private_thread_pool_t.function
70 static void job_processing(private_thread_pool_t
*this)
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());
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
));
85 /* process them here */
93 ike_sa_id_t
*ike_sa_id
;
95 incoming_packet_job_t
*incoming_packet_job
= (incoming_packet_job_t
*)job
;
97 if (incoming_packet_job
->get_packet(incoming_packet_job
,&packet
) != SUCCESS
)
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
));
102 message
= message_create_from_packet(packet
);
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
);
110 status
= message
->parse_header(message
);
111 if (status
!= SUCCESS
)
113 this->logger
->log(this->logger
, CONTROL_MORE
, "thread %u: Message header could not be verified!", pthread_self());
114 message
->destroy(message
);
118 if ((message
->get_major_version(message
) != IKE_MAJOR_VERSION
) || (message
->get_minor_version(message
) != IKE_MINOR_VERSION
))
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 */
124 status
= message
->get_ike_sa_id(message
,&ike_sa_id
);
125 if (status
!= SUCCESS
)
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
);
132 status
= global_ike_sa_manager
->checkout(global_ike_sa_manager
,ike_sa_id
, &ike_sa
);
133 if (status
!= SUCCESS
)
135 this->logger
->log(this->logger
, CONTROL_MORE
, "thread %u: IKE SA could not be checked out", pthread_self());
136 message
->destroy(message
);
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
);
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
);
151 status
= ike_sa
->process_message (ike_sa
,message
);
152 if (status
!= SUCCESS
)
154 this->logger
->log(this->logger
, CONTROL_MORE
, "thread %u: Message could not be processed by IKE SA", pthread_self());
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());
161 message
->destroy(message
);
162 ike_sa_id
->destroy(ike_sa_id
);
166 case INITIATE_IKE_SA
:
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
174 initiate_ike_sa_job_t
*initiate_job
;
175 ike_sa_id_t
*ike_sa_id
;
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
));
183 ike_sa_id
= ike_sa_id_create(0, 0, TRUE
);
184 if (ike_sa_id
== NULL
)
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
));
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
)
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
));
200 status
= ike_sa
->initialize_connection(ike_sa
, initiate_job
->get_configuration_name(initiate_job
));
201 if (status
!= SUCCESS
)
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
);
209 status
= global_ike_sa_manager
->checkin(global_ike_sa_manager
, ike_sa
);
210 if (status
!= SUCCESS
)
212 this->logger
->log(this->logger
, ERROR
, "thread %u: %s could not checkin IKE_SA.",
213 pthread_self(), mapping_find(status_m
, status
));
217 case RETRANSMIT_REQUEST
:
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
));
229 * implementation of thread_pool_t.get_pool_size
231 static size_t get_pool_size(private_thread_pool_t
*this)
233 return this->pool_size
;
237 * Implementation of thread_pool_t.destroy
239 static status_t
destroy(private_thread_pool_t
*this)
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
]);
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
]);
255 global_logger_manager
->destroy_logger(global_logger_manager
, this->logger
);
256 allocator_free(this->threads
);
257 allocator_free(this);
266 thread_pool_t
*thread_pool_create(size_t pool_size
)
270 private_thread_pool_t
*this = allocator_alloc_thing(private_thread_pool_t
);
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
;
276 this->function
= job_processing
;
277 this->pool_size
= pool_size
;
279 this->threads
= allocator_alloc(sizeof(pthread_t
) * pool_size
);
280 if (this->threads
== NULL
)
282 allocator_free(this);
285 this->logger
= global_logger_manager
->create_logger(global_logger_manager
,THREAD_POOL
,NULL
);
286 if (this->threads
== NULL
)
288 allocator_free(this);
289 allocator_free(this->threads
);
293 /* try to create as many threads as possible, up tu pool_size */
294 for (current
= 0; current
< pool_size
; current
++)
296 if (pthread_create(&(this->threads
[current
]), NULL
, (void*(*)(void*))this->function
, this) == 0)
298 this->logger
->log(this->logger
, CONTROL
, "thread %u created", this->threads
[current
]);
302 /* creation failed, is it the first one? */
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);
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
));
314 this->pool_size
= current
;
315 return (thread_pool_t
*)this;
318 return (thread_pool_t
*)this;