sharedptr_registry.hpp: removed ptrs need to not blast contents

See the included unit test update.  Consider:
1) x = lookup_or_create(1, 1)
2) remove(1)
3) y = lookup_or_create(1, 2)
4) x.reset()
5) z = lookup(1)

The bug is that z will be null since x.reset() caused the
cleanup callback to remove y's key value from contents.

To fix this, contents also records the pointer value for
the weak_ptr.  The removal callback only removes the
key from contents if it matches the ptr in contents.

This should work since the pointer passed to the removal
callback must be unique up to that point since it has
not yet been deleted.

This allowed a pg removal -> pg recreation -> pg removal
sequence to cause the second pg removal entry to be
erroneously cleared by the first pg removal's destructor
as it finally made its way through the removal queue.

Fixes: #5951
Signed-off-by: Samuel Just <sam.just@inktank.com>
Reviewed-by: Greg Farnum <greg@inktank.com>

2 files changed, 75 insertions, 25 deletions

diff --git a/src/common/sharedptr_registry.hpp b/src/common/sharedptr_registry.hpp
index 90043001ee7..9fe2fe6be1a 100644
--- a/src/common/sharedptr_registry.hpp
+++ b/src/common/sharedptr_registry.hpp
@@ -33,7 +33,7 @@ public:
 private:
   Mutex lock;
   Cond cond;
