Merge a8eaf2e9d2 into 8482cb11bc

src/crimson/os/futurized_store.h

@@ -52,6 +52,11 @@ public:
       return true;
     }
 
+    // Returns true when the local store is full (failsafe limit); checked at
+    // OSDOp boundary, where data-allocating ops are dropped with -EAGAIN so
+    // the client resends. Default false.
+    virtual bool is_storage_full() const { return false; }
+
     using CollectionRef = boost::intrusive_ptr<FuturizedCollection>;
     using base_errorator = crimson::errorator<crimson::ct_error::input_output_error>;
     using read_errorator = crimson::errorator<crimson::ct_error::enoent,

src/crimson/os/seastore/async_cleaner.cc

@@ -2010,12 +2010,41 @@ void RBMCleaner::mark_space_free(
 	ceph_assert(stats.used_bytes >= len);
 	stats.used_bytes -= len;
 	rbm->mark_space_free(addr, len);
+	background_callback->maybe_wake_blocked_io();
       }
       return;
     }
   }
 }
 
+void RBMCleaner::account_conflict_pending_free(std::size_t bytes)
+{
+  conflict_pending_free_bytes += bytes;
+}
+
+void RBMCleaner::deaccount_conflict_pending_free(std::size_t bytes)
+{
+  ceph_assert(conflict_pending_free_bytes >= bytes);
+  conflict_pending_free_bytes -= bytes;
+  conflict_drained.broadcast();
+  // Pending conflict frees are a release source try_reserve_projected_usage
+  // blocks on; without this wake a blocked IO whose only release source was
+  // a draining conflict free would miss its wakeup.
+  background_callback->maybe_wake_blocked_io();
+}
+
+seastar::future<> RBMCleaner::wait_for_conflict_drain()
+{
+  auto data_capacity = get_total_bytes() - get_journal_bytes();
+  if (data_capacity == 0) {
+    return seastar::now();
+  }
+  auto threshold = data_capacity / 20; // 5% of data capacity
+  return conflict_drained.wait([this, threshold] {
+    return conflict_pending_free_bytes <= threshold;
+  });
+}
+
 void RBMCleaner::commit_space_used(paddr_t addr, extent_len_t len) 
 {
   auto rbms = rb_group->get_rb_managers();
@@ -2033,25 +2062,36 @@ bool RBMCleaner::try_reserve_projected_usage(std::size_t projected_usage)
 {
   assert(background_callback->is_ready());
 
-  // Capacity check. Without this, concurrent transactions over-commit the
-  // RBM device: each reserves but the cleaner has no clean_space() yet, so
-  // a write that physically can't be served reaches the allocator and
-  // surfaces as `unexpected enospc` asserts in the data path (object_data
-  // _handler.cc et al.). Return false so the EPM BackgroundProcess blocks
-  // the IO until committed transactions release space.
-  //
-  // Headroom carves out room for metadata writes (LBA btree, backref) and
-  // for fragmentation slack the allocator can't pack into. 5% is a starting
-  // point; until RBMCleaner::clean_space() exists we cannot reclaim from
-  // fragmented free space, so headroom doubles as a fragmentation guard.
+  // Use the allocator's authoritative view instead of stats.used_bytes.
+  // alloc_paddr and mark_space_used from existing_block_list are unmetered
+  // paths that push stats.used_bytes past the actual allocator state, causing
+  // premature rejection under sustained overwrite.
+  // Headroom is 1% (down from 5%) because the AVL view already accounts for
+  // metadata extents; 1% covers in-flight surge before alloc_paddr commits.
   assert(get_total_bytes() > get_journal_bytes());
   auto data_capacity = get_total_bytes() - get_journal_bytes();
-  auto headroom = data_capacity / 20;
-  auto committed_and_projected = stats.used_bytes
+  auto headroom = data_capacity / 100;
+  auto committed_and_projected = get_allocator_used_bytes()
                                + stats.projected_used_bytes
                                + projected_usage;
   if (committed_and_projected + headroom > data_capacity) {
-    return false;
+    // Gating is only sound while waiting can help. RBM has no cleaner
+    // (can_clean_space() == false): space is only released when in-flight
+    // transactions complete and release what they hold. If no release
+    // source exists, blocking would deadlock (e.g. a sole writer on a
+    // full device waits forever). Admit instead and let the block
+    // allocator stay the final authority: a genuinely full device fails
+    // the allocation with ENOSPC through the existing alloc-failure path
+    // rather than wedging the IO path.
+    //
+    // A zero-usage reservation can never overcommit, and may belong to a
+    // space-releasing transaction (deletes are metadata-only); blocking
+    // those would wedge the escape hatch, so admit them unconditionally.
+    bool has_release_source = stats.projected_used_bytes > 0 ||
+                              conflict_pending_free_bytes > 0;
+    if (projected_usage != 0 && has_release_source) {
+      return false;
+    }
   }
   stats.projected_used_bytes += projected_usage;
   return true;
@@ -2207,10 +2247,11 @@ void RBMCleaner::register_metrics()
 		     sm::description("the size of the space"),
          {sm::label_instance("shard_store_index", std::to_string(store_index))}),
     sm::make_counter("available_bytes",
-		     [this] { return get_total_bytes() - get_journal_bytes() - stats.used_bytes; },
+		     [this] { return get_total_bytes() - get_journal_bytes() - get_used_bytes(); },
 		     sm::description("the size of the space is available"),
          {sm::label_instance("shard_store_index", std::to_string(store_index))}),
-    sm::make_counter("used_bytes", stats.used_bytes,
+    sm::make_counter("used_bytes",
+		     [this] { return get_used_bytes(); },
 		     sm::description("the size of the space occupied by live extents"),
          {sm::label_instance("shard_store_index", std::to_string(store_index))})
   });

