Thrill  0.1
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Groups Pages
prefixsum.hpp
Go to the documentation of this file.
1 /*******************************************************************************
2  * thrill/api/prefixsum.hpp
3  *
4  * Part of Project Thrill - http://project-thrill.org
5  *
6  * Copyright (C) 2015 Alexander Noe <[email protected]>
7  * Copyright (C) 2015 Timo Bingmann <[email protected]>
8  *
9  * All rights reserved. Published under the BSD-2 license in the LICENSE file.
10  ******************************************************************************/
11 
12 #pragma once
13 #ifndef THRILL_API_PREFIXSUM_HEADER
14 #define THRILL_API_PREFIXSUM_HEADER
15 
16 #include <thrill/api/dia.hpp>
17 #include <thrill/api/dop_node.hpp>
18 #include <thrill/common/logger.hpp>
19 #include <thrill/data/file.hpp>
20 
21 namespace thrill {
22 namespace api {
23 
24 /*!
25  * \ingroup api_layer
26  */
27 template <typename ValueType, typename SumFunction>
28 class PrefixSumNode final : public DOpNode<ValueType>
29 {
30  static constexpr bool debug = false;
31 
33  using Super::context_;
34 
35 public:
36  template <typename ParentDIA>
37  PrefixSumNode(const ParentDIA& parent,
38  const SumFunction& sum_function,
39  const ValueType& initial_element)
40  : Super(parent.ctx(), "PrefixSum", { parent.id() }, { parent.node() }),
41  sum_function_(sum_function),
42  local_sum_(initial_element),
43  initial_element_(initial_element),
44  parent_stack_empty_(ParentDIA::stack_empty)
45  {
46  // Hook PreOp(s)
47  auto pre_op_fn = [this](const ValueType& input) {
48  PreOp(input);
49  };
50 
51  auto lop_chain = parent.stack().push(pre_op_fn).fold();
52  parent.node()->AddChild(this, lop_chain);
53  }
54 
55  //! PreOp: compute local prefixsum and store items.
56  void PreOp(const ValueType& input) {
57  LOG << "Input: " << input;
58  local_sum_ = sum_function_(local_sum_, input);
59  writer_.Put(input);
60  }
61 
62  bool OnPreOpFile(const data::File& file, size_t /* parent_index */) final {
63  if (!parent_stack_empty_) {
65  << "PrefixSum rejected File from parent "
66  << "due to non-empty function stack.";
67  return false;
68  }
69  // copy complete Block references to writer_
70  file_ = file.Copy();
71  // read File for prefix sum.
72  auto reader = file_.GetKeepReader();
73  while (reader.HasNext()) {
74  local_sum_ = sum_function_(
75  local_sum_, reader.template Next<ValueType>());
76  }
77  return true;
78  }
79 
80  void StopPreOp(size_t /* parent_index */) final {
81  writer_.Close();
82  }
83 
84  //! Executes the prefixsum operation.
85  void Execute() final {
86  LOG << "MainOp processing";
87 
88  ValueType sum = context_.net.ExPrefixSum(
89  local_sum_, initial_element_, sum_function_);
90 
91  if (context_.my_rank() == 0) {
92  sum = initial_element_;
93  }
94 
95  local_sum_ = sum;
96  }
97 
98  void PushData(bool consume) final {
99  data::File::Reader reader = file_.GetReader(consume);
100 
101  ValueType sum = local_sum_;
102  size_t num_items = file_.num_items();
103 
104  for (size_t i = 0; i < num_items; ++i) {
105  sum = sum_function_(sum, reader.Next<ValueType>());
106  this->PushItem(sum);
107  }
108  }
109 
110  void Dispose() final {
111  file_.Clear();
112  }
113 
114 private:
115  //! The sum function which is applied to two elements.
116  SumFunction sum_function_;
117  //! Local sum to be used in all reduce operation.
118  ValueType local_sum_;
119  //! Initial element.
120  const ValueType initial_element_;
121  //! Whether the parent stack is empty
123 
124  //! Local data file
126  //! Data writer to local file (only active in PreOp).
128 };
129 
130 template <typename ValueType, typename Stack>
131 template <typename SumFunction>
133  const SumFunction& sum_function, const ValueType& initial_element) const {
134  assert(IsValid());
135 
137 
138  static_assert(
139  std::is_convertible<
140  ValueType,
142  >::value,
143  "SumFunction has the wrong input type");
144 
145  static_assert(
146  std::is_convertible<
147  ValueType,
148  typename FunctionTraits<SumFunction>::template arg<1> >::value,
149  "SumFunction has the wrong input type");
150 
151  static_assert(
152  std::is_convertible<
154  ValueType>::value,
155  "SumFunction has the wrong input type");
156 
157  auto node = tlx::make_counting<PrefixSumNode>(
158  *this, sum_function, initial_element);
159 
160  return DIA<ValueType>(node);
161 }
162 
163 } // namespace api
164 } // namespace thrill
165 
166 #endif // !THRILL_API_PREFIXSUM_HEADER
167 
168 /******************************************************************************/
net::FlowControlChannel & net
Definition: context.hpp:443
virtual void Dispose()
Virtual clear method. Triggers actual disposing in sub-classes.
Definition: dia_base.hpp:188
size_t num_items() const
Return the number of items in the file.
Definition: file.hpp:178
DIA is the interface between the user and the Thrill framework.
Definition: dia.hpp:141
ValueType local_sum_
Local sum to be used in all reduce operation.
Definition: prefixsum.hpp:40
ValueType_ ValueType
Definition: dia.hpp:152
size_t my_rank() const
Global rank of this worker among all other workers in the system.
Definition: context.hpp:240
A File is an ordered sequence of Block objects for storing items.
Definition: file.hpp:56
void Clear()
Free all Blocks in the File and deallocate vectors.
Definition: file.cpp:57
PrefixSumNode(const ParentDIA &parent, const SumFunction &sum_function, const ValueType &initial_element)
Definition: prefixsum.hpp:37
static constexpr bool debug
Definition: prefixsum.hpp:30
BlockWriter contains a temporary Block object into which a) any serializable item can be stored or b)...
static constexpr bool g_debug_push_file
Definition: config.hpp:44
T TLX_ATTRIBUTE_WARN_UNUSED_RESULT ExPrefixSum(const T &value, const T &initial=T(), const BinarySumOp &sum_op=BinarySumOp())
Calculates the exclusive prefix sum over all workers, given a certain sum operation.
virtual void PushData(bool consume)=0
Virtual method for pushing data. Triggers actual pushing in sub-classes.
data::File::Writer writer_
Data writer to local file (only active in PreOp).
Definition: prefixsum.hpp:127
const bool parent_stack_empty_
Whether the parent stack is empty.
Definition: prefixsum.hpp:122
const ValueType initial_element_
Initial element.
Definition: prefixsum.hpp:120
KeepReader GetKeepReader(size_t num_prefetch=File::default_prefetch) const
Get BlockReader for beginning of File.
Definition: file.cpp:68
void PushItem(const ValueType &item) const
Method for derived classes to Push a single item to all children.
Definition: dia_node.hpp:147
auto PrefixSum(const SumFunction &sum_function=SumFunction(), const ValueType &initial_element=ValueType()) const
PrefixSum is a DOp, which computes the prefix sum of all elements.
Definition: prefixsum.hpp:132
File Copy() const
Return a copy of the File (explicit copy-constructor)
Definition: file.cpp:42
virtual void StopPreOp(size_t)
Virtual method for preparing end of PushData.
Definition: dia_base.hpp:173
#define LOGC(cond)
Explicitly specify the condition for logging.
Definition: logger.hpp:167
virtual void Execute()=0
Virtual execution method. Triggers actual computation in sub-classes.
int value
Definition: gen_data.py:41
common::FunctionTraits< Function > FunctionTraits
alias for convenience.
Definition: dia.hpp:147
data::File GetFile(size_t dia_id)
Returns a new File object containing a sequence of local Blocks.
Definition: context.hpp:280
Reader GetReader(bool consume, size_t num_prefetch=File::default_prefetch)
Get BlockReader or a consuming BlockReader for beginning of File.
Definition: file.cpp:78
A DOpNode is a typed node representing and distributed operations in Thrill.
Definition: dop_node.hpp:32
TLX_ATTRIBUTE_ALWAYS_INLINE BlockWriter & Put(const T &x)
Put appends a complete item, or fails with a FullException.
data::File file_
Local data file.
Definition: prefixsum.hpp:125
virtual bool OnPreOpFile(const data::File &, size_t)
Definition: dia_base.hpp:168
void Close()
Explicitly close the writer.
DynBlockReader Reader
Definition: file.hpp:60
Writer GetWriter(size_t block_size=default_block_size)
Get BlockWriter.
Definition: file.cpp:63
Context & context_
associated Context
Definition: dia_base.hpp:293
#define LOG
Default logging method: output if the local debug variable is true.
Definition: logger.hpp:172