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.

diff --git a/src/osd/OSD.cc b/src/osd/OSD.cc
index de638d273c113109e5aff679769f18918a998507..89cdc0023663c6aba5906e03a7b356f7fa64f47a 100644 (file)
--- 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 53727c450b56694eac462554ea7091344b333bcc..1a1b3e6f5ba71146930e646f933480fa695c5979 100644 (file)
--- 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 864db6ea0f8f8d0d6b72af50404a43bee5f3b72e..f6ff2807262a07539ad197de3306ec94e18a4a8f 100644 (file)
--- 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 09086ded34ccb43810a0906a0aaeff506ad10c6f..42bed39f15563d8710b5588f3fe01fd32d0028f8 100644 (file)
--- 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 fb2757f6a6189ce769e9593f15172a2b22a91e54..395f48aa27c68853e03f919e68bd71fa3af15eec 100644 (file)
--- 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();