crimson/common: extract parallel_for_each_state out

if `parallel_for_each_state` is defined as a nested class in errorator,
clang fails to compile it:

../src/crimson/common/errorator.h:716:47: error: no class named 'parallel_for_each_state' in 'errorator<AllowedErrors...>'
    friend class errorator<AllowedErrors...>::parallel_for_each_state;
                 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^

and the forward declaration does not help. so we have to extract it
out of the errorator. to speed up the compilation, it is moved into
errorator-loop.h. its name mirrors `include/seastar/core/loop.h`.

we could extract the `errorator<>::parallel_for_each()` out as well,
as its return type can be deduced from the type of Iterator and Func.

Signed-off-by: Kefu Chai <kchai@redhat.com>

diff --git a/src/crimson/common/errorator-loop.h b/src/crimson/common/errorator-loop.h
new file mode 100644 (file)
index 0000000..d2da477
--- /dev/null
+++ b/src/crimson/common/errorator-loop.h
@@ -0,0 +1,60 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:nil -*-
+// vim: ts=8 sw=2 smarttab expandtab
+
+#pragma once
+
+#include <seastar/core/future.hh>
+
+#include "crimson/common/errorator.h"
+
+
+namespace crimson {
+template <class... AllowedErrors>
+class parallel_for_each_state final : private seastar::continuation_base<> {
+  using future_t = typename errorator<AllowedErrors...>::template future<>;
+  std::vector<future_t> _incomplete;
+  seastar::promise<> _result;
+  std::exception_ptr _ex;
+private:
+  void wait_for_one() noexcept {
+    while (!_incomplete.empty() && _incomplete.back().available()) {
+      if (_incomplete.back().failed()) {
+        _ex = _incomplete.back().get_exception();
+      }
+      _incomplete.pop_back();
+    }
+    if (!_incomplete.empty()) {
+      seastar::internal::set_callback(_incomplete.back(), static_cast<continuation_base<>*>(this));
+      _incomplete.pop_back();
+      return;
+    }
+    if (__builtin_expect(bool(_ex), false)) {
+      _result.set_exception(std::move(_ex));
+    } else {
+      _result.set_value();
+    }
+    delete this;
+  }
+  virtual void run_and_dispose() noexcept override {
+    if (_state.failed()) {
+      _ex = std::move(_state).get_exception();
+    }
+    _state = {};
+    wait_for_one();
+  }
+  task* waiting_task() noexcept override { return _result.waiting_task(); }
+public:
+  parallel_for_each_state(size_t n) {
+    _incomplete.reserve(n);
+  }
+  void add_future(future_t&& f) {
+    _incomplete.push_back(std::move(f));
+  }
+  future_t get_future() {
+    auto ret = _result.get_future();
+    wait_for_one();
+    return ret;
+  }
+};
+
+} // namespace crimson

diff --git a/src/crimson/common/errorator.h b/src/crimson/common/errorator.h
index 41183ead19154913962bfe4914903e612ce17989..258124b50945579fdd867f472c876d25dd65a643 100644 (file)
--- a/src/crimson/common/errorator.h
+++ b/src/crimson/common/errorator.h
@@ -321,6 +321,9 @@ static constexpr auto composer(FuncHead&& head, FuncTail&&... tail) {
 template <class ValueT>
 struct errorated_future_marker{};
 
+template <class... AllowedErrors>
+class parallel_for_each_state;
+
 template <class T>
 static inline constexpr bool is_error_v = std::is_base_of_v<error_t<T>, T>;
 
@@ -713,7 +716,7 @@ private:
     friend inline auto ::seastar::internal::do_with_impl(T1&& rv1, T2&& rv2, T3_or_F&& rv3, More&&... more);
     template<typename, typename>
     friend class ::crimson::interruptible::interruptible_future_detail;
-    friend class errorator<AllowedErrors...>::parallel_for_each_state;
+    friend class ::crimson::parallel_for_each_state<AllowedErrors...>;
   };
 
   class Enabler {};
@@ -863,7 +866,7 @@ public:
   template <typename Iterator, typename Func>
   static inline errorator<AllowedErrors...>::future<>
   parallel_for_each(Iterator first, Iterator last, Func&& func) noexcept {
-    parallel_for_each_state* s = nullptr;
+    parallel_for_each_state<AllowedErrors...>* s = nullptr;
     // Process all elements, giving each future the following treatment:
     //   - available, not failed: do nothing
     //   - available, failed: collect exception in ex
@@ -875,7 +878,7 @@ public:
           using itraits = std::iterator_traits<Iterator>;
           auto n = (seastar::internal::iterator_range_estimate_vector_capacity(
                 first, last, typename itraits::iterator_category()) + 1);
-          s = new parallel_for_each_state(n);
+          s = new parallel_for_each_state<AllowedErrors...>(n);
         }
         s->add_future(std::move(f));
       }
@@ -1046,53 +1049,6 @@ private:
     return e.error_t<std::decay_t<ErrorT>>::to_exception_ptr();
   }
 
-  class parallel_for_each_state final : private seastar::continuation_base<> {
-    using future_t = errorator<AllowedErrors...>::future<>;
-    std::vector<future_t> _incomplete;
-    seastar::promise<> _result;
-    std::exception_ptr _ex;
-  private:
-    void wait_for_one() noexcept {
-      while (!_incomplete.empty() && _incomplete.back().available()) {
-        if (_incomplete.back().failed()) {
-          _ex = _incomplete.back().get_exception();
-        }
-        _incomplete.pop_back();
-      }
-      if (!_incomplete.empty()) {
-        seastar::internal::set_callback(_incomplete.back(), static_cast<continuation_base<>*>(this));
-        _incomplete.pop_back();
-        return;
-      }
-      if (__builtin_expect(bool(_ex), false)) {
-        _result.set_exception(std::move(_ex));
-      } else {
-        _result.set_value();
-      }
-      delete this;
-    }
-    virtual void run_and_dispose() noexcept override {
-      if (_state.failed()) {
-        _ex = std::move(_state).get_exception();
-      }
-      _state = {};
-      wait_for_one();
-    }
-    task* waiting_task() noexcept override { return _result.waiting_task(); }
-  public:
-    parallel_for_each_state(size_t n) {
-      _incomplete.reserve(n);
-    }
-    void add_future(future_t&& f) {
-      _incomplete.push_back(std::move(f));
-    }
-    future_t get_future() {
-      auto ret = _result.get_future();
-      wait_for_one();
-      return ret;
-    }
-  };
-
   // needed because of:
   //  * return_errorator_t::template futurize<...> in `safe_then()`,
   //  * conversion to `std::exception_ptr` in `future::future(ErrorT&&)`.

diff --git a/src/test/crimson/test_errorator.cc b/src/test/crimson/test_errorator.cc
index 8e4f09795fc818e3588a5d86dc991e64e430fc95..8d6cf9a00c3048b6339a1720771a6c7268017141 100644 (file)
--- a/src/test/crimson/test_errorator.cc
+++ b/src/test/crimson/test_errorator.cc
@@ -7,6 +7,7 @@
 #include "test/crimson/gtest_seastar.h"
 
 #include "crimson/common/errorator.h"
+#include "crimson/common/errorator-loop.h"
 #include "crimson/common/log.h"
 #include "seastar/core/sleep.hh"