osd/: optionally track every pg ref

This involves three pieces:

For intrusive_ptr type references, we use TrackedIntPtr instead.  This
uses get_with_id and put_with_id to associate an id and backtrace with
each particular ref instance.

For refs taken via direct calls to get() and put(), get and put now
require a tag string.  The PG tracks individual ref counts for each tag
as well as the total.

Finally, PGs register/unregister themselves on construction/destruction
with OSDService.

As a result, on shutdown, we can check for live pgs and determine where
the references are held.

This behavior is compiled out by default, but can be included with the
--enable-pgrefdebugging flag.

Signed-off-by: Samuel Just <sam.just@inktank.com>

7 files changed, 220 insertions, 49 deletions

diff --git a/configure.ac b/configure.ac
index 23e21133e6a..2661d1c456e 100644
--- a/configure.ac
+++ b/configure.ac
@@ -268,6 +268,12 @@ AS_IF([test "x$with_tcmalloc" != xno],
 		  [no tcmalloc found (use --without-tcmalloc to disable)])])])
 AM_CONDITIONAL(WITH_TCMALLOC, [test "$HAVE_LIBTCMALLOC" = "1"])
 
+#set pg ref debugging?
+AC_ARG_ENABLE([pgrefdebugging],
+	    [AS_HELP_STRING([--enable-pgrefdebugging], [enable pg ref debugging])],
+			[AC_DEFINE([PG_DEBUG_REFS], [1], [Defined if you want pg ref debugging])],
+			[])
+
 #
 # Java is painful
 #   - adapted from OMPI wrappers package
diff --git a/src/osd/OSD.cc b/src/osd/OSD.cc
index c61ad0b2d62..5e335422579 100644
--- a/src/osd/OSD.cc
+++ b/src/osd/OSD.cc
@@ -189,6 +189,9 @@ OSDService::OSDService(OSD *osd) :
   cur_ratio(0),
   is_stopping_lock("OSDService::is_stopping_lock"),
   state(NOT_STOPPING)
+#ifdef PG_DEBUG_REFS
+  , pgid_lock("OSDService::pgid_lock")
+#endif
 {}
 
 void OSDService::_start_split(const set<pg_t> &pgs)
@@ -1217,7 +1220,7 @@ int OSD::shutdown()
   g_ceph_context->_conf->set_val("debug_ms", "100");
   g_ceph_context->_conf->apply_changes(NULL);
   
-  // Remove PGs
+  // Shutdown PGs
   for (hash_map<pg_t, PG*>::iterator p = pg_map.begin();
        p != pg_map.end();
        ++p) {
@@ -1227,9 +1230,7 @@ int OSD::shutdown()
     p->second->kick();
     p->second->unlock();
     p->second->osr->flush();
-    p->second->put();
   }
-  pg_map.clear();
   
   // finish ops
   op_wq.drain(); // should already be empty except for lagard PGs
@@ -1308,6 +1309,28 @@ int OSD::shutdown()
     assert(pg_stat_queue.empty());
   }
 
+  peering_wq.clear();
+  // Remove PGs
+#ifdef PG_DEBUG_REFS
+  service.dump_live_pgids();
+#endif
+  for (hash_map<pg_t, PG*>::iterator p = pg_map.begin();
+       p != pg_map.end();
+       ++p) {
+    dout(20) << " kicking pg " << p->first << dendl;
+    p->second->lock();
+    if (p->second->ref.read() != 1) {
+      derr << "pgid " << p->first << " has ref count of "
+	   << p->second->ref.read() << dendl;
+      assert(0);
+    }
+    p->second->unlock();
+    p->second->put("PGMap");
+  }
+  pg_map.clear();
+#ifdef PG_DEBUG_REFS
+  service.dump_live_pgids();
+#endif
   g_conf->remove_observer(this);
 
   monc->shutdown();
@@ -1321,6 +1344,7 @@ int OSD::shutdown()
   cluster_messenger->shutdown();
   hbclient_messenger->shutdown();
   hbserver_messenger->shutdown();
