osd: abstract EC recovery's dispatch_recovery_messages() from IO

Signed-off-by: Radosław Zarzyński <rzarzyns@redhat.com>

diff --git a/src/osd/ECBackend.cc b/src/osd/ECBackend.cc
index 8ae0fe8c5c5c371375efce3dd8ce38769e1302aa..318b1aca1f2fb538fcbaea8bfe6639475af44c20 100644 (file)
--- a/src/osd/ECBackend.cc
+++ b/src/osd/ECBackend.cc
@@ -130,7 +130,7 @@ ECBackend::ECBackend(
   : PGBackend(cct, pg, store, coll, ch),
     read_pipeline(cct, ec_impl, this->sinfo, get_parent()->get_eclistener()),
     rmw_pipeline(cct, ec_impl, this->sinfo, get_parent()->get_eclistener(), *this),
-    recovery_backend(cct, coll, ec_impl, this->sinfo, read_pipeline, unstable_hashinfo_registry, get_parent()->get_eclistener()),
+    recovery_backend(cct, coll, ec_impl, this->sinfo, read_pipeline, unstable_hashinfo_registry, get_parent()),
     ec_impl(ec_impl),
     sinfo(ec_impl->get_data_chunk_count(), stripe_width),
     unstable_hashinfo_registry(cct, ec_impl) {
@@ -143,6 +143,23 @@ PGBackend::RecoveryHandle *ECBackend::open_recovery_op()
   return recovery_backend.open_recovery_op();
 }
 
+ECBackend::RecoveryBackend::RecoveryBackend(
+  CephContext* cct,
+  const coll_t &coll,
+  ceph::ErasureCodeInterfaceRef ec_impl,
+  const ECUtil::stripe_info_t& sinfo,
+  ReadPipeline& read_pipeline,
+  UnstableHashInfoRegistry& unstable_hashinfo_registry,
+  ECListener* parent)
+  : cct(cct),
+    coll(coll),
+    ec_impl(std::move(ec_impl)),
+    sinfo(sinfo),
+    read_pipeline(read_pipeline),
+    unstable_hashinfo_registry(unstable_hashinfo_registry),
+    parent(parent) {
+}
+
 PGBackend::RecoveryHandle *ECBackend::RecoveryBackend::open_recovery_op()
 {
   return new ECRecoveryHandle;
@@ -475,6 +492,19 @@ struct RecoveryReadCompleter : ECCommon::ReadCompleter {
   RecoveryMessages rm;
 };
 
+void ECBackend::ECRecoveryBackend::commit_txn_send_replies(
+  ceph::os::Transaction&& txn,
+  std::map<int, MOSDPGPushReply*> replies)
+{
+  txn.register_on_complete(
+      get_parent()->bless_context(
+        new SendPushReplies(
+          get_parent(),
+          get_osdmap_epoch(),
+          replies)));
+  get_parent()->queue_transaction(std::move(txn));
+}
+
 void ECBackend::RecoveryBackend::dispatch_recovery_messages(RecoveryMessages &m, int priority)
 {
   for (map<pg_shard_t, vector<PushOp> >::iterator i = m.pushes.begin();
@@ -510,15 +540,11 @@ void ECBackend::RecoveryBackend::dispatch_recovery_messages(RecoveryMessages &m,
     replies.insert(make_pair(i->first.osd, msg));
   }
 
+#if 1
   if (!replies.empty()) {
-    (m.t).register_on_complete(
-       get_parent()->bless_context(
-         new SendPushReplies(
-           get_parent(),
-           get_osdmap_epoch(),
-           replies)));
-    get_parent()->queue_transaction(std::move(m.t));
-  } 
+    commit_txn_send_replies(std::move(m.t), std::move(replies));
+  }
+#endif
 
   if (m.recovery_reads.empty())
     return;

diff --git a/src/osd/ECBackend.h b/src/osd/ECBackend.h
index 21f8ed29244c065c57353a6d549d7b88d5ad2b3f..56e1c972fe244b94c4c0ec934f453e350c726de6 100644 (file)
--- a/src/osd/ECBackend.h
+++ b/src/osd/ECBackend.h
@@ -208,20 +208,11 @@ public:
 
     RecoveryBackend(CephContext* cct,
                    const coll_t &coll,
-                ceph::ErasureCodeInterfaceRef ec_impl,
-                const ECUtil::stripe_info_t& sinfo,
-               ReadPipeline& read_pipeline,
-               UnstableHashInfoRegistry& unstable_hashinfo_registry,
-                ECListener* parent)
-      : cct(cct),
-        coll(coll),
-        ec_impl(std::move(ec_impl)),
-        sinfo(sinfo),
-       read_pipeline(read_pipeline),
-       unstable_hashinfo_registry(unstable_hashinfo_registry),
-        parent(parent) {
-    }
-  // <<<----
+                   ceph::ErasureCodeInterfaceRef ec_impl,
+                   const ECUtil::stripe_info_t& sinfo,
+                   ReadPipeline& read_pipeline,
+                   UnstableHashInfoRegistry& unstable_hashinfo_registry,
+                   ECListener* parent);
   struct RecoveryOp {
     hobject_t hoid;
     eversion_t v;
@@ -271,6 +262,10 @@ public:
                        sinfo.get_stripe_width());
   }
 
+  virtual ~RecoveryBackend() = default;
+  virtual void commit_txn_send_replies(
+    ceph::os::Transaction&& txn,
+    std::map<int, MOSDPGPushReply*> replies) = 0;
   void dispatch_recovery_messages(RecoveryMessages &m, int priority);
 
   RecoveryHandle *open_recovery_op();
@@ -305,6 +300,27 @@ public:
   }
   void _failed_push(const hobject_t &hoid, ECCommon::read_result_t &res);
   };
+  struct ECRecoveryBackend : RecoveryBackend {
+    ECRecoveryBackend(CephContext* cct,
+                     const coll_t &coll,
+                     ceph::ErasureCodeInterfaceRef ec_impl,
+                     const ECUtil::stripe_info_t& sinfo,
+                     ReadPipeline& read_pipeline,
+                     UnstableHashInfoRegistry& unstable_hashinfo_registry,
+                     Listener* parent)
+      : RecoveryBackend(cct, coll, std::move(ec_impl), sinfo, read_pipeline, unstable_hashinfo_registry, parent->get_eclistener()),
+       parent(parent) {
+    }
+
+    void commit_txn_send_replies(
+      ceph::os::Transaction&& txn,
+      std::map<int, MOSDPGPushReply*> replies) override;
+
+    Listener *get_parent() const { return parent; }
+
+  private:
+    Listener *parent;
+  };
   friend ostream &operator<<(ostream &lhs, const RecoveryBackend::RecoveryOp &rhs);
   friend struct RecoveryMessages;
   friend struct OnRecoveryReadComplete;
@@ -318,7 +334,7 @@ public:
 public:
   struct ReadPipeline read_pipeline;
   struct RMWPipeline rmw_pipeline;
-  struct RecoveryBackend recovery_backend;
+  struct ECRecoveryBackend recovery_backend;
 
   ceph::ErasureCodeInterfaceRef ec_impl;