Merge pull request ceph#55223 from athanatos/sjust/wip-64055

crimson: clear obc_registry on interval change

Reviewed-by: Matan Breizman <[email protected]>
Reviewed-by: Samuel Just <[email protected]>

Author: athanatos

src/common/intrusive_lru.h

@@ -12,13 +12,12 @@ namespace ceph::common {
 /**
  * intrusive_lru: lru implementation with embedded map and list hook
  *
- * Elements will be stored in an intrusive set. Once an element is no longer
- * referenced it will remain in the set. The unreferenced elements will be
- * evicted from the set once the set size exceeds the `lru_target_size`.
- * Referenced elements will not be evicted as this is a registery with
- * extra caching capabilities.
+ * Elements with live references are guarranteed to remain accessible.
+ * Elements without live references may remain accessible -- implementation
+ * will release unreferenced elements based on lru_target_size.
  *
- * Note, this implementation currently is entirely thread-unsafe.
+ * Accesses, mutations, and references must be confined to a single thread or
+ * serialized via some other mechanism.
  */
 
 template <typename K, typename V, typename VToK>
@@ -43,19 +42,47 @@ void intrusive_ptr_release(intrusive_lru_base<Config> *p);
 
 template <typename Config>
 class intrusive_lru_base {
+  /* object invariants
+   *
+   * 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 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 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;
 
-  // lru points to the corresponding intrusive_lru
-  // which will be set to null if its use_count
-  // is zero (aka unreferenced).
+  // See above, points at intrusive_lru iff referenced
   intrusive_lru<Config> *lru = nullptr;
 
 public:
   bool is_referenced() const {
     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;
@@ -108,9 +135,9 @@ class intrusive_lru {
 
   // when the lru_set exceeds its target size, evict
   // only unreferenced elements from it (if any).
-  void evict() {
+  void evict(unsigned target_size) {
     while (!unreferenced_list.empty() &&
-	   lru_set.size() > lru_target_size) {
+	   lru_set.size() > target_size) {
       auto &evict_target = unreferenced_list.front();
       assert(evict_target.is_unreferenced());
       unreferenced_list.pop_front();
@@ -136,7 +163,7 @@ class intrusive_lru {
     assert(b.is_unreferenced());
     lru_set.insert(b);
     b.lru = this;
-    evict();
+    evict(lru_target_size);
   }
 
   // an element in the lru_set has no users,
@@ -145,7 +172,7 @@ class intrusive_lru {
     assert(b.is_referenced());
     unreferenced_list.push_back(b);
     b.lru = nullptr;
-    evict();
+    evict(lru_target_size);
   }
 
 public:
@@ -189,6 +216,21 @@ class intrusive_lru {
       }
   }
 
+  /// 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) {
@@ -212,7 +254,7 @@ class intrusive_lru {
 
   void set_target_size(size_t target_size) {
     lru_target_size = target_size;
-    evict();
+    evict(lru_target_size);
   }
 
   ~intrusive_lru() {
@@ -226,17 +268,24 @@ class intrusive_lru {
 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
+   * is_referenced() or is_invalidated() objects with live external references */
   assert(p);
   assert(p->use_count > 0);
+  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;
+    }
   }
 }
 

src/crimson/osd/object_context.h

@@ -94,11 +94,13 @@ class ObjectContext : public ceph::common::intrusive_lru_base<
     ceph_assert(is_head());
     obs = std::move(_obs);
     ssc = std::move(_ssc);
+    fully_loaded = true;
   }
 
   void set_clone_state(ObjectState &&_obs) {
     ceph_assert(!is_head());
     obs = std::move(_obs);
+    fully_loaded = true;
   }
 
   /// pass the provided exception to any waiting consumers of this ObjectContext
@@ -110,6 +112,10 @@ class ObjectContext : public ceph::common::intrusive_lru_base<
     }
   }
 
+  bool is_loaded_and_valid() const {
+    return fully_loaded && !invalidated_by_interval_change;
+  }
+
 private:
   tri_mutex lock;
   bool recovery_read_marker = false;
@@ -124,9 +130,13 @@ class ObjectContext : public ceph::common::intrusive_lru_base<
     });
   }
 
-  boost::intrusive::list_member_hook<> list_hook;
+  boost::intrusive::list_member_hook<> obc_accessing_hook;
   uint64_t list_link_cnt = 0;
+  bool fully_loaded = false;
+  bool invalidated_by_interval_change = false;
 
+  friend class ObjectContextRegistry;
+  friend class ObjectContextLoader;
 public:
 
   template <typename ListType>
@@ -147,7 +157,7 @@ class ObjectContext : public ceph::common::intrusive_lru_base<
   using obc_accessing_option_t = boost::intrusive::member_hook<
     ObjectContext,
     boost::intrusive::list_member_hook<>,
