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