Merge pull request ceph#58986 from NitzanMordhai/wip-nitzan-crimson-op_gate-alianstore-multi-core

crimson: use gate per shard for AlienStore and OSD

Reviewed-by: Matan Breizman <[email protected]>
Reviewed-by: Yingxin Cheng <[email protected]>

Author: Matan-B

src/crimson/common/gated.h

@@ -6,6 +6,8 @@
 #include <seastar/core/gate.hh>
 #include <seastar/core/future.hh>
 #include <seastar/core/future-util.hh>
+#include <type_traits>
+#include <vector>
 
 #include "crimson/common/exception.h"
 #include "crimson/common/log.h"
@@ -15,15 +17,27 @@ namespace crimson::common {
 
 class Gated {
  public:
+  Gated() : sid(seastar::this_shard_id()) {}
+  Gated(const seastar::shard_id sid) : sid(sid) {}
+  Gated(const Gated&) = delete;
+  Gated& operator=(const Gated&) = delete;
+  Gated(Gated&&) = default;
+  Gated& operator=(Gated&&) = delete;
+  virtual ~Gated() = default;
+  
   static seastar::logger& gated_logger() {
     return crimson::get_logger(ceph_subsys_osd);
   }
+
   template <typename Func, typename T>
   inline void dispatch_in_background(const char* what, T& who, Func&& func) {
-    (void) dispatch(what, who, func);
+    //ceph_assert(seastar::this_shard_id() == sid);
+    (void) dispatch(what, who, std::forward<Func>(func));
   }
+
   template <typename Func, typename T>
   inline seastar::future<> dispatch(const char* what, T& who, Func&& func) {
+    //ceph_assert(seastar::this_shard_id() == sid);
     return seastar::with_gate(pending_dispatch, std::forward<Func>(func)
     ).handle_exception([what, &who] (std::exception_ptr eptr) {
       if (*eptr.__cxa_exception_type() == typeid(system_shutdown_exception)) {
@@ -42,14 +56,81 @@ class Gated {
     });
   }
 
+  template <typename Func>
+  auto simple_dispatch(const char* what, Func&& func) {
+    //ceph_assert(seastar::this_shard_id() == sid);
+    return seastar::with_gate(pending_dispatch, std::forward<Func>(func));
+  }
+
   seastar::future<> close() {
+    ceph_assert(seastar::this_shard_id() == sid);
     return pending_dispatch.close();
   }
+
   bool is_closed() const {
     return pending_dispatch.is_closed();
   }
+
+  seastar::shard_id get_shard_id() const {
+    return sid;
+  }
  private:
   seastar::gate pending_dispatch;
+  const seastar::shard_id sid;
+};
+
+// gate_per_shard is a class that provides a gate for each shard.
+// It was introduced to provide a way to have gate for each shard
+// in a seastar application since gates are not supposed to be shared
+// across shards. ( https://tracker.ceph.com/issues/64332 )
+class gate_per_shard {
+ public:
+  gate_per_shard() : gates(seastar::smp::count) {
+    std::vector<seastar::future<>> futures;
+    for (unsigned shard = 0; shard < seastar::smp::count; ++shard) {
+      futures.push_back(seastar::smp::submit_to(shard, [this, shard] {
+        gates[shard] = std::make_unique<Gated>();
+      }));
+    }
+    seastar::when_all_succeed(futures.begin(), futures.end()).get();
+  }
+  //explicit gate_per_shard(size_t shard_count) : gates(shard_count) {}
+  gate_per_shard(const gate_per_shard&) = delete;
+  gate_per_shard& operator=(const gate_per_shard&) = delete;
+  gate_per_shard(gate_per_shard&&) = default;
+  gate_per_shard& operator=(gate_per_shard&&) = default;
+  ~gate_per_shard() = default;
+
+  template <typename Func, typename T>
+  inline void dispatch_in_background(const char* what, T& who, Func&& func) {
+    (void) dispatch(what, who, std::forward<Func>(func));
+  }
+
+  template <typename Func, typename T>
+  inline auto dispatch(const char* what, T& who, Func&& func) {
+    return gates[seastar::this_shard_id()]->dispatch(what, who, std::forward<Func>(func));
+  }
+
+  template <typename Func>
+  auto simple_dispatch(const char* what, Func&& func) {
+    return gates[seastar::this_shard_id()]->simple_dispatch(what, std::forward<Func>(func));
+  }
+
+  bool is_closed() const {
+    return gates[seastar::this_shard_id()]->is_closed();
+  }
+
+  seastar::future<> close_all() {
+    ceph_assert(gates.size() == seastar::smp::count);
+    return seastar::parallel_for_each(gates.begin(), gates.end(), [] (std::unique_ptr<Gated>& gate_ptr) {
+      return seastar::smp::submit_to(gate_ptr->get_shard_id(), [gate = gate_ptr.get()] {
+        return gate->close();
+      });
+    });
+  }
+
+ private:
+  std::vector<std::unique_ptr<Gated>> gates;
 };
 
-}// namespace crimson::common
+} // namespace crimson::common

src/crimson/net/io_handler.h

@@ -255,7 +255,7 @@ class IOHandler final : public ConnectionHandler {
   class shard_states_t {
   public:
     shard_states_t(seastar::shard_id _sid, io_state_t state)
-      : sid{_sid}, io_state{state} {}
+      : sid{_sid}, io_state{state}, gate{_sid} {}
 
     seastar::shard_id get_shard_id() const {
       return sid;

src/crimson/os/alienstore/alien_store.cc

@@ -75,7 +75,8 @@ AlienStore::AlienStore(const std::string& type,
                        const ConfigValues& values)
   : type(type),
     path{path},
-    values(values)
+    values(values),
+    op_gates()
 {
 }
 
@@ -142,12 +143,12 @@ AlienStore::exists(
   CollectionRef ch,
   const ghobject_t& oid)
 {
-  return seastar::with_gate(op_gate, [=, this] {
-    return tp->submit(ch->get_cid().hash_to_shard(tp->size()), [=, this] {
-      auto c = static_cast<AlienCollection*>(ch.get());
-      return store->exists(c->collection, oid);
+    return op_gates.simple_dispatch("exists", [=, this] {
+        return tp->submit(ch->get_cid().hash_to_shard(tp->size()), [=, this] {
+            auto c = static_cast<AlienCollection*>(ch.get());
+            return store->exists(c->collection, oid);
+        });
     });
-  });
 }
 
 AlienStore::mount_ertr::future<> AlienStore::mount()
@@ -173,7 +174,7 @@ seastar::future<> AlienStore::umount()
     // not really started yet
     return seastar::now();
   }
-  return op_gate.close().then([this] {
+  return op_gates.close_all().then([this] {
     return tp->submit([this] {
       {
 	std::lock_guard l(coll_map_lock);
@@ -183,10 +184,10 @@ seastar::future<> AlienStore::umount()
 	coll_map.clear();
       }
       return store->umount();
+    }).then([] (int r) {
+      assert(r == 0);
+      return seastar::now();
     });
-  }).then([] (int r) {
-    assert(r == 0);
-    return seastar::now();
   });
 }
 
@@ -477,7 +478,7 @@ seastar::future<> AlienStore::inject_data_error(const ghobject_t& o)
 {
   logger().debug("{}", __func__);
   assert(tp);
-  return seastar::with_gate(op_gate, [=, this] {
+  return op_gates.simple_dispatch("inject_data_error", [=, this] {
     return tp->submit([o, this] {
       return store->inject_data_error(o);
     });
@@ -488,8 +489,8 @@ seastar::future<> AlienStore::inject_mdata_error(const ghobject_t& o)
 {
   logger().debug("{}", __func__);
   assert(tp);
-  return seastar::with_gate(op_gate, [=, this] {
-    return tp->submit([=, this] {
+  return op_gates.simple_dispatch("inject_mdata_error", [=, this] {
+    return tp->submit([o, this] {
       return store->inject_mdata_error(o);
     });
   });
@@ -500,7 +501,7 @@ seastar::future<> AlienStore::write_meta(const std::string& key,
 {
   logger().debug("{}", __func__);
   assert(tp);
-  return seastar::with_gate(op_gate, [=, this] {
+  return op_gates.simple_dispatch("write_meta", [=, this] {
     return tp->submit([=, this] {
       return store->write_meta(key, value);
     }).then([] (int r) {
@@ -515,8 +516,8 @@ AlienStore::read_meta(const std::string& key)
 {
   logger().debug("{}", __func__);
   assert(tp);
-  return seastar::with_gate(op_gate, [this, key] {
-    return tp->submit([this, key] {
+  return op_gates.simple_dispatch("read_meta", [this, key] {
+    return tp->submit([key, this] {
       std::string value;
       int r = store->read_meta(key, &value);
       if (r > 0) {

src/crimson/os/alienstore/alien_store.h

@@ -10,6 +10,7 @@
 #include "os/ObjectStore.h"
 #include "osd/osd_types.h"
 
+#include "crimson/common/gated.h"
 #include "crimson/os/alienstore/thread_pool.h"
 #include "crimson/os/futurized_collection.h"
 #include "crimson/os/futurized_store.h"
@@ -111,9 +112,10 @@ class AlienStore final : public FuturizedStore,
   }
 
 private:
+
   template <class... Args>
   auto do_with_op_gate(Args&&... args) const {
-    return seastar::with_gate(op_gate,
+    return op_gates.simple_dispatch("AlienStore::do_with_op_gate",
       // perfect forwarding in lambda's closure isn't available in C++17
       // using tuple as workaround; see: https://stackoverflow.com/a/49902823
       [args = std::make_tuple(std::forward<Args>(args)...)] () mutable {
@@ -130,7 +132,7 @@ class AlienStore final : public FuturizedStore,
   uint64_t used_bytes = 0;
   std::unique_ptr<ObjectStore> store;
   std::unique_ptr<CephContext> cct;
-  mutable seastar::gate op_gate;
+  mutable crimson::common::gate_per_shard op_gates;
 
   /**
    * coll_map

src/crimson/osd/osd.cc

@@ -718,7 +718,7 @@ seastar::future<> OSD::stop()
     DEBUG("prepared to stop");
     public_msgr->stop();
     cluster_msgr->stop();
-    auto gate_close_fut = gate.close();
+    auto gate_close_fut = gate.close_all();
     return asok->stop().then([this] {
       return heartbeat->stop();
     }).then([this] {

src/crimson/osd/osd.h

@@ -232,7 +232,7 @@ class OSD final : public crimson::net::Dispatcher,
     Ref<MOSDPGUpdateLogMissingReply> m);
 
 private:
-  crimson::common::Gated gate;
+  crimson::common::gate_per_shard gate;
 
   seastar::promise<> stop_acked;
   void got_stop_ack() {