forked from peter-iakovlev/Signals
-
Notifications
You must be signed in to change notification settings - Fork 0
/
SThreadPool.m
128 lines (104 loc) · 3.13 KB
/
SThreadPool.m
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
#import "SThreadPool.h"
#import <libkern/OSAtomic.h>
#import <pthread.h>
#import "SQueue.h"
@interface SThreadPool ()
{
SQueue *_managementQueue;
NSMutableArray *_threads;
NSMutableArray *_queues;
NSMutableArray *_takenQueues;
pthread_mutex_t _mutex;
pthread_cond_t _cond;
}
@end
@implementation SThreadPool
+ (void)threadEntryPoint:(SThreadPool *)threadPool
{
SThreadPoolQueue *queue = nil;
while (true)
{
SThreadPoolTask *task = nil;
pthread_mutex_lock(&threadPool->_mutex);
if (queue != nil)
{
[threadPool->_takenQueues removeObject:queue];
if ([queue _hasTasks])
[threadPool->_queues addObject:queue];
}
while (true)
{
while (threadPool->_queues.count == 0)
pthread_cond_wait(&threadPool->_cond, &threadPool->_mutex);
queue = threadPool->_queues.firstObject;
task = [queue _popFirstTask];
if (queue != nil)
{
[threadPool->_takenQueues addObject:queue];
[threadPool->_queues removeObjectAtIndex:0];
break;
}
}
pthread_mutex_unlock(&threadPool->_mutex);
@autoreleasepool
{
[task execute];
}
}
}
- (instancetype)init
{
return [self initWithThreadCount:2 threadPriority:0.5];
}
- (instancetype)initWithThreadCount:(NSUInteger)threadCount threadPriority:(double)threadPriority
{
self = [super init];
if (self != nil)
{
pthread_mutex_init(&_mutex, 0);
pthread_cond_init(&_cond, 0);
_managementQueue = [[SQueue alloc] init];
[_managementQueue dispatch:^
{
_threads = [[NSMutableArray alloc] init];
_queues = [[NSMutableArray alloc] init];
_takenQueues = [[NSMutableArray alloc] init];
for (NSUInteger i = 0; i < threadCount; i++)
{
NSThread *thread = [[NSThread alloc] initWithTarget:[SThreadPool class] selector:@selector(threadEntryPoint:) object:self];
thread.name = [[NSString alloc] initWithFormat:@"SThreadPool-%p-%d", self, (int)i];
[thread setThreadPriority:threadPriority];
[_threads addObject:thread];
[thread start];
}
}];
}
return self;
}
- (void)dealloc
{
pthread_mutex_destroy(&_mutex);
pthread_cond_destroy(&_cond);
}
- (void)addTask:(SThreadPoolTask *)task
{
SThreadPoolQueue *tempQueue = [self nextQueue];
[tempQueue addTask:task];
}
- (SThreadPoolQueue *)nextQueue
{
return [[SThreadPoolQueue alloc] initWithThreadPool:self];
}
- (void)_workOnQueue:(SThreadPoolQueue *)queue block:(void (^)())block
{
[_managementQueue dispatch:^
{
pthread_mutex_lock(&_mutex);
block();
if (![_queues containsObject:queue] && ![_takenQueues containsObject:queue])
[_queues addObject:queue];
pthread_cond_broadcast(&_cond);
pthread_mutex_unlock(&_mutex);
}];
}
@end