+  peering_wq.clear();
   return r;
 }
 
@@ -1440,7 +1464,7 @@ PG *OSD::_open_lock_pg(
     pg->lock_with_map_lock_held(no_lockdep_check);
   else
     pg->lock(no_lockdep_check);
-  pg->get();  // because it's in pg_map
+  pg->get("PGMap");  // because it's in pg_map
   return pg;
 }
 
@@ -1467,7 +1491,7 @@ PG* OSD::_make_pg(
 void OSD::add_newly_split_pg(PG *pg, PG::RecoveryCtx *rctx)
 {
   epoch_t e(service.get_osdmap()->get_epoch());
-  pg->get();  // For pg_map
+  pg->get("PGMap");  // For pg_map
   pg_map[pg->info.pgid] = pg;
   dout(10) << "Adding newly split pg " << *pg << dendl;
   vector<int> up, acting;
@@ -2981,7 +3005,7 @@ void OSD::send_pg_stats(const utime_t &now)
       ++p;
       if (!pg->is_primary()) {  // we hold map_lock; role is stable.
 	pg->stat_queue_item.remove_myself();
-	pg->put();
+	pg->put("pg_stat_queue");
 	continue;
       }
       pg->pg_stats_lock.Lock();
@@ -3034,7 +3058,7 @@ void OSD::handle_pg_stats_ack(MPGStatsAck *ack)
       if (acked == pg->pg_stats_stable.reported) {
 	dout(25) << " ack on " << pg->info.pgid << " " << pg->pg_stats_stable.reported << dendl;
 	pg->stat_queue_item.remove_myself();
-	pg->put();
+	pg->put("pg_stat_queue");
       } else {
 	dout(25) << " still pending " << pg->info.pgid << " " << pg->pg_stats_stable.reported
 		 << " > acked " << acked << dendl;
@@ -5850,7 +5874,7 @@ void OSD::_remove_pg(PG *pg)
 
   // remove from map
   pg_map.erase(pg->info.pgid);
-  pg->put(); // since we've taken it out of map
+  pg->put("PGMap"); // since we've taken it out of map
 }
 
 
diff --git a/src/osd/OSD.h b/src/osd/OSD.h
index 5166ae74aa4..8a74cd8b630 100644
--- a/src/osd/OSD.h
+++ b/src/osd/OSD.h
@@ -427,6 +427,41 @@ public:
   bool prepare_to_stop();
   void got_stop_ack();
 
+
+#ifdef PG_DEBUG_REFS
+  Mutex pgid_lock;
+  map<pg_t, int> pgid_tracker;
+  map<pg_t, PG*> live_pgs;
+  void add_pgid(pg_t pgid, PG *pg) {
+    Mutex::Locker l(pgid_lock);
+    if (!pgid_tracker.count(pgid)) {
+      pgid_tracker[pgid] = 0;
+      live_pgs[pgid] = pg;
+    }
+    pgid_tracker[pgid]++;
+  }
+  void remove_pgid(pg_t pgid, PG *pg) {
+    Mutex::Locker l(pgid_lock);
+    assert(pgid_tracker.count(pgid));
+    assert(pgid_tracker[pgid] > 0);
+    pgid_tracker[pgid]--;
+    if (pgid_tracker[pgid] == 0) {
+      pgid_tracker.erase(pgid);
+      live_pgs.erase(pgid);
+    }
+  }
+  void dump_live_pgids() {
+    Mutex::Locker l(pgid_lock);
+    derr << "live pgids:" << dendl;
+    for (map<pg_t, int>::iterator i = pgid_tracker.begin();
+	 i != pgid_tracker.end();
+	 ++i) {
+      derr << "\t" << *i << dendl;
+      live_pgs[i->first]->dump_live_ids();
+    }
+  }
+#endif
+
   OSDService(OSD *osd);
 };
 class OSD : public Dispatcher,
@@ -759,14 +794,14 @@ private:
 	   ) {
 	if (*i == pg) {
 	  peering_queue.erase(i++);
-	  pg->put();
+	  pg->put("PeeringWQ");
 	} else {
 	  ++i;
 	}
       }
     }
     bool _enqueue(PG *pg) {
-      pg->get();
+      pg->get("PeeringWQ");
       peering_queue.push_back(pg);
       return true;
     }
