661d0fdb7aaa143d707f87434d4b5eb41b2a119e
[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 <daemon.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 = charon->job_queue->get(charon->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 = charon->ike_sa_manager->checkout(charon->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 if (status == DELETE_ME)
219 {
220 status = charon->ike_sa_manager->checkin_and_delete(charon->ike_sa_manager, ike_sa);
221 }
222 else
223 {
224 status = charon->ike_sa_manager->checkin(charon->ike_sa_manager, ike_sa);
225 }
226
227 if (status != SUCCESS)
228 {
229 this->worker_logger->log(this->worker_logger, ERROR, "checkin of IKE SA failed!");
230 }
231 message->destroy(message);
232 }
233
234 /**
235 * implementation of private_thread_pool_t.process_initiate_ike_sa_job
236 */
237 static void process_initiate_ike_sa_job(private_thread_pool_t *this, initiate_ike_sa_job_t *job)
238 {
239 /*
240 * Initiatie an IKE_SA:
241 * - is defined by a name of a configuration
242 * - create an empty IKE_SA via manager
243 * - call initiate_connection on this sa
244 */
245 ike_sa_t *ike_sa;
246 status_t status;
247
248
249 this->worker_logger->log(this->worker_logger, CONTROL|MOST, "create and checking out IKE SA");
250
251 charon->ike_sa_manager->create_and_checkout(charon->ike_sa_manager, &ike_sa);
252
253 this->worker_logger->log(this->worker_logger, CONTROL|MOST, "initializing connection \"%s\"",
254 job->get_configuration_name(job));
255 status = ike_sa->initialize_connection(ike_sa, job->get_configuration_name(job));
256 if (status != SUCCESS)
257 {
258 this->worker_logger->log(this->worker_logger, ERROR, "%s by initialize_conection, going to delete IKE_SA.",
259 mapping_find(status_m, status));
260 charon->ike_sa_manager->checkin_and_delete(charon->ike_sa_manager, ike_sa);
261 return;
262 }
263
264 this->worker_logger->log(this->worker_logger, CONTROL|MOST, "checking in IKE SA");
265 status = charon->ike_sa_manager->checkin(charon->ike_sa_manager, ike_sa);
266 if (status != SUCCESS)
267 {
268 this->worker_logger->log(this->worker_logger, ERROR, "%s could not checkin IKE_SA.",
269 mapping_find(status_m, status));
270 }
271 }
272
273 /**
274 * implementation of private_thread_pool_t.process_delete_ike_sa_job
275 */
276 static void process_delete_ike_sa_job(private_thread_pool_t *this, delete_ike_sa_job_t *job)
277 {
278 status_t status;
279 ike_sa_id_t *ike_sa_id = job->get_ike_sa_id(job);
280
281 this->worker_logger->log(this->worker_logger, CONTROL|MOST, "deleting IKE SA %lld:%lld, role %s",
282 ike_sa_id->get_initiator_spi(ike_sa_id),
283 ike_sa_id->get_responder_spi(ike_sa_id),
284 ike_sa_id->is_initiator(ike_sa_id) ? "initiator" : "responder");
285
286 status = charon->ike_sa_manager->delete(charon->ike_sa_manager, ike_sa_id);
287 if (status != SUCCESS)
288 {
289 this->worker_logger->log(this->worker_logger, ERROR, "could not delete IKE_SA (%s)",
290 mapping_find(status_m, status));
291 }
292 }
293
294
295 /**
296 * implementation of thread_pool_t.get_pool_size
297 */
298 static size_t get_pool_size(private_thread_pool_t *this)
299 {
300 return this->pool_size;
301 }
302
303 /**
304 * Implementation of thread_pool_t.destroy
305 */
306 static void destroy(private_thread_pool_t *this)
307 {
308 int current;
309 /* flag thread for termination */
310 for (current = 0; current < this->pool_size; current++) {
311 this->pool_logger->log(this->pool_logger, CONTROL, "cancelling worker a thread #%d", current+1);
312 pthread_cancel(this->threads[current]);
313 }
314
315 /* wait for all threads */
316 for (current = 0; current < this->pool_size; current++) {
317 pthread_join(this->threads[current], NULL);
318 this->pool_logger->log(this->pool_logger, CONTROL, "worker thread #%d terminated", current+1);
319 }
320
321 /* free mem */
322 charon->logger_manager->destroy_logger(charon->logger_manager, this->pool_logger);
323 charon->logger_manager->destroy_logger(charon->logger_manager, this->worker_logger);
324 allocator_free(this->threads);
325 allocator_free(this);
326 }
327
328 /*
329 * see header
330 */
331 thread_pool_t *thread_pool_create(size_t pool_size)
332 {
333 int current;
334
335 private_thread_pool_t *this = allocator_alloc_thing(private_thread_pool_t);
336
337 /* fill in public fields */
338 this->public.destroy = (void(*)(thread_pool_t*))destroy;
339 this->public.get_pool_size = (size_t(*)(thread_pool_t*))get_pool_size;
340
341 this->process_jobs = process_jobs;
342 this->process_initiate_ike_sa_job = process_initiate_ike_sa_job;
343 this->process_delete_ike_sa_job = process_delete_ike_sa_job;
344 this->process_incoming_packet_job = process_incoming_packet_job;
345 this->pool_size = pool_size;
346
347 this->threads = allocator_alloc(sizeof(pthread_t) * pool_size);
348
349 this->pool_logger = charon->logger_manager->create_logger(charon->logger_manager,THREAD_POOL,NULL);
350
351 this->worker_logger = charon->logger_manager->create_logger(charon->logger_manager,WORKER,NULL);
352
353 /* try to create as many threads as possible, up tu pool_size */
354 for (current = 0; current < pool_size; current++)
355 {
356 if (pthread_create(&(this->threads[current]), NULL, (void*(*)(void*))this->process_jobs, this) == 0)
357 {
358 this->pool_logger->log(this->pool_logger, CONTROL, "created worker thread #%d", current+1);
359 }
360 else
361 {
362 /* creation failed, is it the first one? */
363 if (current == 0)
364 {
365 this->pool_logger->log(this->pool_logger, ERROR, "could not create any thread");
366 charon->logger_manager->destroy_logger(charon->logger_manager, this->pool_logger);
367 charon->logger_manager->destroy_logger(charon->logger_manager, this->worker_logger);
368 allocator_free(this->threads);
369 allocator_free(this);
370 return NULL;
371 }
372 /* not all threads could be created, but at least one :-/ */
373 this->pool_logger->log(this->pool_logger, ERROR, "could only create %d from requested %d threads!", current, pool_size);
374
375 this->pool_size = current;
376 return (thread_pool_t*)this;
377 }
378 }
379 return (thread_pool_t*)this;
380 }