src/crimson/os/seastore/async_cleaner.h

@@ -4,12 +4,14 @@
 #pragma once
 
 #include <boost/intrusive/set.hpp>
+#include <seastar/core/condition-variable.hh>
 #include <seastar/core/metrics_types.hh>
 
 #include "common/ceph_time.h"
 
 #include "osd/osd_types.h"
 
+#include "crimson/common/config_proxy.h"
 #include "crimson/os/seastore/cached_extent.h"
 #include "crimson/os/seastore/seastore_types.h"
 #include "crimson/os/seastore/segment_manager.h"
@@ -1292,6 +1294,18 @@ public:
 
   virtual void release_projected_usage(std::size_t) = 0;
 
+  // Returns true when the allocator is full (failsafe limit) and writes should
+  // be refused. Only RBMCleaner overrides this; SegmentCleaner returns false.
+  virtual bool is_storage_full() const { return false; }
+
+  // Backpressure gate for deferred-free accumulation (RBM only).
+  // Base-class no-ops so SegmentCleaner callers compile without change.
+  virtual void account_conflict_pending_free(std::size_t) {}
+  virtual void deaccount_conflict_pending_free(std::size_t) {}
+  virtual seastar::future<> wait_for_conflict_drain() {
+    return seastar::now();
+  }
+
   virtual bool should_block_io_on_clean() const = 0;
 
   virtual bool can_clean_space() const = 0;
@@ -1847,10 +1861,11 @@ public:
 
   store_statfs_t get_stat() const final {
     store_statfs_t st;
+    auto used = get_used_bytes();
     st.total = get_total_bytes();
-    st.available = get_total_bytes() - get_journal_bytes() - stats.used_bytes;
-    st.allocated = get_journal_bytes() + stats.used_bytes;
-    st.data_stored = get_journal_bytes() + stats.used_bytes;
+    st.available = get_total_bytes() - get_journal_bytes() - used;
+    st.allocated = get_journal_bytes() + used;
+    st.data_stored = get_journal_bytes() + used;
     return st;
   }
 
@@ -1866,6 +1881,10 @@ public:
 
   void mark_space_free(paddr_t, extent_len_t) final;
 
+  void account_conflict_pending_free(std::size_t bytes) final;
+  void deaccount_conflict_pending_free(std::size_t bytes) final;
+  seastar::future<> wait_for_conflict_drain() final;
+
   void commit_space_used(paddr_t, extent_len_t) final;
 
   bool try_reserve_projected_usage(std::size_t) final;
@@ -1959,6 +1978,43 @@ public:
     return total;
   }
 
