osdc: Extend op_post_submit to cope with successful Ops and move SplitOp decision...

The locking situation in Objecter is complex. When ops are completed whether with success or otherwise, some locks are held. For split ops, this is particularly complex, since multiple sessions are involved in the completion.

To avoid all these deadlock issues, splitOps choose to schedule a completion task using asio::post, which can then take the appropriate locks before completing the IO, without risk of deadlock.

Usage of this will be added in a refactor of SplitOps.

In addition, previously split ops was being calculated immediately as soon as the op was submitted.  Here we move the submit down to below the throttling and timeout code.  This way we throttle/timeout the original op.

Handling the timeout (op_cancel) will be handled in a later commit.

As part of this commit we also introduce a SplitOp session. This allows us to keep track of the parent ops while the child ops have been submitted and redrive the correct op(s) when necessary.

Signed-off-by: Alex Ainscow <aainscow@uk.ibm.com>
Signed-off-by: Jon Bailey <jonathan.bailey1@ibm.com>

diff --git a/src/osdc/Objecter.cc b/src/osdc/Objecter.cc
index bf571c6ef5826db714c7460b32fc361864c3bced..a368beaa8051382e90ed3c752898fed6e1f15c3e 100644 (file)
--- a/src/osdc/Objecter.cc
+++ b/src/osdc/Objecter.cc
@@ -539,6 +539,18 @@ void Objecter::shutdown()
     cop->put();
   }
 
+  // Split ops can only contain standard ops...
+  while(!splitop_session->ops.empty()) {
+    auto i = splitop_session->ops.begin();
+    ldout(cct, 10) << " op " << i->first << dendl;
+    auto op = i->second;
+    {
+      std::unique_lock swl(splitop_session->lock);
+      _session_op_remove(splitop_session, op);
+    }
+    op->put();
+  }
+
   if (tick_event) {
     if (timer.cancel_event(tick_event)) {
       ldout(cct, 10) <<  " successfully canceled tick" << dendl;
@@ -2365,10 +2377,35 @@ void Objecter::resend_mon_ops()
 
 // read | write ---------------------------
 
+void Objecter::op_post_split_op_complete(Op* op, bs::error_code ec, int rc) {
+  ceph_assert(op->session == splitop_session);
+
+  op->get();  // Keep alive during async operation
+
+  boost::asio::post(service, [this, op, ec, rc]() {
+    shunique_lock rl(rwlock, ceph::acquire_shared);
+
+    bool freed = op->get_nref() == 1;
+    op->put();
+
+    if (freed || op->session != splitop_session) {
+      ceph_assert(!initialized); // Should only happen during shutdown.
+      return;
+    }
+
+    unique_lock sl(op->session->lock);
 
-void Objecter::op_post_submit(Op* op) {
-  boost::asio::post(service, [this, op]() {
-    op_submit(op);
+    if (rc != -EAGAIN) {
+      op->trace.event("post op complete");
+      // This removes from session and unlocks sl.
+      complete_op_reply(op, ec, op->session, sl, rc);
+    } else {
+      _session_op_remove(op->session, op);
+      sl.unlock();
+      op->split_op_tids.reset();
+      ceph_tid_t tid = 0;
+      _op_submit(op, rl, &tid);
+    }
   });
 }
 
@@ -2380,11 +2417,17 @@ void Objecter::op_submit(Op *op, ceph_tid_t *ptid, int *ctx_budget)
     ptid = &tid;
   op->trace.event("op submit");
 
-  bool was_split = SplitOp::create(op, *this, rl, ptid, ctx_budget, cct);
+  _op_submit_with_budget(op, rl, ptid, ctx_budget);
+}
 
-  if (!was_split) {
-    _op_submit_with_budget(op, rl, ptid, ctx_budget);
+void Objecter::add_op_to_splitop_session(Op *op) {
+  unique_lock sl(splitop_session->lock);
+  if (op->tid == 0) {
+    op->tid = ++last_tid;
   }
+  _session_op_assign(splitop_session, op);
+  inflight_ops++;
+  sl.unlock();
 }
 
 void Objecter::_op_submit_with_budget(Op *op,
@@ -2418,7 +2461,14 @@ void Objecter::_op_submit_with_budget(Op *op,
                                      op_cancel(tid, -ETIMEDOUT); });
   }
 
-  _op_submit(op, sul, ptid);
+
+  bool was_split = SplitOp::create(op, *this, sul, cct);
+
+  if (was_split) {
+    *ptid = op->tid;
+  } else {
+    _op_submit(op, sul, ptid);
+  }
 }
 
 void Objecter::_send_op_account(Op *op)
@@ -2706,17 +2756,32 @@ start:
     }
   }
 
-  // Handle case where the op is in homeless session
-  shared_lock sl(homeless_session->lock);
-  while (auto tid = next_subsystem_op(homeless_session->ops)) {
+  {
+    // Handle case where the op is in homeless session
+    shared_lock sl(homeless_session->lock);
+    while (auto tid = next_subsystem_op(homeless_session->ops)) {
+      sl.unlock();
+      auto ret = op_cancel(homeless_session, *tid, ceph::from_error_code(ec), ec);
+      if (ret == -ENOENT) {
+        /* oh no! raced, maybe tid moved to another session, restarting */
+        goto start;
+      }
+    }
     sl.unlock();
-    auto ret = op_cancel(homeless_session, *tid, ceph::from_error_code(ec), ec);
-    if (ret == -ENOENT) {
-      /* oh no! raced, maybe tid moved to another session, restarting */
-      goto start;
+  }
+  {
+    // Handle case where the op is in splitop session
+    shared_lock sl(splitop_session->lock);
+    while (auto tid = next_subsystem_op(splitop_session->ops)) {
+      sl.unlock();
+      auto ret = op_cancel(splitop_session, *tid, ceph::from_error_code(ec), ec);
+      if (ret == -ENOENT) {
+        /* oh no! raced, maybe tid moved to another session, restarting */
+        goto start;
+      }
     }
+    sl.unlock();
   }
-  sl.unlock();
 }
 
 int Objecter::op_cancel(ceph_tid_t tid, int r)
@@ -2766,19 +2831,36 @@ start:
   ldout(cct, 5) << __func__ << ": tid " << tid
                << " not found in live sessions" << dendl;
 
-  // Handle case where the op is in homeless session
-  shared_lock sl(homeless_session->lock);
-  if (homeless_session->ops.find(tid) != homeless_session->ops.end()) {
-    sl.unlock();
-    ret = op_cancel(homeless_session, tid, r, osdcode(r));
-    if (ret == -ENOENT) {
-      /* oh no! raced, maybe tid moved to another session, restarting */
-      goto start;
+  {
+    // Handle case where the op is in homeless session
+    shared_lock sl(homeless_session->lock);
+    if (homeless_session->ops.find(tid) != homeless_session->ops.end()) {
+      sl.unlock();
+      ret = op_cancel(homeless_session, tid, r, osdcode(r));
+      if (ret == -ENOENT) {
+        /* oh no! raced, maybe tid moved to another session, restarting */
+        goto start;
+      } else {
+        return ret;
+      }
     } else {
-      return ret;
+      sl.unlock();
+    }
+  }
+  {
+    shared_lock sl(splitop_session->lock);
+    if (splitop_session->ops.find(tid) != splitop_session->ops.end()) {
+      sl.unlock();
+      ret = op_cancel(splitop_session, tid, r, osdcode(r));
+      if (ret == -ENOENT) {
+        /* oh no! raced, maybe tid moved to another session, restarting */
+        goto start;
+      } else {
+        return ret;
+      }
+    } else {
+      sl.unlock();
     }
-  } else {
-    sl.unlock();
   }
 
   ldout(cct, 5) << __func__ << ": tid " << tid
@@ -4868,6 +4950,7 @@ void Objecter::_dump_active()
     sl.unlock();
   }
   _dump_active(homeless_session);
+  _dump_active(splitop_session);
 }
 
 void Objecter::dump_active()
@@ -4927,6 +5010,7 @@ void Objecter::dump_ops(Formatter *fmt)
     sl.unlock();
   }
   _dump_ops(homeless_session, fmt);
+  _dump_ops(splitop_session, fmt);
   fmt->close_section(); // ops array
 }
 
@@ -4955,6 +5039,7 @@ void Objecter::dump_linger_ops(Formatter *fmt)
     sl.unlock();
   }
   _dump_linger_ops(homeless_session, fmt);
+  // No linger ops in splitop_session
   fmt->close_section(); // linger_ops array
 }
 
@@ -4989,6 +5074,7 @@ void Objecter::dump_command_ops(Formatter *fmt)
     sl.unlock();
   }
   _dump_command_ops(homeless_session, fmt);
+  // No command_ops for splitops session.
   fmt->close_section(); // command_ops array
 }
 
@@ -5364,8 +5450,10 @@ Objecter::Objecter(CephContext *cct,
 Objecter::~Objecter()
 {
   ceph_assert(homeless_session->get_nref() == 1);
+  ceph_assert(splitop_session->get_nref() == 1);
   ceph_assert(num_homeless_ops == 0);
   homeless_session->put();
+  splitop_session->put();
 
   ceph_assert(osd_sessions.empty());
   ceph_assert(poolstat_ops.empty());

diff --git a/src/osdc/Objecter.h b/src/osdc/Objecter.h
index 2620c3cbc4852067dfb59ab62d91d5c52fe77d65..def11791a738dfbd65802e74338b4af6a1f0ef48 100644 (file)
--- a/src/osdc/Objecter.h
+++ b/src/osdc/Objecter.h
@@ -2552,6 +2552,7 @@ public:
   std::atomic<unsigned> num_homeless_ops{0};
 
   OSDSession* homeless_session = new OSDSession(cct, -1);
+  OSDSession* splitop_session = new OSDSession(cct, -2); // -2 to differentiate from homeless
 
 
   // ops waiting for an osdmap with a new pool or confirmation that
@@ -2841,13 +2842,14 @@ private:
   // low-level
   void _op_submit(Op *op, ceph::shunique_lock<ceph::shared_mutex>& lc,
                  ceph_tid_t *ptid);
+  void add_op_to_splitop_session(Op *op);
   void _op_submit_with_budget(Op *op,
                              ceph::shunique_lock<ceph::shared_mutex>& lc,
                              ceph_tid_t *ptid,
                              int *ctx_budget = NULL);
   // public interface
 public:
-  void op_post_submit(Op *op);
+  void op_post_split_op_complete(Op* op, boost::system::error_code ec, int rc);
   void op_submit(Op *op, ceph_tid_t *ptid = NULL, int *ctx_budget = NULL);
   bool is_active() {
     std::shared_lock l(rwlock);