-  map<K, WeakVPtr> contents;
+  map<K, pair<WeakVPtr, V*> > contents;
 
   class OnRemoval {
     SharedPtrRegistry<K,V> *parent;
@@ -44,8 +44,13 @@ private:
     void operator()(V *to_remove) {
       {
 	Mutex::Locker l(parent->lock);
-	parent->contents.erase(key);
-	parent->cond.Signal();
+	typename map<K, pair<WeakVPtr, V*> >::iterator i =
+	  parent->contents.find(key);
+	if (i != parent->contents.end() &&
+	    i->second.second == to_remove) {
+	  parent->contents.erase(i);
+	  parent->cond.Signal();
+	}
       }
       delete to_remove;
     }
@@ -68,9 +73,10 @@ public:
     {
       Mutex::Locker l(lock);
       VPtr next_val;
-      typename map<K, WeakVPtr>::iterator i = contents.upper_bound(key);
+      typename map<K, pair<WeakVPtr, V*> >::iterator i =
+	contents.upper_bound(key);
       while (i != contents.end() &&
-	     !(next_val = i->second.lock()))
+	     !(next_val = i->second.first.lock()))
 	++i;
       if (i == contents.end())
 	return false;
@@ -86,9 +92,10 @@ public:
   bool get_next(const K &key, pair<K, V> *next) {
     VPtr next_val;
     Mutex::Locker l(lock);
-    typename map<K, WeakVPtr>::iterator i = contents.upper_bound(key);
+    typename map<K, pair<WeakVPtr, V*> >::iterator i =
+      contents.upper_bound(key);
     while (i != contents.end() &&
-	   !(next_val = i->second.lock()))
+	   !(next_val = i->second.first.lock()))
       ++i;
     if (i == contents.end())
       return false;
@@ -101,8 +108,10 @@ public:
     Mutex::Locker l(lock);
     waiting++;
     while (1) {
-      if (contents.count(key)) {
-	VPtr retval = contents[key].lock();
+      typename map<K, pair<WeakVPtr, V*> >::iterator i =
+	contents.find(key);
+      if (i != contents.end()) {
+	VPtr retval = i->second.first.lock();
 	if (retval) {
 	  waiting--;
 	  return retval;
@@ -120,8 +129,10 @@ public:
     Mutex::Locker l(lock);
     waiting++;
     while (1) {
-      if (contents.count(key)) {
-	VPtr retval = contents[key].lock();
+      typename map<K, pair<WeakVPtr, V*> >::iterator i =
+	contents.find(key);
+      if (i != contents.end()) {
+	VPtr retval = i->second.first.lock();
 	if (retval) {
 	  waiting--;
 	  return retval;
@@ -131,8 +142,9 @@ public:
       }
       cond.Wait(lock);
     }
-    VPtr retval(new V(), OnRemoval(this, key));
-    contents[key] = retval;
+    V *ptr = new V();
+    VPtr retval(ptr, OnRemoval(this, key));
+    contents.insert(make_pair(key, make_pair(retval, ptr)));
     waiting--;
     return retval;
   }
@@ -148,8 +160,10 @@ public:
     Mutex::Locker l(lock);
     waiting++;
     while (1) {
-      if (contents.count(key)) {
-	VPtr retval = contents[key].lock();
+      typename map<K, pair<WeakVPtr, V*> >::iterator i =
+	contents.find(key);
+      if (i != contents.end()) {
+	VPtr retval = i->second.first.lock();
 	if (retval) {
 	  waiting--;
 	  return retval;
@@ -159,8 +173,9 @@ public:
       }
       cond.Wait(lock);
     }
-    VPtr retval(new V(arg), OnRemoval(this, key));
-    contents[key] = retval;
+    V *ptr = new V(arg);
+    VPtr retval(ptr, OnRemoval(this, key));
+    contents.insert(make_pair(key, make_pair(retval, ptr)));
     waiting--;
     return retval;
   }
diff --git a/src/test/common/test_sharedptr_registry.cc b/src/test/common/test_sharedptr_registry.cc
index b1713a9bd9f..6121b6335b8 100644
--- a/src/test/common/test_sharedptr_registry.cc
+++ b/src/test/common/test_sharedptr_registry.cc
@@ -32,7 +32,9 @@ using namespace std::tr1;
 class SharedPtrRegistryTest : public SharedPtrRegistry<unsigned int, int> {
 public:
   Mutex &get_lock() { return lock; }
-  map<unsigned int, weak_ptr<int> > &get_contents() { return contents; }
+  map<unsigned int, pair<weak_ptr<int>, int*> > &get_contents() {
+    return contents;
+  }
 };
 
 class SharedPtrRegistry_all : public ::testing::Test {
@@ -125,9 +127,9 @@ TEST_F(SharedPtrRegistry_all, wait_lookup_or_create) {
     unsigned int key = 1;
     {
       shared_ptr<int> ptr(new int);
-      registry.get_contents()[key] = ptr;
+      registry.get_contents()[key] = make_pair(ptr, ptr.get());
     }
-    EXPECT_FALSE(registry.get_contents()[key].lock());
+    EXPECT_FALSE(registry.get_contents()[key].first.lock());
 
     Thread_wait t(registry, key, 0, Thread_wait::LOOKUP_OR_CREATE);
     t.create();
@@ -145,9 +147,9 @@ TEST_F(SharedPtrRegistry_all, wait_lookup_or_create) {
     int value = 3;
     {
       shared_ptr<int> ptr(new int);
-      registry.get_contents()[key] = ptr;
+      registry.get_contents()[key] = make_pair(ptr, ptr.get());
     }
-    EXPECT_FALSE(registry.get_contents()[key].lock());
+    EXPECT_FALSE(registry.get_contents()[key].first.lock());
 
     Thread_wait t(registry, key, value, Thread_wait::LOOKUP_OR_CREATE);
     t.create();
@@ -188,9 +190,9 @@ TEST_F(SharedPtrRegistry_all, wait_lookup) {
   int value = 2;
   {
     shared_ptr<int> ptr(new int);
-    registry.get_contents()[key] = ptr;
+    registry.get_contents()[key] = make_pair(ptr, ptr.get());
   }
-  EXPECT_FALSE(registry.get_contents()[key].lock());
+  EXPECT_FALSE(registry.get_contents()[key].first.lock());
 
   Thread_wait t(registry, key, value, Thread_wait::LOOKUP);
   t.create();
@@ -221,7 +223,7 @@ TEST_F(SharedPtrRegistry_all, get_next) {
 
     // entries with expired pointers are silentely ignored
     const unsigned int key_gone = 222;
-    registry.get_contents()[key_gone] = shared_ptr<int>();
+    registry.get_contents()[key_gone] = make_pair(shared_ptr<int>(), (int*)0);
 
     const unsigned int key1 = 111;
     shared_ptr<int> ptr1 = registry.lookup_or_create(key1);
@@ -258,6 +260,39 @@ TEST_F(SharedPtrRegistry_all, get_next) {
   }
 }
 
+TEST_F(SharedPtrRegistry_all, remove) {
+  {
+    SharedPtrRegistryTest registry;
+    const unsigned int key1 = 1;
+    shared_ptr<int> ptr1 = registry.lookup_or_create(key1);
+    *ptr1 = 400;
+    registry.remove(key1);
+
+    shared_ptr<int> ptr2 = registry.lookup_or_create(key1);
+    *ptr2 = 500;
+
+    ptr1 = shared_ptr<int>();
+    shared_ptr<int> res = registry.lookup(key1);
+    assert(res);
+    assert(res == ptr2);
+    assert(*res == 500);
+  }
+  {
+    SharedPtrRegistryTest registry;
+    const unsigned int key1 = 1;
+    shared_ptr<int> ptr1 = registry.lookup_or_create(key1, 400);
+    registry.remove(key1);
+
+    shared_ptr<int> ptr2 = registry.lookup_or_create(key1, 500);
+
+    ptr1 = shared_ptr<int>();
+    shared_ptr<int> res = registry.lookup(key1);
+    assert(res);
+    assert(res == ptr2);
+    assert(*res == 500);
+  }
+}
+
 class SharedPtrRegistry_destructor : public ::testing::Test {
 public: