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