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
|
#include "stdafx.h"
#include "WorkQueue.h"
#include "Server.h"
#include "OS/UThread.h"
namespace storm {
namespace server {
WorkItem::WorkItem(File *file) : file(file) {}
Range WorkItem::run(WorkQueue *q) {
return Range();
}
Bool WorkItem::merge(WorkItem *o) {
return runtime::typeOf(this) == runtime::typeOf(o)
&& file == o->file;
}
/**
* Work queue.
*/
WorkQueue::WorkQueue(Server *callbackTo) : callbackTo(callbackTo), running(false), quit(false) {
event = new (this) Event();
work = new (this) Array<WorkItem *>();
idleTime = defaultIdleTime;
}
void WorkQueue::start() {
if (running)
return;
running = true;
quit = false;
WorkQueue *me = this;
os::FnCall<void, 1> params = os::fnCall().add(me);
os::UThread::spawn(address(&WorkQueue::workerMain), true, params);
}
void WorkQueue::stop() {
if (!running)
return;
quit = true;
event->set();
while (running)
os::UThread::leave();
}
void WorkQueue::poke() {
startWork = Moment() + time::ms(idleTime);
}
void WorkQueue::post(WorkItem *item) {
// Linear search is good enough as we do not expect more than ~10 events in the queue at any given time.
bool found = false;
for (Nat i = 0; i < work->count(); i++) {
if (work->at(i)->merge(item)) {
found = true;
break;
}
}
if (!found)
work->push(item);
event->set();
}
void WorkQueue::workerMain() {
poke();
while (!quit) {
event->wait();
if (work->empty()) {
event->clear();
continue;
}
while (Moment() < startWork)
os::UThread::sleep(idleTime * 3 / 4);
Array<WorkItem *> *work = this->work;
event->clear();
this->work = new (this) Array<WorkItem *>();
for (Nat i = 0; i < work->count(); i++) {
callbackTo->runWork(work->at(i));
}
poke();
}
running = false;
}
}
}
|
