mds: cap writeback authlock fields as well as filelock fields

No xattrs yet.

diff --git a/src/mds/Locker.cc b/src/mds/Locker.cc
index b7968f743bef29e8170cd97f8daefeaea2ee72a5..6d93884851742d88f66a1d73ea04b8f697fb4f49 100644 (file)
--- a/src/mds/Locker.cc
+++ b/src/mds/Locker.cc
@@ -999,7 +999,10 @@ void Locker::handle_client_caps(MClientCaps *m)
   assert(cap);
 
   // freezing|frozen?
-  if ((in->is_freezing() && in->filelock.is_stable()) ||  // continue if freezing and lock is unstable
+  if ((in->is_freezing() && (in->filelock.is_stable() &&
+                            in->authlock.is_stable() &&
+                            in->xattrlock.is_stable() &&
+                            in->linklock.is_stable())) ||  // continue if freezing and lock is unstable
       in->is_frozen()) {
     dout(7) << "handle_client_caps freezing|frozen on " << *in << dendl;
     in->add_waiter(CInode::WAIT_UNFREEZE, new C_MDS_RetryMessage(mds, m));
@@ -1145,22 +1148,33 @@ void Locker::_do_cap_update(CInode *in, int had, int all_wanted, snapid_t follow
 
   // atime|mtime|size?
   bool had_or_has_wr = had & (CEPH_CAP_GWR << CEPH_CAP_SFILE);
-  bool excl = had & CEPH_CAP_ANY_EXCL;
+  bool file_excl = had & (CEPH_CAP_GEXCL << CEPH_CAP_SFILE);
   bool dirty_atime = false;
   bool dirty_mtime = false;
   bool dirty_ctime = false;
   bool dirty_size = false;
-  if (had_or_has_wr || excl) {
-    if (mtime > latest->mtime || (excl && mtime != latest->mtime)) 
-      dirty_mtime = true;
+  bool dirty_file = false;
+  if (had_or_has_wr || file_excl) {
+    if (mtime > latest->mtime || (file_excl && mtime != latest->mtime)) 
+      dirty_file = dirty_mtime = true;
     if (ctime > latest->ctime)
-      dirty_ctime = true;
+      dirty_file = dirty_ctime = true;
     if (size > latest->size) 
-      dirty_size = true;
-  }
-  if (excl && atime != latest->atime)
-    dirty_atime = true;
-  bool dirty = dirty_atime || dirty_mtime || dirty_ctime || dirty_size;
+      dirty_file = dirty_size = true;
+  }
+  if (file_excl && atime != latest->atime)
+    dirty_file = dirty_atime = true;
+  bool dirty_mode = false;
+  bool dirty_owner = false;
+  bool dirty_auth = false;
+  if (had & (CEPH_CAP_GEXCL << CEPH_CAP_SAUTH)) {
+    if (m->head.uid != latest->uid ||
+       m->head.gid != latest->gid)
+      dirty_auth = dirty_owner = true;
+    if (m->head.mode != latest->mode)
+      dirty_auth = dirty_mode = true;
+  }
+  bool dirty = dirty_file || dirty_auth;
   
   // increase or zero max_size?
   bool change_max = false;
@@ -1224,14 +1238,30 @@ void Locker::_do_cap_update(CInode *in, int had, int all_wanted, snapid_t follow
              << " for " << *in << dendl;
       pi->atime = atime;
     }
-    if (excl && pi->time_warp_seq < m->get_time_warp_seq()) {
+    if (file_excl && pi->time_warp_seq < m->get_time_warp_seq()) {
       dout(7) << "  time_warp_seq " << pi->time_warp_seq << " -> " << m->get_time_warp_seq()
              << " for " << *in << dendl;
       pi->time_warp_seq = m->get_time_warp_seq();
     }
+    if (dirty_owner) {
+      dout(7) << "  uid.gid " << pi->uid << "." << pi->gid
+             << " -> " << m->head.uid << "." << m->head.gid
+             << " for " << *in << dendl;
+      pi->uid = m->head.uid;
+      pi->gid = m->head.gid;
+    }
+    if (dirty_mode) {
+      dout(7) << "  mode " << oct << pi->mode
+             << " -> " << m->head.mode << dec
+             << " for " << *in << dendl;
+      pi->mode = m->head.mode;
+    }
     Mutation *mut = new Mutation;
     mut->ls = mds->mdlog->get_current_segment();
-    file_wrlock_force(&in->filelock, mut);  // wrlock for duration of journal
+    if (dirty_file)
+      file_wrlock_force(&in->filelock, mut);  // wrlock for duration of journal
+    if (dirty_auth)
+      simple_wrlock_force(&in->authlock, mut);
     mut->auth_pin(in);
     mdcache->predirty_journal_parents(mut, &le->metablob, in, 0, PREDIRTY_PRIMARY, 0, follows);
     mdcache->journal_dirty_inode(mut, &le->metablob, in, follows);
@@ -1250,12 +1280,6 @@ void Locker::_do_cap_update(CInode *in, int had, int all_wanted, snapid_t follow
     } else if (ack)
       mds->send_message_client(ack, client);
   }
-
-  // reevaluate, waiters
-  if (!in->filelock.is_stable())
-    file_eval_gather(&in->filelock);
-  else if (in->is_auth())
-    file_eval(&in->filelock);
 }
 
 
@@ -1805,6 +1829,16 @@ bool Locker::simple_rdlock_start(SimpleLock *lock, MDRequest *mut)
   return false;
 }
 
+bool Locker::simple_wrlock_force(SimpleLock *lock, Mutation *mut)
+{
+  dout(7) << "simple_wrlock_force  on " << *lock
+         << " on " << *lock->get_parent() << dendl;  
+  lock->get_wrlock(true);
+  mut->wrlocks.insert(lock);
+  mut->locks.insert(lock);
+  return true;
+}
+
 bool Locker::simple_wrlock_start(SimpleLock *lock, MDRequest *mut)
 {
   dout(7) << "simple_wrlock_start  on " << *lock

diff --git a/src/mds/Locker.h b/src/mds/Locker.h
index 5ba08c0785051ff23783997aac6fef575d8f34d3..120fdeb78e7050f38ddafefbee003800f2e455e4 100644 (file)
--- a/src/mds/Locker.h
+++ b/src/mds/Locker.h
@@ -107,6 +107,7 @@ protected:
   void simple_lock(SimpleLock *lock);
   bool simple_rdlock_start(SimpleLock *lock, MDRequest *mut);
   void simple_rdlock_finish(SimpleLock *lock, Mutation *mut);
+  bool simple_wrlock_force(SimpleLock *lock, Mutation *mut);
   bool simple_wrlock_start(SimpleLock *lock, MDRequest *mut);
   void simple_wrlock_finish(SimpleLock *lock, Mutation *mut);
   bool simple_xlock_start(SimpleLock *lock, MDRequest *mut);

diff --git a/src/messages/MClientCaps.h b/src/messages/MClientCaps.h
index 2bc3da20e01f4e2c5fb8d80ebdc68d928df00d23..0cba925ae9e16f7bb1ce091e0bfe64eba6463ba1 100644 (file)
--- a/src/messages/MClientCaps.h
+++ b/src/messages/MClientCaps.h
@@ -22,6 +22,7 @@ class MClientCaps : public Message {
  public:
   struct ceph_mds_caps head;
   bufferlist snapbl;
+  bufferlist xattrbl;
 
   int      get_caps() { return head.caps; }
   int      get_wanted() { return head.wanted; }
@@ -73,11 +74,18 @@ class MClientCaps : public Message {
     head.seq = seq;
     head.caps = caps;
     head.wanted = wanted;
+    head.migrate_seq = mseq;
+
+    head.uid = inode.uid;
+    head.gid = inode.gid;
+    head.mode = inode.mode;
+
+    head.xattr_len = 0; // FIXME
+
     head.layout = inode.layout;
     head.size = inode.size;
     head.max_size = inode.max_size;
     head.truncate_seq = inode.truncate_seq;
-    head.migrate_seq = mseq;
     inode.mtime.encode_timeval(&head.mtime);
     inode.atime.encode_timeval(&head.atime);
     inode.ctime.encode_timeval(&head.ctime);
@@ -116,11 +124,13 @@ class MClientCaps : public Message {
     bufferlist::iterator p = payload.begin();
     ::decode(head, p);
     ::decode_nohead(head.snap_trace_len, snapbl, p);
+    ::decode_nohead(head.xattr_len, xattrbl, p);
   }
   void encode_payload() {
     head.snap_trace_len = snapbl.length();
     ::encode(head, payload);
     ::encode_nohead(snapbl, payload);
+    ::encode_nohead(xattrbl, payload);
   }
 };