@@ -795,7 +830,7 @@ private:
       for (list<PG *>::const_iterator i = pgs.begin();
 	   i != pgs.end();
 	   ++i) {
-	(*i)->put();
+	(*i)->put("PeeringWQ");
       }
     }
     void _process_finish(const list<PG *> &pgs) {
@@ -1019,7 +1054,7 @@ protected:
   void pg_stat_queue_enqueue(PG *pg) {
     pg_stat_queue_lock.Lock();
     if (pg->is_primary() && !pg->stat_queue_item.is_on_list()) {
-      pg->get();
+      pg->get("pg_stat_queue");
       pg_stat_queue.push_back(&pg->stat_queue_item);
     }
     osd_stat_updated = true;
@@ -1028,7 +1063,7 @@ protected:
   void pg_stat_queue_dequeue(PG *pg) {
     pg_stat_queue_lock.Lock();
     if (pg->stat_queue_item.remove_myself())
-      pg->put();
+      pg->put("pg_stat_queue");
     pg_stat_queue_lock.Unlock();
   }
   void clear_pg_stat_queue() {
@@ -1036,7 +1071,7 @@ protected:
     while (!pg_stat_queue.empty()) {
       PG *pg = pg_stat_queue.front();
       pg_stat_queue.pop_front();
-      pg->put();
+      pg->put("pg_stat_queue");
     }
     pg_stat_queue_lock.Unlock();
   }
@@ -1159,7 +1194,7 @@ protected:
     }
     bool _enqueue(PG *pg) {
       if (!pg->recovery_item.is_on_list()) {
-	pg->get();
+	pg->get("RecoveryWQ");
 	osd->recovery_queue.push_back(&pg->recovery_item);
 
 	if (g_conf->osd_recovery_delay_start > 0) {
@@ -1172,7 +1207,7 @@ protected:
     }
     void _dequeue(PG *pg) {
       if (pg->recovery_item.remove_myself())
-	pg->put();
+	pg->put("RecoveryWQ");
     }
     PG *_dequeue() {
       if (osd->recovery_queue.empty())
@@ -1187,19 +1222,19 @@ protected:
     }
     void _queue_front(PG *pg) {
       if (!pg->recovery_item.is_on_list()) {
-	pg->get();
+	pg->get("RecoveryWQ");
 	osd->recovery_queue.push_front(&pg->recovery_item);
       }
     }
     void _process(PG *pg) {
       osd->do_recovery(pg);
-      pg->put();
+      pg->put("RecoveryWQ");
     }
     void _clear() {
       while (!osd->recovery_queue.empty()) {
 	PG *pg = osd->recovery_queue.front();
 	osd->recovery_queue.pop_front();
-	pg->put();
+	pg->put("RecoveryWQ");
       }
     }
   } recovery_wq;
@@ -1231,13 +1266,13 @@ protected:
     bool _enqueue(PG *pg) {
       if (pg->snap_trim_item.is_on_list())
 	return false;
-      pg->get();
+      pg->get("SnapTrimWQ");
       osd->snap_trim_queue.push_back(&pg->snap_trim_item);
       return true;
     }
     void _dequeue(PG *pg) {
       if (pg->snap_trim_item.remove_myself())
-	pg->put();
+	pg->put("SnapTrimWQ");
     }
     PG *_dequeue() {
       if (osd->snap_trim_queue.empty())
@@ -1248,7 +1283,7 @@ protected:
     }
     void _process(PG *pg) {
       pg->snap_trimmer();
-      pg->put();
+      pg->put("SnapTrimWQ");
     }
     void _clear() {
       osd->snap_trim_queue.clear();
@@ -1275,13 +1310,13 @@ protected:
       if (pg->scrub_item.is_on_list()) {
 	return false;
       }
-      pg->get();
+      pg->get("ScrubWQ");
       osd->scrub_queue.push_back(&pg->scrub_item);
       return true;
     }
     void _dequeue(PG *pg) {
       if (pg->scrub_item.remove_myself()) {
-	pg->put();
+	pg->put("ScrubWQ");
       }
     }
     PG *_dequeue() {
@@ -1293,13 +1328,13 @@ protected:
     }
     void _process(PG *pg) {
       pg->scrub();
-      pg->put();
+      pg->put("ScrubWQ");
     }
     void _clear() {
       while (!osd->scrub_queue.empty()) {
 	PG *pg = osd->scrub_queue.front();
 	osd->scrub_queue.pop_front();
-	pg->put();
+	pg->put("ScrubWQ");
       }
     }
   } scrub_wq;
@@ -1320,13 +1355,13 @@ protected:
       if (pg->scrub_finalize_item.is_on_list()) {
 	return false;
       }
-      pg->get();
+      pg->get("ScrubFinalizeWQ");
       scrub_finalize_queue.push_back(&pg->scrub_finalize_item);
       return true;
     }
     void _dequeue(PG *pg) {
       if (pg->scrub_finalize_item.remove_myself()) {
-	pg->put();
+	pg->put("ScrubFinalizeWQ");
       }
     }
     PG *_dequeue() {
@@ -1338,13 +1373,13 @@ protected:
     }
     void _process(PG *pg) {
       pg->scrub_finalize();
-      pg->put();
+      pg->put("ScrubFinalizeWQ");
     }
     void _clear() {
       while (!scrub_finalize_queue.empty()) {
 	PG *pg = scrub_finalize_queue.front();
 	scrub_finalize_queue.pop_front();
-	pg->put();
+	pg->put("ScrubFinalizeWQ");
       }
     }
   } scrub_finalize_wq;
