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
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
|
#include "threadpool.h"
#include "globals.h"
#include "helper.h"
#include "locks.h"
typedef struct _entry_t {
pthread_t thread;
dnbd3_signal_t* signal;
void *(*startRoutine)(void *);
void * arg;
const char *name;
} entry_t;
static void *threadpool_worker(void *entryPtr);
static pthread_attr_t threadAttrs;
static atomic_int maxIdleThreads = -1;
static _Atomic(entry_t *) *pool = NULL;
static atomic_int activeThreads = 0;
bool threadpool_init(int maxIdle)
{
if ( maxIdle < 0 )
return false;
int exp = -1;
if ( !atomic_compare_exchange_strong( &maxIdleThreads, &exp, maxIdle ) )
return false;
pool = malloc( maxIdle * sizeof(*pool) );
for ( int i = 0; i < maxIdle; ++i ) {
atomic_init( &pool[i], NULL );
}
pthread_attr_init( &threadAttrs );
pthread_attr_setdetachstate( &threadAttrs, PTHREAD_CREATE_DETACHED );
return true;
}
void threadpool_close()
{
int max = atomic_exchange( &maxIdleThreads, -1 );
if ( max <= 0 )
return;
for ( int i = 0; i < max; ++i ) {
entry_t *cur = pool[i];
if ( cur != NULL && atomic_compare_exchange_strong( &pool[i], &cur, NULL ) ) {
signal_call( cur->signal );
}
}
}
void threadpool_waitEmpty()
{
if ( activeThreads == 0 )
return;
do {
sleep( 1 );
logadd( LOG_INFO, "Threadpool: %d threads still active", (int)activeThreads );
} while ( activeThreads != 0 );
}
bool threadpool_run(void *(*startRoutine)(void *), void *arg, const char *name)
{
if ( unlikely( _shutdown ) ) {
logadd( LOG_MINOR, "Cannot submit work to threadpool while shutting down!" );
return false;
}
#ifdef DEBUG
if ( unlikely( startRoutine == NULL ) ) {
logadd( LOG_ERROR, "Trying to queue work for thread pool with NULL startRoutine" );
return false; // Or bail out!?
}
#endif
entry_t *entry = NULL;
for ( int i = 0; i < maxIdleThreads; ++i ) {
entry = atomic_exchange( &pool[i], NULL );
if ( entry != NULL ) {
break;
}
}
if ( unlikely( entry == NULL ) ) {
entry = malloc( sizeof(entry_t) );
if ( entry == NULL ) {
logadd( LOG_WARNING, "Could not alloc entry_t for new thread\n" );
return false;
}
entry->signal = signal_newBlocking();
if ( entry->signal == NULL ) {
logadd( LOG_WARNING, "Could not create signal for new thread pool thread\n" );
free( entry );
return false;
}
if ( 0 != thread_create( &(entry->thread), &threadAttrs, threadpool_worker, (void*)entry ) ) {
logadd( LOG_WARNING, "Could not create new thread for thread pool (%d active)\n", (int)activeThreads );
signal_close( entry->signal );
free( entry );
return false;
}
activeThreads++;
}
entry->startRoutine = startRoutine;
entry->arg = arg;
entry->name = name;
atomic_thread_fence( memory_order_release );
signal_call( entry->signal );
return true;
}
/**
* This is a worker thread of our thread pool.
*/
static void *threadpool_worker(void *entryPtr)
{
blockNoncriticalSignals();
entry_t *entry = (entry_t*)entryPtr;
int ret;
for ( ;; ) {
keep_going:;
// Wait for signal from outside that we have work to do
ret = signal_clear( entry->signal );
atomic_thread_fence( memory_order_acquire );
if ( _shutdown )
break;
if ( ret <= 0 ) {
logadd( LOG_DEBUG1, "Unexpected return value %d for signal_wait in threadpool worker!", ret );
continue;
}
#ifdef DEBUG
if ( entry->startRoutine == NULL ) {
logadd( LOG_ERROR, "Worker woke up but has no work to do!" );
exit( 1 );
}
if ( entry->name != NULL ) {
setThreadName( entry->name );
}
#endif
// Start assigned work
(*entry->startRoutine)( entry->arg );
// Reset vars for safety
entry->startRoutine = NULL;
entry->arg = NULL;
atomic_thread_fence( memory_order_release );
if ( _shutdown )
break;
// Put thread back into pool
setThreadName( "[pool]" );
for ( int i = 0; i < maxIdleThreads; ++i ) {
entry_t *exp = NULL;
if ( atomic_compare_exchange_weak( &pool[i], &exp, entry ) ) {
goto keep_going;
}
}
// Reaching here means pool is full; just let the thread exit
break;
}
setThreadName( "[dead]" );
signal_close( entry->signal );
free( entry );
activeThreads--;
return NULL;
}
|
