mds: don't delay processing replica buffer in slave request

Replicated objects need to be added into the cache immediately

Signed-off-by: Yan, Zheng <zheng.z.yan@intel.com>
Reviewed-by: Greg Farnum <greg@inktank.com>

diff --git a/src/mds/MDCache.cc b/src/mds/MDCache.cc
index 71e333bc1c44697f3abeaf964f1d0cae8baf0724..f42c018352f46bc14c55616b9f96c00eef714b7d 100644 (file)
--- a/src/mds/MDCache.cc
+++ b/src/mds/MDCache.cc
@@ -7726,6 +7726,18 @@ void MDCache::_find_ino_dir(inodeno_t ino, Context *fin, bufferlist& bl, int r)
 
 /* ---------------------------- */
 
+int MDCache::get_num_client_requests()
+{
+  int count = 0;
+  for (hash_map<metareqid_t, MDRequest*>::iterator p = active_requests.begin();
+      p != active_requests.end();
+      ++p) {
+    if (p->second->reqid.name.is_client() && !p->second->is_slave())
+      count++;
+  }
+  return count;
+}
+
 /* This function takes over the reference to the passed Message */
 MDRequest *MDCache::request_start(MClientRequest *req)
 {

diff --git a/src/mds/MDCache.h b/src/mds/MDCache.h
index 2ae5b4c39a894719cd0ccf824c078711b09a9653..1666a3dff8d48b5d16884d57440da1d48fc870f8 100644 (file)
--- a/src/mds/MDCache.h
+++ b/src/mds/MDCache.h
@@ -240,7 +240,7 @@ protected:
   hash_map<metareqid_t, MDRequest*> active_requests; 
 
 public:
-  int get_num_active_requests() { return active_requests.size(); }
+  int get_num_client_requests();
 
   MDRequest* request_start(MClientRequest *req);
   MDRequest* request_start_slave(metareqid_t rid, __u32 attempt, int by);

diff --git a/src/mds/MDS.cc b/src/mds/MDS.cc
index 2d48815b247da00e9c293a5581dc4eebc17846d0..99b14d963ea912d091e49f4d4833985b28600eb3 100644 (file)
--- a/src/mds/MDS.cc
+++ b/src/mds/MDS.cc
@@ -1905,9 +1905,9 @@ bool MDS::_dispatch(Message *m)
       mdcache->is_open() &&
       replay_queue.empty() &&
       want_state == MDSMap::STATE_CLIENTREPLAY) {
-    dout(10) << " still have " << mdcache->get_num_active_requests()
-            << " active replay requests" << dendl;
-    if (mdcache->get_num_active_requests() == 0)
+    int num_requests = mdcache->get_num_client_requests();
+    dout(10) << " still have " << num_requests << " active replay requests" << dendl;
+    if (num_requests == 0)
       clientreplay_done();
   }
 

diff --git a/src/mds/Server.cc b/src/mds/Server.cc
index cfed235a927bcbda7cec76dda93cf2643ae26c6f..9e0ad7aff9e3c79f437997a7a7cf02f8f2441f37 100644 (file)
--- a/src/mds/Server.cc
+++ b/src/mds/Server.cc
@@ -110,10 +110,8 @@ void Server::dispatch(Message *m)
                (m->get_type() == CEPH_MSG_CLIENT_REQUEST &&
                 (static_cast<MClientRequest*>(m))->is_replay()))) {
       // replaying!
-    } else if (mds->is_clientreplay() && m->get_type() == MSG_MDS_SLAVE_REQUEST &&
-              ((static_cast<MMDSSlaveRequest*>(m))->is_reply() ||
-               !mds->mdsmap->is_active(m->get_source().num()))) {
-      // slave reply or the master is also in the clientreplay stage
+    } else if (m->get_type() == MSG_MDS_SLAVE_REQUEST) {
+      // handle_slave_request() will wait if necessary
     } else {
       dout(3) << "not active yet, waiting" << dendl;
       mds->wait_for_active(new C_MDS_RetryMessage(mds, m));
@@ -1294,6 +1292,13 @@ void Server::handle_slave_request(MMDSSlaveRequest *m)
   if (m->is_reply())
     return handle_slave_request_reply(m);
 
+  CDentry *straydn = NULL;
+  if (m->stray.length() > 0) {
+    straydn = mdcache->add_replica_stray(m->stray, from);
+    assert(straydn);
+    m->stray.clear();
+  }
+
   // am i a new slave?
   MDRequest *mdr = NULL;
   if (mdcache->have_request(m->get_reqid())) {
@@ -1329,9 +1334,26 @@ void Server::handle_slave_request(MMDSSlaveRequest *m)
       m->put();
       return;
     }
-    mdr = mdcache->request_start_slave(m->get_reqid(), m->get_attempt(), m->get_source().num());
+    mdr = mdcache->request_start_slave(m->get_reqid(), m->get_attempt(), from);
   }
   assert(mdr->slave_request == 0);     // only one at a time, please!  
