ssd_streaming_backend: scan-resistant eviction and look-ahead prefetch

bench/ssd_streaming_pipeline_bench.cpp

@@ -0,0 +1,125 @@
+/*
+ * Look-ahead prefetch pipeline benchmark for the SSD streaming backend.
+ *
+ * Models the CXL data-supply path of a ternary-weight LLM decode loop: every
+ * token streams a run of use-once weight tiles; each tile costs a modelled
+ * backing read latency plus a modelled compute time. It compares demand-only
+ * loading against a look-ahead prefetcher that stays a fixed depth ahead over
+ * several parallel channels, so the backing latency overlaps the compute of
+ * earlier tiles. The latency is a model parameter (backing_latency_ns), not a
+ * measured device latency, so the result is deterministic on any host.
+ * Args: [lat_ns] [comp_ns] [depth] [threads].
+ */
+#include "../include/ssd_streaming_backend.h"
+
+#include <atomic>
+#include <chrono>
+#include <cstdint>
+#include <cstdio>
+#include <cstdlib>
+#include <thread>
+#include <vector>
+
+using clk = std::chrono::steady_clock;
+
+static void spin_ns(uint64_t ns)
+{
+    const auto deadline = clk::now() + std::chrono::nanoseconds(ns);
+    while (clk::now() < deadline) {
+    }
+}
+
+int main(int argc, char **argv)
+{
+    const uint32_t ps = 4096;
+    const uint64_t weight_pages = 512;
+    const int tokens = 8;
+    const uint64_t lat = argc > 1 ? strtoull(argv[1], nullptr, 10) : 3000;
+    const uint64_t comp = argc > 2 ? strtoull(argv[2], nullptr, 10) : 2000;
+    const int depth = argc > 3 ? atoi(argv[3]) : 8;
+    const int pt = argc > 4 ? atoi(argv[4]) : 1;
+
+    std::vector<uint64_t> seq;
+    seq.reserve((size_t)tokens * weight_pages);
+    for (int t = 0; t < tokens; t++) {
+        for (uint64_t p = 0; p < weight_pages; p++) {
+            seq.push_back(p);
+        }
+    }
+    const size_t n = seq.size();
+
+    SsdStreamingConfig cfg;
+    cfg.backing_path = "/home/heke/temp/qemu-camp/scratch/cxl_pipe.img";
+    cfg.capacity_bytes = (weight_pages + 16) * ps;
+    cfg.page_size = ps;
+    cfg.cache_pages = 256;
+    cfg.read_ahead_pages = 0;
+    cfg.backing_latency_ns = lat;
+
+    auto run = [&](bool pipelined) {
+        SsdStreamingBackend be(cfg);
+        be.initialize();
+        std::vector<uint8_t> buf(ps);
+        std::atomic<size_t> consumed{0};
+        std::atomic<size_t> next{0};
+        std::atomic<bool> done{false};
+
+        std::vector<std::thread> pf;
+        if (pipelined) {
+            for (int k = 0; k < pt; k++) {
+                pf.emplace_back([&] {
+                    for (;;) {
+                        size_t i = next.fetch_add(1);
+                        if (i >= n || done.load()) {
+                            break;
+                        }
+                        while (i > consumed.load() + depth && !done.load()) {
+                            std::this_thread::yield();
+                        }
+                        be.prefetch(seq[i] * ps, ps);
+                    }
+                });
+            }
+        }
+
+        auto t0 = clk::now();
+        for (size_t i = 0; i < n; i++) {
+            uint64_t addr = seq[i] * ps;
+            be.set_streaming(addr, ps);
+            be.read(addr, buf.data(), ps);
+            spin_ns(comp);
+            consumed.store(i + 1);
+        }
+        auto t1 = clk::now();
+        done.store(true);
+        for (auto &t : pf) {
+            t.join();
+        }
+
+        double ms = std::chrono::duration<double, std::milli>(t1 - t0).count();
+        SsdStreamingStats st = be.get_stats();
+        be.shutdown();
+        return std::make_pair(ms, st);
+    };
+
+    /* warm the backing file so both arms fault the same way */
+    run(false);
+
+    auto d = run(false);
+    auto p = run(true);
+    double ms_demand = d.first, ms_pipe = p.first;
+    double bw_demand = (double)n * ps / (ms_demand / 1e3) / (1024 * 1024);
+    double bw_pipe = (double)n * ps / (ms_pipe / 1e3) / (1024 * 1024);
+
+    printf("pages=%zu lat=%lu ns comp=%lu ns depth=%d threads=%d\n",
+           n, (unsigned long)lat, (unsigned long)comp, depth, pt);
+    printf("demand-only : %8.2f ms  faults=%lu hits=%lu  supply=%.1f MiB/s\n",
+           ms_demand, (unsigned long)d.second.page_faults,
+           (unsigned long)d.second.cache_hits, bw_demand);
+    printf("pipelined   : %8.2f ms  faults=%lu hits=%lu  supply=%.1f MiB/s\n",
+           ms_pipe, (unsigned long)p.second.page_faults,
+           (unsigned long)p.second.cache_hits, bw_pipe);
+    printf("speedup     : %.2fx   supply gain: %.2fx\n",
+           ms_demand / ms_pipe, bw_pipe / bw_demand);
+    return 0;
+}

