Thrill  0.1
prefix_sum.hpp
Go to the documentation of this file.
1 /*******************************************************************************
2  * thrill/api/prefix_sum.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_PREFIX_SUM_HEADER
14 #define THRILL_API_PREFIX_SUM_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, bool Inclusive>
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 char* label,
39  const SumFunction& sum_function,
40  const ValueType& initial_element)
41  : Super(parent.ctx(), label, { parent.id() }, { parent.node() }),
42  sum_function_(sum_function),
43  local_sum_(),
44  initial_element_(initial_element),
45  parent_stack_empty_(ParentDIA::stack_empty) {
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;
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()) {
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 
90  }
91 
92  void PushData(bool consume) final {
93  data::File::Reader reader = file_.GetReader(consume);
94  size_t num_items = file_.num_items();
95 
96  if (Inclusive) {
97  ValueType sum = local_sum_;
98  for (size_t i = 0; i < num_items; ++i) {
99  sum = sum_function_(sum, reader.Next<ValueType>());
100  this->PushItem(sum);
101  }
102  }
103  else {
104  ValueType sum = local_sum_;
105  for (size_t i = 0; i < num_items; ++i) {
106  this->PushItem(sum);
107  sum = sum_function_(sum, reader.Next<ValueType>());
108  }
109  }
110  }
111 
112  void Dispose() final {
113  file_.Clear();
114  }
115 
116 private:
117  //! The sum function which is applied to two elements.
118  SumFunction sum_function_;
119  //! Local sum to be used in all reduce operation.
120  ValueType local_sum_;
121  //! Initial element.
122  const ValueType initial_element_;
123  //! Whether the parent stack is empty
125 
126  //! Local data file
128  //! Data writer to local file (only active in PreOp).
130 };
131 
132 template <typename ValueType, typename Stack>
133 template <typename SumFunction>
135  const SumFunction& sum_function, const ValueType& initial_element) const {
136  assert(IsValid());
137 
139  ValueType, SumFunction, /* Inclusive */ true>;
140 
141  static_assert(
142  std::is_convertible<
143  ValueType,
145  >::value,
146  "SumFunction has the wrong input type");
147 
148  static_assert(
149  std::is_convertible<
150  ValueType,
151  typename FunctionTraits<SumFunction>::template arg<1> >::value,
152  "SumFunction has the wrong input type");
153 
154  static_assert(
155  std::is_convertible<
157  ValueType>::value,
158  "SumFunction has the wrong input type");
159 
160  auto node = tlx::make_counting<PrefixSumNode>(
161  *this, "PrefixSum", sum_function, initial_element);
162 
163  return DIA<ValueType>(node);
164 }
165 
166 } // namespace api
167 } // namespace thrill
168 
169 #endif // !THRILL_API_PREFIX_SUM_HEADER
170 
171 /******************************************************************************/
net::FlowControlChannel & net
Definition: context.hpp:446
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: prefix_sum.hpp:120
void PushItem(const ValueType &item) const
Method for derived classes to Push a single item to all children.
Definition: dia_node.hpp:147
KeepReader GetKeepReader(size_t prefetch_size=File::default_prefetch_size_) const
Get BlockReader for beginning of File.
Definition: file.cpp:68
void StopPreOp(size_t) final
Virtual method for preparing end of PushData.
Definition: prefix_sum.hpp:80
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
TLX_ATTRIBUTE_ALWAYS_INLINE T Next()
Next() reads a complete item T.
const char * label() const
return label() of DIANode subclass as stored by StatsNode
Definition: dia_base.hpp:218
T TLX_ATTRIBUTE_WARN_UNUSED_RESULT ExPrefixSum(const T &value, const BinarySumOp &sum_op=BinarySumOp(), const T &initial=T())
Calculates the exclusive prefix sum over all workers, given a certain sum operation.
static constexpr bool debug
Definition: prefix_sum.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
data::File::Writer writer_
Data writer to local file (only active in PreOp).
Definition: prefix_sum.hpp:129
auto PrefixSum(const SumFunction &sum_function=SumFunction(), const ValueType &initial_element=ValueType()) const
PrefixSum is a DOp, which computes the (inclusive) prefix sum of all elements.
Definition: prefix_sum.hpp:134
const bool parent_stack_empty_
Whether the parent stack is empty.
Definition: prefix_sum.hpp:124
const ValueType initial_element_
Initial element.
Definition: prefix_sum.hpp:122
void Dispose() final
Virtual clear method. Triggers actual disposing in sub-classes.
Definition: prefix_sum.hpp:112
void PreOp(const ValueType &input)
PreOp: compute local prefixsum and store items.
Definition: prefix_sum.hpp:56
int value
Definition: gen_data.py:41
BlockReader takes Block objects from BlockSource and allows reading of a) serializable Items or b) ar...
common::FunctionTraits< Function > FunctionTraits
alias for convenience.
Definition: dia.hpp:147
void PushData(bool consume) final
Virtual method for pushing data. Triggers actual pushing in sub-classes.
Definition: prefix_sum.hpp:92
data::File GetFile(size_t dia_id)
Returns a new File object containing a sequence of local Blocks.
Definition: context.hpp:283
A DOpNode is a typed node representing and distributed operations in Thrill.
Definition: dop_node.hpp:32
Reader GetReader(bool consume, size_t prefetch_size=File::default_prefetch_size_)
Get BlockReader or a consuming BlockReader for beginning of File.
Definition: file.cpp:78
void Execute() final
Executes the prefixsum operation.
Definition: prefix_sum.hpp:85
File Copy() const
Return a copy of the File (explicit copy-constructor)
Definition: file.cpp:42
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: prefix_sum.hpp:127
SumFunction sum_function_
The sum function which is applied to two elements.
Definition: prefix_sum.hpp:118
size_t num_items() const
Return the number of items in the file.
Definition: file.hpp:180
void Close()
Explicitly close the writer.
#define LOG
Default logging method: output if the local debug variable is true.
Definition: logger.hpp:24
PrefixSumNode(const ParentDIA &parent, const char *label, const SumFunction &sum_function, const ValueType &initial_element)
Definition: prefix_sum.hpp:37
Writer GetWriter(size_t block_size=default_block_size)
Get BlockWriter.
Definition: file.cpp:63
bool OnPreOpFile(const data::File &file, size_t) final
Definition: prefix_sum.hpp:62
Context & context_
associated Context
Definition: dia_base.hpp:293
#define LOGC(cond)
Explicitly specify the condition for logging.
Definition: logger.hpp:21