Thrill  0.1
flow_control_channel.cpp
Go to the documentation of this file.
1 /*******************************************************************************
2  * thrill/net/flow_control_channel.cpp
3  *
4  * Part of Project Thrill - http://project-thrill.org
5  *
6  * Copyright (C) 2016 Timo Bingmann <[email protected]>
7  *
8  * All rights reserved. Published under the BSD-2 license in the LICENSE file.
9  ******************************************************************************/
10 
11 #include <thrill/net/flow_control_channel.hpp>
12 
13 #include <functional>
14 
15 namespace thrill {
16 namespace net {
17 
18 /******************************************************************************/
19 // FlowControlChannel
20 
21 FlowControlChannel::FlowControlChannel(
22  Group& group, size_t local_id, size_t thread_count,
23  common::ThreadBarrier& barrier, LocalData* shmem,
24  std::atomic<size_t>& generation)
25  : group_(group),
26  host_rank_(group_.my_host_rank()), num_hosts_(group_.num_hosts()),
27  local_id_(local_id),
28  thread_count_(thread_count),
29  barrier_(barrier), shmem_(shmem), generation_(generation) { }
30 
31 FlowControlChannel::~FlowControlChannel() {
32  sLOGC(enable_stats)
33  << "FCC worker" << my_rank() << ":"
34  << "prefixsum"
35  << count_prefixsum_ << "in" << timer_prefixsum_
36  << "broadcast"
37  << count_broadcast_ << "in" << timer_broadcast_
38  << "reduce"
39  << count_reduce_ << "in" << timer_reduce_
40  << "allreduce"
41  << count_allreduce_ << "in" << timer_allreduce_
42  << "predecessor"
43  << count_predecessor_ << "in" << timer_predecessor_
44  << "barrier"
45  << count_barrier_ << "in" << timer_barrier_
46  << "communication"
47  << timer_communication_;
48 }
49 
50 void FlowControlChannel::Barrier() {
51  RunTimer run_timer(timer_barrier_);
52  if (enable_stats || debug) ++count_barrier_;
53 
54  LOG << "FCC::Barrier() ENTER count=" << count_barrier_;
55 
56  barrier_.wait(
57  [&]() {
58  RunTimer net_timer(timer_communication_);
59 
60  LOG << "FCC::Barrier() COMMUNICATE BEGIN"
61  << " count=" << count_barrier_;
62 
63  // Global all reduce
64  size_t i = 0;
65  group_.AllReduce(i);
66 
67  LOG << "FCC::Barrier() COMMUNICATE END"
68  << " count=" << count_barrier_;
69  });
70 
71  LOG << "FCC::Barrier() EXIT count=" << count_barrier_;
72 }
73 
74 void FlowControlChannel::LocalBarrier() {
75  barrier_.wait();
76 }
77 
78 /******************************************************************************/
79 // template instantiations
80 
81 template size_t FlowControlChannel::PrefixSumBase(
82  const size_t&, const std::plus<size_t>&, const size_t&, bool);
83 
84 template std::array<size_t, 2> FlowControlChannel::PrefixSumBase(
85  const std::array<size_t, 2>&,
86  const common::ComponentSum<std::array<size_t, 2> >&,
87  const std::array<size_t, 2>&, bool);
88 template std::array<size_t, 3> FlowControlChannel::PrefixSumBase(
89  const std::array<size_t, 3>&,
90  const common::ComponentSum<std::array<size_t, 3> >&,
91  const std::array<size_t, 3>&, bool);
92 template std::array<size_t, 4> FlowControlChannel::PrefixSumBase(
93  const std::array<size_t, 4>&,
94  const common::ComponentSum<std::array<size_t, 4> >&,
95  const std::array<size_t, 4>&, bool);
96 
97 template size_t FlowControlChannel::ExPrefixSumTotal(
98  size_t&, const std::plus<size_t>&, const size_t&);
99 
100 template std::array<size_t, 2> FlowControlChannel::ExPrefixSumTotal(
101  std::array<size_t, 2>&,
102  const common::ComponentSum<std::array<size_t, 2> >&,
103  const std::array<size_t, 2>&);
104 template std::array<size_t, 3> FlowControlChannel::ExPrefixSumTotal(
105  std::array<size_t, 3>&,
106  const common::ComponentSum<std::array<size_t, 3> >&,
107  const std::array<size_t, 3>&);
108 template std::array<size_t, 4> FlowControlChannel::ExPrefixSumTotal(
109  std::array<size_t, 4>&,
110  const common::ComponentSum<std::array<size_t, 4> >&,
111  const std::array<size_t, 4>&);
112 
113 template size_t FlowControlChannel::Broadcast(const size_t&, size_t);
114 
115 template std::array<size_t, 2> FlowControlChannel::Broadcast(
116  const std::array<size_t, 2>&, size_t);
117 template std::array<size_t, 3> FlowControlChannel::Broadcast(
118  const std::array<size_t, 3>&, size_t);
119 template std::array<size_t, 4> FlowControlChannel::Broadcast(
120  const std::array<size_t, 4>&, size_t);
121 
122 template size_t FlowControlChannel::AllReduce(
123  const size_t&, const std::plus<size_t>&);
124 
125 } // namespace net
126 } // namespace thrill
127 
128 /******************************************************************************/
thrill::net::FlowControlChannel::count_broadcast_
common::AtomicMovable< size_t > count_broadcast_
Definition: flow_control_channel.hpp:86
thrill::net::FlowControlChannel::Barrier
void Barrier()
A trivial global barrier.
Definition: flow_control_channel.cpp:50
thrill::net::FlowControlChannel::timer_prefixsum_
Timer timer_prefixsum_
Synchronization timer.
Definition: flow_control_channel.hpp:75
sLOGC
#define sLOGC(cond)
Explicitly specify the condition for logging.
Definition: logger.hpp:31
thrill::net::FlowControlChannel::count_barrier_
common::AtomicMovable< size_t > count_barrier_
Definition: flow_control_channel.hpp:90
flow_control_channel.hpp
thrill::common::RunTimer
RIAA class for running the timer until destruction.
Definition: stats_timer.hpp:298
thrill::net::FlowControlChannel::count_allreduce_
common::AtomicMovable< size_t > count_allreduce_
Definition: flow_control_channel.hpp:88
thrill::net::FlowControlChannel::count_prefixsum_
common::AtomicMovable< size_t > count_prefixsum_
Synchronization counters.
Definition: flow_control_channel.hpp:85
thrill::net::FlowControlChannel::AllReduce
T TLX_ATTRIBUTE_WARN_UNUSED_RESULT AllReduce(const T &value, const BinarySumOp &sum_op=BinarySumOp())
Reduces a value of a serializable type T over all workers given a certain reduce function.
Definition: flow_control_channel.hpp:599
thrill::net::FlowControlChannel::PrefixSumBase
T TLX_ATTRIBUTE_WARN_UNUSED_RESULT PrefixSumBase(const T &value, const BinarySumOp &sum_op=BinarySumOp(), const T &initial=T(), bool inclusive=true)
Calculates the prefix sum over all workers, given a certain sum operation.
Definition: flow_control_channel.hpp:236
thrill::common::ComponentSum
template for computing the component-wise sum of std::array or std::vector.
Definition: functional.hpp:94
thrill::net::Group::AllReduce
void AllReduce(T &value, BinarySumOp sum_op=BinarySumOp())
Reduce a value from all workers to all workers.
Definition: collective.hpp:568
thrill::net::FlowControlChannel::group_
Group & group_
The group associated with this channel.
Definition: flow_control_channel.hpp:55
thrill::net::FlowControlChannel::count_predecessor_
common::AtomicMovable< size_t > count_predecessor_
Definition: flow_control_channel.hpp:89
thrill::net::FlowControlChannel::LocalBarrier
void LocalBarrier()
A trivial local thread barrier.
Definition: flow_control_channel.cpp:74
thrill::net::FlowControlChannel::ExPrefixSumTotal
T TLX_ATTRIBUTE_WARN_UNUSED_RESULT ExPrefixSumTotal(T &value, const BinarySumOp &sum_op=BinarySumOp(), const T &initial=T())
Calculates the exclusive prefix sum over all workers, and delivers the total sum as well...
Definition: flow_control_channel.hpp:351
thrill::net::Group
A network Group is a collection of enumerated communication links, which provides point-to-point comm...
Definition: group.hpp:47
tlx::ThreadBarrierSpin::wait
void wait(Lambda lambda=Lambda())
Waits for n threads to arrive.
Definition: thread_barrier_spin.hpp:47
thrill::net::FlowControlChannel::my_rank
size_t my_rank() const
Return the worker&#39;s global rank.
Definition: flow_control_channel.hpp:197
thrill::net::FlowControlChannel::Broadcast
T TLX_ATTRIBUTE_WARN_UNUSED_RESULT Broadcast(const T &value, size_t origin=0)
Broadcasts a value of a serializable type T from the master (the worker with id 0) to all other worke...
Definition: flow_control_channel.hpp:424
LOG
#define LOG
Default logging method: output if the local debug variable is true.
Definition: logger.hpp:24
thrill::net::FlowControlChannel::count_reduce_
common::AtomicMovable< size_t > count_reduce_
Definition: flow_control_channel.hpp:87
tlx::ThreadBarrierSpin
Implements a thread barrier using atomics and a spin lock that can be used to synchronize threads...
Definition: thread_barrier_spin.hpp:29
thrill::net::FlowControlChannel::FlowControlChannel
FlowControlChannel(Group &group, size_t local_id, size_t thread_count, common::ThreadBarrier &barrier, LocalData *shmem, std::atomic< size_t > &generation)
Creates a new instance of this class, wrapping a net::Group.
Definition: flow_control_channel.cpp:21