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
reduce.hpp
1 #ifndef DIY_REDUCE_HPP
2 #define DIY_REDUCE_HPP
3
4 #include <vector>
5 #include "master.hpp"
6 #include "assigner.hpp"
7 #include "detail/block_traits.hpp"
8 #include "log.hpp"
9
10 namespace diy
11 {
15 struct ReduceProxy: public Master::Proxy
16 {
17 typedef std::vector<int> GIDVector;
18
19 ReduceProxy(const Master::Proxy& proxy,
20 void* block,
21 unsigned round,
22 const Assigner& assigner,
23 const GIDVector& incoming_gids,
24 const GIDVector& outgoing_gids):
25 Master::Proxy(proxy),
26 block_(block),
27 round_(round),
28 assigner_(assigner)
29 {
30 // setup in_link
31 for (unsigned i = 0; i < incoming_gids.size(); ++i)
32 {
33 BlockID nbr;
34 nbr.gid = incoming_gids[i];
35 nbr.proc = assigner.rank(nbr.gid);
36 in_link_.add_neighbor(nbr);
37 }
38
39 // setup out_link
40 for (unsigned i = 0; i < outgoing_gids.size(); ++i)
41 {
42 BlockID nbr;
43 nbr.gid = outgoing_gids[i];
44 nbr.proc = assigner.rank(nbr.gid);
45 out_link_.add_neighbor(nbr);
46 }
47 }
48
49 ReduceProxy(const Master::Proxy& proxy,
50 void* block,
51 unsigned round,
52 const Assigner& assigner,
53 const Link& in_link,
54 const Link& out_link):
55 Master::Proxy(proxy),
56 block_(block),
57 round_(round),
58 assigner_(assigner),
59 in_link_(in_link),
60 out_link_(out_link)
61 {}
62
64 void* block() const { return block_; }
66 unsigned round() const { return round_; }
68 const Link& in_link() const { return in_link_; }
70 const Link& out_link() const { return out_link_; }
72 int nblocks() const { return assigner_.nblocks(); }
74 const Assigner& assigner() const { return assigner_; }
75
77 void set_round(unsigned r) { round_ = r; }
78
79 private:
80 void* block_;
81 unsigned round_;
82 const Assigner& assigner_;
83
84 Link in_link_;
85 Link out_link_;
86 };
87
88 namespace detail
89 {
90 template<class Block, class Partners>
91 struct ReductionFunctor;
92
93 template<class Partners, class Skip>
94 struct SkipInactiveOr;
95
96 struct ReduceNeverSkip
97 {
98 bool operator()(int, int, const Master&) const { return false; }
99 };
100 }
101
108 template<class Reduce, class Partners, class Skip>
109 void reduce(Master& master,
110 const Assigner& assigner,
111 const Partners& partners,
112 const Reduce& reduce,
113 const Skip& skip)
114 {
115 auto log = get_logger();
116
117 int original_expected = master.expected();
118
119 using Block = typename detail::block_traits<Reduce>::type;
120
121 unsigned round;
122 for (round = 0; round < partners.rounds(); ++round)
123 {
124 log->debug("Round {}", round);
125 master.foreach(detail::ReductionFunctor<Block,Partners>(round, reduce, partners, assigner),
126 detail::SkipInactiveOr<Partners,Skip>(round, partners, skip));
127 master.execute();
128
129 int expected = 0;
130 for (unsigned i = 0; i < master.size(); ++i)
131 {
132 if (partners.active(round + 1, master.gid(i), master))
133 {
134 std::vector<int> incoming_gids;
135 partners.incoming(round + 1, master.gid(i), incoming_gids, master);
136 expected += static_cast<int>(incoming_gids.size());
137 master.incoming(master.gid(i)).clear();
138 }
139 }
140 master.set_expected(expected);
141 master.flush();
142 }
143 // final round
144 log->debug("Round {}", round);
145 master.foreach(detail::ReductionFunctor<Block,Partners>(round, reduce, partners, assigner),
146 detail::SkipInactiveOr<Partners,Skip>(round, partners, skip));
147
148 master.set_expected(original_expected);
149 }
150
157 template<class Reduce, class Partners>
158 void reduce(Master& master,
159 const Assigner& assigner,
160 const Partners& partners,
161 const Reduce& reducer)
162 {
163 reduce(master, assigner, partners, reducer, detail::ReduceNeverSkip());
164 }
165
166 namespace detail
167 {
168 template<class Block, class Partners>
169 struct ReductionFunctor
170 {
171 using Callback = std::function<void(Block*, const ReduceProxy&, const Partners&)>;
172
173 ReductionFunctor(unsigned round_, const Callback& reduce_, const Partners& partners_, const Assigner& assigner_):
174 round(round_), reduce(reduce_), partners(partners_), assigner(assigner_) {}
175
176 void operator()(Block* b, const Master::ProxyWithLink& cp) const
177 {
178 if (!partners.active(round, cp.gid(), *cp.master())) return;
179
180 std::vector<int> incoming_gids, outgoing_gids;
181 if (round > 0)
182 partners.incoming(round, cp.gid(), incoming_gids, *cp.master()); // receive from the previous round
183 if (round < partners.rounds())
184 partners.outgoing(round, cp.gid(), outgoing_gids, *cp.master()); // send to the next round
185
186 ReduceProxy rp(cp, b, round, assigner, incoming_gids, outgoing_gids);
187 reduce(b, rp, partners);
188
189 // touch the outgoing queues to make sure they exist
190 Master::OutgoingQueues& outgoing = *cp.outgoing();
191 if (outgoing.size() < (size_t) rp.out_link().size())
192 for (int j = 0; j < rp.out_link().size(); ++j)
193 outgoing[rp.out_link().target(j)]; // touch the outgoing queue, creating it if necessary
194 }
195
196 unsigned round;
197 Callback reduce;
198 Partners partners;
199 const Assigner& assigner;
200 };
201
202 template<class Partners, class Skip>
203 struct SkipInactiveOr
204 {
205 SkipInactiveOr(int round_, const Partners& partners_, const Skip& skip_):
206 round(round_), partners(partners_), skip(skip_) {}
207 bool operator()(int i, const Master& master) const { return !partners.active(round, master.gid(i), master) || skip(round, i, master); }
208 int round;
209 const Partners& partners;
210 Skip skip;
211 };
212 }
213
214 } // diy
215
216 #endif // DIY_REDUCE_HPP
diy::ReduceProxy::out_link
const Link & out_link() const
returns outgoing link
Definition: reduce.hpp:70
diy::ReduceProxy
Enables communication within a group during a reduction. DIY creates the ReduceProxy for you in diy::...
Definition: reduce.hpp:15
diy::ReduceProxy::round
unsigned round() const
returns current round number
Definition: reduce.hpp:66
diy::BlockID
Definition: types.hpp:10
diy::Assigner::nblocks
int nblocks() const
returns the total number of global blocks
Definition: assigner.hpp:26
diy::Master::gid
int gid(int i) const
return gid of the i-th block
Definition: master.hpp:219
diy::reduce
void reduce(Master &master, const Assigner &assigner, const Partners &partners, const Reduce &reduce, const Skip &skip)
Implementation of the reduce communication pattern (includes swap-reduce, merge-reduce, and any other global communication).
Definition: reduce.hpp:109
diy::Master::size
unsigned int size() const
return the number of local blocks
Definition: master.hpp:233
diy::Assigner::rank
virtual int rank(int gid) const =0
returns the process rank of the block with global id gid (need not be local)
diy::ReduceProxy::set_round
void set_round(unsigned r)
advanced: change current round number
Definition: reduce.hpp:77
diy::Assigner
Definition: assigner.hpp:11
diy::ReduceProxy::ReduceProxy
ReduceProxy(const Master::Proxy &proxy, void *block, unsigned round, const Assigner &assigner, const Link &in_link, const Link &out_link)
Definition: reduce.hpp:49
diy::ReduceProxy::in_link
const Link & in_link() const
returns incoming link
Definition: reduce.hpp:68
diy::Master
Definition: master.hpp:35
diy::Link
Definition: link.hpp:15
diy::Master::Proxy
Communication proxy, used for enqueueing and dequeueing items for future exchange.
Definition: proxy.hpp:8
diy::ReduceProxy::ReduceProxy
ReduceProxy(const Master::Proxy &proxy, void *block, unsigned round, const Assigner &assigner, const GIDVector &incoming_gids, const GIDVector &outgoing_gids)
Definition: reduce.hpp:19
diy::ReduceProxy::block
void * block() const
returns pointer to block
Definition: reduce.hpp:64
diy::ReduceProxy::nblocks
int nblocks() const
returns total number of blocks
Definition: reduce.hpp:72
diy::ReduceProxy::assigner
const Assigner & assigner() const
returns the assigner
Definition: reduce.hpp:74