forked from Tencent/libco
/
worker_pool.cpp
119 lines (103 loc) · 3 KB
/
worker_pool.cpp
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
#include <sys/mman.h>
#include <unistd.h>
#include <deque>
#include <map>
#include <stack>
#include "co_routine.h"
#include "co_routine_inner.h"
#include "worker_pool.h"
static const int kPageSize = 4 * 1024;
class CWorker {
public:
CWorker(stCoRoutine_t* co) : co_(co), func_(NULL){};
stCoRoutine_t* co_;
std::function<void(void)> func_;
};
class CWorkerPool {
public:
CWorkerPool() {
soft_limit_stack_size_ = kDefaultSoftLimitStackSize;
hard_limit_stack_size_ = kDefaultHardLimitStackSize;
gc_stack_size_ = kDefaultHardLimitStackSize - kDefaultSoftLimitStackSize;
}
std::stack<CWorker*> workers_;
int soft_limit_stack_size_;
int hard_limit_stack_size_;
int gc_stack_size_;
};
static CWorkerPool g_worker_pool;
int InitWorkerPool(int soft_limit_stack_size, int hard_limit_stack_size) {
static bool has_init = false;
if (has_init) {
return 0;
}
has_init = true;
soft_limit_stack_size =
(soft_limit_stack_size + kPageSize - 1) / kPageSize * kPageSize;
hard_limit_stack_size =
(hard_limit_stack_size + kPageSize - 1) / kPageSize * kPageSize;
if (soft_limit_stack_size < kDefaultSoftLimitStackSize) {
soft_limit_stack_size = kDefaultSoftLimitStackSize;
}
if (hard_limit_stack_size > kDefaultHardLimitStackSize) {
hard_limit_stack_size = kDefaultHardLimitStackSize;
}
if (hard_limit_stack_size < soft_limit_stack_size) {
hard_limit_stack_size = soft_limit_stack_size;
}
// init g_worker_pool
g_worker_pool.soft_limit_stack_size_ = soft_limit_stack_size;
g_worker_pool.hard_limit_stack_size_ = hard_limit_stack_size;
g_worker_pool.gc_stack_size_ = hard_limit_stack_size - soft_limit_stack_size;
return 0;
}
static void FreeStackRss(void* stack_buffer) {
int ret = 0;
unsigned char* p1 = (unsigned char*)stack_buffer;
p1 = p1 + g_worker_pool.gc_stack_size_ - kPageSize;
unsigned char vec[1] = {0};
ret = mincore((void*)p1, kPageSize, vec);
if (ret != 0) {
return;
}
// no need to free
if ((vec[0] & 0x1) == 0) {
return;
}
madvise(stack_buffer, g_worker_pool.gc_stack_size_, MADV_DONTNEED);
return;
}
static void* worker_main(void* arg) {
CWorker* worker = (CWorker*)arg;
while (true) {
if (worker->func_) {
worker->func_();
}
worker->func_ = NULL;
if (g_worker_pool.gc_stack_size_ > 0) {
FreeStackRss(worker->co_->stack_mem->stack_buffer);
}
g_worker_pool.workers_.push(worker);
co_yield_ct();
}
return NULL;
}
void ProcessByWorker(std::function<void(void)> func1) {
CWorker* worker = NULL;
if (g_worker_pool.workers_.empty()) {
stCoRoutineAttr_t attr;
attr.stack_size = g_worker_pool.hard_limit_stack_size_;
stCoRoutine_t* worker_routine = NULL;
worker = new CWorker(worker_routine);
co_create(&worker_routine, &attr, worker_main, worker);
worker->co_ = worker_routine;
} else {
worker = g_worker_pool.workers_.top();
g_worker_pool.workers_.pop();
}
if (worker == NULL) {
return;
}
worker->func_ = func1;
co_resume(worker->co_);
}