From 2a45bcad02d6ec2196961281c369ca762c49fdb6 Mon Sep 17 00:00:00 2001
From: Sage Weil <sage@newdream.net>
Date: Thu, 9 Jun 2011 10:41:53 -0700
Subject: [PATCH] mds: issue caps from drop_locks

In drop_locks, build a set of inodes we need to issue caps on.  Then do it
all at once.  This does two things:

 - it fixes the fact that currently a dropped lock leading to an eval and
   lock state change will not issue caps _at_all_
 - it ensure we only issue_caps once for each inode, even when we are
   dropping multiple locks on it.

This should fix #1084.

Signed-off-by: Sage Weil <sage@newdream.net>
---
 src/mds/Locker.cc | 51 +++++++++++++++++++++++++++++++++++++----------
 src/mds/Locker.h  |  5 +++--
 2 files changed, 44 insertions(+), 12 deletions(-)

diff --git a/src/mds/Locker.cc b/src/mds/Locker.cc
index f5a5fa60904cf..a0370e76b5a60 100644
--- a/src/mds/Locker.cc
+++ b/src/mds/Locker.cc
@@ -409,15 +409,37 @@ void Locker::set_xlocks_done(Mutation *mut)
   }
 }
 
-void Locker::drop_locks(Mutation *mut)
+void Locker::drop_locks(Mutation *mut, set<CInode*> *pneed_issue)
 {
   // leftover locks
-  while (!mut->xlocks.empty()) 
-    xlock_finish(*mut->xlocks.begin(), mut);
-  while (!mut->rdlocks.empty()) 
-    rdlock_finish(*mut->rdlocks.begin(), mut);
-  while (!mut->wrlocks.empty()) 
-    wrlock_finish(*mut->wrlocks.begin(), mut);
+  set<CInode*> my_need_issue;
+  if (!pneed_issue)
+    pneed_issue = &my_need_issue;
+
+  while (!mut->xlocks.empty()) {
+    bool ni = false;
+    MDSCacheObject *p = (*mut->xlocks.begin())->get_parent();
+    xlock_finish(*mut->xlocks.begin(), mut, &ni);
+    if (ni)
+      pneed_issue->insert((CInode*)p);
+  }
+  while (!mut->rdlocks.empty()) {
+    bool ni = false;
+    MDSCacheObject *p = (*mut->rdlocks.begin())->get_parent();
+    rdlock_finish(*mut->rdlocks.begin(), mut, &ni);
+    if (ni)
+      pneed_issue->insert((CInode*)p);
+  }	 
+  while (!mut->wrlocks.empty()) {
+    bool ni = false;
+    MDSCacheObject *p = (*mut->wrlocks.begin())->get_parent();
+    wrlock_finish(*mut->wrlocks.begin(), mut, &ni);
+    if (ni)
+      pneed_issue->insert((CInode*)p);
+  }
+
+  if (pneed_issue == &my_need_issue)
+    issue_caps_set(*pneed_issue);
   mut->done_locking = false;
 }
 
@@ -714,7 +736,7 @@ void Locker::try_eval(CInode *in, int mask)
   eval(in, mask);
 }
 
-void Locker::eval_cap_gather(CInode *in)
+void Locker::eval_cap_gather(CInode *in, set<CInode*> *issue_set)
 {
   bool need_issue = false;
   list<Context*> finishers;
@@ -729,8 +751,12 @@ void Locker::eval_cap_gather(CInode *in)
   if (!in->xattrlock.is_stable())
     eval_gather(&in->xattrlock, false, &need_issue, &finishers);
 
-  if (need_issue && in->is_head())
-    issue_caps(in);
+  if (need_issue && in->is_head()) {
+    if (issue_set)
+      issue_set->insert(in);
+    else
+      issue_caps(in);
+  }
 
   finish_contexts(finishers);
 }
@@ -1346,6 +1372,11 @@ Capability* Locker::issue_new_caps(CInode *in,
 }
 
 
+void Locker::issue_caps_set(set<CInode*>& inset)
+{
+  for (set<CInode*>::iterator p = inset.begin(); p != inset.end(); ++p)
+    issue_caps(*p);
+}
 
 bool Locker::issue_caps(CInode *in, Capability *only_cap)
 {
diff --git a/src/mds/Locker.h b/src/mds/Locker.h
index 2c849dc2a253b..61f501b479b85 100644
--- a/src/mds/Locker.h
+++ b/src/mds/Locker.h
@@ -87,7 +87,7 @@ public:
 		     set<SimpleLock*> &wrlocks,
 		     set<SimpleLock*> &xlocks);
 
-  void drop_locks(Mutation *mut);
+  void drop_locks(Mutation *mut, set<CInode*> *pneed_issue=0);
   void set_xlocks_done(Mutation *mut);
   void drop_non_rdlocks(Mutation *mut);
   void drop_rdlocks(Mutation *mut);
@@ -115,7 +115,7 @@ public:
   };
   void eval_scatter_gathers(CInode *in);
 
-  void eval_cap_gather(CInode *in);
+  void eval_cap_gather(CInode *in, set<CInode*> *issue_set=0);
 
   bool eval(CInode *in, int mask);
   void try_eval(CInode *in, int mask);
@@ -248,6 +248,7 @@ public:
   version_t issue_file_data_version(CInode *in);
   Capability* issue_new_caps(CInode *in, int mode, Session *session, SnapRealm *conrealm, bool is_replay);
   bool issue_caps(CInode *in, Capability *only_cap=0);
+  void issue_caps_set(set<CInode*>& inset);
   void issue_truncate(CInode *in);
   void revoke_stale_caps(Session *session);
   void resume_stale_caps(Session *session);
-- 
2.39.5