OmniSciDB  471d68cefb
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Groups Pages
threading_serial.h
Go to the documentation of this file.
1 #include "threading_std.h"
2 
3 namespace threading_serial {
4 
5 using namespace threading_common;
6 using std::future;
7 
8 template <typename Fn,
9  typename... Args,
10  typename Result = std::result_of_t<Fn && (Args && ...)>>
11 future<Result> async(Fn&& fn, Args&&... args) {
12  std::promise<Result> pr;
13  if constexpr (std::is_same<void, Result>::value) {
14  fn(std::forward<Args>(args)...);
15  pr.set_value();
16  } else {
17  pr.set_value(fn(std::forward<Args>(args)...));
18  }
19  return pr.get_future();
20 }
21 
22 class task_group {
23  public:
24  template <typename F>
25  void run(F&& f) {
26  f();
27  }
28  void cancel() { /*not implemented*/
29  }
30  void wait() {}
31 }; // class task_group
32 
33 template <typename Int, typename Body, typename Partitioner = auto_partitioner>
34 void parallel_for(const blocked_range<Int>& range,
35  const Body& body,
36  const Partitioner& p = Partitioner()) {
37  const Int worker_count = cpu_threads();
38 
39  for (Int i = 0,
40  start_entry = range.begin(),
41  stop_entry = range.end(),
42  stride = (range.size() + worker_count - 1) / worker_count;
43  i < worker_count && start_entry < stop_entry;
44  ++i, start_entry += stride) {
45  const auto end_entry = std::min(start_entry + stride, stop_entry);
46  body(blocked_range<Int>(start_entry, end_entry));
47  }
48 }
49 
52 template <typename Index, typename Function, typename Partitioner = auto_partitioner>
53 void parallel_for(Index first,
54  Index last,
55  const Function& f,
56  const Partitioner& p = Partitioner()) {
58  blocked_range<Index>(first, last),
59  [&f](const blocked_range<Index>& r) {
60  for (auto i = r.begin(), e = r.end(); i < e; i++) {
61  f(i);
62  }
63  },
64  p);
65 }
66 
68 
69 template <typename Int,
70  typename Value,
71  typename RealBody,
72  typename Reduction,
73  typename Partitioner = auto_partitioner>
75  const Value& identity,
76  const RealBody& real_body,
77  const Reduction& reduction,
78  const Partitioner& p = Partitioner()) {
79  const size_t worker_count = cpu_threads();
80  std::vector<Value> worker_threads;
81  worker_threads.reserve(worker_count);
82 
83  for (Int i = 0,
84  start_entry = range.begin(),
85  stop_entry = range.end(),
86  stride = (range.size() + worker_count - 1) / worker_count;
87  i < worker_count && start_entry < stop_entry;
88  ++i, start_entry += stride) {
89  const auto end_entry = std::min(start_entry + stride, stop_entry);
90  // TODO grainsize?
91  worker_threads.emplace_back(
92  real_body(blocked_range<Int>(start_entry, end_entry), Value{}));
93  }
94  Value v = identity;
95  for (auto& child : worker_threads) {
96  v = reduction(v, child);
97  }
98 
99  return v;
100 }
101 
102 } // namespace threading_serial
size_type size() const
Size of the range.
Definition: threading_std.h:47
future< Result > async(Fn &&fn, Args &&...args)
A range over which to iterate.
Definition: threading_std.h:21
Value parallel_reduce(const blocked_range< Int > &range, const Value &identity, const RealBody &real_body, const Reduction &reduction, const Partitioner &p=Partitioner())
Parallel iteration with reduction.
void parallel_for(const blocked_range< Int > &range, const Body &body, const Partitioner &p=Partitioner())
const_iterator end() const
One past last value in range.
Definition: threading_std.h:43
const_iterator begin() const
Beginning of range.
Definition: threading_std.h:40
char * f
int cpu_threads()
Definition: thread_count.h:24