1 files changed, 142 insertions, 0 deletions
diff --git a/src/crimson/osd/osd_operations/peering_event.h b/src/crimson/osd/osd_operations/peering_event.h
new file mode 100644
index 000000000..3a6c0678c
--- /dev/null
+++ b/src/crimson/osd/osd_operations/peering_event.h
@@ -0,0 +1,142 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#pragma once
+
+#include <iostream>
+#include <seastar/core/future.hh>
+
+#include "crimson/osd/osd_operation.h"
+#include "osd/osd_types.h"
+#include "osd/PGPeeringEvent.h"
+#include "osd/PeeringState.h"
+
+namespace ceph {
+  class Formatter;
+}
+
+namespace crimson::osd {
+
+class OSD;
+class ShardServices;
+class PG;
+
+class PeeringEvent : public OperationT<PeeringEvent> {
+public:
+  static constexpr OperationTypeCode type = OperationTypeCode::peering_event;
+
+  class PGPipeline {
+    OrderedPipelinePhase await_map = {
+      "PeeringEvent::PGPipeline::await_map"
+    };
+    OrderedPipelinePhase process = {
+      "PeeringEvent::PGPipeline::process"
+    };
+    friend class PeeringEvent;
+    friend class PGAdvanceMap;
+  };
+
+protected:
+  OrderedPipelinePhase::Handle handle;
+  PGPipeline &pp(PG &pg);
+
+  ShardServices &shard_services;
+  PeeringCtx ctx;
+  pg_shard_t from;
+  spg_t pgid;
+  float delay = 0;
+  PGPeeringEvent evt;
+
+  const pg_shard_t get_from() const {
+    return from;
+  }
+
+  const spg_t get_pgid() const {
+    return pgid;
+  }
+
+  const PGPeeringEvent &get_event() const {
+    return evt;
+  }
+
+  virtual void on_pg_absent();
+  virtual seastar::future<> complete_rctx(Ref<PG>);
+  virtual seastar::future<Ref<PG>> get_pg() = 0;
+
+public:
+  template <typename... Args>
+  PeeringEvent(
+    ShardServices &shard_services, const pg_shard_t &from, const spg_t &pgid,
+    Args&&... args) :
+    shard_services(shard_services),
+    ctx{ceph_release_t::octopus},
+    from(from),
+    pgid(pgid),
+    evt(std::forward<Args>(args)...)
+  {}
+  template <typename... Args>
+  PeeringEvent(
+    ShardServices &shard_services, const pg_shard_t &from, const spg_t &pgid,
+    float delay, Args&&... args) :
+    shard_services(shard_services),
+    ctx{ceph_release_t::octopus},
+    from(from),
+    pgid(pgid),
+    delay(delay),
+    evt(std::forward<Args>(args)...)
+  {}
+
+  void print(std::ostream &) const final;
+  void dump_detail(ceph::Formatter* f) const final;
+  seastar::future<> start();
+};
+
+class RemotePeeringEvent : public PeeringEvent {
+protected:
+  OSD &osd;
+  crimson::net::ConnectionRef conn;
+
+  void on_pg_absent() final;
+  seastar::future<> complete_rctx(Ref<PG> pg) override;
+  seastar::future<Ref<PG>> get_pg() final;
+
+public:
+  class ConnectionPipeline {
+    OrderedPipelinePhase await_map = {
+      "PeeringRequest::ConnectionPipeline::await_map"
+    };
+    OrderedPipelinePhase get_pg = {
+      "PeeringRequest::ConnectionPipeline::get_pg"
+    };
+    friend class RemotePeeringEvent;
+  };
+
+  template <typename... Args>
+  RemotePeeringEvent(OSD &osd, crimson::net::ConnectionRef conn, Args&&... args) :
+    PeeringEvent(std::forward<Args>(args)...),
+    osd(osd),
+    conn(conn)
+  {}
+
+private:
+  ConnectionPipeline &cp();
+};
+
+class LocalPeeringEvent final : public PeeringEvent {
+protected:
+  seastar::future<Ref<PG>> get_pg() final;
+
+  Ref<PG> pg;
+
+public:
+  template <typename... Args>
+  LocalPeeringEvent(Ref<PG> pg, Args&&... args) :
+    PeeringEvent(std::forward<Args>(args)...),
+    pg(pg)
+  {}
+
+  virtual ~LocalPeeringEvent();
+};
+
+
+}