From 91fe2a91728cf94de2fb767cf5b5822f755b5314 Mon Sep 17 00:00:00 2001
From: Sage Weil <sage@newdream.net>
Date: Sun, 30 Nov 2008 13:06:37 -0800
Subject: [PATCH] osd: fix lock inversion on workqueue shutdown

Simplify PG get/put vs lock/unlock.  Since we are reference counting with
an atomic_t, we don't need to re-use PG::lock to protect the reference
count.
---
 src/osd/OSD.cc          | 12 ++++++++----
 src/osd/OSD.h           | 11 ++++-------
 src/osd/PG.cc           |  3 ++-
 src/osd/PG.h            | 11 ++---------
 src/osd/ReplicatedPG.cc |  6 ++++--
 5 files changed, 20 insertions(+), 23 deletions(-)

diff --git a/src/osd/OSD.cc b/src/osd/OSD.cc
index de638d273c113..89cdc0023663c 100644
--- a/src/osd/OSD.cc
+++ b/src/osd/OSD.cc
@@ -534,7 +534,8 @@ int OSD::shutdown()
        p++) {
     PG *pg = p->second;
     pg->lock();
-    pg->put_unlock();
+    pg->unlock();
+    pg->put();
   }
   pg_map.clear();
 
@@ -724,7 +725,8 @@ void OSD::_remove_unlock_pg(PG *pg)
   pg_map.erase(pgid);
 
   // unlock, and probably delete
-  pg->put_unlock();     // will delete, if last reference
+  pg->unlock();
+  pg->put();  // will delete, if last reference
   dout(10) << "_remove_unlock_pg " << pgid << " done" << dendl;
 }
 
@@ -3021,7 +3023,8 @@ void OSD::do_recovery(PG *pg)
   if (started < max)
     pg->recovery_item.remove_myself();
 
-  pg->put_unlock();
+  pg->unlock();
+  pg->put();
 }
 
 
@@ -3392,7 +3395,8 @@ void OSD::dequeue_op(PG *pg)
     assert(0);
 
   // unlock and put pg
-  pg->put_unlock();
+  pg->unlock();
+  pg->put();
   
   //#warning foo
   //scrub_wq.queue(pg);
diff --git a/src/osd/OSD.h b/src/osd/OSD.h
index 53727c450b566..1a1b3e6f5ba71 100644
--- a/src/osd/OSD.h
+++ b/src/osd/OSD.h
@@ -459,9 +459,8 @@ private:
     void _clear() {
       while (!osd->recovery_queue.empty()) {
 	PG *pg = osd->recovery_queue.front();
-	pg->lock();
-	pg->put_unlock();
 	osd->recovery_queue.pop_front();
+	pg->put();
       }
     }
   } recovery_wq;
@@ -524,9 +523,8 @@ private:
     void _clear() {
       while (!osd->snap_trim_queue.empty()) {
 	PG *pg = osd->snap_trim_queue.front();
-	pg->lock();
-	pg->put_unlock();
 	osd->snap_trim_queue.pop_front();
+	pg->put();
       }
     }
   } snap_trim_wq;
@@ -559,11 +557,10 @@ private:
       pg->scrub();
     }
     void _clear() {
-       while (!osd->scrub_queue.empty()) {
+      while (!osd->scrub_queue.empty()) {
 	PG *pg = osd->scrub_queue.front();
-	pg->lock();
-	pg->put_unlock();
 	osd->scrub_queue.pop_front();
+	pg->put();
       }
     }
   } scrub_wq;
diff --git a/src/osd/PG.cc b/src/osd/PG.cc
index 864db6ea0f8f8..f6ff2807262a0 100644
--- a/src/osd/PG.cc
+++ b/src/osd/PG.cc
@@ -1356,7 +1356,8 @@ void PG::_finish_recovery(Context *c)
   }
   osd->map_lock.put_read();
   osd->finish_recovery_op(this, recovery_ops_active, true);
-  put_unlock();
+  unlock();
+  put();
 }
 
 void PG::defer_recovery()
diff --git a/src/osd/PG.h b/src/osd/PG.h
index 09086ded34ccb..42bed39f15563 100644
--- a/src/osd/PG.h
+++ b/src/osd/PG.h
@@ -509,15 +509,8 @@ public:
   void put() { 
     //generic_dout(0) << this << " " << info.pgid << " put " << ref.test() << dendl;
     assert(_lock.is_locked());
-    ref.dec();
-    assert(ref.test() > 0);  // last put must be a put_unlock.
-  }
-  void put_unlock() { 
-    //generic_dout(0) << this << " " << info.pgid << " put_unlock " << ref.test() << dendl;
-    assert(_lock.is_locked());
-    int last = ref.dec();
-    _lock.Unlock();
-    if (last == 0) delete this;
+    if (ref.dec() == 0)
+      delete this;
   }
 
 
diff --git a/src/osd/ReplicatedPG.cc b/src/osd/ReplicatedPG.cc
index fb2757f6a6189..395f48aa27c68 100644
--- a/src/osd/ReplicatedPG.cc
+++ b/src/osd/ReplicatedPG.cc
@@ -1198,7 +1198,8 @@ public:
     if (!pg->is_deleted()) 
       pg->op_modify_ondisk(repop);
     repop->put();
-    pg->put_unlock();
+    pg->unlock();
+    pg->put();
   }
 };
 
@@ -1641,7 +1642,8 @@ public:
 
     pg->lock();
     pg->sub_op_modify_ondisk(op, destosd, pg_last_complete);
-    pg->put_unlock();
+    pg->unlock();
+    pg->put();
   }
   void ack() {
     lock.Lock();
-- 
2.39.5