Merge branch 'master' of github.com:cloudius-systems/seastar into db

core/future-util.hh

@@ -297,7 +297,7 @@ map_reduce(Iterator begin, Iterator end, Mapper&& mapper, Reducer&& r)
 //   Initial
 template <typename Iterator, typename Mapper, typename Initial, typename Reduce>
 inline
-Initial
+future<Initial>
 map_reduce(Iterator begin, Iterator end, Mapper&& mapper, Initial initial, Reduce reduce) {
     struct state {
         Initial result;
@@ -306,12 +306,13 @@ map_reduce(Iterator begin, Iterator end, Mapper&& mapper, Initial initial, Reduc
     auto s = make_lw_shared(state{std::move(initial), std::move(reduce)});
     future<> ret = make_ready_future<>();
     while (begin != end) {
-        ret = mapper(*begin++).then([s = s.get()] (auto&& value) mutable {
+        ret = mapper(*begin++).then([s = s.get(), ret = std::move(ret)] (auto&& value) mutable {
             s->result = s->reduce(std::move(s->result), std::move(value));
+            return std::move(ret);
         });
     }
     return ret.then([s] {
-        return std::move(s->result);
+        return make_ready_future<Initial>(std::move(s->result));
     });
 }
 

tests/futures_test.cc

@@ -24,6 +24,7 @@
 #include "core/shared_ptr.hh"
 #include "core/semaphore.hh"
 #include "core/future-util.hh"
+#include <boost/iterator/counting_iterator.hpp>
 
 class expected_exception : std::runtime_error {
 public:
@@ -277,3 +278,13 @@ SEASTAR_TEST_CASE(test_when_all_iterator_range) {
         BOOST_REQUIRE(std::all_of(ret.begin(), ret.end(), [&ret] (auto& f) { return std::get<0>(f.get()) == size_t(&f - ret.data()); }));
     });
 }
+
+SEASTAR_TEST_CASE(test_map_reduce) {
+    auto square = [] (long x) { return make_ready_future<long>(x*x); };
+    long n = 1000;
+    return map_reduce(boost::make_counting_iterator<long>(0), boost::make_counting_iterator<long>(n),
+            square, long(0), std::plus<long>()).then([n] (auto result) {
+        auto m = n - 1; // counting does not include upper bound
+        BOOST_REQUIRE_EQUAL(result, (m * (m + 1) * (2*m + 1)) / 6);
+    });
+}