common/intrusive_lru: add clear() mechanism and state for removed items with live...

We need to be able to clear the lru without waiting for all outstanding
references to be released.  Items with such references enter an
invalidated state where references can still be added and removed with
deletion at use_count == 0, but can't be accessed any longer via the lru.

Signed-off-by: Samuel Just <sjust@redhat.com>

diff --git a/src/common/intrusive_lru.h b/src/common/intrusive_lru.h
index fb94b5fd9ead1da94136d1a70caae297f46c224b..564cceef1cc2f6eace6347ddf3d6986caf06ab4c 100644 (file)
--- a/src/common/intrusive_lru.h
+++ b/src/common/intrusive_lru.h
@@ -44,20 +44,30 @@ template <typename Config>
 class intrusive_lru_base {
   /* object invariants
    *
-   * intrusive_lru objects may be in one of two states:
+   * intrusive_lru objects may be in one of three states:
    * 1. referenced
+   *    - accessible via intrusive_lru
    *    - intrusive_lru_base::lru is points to parent intrusive_lru
-   *    - present in lru_set
+   *    - present in intrusive_lru::lru_set
+   *    - absent from intrusive_lru::unreferenced_list
    *    - use_count > 0
    *    - not eligible for eviction
    *    - intrusive_lru_release may be invoked externally
    * 2. unreferenced
+   *    - accessible via intrusive_lru
    *    - intrusive_lru_base::lru is null
-   *    - present in lru_set
+   *    - present in intrusive_lru::lru_set
    *    - present in intrusive_lru::unreferenced_list
    *    - use_count == 0
    *    - eligible for eviction
    *    - intrusive_lru_release cannot be invoked
+   * 3. invalidated
+   *    - inaccessible via intrusive_lru
+   *    - intrusive_lru_base::lru is null
+   *    - absent from intrusive_lru::lru_set
+   *    - absent from intrusive_lru::unreferenced_list
+   *    - use_count > 0
+   *    - intrusive_lru_release may be invoked externally
    */
   unsigned use_count = 0;
 
@@ -69,7 +79,10 @@ public:
     return static_cast<bool>(lru);
   }
   bool is_unreferenced() const {
-    return !is_referenced();
+    return !is_referenced() && use_count == 0;
+  }
+  bool is_invalidated() const {
+    return !is_referenced() && use_count > 0;
   }
   boost::intrusive::set_member_hook<> set_hook;
   boost::intrusive::list_member_hook<> list_hook;
@@ -203,6 +216,21 @@ public:
       }
   }
 
