map_reduce: prevent mapper exception from poisoning state

In map_reduce, if the mapper throws, we can end up with poisoned
state. This is because the compiler can choose to move s->result
before evaluating f.get(), so s->result gets broken (if the mapped
type becomes invalid after move).

Fix by triggering the exception before the moving s->result, with
a sequence point to force the ordering.

A unit test is added (which does fail with clang before the patch,
though if the arguments are evaluated in reverse order, it can pass
with a different compiler).

Author: avikivity

include/seastar/core/map_reduce.hh

@@ -191,7 +191,8 @@ map_reduce(Iterator begin, Iterator end, Mapper&& mapper, Initial initial, Reduc
     while (begin != end) {
         ret = futurize_invoke(s->mapper, *begin++).then_wrapped([s = s.get(), ret = std::move(ret)] (auto f) mutable {
             try {
-                s->result = s->reduce(std::move(s->result), f.get());
+                auto mapped_value = f.get(); // Extract exception before touching s->result
+                s->result = s->reduce(std::move(s->result), std::move(mapped_value));
                 return std::move(ret);
             } catch (...) {
                 return std::move(ret).then_wrapped([ex = std::current_exception()] (auto f) {

tests/unit/futures_test.cc

@@ -862,6 +862,32 @@ SEASTAR_TEST_CASE(test_map_reduce1_lifetime) {
     });
 }
 
+SEASTAR_TEST_CASE(map_reduce_with_throwing_mapper) {
+    try {
+        auto vec = std::vector<int>{1, 2, 3, 4, 5, 6, 7};
+        co_await map_reduce(
+            vec,
+            // Mapper: identity function, but throws
+            [] (int x) -> future<int> {
+                if (x == 5) {
+                    throw std::runtime_error("test");
+                }
+                co_return x;
+            },
+            // Initial value (and accumulator): move-only type
+            std::make_unique<int>(0),
+            // Reducer: test that it does not act on a moved-from value
+            [] (std::unique_ptr<int> acc, int x) -> std::unique_ptr<int> {
+                BOOST_REQUIRE(bool(acc));
+                *acc += x;
+                return acc;
+            }
+        );
+    } catch (...) {
+        // Exception is expected and uninteresting
+    }
+}
+
 // This test doesn't actually test anything - it just waits for the future
 // returned by sleep to complete. However, a bug we had in sleep() caused
 // this test to fail the sanitizer in the debug build, so this is a useful