mds: track which map batch_op is in-use in MDRequest

batch_op is in either CInode::batch_ops or CDentry::batch_ops. Current
code gets the map by dereferencing mdr->in[0] or mdr->dn[0].back().
The problem is that MDCache::path_traverse() may clear mdr->in[0] and
mdr->dn[0]. batch_op leaks when mds kills a request whose mdr->in[0] or
mdr->dn[0] were cleared.

Signed-off-by: "Yan, Zheng" <zyan@redhat.com>
(cherry picked from commit 7768d2c07d66ac015b2e235d38b7ac3616c47c43)

Conflicts:
	src/mds/Mutation.h
	  - octopus does not use the specific ns, use the original one

diff --git a/src/mds/MDCache.cc b/src/mds/MDCache.cc
index 88eece19f00bb485c42581b7cc838641626986bd..1cc398c48d8ed13040288339d67b33d778850625 100644 (file)
--- a/src/mds/MDCache.cc
+++ b/src/mds/MDCache.cc
@@ -9629,37 +9629,8 @@ void MDCache::request_forward(MDRequestRef& mdr, mds_rank_t who, int port)
   if (mdr->client_request && mdr->client_request->get_source().is_client()) {
     dout(7) << "request_forward " << *mdr << " to mds." << who << " req "
             << *mdr->client_request << dendl;
-    if (mdr->is_batch_head) {
-      int mask = mdr->client_request->head.args.getattr.mask;
-
-      switch (mdr->client_request->get_op()) {
-       case CEPH_MDS_OP_GETATTR:
-         {
-           CInode* in = mdr->in[0];
-           if (in) {
-             auto it = in->batch_ops.find(mask);
-             if (it != in->batch_ops.end()) {
-                it->second->forward(who);
-                in->batch_ops.erase(it);
-             }
-           }
-           break;
-         }
-       case CEPH_MDS_OP_LOOKUP:
-         {
-           if (mdr->dn[0].size()) {
-             CDentry* dn = mdr->dn[0].back();
-             auto it = dn->batch_ops.find(mask);
-             if (it != dn->batch_ops.end()) {
-               it->second->forward(who);
-               dn->batch_ops.erase(it);
-             }
-           }
-           break;
-         }
-       default:
-         ceph_abort();
-      }
+    if (mdr->is_batch_head()) {
+      mdr->release_batch_op()->forward(who);
     } else {
       mds->forward_message_mds(mdr->release_client_request(), who);
     }

diff --git a/src/mds/Mutation.cc b/src/mds/Mutation.cc
index d9f3f7cadcc60b049014a2096d964b0027d5f56f..acb9db218679254fb1cdac371abd685cdfa795d6 100644 (file)
--- a/src/mds/Mutation.cc
+++ b/src/mds/Mutation.cc
@@ -426,6 +426,15 @@ bool MDRequestImpl::is_batch_op()
      client_request->get_filepath().depth() == 0);
 }
 
