replace in_method_t with a counter

A single counter ( waiting ) accurately reflects the number of
waiters, regardless of the method waiting. It is enough to allow
unit tests to synthetise all situations, including:

T1: x = lookup_or_create(0)
T1: release x part 1 (weak_ptrs now fail to lock)
T2: y = lookup_or_create(0)
T2: block in lookup_or_create (waiting == 1)
T1: z = lookup_or_create(1) (does not block because the key is different)
    while holding the lock it waiting++ and waiting == 2
    and before returning it waiting-- and waiting is back to == 1
T1: complete release x
T2: complete lookup_or_create(0) (waiting == 0)

The unit tests are modified to add a lookup on an unrelated key to
demonstrate that it does not reset waiting counter.

http://tracker.ceph.com/issues/5527 refs #5527

Signed-off-by: Loic Dachary <loic@dachary.org>

2 files changed, 42 insertions, 32 deletions

diff --git a/src/common/sharedptr_registry.hpp b/src/common/sharedptr_registry.hpp
index 9e2cdc7b0dd..a62aa0d9ce3 100644
--- a/src/common/sharedptr_registry.hpp
+++ b/src/common/sharedptr_registry.hpp
@@ -29,7 +29,7 @@ class SharedPtrRegistry {
 public:
   typedef std::tr1::shared_ptr<V> VPtr;
   typedef std::tr1::weak_ptr<V> WeakVPtr;
-  enum in_method_t { UNDEFINED, LOOKUP, LOOKUP_OR_CREATE } in_method;
+  int waiting;
 private:
   Mutex lock;
   Cond cond;
@@ -54,7 +54,7 @@ private:
 
 public:
   SharedPtrRegistry() :
-    in_method(UNDEFINED),
+    waiting(0),
     lock("SharedPtrRegistry::lock")
   {}
 
@@ -74,12 +74,12 @@ public:
 
   VPtr lookup(const K &key) {
     Mutex::Locker l(lock);
-    in_method = LOOKUP;
+    waiting++;
     while (1) {
       if (contents.count(key)) {
 	VPtr retval = contents[key].lock();
 	if (retval) {
-	  in_method = UNDEFINED;
+	  waiting--;
 	  return retval;
 	}
       } else {
@@ -87,18 +87,18 @@ public:
       }
       cond.Wait(lock);
     }
-    in_method = UNDEFINED;
+    waiting--;
     return VPtr();
   }
 
   VPtr lookup_or_create(const K &key) {
     Mutex::Locker l(lock);
-    in_method = LOOKUP_OR_CREATE;
+    waiting++;
     while (1) {
       if (contents.count(key)) {
 	VPtr retval = contents[key].lock();
 	if (retval) {
-	  in_method = UNDEFINED;
+	  waiting--;
 	  return retval;
 	}
       } else {
@@ -108,7 +108,7 @@ public:
     }
     VPtr retval(new V(), OnRemoval(this, key));
     contents[key] = retval;
-    in_method = UNDEFINED;
+    waiting--;
     return retval;
   }
 
@@ -121,12 +121,12 @@ public:
   template<class A>
   VPtr lookup_or_create(const K &key, const A &arg) {
     Mutex::Locker l(lock);
-    in_method = LOOKUP_OR_CREATE;
+    waiting++;
     while (1) {
       if (contents.count(key)) {
 	VPtr retval = contents[key].lock();
 	if (retval) {
-	  in_method = UNDEFINED;
+	  waiting--;
 	  return retval;
 	}
       } else {
@@ -136,7 +136,7 @@ public:
     }
     VPtr retval(new V(arg), OnRemoval(this, key));
     contents[key] = retval;
-    in_method = UNDEFINED;
+    waiting--;
     return retval;
   }
 
diff --git a/src/test/common/test_sharedptr_registry.cc b/src/test/common/test_sharedptr_registry.cc
index 3ca7b49beb9..aec2107c9e5 100644
--- a/src/test/common/test_sharedptr_registry.cc
+++ b/src/test/common/test_sharedptr_registry.cc
@@ -44,9 +44,9 @@ public:
     unsigned int key;
     int value;
     shared_ptr<int> ptr;
-    SharedPtrRegistryTest::in_method_t in_method;
+    enum in_method_t { LOOKUP, LOOKUP_OR_CREATE } in_method;
 
-    Thread_wait(SharedPtrRegistryTest& _registry, unsigned int _key, int _value, SharedPtrRegistryTest::in_method_t _in_method) : 
+    Thread_wait(SharedPtrRegistryTest& _registry, unsigned int _key, int _value, in_method_t _in_method) : 
       registry(_registry),
       key(_key),
       value(_value),
@@ -56,19 +56,17 @@ public:
     
     virtual void *entry() {
       switch(in_method) {
-      case SharedPtrRegistryTest::LOOKUP_OR_CREATE:
+      case LOOKUP_OR_CREATE:
 	if (value) 
 	  ptr = registry.lookup_or_create<int>(key, value);
 	else
 	  ptr = registry.lookup_or_create(key);
 	break;
-      case SharedPtrRegistryTest::LOOKUP:
+      case LOOKUP:
 	ptr = shared_ptr<int>(new int);
 	*ptr = value;
 	ptr = registry.lookup(key);
 	break;
-      case SharedPtrRegistryTest::UNDEFINED:
-	break;
       }
       return NULL;
     }
@@ -77,7 +75,7 @@ public:
   static const useconds_t DELAY_MAX = 20 * 1000 * 1000;
   static useconds_t delay;
 
-  bool wait_for(SharedPtrRegistryTest &registry, SharedPtrRegistryTest::in_method_t method) {
+  bool wait_for(SharedPtrRegistryTest &registry, int waiting) {
     do {
       //
       // the delay variable is supposed to be initialized to zero. It would be fine
@@ -89,7 +87,7 @@ public:
 	usleep(delay);
       {
 	Mutex::Locker l(registry.get_lock());
-	if (registry.in_method == method) 
+	if (registry.waiting == waiting) 
 	  break;
       }
       if (delay > 0)
@@ -118,10 +116,10 @@ TEST_F(SharedPtrRegistry_all, wait_lookup_or_create) {
   // out of scope and the shared_ptr object is about to be removed and
   // marked as such. The weak_ptr stored in the registry will show
   // that it has expired(). However, the SharedPtrRegistry::OnRemoval
-  // object not yet been called and did not get a chance to acquire
-  // the lock. The lookup_or_create and lookup methods must detect
-  // that situation and wait until the weak_ptr is removed from the
-  // registry.
+  // object has not yet been called and did not get a chance to
+  // acquire the lock. The lookup_or_create and lookup methods must
+  // detect that situation and wait until the weak_ptr is removed from
+  // the registry.
   //
   {
     unsigned int key = 1;
@@ -131,12 +129,14 @@ TEST_F(SharedPtrRegistry_all, wait_lookup_or_create) {
     }
     EXPECT_FALSE(registry.get_contents()[key].lock());
 
-    Thread_wait t(registry, key, 0, SharedPtrRegistryTest::LOOKUP_OR_CREATE);
+    Thread_wait t(registry, key, 0, Thread_wait::LOOKUP_OR_CREATE);
     t.create();
-    ASSERT_TRUE(wait_for(registry, SharedPtrRegistryTest::LOOKUP_OR_CREATE));
+    ASSERT_TRUE(wait_for(registry, 1));
     EXPECT_FALSE(t.ptr);
+    // waiting on a key does not block lookups on other keys
+    EXPECT_TRUE(registry.lookup_or_create(key + 12345));
     registry.remove(key);
-    ASSERT_TRUE(wait_for(registry, SharedPtrRegistryTest::UNDEFINED));
+    ASSERT_TRUE(wait_for(registry, 0));
     EXPECT_TRUE(t.ptr);
     t.join();
   }
@@ -149,12 +149,20 @@ TEST_F(SharedPtrRegistry_all, wait_lookup_or_create) {
     }
     EXPECT_FALSE(registry.get_contents()[key].lock());
 
-    Thread_wait t(registry, key, value, SharedPtrRegistryTest::LOOKUP_OR_CREATE);
+    Thread_wait t(registry, key, value, Thread_wait::LOOKUP_OR_CREATE);
     t.create();
-    ASSERT_TRUE(wait_for(registry, SharedPtrRegistryTest::LOOKUP_OR_CREATE));
+    ASSERT_TRUE(wait_for(registry, 1));
     EXPECT_FALSE(t.ptr);
+    // waiting on a key does not block lookups on other keys
+    {
+      int other_value = value + 1;
+      unsigned int other_key = key + 1;
+      shared_ptr<int> ptr = registry.lookup_or_create<int>(other_key, other_value);
+      EXPECT_TRUE(ptr);
+      EXPECT_EQ(other_value, *ptr);
+    }
     registry.remove(key);
-    ASSERT_TRUE(wait_for(registry, SharedPtrRegistryTest::UNDEFINED));
+    ASSERT_TRUE(wait_for(registry, 0));
     EXPECT_TRUE(t.ptr);
     EXPECT_EQ(value, *t.ptr);
     t.join();
@@ -184,12 +192,14 @@ TEST_F(SharedPtrRegistry_all, wait_lookup) {
   }
   EXPECT_FALSE(registry.get_contents()[key].lock());
 
-  Thread_wait t(registry, key, value, SharedPtrRegistryTest::LOOKUP);
+  Thread_wait t(registry, key, value, Thread_wait::LOOKUP);
   t.create();
-  ASSERT_TRUE(wait_for(registry, SharedPtrRegistryTest::LOOKUP));
+  ASSERT_TRUE(wait_for(registry, 1));
   EXPECT_EQ(value, *t.ptr);
+  // waiting on a key does not block lookups on other keys
+  EXPECT_FALSE(registry.lookup(key + 12345));
   registry.remove(key);
-  ASSERT_TRUE(wait_for(registry, SharedPtrRegistryTest::UNDEFINED));
+  ASSERT_TRUE(wait_for(registry, 0));
   EXPECT_FALSE(t.ptr);
   t.join();
 }