crimson/common/operation: release pipeline stages on the core they are on

Otherwise, tasks waiting on the stage will wake up on the wrong core.
Later, we may choose to statically enable this behavior only for stages
that can actually span cores, but this is ok for now.

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

diff --git a/src/crimson/common/operation.h b/src/crimson/common/operation.h
index 6a6a48a2fd919e3b4fdf0c53d90e4baa76d2a066..93dbe9ee5b94f71da7628f768dd6afd40dc086de 100644 (file)
--- a/src/crimson/common/operation.h
+++ b/src/crimson/common/operation.h
@@ -20,6 +20,7 @@
 #include "include/utime.h"
 #include "common/Clock.h"
 #include "crimson/common/interruptible_future.h"
+#include "crimson/common/smp_helpers.h"
 #include "crimson/common/log.h"
 
 namespace ceph {
@@ -471,7 +472,10 @@ public:
 
 template <class T>
 class PipelineStageIT : public BlockerT<T> {
+  const core_id_t core = seastar::this_shard_id();
 public:
+  core_id_t get_core() const { return core; }
+
   template <class... Args>
   decltype(auto) enter(Args&&... args) {
     return static_cast<T*>(this)->enter(std::forward<Args>(args)...);
@@ -502,6 +506,7 @@ public:
   template <typename OpT, typename T>
   seastar::future<>
   enter(T &stage, typename T::BlockingEvent::template Trigger<OpT>&& t) {
+    ceph_assert(stage.get_core() == seastar::this_shard_id());
     return wait_barrier().then([this, &stage, t=std::move(t)] () mutable {
       auto fut = t.maybe_record_blocking(stage.enter(t), stage);
       exit();
@@ -554,8 +559,13 @@ class OrderedExclusivePhaseT : public PipelineStageIT<T> {
 
     void exit() final {
       if (phase) {
-       phase->exit();
+       auto *p = phase;
        phase = nullptr;
+       std::ignore = seastar::smp::submit_to(
+         p->get_core(),
+         [p] {
+           p->exit();
+         });
       }
     }
 
@@ -644,8 +654,12 @@ private:
        phase = nullptr;
       }
       if (phase) {
-       phase->mutex.unlock();
-       phase = nullptr;
+       std::ignore = seastar::smp::submit_to(
+         phase->get_core(),
+         [this] {
+           phase->mutex.unlock();
+           phase = nullptr;
+         });
       }
     }