mds: get_linkage() should return projected if xlocked by self

Also, move set_xlocks_done() to _below_ set_trace_dist, in case some
other request comes in and xlocks the dentry.

Allow read_projected in XLOCK state too.  Note that other requests
_still_ won't be able to re-xlock or rdlock until the prior request
set xlock done.

diff --git a/src/mds/CDentry.h b/src/mds/CDentry.h
index b3db9b93dea177c0ad450fa0b231e8fb53c13fe2..67c2b5753c03a70f79bfe6dcec9dcbe3b6c84cce 100644 (file)
--- a/src/mds/CDentry.h
+++ b/src/mds/CDentry.h
@@ -207,11 +207,12 @@ public:
     return get_projected_linkage()->inode;
   }
 
-  bool use_projected(int client) {
-    return lock.can_read_projected(client);
+  bool use_projected(int client, Mutation *mut) {
+    return lock.can_read_projected(client) || 
+      lock.get_xlocked_by() == mut;
   }
-  linkage_t *get_linkage(int client) {
-    return use_projected(client) ? get_projected_linkage() : get_linkage();
+  linkage_t *get_linkage(int client, Mutation *mut) {
+    return use_projected(client, mut) ? get_projected_linkage() : get_linkage();
   }
 
   // ref counts: pin ourselves in the LRU when we're pinned.

diff --git a/src/mds/Server.cc b/src/mds/Server.cc
index deb5db6ec2e16ad912fd5f6407f7ad86e39cdca4..faffe5ba360f4a1a9239c61bb32e87d9f5d1d1bb 100644 (file)
--- a/src/mds/Server.cc
+++ b/src/mds/Server.cc
@@ -525,9 +525,6 @@ void Server::early_reply(MDRequest *mdr, CInode *tracei, CDentry *tracedn)
     return;
   }
 
-  // mark xlocks "done", indicating that we are exposing uncommitted changes
-  mds->locker->set_xlocks_done(mdr);
-
   MClientRequest *req = mdr->client_request;
   entity_inst_t client_inst = req->get_orig_source_inst();
   if (client_inst.name.is_mds())
@@ -545,9 +542,12 @@ void Server::early_reply(MDRequest *mdr, CInode *tracei, CDentry *tracedn)
   if (tracei || tracedn)
     set_trace_dist(mdr->session, reply, tracei, tracedn, snapid, snapdiri, true, mdr);
 
-  mdr->did_early_reply = true;
-
   messenger->send_message(reply, client_inst);
+
+  // mark xlocks "done", indicating that we are exposing uncommitted changes
+  mds->locker->set_xlocks_done(mdr);
+
+  mdr->did_early_reply = true;
 }
 
 /*
@@ -685,7 +685,7 @@ void Server::set_trace_dist(Session *session, MClientReply *reply, CInode *in, C
   // start with dentry or inode?
   if (!in) {
     assert(dn);
-    in = dn->get_linkage(client)->get_inode();
+    in = dn->get_linkage(client, mdr)->get_inode();
     goto dentry;
   }
 
@@ -725,7 +725,7 @@ void Server::set_trace_dist(Session *session, MClientReply *reply, CInode *in, C
 
   if (!dn) {
     dn = in->get_projected_parent_dn();
-    if (dn && !dn->use_projected(client))
+    if (dn && !dn->use_projected(client, mdr))
       dn = NULL;
     if (!dn)
       dn = in->get_parent_dn();
@@ -1309,7 +1309,7 @@ CDentry* Server::prepare_null_dentry(MDRequest *mdr, CDir *dir, const string& dn
       return 0;
     }
     */
