Seastar
High performance C++ framework for concurrent servers
map_reduce.hh
1 /*
2  * This file is open source software, licensed to you under the terms
3  * of the Apache License, Version 2.0 (the "License"). See the NOTICE file
4  * distributed with this work for additional information regarding copyright
5  * ownership. You may not use this file except in compliance with the License.
6  *
7  * You may obtain a copy of the License at
8  *
9  * http://www.apache.org/licenses/LICENSE-2.0
10  *
11  * Unless required by applicable law or agreed to in writing,
12  * software distributed under the License is distributed on an
13  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14  * KIND, either express or implied. See the License for the
15  * specific language governing permissions and limitations
16  * under the License.
17  */
18 
19 /*
20  * Copyright (C) 2020 ScyllaDB.
21  */
22 
23 #pragma once
24 
25 #include <iterator>
26 
27 #include <seastar/core/future.hh>
28 #include <seastar/core/shared_ptr.hh>
29 
30 namespace seastar {
31 
34 
36 
37 template <typename T, typename Ptr, bool IsFuture>
38 struct reducer_with_get_traits;
39 
40 template <typename T, typename Ptr>
41 struct reducer_with_get_traits<T, Ptr, false> {
42  using result_type = decltype(std::declval<T>().get());
43  using future_type = future<result_type>;
44  static future_type maybe_call_get(future<> f, Ptr r) {
45  return f.then([r = std::move(r)] () mutable {
46  return make_ready_future<result_type>(std::move(r->reducer).get());
47  });
48  }
49 };
50 
51 template <typename T, typename Ptr>
52 struct reducer_with_get_traits<T, Ptr, true> {
53  using future_type = decltype(std::declval<T>().get());
54  static future_type maybe_call_get(future<> f, Ptr r) {
55  return f.then([r = std::move(r)] () mutable {
56  return r->reducer.get();
57  }).then_wrapped([r] (future_type f) {
58  return f;
59  });
60  }
61 };
62 
63 template <typename T, typename Ptr = lw_shared_ptr<T>, typename V = void>
64 struct reducer_traits {
65  using future_type = future<>;
66  static future_type maybe_call_get(future<> f, Ptr r) {
67  return f.then([r = std::move(r)] {});
68  }
69 };
70 
71 template <typename T, typename Ptr>
72 struct reducer_traits<T, Ptr, decltype(std::declval<T>().get(), void())> : public reducer_with_get_traits<T, Ptr, is_future<std::invoke_result_t<decltype(&T::get),T>>::value> {};
73 
75 
85 
86 // TODO: specialize for non-deferring reducer
87 template <typename Iterator, typename Mapper, typename Reducer>
88 SEASTAR_CONCEPT( requires requires (Iterator i, Mapper mapper, Reducer reduce) {
89  *i++;
90  { i != i } -> std::convertible_to<bool>;
91  mapper(*i);
92  reduce(futurize_invoke(mapper, *i).get0());
93 } )
94 inline
95 auto
96 map_reduce(Iterator begin, Iterator end, Mapper&& mapper, Reducer&& r)
97  -> typename reducer_traits<Reducer>::future_type
98 {
99  struct state {
100  Mapper mapper;
101  Reducer reducer;
102  };
103  auto s = make_lw_shared(state{std::forward<Mapper>(mapper), std::forward<Reducer>(r)});
104  future<> ret = make_ready_future<>();
105  while (begin != end) {
106  ret = futurize_invoke(s->mapper, *begin++).then_wrapped([ret = std::move(ret), s] (auto f) mutable {
107  return ret.then_wrapped([f = std::move(f), s] (auto rf) mutable {
108  if (rf.failed()) {
109  f.ignore_ready_future();
110  return rf;
111  } else {
112  return futurize_invoke(s->reducer, std::move(f.get0()));
113  }
114  });
115  });
116  }
117  return reducer_traits<Reducer, lw_shared_ptr<state>>::maybe_call_get(std::move(ret), s);
118 }
119 
155 template <typename Iterator, typename Mapper, typename Initial, typename Reduce>
156 SEASTAR_CONCEPT( requires requires (Iterator i, Mapper mapper, Initial initial, Reduce reduce) {
157  *i++;
158  { i != i} -> std::convertible_to<bool>;
159  mapper(*i);
160  requires is_future<decltype(mapper(*i))>::value;
161  { reduce(std::move(initial), mapper(*i).get0()) } -> std::convertible_to<Initial>;
162 } )
163 inline
164 future<Initial>
165 map_reduce(Iterator begin, Iterator end, Mapper&& mapper, Initial initial, Reduce reduce) {
166  struct state {
167  Mapper mapper;
168  Initial result;
169  Reduce reduce;
170  };
171  auto s = make_lw_shared(state{std::forward<Mapper>(mapper), std::move(initial), std::move(reduce)});
172  future<> ret = make_ready_future<>();
173  while (begin != end) {
174  ret = futurize_invoke(s->mapper, *begin++).then_wrapped([s = s.get(), ret = std::move(ret)] (auto f) mutable {
175  try {
176  s->result = s->reduce(std::move(s->result), std::move(f.get0()));
177  return std::move(ret);
178  } catch (...) {
179  return std::move(ret).then_wrapped([ex = std::current_exception()] (auto f) {
181  return make_exception_future<>(ex);
182  });
183  }
184  });
185  }
186  return ret.then([s] {
187  return make_ready_future<Initial>(std::move(s->result));
188  });
189 }
190 
226 template <typename Range, typename Mapper, typename Initial, typename Reduce>
227 SEASTAR_CONCEPT( requires requires (Range range, Mapper mapper, Initial initial, Reduce reduce) {
228  std::begin(range);
229  std::end(range);
230  mapper(*std::begin(range));
231  requires is_future<std::remove_reference_t<decltype(mapper(*std::begin(range)))>>::value;
232  { reduce(std::move(initial), mapper(*std::begin(range)).get0()) } -> std::convertible_to<Initial>;
233 } )
234 inline
235 future<Initial>
236 map_reduce(Range&& range, Mapper&& mapper, Initial initial, Reduce reduce) {
237  return map_reduce(std::begin(range), std::end(range), std::forward<Mapper>(mapper),
238  std::move(initial), std::move(reduce));
239 }
240 
243 template <typename Result, typename Addend = Result>
244 class adder {
245 private:
246  Result _result;
247 public:
248  future<> operator()(const Addend& value) {
249  _result += value;
250  return make_ready_future<>();
251  }
252  Result get() && {
253  return std::move(_result);
254  }
255 };
256 
258 
259 } // namespace seastar
Definition: map_reduce.hh:244
Result then(Func &&func) noexcept
Schedule a block of code to run when the future is ready.
Definition: future.hh:1527
void ignore_ready_future() noexcept
Ignore any result hold by this future.
Definition: future.hh:1885
auto map_reduce(Iterator begin, Iterator end, Mapper &&mapper, Reducer &&r) -> typename reducer_traits< Reducer >::future_type
Definition: map_reduce.hh:96
future< Initial > map_reduce(Range &&range, Mapper &&mapper, Initial initial, Reduce reduce)
Definition: map_reduce.hh:236
Seastar API namespace.
Definition: abort_on_ebadf.hh:24