+std::unique_ptr<BatchOp> MDRequestImpl::release_batch_op()
+{
+  int mask = client_request->head.args.getattr.mask;
+  auto it = batch_op_map->find(mask);
+  std::unique_ptr<BatchOp> bop = std::move(it->second);
+  batch_op_map->erase(it);
+  return bop;
+}
+
 int MDRequestImpl::compare_paths()
 {
   if (dir_root[0] < dir_root[1])

diff --git a/src/mds/Mutation.h b/src/mds/Mutation.h
index 8855d0af6b4056f74c80b4ad6a94a40a4a432f1b..b80aef2866fb47dc3c65091585f207ecd879265b 100644 (file)
--- a/src/mds/Mutation.h
+++ b/src/mds/Mutation.h
@@ -24,6 +24,7 @@
 
 #include "SimpleLock.h"
 #include "Capability.h"
+#include "BatchOp.h"
 
 #include "common/TrackedOp.h"
 #include "messages/MClientRequest.h"
@@ -381,9 +382,14 @@ struct MDRequestImpl : public MutationImpl {
   void set_filepath(const filepath& fp);
   void set_filepath2(const filepath& fp);
   bool is_queued_for_replay() const;
-  bool is_batch_op();
   int compare_paths();
 
+  bool is_batch_op();
+  bool is_batch_head() {
+    return batch_op_map != nullptr;
+  }
+  std::unique_ptr<BatchOp> release_batch_op();
+
   void print(ostream &out) const override;
   void dump(Formatter *f) const override;
 
@@ -435,7 +441,7 @@ struct MDRequestImpl : public MutationImpl {
   // indicates how may retries of request have been made
   int retry = 0;
 
-  bool is_batch_head = false;
+  std::map<int, std::unique_ptr<BatchOp> > *batch_op_map = nullptr;
 
   // indicator for vxattr osdmap update
   bool waited_for_osdmap = false;

diff --git a/src/mds/Server.cc b/src/mds/Server.cc
index d636646807d850628dc152cb98203c1995fadecf..d2895f0238ee67e022555c94fb68713ababa9e3c 100644 (file)
--- a/src/mds/Server.cc
+++ b/src/mds/Server.cc
@@ -82,10 +82,15 @@ class Batch_Getattr_Lookup : public BatchOp {
 protected:
   Server* server;
   ceph::ref_t<MDRequestImpl> mdr;
-  MDCache* mdcache;
   int res = 0;
 public:
-  Batch_Getattr_Lookup(Server* s, ceph::ref_t<MDRequestImpl> r, MDCache* mdc) : server(s), mdr(std::move(r)), mdcache(mdc) {}
+  Batch_Getattr_Lookup(Server* s, const ceph::ref_t<MDRequestImpl>& r)
+    : server(s), mdr(r) {
+    if (mdr->client_request->get_op() == CEPH_MDS_OP_LOOKUP)
+      mdr->batch_op_map = &mdr->dn[0].back()->batch_ops;
+    else
+      mdr->batch_op_map = &mdr->in[0]->batch_ops;
+  }
   void add_request(const ceph::ref_t<MDRequestImpl>& m) override {
     mdr->batch_reqs.push_back(m);
   }
@@ -93,6 +98,7 @@ public:
     mdr = m;
   }
   void _forward(mds_rank_t t) override {
+    MDCache* mdcache = server->mdcache;
     mdcache->mds->forward_message_mds(mdr->release_client_request(), t);
     mdr->set_mds_stamp(ceph_clock_now());
     for (auto& m : mdr->batch_reqs) {
@@ -1865,23 +1871,9 @@ void Server::submit_mdlog_entry(LogEvent *le, MDSLogContextBase *fin, MDRequestR
 void Server::respond_to_request(MDRequestRef& mdr, int r)
 {
   if (mdr->client_request) {
-    if (mdr->is_batch_op() && mdr->is_batch_head) {
-      int mask = mdr->client_request->head.args.getattr.mask;
-
-      std::unique_ptr<BatchOp> bop;
-      if (mdr->client_request->get_op() == CEPH_MDS_OP_GETATTR) {
-       dout(20) << __func__ << ": respond other getattr ops. " << *mdr << dendl;
-        auto it = mdr->in[0]->batch_ops.find(mask);
-        bop = std::move(it->second);
-       mdr->in[0]->batch_ops.erase(it);
-      } else {
-       dout(20) << __func__ << ": respond other lookup ops. " << *mdr << dendl;
-       auto it = mdr->dn[0].back()->batch_ops.find(mask);
-        bop = std::move(it->second);
-       mdr->dn[0].back()->batch_ops.erase(it);
-      }
-
-      bop->respond(r);
+    if (mdr->is_batch_head()) {
+      dout(20) << __func__ << " batch head " << *mdr << dendl;
+      mdr->release_batch_op()->respond(r);
     } else {
      reply_client_request(mdr, make_message<MClientReply>(*mdr->client_request, r));
     }
@@ -2402,16 +2394,6 @@ void Server::handle_osd_map()
     });
 }
 
-void Server::clear_batch_ops(const MDRequestRef& mdr)
-{
-  int mask = mdr->client_request->head.args.getattr.mask;
-  if (mdr->client_request->get_op() == CEPH_MDS_OP_GETATTR && mdr->in[0]) {
-    mdr->in[0]->batch_ops.erase(mask);
-  } else if (mdr->client_request->get_op() == CEPH_MDS_OP_LOOKUP && mdr->dn[0].size()) {
-    mdr->dn[0].back()->batch_ops.erase(mask);
-  }
-}
-
 void Server::dispatch_client_request(MDRequestRef& mdr)
 {
   // we shouldn't be waiting on anyone.
@@ -2421,7 +2403,10 @@ void Server::dispatch_client_request(MDRequestRef& mdr)
     dout(10) << "request " << *mdr << " was killed" << dendl;
     //if the mdr is a "batch_op" and it has followers, pick a follower as
     //the new "head of the batch ops" and go on processing the new one.
-    if (mdr->is_batch_op() && mdr->is_batch_head ) {
+    if (mdr->is_batch_head()) {
+      int mask = mdr->client_request->head.args.getattr.mask;
+      auto it = mdr->batch_op_map->find(mask);
+
       if (!mdr->batch_reqs.empty()) {
        MDRequestRef new_batch_head;
        for (auto itr = mdr->batch_reqs.cbegin(); itr != mdr->batch_reqs.cend();) {
@@ -2432,26 +2417,18 @@ void Server::dispatch_client_request(MDRequestRef& mdr)
            break;
          }
        }
-
        if (!new_batch_head) {
-         clear_batch_ops(mdr);
+         mdr->batch_op_map->erase(it);
          return;
        }
 
        new_batch_head->batch_reqs = std::move(mdr->batch_reqs);
-
+       new_batch_head->batch_op_map = mdr->batch_op_map;
+       mdr->batch_op_map = nullptr;
+       it->second->set_request(new_batch_head);
        mdr = new_batch_head;
-       mdr->is_batch_head = true;
-       int mask = mdr->client_request->head.args.getattr.mask;
-       if (mdr->client_request->get_op() == CEPH_MDS_OP_GETATTR) {
-         auto& fin = mdr->in[0]->batch_ops[mask];
-         fin->set_request(new_batch_head);
-       } else if (mdr->client_request->get_op() == CEPH_MDS_OP_LOOKUP) {
-         auto& fin = mdr->dn[0].back()->batch_ops[mask];
-         fin->set_request(new_batch_head);
-       }
       } else {
-       clear_batch_ops(mdr);
+       mdr->batch_op_map->erase(it);
        return;
       }
     } else {
@@ -3754,11 +3731,12 @@ void Server::handle_client_getattr(MDRequestRef& mdr, bool is_lookup)
 
   mdr->getattr_caps = mask;
 
-  if (mdr->snapid == CEPH_NOSNAP && !mdr->is_batch_head && mdr->is_batch_op()) {
+  if (mdr->snapid == CEPH_NOSNAP && !mdr->is_batch_head() && mdr->is_batch_op()) {
     if (!is_lookup) {
+      mdr->pin(ref);
       auto em = ref->batch_ops.emplace(std::piecewise_construct, std::forward_as_tuple(mask), std::forward_as_tuple());
       if (em.second) {
-        em.first->second = std::make_unique<Batch_Getattr_Lookup>(this, mdr, mdcache);
+       em.first->second = std::make_unique<Batch_Getattr_Lookup>(this, mdr);
       } else {
        dout(20) << __func__ << ": GETATTR op, wait for previous same getattr ops to respond. " << *mdr << dendl;
        em.first->second->add_request(mdr);
@@ -3766,17 +3744,16 @@ void Server::handle_client_getattr(MDRequestRef& mdr, bool is_lookup)
       }
     } else {
       CDentry* dn = mdr->dn[0].back();
+      mdr->pin(dn);
       auto em = dn->batch_ops.emplace(std::piecewise_construct, std::forward_as_tuple(mask), std::forward_as_tuple());
       if (em.second) {
-       em.first->second = std::make_unique<Batch_Getattr_Lookup>(this, mdr, mdcache);
-       mdr->pin(dn);
+       em.first->second = std::make_unique<Batch_Getattr_Lookup>(this, mdr);
       } else {
        dout(20) << __func__ << ": LOOKUP op, wait for previous same getattr ops to respond. " << *mdr << dendl;
        em.first->second->add_request(mdr);
        return;
       }
     }
-    mdr->is_batch_head = true;
   }
 
   /*

diff --git a/src/mds/Server.h b/src/mds/Server.h
index 12a9be819421d9d4c6eef8502960e7179c934d1e..98fd2da10a376aab3982728ff6b6d8cb922725be 100644 (file)
--- a/src/mds/Server.h
+++ b/src/mds/Server.h
@@ -164,7 +164,6 @@ public:
   void set_trace_dist(const ref_t<MClientReply> &reply, CInode *in, CDentry *dn,
                      MDRequestRef& mdr);
 
-
   void handle_slave_request(const cref_t<MMDSSlaveRequest> &m);
   void handle_slave_request_reply(const cref_t<MMDSSlaveRequest> &m);
   void dispatch_slave_request(MDRequestRef& mdr);
@@ -320,7 +319,6 @@ private:
 
   void reply_client_request(MDRequestRef& mdr, const ref_t<MClientReply> &reply);
   void flush_session(Session *session, MDSGatherBuilder *gather);
-  void clear_batch_ops(const MDRequestRef& mdr);
 
   MDSRank *mds;
   MDCache *mdcache;