-    &ObjectContext::list_hook>;
+    &ObjectContext::obc_accessing_hook>;
 
   template<RWState::State Type, typename InterruptCond = void, typename Func>
   auto with_lock(Func&& func) {
@@ -260,6 +270,12 @@ class ObjectContextRegistry : public md_config_obs_t  {
     obc_lru.clear_range(from, to);
   }
 
+  void invalidate_on_interval_change() {
+    obc_lru.clear([](auto &obc) {
+      obc.invalidated_by_interval_change = true;
+    });
+  }
+
   template <class F>
   void for_each(F&& f) {
     obc_lru.for_each(std::forward<F>(f));

src/crimson/osd/object_context_loader.cc

@@ -168,6 +168,15 @@ using crimson::common::local_conf;
     auto loaded =
       load_obc_iertr::make_ready_future<ObjectContextRef>(obc);
     if (existed) {
+      if (!obc->is_loaded_and_valid()) {
+	ERRORDPP(
+	  "obc for {} invalid -- fully_loaded={}, "
+	  "invalidated_by_interval_change={}",
+	  dpp, obc->get_oid(),
+	  obc->fully_loaded, obc->invalidated_by_interval_change
+	);
+      }
+      ceph_assert(obc->is_loaded_and_valid());
       DEBUGDPP("cache hit on {}", dpp, obc->get_oid());
     } else {
       DEBUGDPP("cache miss on {}", dpp, obc->get_oid());

src/crimson/osd/ops_executer.cc

@@ -931,12 +931,12 @@ std::unique_ptr<OpsExecuter::CloningContext> OpsExecuter::execute_clone(
     return std::vector<snapid_t>{std::begin(snapc.snaps), last};
   }();
 
-  auto [snap_oi, clone_obc] = prepare_clone(coid);
+  auto clone_obc = prepare_clone(coid);
   // make clone
-  backend.clone(snap_oi, initial_obs, clone_obc->obs, txn);
+  backend.clone(clone_obc->obs.oi, initial_obs, clone_obc->obs, txn);
 
   delta_stats.num_objects++;
-  if (snap_oi.is_omap()) {
+  if (clone_obc->obs.oi.is_omap()) {
     delta_stats.num_objects_omap++;
   }
   delta_stats.num_object_clones++;
@@ -960,7 +960,7 @@ std::unique_ptr<OpsExecuter::CloningContext> OpsExecuter::execute_clone(
   cloning_ctx->log_entry = {
     pg_log_entry_t::CLONE,
     coid,
-    snap_oi.version,
+    clone_obc->obs.oi.version,
     initial_obs.oi.version,
     initial_obs.oi.user_version,
     osd_reqid_t(),
@@ -1028,32 +1028,23 @@ OpsExecuter::flush_clone_metadata(
   });
 }
 
-// TODO: make this static
-std::pair<object_info_t, ObjectContextRef> OpsExecuter::prepare_clone(
+ObjectContextRef OpsExecuter::prepare_clone(
   const hobject_t& coid)
 {
-  object_info_t static_snap_oi(coid);
-  static_snap_oi.version = osd_op_params->at_version;
-  static_snap_oi.prior_version = obc->obs.oi.version;
-  static_snap_oi.copy_user_bits(obc->obs.oi);
-  if (static_snap_oi.is_whiteout()) {
-    // clone shouldn't be marked as whiteout
-    static_snap_oi.clear_flag(object_info_t::FLAG_WHITEOUT);
-  }
-
-  ObjectContextRef clone_obc;
-  if (pg->is_primary()) {
-    // lookup_or_create
-    auto [c_obc, existed] =
-      pg->obc_registry.get_cached_obc(std::move(coid));
-    assert(!existed);
-    c_obc->obs.oi = static_snap_oi;
-    c_obc->obs.exists = true;
-    c_obc->ssc = obc->ssc;
-    logger().debug("clone_obc: {}", c_obc->obs.oi);
-    clone_obc = std::move(c_obc);
-  }
-  return std::make_pair(std::move(static_snap_oi), std::move(clone_obc));
+  ceph_assert(pg->is_primary());
+  ObjectState clone_obs{coid};
+  clone_obs.exists = true;
+  clone_obs.oi.version = osd_op_params->at_version;
+  clone_obs.oi.prior_version = obc->obs.oi.version;
+  clone_obs.oi.copy_user_bits(obc->obs.oi);
+  clone_obs.oi.clear_flag(object_info_t::FLAG_WHITEOUT);
+
+  auto [clone_obc, existed] = pg->obc_registry.get_cached_obc(std::move(coid));
+  ceph_assert(!existed);
+
+  clone_obc->set_clone_state(std::move(clone_obs));
+  clone_obc->ssc = obc->ssc;
+  return clone_obc;
 }
 
 void OpsExecuter::apply_stats()

src/crimson/osd/ops_executer.h

@@ -449,7 +449,7 @@ class OpsExecuter : public seastar::enable_lw_shared_from_this<OpsExecuter> {
 
   version_t get_last_user_version() const;
 
-  std::pair<object_info_t, ObjectContextRef> prepare_clone(
+  ObjectContextRef prepare_clone(
     const hobject_t& coid);
 
   void apply_stats();

src/crimson/osd/pg.cc

@@ -1529,6 +1529,7 @@ void PG::on_change(ceph::os::Transaction &t) {
     client_request_orderer.clear_and_cancel(*this);
   }
   scrubber.on_interval_change();
+  obc_registry.invalidate_on_interval_change();
 }
 
 void PG::context_registry_on_change() {