DIY 3.0 data-parallel out-of-core C++ library

- Main Page
- Modules
- Namespaces
- Classes
- Files
- Examples

- File List

All Classes Namespaces Functions Typedefs Groups Pages
sort.hpp
1 #ifndef DIY_DETAIL_ALGORITHMS_SORT_HPP
2 #define DIY_DETAIL_ALGORITHMS_SORT_HPP
3
4 #include <functional>
5 #include <algorithm>
6
7 namespace diy
8 {
9
10 namespace detail
11 {
12
13 template<class Block, class T, class Cmp>
14 struct SampleSort
15 {
16 typedef std::vector<T> Block::*ValuesVector;
17 struct Sampler;
18 struct Exchanger;
19
20 SampleSort(ValuesVector values_, ValuesVector samples_, const Cmp& cmp_, size_t num_samples_):
21 values(values_), samples(samples_),
22 cmp(cmp_), num_samples(num_samples_) {}
23
24 Sampler sample() const { return Sampler(values, samples, cmp, num_samples); }
25 Exchanger exchange() const { return Exchanger(values, samples, cmp); }
26
27 static void dequeue_values(std::vector<T>& v, const ReduceProxy& rp, bool skip_self = true)
28 {
29 auto log = get_logger();
30
31 int k_in = rp.in_link().size();
32
33 log->trace("dequeue_values(): gid={}, round={}; v.size()={}", rp.gid(), rp.round(), v.size());
34
35 if (detail::is_default< Serialization<T> >::value)
36 {
37 // add up sizes
38 size_t sz = 0;
39 size_t end = v.size();
40 for (int i = 0; i < k_in; ++i)
41 {
42 log->trace(" incoming size from {}: {}", rp.in_link().target(i).gid, sz);
43 if (skip_self && rp.in_link().target(i).gid == rp.gid()) continue;
44 MemoryBuffer& in = rp.incoming(rp.in_link().target(i).gid);
45 sz += in.size() / sizeof(T);
46 }
47 log->trace(" incoming size: {}", sz);
48 v.resize(end + sz);
49
50 for (int i = 0; i < k_in; ++i)
51 {
52 if (skip_self && rp.in_link().target(i).gid == rp.gid()) continue;
53 MemoryBuffer& in = rp.incoming(rp.in_link().target(i).gid);
54 size_t sz = in.size() / sizeof(T);
55 T* bg = (T*) &in.buffer[0];
56 std::copy(bg, bg + sz, &v[end]);
57 end += sz;
58 }
59 } else
60 {
61 for (int i = 0; i < k_in; ++i)
62 {
63 if (skip_self && rp.in_link().target(i).gid == rp.gid()) continue;
64 MemoryBuffer& in = rp.incoming(rp.in_link().target(i).gid);
65 while(in)
66 {
67 T x;
68 diy::load(in, x);
69 v.emplace_back(std::move(x));
70 }
71 }
72 }
73 log->trace(" v.size()={}", v.size());
74 }
75
76 ValuesVector values;
77 ValuesVector samples;
78 Cmp cmp;
79 size_t num_samples;
80 };
81
82 template<class Block, class T, class Cmp>
83 struct SampleSort<Block,T,Cmp>::Sampler
84 {
85 Sampler(ValuesVector values_, ValuesVector dividers_, const Cmp& cmp_, size_t num_samples_):
86 values(values_), dividers(dividers_), cmp(cmp_), num_samples(num_samples_) {}
87
88 void operator()(Block* b, const ReduceProxy& srp, const RegularSwapPartners& partners) const
89 {
90 int k_in = srp.in_link().size();
91 int k_out = srp.out_link().size();
92
93 std::vector<T> samples;
94
95 if (k_in == 0)
96 {
97 // draw random samples
98 for (size_t i = 0; i < num_samples; ++i)
99 samples.push_back((b->*values)[std::rand() % (b->*values).size()]);
100 } else
101 dequeue_values(samples, srp, false);
102
103 if (k_out == 0)
104 {
105 // pick subsamples that separate quantiles
106 std::sort(samples.begin(), samples.end(), cmp);
107 std::vector<T> subsamples(srp.nblocks() - 1);
108 int step = samples.size() / srp.nblocks(); // NB: subsamples.size() + 1
109 for (size_t i = 0; i < subsamples.size(); ++i)
110 subsamples[i] = samples[(i+1)*step];
111 (b->*dividers).swap(subsamples);
112 }
113 else
114 {
115 for (int i = 0; i < k_out; ++i)
116 {
117 MemoryBuffer& out = srp.outgoing(srp.out_link().target(i));
118 save(out, &samples[0], samples.size());
119 }
120 }
121 }
122
123 ValuesVector values;
124 ValuesVector dividers;
125 Cmp cmp;
126 size_t num_samples;
127 };
128
129 template<class Block, class T, class Cmp>
130 struct SampleSort<Block,T,Cmp>::Exchanger
131 {
132 Exchanger(ValuesVector values_, ValuesVector samples_, const Cmp& cmp_):
133 values(values_), samples(samples_), cmp(cmp_) {}
134
135 void operator()(Block* b, const ReduceProxy& rp) const
136 {
137 if (rp.round() == 0)
138 {
139 // enqueue values to the correct locations
140 for (size_t i = 0; i < (b->*values).size(); ++i)
141 {
142 int to = std::lower_bound((b->*samples).begin(), (b->*samples).end(), (b->*values)[i], cmp) - (b->*samples).begin();
143 rp.enqueue(rp.out_link().target(to), (b->*values)[i]);
144 }
145 (b->*values).clear();
146 } else
147 {
148 dequeue_values((b->*values), rp, false);
149 std::sort((b->*values).begin(), (b->*values).end(), cmp);
150 }
151 }
152
153 ValuesVector values;
154 ValuesVector samples;
155 Cmp cmp;
156 };
157
158 }
159
160 }
161
162 #endif
diy::load
void load(BinaryBuffer &bb, T &x)
Loads x from bb by calling diy::Serialization<T>::load(bb,x).
Definition: serialization.hpp:106
diy::save
void save(BinaryBuffer &bb, const T &x)
Saves x to bb by calling diy::Serialization<T>::save(bb,x).
Definition: serialization.hpp:102
diy::in
void in(const RegularLink< Bounds > &link, const Point &p, OutIter out, const Bounds &domain)
Finds the neighbor(s) containing the target point.
Definition: pick.hpp:102
diy::sort
void sort(Master &master, const Assigner &assigner, std::vector< T > Block::*values, std::vector< T > Block::*samples, size_t num_samples, const Cmp &cmp, int k=2, bool samples_only=false)
sample sort values of each block, store the boundaries between blocks in samples
Definition: algorithms.hpp:25