mds: be more careful about updating inode.max_size

diff --git a/src/mds/Locker.cc b/src/mds/Locker.cc
index 40aa540ac71680eee08e82716247420df9defb27..248dbb86b3157fd52c3504681b2639a283e12b35 100644 (file)
--- a/src/mds/Locker.cc
+++ b/src/mds/Locker.cc
@@ -482,7 +482,7 @@ void Locker::file_update_finish(CInode *in, LogSegment *ls, bool share)
   in->put(CInode::PIN_PTRWAITER);
   file_wrlock_finish(&in->filelock);
   if (share && in->is_auth() && in->filelock.is_stable())
-    share_new_file_max(in);
+    share_inode_max_size(in);
 }
 
 Capability* Locker::issue_new_caps(CInode *in,
@@ -545,6 +545,8 @@ Capability* Locker::issue_new_caps(CInode *in,
 }
 
 
+
+
 bool Locker::issue_caps(CInode *in)
 {
   // allowed caps are determined by the lock mode.
@@ -556,25 +558,8 @@ bool Locker::issue_caps(CInode *in)
   int nissued = 0;        
 
   // should we increase max_size?
-  if (!in->is_dir() && (allowed & CEPH_CAP_WR)) {
-    inode_t *latest = in->get_projected_inode();
-    if ((latest->size << 1) >= latest->max_size) {
-      int64_t new_max = latest->max_size ? (latest->max_size << 1):in->get_layout_size_increment();
-      dout(10) << "increasing max_size " << latest->max_size << " -> " << new_max << dendl;
-      
-      inode_t *pi = in->project_inode();
-      pi->version = in->pre_dirty();
-      pi->max_size = new_max;
-      EOpen *le = new EOpen(mds->mdlog);
-      le->metablob.add_dir_context(in->get_parent_dir());
-      le->metablob.add_primary_dentry(in->parent, true, 0, pi);
-      LogSegment *ls = mds->mdlog->get_current_segment();
-      le->add_ino(in->ino());
-      ls->open_files.push_back(&in->xlist_open_file);
-      mds->mdlog->submit_entry(le, new C_Locker_FileUpdate_finish(this, in, ls, true));
-      file_wrlock_start(&in->filelock);  // wrlock for duration of journal
-    }
-  }
+  if (!in->is_dir() && (allowed & CEPH_CAP_WR) && in->is_auth())
+    check_inode_max_size(in);
 
   // client caps
   for (map<int, Capability*>::iterator it = in->client_caps.begin();
@@ -778,23 +763,74 @@ void Locker::handle_inode_file_caps(MInodeFileCaps *m)
 }
 
 
+class C_MDL_CheckMaxSize : public Context {
+  Locker *locker;
+  CInode *in;
+public:
+  C_MDL_CheckMaxSize(Locker *l, CInode *i) : locker(l), in(i) {
+    in->get(CInode::PIN_PTRWAITER);
+  }
+  void finish(int r) {
+    in->put(CInode::PIN_PTRWAITER);
+    if (in->is_auth())
+      locker->check_inode_max_size(in);
+  }
+};
 
 
-void Locker::share_new_file_max(CInode *in)
+void Locker::check_inode_max_size(CInode *in)
+{
+  assert(in->is_auth());
+  if (!in->filelock.can_wrlock()) {
+    // try again later
+    in->filelock.add_waiter(SimpleLock::WAIT_WR, new C_MDL_CheckMaxSize(this, in));
+    dout(10) << "check_inode_max_size can't wrlock, waiting on " << *in << dendl;
+    return;    
+  }
+
+  inode_t *latest = in->get_projected_inode();
+  int64_t new_max;
+  if (!in->is_any_caps())
+    new_max = 0;
+  else if ((latest->size << 1) >= latest->max_size)
+    new_max = latest->max_size ? (latest->max_size << 1):in->get_layout_size_increment();
+  else
+    return;  // nothing.
+  if (new_max == latest->max_size)
+    return;  // no change.
+
+  dout(10) << "check_inode_max_size " << latest->max_size << " -> " << new_max
+          << " on " << *in << dendl;
+    
+  inode_t *pi = in->project_inode();
+  pi->version = in->pre_dirty();
+  pi->max_size = new_max;
+  EOpen *le = new EOpen(mds->mdlog);
+  le->metablob.add_dir_context(in->get_parent_dir());
+  le->metablob.add_primary_dentry(in->parent, true, 0, pi);
+  LogSegment *ls = mds->mdlog->get_current_segment();
+  le->add_ino(in->ino());
+  ls->open_files.push_back(&in->xlist_open_file);
+  mds->mdlog->submit_entry(le, new C_Locker_FileUpdate_finish(this, in, ls, true));
+  file_wrlock_start(&in->filelock);  // wrlock for duration of journal
+}
+
+
+void Locker::share_inode_max_size(CInode *in)
 {
   /*
    * only share if currently issued a WR cap.  if client doesn't have it,
    * file_max doesn't matter, and the client will get it if/when they get
    * the cap later.
    */
-  dout(10) << "share_new_file_max on " << *in << dendl;
+  dout(10) << "share_inode_max_size on " << *in << dendl;
   for (map<int,Capability*>::iterator it = in->client_caps.begin();
        it != in->client_caps.end();
        it++) {
     const int client = it->first;
     Capability *cap = it->second;
     if (cap->pending() & CEPH_CAP_WR) {
-      dout(10) << "share_new_file_max with client" << client << dendl;
+      dout(10) << "share_inode_max_size with client" << client << dendl;
       mds->send_message_client(new MClientFileCaps(CEPH_CAP_OP_GRANT,
                                                   in->inode,
                                                   cap->get_last_seq(),
@@ -874,11 +910,6 @@ void Locker::handle_client_file_caps(MClientFileCaps *m)
 
   inode_t *latest = in->get_projected_inode();
 
-  // no more writers?
-  bool no_wr = false;
-  if (latest->max_size && (wanted & (CEPH_CAP_WR|CEPH_CAP_WRBUFFER)) == 0)
-    no_wr = true;
-
   utime_t atime = m->get_atime();
   utime_t mtime = m->get_mtime();
   utime_t ctime = m->get_ctime();
@@ -903,34 +934,43 @@ void Locker::handle_client_file_caps(MClientFileCaps *m)
     dirty_atime = true;
   bool dirty = dirty_atime || dirty_mtime || dirty_ctime || dirty_size;
   
-  // increase max_size?
-  bool increase_max = false;
+  // increase or zero max_size?
+  bool change_max = false;
   int64_t new_max = latest->max_size;
-  if ((wanted & (CEPH_CAP_WR|CEPH_CAP_WRBUFFER|CEPH_CAP_WREXTEND)) &&
-      (size << 1) >= latest->max_size &&
-      in->filelock.can_wrlock()) {
-    dout(10) << "wr caps wanted, and size " << size
-            << " *2 >= max " << latest->max_size << ", increasing" << dendl;
-    increase_max = true;
-    new_max = latest->max_size ? (latest->max_size << 1):in->get_layout_size_increment();
-  }
-  if ((wanted & (CEPH_CAP_WR|CEPH_CAP_WRBUFFER|CEPH_CAP_WREXTEND)) &&
-      m->get_max_size() > new_max) {
-    dout(10) << "client requests file_max " << m->get_max_size()
-            << " > max " << latest->max_size << dendl;
-    increase_max = true;
-    new_max = (m->get_max_size() << 1) & ~(in->get_layout_size_increment() - 1);
+
+  if (in->is_auth()) {
+    if (latest->max_size && (wanted & (CEPH_CAP_WR|CEPH_CAP_WRBUFFER)) == 0) {
+      change_max = true;
+      new_max = 0;
+    }
+    else if ((wanted & (CEPH_CAP_WR|CEPH_CAP_WRBUFFER|CEPH_CAP_WREXTEND)) &&
+       (size << 1) >= latest->max_size) {
+      dout(10) << "wr caps wanted, and size " << size
+              << " *2 >= max " << latest->max_size << ", increasing" << dendl;
+      change_max = true;
+      new_max = latest->max_size ? (latest->max_size << 1):in->get_layout_size_increment();
+    }
+    if ((wanted & (CEPH_CAP_WR|CEPH_CAP_WRBUFFER|CEPH_CAP_WREXTEND)) &&
+       m->get_max_size() > new_max) {
+      dout(10) << "client requests file_max " << m->get_max_size()
+              << " > max " << latest->max_size << dendl;
+      change_max = true;
+      new_max = (m->get_max_size() << 1) & ~(in->get_layout_size_increment() - 1);
+    }
+
+    if (change_max && !in->filelock.can_wrlock()) {
+      dout(10) << "want to change file_max, but lock won't allow it; will retry" << dendl;
+      check_inode_max_size(in);  // this will fail, and schedule a waiter.
+      change_max = false;
+    }
   }
 
-  if ((dirty || no_wr || increase_max) &&
+  if ((dirty || change_max) &&
       !in->is_base()) {              // FIXME.. what about root inode mtime/atime?
     EUpdate *le = new EUpdate(mds->mdlog, "size|max_size|mtime|ctime|atime update");
     inode_t *pi = in->project_inode();
     pi->version = in->pre_dirty();
-    if (no_wr) {
-      dout(7) << " max_size " << pi->max_size << " -> 0 (last writer released)" << dendl;
-      pi->max_size = 0;
-    } else if (increase_max) {
+    if (change_max) {
       dout(7) << " max_size " << pi->max_size << " -> " << new_max << dendl;
       pi->max_size = new_max;
     }    
@@ -957,7 +997,7 @@ void Locker::handle_client_file_caps(MClientFileCaps *m)
     le->metablob.add_dir_context(in->get_parent_dir());
     le->metablob.add_primary_dentry(in->parent, true, 0, pi);
     LogSegment *ls = mds->mdlog->get_current_segment();
-    mds->mdlog->submit_entry(le, new C_Locker_FileUpdate_finish(this, in, ls, increase_max));
+    mds->mdlog->submit_entry(le, new C_Locker_FileUpdate_finish(this, in, ls, change_max));
     file_wrlock_start(&in->filelock);  // wrlock for duration of journal
   }
 

diff --git a/src/mds/Locker.h b/src/mds/Locker.h
index 0d32e9883065d79a60aae0886db413e61f92b6de..de43036e0c95f2708fb8dfef3c609a4ea20f8178 100644 (file)
--- a/src/mds/Locker.h
+++ b/src/mds/Locker.h
@@ -196,8 +196,10 @@ protected:
   void handle_inode_file_caps(class MInodeFileCaps *m);
 
   void file_update_finish(CInode *in, LogSegment *ls, bool share);
-  void share_new_file_max(CInode *in);
+  void check_inode_max_size(CInode *in);
+  void share_inode_max_size(CInode *in);
 
+  friend class C_MDL_CheckMaxSize;
   friend class C_MDL_RequestInodeFileCaps;
   friend class C_Locker_FileUpdate_finish;