Merge pull request #26656 from WillemKauf/backport-pr-26513-v25.1.x-81

[v25.1.x] `storage`: set cached disk usage after compaction rounds & more cached disk usage improvements (MANUAL BACKPORT)

Author: WillemKauf

src/v/storage/compacted_index_writer.h

@@ -78,6 +78,7 @@ class compacted_index_writer {
     virtual void set_flag(compacted_index::footer_flags) = 0;
     virtual void print(std::ostream&) const = 0;
     const ss::sstring& filename() const { return _name; }
+    virtual size_t size_bytes() const = 0;
 
 private:
     friend std::ostream&

src/v/storage/disk_log_impl.cc

@@ -530,7 +530,6 @@ ss::future<> disk_log_impl::adjacent_merge_compact(
           segment->reader().filename(),
           result);
         if (result.did_compact()) {
-            segment->clear_cached_disk_usage();
             _compaction_ratio.update(result.compaction_ratio());
             const ssize_t removed_bytes = ssize_t(result.size_before)
                                           - ssize_t(result.size_after);
@@ -1064,7 +1063,12 @@ ss::future<compaction_result> disk_log_impl::do_compact_adjacent_segments(
     }
     // transfer segment state from replacement to target
     locks = co_await internal::transfer_segment(
-      target, replacement, cfg, *_probe, std::move(locks));
+      target,
+      replacement,
+      cfg,
+      *_probe,
+      std::move(locks),
+      ret.cmp_idx_size_after);
 
     auto segment_to_remove = segments.back();
     // We are going to remove one of the segments, but its bytes have not
@@ -1576,7 +1580,8 @@ ss::future<> disk_log_impl::rewrite_segment_with_offset_map(
     co_await seg->index().drop_all_data();
 
     // Rename the data file.
-    co_await internal::do_swap_data_file_handles(tmpname, seg, cfg, *_probe);
+    co_await internal::do_swap_data_file_handles(
+      tmpname, seg, cfg, *_probe, compacted_idx_writer->size_bytes());
 
     // Persist the state of our indexes in their new names.
     seg->index().swap_index_state(std::move(new_idx));

src/v/storage/segment.cc

@@ -120,7 +120,9 @@ ss::future<usage> segment::persistent_size() {
      * segment appender will transparently extend the size of the segment file
      * using fallocate, always stat the on disk size for the head partition.
      */
-    if (!_appender && _data_disk_usage_size.has_value()) {
+    if (_appender) {
+        u.data = _appender->size_bytes();
+    } else if (_data_disk_usage_size.has_value()) {
         u.data = _data_disk_usage_size.value();
     } else {
         try {
@@ -139,7 +141,9 @@ ss::future<usage> segment::persistent_size() {
      * however, we pay for that stat() with no guarantee that the information
      * will be used.
      */
-    if (_compaction_index_size.has_value()) {
+    if (_compaction_index) {
+        u.compaction = _compaction_index.value()->size_bytes();
+    } else if (_compaction_index_size.has_value()) {
         u.compaction = _compaction_index_size.value();
     } else {
         auto path = reader().path().to_compacted_index();
@@ -189,6 +193,14 @@ void segment::clear_cached_disk_usage() {
     _compaction_index_size.reset();
 }
 
+void segment::set_cached_disk_usage(
+  size_t new_seg_size, std::optional<size_t> new_compacted_index_size) {
+    _data_disk_usage_size = new_seg_size;
+    if (new_compacted_index_size.has_value()) {
+        _compaction_index_size = new_compacted_index_size.value();
+    }
+}
+
 ss::future<size_t> segment::remove_persistent_state() {
     vassert(is_closed(), "Cannot clear state from unclosed segment");
 
@@ -243,13 +255,22 @@ ss::future<> segment::do_release_appender(
         std::optional<std::unique_ptr<compacted_index_writer>>&
           compacted_index) {
           return appender->close()
-            .then([this] { return _idx.flush(); })
-            .then([this, &compacted_index] {
+            .then([this] {
+                clear_cached_disk_usage();
+                return _idx.flush();
+            })
+            .then([&compacted_index] {
                 if (compacted_index) {
                     return compacted_index.value()->close();
                 }
-                clear_cached_disk_usage();
                 return ss::now();
+            })
+            .then([this, &compacted_index, &appender] {
+                std::optional<size_t> cmp_idx_size{std::nullopt};
+                if (compacted_index) {
+                    cmp_idx_size = compacted_index.value()->size_bytes();
+                }
+                set_cached_disk_usage(appender->size_bytes(), cmp_idx_size);
             });
       });
 }

src/v/storage/segment.h

@@ -37,6 +37,10 @@ struct segment_closed_exception final : std::exception {
     }
 };
 
+namespace testing_details {
+class segment_accessor;
+}; // namespace testing_details
+
 class segment {
 public:
     using generation_id = named_type<uint64_t, struct segment_gen_tag>;
@@ -271,6 +275,17 @@ class segment {
     }
 
     void clear_cached_disk_usage();
+    // Sets the cached disk usage for the `segment` and optionally the
+    // `compacted_index`. These are usually the values from the underlying
+    // `segment_{appender/reader}` for either of these objects. The base
+    // `segment_index` sets its disk usage in `flush_to_file()`.
+    //
+    // This function is widely used within the compaction subsystem to cut down
+    // on future calls to `stat()`, since the cached disk usage is cleared
+    // during every round of compaction, and we know from in-memory state how
+    // large these objects are going to be on disk.
+    void set_cached_disk_usage(
+      size_t new_seg_size, std::optional<size_t> new_compacted_index_size);
 
     /// Fallback timestamp method, for use if the timestamps in the index
     /// appear to be invalid (e.g. too far in the future)
@@ -350,6 +365,7 @@ class segment {
     std::optional<ss::lowres_clock::time_point> _first_write;
 
     friend std::ostream& operator<<(std::ostream&, const segment&);
+    friend class testing_details::segment_accessor;
 };
 
 /**

src/v/storage/segment_appender.cc

@@ -326,7 +326,12 @@ ss::future<> segment_appender::close() {
     _closed = true;
     return hard_flush()
       .then([this] { return do_truncation(_committed_offset); })
-      .then([this] { return _out.close(); });
+      .then([this] {
+          _fallocation_offset = _committed_offset;
+          _flushed_offset = _committed_offset;
+          _stable_offset = _committed_offset;
+          return _out.close();
+      });
 }
 
 ss::future<> segment_appender::do_next_adaptive_fallocation() {

src/v/storage/segment_appender.h

@@ -89,6 +89,10 @@ class segment_appender {
         return _committed_offset + _bytes_flush_pending;
     }
 
+    // The size of the underlying file on disk, which is the
+    // _fallocation_offset.
+    size_t size_bytes() const { return _fallocation_offset; }
+
     /**
      * @brief Appends a batch.
      *

src/v/storage/segment_index.cc

@@ -181,6 +181,7 @@ ss::future<bool> segment_index::materialize_index() {
 ss::future<bool> segment_index::materialize_index_from_file(ss::file f) {
     auto size = co_await f.size();
     auto buf = co_await f.dma_read_bulk<char>(0, size);
+    _disk_usage_size = size;
     if (buf.empty()) {
         co_return false;
     }
@@ -229,6 +230,9 @@ ss::future<> segment_index::flush_to_file(ss::file backing_file) {
     for (const auto& f : b) {
         co_await out.write(f.get(), f.size());
     }
+
+    _disk_usage_size = b.size_bytes();
+
     co_await out.flush();
 }
 

src/v/storage/segment_reader.cc

@@ -51,8 +51,8 @@ segment_reader::~segment_reader() noexcept {
 ss::future<> segment_reader::load_size() {
     ss::gate::holder guard{_gate};
 
-    auto s = co_await stat();
-    set_file_size(s.st_size);
+    auto s = co_await fsize();
+    set_file_size(s);
 };
 
 ss::future<segment_reader_handle>
@@ -132,6 +132,15 @@ ss::future<> segment_reader::put() {
     }
 }
 
+ss::future<size_t> segment_reader::fsize() {
+    ss::gate::holder guard{_gate};
+
+    auto handle = co_await get();
+    auto r = co_await _data_file.size();
+    co_await handle.close();
+    co_return r;
+}
+
 ss::future<struct stat> segment_reader::stat() {
     ss::gate::holder guard{_gate};
 

src/v/storage/segment_reader.h

@@ -132,6 +132,9 @@ class segment_reader {
     /// close the underlying file handle
     ss::future<> close();
 
+    // Performs a syscall to get the size of _data_file on disk.
+    ss::future<size_t> fsize();
+
     /// perform syscall stat
     ss::future<struct stat> stat();
 

src/v/storage/segment_set.cc

@@ -281,8 +281,8 @@ static ss::future<segment_set> unsafe_do_recover(
           to_recover.begin(),
           to_recover.end(),
           [](ss::lw_shared_ptr<segment>& segment) {
-              auto stat = segment->reader().stat().get();
-              if (stat.st_size != 0) {
+              auto size = segment->reader().fsize().get();
+              if (size != 0) {
                   return true;
               }
               vlog(stlog.info, "Removing empty segment: {}", segment);