+  /// drop all elements from lru, invoke f on any with outstanding references
+  template <typename F>
+  void clear(F &&f) {
+    evict(0);
+    assert(unreferenced_list.empty());
+    for (auto &i: lru_set) {
+      std::invoke(f, static_cast<T&>(i));
+      i.lru = nullptr;
+      assert(i.is_invalidated());
+    }
+    lru_set.clear_and_dispose([](auto *i){
+      assert(i->use_count > 0); /* don't delete, still has a ref count */
+    });
+  }
+
   template <class F>
   void for_each(F&& f) {
     for (auto& v : lru_set) {
@@ -240,20 +268,24 @@ public:
 template <typename Config>
 void intrusive_ptr_add_ref(intrusive_lru_base<Config> *p) {
   assert(p);
-  assert(p->lru);
   p->use_count++;
+  assert(p->is_referenced() || p->is_invalidated());
 }
 
 template <typename Config>
 void intrusive_ptr_release(intrusive_lru_base<Config> *p) {
   /* See object invariants above -- intrusive_ptr_release can only be invoked on
-   * referenced objects */
+   * is_referenced() or is_invalidated() objects with live external references */
   assert(p);
   assert(p->use_count > 0);
-  assert(is_referenced());
+  assert(p->is_referenced() || p->is_invalidated());
   --p->use_count;
   if (p->use_count == 0) {
-    p->lru->mark_as_unreferenced(*p);
+    if (p->lru) {
+      p->lru->mark_as_unreferenced(*p);
+    } else {
+      delete p;
+    }
   }
 }
 

diff --git a/src/test/common/test_intrusive_lru.cc b/src/test/common/test_intrusive_lru.cc
index 0654bd97d81ecaa51ddf8552f672cfff4e65364e..af8edb8e2bf3a2b62e95c8f7df66a7672e90ffb5 100644 (file)
--- a/src/test/common/test_intrusive_lru.cc
+++ b/src/test/common/test_intrusive_lru.cc
@@ -13,14 +13,21 @@ struct item_to_unsigned {
   }
 };
 
+
+static int LIVE_TEST_LRU_ITEMS = 0;
 struct TestLRUItem : public ceph::common::intrusive_lru_base<
   ceph::common::intrusive_lru_config<
     unsigned, TestLRUItem, item_to_unsigned<TestLRUItem>>> {
   unsigned key = 0;
   int value = 0;
+  bool invalidated = false;
 
-  TestLRUItem(unsigned key) : key(key) {}
+  TestLRUItem(unsigned key) : key(key) {
+    ++LIVE_TEST_LRU_ITEMS;
+  }
+  ~TestLRUItem() { --LIVE_TEST_LRU_ITEMS; }
 };
+using TestLRUItemRef = boost::intrusive_ptr<TestLRUItem>;
 
 class LRUTest : public TestLRUItem::lru_t {
 public:
@@ -206,3 +213,72 @@ TEST(LRU, clear_range) {
     ASSERT_FALSE(existed);
   }
 }
+
+TEST(LRU, clear) {
+  LRUTest cache;
+  const unsigned SIZE = 10;
+  cache.set_target_size(SIZE);
+  
+  std::vector<TestLRUItemRef> refs;
+  for (unsigned i = 0; i < 100; ++i) {
+    auto [ref, existed] = cache.add(i, i);
+    ASSERT_FALSE(existed);
+    if ((i % 2) == 0) {
+      refs.push_back(ref);
+    }
+  }
+
+  for (unsigned i = 0; i < 100; i += 2) {
+    auto [ref, existed] = cache.add(i, i);
+    ASSERT_TRUE(existed);
+  }
+
+  cache.clear([](auto &i) { i.invalidated = true; });
+  ASSERT_EQ(refs.size(), LIVE_TEST_LRU_ITEMS);
+
+  for (auto &i: refs) {
+    ASSERT_TRUE(i->invalidated);
+  }
+
+  std::vector<TestLRUItemRef> refs_new;
+  for (unsigned i = 0; i < 100; ++i) {
+    auto [ref, existed] = cache.add(i, i);
+    ASSERT_FALSE(existed);
+    ASSERT_FALSE(ref->invalidated);
+    if ((i % 2) == 0) {
+      refs_new.push_back(ref);
+    }
+  }
+
+  for (unsigned i = 0; i < 100; i += 2) {
+    auto [ref, existed] = cache.add(i, i);
+    ASSERT_TRUE(existed);
+    ASSERT_FALSE(ref->invalidated);
+  }
+
+  refs.clear();
+  cache.set_target_size(0);
+  ASSERT_EQ(refs_new.size(), LIVE_TEST_LRU_ITEMS);
+  cache.set_target_size(SIZE);
+
+  for (unsigned i = 100; i < 200; ++i) {
+    auto [ref, existed] = cache.add(i, i);
+    ASSERT_FALSE(existed);
+    ASSERT_FALSE(ref->invalidated);
+    if ((i % 2) == 0) {
+      refs_new.push_back(ref);
+    }
+  }
+
+  for (unsigned i = 0; i < 200; i += 2) {
+    auto [ref, existed] = cache.add(i, i);
+    ASSERT_TRUE(existed);
+    ASSERT_FALSE(ref->invalidated);
+  }
+
+  ASSERT_EQ(refs_new.size(), LIVE_TEST_LRU_ITEMS);
+  refs_new.clear();
+  ASSERT_EQ(SIZE, LIVE_TEST_LRU_ITEMS);
+  cache.set_target_size(0);
+  ASSERT_EQ(0, LIVE_TEST_LRU_ITEMS);
+}