diff --git a/src/osd/PG.cc b/src/osd/PG.cc
index 201d2d93d63..a2fdecc89ae 100644
--- a/src/osd/PG.cc
+++ b/src/osd/PG.cc
@@ -33,6 +33,7 @@
 
 #include "messages/MOSDSubOp.h"
 #include "messages/MOSDSubOpReply.h"
+#include "common/BackTrace.h"
 
 #include <sstream>
 
@@ -43,6 +44,73 @@ static ostream& _prefix(std::ostream *_dout, const PG *pg) {
   return *_dout << pg->gen_prefix();
 }
 
+void PG::get(const string &tag) {
+  ref.inc();
+#ifdef PG_DEBUG_REFS
+  Mutex::Locker l(_ref_id_lock);
+  if (!_tag_counts.count(tag)) {
+    _tag_counts[tag] = 0;
+  }
+  _tag_counts[tag]++;
+#endif
+}
+void PG::put(const string &tag) {
+#ifdef PG_DEBUG_REFS
+  {
+    Mutex::Locker l(_ref_id_lock);
+    assert(_tag_counts.count(tag));
+    _tag_counts[tag]--;
+    if (_tag_counts[tag] == 0) {
+      _tag_counts.erase(tag);
+    }
+  }
+#endif
+  if (ref.dec() == 0)
+    delete this;
+}
+
+#ifdef PG_DEBUG_REFS
+uint64_t PG::get_with_id() {
+  ref.inc();
+  Mutex::Locker l(_ref_id_lock);
+  uint64_t id = ++_ref_id;
+  BackTrace bt(0);
+  stringstream ss;
+  bt.print(ss);
+  dout(20) << __func__ << ": " << info.pgid << " got id " << id << dendl;
+  assert(!_live_ids.count(id));
+  _live_ids.insert(make_pair(id, ss.str()));
+  return id;
+}
+
+void PG::put_with_id(uint64_t id) {
+  dout(20) << __func__ << ": " << info.pgid << " put id " << id << dendl;
+  {
+    Mutex::Locker l(_ref_id_lock);
+    assert(_live_ids.count(id));
+    _live_ids.erase(id);
+  }
+  if (ref.dec() == 0)
+    delete this;
+}
+
+void PG::dump_live_ids() {
+  Mutex::Locker l(_ref_id_lock);
+  dout(0) << "\t" << __func__ << ": " << info.pgid << " live ids:" << dendl;
+  for (map<uint64_t, string>::iterator i = _live_ids.begin();
+       i != _live_ids.end();
+       ++i) {
+    dout(0) << "\t\tid: " << *i << dendl;
+  }
+  dout(0) << "\t" << __func__ << ": " << info.pgid << " live tags:" << dendl;
+  for (map<string, uint64_t>::iterator i = _tag_counts.begin();
+       i != _tag_counts.end();
+       ++i) {
+    dout(0) << "\t\tid: " << *i << dendl;
+  }
+}
+#endif
+
 void PGPool::update(OSDMapRef map)
 {
   const pg_pool_t *pi = map->get_pg_pool(id);
@@ -72,7 +140,11 @@ PG::PG(OSDService *o, OSDMapRef curmap,
     _pool.id),
   osdmap_ref(curmap), pool(_pool),
   _lock("PG::_lock"),
-  ref(0), deleting(false), dirty_info(false), dirty_big_info(false), dirty_log(false),
+  ref(0),
+  #ifdef PG_DEBUG_REFS
+  _ref_id_lock("PG::_ref_id_lock"), _ref_id(0),
+  #endif
+  deleting(false), dirty_info(false), dirty_big_info(false), dirty_log(false),
   info(p),
   info_struct_v(0),
   coll(p), log_oid(loid), biginfo_oid(ioid),
@@ -98,10 +170,16 @@ PG::PG(OSDService *o, OSDMapRef curmap,
   active_pushes(0),
   recovery_state(this)
 {
+#ifdef PG_DEBUG_REFS
+  osd->add_pgid(p, this);
+#endif
 }
 
 PG::~PG()
 {
+#ifdef PG_DEBUG_REFS
+  osd->remove_pgid(info.pgid, this);
+#endif
 }
 
 void PG::lock(bool no_lockdep)
@@ -7630,5 +7708,10 @@ bool PG::PriorSet::affected_by_map(const OSDMapRef osdmap, const PG *debug_pg) c
   return false;
 }
 
-void intrusive_ptr_add_ref(PG *pg) { pg->get(); }
-void intrusive_ptr_release(PG *pg) { pg->put(); }
+void intrusive_ptr_add_ref(PG *pg) { pg->get("intptr"); }
+void intrusive_ptr_release(PG *pg) { pg->put("intptr"); }
+
+#ifdef PG_DEBUG_REFS
+  uint64_t get_with_id(PG *pg) { return pg->get_with_id(); }
+  void put_with_id(PG *pg, uint64_t id) { return pg->put_with_id(id); }
+#endif
diff --git a/src/osd/PG.h b/src/osd/PG.h
index 716ac2b346a..3ac51b87b94 100644
--- a/src/osd/PG.h
+++ b/src/osd/PG.h
@@ -42,6 +42,7 @@
 #include "msg/Messenger.h"
 #include "messages/MOSDRepScrub.h"
 #include "messages/MOSDPGLog.h"
+#include "common/tracked_int_ptr.hpp"
 
 #include <list>
 #include <memory>
@@ -69,7 +70,13 @@ class PG;
 void intrusive_ptr_add_ref(PG *pg);
 void intrusive_ptr_release(PG *pg);
 
+#ifdef PG_DEBUG_REFS
+  uint64_t get_with_id(PG *pg);
+  void put_with_id(PG *pg, uint64_t id);
+  typedef TrackedIntPtr<PG> PGRef;
+#else
   typedef boost::intrusive_ptr<PG> PGRef;
+#endif
 
 struct PGRecoveryStats {
   struct per_state_info {
@@ -394,6 +401,13 @@ protected:
   Cond _cond;
   atomic_t ref;
 
+#ifdef PG_DEBUG_REFS
+  Mutex _ref_id_lock;
+  map<uint64_t, string> _live_ids;
+  map<string, uint64_t> _tag_counts;
+  uint64_t _ref_id;
+#endif
+
 public:
   bool deleting;  // true while in removing or OSD is shutting down
 
@@ -428,17 +442,13 @@ public:
     _cond.Signal();
   }
 
-  void get() {
-    //generic_dout(0) << this << " " << info.pgid << " get " << ref.test() << dendl;
-    //assert(_lock.is_locked());
-    ref.inc();
-  }
-  void put() { 
-    //generic_dout(0) << this << " " << info.pgid << " put " << ref.test() << dendl;
-    if (ref.dec() == 0)
-      delete this;
-  }
-
+#ifdef PG_DEBUG_REFS
+  uint64_t get_with_id();
+  void put_with_id(uint64_t);
+  void dump_live_ids();
+#endif
+  void get(const string &tag);
+  void put(const string &tag);
 
   bool dirty_info, dirty_big_info, dirty_log;
 
diff --git a/src/osd/ReplicatedPG.cc b/src/osd/ReplicatedPG.cc
index 427dce0d4c8..c5f50b9d235 100644
--- a/src/osd/ReplicatedPG.cc
+++ b/src/osd/ReplicatedPG.cc
@@ -4627,7 +4627,7 @@ void ReplicatedPG::sub_op_modify(OpRequestRef op)
 
   RepModify *rm = new RepModify;
   rm->pg = this;
-  get();
+  get("RepModify");
   rm->op = op;
   rm->ctx = 0;
   rm->ackerosd = ackerosd;
@@ -4760,7 +4760,7 @@ void ReplicatedPG::sub_op_modify_applied(RepModify *rm)
   if (done) {
     delete rm->ctx;
     delete rm;
-    put();
+    put("RepModify");
   }
 }
 
@@ -4795,7 +4795,7 @@ void ReplicatedPG::sub_op_modify_commit(RepModify *rm)
   if (done) {
     delete rm->ctx;
     delete rm;
-    put();
+    put("RepModify");
   }
 }
 