+
+  if (straydn) {
+    mdr->pin(straydn);
+    mdr->straydn = straydn;
+  }
+
+  if (!mds->is_clientreplay() && !mds->is_active() && !mds->is_stopping()) {
+    dout(3) << "not clientreplay|active yet, waiting" << dendl;
+    mds->wait_for_replay(new C_MDS_RetryMessage(mds, m));
+    return;
+  } else if (mds->is_clientreplay() && !mds->mdsmap->is_clientreplay(from) &&
+            mdr->locks.empty()) {
+    dout(3) << "not active yet, waiting" << dendl;
+    mds->wait_for_active(new C_MDS_RetryMessage(mds, m));
+    return;
+  }
+
   mdr->slave_request = m;
   
   dispatch_slave_request(mdr);
@@ -1342,6 +1364,12 @@ void Server::handle_slave_request_reply(MMDSSlaveRequest *m)
 {
   int from = m->get_source().num();
   
+  if (!mds->is_clientreplay() && !mds->is_active() && !mds->is_stopping()) {
+    dout(3) << "not clientreplay|active yet, waiting" << dendl;
+    mds->wait_for_replay(new C_MDS_RetryMessage(mds, m));
+    return;
+  }
+
   if (m->get_op() == MMDSSlaveRequest::OP_COMMITTED) {
     metareqid_t r = m->get_reqid();
     mds->mdcache->committed_master_slave(r, from);
@@ -5176,10 +5204,8 @@ void Server::handle_slave_rmdir_prep(MDRequest *mdr)
   dout(10) << " dn " << *dn << dendl;
   mdr->pin(dn);
 
-  assert(mdr->slave_request->stray.length() > 0);
-  CDentry *straydn = mdcache->add_replica_stray(mdr->slave_request->stray, mdr->slave_to_mds);
-  assert(straydn);
-  mdr->pin(straydn);
+  assert(mdr->straydn);
+  CDentry *straydn = mdr->straydn;
   dout(10) << " straydn " << *straydn << dendl;
   
   mdr->now = mdr->slave_request->now;
@@ -5246,6 +5272,7 @@ void Server::_logged_slave_rmdir(MDRequest *mdr, CDentry *dn, CDentry *straydn)
   // done.
   mdr->slave_request->put();
   mdr->slave_request = 0;
+  mdr->straydn = 0;
 }
 
 void Server::handle_slave_rmdir_prep_ack(MDRequest *mdr, MMDSSlaveRequest *ack)
@@ -6506,15 +6533,12 @@ void Server::handle_slave_rename_prep(MDRequest *mdr)
   // stray?
   bool linkmerge = (srcdnl->get_inode() == destdnl->get_inode() &&
                    (srcdnl->is_primary() || destdnl->is_primary()));
-  CDentry *straydn = 0;
-  if (destdnl->is_primary() && !linkmerge) {
-    assert(mdr->slave_request->stray.length() > 0);
-    straydn = mdcache->add_replica_stray(mdr->slave_request->stray, mdr->slave_to_mds);
+  CDentry *straydn = mdr->straydn;
+  if (destdnl->is_primary() && !linkmerge)
     assert(straydn);
-    mdr->pin(straydn);
-  }
 
   mdr->now = mdr->slave_request->now;
+  mdr->more()->srcdn_auth_mds = srcdn->authority().first;
 
   // set up commit waiter (early, to clean up any freezing etc we do)
   if (!mdr->more()->slave_commit)
@@ -6697,6 +6721,7 @@ void Server::_logged_slave_rename(MDRequest *mdr,
   // done.
   mdr->slave_request->put();
   mdr->slave_request = 0;
+  mdr->straydn = 0;
 }
 
 void Server::_commit_slave_rename(MDRequest *mdr, int r,