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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
|
#ifndef __RCPP_PARALLEL_TINYTHREAD__
#define __RCPP_PARALLEL_TINYTHREAD__
#include <cstdlib>
#include <cstdio>
#include "Common.h"
#include <tthread/tinythread.h>
#include <vector>
namespace RcppParallel {
namespace {
// Class which represents a range of indexes to perform work on
// (worker functions are passed this range so they know which
// elements are safe to read/write to)
class IndexRange {
public:
// Initizlize with a begin and (exclusive) end index
IndexRange(std::size_t begin, std::size_t end)
: begin_(begin), end_(end)
{
}
// Access begin() and end()
std::size_t begin() const { return begin_; }
std::size_t end() const { return end_; }
std::size_t size() const { return end_ - begin_ ; }
private:
std::size_t begin_;
std::size_t end_;
};
// Because tinythread allows us to pass only a plain C function
// we need to pass our worker and range within a struct that we
// can cast to/from void*
struct Work {
Work(IndexRange range, Worker& worker)
: range(range), worker(worker)
{
}
IndexRange range;
Worker& worker;
};
// Thread which performs work (then deletes the work object
// when it's done)
extern "C" inline void workerThread(void* data) {
try
{
Work* pWork = static_cast<Work*>(data);
pWork->worker(pWork->range.begin(), pWork->range.end());
delete pWork;
}
catch(...)
{
}
}
// Function to calculate the ranges for a given input
std::vector<IndexRange> splitInputRange(const IndexRange& range,
std::size_t grainSize) {
// determine max number of threads
std::size_t threads = tthread::thread::hardware_concurrency();
char* numThreads = ::getenv("RCPP_PARALLEL_NUM_THREADS");
if (numThreads != NULL) {
int parsedThreads = ::atoi(numThreads);
if (parsedThreads > 0)
threads = parsedThreads;
}
// compute grainSize (including enforcing requested minimum)
std::size_t length = range.end() - range.begin();
if (threads == 1)
grainSize = length;
else if ((length % threads) == 0) // perfect division
grainSize = std::max(length / threads, grainSize);
else // imperfect division, divide by threads - 1
grainSize = std::max(length / (threads-1), grainSize);
// allocate ranges
std::vector<IndexRange> ranges;
std::size_t begin = range.begin();
std::size_t end = begin;
while (begin < range.end()) {
if ((range.end() - (begin + grainSize)) < grainSize)
end = range.end();
else
end = std::min(begin + grainSize, range.end());
ranges.push_back(IndexRange(begin, end));
begin = end;
}
// return ranges
return ranges;
}
} // anonymous namespace
// Execute the Worker over the IndexRange in parallel
inline void ttParallelFor(std::size_t begin,
std::size_t end,
Worker& worker,
std::size_t grainSize = 1)
{
// split the work
IndexRange inputRange(begin, end);
std::vector<IndexRange> ranges = splitInputRange(inputRange, grainSize);
// create threads
std::vector<tthread::thread*> threads;
for (std::size_t i = 0; i<ranges.size(); ++i) {
threads.push_back(new tthread::thread(workerThread, new Work(ranges[i], worker)));
}
// join and delete them
for (std::size_t i = 0; i<threads.size(); ++i) {
threads[i]->join();
delete threads[i];
}
}
// Execute the IWorker over the range in parallel then join results
template <typename Reducer>
inline void ttParallelReduce(std::size_t begin,
std::size_t end,
Reducer& reducer,
std::size_t grainSize = 1)
{
// split the work
IndexRange inputRange(begin, end);
std::vector<IndexRange> ranges = splitInputRange(inputRange, grainSize);
// create threads (split for each thread and track the allocated workers)
std::vector<tthread::thread*> threads;
std::vector<Worker*> workers;
for (std::size_t i = 0; i<ranges.size(); ++i) {
Reducer* pReducer = new Reducer(reducer, RcppParallel::Split());
workers.push_back(pReducer);
threads.push_back(new tthread::thread(workerThread, new Work(ranges[i], *pReducer)));
}
// wait for each thread, join it's results, then delete the worker & thread
for (std::size_t i = 0; i<threads.size(); ++i) {
// wait for thread
threads[i]->join();
// join the results
reducer.join(static_cast<Reducer&>(*workers[i]));
// delete the worker (which we split above) and the thread
delete workers[i];
delete threads[i];
}
}
} // namespace RcppParallel
#endif // __RCPP_PARALLEL_TINYTHREAD__
|
