blob: 04a8e860ec8a12e06c481ba81d85542e8c76c83c (
plain) (
blame)
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
|
#include "globals.h"
#include "helper.h"
#include "threadpool.h"
#include "signal.h"
#include "locks.h"
#include <pthread.h>
typedef struct _entry_t {
struct _entry_t *next;
pthread_t thread;
int signalFd;
void *(*startRoutine)(void *);
void * volatile arg;
} entry_t;
static void *threadpool_worker(void *entryPtr);
static pthread_attr_t threadAttrs;
static int maxIdleThreads = -1;
static entry_t *pool = NULL;
static pthread_spinlock_t poolLock;
bool threadpool_init(int maxIdle)
{
if ( maxIdle < 0 || maxIdleThreads >= 0 ) return false;
maxIdleThreads = maxIdle;
spin_init( &poolLock, PTHREAD_PROCESS_PRIVATE );
pthread_attr_init( &threadAttrs );
pthread_attr_setdetachstate( &threadAttrs, PTHREAD_CREATE_DETACHED );
return true;
}
bool threadpool_run(void *(*startRoutine)(void *), void *arg)
{
spin_lock( &poolLock );
entry_t *entry = pool;
if ( entry != NULL ) pool = entry->next;
spin_unlock( &poolLock );
if ( entry == NULL ) {
entry = (entry_t*)malloc( sizeof(entry_t) );
entry->signalFd = signal_new();
if ( entry->signalFd < 0 ) {
printf( "[WARNING] Could not create signalFd for new thread pool thread\n" );
return false;
}
if ( 0 != thread_create( &(entry->thread), &threadAttrs, threadpool_worker, (void*)entry ) ) {
printf( "[WARNING] Could not create new thread for thread pool\n" );
signal_close( entry->signalFd );
free( entry );
return false;
}
printf( "[DEBUG] Thread created!\n" );
}
entry->next = NULL;
entry->startRoutine = startRoutine;
entry->arg = arg;
signal_call( entry->signalFd );
return true;
}
/**
* This is a worker thread of our thread pool.
*/
static void *threadpool_worker(void *entryPtr)
{
blockNoncriticalSignals();
entry_t *entry = (entry_t*)entryPtr;
while ( !_shutdown ) {
// Wait for signal from outside that we have work to do
int ret = signal_wait( entry->signalFd, -1 );
if ( ret > 0 ) {
if ( entry->startRoutine == NULL ) {
printf( "[DEBUG] Worker woke up but has no work to do!\n" );
continue;
}
// Start assigned work
(*entry->startRoutine)( entry->arg );
// Reset vars for safety
entry->startRoutine = NULL;
entry->arg = NULL;
// Put thread back into pool if there are less than maxIdleThreds threads, just die otherwise
int threadCount = 0;
spin_lock( &poolLock );
entry_t *ptr = pool;
while ( ptr != NULL ) {
threadCount++;
ptr = ptr->next;
}
if ( threadCount >= maxIdleThreads ) {
spin_unlock( &poolLock );
signal_close( entry->signalFd );
free( entry );
printf(" [DEBUG] Thread killed!\n" );
return NULL;
}
entry->next = pool;
pool = entry;
spin_unlock( &poolLock );
setThreadName( "[pool]" );
} else {
printf( "[DEBUG] Unexpected return value %d for signal_wait in threadpool worker!\n", ret );
}
}
return NULL;
}
|
