7054ae88caa62926dd7b7b51f752e48e211ad62f
[strongswan.git] / 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 }