![[BACK]](/icons/cvsweb/back.gif){kind=icon}
Return to processor.c CVS log ![[TXT]](/icons/cvsweb/text.gif){kind=icon}
![[DIR]](/icons/cvsweb/dir.gif){kind=icon}
Up to [ELWIX - Embedded LightWeight unIX -] / embedaddon / strongswan / src / libstrongswan / processing|
Return to processor.c CVS log | Up to [ELWIX - Embedded LightWeight unIX -] / embedaddon / strongswan / src / libstrongswan / processing |
1.1 misho 1: /*
2: * Copyright (C) 2005-2011 Martin Willi
3: * Copyright (C) 2011 revosec AG
4: * Copyright (C) 2008-2013 Tobias Brunner
5: * Copyright (C) 2005 Jan Hutter
6: * HSR Hochschule fuer Technik Rapperswil
7: *
8: * This program is free software; you can redistribute it and/or modify it
9: * under the terms of the GNU General Public License as published by the
10: * Free Software Foundation; either version 2 of the License, or (at your
11: * option) any later version. See <http://www.fsf.org/copyleft/gpl.txt>.
12: *
13: * This program is distributed in the hope that it will be useful, but
14: * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
15: * or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
16: * for more details.
17: */
18:
19: #include <stdlib.h>
20: #include <string.h>
21: #include <errno.h>
22:
23: #include "processor.h"
24:
25: #include <utils/debug.h>
26: #include <threading/thread.h>
27: #include <threading/condvar.h>
28: #include <threading/mutex.h>
29: #include <threading/thread_value.h>
30: #include <collections/linked_list.h>
31:
32: typedef struct private_processor_t private_processor_t;
33:
34: /**
35: * Private data of processor_t class.
36: */
37: struct private_processor_t {
38:
39: /**
40: * Public processor_t interface.
41: */
42: processor_t public;
43:
44: /**
45: * Number of running threads
46: */
47: u_int total_threads;
48:
49: /**
50: * Desired number of threads
51: */
52: u_int desired_threads;
53:
54: /**
55: * Number of threads currently working, for each priority
56: */
57: u_int working_threads[JOB_PRIO_MAX];
58:
59: /**
60: * All threads managed in the pool (including threads that have been
61: * canceled, this allows to join them later), as worker_thread_t
62: */
63: linked_list_t *threads;
64:
65: /**
66: * A list of queued jobs for each priority
67: */
68: linked_list_t *jobs[JOB_PRIO_MAX];
69:
70: /**
71: * Threads reserved for each priority
72: */
73: int prio_threads[JOB_PRIO_MAX];
74:
75: /**
76: * access to job lists is locked through this mutex
77: */
78: mutex_t *mutex;
79:
80: /**
81: * Condvar to wait for new jobs
82: */
83: condvar_t *job_added;
84:
85: /**
86: * Condvar to wait for terminated threads
87: */
88: condvar_t *thread_terminated;
89: };
90:
91: /**
92: * Worker thread
93: */
94: typedef struct {
95:
96: /**
97: * Reference to the processor
98: */
99: private_processor_t *processor;
100:
101: /**
102: * The actual thread
103: */
104: thread_t *thread;
105:
106: /**
107: * Job currently being executed by this worker thread
108: */
109: job_t *job;
110:
111: /**
112: * Priority of the current job
113: */
114: job_priority_t priority;
115:
116: } worker_thread_t;
117:
118: static void process_jobs(worker_thread_t *worker);
119:
120: /**
121: * restart a terminated thread
122: */
123: static void restart(worker_thread_t *worker)
124: {
125: private_processor_t *this = worker->processor;
126: job_t *job;
127:
128: DBG2(DBG_JOB, "terminated worker thread %.2u", thread_current_id());
129:
130: this->mutex->lock(this->mutex);
131: /* cleanup worker thread */
132: this->working_threads[worker->priority]--;
133: worker->job->status = JOB_STATUS_CANCELED;
134: job = worker->job;
135: /* unset the job before releasing the mutex, otherwise cancel() might
136: * interfere */
137: worker->job = NULL;
138: /* release mutex to avoid deadlocks if the same lock is required
139: * during queue_job() and in the destructor called here */
140: this->mutex->unlock(this->mutex);
141: job->destroy(job);
142: this->mutex->lock(this->mutex);
143:
144: /* respawn thread if required */
145: if (this->desired_threads >= this->total_threads)
146: {
147: worker_thread_t *new_worker;
148:
149: INIT(new_worker,
150: .processor = this,
151: );
152: new_worker->thread = thread_create((thread_main_t)process_jobs,
153: new_worker);
154: if (new_worker->thread)
155: {
156: this->threads->insert_last(this->threads, new_worker);
157: this->mutex->unlock(this->mutex);
158: return;
159: }
160: free(new_worker);
161: }
162: this->total_threads--;
163: this->thread_terminated->signal(this->thread_terminated);
164: this->mutex->unlock(this->mutex);
165: }
166:
167: /**
168: * Get number of idle threads, non-locking variant
169: */
170: static u_int get_idle_threads_nolock(private_processor_t *this)
171: {
172: u_int count, i;
173:
174: count = this->total_threads;
175: for (i = 0; i < JOB_PRIO_MAX; i++)
176: {
177: count -= this->working_threads[i];
178: }
179: return count;
180: }
181:
182: /**
183: * Get a job from any job queue, starting with the highest priority.
184: *
185: * this->mutex is expected to be locked.
186: */
187: static bool get_job(private_processor_t *this, worker_thread_t *worker)
188: {
189: int i, reserved = 0, idle;
190:
191: idle = get_idle_threads_nolock(this);
192:
193: for (i = 0; i < JOB_PRIO_MAX; i++)
194: {
195: if (reserved && reserved >= idle)
196: {
197: DBG2(DBG_JOB, "delaying %N priority jobs: %d threads idle, "
198: "but %d reserved for higher priorities",
199: job_priority_names, i, idle, reserved);
200: /* wait until a job of higher priority gets queued */
201: return FALSE;
202: }
203: if (this->working_threads[i] < this->prio_threads[i])
204: {
205: reserved += this->prio_threads[i] - this->working_threads[i];
206: }
207: if (this->jobs[i]->remove_first(this->jobs[i],
208: (void**)&worker->job) == SUCCESS)
209: {
210: worker->priority = i;
211: return TRUE;
212: }
213: }
214: return FALSE;
215: }
216:
217: /**
218: * Process a single job (provided in worker->job, worker->priority is also
219: * expected to be set)
220: *
221: * this->mutex is expected to be locked.
222: */
223: static void process_job(private_processor_t *this, worker_thread_t *worker)
224: {
225: job_t *to_destroy = NULL;
226: job_requeue_t requeue;
227:
228: this->working_threads[worker->priority]++;
229: worker->job->status = JOB_STATUS_EXECUTING;
230: this->mutex->unlock(this->mutex);
231: /* canceled threads are restarted to get a constant pool */
232: thread_cleanup_push((thread_cleanup_t)restart, worker);
233: while (TRUE)
234: {
235: requeue = worker->job->execute(worker->job);
236: if (requeue.type != JOB_REQUEUE_TYPE_DIRECT)
237: {
238: break;
239: }
240: else if (!worker->job->cancel)
241: { /* only allow cancelable jobs to requeue directly */
242: requeue.type = JOB_REQUEUE_TYPE_FAIR;
243: break;
244: }
245: }
246: thread_cleanup_pop(FALSE);
247: this->mutex->lock(this->mutex);
248: this->working_threads[worker->priority]--;
249: if (worker->job->status == JOB_STATUS_CANCELED)
250: { /* job was canceled via a custom cancel() method or did not
251: * use JOB_REQUEUE_TYPE_DIRECT */
252: to_destroy = worker->job;
253: }
254: else
255: {
256: switch (requeue.type)
257: {
258: case JOB_REQUEUE_TYPE_NONE:
259: worker->job->status = JOB_STATUS_DONE;
260: to_destroy = worker->job;
261: break;
262: case JOB_REQUEUE_TYPE_FAIR:
263: worker->job->status = JOB_STATUS_QUEUED;
264: this->jobs[worker->priority]->insert_last(
265: this->jobs[worker->priority], worker->job);
266: this->job_added->signal(this->job_added);
267: break;
268: case JOB_REQUEUE_TYPE_SCHEDULE:
269: /* scheduler_t does not hold its lock when queuing jobs
270: * so this should be safe without unlocking our mutex */
271: switch (requeue.schedule)
272: {
273: case JOB_SCHEDULE:
274: lib->scheduler->schedule_job(lib->scheduler,
275: worker->job, requeue.time.rel);
276: break;
277: case JOB_SCHEDULE_MS:
278: lib->scheduler->schedule_job_ms(lib->scheduler,
279: worker->job, requeue.time.rel);
280: break;
281: case JOB_SCHEDULE_TV:
282: lib->scheduler->schedule_job_tv(lib->scheduler,
283: worker->job, requeue.time.abs);
284: break;
285: }
286: break;
287: default:
288: break;
289: }
290: }
291: /* unset the current job to avoid interference with cancel() when
292: * destroying the job below */
293: worker->job = NULL;
294:
295: if (to_destroy)
296: { /* release mutex to avoid deadlocks if the same lock is required
297: * during queue_job() and in the destructor called here */
298: this->mutex->unlock(this->mutex);
299: to_destroy->destroy(to_destroy);
300: this->mutex->lock(this->mutex);
301: }
302: }
303:
304: /**
305: * Process queued jobs, called by the worker threads
306: */
307: static void process_jobs(worker_thread_t *worker)
308: {
309: private_processor_t *this = worker->processor;
310:
311: /* worker threads are not cancelable by default */
312: thread_cancelability(FALSE);
313:
314: DBG2(DBG_JOB, "started worker thread %.2u", thread_current_id());
315:
316: this->mutex->lock(this->mutex);
317: while (this->desired_threads >= this->total_threads)
318: {
319: if (get_job(this, worker))
320: {
321: process_job(this, worker);
322: }
323: else
324: {
325: this->job_added->wait(this->job_added, this->mutex);
326: }
327: }
328: this->total_threads--;
329: this->thread_terminated->signal(this->thread_terminated);
330: this->mutex->unlock(this->mutex);
331: }
332:
333: METHOD(processor_t, get_total_threads, u_int,
334: private_processor_t *this)
335: {
336: u_int count;
337:
338: this->mutex->lock(this->mutex);
339: count = this->total_threads;
340: this->mutex->unlock(this->mutex);
341: return count;
342: }
343:
344: METHOD(processor_t, get_idle_threads, u_int,
345: private_processor_t *this)
346: {
347: u_int count;
348:
349: this->mutex->lock(this->mutex);
350: count = get_idle_threads_nolock(this);
351: this->mutex->unlock(this->mutex);
352: return count;
353: }
354:
355: /**
356: * Check priority bounds
357: */
358: static job_priority_t sane_prio(job_priority_t prio)
359: {
360: if ((int)prio < 0 || prio >= JOB_PRIO_MAX)
361: {
362: return JOB_PRIO_MAX - 1;
363: }
364: return prio;
365: }
366:
367: METHOD(processor_t, get_working_threads, u_int,
368: private_processor_t *this, job_priority_t prio)
369: {
370: u_int count;
371:
372: this->mutex->lock(this->mutex);
373: count = this->working_threads[sane_prio(prio)];
374: this->mutex->unlock(this->mutex);
375: return count;
376: }
377:
378: METHOD(processor_t, get_job_load, u_int,
379: private_processor_t *this, job_priority_t prio)
380: {
381: u_int load;
382:
383: prio = sane_prio(prio);
384: this->mutex->lock(this->mutex);
385: load = this->jobs[prio]->get_count(this->jobs[prio]);
386: this->mutex->unlock(this->mutex);
387: return load;
388: }
389:
390: METHOD(processor_t, queue_job, void,
391: private_processor_t *this, job_t *job)
392: {
393: job_priority_t prio;
394:
395: prio = sane_prio(job->get_priority(job));
396: job->status = JOB_STATUS_QUEUED;
397:
398: this->mutex->lock(this->mutex);
399: this->jobs[prio]->insert_last(this->jobs[prio], job);
400: this->job_added->signal(this->job_added);
401: this->mutex->unlock(this->mutex);
402: }
403:
404: METHOD(processor_t, execute_job, void,
405: private_processor_t *this, job_t *job)
406: {
407: job_priority_t prio;
408: bool queued = FALSE;
409:
410: this->mutex->lock(this->mutex);
411: if (this->desired_threads && get_idle_threads_nolock(this))
412: {
413: prio = sane_prio(job->get_priority(job));
414: job->status = JOB_STATUS_QUEUED;
415: /* insert job in front to execute it immediately */
416: this->jobs[prio]->insert_first(this->jobs[prio], job);
417: queued = TRUE;
418: }
419: this->job_added->signal(this->job_added);
420: this->mutex->unlock(this->mutex);
421:
422: if (!queued)
423: {
424: job->execute(job);
425: job->destroy(job);
426: }
427: }
428:
429: METHOD(processor_t, set_threads, void,
430: private_processor_t *this, u_int count)
431: {
432: int i;
433:
434: this->mutex->lock(this->mutex);
435: for (i = 0; i < JOB_PRIO_MAX; i++)
436: {
437: this->prio_threads[i] = lib->settings->get_int(lib->settings,
438: "%s.processor.priority_threads.%N", 0, lib->ns,
439: job_priority_names, i);
440: }
441: if (count > this->total_threads)
442: { /* increase thread count */
443: worker_thread_t *worker;
444: int i;
445:
446: this->desired_threads = count;
447: DBG1(DBG_JOB, "spawning %d worker threads", count - this->total_threads);
448: for (i = this->total_threads; i < count; i++)
449: {
450: INIT(worker,
451: .processor = this,
452: );
453: worker->thread = thread_create((thread_main_t)process_jobs, worker);
454: if (worker->thread)
455: {
456: this->threads->insert_last(this->threads, worker);
457: this->total_threads++;
458: }
459: else
460: {
461: free(worker);
462: }
463: }
464: }
465: else if (count < this->total_threads)
466: { /* decrease thread count */
467: this->desired_threads = count;
468: }
469: this->job_added->broadcast(this->job_added);
470: this->mutex->unlock(this->mutex);
471: }
472:
473: METHOD(processor_t, cancel, void,
474: private_processor_t *this)
475: {
476: enumerator_t *enumerator;
477: worker_thread_t *worker;
478: job_t *job;
479: int i;
480:
481: this->mutex->lock(this->mutex);
482: this->desired_threads = 0;
483: /* cancel potentially blocking jobs */
484: enumerator = this->threads->create_enumerator(this->threads);
485: while (enumerator->enumerate(enumerator, (void**)&worker))
486: {
487: if (worker->job && worker->job->cancel)
488: {
489: worker->job->status = JOB_STATUS_CANCELED;
490: if (!worker->job->cancel(worker->job))
491: { /* job requests to be canceled explicitly, otherwise we assume
492: * the thread terminates itself and can be joined */
493: worker->thread->cancel(worker->thread);
494: }
495: }
496: }
497: enumerator->destroy(enumerator);
498: while (this->total_threads > 0)
499: {
500: this->job_added->broadcast(this->job_added);
501: this->thread_terminated->wait(this->thread_terminated, this->mutex);
502: }
503: while (this->threads->remove_first(this->threads,
504: (void**)&worker) == SUCCESS)
505: {
506: worker->thread->join(worker->thread);
507: free(worker);
508: }
509: for (i = 0; i < JOB_PRIO_MAX; i++)
510: {
511: while (this->jobs[i]->remove_first(this->jobs[i],
512: (void**)&job) == SUCCESS)
513: {
514: job->destroy(job);
515: }
516: }
517: this->mutex->unlock(this->mutex);
518: }
519:
520: METHOD(processor_t, destroy, void,
521: private_processor_t *this)
522: {
523: int i;
524:
525: cancel(this);
526: this->thread_terminated->destroy(this->thread_terminated);
527: this->job_added->destroy(this->job_added);
528: this->mutex->destroy(this->mutex);
529: for (i = 0; i < JOB_PRIO_MAX; i++)
530: {
531: this->jobs[i]->destroy(this->jobs[i]);
532: }
533: this->threads->destroy(this->threads);
534: free(this);
535: }
536:
537: /*
538: * Described in header.
539: */
540: processor_t *processor_create()
541: {
542: private_processor_t *this;
543: int i;
544:
545: INIT(this,
546: .public = {
547: .get_total_threads = _get_total_threads,
548: .get_idle_threads = _get_idle_threads,
549: .get_working_threads = _get_working_threads,
550: .get_job_load = _get_job_load,
551: .queue_job = _queue_job,
552: .execute_job = _execute_job,
553: .set_threads = _set_threads,
554: .cancel = _cancel,
555: .destroy = _destroy,
556: },
557: .threads = linked_list_create(),
558: .mutex = mutex_create(MUTEX_TYPE_DEFAULT),
559: .job_added = condvar_create(CONDVAR_TYPE_DEFAULT),
560: .thread_terminated = condvar_create(CONDVAR_TYPE_DEFAULT),
561: );
562:
563: for (i = 0; i < JOB_PRIO_MAX; i++)
564: {
565: this->jobs[i] = linked_list_create();
566: }
567: return &this->public;
568: }