-
Notifications
You must be signed in to change notification settings - Fork 0
/
Stage.hpp
55 lines (44 loc) · 1.71 KB
/
Stage.hpp
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
#pragma once
#include <thread>
#include <boost/optional.hpp>
#include "Queue.hpp"
#include "StageProcessor.hpp"
namespace beton {
/// Represents a single stage in the processing pipeline, parameterized by the type of input
/// data and type of output data. Contains one or more StageProcessors, each running on a
/// separate thread.
template <typename TIn, typename TOut>
class Stage {
std::vector<std::thread> threads;
std::vector<std::shared_ptr<StageProcessor<TIn, TOut>>> processors;
std::atomic<bool> running;
QueueRef<TIn> inQueue;
QueueRef<TOut> outQueue;
public:
Stage(QueueRef<TIn> inQueue, QueueRef<TOut> outQueue) :
inQueue(inQueue), outQueue(outQueue), running(true) {}
~Stage() {
running = false;
for (auto& thread : threads) {
thread.join();
}
}
const QueueRef<TIn> getInQueue() { return inQueue; }
const QueueRef<TOut> getOutQueue() { return outQueue; }
void pushProcessor(std::shared_ptr<StageProcessor<TIn, TOut>> processor) {
processors.push_back(processor);
threads.push_back(std::thread([=] () {
processor->threadSetup();
while (running) {
// Wait with timeout, otherwise thread can't be joined.
boost::optional<TIn> dataIn = inQueue->tryPop(std::chrono::milliseconds(1000));
if (dataIn) {
auto data = *dataIn;
auto result = processor->process(data);
outQueue->push(result);
}
}
}));
}
};
}