+  // Bytes allocated in the AVL allocator across all RBM devices.
+  // get_size() returns the data section only (journal already excluded).
+  uint64_t get_allocator_used_bytes() const {
+    uint64_t used = 0;
+    for (auto *rbm : rb_group->get_rb_managers()) {
+      uint64_t data = rbm->get_size();
+      uint64_t free = static_cast<uint64_t>(rbm->get_free_blocks())
+                     * rbm->get_block_size();
+      used += (data > free) ? (data - free) : 0;
+    }
+    return used;
+  }
+
+  // Used bytes as reported to statfs and metrics: the allocator's
+  // authoritative view, matching the admission gating in
+  // try_reserve_projected_usage (stats.used_bytes drifts on unmetered
+  // paths). Falls back to stats.used_bytes until the allocator is
+  // populated during mount (mirrors is_storage_full).
+  uint64_t get_used_bytes() const {
+    return background_callback->is_ready()
+         ? get_allocator_used_bytes() : stats.used_bytes;
+  }
+
+  bool is_storage_full() const final {
+    // The allocator view is not authoritative until the background process
+    // has finished scanning space at mount; never report full before then
+    // (mirrors the is_ready assertion in try_reserve_projected_usage).
+    if (!background_callback->is_ready()) {
+      return false;
+    }
+    auto data_capacity = get_total_bytes() - get_journal_bytes();
+    if (data_capacity == 0) return false;
+    auto headroom = data_capacity / 100;
+    return get_allocator_used_bytes() + stats.projected_used_bytes
+         + headroom > data_capacity;
+  }
+
   // Testing interfaces
 
   bool check_usage(bool has_cold_tier) final;
@@ -1994,6 +2050,12 @@ private:
      */
     uint64_t projected_used_bytes = 0;
   } stats;
+
+  // Bytes held in deferred-free state (conflicted OOL writes still in flight).
+  // Gate: new OOL writes wait when this exceeds 5% of data_capacity.
+  std::size_t conflict_pending_free_bytes = 0;
+  seastar::condition_variable conflict_drained;
+
   seastar::metrics::metric_group metrics;
   void register_metrics();
 

src/crimson/os/seastore/cache.cc

