crimson: OpsExecuter::submit_changes() operates on errorated future.

Signed-off-by: Radoslaw Zarzynski <rzarzyns@redhat.com>

diff --git a/src/crimson/osd/ops_executer.h b/src/crimson/osd/ops_executer.h
index 1ce9c29576f8eb553e7f8e786f5cf0dbed2781e0..2606bb8190817f7443095f71c538ddb4696c670f 100644 (file)
--- a/src/crimson/osd/ops_executer.h
+++ b/src/crimson/osd/ops_executer.h
@@ -37,6 +37,24 @@ class OSDOp;
 
 namespace crimson::osd {
 class OpsExecuter {
+  using call_errorator = crimson::errorator<
+    crimson::stateful_ec,
+    crimson::ct_error::enoent,
+    crimson::ct_error::invarg,
+    crimson::ct_error::permission_denied,
+    crimson::ct_error::operation_not_supported,
+    crimson::ct_error::input_output_error>;
+  using read_errorator = PGBackend::read_errorator;
+  using get_attr_errorator = PGBackend::get_attr_errorator;
+
+public:
+  using osd_op_errorator = crimson::compound_errorator_t<
+    call_errorator,
+    read_errorator,
+    get_attr_errorator,
+    PGBackend::stat_errorator>;
+
+private:
   // an operation can be divided into two stages: main and effect-exposing
   // one. The former is performed immediately on call to `do_osd_op()` while
   // the later on `submit_changes()` – after successfully processing main
@@ -44,7 +62,7 @@ class OpsExecuter {
   // scheduled effect-exposing stages will be executed.
   // when operation requires this division, `with_effect()` should be used.
   struct effect_t {
-    virtual seastar::future<> execute() = 0;
+    virtual osd_op_errorator::future<> execute() = 0;
     virtual ~effect_t() = default;
   };
 
@@ -65,13 +83,6 @@ class OpsExecuter {
   template <class Context, class MainFunc, class EffectFunc>
   auto with_effect(Context&& ctx, MainFunc&& main_func, EffectFunc&& effect_func);
 
-  using call_errorator = crimson::errorator<
-    crimson::stateful_ec,
-    crimson::ct_error::enoent,
-    crimson::ct_error::invarg,
-    crimson::ct_error::permission_denied,
-    crimson::ct_error::operation_not_supported,
-    crimson::ct_error::input_output_error>;
   call_errorator::future<> do_op_call(class OSDOp& osd_op);
 
   template <class Func>
@@ -104,9 +115,6 @@ class OpsExecuter {
     return crimson::ct_error::operation_not_supported::make();
   }
 
-  using read_errorator = PGBackend::read_errorator;
-  using get_attr_errorator = PGBackend::get_attr_errorator;
-
 public:
   OpsExecuter(PGBackend::cached_os_t os, PG& pg, Ref<MOSDOp> msg)
     : os(std::move(os)),
@@ -118,16 +126,11 @@ public:
     : OpsExecuter{PGBackend::cached_os_t{}, pg, std::move(msg)}
   {}
 
-  using osd_op_errorator = crimson::compound_errorator_t<
-    call_errorator,
-    read_errorator,
-    get_attr_errorator,
-    PGBackend::stat_errorator>;
   osd_op_errorator::future<> execute_osd_op(class OSDOp& osd_op);
   seastar::future<> execute_pg_op(class OSDOp& osd_op);
 
   template <typename Func>
-  seastar::future<> submit_changes(Func&& f) &&;
+  osd_op_errorator::future<> submit_changes(Func&& f) &&;
 
   const auto& get_message() const {
     return *msg;
@@ -152,7 +155,7 @@ auto OpsExecuter::with_effect(
 
     task_t(Context&& ctx, EffectFunc&& effect_func)
        : ctx(std::move(ctx)), effect_func(std::move(effect_func)) {}
-    seastar::future<> execute() final {
+    osd_op_errorator::future<> execute() final {
       return std::move(effect_func)(std::move(ctx));
     }
   };
@@ -164,18 +167,16 @@ auto OpsExecuter::with_effect(
 }
 
 template <typename Func>
-seastar::future<> OpsExecuter::submit_changes(Func&& f) && {
-  return std::forward<Func>(f)(std::move(txn), std::move(os)).then(
-    // NOTE: this lambda could be scheduled conditionally (if_then?)
-    [this] {
-      return seastar::do_until(
-        [this] { return op_effects.empty(); },
-        [this] {
-          auto fut = op_effects.front()->execute();
-          op_effects.pop_front();
-          return fut;
-        });
+OpsExecuter::osd_op_errorator::future<> OpsExecuter::submit_changes(Func&& f) && {
+  if (__builtin_expect(op_effects.empty(), true)) {
+    return std::forward<Func>(f)(std::move(txn), std::move(os));
+  }
+  return std::forward<Func>(f)(std::move(txn), std::move(os)).safe_then([this] {
+    // let's do the cleaning of `op_effects` in destructor
+    return crimson::do_for_each(op_effects, [] (auto& op_effect) {
+      return op_effect->execute();
     });
+  });
 }
 
 } // namespace crimson::osd

diff --git a/src/crimson/osd/pg.cc b/src/crimson/osd/pg.cc
index b14bb103c594eacc859ca961d940e3ca63ae081d..f270d2669848747f9d13808f0e9cb6ce191c71e6 100644 (file)
--- a/src/crimson/osd/pg.cc
+++ b/src/crimson/osd/pg.cc
@@ -435,6 +435,7 @@ seastar::future<> PG::submit_transaction(boost::local_shared_ptr<ObjectState>&&
 
 seastar::future<Ref<MOSDOpReply>> PG::do_osd_ops(Ref<MOSDOp> m)
 {
+  using osd_op_errorator = OpsExecuter::osd_op_errorator;
   const auto oid = m->get_snapid() == CEPH_SNAPDIR ? m->get_hobj().get_head()
                                                    : m->get_hobj();
   return backend->get_object_state(oid).safe_then([this, m](auto os) mutable {
@@ -445,11 +446,11 @@ seastar::future<Ref<MOSDOpReply>> PG::do_osd_ops(Ref<MOSDOp> m)
         return ox.execute_osd_op(osd_op);
       }).safe_then([this, m, &ox] {
         logger().debug("all operations have been executed successfully");
-        return std::move(ox).submit_changes([this, m] (auto&& txn, auto&& os) {
+        return std::move(ox).submit_changes([this, m] (auto&& txn, auto&& os) -> osd_op_errorator::future<> {
           // XXX: the entire lambda could be scheduled conditionally. ::if_then()?
          if (txn.empty()) {
             logger().debug("txn is empty, bypassing mutate");
-           return seastar::now();
+           return osd_op_errorator::now();
          } else {
            return submit_transaction(std::move(os), std::move(txn), *m);
          }