bench/ssd_streaming_scan_bench.cpp

@@ -0,0 +1,44 @@
+#include "ssd_streaming_backend.h"
+#include <chrono>
+#include <cstdio>
+#include <vector>
+
+int main(int argc, char **argv) {
+    SsdStreamingConfig cfg;
+    cfg.backing_path = "/home/heke/temp/qemu-camp/scratch/cxl_bench.img";
+    cfg.page_size = 4096;
+    cfg.capacity_bytes = 256ull * 1024 * 1024;
+    cfg.cache_pages = 512;
+    cfg.read_ahead_pages = 0;
+    cfg.use_io_uring = false;
+    cfg.use_odirect = false;
+    SsdStreamingBackend be(cfg);
+    if (!be.initialize()) { fprintf(stderr, "init failed\n"); return 1; }
+
+    const uint64_t PS = cfg.page_size;
+    const uint64_t hot_pages = 200;      /* reused KV/hot weights */
+    const uint64_t weight_base = 4096;   /* streamed ternary weights */
+    const uint64_t weight_pages = 3000;
+    const int tokens = 10;
+    std::vector<uint8_t> buf(PS);
+
+    auto t0 = std::chrono::steady_clock::now();
+    for (int tok = 0; tok < tokens; tok++) {
+        for (int r = 0; r < 3; r++)
+            for (uint64_t p = 0; p < hot_pages; p++) be.read(p * PS, buf.data(), PS);
+        for (uint64_t i = 0; i < weight_pages; i++) {
+            uint64_t addr = (weight_base + i) * PS;
+            be.set_streaming(addr, PS);
+            be.read(addr, buf.data(), PS);
+        }
+    }
+    auto t1 = std::chrono::steady_clock::now();
+    auto st = be.get_stats();
+    double ms = std::chrono::duration<double, std::milli>(t1 - t0).count();
+    printf("reads=%lu cache_hits=%lu page_faults=%lu evictions=%lu bytes_read=%lu hit_rate=%.1f%% time_ms=%.0f\n",
+           (unsigned long)st.reads, (unsigned long)st.cache_hits, (unsigned long)st.page_faults,
+           (unsigned long)st.evictions, (unsigned long)st.bytes_read,
+           100.0 * st.cache_hits / (st.cache_hits + st.page_faults), ms);
+    be.shutdown();
+    return 0;
+}

include/ssd_streaming_backend.h

@@ -36,6 +36,13 @@ struct SsdStreamingConfig {
     uint32_t read_ahead_pages = 16;
     bool use_io_uring = true;
     bool use_odirect = true;
+    /*
+     * Modelled CXL/SSD backing-read latency per page fault, in nanoseconds.
+     * When non-zero read_page() adds this deterministic delay so the
+     * data-supply path can be studied independently of the noisy host disk.
+     * This is a model parameter, not a measured device latency.
+     */
+    uint64_t backing_latency_ns = 0;
 };
 
 struct SsdStreamingStats {

src/ssd_streaming_backend.cpp

@@ -7,12 +7,14 @@
 #include <algorithm>
 #include <atomic>
 #include <cerrno>
+#include <chrono>
 #include <cstdlib>
 #include <cstring>
 #include <fcntl.h>
 #include <iostream>
 #include <limits>
 #include <memory>
+#include <thread>
 #include <sys/stat.h>
 #include <sys/types.h>
 #include <utility>
@@ -421,6 +423,16 @@ bool SsdStreamingBackend::submit_io_uring_rw(uint8_t opcode, uint8_t *buffer, of
 }
 
 bool SsdStreamingBackend::read_page(uint8_t *dst, off_t offset) {
+    if (config_.backing_latency_ns) {
+        /*
+         * Model async backing-store read latency: sleep rather than spin so the
+         * CPU is released for compute during the transfer, as a real io_uring
+         * DMA / CXL backing read would be. The metadata mutex is already dropped
+         * by the caller, so a concurrent prefetch overlaps foreground compute.
+         */
+        std::this_thread::sleep_for(
+            std::chrono::nanoseconds(config_.backing_latency_ns));
+    }
 #ifdef __linux__
     if (io_uring_enabled_) {
         std::lock_guard<std::mutex> io_lock(io_mutex_);
@@ -740,6 +752,17 @@ bool SsdStreamingBackend::read(uint64_t addr, uint8_t *dst, size_t size) {
         PagePtr page = pages_.at(page_no);
         memcpy(dst + copied, page->data.get() + offset, chunk);
         copied += chunk;
+
+        /*
+         * Scan-resistant streaming: a set_streaming()-hinted page is a
+         * use-once weight-scan page, so drop it immediately after it is
+         * served instead of letting it linger and evict hot/reused pages
+         * (KV cache, shared weights). Only clean, unreferenced pages.
+         */
+        if (page->streaming && !page->dirty && page->refcnt == 0 &&
+            page->state == SsdPageState::ResidentClean) {
+            evict_page_locked(page_no, lock);
+        }
     }
 
     stats_.reads++;