1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
|
/**
* @file thread_pool.c
*
* @brief Thread-pool with some threads processing the job_queue
*
*/
/*
* Copyright (C) 2005 Jan Hutter, Martin Willi
* Hochschule fuer Technik Rapperswil
*
* This program is free software; you can redistribute it and/or modify it
* under the terms of the GNU General Public License as published by the
* Free Software Foundation; either version 2 of the License, or (at your
* option) any later version. See <http://www.fsf.org/copyleft/gpl.txt>.
*
* This program is distributed in the hope that it will be useful, but
* WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
* or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
* for more details.
*/
#include <stdlib.h>
#include <freeswan.h>
#include <pluto/constants.h>
#include <pluto/defs.h>
#include <pthread.h>
#include "allocator.h"
#include "thread_pool.h"
#include "job_queue.h"
#include "globals.h"
/**
* structure with private members for thread_pool
*/
typedef struct {
/**
* inclusion of public members
*/
thread_pool_t public;
/**
* number of running threads
*/
size_t pool_size;
/**
* array of thread ids
*/
pthread_t *threads;
} private_thread_pool_t;
static void job_processing(private_thread_pool_t *this)
{
/* cancellation disabled by default */
pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);
for (;;) {
job_t *job;
global_job_queue->get(global_job_queue, &job);
/* process them here */
job->destroy(job);
}
}
/**
* Implementation of thread_pool_t.get_pool_size
*/
static status_t get_pool_size(private_thread_pool_t *this, size_t *size)
{
*size = this->pool_size;
return SUCCESS;
}
/**
* Implementation of thread_pool_t.destroy
*/
static status_t destroy(private_thread_pool_t *this)
{
int current;
/* flag thread for termination */
for (current = 0; current < this->pool_size; current++) {
pthread_cancel(this->threads[current]);
}
/* wait for all threads */
for (current = 0; current < this->pool_size; current++) {
pthread_join(this->threads[current], NULL);
}
/* free mem */
allocator_free(this->threads);
allocator_free(this);
return SUCCESS;
}
/**
* Implementation of default constructor for thread_pool_t
*/
thread_pool_t *thread_pool_create(size_t pool_size)
{
int current;
private_thread_pool_t *this = allocator_alloc_thing(private_thread_pool_t);
/* fill in public fields */
this->public.destroy = (status_t(*)(thread_pool_t*))destroy;
this->public.get_pool_size = (status_t(*)(thread_pool_t*, size_t*))get_pool_size;
this->pool_size = pool_size;
this->threads = allocator_alloc(sizeof(pthread_t) * pool_size);
/* try to create as many threads as possible, up tu pool_size */
for (current = 0; current < pool_size; current++) {
if (pthread_create(&(this->threads[current]), NULL, (void*(*)(void*))job_processing, this)) {
/* did we get any? */
if (current == 0) {
allocator_free(this->threads);
allocator_free(this);
return NULL;
}
/* not all threads could be created, but at least one :-/ */
this->pool_size = current;
return (thread_pool_t*)this;
}
}
return (thread_pool_t*)this;
}
|