@@ -7449,5 +7449,10 @@ boost::statechart::result ReplicatedPG::WaitingOnReplicas::react(const SnapTrim&
   return transit< NotTrimming >();
 }
 
-void intrusive_ptr_add_ref(ReplicatedPG *pg) { pg->get(); }
-void intrusive_ptr_release(ReplicatedPG *pg) { pg->put(); }
+void intrusive_ptr_add_ref(ReplicatedPG *pg) { pg->get("intptr"); }
+void intrusive_ptr_release(ReplicatedPG *pg) { pg->put("intptr"); }
+
+#ifdef PG_DEBUG_REFS
+uint64_t get_with_id(ReplicatedPG *pg) { return pg->get_with_id(); }
+void put_with_id(ReplicatedPG *pg, uint64_t id) { return pg->put_with_id(id); }
+#endif
diff --git a/src/osd/ReplicatedPG.h b/src/osd/ReplicatedPG.h
index 0bc8a6beff6..2ef2627ba70 100644
--- a/src/osd/ReplicatedPG.h
+++ b/src/osd/ReplicatedPG.h
@@ -31,7 +31,15 @@ class MOSDSubOpReply;
 class ReplicatedPG;
 void intrusive_ptr_add_ref(ReplicatedPG *pg);
 void intrusive_ptr_release(ReplicatedPG *pg);
+uint64_t get_with_id(ReplicatedPG *pg);
+void put_with_id(ReplicatedPG *pg, uint64_t id);
+
+#ifdef PG_DEBUG_REFS
+  typedef TrackedIntPtr<ReplicatedPG> ReplicatedPGRef;
+#else
   typedef boost::intrusive_ptr<ReplicatedPG> ReplicatedPGRef;
+#endif
+
 class PGLSFilter {
 protected:
   string xattr;