mds: Locker needs to remember requested max_size changes from clients

Previously, if we received an MClientCaps request containing a change
in the inode's max size, and _do_cap_update() was unable to process
the request immediately (due to a locking issue), we would wait-list
the request by adding a call to check_inode_max_size() once the lock
became stable. However, we then tossed out the message without in any
way propagating the new max size which had been requested!

Handle this by extending check_inode_max_size to also accept parameters
for increasing the max size, and by storing all the parameters explicitly
in the C_MDL_CheckMaxSize Context instead of relying on defaults. That
gets us to the point where we *can* notice we need to increase the max. To
actually do so, we now pass calc_new_client_ranges() the requested max
size instead of the actual size if we're doing an update.

Notice that as a side effect of this, all clients get to see the max size
increase instead of just the requester. This should be okay, but it is
chattier than in the optimal case (where we don't get stuck on a lock).

Fixes #3637

Signed-off-by: Greg Farnum <greg@inktank.com>

3 files changed, 43 insertions, 10 deletions

diff --git a/src/mds/Locker.cc b/src/mds/Locker.cc
index eab7ed686bc..ea923c3eba1 100644
--- a/src/mds/Locker.cc
+++ b/src/mds/Locker.cc
@@ -1956,14 +1956,27 @@ void Locker::handle_inode_file_caps(MInodeFileCaps *m)
 class C_MDL_CheckMaxSize : public Context {
   Locker *locker;
   CInode *in;
+  bool update_size;
+  uint64_t newsize;
+  bool update_max;
+  uint64_t new_max_size;
+  utime_t mtime;
+
 public:
-  C_MDL_CheckMaxSize(Locker *l, CInode *i) : locker(l), in(i) {
+  C_MDL_CheckMaxSize(Locker *l, CInode *i, bool _update_size, uint64_t _newsize,
+                     bool _update_max, uint64_t _new_max_size, utime_t _mtime) :
+    locker(l), in(i),
+    update_size(_update_size), newsize(_newsize),
+    update_max(_update_max), new_max_size(_new_max_size),
+    mtime(_mtime)
+  {
     in->get(CInode::PIN_PTRWAITER);
   }
   void finish(int r) {
     in->put(CInode::PIN_PTRWAITER);
     if (in->is_auth())
-      locker->check_inode_max_size(in);
+      locker->check_inode_max_size(in, false, update_size, newsize,
+                                   update_max, new_max_size, mtime);
   }
 };
 
@@ -1995,7 +2008,9 @@ void Locker::calc_new_client_ranges(CInode *in, uint64_t size, map<client_t,clie
 }
 
 bool Locker::check_inode_max_size(CInode *in, bool force_wrlock,
-				  bool update_size, uint64_t new_size, utime_t new_mtime)
+				  bool update_size, uint64_t new_size,
+				  bool update_max, uint64_t new_max_size,
+				  utime_t new_mtime)
 {
   assert(in->is_auth());
 
@@ -2004,9 +2019,11 @@ bool Locker::check_inode_max_size(CInode *in, bool force_wrlock,
   uint64_t size = latest->size;
   if (update_size)
     size = new_size;
-  bool new_max = false;
+  bool new_max = update_max;
+
+  uint64_t client_range_size = update_max ? new_max_size : size;
 
-  calc_new_client_ranges(in, size, new_ranges);
+  calc_new_client_ranges(in, client_range_size, new_ranges);
 
   if (latest->client_ranges != new_ranges)
     new_max = true;
@@ -2022,7 +2039,11 @@ bool Locker::check_inode_max_size(CInode *in, bool force_wrlock,
 
   if (in->is_frozen()) {
     dout(10) << "check_inode_max_size frozen, waiting on " << *in << dendl;
-    in->add_waiter(CInode::WAIT_UNFREEZE, new C_MDL_CheckMaxSize(this, in));
+    C_MDL_CheckMaxSize *cms = new C_MDL_CheckMaxSize(this, in,
+                                                     update_size, new_size,
+                                                     update_max, new_max_size,
+                                                     new_mtime);
+    in->add_waiter(CInode::WAIT_UNFREEZE, cms);
     return false;
   }
   if (!force_wrlock && !in->filelock.can_wrlock(in->get_loner())) {
@@ -2035,7 +2056,12 @@ bool Locker::check_inode_max_size(CInode *in, bool force_wrlock,
     }
     if (!in->filelock.can_wrlock(in->get_loner())) {
       // try again later
-      in->filelock.add_waiter(SimpleLock::WAIT_STABLE, new C_MDL_CheckMaxSize(this, in));
+      C_MDL_CheckMaxSize *cms = new C_MDL_CheckMaxSize(this, in,
+                                                       update_size, new_size,
+                                                       update_max, new_max_size,
+                                                       new_mtime);
+
+      in->filelock.add_waiter(SimpleLock::WAIT_STABLE, cms);
       dout(10) << "check_inode_max_size can't wrlock, waiting on " << *in << dendl;
       return false;    
     }
@@ -2755,7 +2781,12 @@ bool Locker::_do_cap_update(CInode *in, Capability *cap,
       }
       if (!in->filelock.can_wrlock(client) &&
 	  !in->filelock.can_force_wrlock(client)) {
-	in->filelock.add_waiter(SimpleLock::WAIT_STABLE, new C_MDL_CheckMaxSize(this, in));
+	C_MDL_CheckMaxSize *cms = new C_MDL_CheckMaxSize(this, in,
+	                                                 false, 0,
+	                                                 change_max, new_max,
+	                                                 utime_t());
+
+	in->filelock.add_waiter(SimpleLock::WAIT_STABLE, cms);
 	change_max = false;
       }
     }
diff --git a/src/mds/Locker.h b/src/mds/Locker.h
index d98104fc435..f4d9861a384 100644
--- a/src/mds/Locker.h
+++ b/src/mds/Locker.h
@@ -274,7 +274,9 @@ protected:
 			  MClientCaps *ack);
 public:
   void calc_new_client_ranges(CInode *in, uint64_t size, map<client_t, client_writeable_range_t>& new_ranges);
-  bool check_inode_max_size(CInode *in, bool force_wrlock=false, bool update_size=false, uint64_t newsize=0,
+  bool check_inode_max_size(CInode *in, bool force_wrlock=false,
+                            bool update_size=false, uint64_t newsize=0,
+                            bool update_max=false, uint64_t newmax=0,
 			    utime_t mtime=utime_t());
   void share_inode_max_size(CInode *in, Capability *only_cap=0);
 
diff --git a/src/mds/MDCache.cc b/src/mds/MDCache.cc
index 3129ed7c267..9df91b47961 100644
--- a/src/mds/MDCache.cc
+++ b/src/mds/MDCache.cc
@@ -5667,7 +5667,7 @@ void MDCache::_recovered(CInode *in, int r, uint64_t size, utime_t mtime)
     remove_inode(in);
   } else {
     // journal
-    mds->locker->check_inode_max_size(in, true, true, size, mtime);
+    mds->locker->check_inode_max_size(in, true, true, size, false, 0, mtime);
     in->auth_unpin(this);
   }