-    if (!dn->get_linkage(client)->is_null()) {
+    if (!dn->get_linkage(client, mdr)->is_null()) {
       // name already exists
       dout(10) << "dentry " << dname << " exists in " << *dir << dendl;
       if (!okexist) {
@@ -1510,7 +1510,7 @@ CInode* Server::rdlock_path_pin_ref(MDRequest *mdr,
     ref = mdcache->get_inode(refpath.get_ino());
   else {
     CDentry *dn = trace[trace.size()-1];
-    bool dnp = dn->use_projected(client);
+    bool dnp = dn->use_projected(client, mdr);
     CDentry::linkage_t *dnl = dnp ? dn->get_projected_linkage() : dn->get_linkage();
 
     // if no inode (null or unattached remote), fw to dentry auth?
@@ -1635,7 +1635,7 @@ CDentry* Server::rdlock_path_xlock_dentry(MDRequest *mdr, bool okexist, bool mus
     }
       
     // exists?
-    if (!dn || dn->get_linkage(client)->is_null()) {
+    if (!dn || dn->get_linkage(client, mdr)->is_null()) {
       dout(7) << "dentry " << dname << " dne in " << *dir << dendl;
       reply_request(mdr, -ENOENT);
       return 0;
@@ -1651,7 +1651,7 @@ CDentry* Server::rdlock_path_xlock_dentry(MDRequest *mdr, bool okexist, bool mus
 
   for (int i=0; i<(int)trace.size(); i++) 
     rdlocks.insert(&trace[i]->lock);
-  if (dn->get_linkage(client)->is_null())
+  if (dn->get_linkage(client, mdr)->is_null())
     xlocks.insert(&dn->lock);                 // new dn, xlock
   else
     rdlocks.insert(&dn->lock);  // existing dn, rdlock
@@ -2207,7 +2207,7 @@ void Server::handle_client_readdir(MDRequest *mdr)
     CDentry *dn = it->second;
     it++;
 
-    bool dnp = dn->use_projected(client);
+    bool dnp = dn->use_projected(client, mdr);
     CDentry::linkage_t *dnl = dnp ? dn->get_projected_linkage() : dn->get_linkage();
 
     if (dnl->is_null())
@@ -3107,7 +3107,7 @@ void Server::handle_client_unlink(MDRequest *mdr)
     return;
   }
 
-  CDentry::linkage_t *dnl = dn->get_linkage(client);
+  CDentry::linkage_t *dnl = dn->get_linkage(client, mdr);
   if (dnl->is_null()) {
     reply_request(mdr, -ENOENT);
     return;
@@ -5223,7 +5223,7 @@ void Server::handle_client_lssnap(MDRequest *mdr)
     mdcache->request_forward(mdr, dn->authority().first);
     return;
   }
-  CDentry::linkage_t *dnl = dn->get_linkage(client);
+  CDentry::linkage_t *dnl = dn->get_linkage(client, mdr);
 
   // dir only
   CInode *diri = dnl->get_inode();

diff --git a/src/mds/locks.c b/src/mds/locks.c
index 51961f4fde2f4418ab42596d470fba3b9faec878..e66c68100959db5223654b487a56bfb1d54ef155 100644 (file)
--- a/src/mds/locks.c
+++ b/src/mds/locks.c
@@ -22,9 +22,9 @@ struct sm_state_t simplelock[20] = {
     [LOCK_SYNC_LOCK] = { LOCK_LOCK, false, LOCK_LOCK, ANY,  0,   0,   0,   0,   0,   0,0,0 }, 
     [LOCK_EXCL_LOCK] = { LOCK_LOCK, false, LOCK_LOCK, 0,    0,   0,   0,   0,   0,   0,0,0 },
 
-    [LOCK_XLOCK]     = { 0,         false, LOCK_LOCK, 0,    0,   0,   0,   0,   0,   0,0,0 },
-    [LOCK_LOCK_XLOCK]= { LOCK_XLOCK,false, LOCK_LOCK, 0,    0,   0,   0,   0,   0,   0,0,0 },
+    [LOCK_XLOCK]     = { 0,         false, LOCK_LOCK, 0,    XCL, 0,   0,   0,   0,   0,0,0 },
     [LOCK_XLOCKDONE] = { LOCK_SYNC, false, LOCK_LOCK, XCL,  XCL, XCL, 0,   XCL, XCL, 0,0,0 },
+    [LOCK_LOCK_XLOCK]= { LOCK_XLOCK,false, LOCK_LOCK, 0,    0,   0,   0,   0,   0,   0,0,0 },
 
     [LOCK_EXCL]      = { 0,         true,  LOCK_LOCK, 0,    0,   0,   0,   0,   0,   0,CEPH_CAP_GEXCL|CEPH_CAP_GRDCACHE,0 },
     [LOCK_SYNC_EXCL] = { LOCK_EXCL, true,  LOCK_LOCK, ANY,  0,   0,   0,   0,   0,   0,CEPH_CAP_GRDCACHE,0 },
@@ -88,7 +88,7 @@ struct sm_state_t filelock[30] = {
     [LOCK_EXCL_LOCK] = { LOCK_LOCK, false, LOCK_LOCK, 0,    0,   0,   0,   0,   0,   CEPH_CAP_GRDCACHE|CEPH_CAP_GWRBUFFER,0,CEPH_CAP_GRDCACHE },
     [LOCK_MIX_LOCK]  = { LOCK_LOCK, false, LOCK_LOCK, AUTH, 0,   0,   0,   0,   0,   0,0,0 },
 
-    [LOCK_XLOCK]     = { 0,         false, LOCK_LOCK, 0,    0,   0,   0,   0,   0,   0,0,0 },
+    [LOCK_XLOCK]     = { 0,         false, LOCK_LOCK, 0,    XCL, 0,   0,   0,   0,   0,0,0 },
     [LOCK_XLOCKDONE] = { LOCK_SYNC, false, LOCK_LOCK, XCL,  XCL, XCL, 0,   XCL, XCL, 0,0,0 },
 
     [LOCK_MIX]       = { 0,         false, LOCK_MIX,  0,    0,   FW,  ANY, 0,   0,   CEPH_CAP_GRD|CEPH_CAP_GWR,0,CEPH_CAP_GRD },