@@ -1128,7 +1128,17 @@ void Cache::mark_transaction_conflicted(
     efforts.mutate_delta_bytes += delta_stat.bytes;
 
     if (t.get_pending_ool()) {
+      // Paddrs are still in-flight; mark_space_free happens in
+      // do_write().finally() once the write completes (safe for all devices,
+      // including NVMe MQ which may reorder overlapping writes).
       t.get_pending_ool()->is_conflicted = true;
+      std::size_t pending_bytes = 0;
+      for (auto &e : t.pre_alloc_list) {
+        pending_bytes += e->get_length();
+      }
+      if (pending_bytes > 0) {
+        epm.account_conflict_pending_free(pending_bytes);
+      }
     } else {
       for (auto &i: t.pre_alloc_list) {
 	epm.mark_space_free(i->get_paddr(), i->get_length());

src/crimson/os/seastore/extent_placement_manager.cc

@@ -1285,18 +1285,37 @@ RandomBlockOolWriter::alloc_write_ool_extents(
     try {
       co_await trans_intr::make_interruptible(
         token_bucket.get(size));
+      // Apply backpressure if the accumulated deferred-free byte count of
+      // conflicted OOL writes exceeds the safety threshold, preventing
+      // allocator exhaustion.
+      co_await trans_intr::make_interruptible(
+        rb_cleaner->wait_for_conflict_drain());
       seastar::lw_shared_ptr<rbm_pending_ool_t> ptr =
           seastar::make_lw_shared<rbm_pending_ool_t>();
       ptr->pending_extents = t.get_pre_alloc_list();
       assert(!t.is_conflicted());
       t.set_pending_ool(ptr);
+      // The conflicted check, the deferred frees and clear_pending_ool() must
+      // all run in the same continuation that completes the write: it makes
+      // them atomic w.r.t. reactor task interleaving (a conflict marked between
+      // write completion and coroutine resumption could otherwise observe a
+      // stale pending_ool whose frees were already decided), and it guarantees
+      // pending_ool is cleared on every exit path, including errors and
+      // conflict-driven unwinds. t outlives the future: the submit fiber is
+      // suspended awaiting this coroutine.
       co_await do_write(t, extents
-      ).finally([this, ptr=ptr] {
+      ).finally([this, ptr=ptr, &t] {
         if (ptr->is_conflicted) {
+          std::size_t freed = 0;
           for (auto &e : ptr->pending_extents) {
             rb_cleaner->mark_space_free(e->get_paddr(), e->get_length());
+            freed += e->get_length();
+          }
+          if (freed > 0) {
+            rb_cleaner->deaccount_conflict_pending_free(freed);
           }
         }
+        t.clear_pending_ool();
       });
     } catch (...) {
       token_bucket.release(size);
@@ -1401,7 +1420,7 @@ RandomBlockOolWriter::do_write(
       trans_stats.num_records += writes.size();
       return alloc_write_ertr::parallel_for_each(writes,
         [](auto& info) {
-        return info.rbm->write(info.offset, info.bp
+        return info.rbm->write(info.offset, std::move(info.bp)
         ).handle_error(
           alloc_write_ertr::pass_further{},
           crimson::ct_error::assert_all(

src/crimson/os/seastore/extent_placement_manager.h

@@ -694,6 +694,15 @@ public:
     background_process.release_projected_usage(usage);
   }
 
+  bool is_storage_full() const {
+    auto *cleaner = background_process.get_main_cleaner();
+    return cleaner ? cleaner->is_storage_full() : false;
+  }
+
+  void account_conflict_pending_free(std::size_t bytes) {
+    background_process.account_conflict_pending_free(bytes);
+  }
+
   backend_type_t get_main_backend_type() const {
     if (!background_process.is_no_background()) {
       return background_process.get_main_backend_type();
@@ -1041,6 +1050,12 @@ private:
       }
     }
 
+    void account_conflict_pending_free(std::size_t bytes) {
+      if (main_cleaner) {
+        main_cleaner->account_conflict_pending_free(bytes);
+      }
+    }
+
     void commit_space_used(paddr_t addr, extent_len_t len) {
       if (state < state_t::SCAN_SPACE) {
         return;
@@ -1112,6 +1127,10 @@ private:
       return main_cleaner->get_alive_ratio() >= 0.99;
     }
 
+    const AsyncCleaner* get_main_cleaner() const {
+      return main_cleaner.get();
+    }
+
     void maybe_wake_background() final {
       if (!is_running()) {
         return;

src/crimson/os/seastore/random_block_manager/block_rb_manager.cc

@@ -142,7 +142,7 @@ BlockRBManager::write_ertr::future<> BlockRBManager::write(
   }
   co_return co_await device->write(
     addr,
-    bptr);
+    std::move(bptr));
 }
 
 BlockRBManager::read_ertr::future<> BlockRBManager::read(

src/crimson/os/seastore/random_block_manager/block_rb_manager.h

@@ -83,10 +83,11 @@ public:
     return device->get_device_id();
   }
 
+  // Returns the current count of free blocks based on the authoritative
+  // allocator state (or 0 if the allocator is not yet initialized).
   uint64_t get_free_blocks() const override { 
-    // TODO: return correct free blocks after block allocator is introduced
     assert(device);
-    return get_size() / get_block_size();
+    return allocator ? (allocator->get_available_size() / get_block_size()) : 0;
   }
   const seastore_meta_t &get_meta() const override {
     return device->get_meta();

src/crimson/os/seastore/seastore.h

@@ -207,6 +207,10 @@ public:
       return 256;
     }
 
+    bool is_storage_full() const override final {
+      return transaction_manager ? transaction_manager->is_storage_full() : false;
+    }
+
     omap_root_t select_log_omap_root(Onode& onode) const;
 
   // only exposed to SeaStore

src/crimson/os/seastore/transaction.h

@@ -601,6 +601,8 @@ public:
     ool_block_list.clear();
     inplace_ool_block_list.clear();
     pre_alloc_list.clear();
+    // Clear out-of-line write handle to prevent leakage upon transaction reuse
+    pending_ool = nullptr;
     pre_inplace_rewrite_list.clear();
     retired_set.clear();
     existing_block_list.clear();
@@ -722,10 +724,19 @@ public:
     return static_cast<T&>(*view);
   }
 
+  // Track pending out-of-line writes. Once the write completes successfully,
+  // clear_pending_ool() must be called to disassociate the transaction.
   void set_pending_ool(seastar::lw_shared_ptr<rbm_pending_ool_t> ptr) {
+    assert(!pending_ool);
     pending_ool = ptr;
   }
 
+  // Prevents double-freeing physical extents if the transaction is conflicted
+  // after the physical write has already completed or failed.
+  void clear_pending_ool() {
+    pending_ool = nullptr;
+  }
+
   seastar::lw_shared_ptr<rbm_pending_ool_t> get_pending_ool() {
     return pending_ool;
   }

src/crimson/os/seastore/transaction_manager.h

@@ -1203,6 +1203,10 @@ public:
     return epm->get_stat();
   }
 
+  bool is_storage_full() const {
+    return epm->is_storage_full();
+  }
+
   ExtentTransViewRetriever& get_etvr() {
     return *cache;
   }

src/crimson/osd/ops_executer.cc

@@ -609,6 +609,28 @@ OpsExecuter::do_execute_op(OSDOp& osd_op)
     "handling op {} on object {}",
     ceph_osd_op_name(osd_op.op.op),
     get_target());
+  // Best-effort capacity gate (failsafe): drop data-allocating ops before a
+  // transaction is created if the local store is full. This is Crimson's
+  // analog to classic's osd_failsafe_full_ratio check (PrimaryLogPG.cc:2162),
+  // preventing allocator exhaustion assert aborts when the local store fills
+  // before the monitor's pool FLAG_FULL has propagated. We reply -EAGAIN so
+  // the client resends, matching both classic's silent drop and Crimson's own
+  // mon-driven full path (PG::run_executer_fut, pg.cc), rather than returning
+  // a result that depends on whether full flags have reached the client yet.
+  switch (osd_op.op.op) {
+  case CEPH_OSD_OP_CREATE:
+  case CEPH_OSD_OP_WRITE:
+  case CEPH_OSD_OP_WRITEFULL:
+  case CEPH_OSD_OP_WRITESAME:
+  case CEPH_OSD_OP_APPEND:
+  case CEPH_OSD_OP_COPY_FROM2:
+    if (pg->is_local_store_full() && !get_message().has_flag(CEPH_OSD_FLAG_FULL_TRY)) {
+      return crimson::ct_error::eagain::make();
+    }
+    break;
+  default:
+    break;
+  }
   switch (const ceph_osd_op& op = osd_op.op; op.op) {
   case CEPH_OSD_OP_SYNC_READ:
     [[fallthrough]];

src/crimson/osd/pg.h

@@ -240,6 +240,10 @@ public:
     return false;
   }
 
+  bool is_local_store_full() const {
+    return shard_services.is_local_storage_full(get_store_index());
+  }
+
   epoch_t get_last_peering_reset_epoch() const final {
     return get_last_peering_reset();
   }
@@ -269,7 +273,7 @@ public:
     return pgid;
   }
 
-  const unsigned int get_store_index() {
+  unsigned get_store_index() const {
     return store_index;
   }
   PGBackend& get_backend() {

src/crimson/osd/shard_services.h

@@ -533,6 +533,12 @@ public:
     return store;
   }
 
+  bool is_local_storage_full(store_index_t store_index) const {
+    const auto &shard_store =
+        local_state.b_store.f_store.get_sharded_store(store_index);
+    return shard_store.is_storage_full();
+  }
+
   struct shard_stats_t {
     double reactor_utilization;
   };

src/test/crimson/seastore/test_transaction_manager.cc

@@ -1913,6 +1913,46 @@ TEST_P(tm_random_block_device_test_t, scatter_allocation)
   });
 }
 
+TEST_P(tm_random_block_device_test_t, storage_full_failsafe_threshold)
+{
+  run_async([this] {
+    // A fresh store is far below the default failsafe ratio (0.97).
+    EXPECT_FALSE(tm->is_storage_full());
+
+    auto st = tm->store_stat();
+    ASSERT_GT(st.total, 0);
+    ASSERT_GE(st.total, st.available);
+
+    // Lower the ratio to just above the current used fraction, leaving half
+    // of the upcoming allocation as margin on each side of the threshold, so
+    // the allocation below is what crosses it.
+    constexpr extent_len_t EXTENT_SIZE = 16384;
+    constexpr unsigned NUM_EXTENTS = 64;
+    constexpr uint64_t alloc_bytes = uint64_t(EXTENT_SIZE) * NUM_EXTENTS;
+    const double used_fraction =
+      double(st.total - st.available) / double(st.total);
+    const double ratio =
+      used_fraction + double(alloc_bytes) / (2.0 * double(st.total));
+    crimson::common::local_conf().set_val(
+      "osd_failsafe_full_ratio", std::to_string(ratio)).get();
+    EXPECT_FALSE(tm->is_storage_full());
+
+    laddr_t ADDR = get_laddr_hint(0xFF * 4096);
+    auto t = create_transaction();
+    for (unsigned i = 0; i < NUM_EXTENTS; i++) {
+      alloc_extents(
+	t, (ADDR + i * EXTENT_SIZE).checked_to_laddr(), EXTENT_SIZE, 'a');
+    }
+    submit_transaction(std::move(t));
+    EXPECT_TRUE(tm->is_storage_full());
+
+    // The threshold tracks the config knob: restoring the default clears it.
+    crimson::common::local_conf().set_val(
+      "osd_failsafe_full_ratio", "0.97").get();
+    EXPECT_FALSE(tm->is_storage_full());
+  });
+}
+
 TEST_P(tm_single_device_test_t, basic)
 {
   constexpr size_t SIZE = 4096;