crimson: futures from flush_changes_n_do_ops_effects must not fail

The return signature previously suggested that the second future
returned could be an error.  This seemed necessary due to how
effects are handled:

template <typename MutFunc>
OpsExecuter::rep_op_fut_t
OpsExecuter::flush_changes_n_do_ops_effects(
  const std::vector<OSDOp>& ops,
  SnapMapper& snap_mapper,
  OSDriver& osdriver,
  MutFunc mut_func) &&
{
...
    all_completed =
      std::move(all_completed).then_interruptible([this, pg=this->pg] {
      // let's do the cleaning of `op_effects` in destructor
      return interruptor::do_for_each(op_effects,
        [pg=std::move(pg)](auto& op_effect) {
        return op_effect->execute(pg);
      });

However, all of the actual execute implementations (created via
OpsExecuter::with_effect_on_obc) return a bare seastar::future and
cannot fail.

In a larger sense, it's actually critical that neither future returned
from flush_changes_n_do_ops_effects may fail -- they represent applying
the transaction locally and remotely.  If either portion fails, there
would need to be an interval change to recover.

Signed-off-by: Samuel Just <sjust@redhat.com>

diff --git a/src/crimson/osd/ops_executer.h b/src/crimson/osd/ops_executer.h
index 0b61f80b9983b283ba33fbc9797dc3e723f61c56..185ead24e7550d4df69e032b76b7088a5e111ded 100644 (file)
--- a/src/crimson/osd/ops_executer.h
+++ b/src/crimson/osd/ops_executer.h
@@ -179,7 +179,7 @@ private:
   // should be used.
   struct effect_t {
     // an effect can affect PG, i.e. create a watch timeout
-    virtual osd_op_errorator::future<> execute(Ref<PG> pg) = 0;
+    virtual seastar::future<> execute(Ref<PG> pg) = 0;
     virtual ~effect_t() = default;
   };
 
@@ -400,7 +400,7 @@ public:
   execute_op(OSDOp& osd_op);
 
   using rep_op_fut_tuple =
-    std::tuple<interruptible_future<>, osd_op_ierrorator::future<>>;
+    std::tuple<interruptible_future<>, interruptible_future<>>;
   using rep_op_fut_t =
     interruptible_future<rep_op_fut_tuple>;
   template <typename MutFunc>
@@ -475,7 +475,7 @@ auto OpsExecuter::with_effect_on_obc(
          effect_func(std::move(effect_func)),
          obc(std::move(obc)) {
     }
-    osd_op_errorator::future<> execute(Ref<PG> pg) final {
+    seastar::future<> execute(Ref<PG> pg) final {
       return std::move(effect_func)(std::move(ctx),
                                     std::move(obc),
                                     std::move(pg));
@@ -502,8 +502,7 @@ OpsExecuter::flush_changes_n_do_ops_effects(
   assert(obc);
 
   auto submitted = interruptor::now();
-  auto all_completed =
-    interruptor::make_interruptible(osd_op_errorator::now());
+  auto all_completed = interruptor::now();
 
   if (cloning_ctx) {
     ceph_assert(want_mutate);
@@ -536,7 +535,7 @@ OpsExecuter::flush_changes_n_do_ops_effects(
     // need extra ref pg due to apply_stats() which can be executed after
     // informing snap mapper
     all_completed =
-      std::move(all_completed).safe_then_interruptible([this, pg=this->pg] {
+      std::move(all_completed).then_interruptible([this, pg=this->pg] {
       // let's do the cleaning of `op_effects` in destructor
       return interruptor::do_for_each(op_effects,
         [pg=std::move(pg)](auto& op_effect) {

diff --git a/src/crimson/osd/pg.cc b/src/crimson/osd/pg.cc
index 97d48c1fa454c1d93438c0f7de83ff9a21a77369..8ab4e4e899b8e84050cdf3953e423f734573a2be 100644 (file)
--- a/src/crimson/osd/pg.cc
+++ b/src/crimson/osd/pg.cc
@@ -999,6 +999,28 @@ PG::do_osd_ops_execute(
       ceph_osd_op_name(osd_op.op.op));
     return ox->execute_op(osd_op);
   }).safe_then_interruptible([this, ox, &ops] {
+    /* flush_changes_n_do_ops_effects now returns
+     *
+     * interruptible_future<
+     *    tuple<interruptible_future<>, interruptible_future<>>>
+     *
+     * Previously, this lambda relied on the second element of that tuple to
+     * include OpsExecutor::osd_op_errorator in order to propogate the
+     * following three errors to the next callback.  This is actually quite
+     * awkward as the second future is the completion future, which really
+     * cannot fail (for it to do so would require an interval change to
+     * correct).
+     *
+     * Rather than reworking this now, I'll leave it as is and refactor it
+     * later.
+     */
+    using complete_iertr = crimson::interruptible::interruptible_errorator<
+      ::crimson::osd::IOInterruptCondition,
+      OpsExecuter::osd_op_errorator>;
+    using ret_t = std::tuple<
+      interruptible_future<>,
+      complete_iertr::future<>>;
+
     logger().debug(
       "do_osd_ops_execute: object {} all operations successful",
       ox->get_target());
@@ -1014,22 +1036,22 @@ PG::do_osd_ops_execute(
         // they tried, they failed.
         logger().info(" full, replying to FULL_TRY op");
         if (get_pgpool().info.has_flag(pg_pool_t::FLAG_FULL_QUOTA))
-          return interruptor::make_ready_future<OpsExecuter::rep_op_fut_tuple>(
-            seastar::now(),
-            OpsExecuter::osd_op_ierrorator::future<>(
-              crimson::ct_error::edquot::make()));
+         return interruptor::make_ready_future<ret_t>(
+            interruptor::now(),
+           complete_iertr::future<>(
+             crimson::ct_error::edquot::make()));
         else
-          return interruptor::make_ready_future<OpsExecuter::rep_op_fut_tuple>(
-            seastar::now(),
-            OpsExecuter::osd_op_ierrorator::future<>(
-              crimson::ct_error::enospc::make()));
+         return interruptor::make_ready_future<ret_t>(
+            interruptor::now(),
+           complete_iertr::future<>(
+             crimson::ct_error::enospc::make()));
       } else {
         // drop request
         logger().info(" full, dropping request (bad client)");
-        return interruptor::make_ready_future<OpsExecuter::rep_op_fut_tuple>(
-          seastar::now(),
-          OpsExecuter::osd_op_ierrorator::future<>(
-            crimson::ct_error::eagain::make()));
+       return interruptor::make_ready_future<ret_t>(
+         interruptor::now(),
+         complete_iertr::future<>(
+           crimson::ct_error::eagain::make()));
       }
     }
     return std::move(*ox).flush_changes_n_do_ops_effects(
@@ -1049,7 +1071,12 @@ PG::do_osd_ops_execute(
           std::move(txn),
           std::move(osd_op_p),
           std::move(log_entries));
-    });
+      }).then_interruptible([](auto &&futs) {
+       auto &&[submitted, completed] = std::move(futs);
+       return interruptor::make_ready_future<ret_t>(
+         std::move(submitted),
+         std::move(completed));
+      });
   }).safe_then_unpack_interruptible(
     [success_func=std::move(success_func), rollbacker, this, failure_func_ptr, obc]
     (auto submitted_fut, auto _all_completed_fut) mutable {