Dead Page Reference Count Bug Fix

src/llfs/committable_page_cache_job.cpp

@@ -208,7 +208,12 @@ Status CommittablePageCacheJob::commit_impl(const JobCommitParams& params, u64 c
   const PageCacheJob* job = this->job_.get();
   BATT_CHECK_NOT_NULLPTR(job);
 
-  LLFS_VLOG(1) << "commit(PageCacheJob): entered";
+  if (durable_caller_slot) {
+    LLFS_VLOG(1) << "commit(PageCacheJob): entered" << BATT_INSPECT(prev_caller_slot)
+                 << BATT_INSPECT(*durable_caller_slot);
+  } else {
+    LLFS_VLOG(1) << "commit(PageCacheJob): entered" << BATT_INSPECT(prev_caller_slot);
+  }
 
   // Make sure the job is pruned!
   //
@@ -426,7 +431,7 @@ auto CommittablePageCacheJob::start_ref_count_updates(const JobCommitParams& par
                                                       PageRefCountUpdates& updates, u64 /*callers*/)
     -> StatusOr<DeadPages>
 {
-  LLFS_VLOG(1) << "commit(PageCacheJob): updating ref counts";
+  LLFS_VLOG(1) << "commit(PageCacheJob): updating ref counts" << BATT_INSPECT(params.caller_slot);
 
   DeadPages dead_pages;
 
@@ -579,10 +584,11 @@ Status CommittablePageCacheJob::recycle_dead_pages(const JobCommitParams& params
 
   BATT_ASSIGN_OK_RESULT(
       slot_offset_type recycler_sync_point,
-      params.recycler.recycle_pages(as_slice(dead_pages.ids), params.recycle_grant,
-                                    params.recycle_depth + 1));
+      params.recycler.recycle_pages(as_slice(dead_pages.ids), params.caller_slot,
+                                    params.recycle_grant, params.recycle_depth + 1));
 
-  LLFS_VLOG(1) << "commit(PageCacheJob): waiting for PageRecycler sync point";
+  LLFS_VLOG(1) << "commit(PageCacheJob): waiting for PageRecycler sync point"
+               << BATT_INSPECT(params.caller_slot);
 
   return params.recycler.await_flush(recycler_sync_point);
   //

src/llfs/page_allocator_state.cpp

@@ -458,11 +458,14 @@ PageAllocatorState::ProposalStatus PageAllocatorState::propose(PackedPageAllocat
   // If this is a valid proposal that will cause state change, go through and change the deltas to
   // the new ref count values.
   //
+  LLFS_VLOG(1) << "propose (start): txn-ref-count= " << txn->ref_counts.size();
   if (status == ProposalStatus::kValid) {
     for (PackedPageRefCount& prc : txn->ref_counts) {
-      prc.ref_count = this->calculate_new_ref_count(prc);
+      LLFS_VLOG(1) << "reference count: " << prc;
+      prc.ref_count = this->calculate_new_ref_count(prc, *user_index);
     }
   }
+  LLFS_VLOG(1) << "propose (end): txn-ref-count= " << txn->ref_counts.size();
 
   return status;
 }
@@ -556,13 +559,13 @@ namespace {
 void run_ref_count_update_sanity_checks(const PageIdFactory& id_factory,
                                         const PackedPageRefCount& delta,
                                         const PageAllocatorRefCount& obj, i32 old_count,
-                                        i32 new_count)
+                                        i32 new_count, const u32 index)
 {
   const auto debug_info = [&](std::ostream& out) {
     const page_generation_int delta_generation = id_factory.get_generation(delta.page_id.unpack());
 
     out << "(" << BATT_INSPECT(delta) << BATT_INSPECT(obj) << BATT_INSPECT(old_count)
-        << BATT_INSPECT(new_count) << ")" << BATT_INSPECT(delta_generation);
+        << BATT_INSPECT(new_count) << ")" << BATT_INSPECT(delta_generation) << BATT_INSPECT(index);
   };
 
   LLFS_VLOG(2) << debug_info;
@@ -639,7 +642,8 @@ void run_ref_count_update_sanity_checks(const PageIdFactory& id_factory,
 
 //==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- -  -  -   -
 //
-i32 PageAllocatorState::calculate_new_ref_count(const PackedPageRefCount& delta) const
+i32 PageAllocatorState::calculate_new_ref_count(const PackedPageRefCount& delta,
+                                                const u32 index) const
 {
   const PageId page_id = delta.page_id.unpack();
   const page_id_int physical_page = this->page_ids_.get_physical_page(page_id);
@@ -662,7 +666,7 @@ i32 PageAllocatorState::calculate_new_ref_count(const PackedPageRefCount& delta)
     new_count = old_count + delta.ref_count;
   }
 
-  run_ref_count_update_sanity_checks(this->page_ids_, delta, obj, old_count, new_count);
+  run_ref_count_update_sanity_checks(this->page_ids_, delta, obj, old_count, new_count, index);
 
   return new_count;
 }

src/llfs/page_allocator_state.hpp

@@ -242,7 +242,7 @@ class PageAllocatorState : public PageAllocatorStateNoLock
 
   // Returns the new ref count that will result from applying the delta to the passed obj.
   //
-  i32 calculate_new_ref_count(const PackedPageRefCount& delta) const;
+  i32 calculate_new_ref_count(const PackedPageRefCount& delta, const u32 index) const;
 
   /** \brief If the given ref count object has a positive ref count but *is* in the free pool, then
    * this function removes it; otherwise if the object has a zero ref count but is *not* in the free

src/llfs/page_recycler.cpp

@@ -72,23 +72,41 @@ StatusOr<SlotRange> refresh_recycler_info_slot(TypedSlotWriter<PageRecycleEvent>
 //
 /*static*/ u64 PageRecycler::calculate_log_size(const PageRecyclerOptions& options,
                                                 Optional<PageCount> max_buffered_page_count)
+{
+  const usize log_size_raw =
+      PageRecycler::calculate_log_size_no_padding(options, max_buffered_page_count);
+  const usize padding_bytes = 1 * kKiB;
+
+  return round_up_to_page_size_multiple(log_size_raw + padding_bytes);
+}
+
+//==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- -  -  -   -
+//
+/*static*/ u64 PageRecycler::calculate_log_size_no_padding(
+    const PageRecyclerOptions& options, Optional<PageCount> max_buffered_page_count)
 {
   static const PackedPageRecyclerInfo info = {};
 
-  return round_up_to_page_size_multiple(
-      options.total_page_grant_size() *
-          (1 + max_buffered_page_count.value_or(
-                   PageRecycler::default_max_buffered_page_count(options))) +
-      options.recycle_task_target() + packed_sizeof_slot(info) * (options.info_refresh_rate() + 1) +
-      1 * kKiB);
+  const usize bytes_per_buffered_page = options.total_page_grant_size();
+
+  const usize minimum_required_buffered_pages =
+      1 + max_buffered_page_count.value_or(PageRecycler::default_max_buffered_page_count(options));
+
+  const usize bytes_for_minimum_required_buffered_pages =
+      bytes_per_buffered_page * minimum_required_buffered_pages;
+
+  const usize fixed_size_overhead_bytes =
+      options.recycle_task_target() + packed_sizeof_slot(info) * (options.info_refresh_rate() + 1);
+
+  return bytes_for_minimum_required_buffered_pages + fixed_size_overhead_bytes;
 }
 
 //==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- -  -  -   -
 //
 /*static*/ PageCount PageRecycler::calculate_max_buffered_page_count(
     const PageRecyclerOptions& options, u64 log_size)
 {
-  const u64 required_log_size = PageRecycler::calculate_log_size(options, PageCount{0});
+  const u64 required_log_size = PageRecycler::calculate_log_size_no_padding(options, PageCount{0});
   if (log_size <= required_log_size) {
     return PageCount{0};
   }
@@ -107,6 +125,7 @@ StatusOr<SlotRange> refresh_recycler_info_slot(TypedSlotWriter<PageRecycleEvent>
 {
   initialize_status_codes();
 
+  LLFS_VLOG(1) << "PageRecycler:recover() start" << BATT_INSPECT(name);
   PageRecyclerRecoveryVisitor visitor{default_options};
 
   // Read the log, scanning its contents.
@@ -153,17 +172,20 @@ StatusOr<SlotRange> refresh_recycler_info_slot(TypedSlotWriter<PageRecycleEvent>
 
   state->bulk_load(as_slice(visitor.recovered_pages()));
 
-  return std::unique_ptr<PageRecycler>{new PageRecycler(scheduler, std::string{name}, page_deleter,
-                                                        std::move(*recovered_log),
-                                                        std::move(latest_batch), std::move(state))};
+  LLFS_VLOG(1) << "PageRecycler:recover() end" << BATT_INSPECT(name);
+
+  return std::unique_ptr<PageRecycler>{
+      new PageRecycler(scheduler, std::string{name}, page_deleter, std::move(*recovered_log),
+                       std::move(latest_batch), std::move(state), visitor.largest_unique_offset())};
 }
 
 //==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- -  -  -   -
 //
 PageRecycler::PageRecycler(batt::TaskScheduler& scheduler, const std::string& name,
                            PageDeleter& page_deleter, std::unique_ptr<LogDevice>&& wal_device,
                            Optional<Batch>&& recovered_batch,
-                           std::unique_ptr<PageRecycler::State>&& state) noexcept
+                           std::unique_ptr<PageRecycler::State>&& state,
+                           llfs::slot_offset_type largest_offset_as_unique_identifier_init) noexcept
     : scheduler_{scheduler}
     , name_{name}
     , page_deleter_{page_deleter}
@@ -177,6 +199,7 @@ PageRecycler::PageRecycler(batt::TaskScheduler& scheduler, const std::string& na
     , recycle_task_{}
     , metrics_{}
     , prepared_batch_{std::move(recovered_batch)}
+    , largest_offset_as_unique_identifier_{largest_offset_as_unique_identifier_init}
 {
   const PageRecyclerOptions& options = this->state_.no_lock().options;
 
@@ -202,6 +225,34 @@ PageRecycler::PageRecycler(batt::TaskScheduler& scheduler, const std::string& na
 #undef ADD_METRIC_
 }
 
+//==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- -  -  -   -
+/** \brief This infrastructure is to collect metrics for PageRecycler submodule.
+ * This metric collection is currently used by test suit to assess execution behavior of internal
+ * flows. This is static metric infrastructure so that any user level code could access it.
+ *
+ */
+auto PageRecycler::metrics_export() -> MetricsExported&
+{
+  static MetricsExported& metrics_ = [&]() -> MetricsExported& {
+    static MetricsExported metrics_;
+
+    LLFS_VLOG(1) << "Registering PageRecycler metrics...";
+    const auto metric_name = [](std::string_view property) {
+      return batt::to_string("PageRecycler_", property);
+    };
+
+#define ADD_METRIC_(n) global_metric_registry().add(metric_name(#n), metrics_.n)
+
+    ADD_METRIC_(page_id_deletion_reissue);
+
+#undef ADD_METRIC_
+
+    return metrics_;
+  }();
+
+  return metrics_;
+}
+
 //==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- -  -  -   -
 //
 PageRecycler::~PageRecycler() noexcept
@@ -270,20 +321,37 @@ void PageRecycler::join()
 
 //==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- -  -  -   -
 //
-StatusOr<slot_offset_type> PageRecycler::recycle_pages(const Slice<const PageId>& page_ids,
-                                                       batt::Grant* grant, i32 depth)
+StatusOr<slot_offset_type> PageRecycler::recycle_pages(
+    const Slice<const PageId>& page_ids, llfs::slot_offset_type offset_as_unique_identifier,
+    batt::Grant* grant, i32 depth)
 {
   BATT_CHECK_GE(depth, 0);
 
   LLFS_VLOG(1) << "PageRecycler::recycle_pages(page_ids=" << batt::dump_range(page_ids) << "["
                << page_ids.size() << "]"
                << ", grant=[" << (grant ? grant->size() : usize{0}) << "], depth=" << depth << ") "
-               << this->name_;
+               << this->name_ << BATT_INSPECT(offset_as_unique_identifier)
+               << BATT_INSPECT(this->largest_offset_as_unique_identifier_);
 
   if (page_ids.empty()) {
     return this->wal_device_->slot_range(LogReadMode::kDurable).upper_bound;
   }
 
+  // Check to make sure request was never seen before. Note that we do unique identifier check only
+  // for depth=0 as that's coming from the external client side. For all recycler internal calls
+  // (with depth > 0) we will simply accept the request.
+  //
+  if (this->largest_offset_as_unique_identifier_ < offset_as_unique_identifier || depth != 0) {
+    // Update the largest unique identifier and continue.
+    //
+    this->largest_offset_as_unique_identifier_ = offset_as_unique_identifier;
+  } else {
+    // Look like this is a repost of some old request thus update some status and ignore/return.
+    //
+    this->metrics_export().page_id_deletion_reissue.fetch_add(1);
+    return this->wal_device_->slot_range(LogReadMode::kDurable).upper_bound;
+  }
+
   Optional<slot_offset_type> sync_point = None;
 
   if (depth == 0) {
@@ -297,6 +365,7 @@ StatusOr<slot_offset_type> PageRecycler::recycle_pages(const Slice<const PageId>
 
     const PageRecyclerOptions& options = this->state_.no_lock().options;
 
+    LLFS_VLOG(1) << "recycle_pages entered grant==NULL case";
     for (PageId page_id : page_ids) {
       StatusOr<batt::Grant> local_grant = [&] {
         const usize needed_size = options.insert_grant_size();
@@ -319,8 +388,9 @@ StatusOr<slot_offset_type> PageRecycler::recycle_pages(const Slice<const PageId>
       BATT_REQUIRE_OK(local_grant);
       {
         auto locked_state = this->state_.lock();
-        StatusOr<slot_offset_type> append_slot =
-            this->insert_to_log(*local_grant, page_id, depth, locked_state);
+        // Writing to recycler log.
+        StatusOr<slot_offset_type> append_slot = this->insert_to_log(
+            *local_grant, page_id, depth, offset_as_unique_identifier, locked_state);
         BATT_REQUIRE_OK(append_slot);
 
         clamp_min_slot(&sync_point, *append_slot);
@@ -329,10 +399,13 @@ StatusOr<slot_offset_type> PageRecycler::recycle_pages(const Slice<const PageId>
   } else {
     BATT_CHECK_LT(depth, (i32)kMaxPageRefDepth) << BATT_INSPECT_RANGE(page_ids);
 
+    LLFS_VLOG(1) << "recycle_pages entered the valid grant case";
+
     auto locked_state = this->state_.lock();
     for (PageId page_id : page_ids) {
+      // Writing to recycler log.
       StatusOr<slot_offset_type> append_slot =
-          this->insert_to_log(*grant, page_id, depth, locked_state);
+          this->insert_to_log(*grant, page_id, depth, offset_as_unique_identifier, locked_state);
       BATT_REQUIRE_OK(append_slot);
 
       clamp_min_slot(&sync_point, *append_slot);
@@ -345,16 +418,18 @@ StatusOr<slot_offset_type> PageRecycler::recycle_pages(const Slice<const PageId>
 
 //==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- -  -  -   -
 //
-StatusOr<slot_offset_type> PageRecycler::recycle_page(PageId page_id, batt::Grant* grant, i32 depth)
+StatusOr<slot_offset_type> PageRecycler::recycle_page(PageId page_id,
+                                                      slot_offset_type unique_offset,
+                                                      batt::Grant* grant, i32 depth)
 {
   std::array<PageId, 1> page_ids{page_id};
-  return this->recycle_pages(batt::as_slice(page_ids), grant, depth);
+  return this->recycle_pages(batt::as_slice(page_ids), unique_offset, grant, depth);
 }
 
 //==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- -  -  -   -
 //
 StatusOr<slot_offset_type> PageRecycler::insert_to_log(
-    batt::Grant& grant, PageId page_id, i32 depth,
+    batt::Grant& grant, PageId page_id, i32 depth, slot_offset_type offset_as_unique_identifier,
     batt::Mutex<std::unique_ptr<State>>::Lock& locked_state)
 {
   BATT_CHECK(locked_state.is_held());
@@ -370,6 +445,7 @@ StatusOr<slot_offset_type> PageRecycler::insert_to_log(
               .refresh_slot = None,
               .batch_slot = None,
               .depth = depth,
+              .offset_as_unique_identifier = offset_as_unique_identifier,
           },
           [&](const batt::SmallVecBase<PageToRecycle*>& to_append) -> StatusOr<slot_offset_type> {
             if (to_append.empty()) {
@@ -387,7 +463,7 @@ StatusOr<slot_offset_type> PageRecycler::insert_to_log(
               BATT_REQUIRE_OK(append_slot);
               item->refresh_slot = append_slot->lower_bound;
               last_slot = slot_max(last_slot, append_slot->upper_bound);
-              LLFS_VLOG(1) << "Write " << item << " to the log; last_slot=" << last_slot;
+              LLFS_VLOG(1) << "Write " << *item << " to the log; last_slot=" << last_slot;
             }
             BATT_CHECK_NE(this->slot_writer_.slot_offset(), current_slot);