diff --git a/bench/ssd_streaming_pipeline_bench.cpp b/bench/ssd_streaming_pipeline_bench.cpp new file mode 100644 index 0000000..731cf1b --- /dev/null +++ b/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 +#include +#include +#include +#include +#include +#include + +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 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 buf(ps); + std::atomic consumed{0}; + std::atomic next{0}; + std::atomic done{false}; + + std::vector 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(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; +} diff --git a/bench/ssd_streaming_scan_bench.cpp b/bench/ssd_streaming_scan_bench.cpp new file mode 100644 index 0000000..112fe88 --- /dev/null +++ b/bench/ssd_streaming_scan_bench.cpp @@ -0,0 +1,44 @@ +#include "ssd_streaming_backend.h" +#include +#include +#include + +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 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(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; +} diff --git a/include/ssd_streaming_backend.h b/include/ssd_streaming_backend.h index 7102387..e0241f2 100644 --- a/include/ssd_streaming_backend.h +++ b/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 { diff --git a/src/ssd_streaming_backend.cpp b/src/ssd_streaming_backend.cpp index 041abff..95eb68c 100644 --- a/src/ssd_streaming_backend.cpp +++ b/src/ssd_streaming_backend.cpp @@ -7,12 +7,14 @@ #include #include #include +#include #include #include #include #include #include #include +#include #include #include #include @@ -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 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++;