From: Samuel Just <sjust@redhat.com>
Date: Tue, 19 Jan 2021 03:46:30 +0000 (-0800)
Subject: crimson/common/operation: add OrderedConcurrentPhase
X-Git-Tag: v17.1.0~3095^2~3
X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=ad14ecedb60f535c639df10aa685f945abab87f6;p=ceph-ci.git

crimson/common/operation: add OrderedConcurrentPhase

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

diff --git a/src/crimson/common/operation.cc b/src/crimson/common/operation.cc
index 0a4c1a0cd0f..7f28ce31f07 100644
--- a/src/crimson/common/operation.cc
+++ b/src/crimson/common/operation.cc
@@ -66,4 +66,8 @@ void OrderedExclusivePhase::dump_detail(ceph::Formatter* f) const
 {
 }
 
+void OrderedConcurrentPhase::dump_detail(ceph::Formatter* f) const
+{
+}
+
 }
diff --git a/src/crimson/common/operation.h b/src/crimson/common/operation.h
index d81ff0075e6..07806768a93 100644
--- a/src/crimson/common/operation.h
+++ b/src/crimson/common/operation.h
@@ -414,4 +414,66 @@ private:
   seastar::shared_mutex mutex;
 };
 
+/**
+ * Permits multiple ops to inhabit the stage concurrently, but ensures that
+ * they will proceed to the next stage in the order in which they called
+ * enter.
+ */
+class OrderedConcurrentPhase : public PipelineStageI {
+  void dump_detail(ceph::Formatter *f) const final;
+  const char *get_type_name() const final {
+    return name;
+  }
+
+  class ExitBarrier : public PipelineExitBarrierI {
+    OrderedConcurrentPhase *phase;
+    std::optional<seastar::future<>> barrier;
+  public:
+    ExitBarrier(
+      OrderedConcurrentPhase *phase,
+      seastar::future<> &&barrier) : phase(phase), barrier(std::move(barrier)) {}
+
+    seastar::future<> wait() final {
+      assert(phase);
+      assert(barrier);
+      auto ret = std::move(*barrier);
+      barrier = std::nullopt;
+      return ret;
+    }
+
+    void exit() final {
+      if (barrier) {
+	static_cast<void>(
+	  std::move(*barrier).then([phase=this->phase] { phase->mutex.unlock(); }));
+	barrier = std::nullopt;
+	phase = nullptr;
+      }
+      if (phase) {
+	phase->mutex.unlock();
+	phase = nullptr;
+      }
+    }
+
+    void cancel() final {
+      exit();
+    }
+
+    ~ExitBarrier() final {
+      cancel();
+    }
+  };
+
+public:
+  seastar::future<PipelineExitBarrierI::Ref> enter() final {
+    return seastar::make_ready_future<PipelineExitBarrierI::Ref>(
+      new ExitBarrier{this, mutex.lock()});
+  }
+
+  OrderedConcurrentPhase(const char *name) : name(name) {}
+
+private:
+  const char * name;
+  seastar::shared_mutex mutex;
+};
+
 }