1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
|
/*
* This file is open source software, licensed to you under the terms
* of the Apache License, Version 2.0 (the "License"). See the NOTICE file
* distributed with this work for additional information regarding copyright
* ownership. You may not use this file except in compliance with the License.
*
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
/*
* Copyright (C) 2020 ScyllaDB.
*/
#pragma once
#include <iterator>
#include <seastar/core/future.hh>
#include <seastar/core/shared_ptr.hh>
namespace seastar {
/// \addtogroup future-util
/// @{
/// \cond internal
template <typename T, bool IsFuture>
struct reducer_with_get_traits;
template <typename T>
struct reducer_with_get_traits<T, false> {
using result_type = decltype(std::declval<T>().get());
using future_type = future<result_type>;
static future_type maybe_call_get(future<> f, lw_shared_ptr<T> r) {
return f.then([r = std::move(r)] () mutable {
return make_ready_future<result_type>(std::move(*r).get());
});
}
};
template <typename T>
struct reducer_with_get_traits<T, true> {
using future_type = decltype(std::declval<T>().get());
static future_type maybe_call_get(future<> f, lw_shared_ptr<T> r) {
return f.then([r = std::move(r)] () mutable {
return r->get();
}).then_wrapped([r] (future_type f) {
return f;
});
}
};
template <typename T, typename V = void>
struct reducer_traits {
using future_type = future<>;
static future_type maybe_call_get(future<> f, lw_shared_ptr<T> r) {
return f.then([r = std::move(r)] {});
}
};
template <typename T>
struct reducer_traits<T, decltype(std::declval<T>().get(), void())> : public reducer_with_get_traits<T, is_future<std::result_of_t<decltype(&T::get)(T)>>::value> {};
/// \endcond
/// Map a function over a range and reduce the result.
///
/// \param begin an \c InputIterator designating the beginning of the range
/// \param end an \c InputIterator designating the end of the range
/// \param mapper is a callable which transforms values from the iterator range into a future<T>
/// \param r is an object which can be called with T as parameter and yields a future<>
/// It may have a get() method which returns a value of type U which holds the result of reduction.
/// \return Th reduced value wrapped in a future.
/// If the reducer has no get() method then this function returns future<>.
// TODO: specialize for non-deferring reducer
template <typename Iterator, typename Mapper, typename Reducer>
SEASTAR_CONCEPT( requires requires (Iterator i, Mapper mapper, Reducer reduce) {
*i++;
{ i != i } -> std::convertible_to<bool>;
mapper(*i);
reduce(futurize_invoke(mapper, *i).get0());
} )
inline
auto
map_reduce(Iterator begin, Iterator end, Mapper&& mapper, Reducer&& r)
-> typename reducer_traits<Reducer>::future_type
{
auto r_ptr = make_lw_shared(std::forward<Reducer>(r));
future<> ret = make_ready_future<>();
while (begin != end) {
ret = futurize_invoke(mapper, *begin++).then_wrapped([ret = std::move(ret), r_ptr] (auto f) mutable {
return ret.then_wrapped([f = std::move(f), r_ptr] (auto rf) mutable {
if (rf.failed()) {
f.ignore_ready_future();
return std::move(rf);
} else {
return futurize_invoke(*r_ptr, std::move(f.get0()));
}
});
});
}
return reducer_traits<Reducer>::maybe_call_get(std::move(ret), r_ptr);
}
/// Asynchronous map/reduce transformation.
///
/// Given a range of objects, an asynchronous unary function
/// operating on these objects, an initial value, and a
/// binary function for reducing, map_reduce() will
/// transform each object in the range, then invoke
/// the the reducing function with the result.
///
/// Example:
///
/// Calculate the total size of several files:
///
/// \code
/// map_reduce(files.begin(), files.end(),
/// std::mem_fn(file::size),
/// size_t(0),
/// std::plus<size_t>())
/// \endcode
///
/// Requirements:
/// - Iterator: an InputIterator.
/// - Mapper: unary function taking Iterator::value_type and producing a future<...>.
/// - Initial: any value type
/// - Reduce: a binary function taking two Initial values and returning an Initial
///
/// Return type:
/// - future<Initial>
///
/// \param begin beginning of object range to operate on
/// \param end end of object range to operate on
/// \param mapper map function to call on each object, returning a future
/// \param initial initial input value to reduce function
/// \param reduce binary function for merging two result values from \c mapper
///
/// \return equivalent to \c reduce(reduce(initial, mapper(obj0)), mapper(obj1)) ...
template <typename Iterator, typename Mapper, typename Initial, typename Reduce>
SEASTAR_CONCEPT( requires requires (Iterator i, Mapper mapper, Initial initial, Reduce reduce) {
*i++;
{ i != i} -> std::convertible_to<bool>;
mapper(*i);
requires is_future<decltype(mapper(*i))>::value;
{ reduce(std::move(initial), mapper(*i).get0()) } -> std::convertible_to<Initial>;
} )
inline
future<Initial>
map_reduce(Iterator begin, Iterator end, Mapper&& mapper, Initial initial, Reduce reduce) {
struct state {
Initial result;
Reduce reduce;
};
auto s = make_lw_shared(state{std::move(initial), std::move(reduce)});
future<> ret = make_ready_future<>();
while (begin != end) {
ret = futurize_invoke(mapper, *begin++).then_wrapped([s = s.get(), ret = std::move(ret)] (auto f) mutable {
try {
s->result = s->reduce(std::move(s->result), std::move(f.get0()));
return std::move(ret);
} catch (...) {
return std::move(ret).then_wrapped([ex = std::current_exception()] (auto f) {
f.ignore_ready_future();
return make_exception_future<>(ex);
});
}
});
}
return ret.then([s] {
return make_ready_future<Initial>(std::move(s->result));
});
}
/// Asynchronous map/reduce transformation (range version).
///
/// Given a range of objects, an asynchronous unary function
/// operating on these objects, an initial value, and a
/// binary function for reducing, map_reduce() will
/// transform each object in the range, then invoke
/// the the reducing function with the result.
///
/// Example:
///
/// Calculate the total size of several files:
///
/// \code
/// std::vector<file> files = ...;
/// map_reduce(files,
/// std::mem_fn(file::size),
/// size_t(0),
/// std::plus<size_t>())
/// \endcode
///
/// Requirements:
/// - Iterator: an InputIterator.
/// - Mapper: unary function taking Iterator::value_type and producing a future<...>.
/// - Initial: any value type
/// - Reduce: a binary function taking two Initial values and returning an Initial
///
/// Return type:
/// - future<Initial>
///
/// \param range object range to operate on
/// \param mapper map function to call on each object, returning a future
/// \param initial initial input value to reduce function
/// \param reduce binary function for merging two result values from \c mapper
///
/// \return equivalent to \c reduce(reduce(initial, mapper(obj0)), mapper(obj1)) ...
template <typename Range, typename Mapper, typename Initial, typename Reduce>
SEASTAR_CONCEPT( requires requires (Range range, Mapper mapper, Initial initial, Reduce reduce) {
std::begin(range);
std::end(range);
mapper(*std::begin(range));
requires is_future<std::remove_reference_t<decltype(mapper(*std::begin(range)))>>::value;
{ reduce(std::move(initial), mapper(*std::begin(range)).get0()) } -> std::convertible_to<Initial>;
} )
inline
future<Initial>
map_reduce(Range&& range, Mapper&& mapper, Initial initial, Reduce reduce) {
return map_reduce(std::begin(range), std::end(range), std::forward<Mapper>(mapper),
std::move(initial), std::move(reduce));
}
/// Implements @Reducer concept. Calculates the result by
/// adding elements to the accumulator.
template <typename Result, typename Addend = Result>
class adder {
private:
Result _result;
public:
future<> operator()(const Addend& value) {
_result += value;
return make_ready_future<>();
}
Result get() && {
return std::move(_result);
}
};
/// @}
} // namespace seastar
|
