e14679b14f8a5d1f17b8a2e89b02a823ec24fc4e
[strongswan.git] / src / charon / processing / processor.c
1 /*
2 * Copyright (C) 2005-2007 Martin Willi
3 * Copyright (C) 2005 Jan Hutter
4 * Hochschule fuer Technik Rapperswil
5 *
6 * This program is free software; you can redistribute it and/or modify it
7 * under the terms of the GNU General Public License as published by the
8 * Free Software Foundation; either version 2 of the License, or (at your
9 * option) any later version. See <http://www.fsf.org/copyleft/gpl.txt>.
10 *
11 * This program is distributed in the hope that it will be useful, but
12 * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
13 * or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
14 * for more details.
15 *
16 * $Id$
17 */
18
19 #include <stdlib.h>
20 #include <pthread.h>
21 #include <string.h>
22 #include <errno.h>
23
24 #include "processor.h"
25
26 #include <daemon.h>
27 #include <utils/linked_list.h>
28
29
30 typedef struct private_processor_t private_processor_t;
31
32 /**
33 * Private data of processor_t class.
34 */
35 struct private_processor_t {
36 /**
37 * Public processor_t interface.
38 */
39 processor_t public;
40
41 /**
42 * Number of running threads
43 */
44 u_int total_threads;
45
46 /**
47 * Desired number of threads
48 */
49 u_int desired_threads;
50
51 /**
52 * Number of threads waiting for work
53 */
54 u_int idle_threads;
55
56 /**
57 * The jobs are stored in a linked list
58 */
59 linked_list_t *list;
60
61 /**
62 * access to linked_list is locked through this mutex
63 */
64 pthread_mutex_t mutex;
65
66 /**
67 * Condvar to wait for new jobs
68 */
69 pthread_cond_t condvar;
70 };
71
72 static void process_jobs(private_processor_t *this);
73
74 /**
75 * restart a terminated thread
76 */
77 static void restart(private_processor_t *this)
78 {
79 pthread_t thread;
80
81 if (pthread_create(&thread, NULL, (void*)process_jobs, this) != 0)
82 {
83 this->total_threads--;
84 }
85 }
86
87 /**
88 * Process queued jobs, called by the worker threads
89 */
90 static void process_jobs(private_processor_t *this)
91 {
92 int oldstate;
93
94 pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &oldstate);
95
96 DBG2(DBG_JOB, "started worker thread, thread_ID: %06u", (int)pthread_self());
97
98 pthread_mutex_lock(&this->mutex);
99 while (this->desired_threads >= this->total_threads)
100 {
101 job_t *job;
102
103 if (this->list->get_count(this->list) == 0)
104 {
105 this->idle_threads++;
106 pthread_cond_wait(&this->condvar, &this->mutex);
107 this->idle_threads--;
108 continue;
109 }
110 this->list->remove_first(this->list, (void**)&job);
111 pthread_mutex_unlock(&this->mutex);
112 /* terminated threads are restarted, so we have a constant pool */
113 pthread_cleanup_push((void*)restart, this);
114 job->execute(job);
115 pthread_cleanup_pop(0);
116 pthread_mutex_lock(&this->mutex);
117 }
118 this->total_threads--;
119 pthread_cond_broadcast(&this->condvar);
120 pthread_mutex_unlock(&this->mutex);
121 }
122
123 /**
124 * Implementation of processor_t.get_total_threads.
125 */
126 static u_int get_total_threads(private_processor_t *this)
127 {
128 return this->total_threads;
129 }
130
131 /**
132 * Implementation of processor_t.get_idle_threads.
133 */
134 static u_int get_idle_threads(private_processor_t *this)
135 {
136 return this->idle_threads;
137 }
138
139 /**
140 * implements processor_t.get_job_load
141 */
142 static u_int get_job_load(private_processor_t *this)
143 {
144 u_int load;
145 pthread_mutex_lock(&this->mutex);
146 load = this->list->get_count(this->list);
147 pthread_mutex_unlock(&this->mutex);
148 return load;
149 }
150
151 /**
152 * implements function processor_t.queue_job
153 */
154 static void queue_job(private_processor_t *this, job_t *job)
155 {
156 pthread_mutex_lock(&this->mutex);
157 this->list->insert_last(this->list, job);
158 pthread_mutex_unlock(&this->mutex);
159 pthread_cond_signal(&this->condvar);
160 }
161
162 /**
163 * Implementation of processor_t.set_threads.
164 */
165 static void set_threads(private_processor_t *this, u_int count)
166 {
167 pthread_mutex_lock(&this->mutex);
168 if (count > this->total_threads)
169 { /* increase thread count */
170 int i;
171 pthread_t current;
172
173 this->desired_threads = count;
174 DBG1(DBG_JOB, "spawning %d worker threads", count - this->total_threads);
175 for (i = this->total_threads; i < count; i++)
176 {
177 if (pthread_create(&current, NULL, (void*)process_jobs, this) == 0)
178 {
179 this->total_threads++;
180 }
181 }
182 }
183 else if (count < this->total_threads)
184 { /* decrease thread count */
185 this->desired_threads = count;
186 }
187 pthread_mutex_unlock(&this->mutex);
188 }
189
190 /**
191 * Implementation of processor_t.destroy.
192 */
193 static void destroy(private_processor_t *this)
194 {
195 set_threads(this, 0);
196 pthread_mutex_lock(&this->mutex);
197 while (this->total_threads > 0)
198 {
199 pthread_cond_broadcast(&this->condvar);
200 pthread_cond_wait(&this->condvar, &this->mutex);
201 }
202 pthread_mutex_unlock(&this->mutex);
203 this->list->destroy_offset(this->list, offsetof(job_t, destroy));
204 free(this);
205 }
206
207 /*
208 * Described in header.
209 */
210 processor_t *processor_create(size_t pool_size)
211 {
212 private_processor_t *this = malloc_thing(private_processor_t);
213
214 this->public.get_total_threads = (u_int(*)(processor_t*))get_total_threads;
215 this->public.get_idle_threads = (u_int(*)(processor_t*))get_idle_threads;
216 this->public.get_job_load = (u_int(*)(processor_t*))get_job_load;
217 this->public.queue_job = (void(*)(processor_t*, job_t*))queue_job;
218 this->public.set_threads = (void(*)(processor_t*, u_int))set_threads;
219 this->public.destroy = (void(*)(processor_t*))destroy;
220
221 this->list = linked_list_create();
222 pthread_mutex_init(&this->mutex, NULL);
223 pthread_cond_init(&this->condvar, NULL);
224 this->total_threads = 0;
225 this->desired_threads = 0;
226 this->idle_threads = 0;
227
228 return &this->public;
229 }
230