forked from mooncake-track/Mooncake
Compare commits
No commits in common. "feature/memory-fabric-abstraction" and "main" have entirely different histories.
feature/me
...
main
|
|
@ -1,156 +0,0 @@
|
|||
# Memory Fabric Capability Layer
|
||||
|
||||
## Overview
|
||||
|
||||
The Transfer Engine dispatches a transfer to a backend by matching the target
|
||||
segment's `protocol` string (`rdma`, `tcp`, `cxl`, `nvlink`, `ub`, ...) against
|
||||
the installed transports. What a backend can do, and whether it is usable right
|
||||
now, is implicit today — encoded in which methods it overrides and which build
|
||||
flag is set. There is no machine-readable capability descriptor, no single test
|
||||
that proves every backend honours the same contract, and no capability-driven
|
||||
selector that can route around a failed link without code changes.
|
||||
|
||||
This layer adds that missing piece as a thin, additive extension of the existing
|
||||
`Transport` base class:
|
||||
|
||||
- a uniform **`Capability`** descriptor returned by `Transport::probe()` and a
|
||||
**`HealthStatus`** returned by `Transport::health()`;
|
||||
- a backend-agnostic **conformance suite** that runs the same cases against every
|
||||
transport through the real `TransferEngine`;
|
||||
- a capability- and health-driven **`PathSelector`** that turns the probed view
|
||||
of the installed backends into an explainable routing decision with fallback;
|
||||
- a versioned **plugin ABI** so a transport can ship as a `.so` and be loaded,
|
||||
probed, and conformance-checked at runtime.
|
||||
|
||||
It also promotes **CXL / NUMA-remote memory to a first-class Store tier** (see
|
||||
[Memory Tiering](mooncake-store.md)) and adds an honestly-emulated **UALink**
|
||||
backend so the contract can be exercised end to end on hosts without that
|
||||
silicon.
|
||||
|
||||
Everything is additive: `probe()` and `health()` are non-pure virtuals with
|
||||
sensible defaults, so existing transports compile and behave unchanged.
|
||||
|
||||
---
|
||||
|
||||
## Capability descriptor
|
||||
|
||||
`Capability` (`include/fabric/capability.h`) is a flat POD so it can be
|
||||
serialised to JSON, compared, and shipped across a wire for remote topology
|
||||
assembly:
|
||||
|
||||
```cpp
|
||||
struct Capability {
|
||||
MemoryFabricKind kind; // TCP/RDMA/NVLINK/PCIE/CXL/UB/UALINK/SHM/NUMA
|
||||
std::string name; // backend instance, e.g. "rdma:mlx5_0"
|
||||
|
||||
bool supports_device_memory, supports_host_memory, supports_p2p;
|
||||
bool supports_ordered_write, supports_atomic, supports_zero_copy;
|
||||
bool supports_multi_rail, remote_capable;
|
||||
|
||||
bool emulated; // honesty flag: not real silicon on this host
|
||||
|
||||
uint64_t max_transfer_size, alignment;
|
||||
uint64_t max_bandwidth_mbps, latency_ns; // calibrated by probe()
|
||||
int numa_node, device_id;
|
||||
std::string notes;
|
||||
};
|
||||
```
|
||||
|
||||
`HealthStatus ∈ {UNINITIALIZED, HEALTHY, DEGRADED, UNREACHABLE}` generalises the
|
||||
intent of `TransferEngine::probePeerAliveByID()` to a single uniform signal.
|
||||
|
||||
`probe()` *calibrates honestly*: where it reports bandwidth and latency it times
|
||||
real transfers through the backend's own data path (a shared helper in
|
||||
`include/fabric/probe_calibration.h`), so an emulated backend reports the
|
||||
throughput of its emulation, never an invented hardware figure. Backends without
|
||||
real silicon on the host set `emulated = true` and explain how in `notes`.
|
||||
|
||||
---
|
||||
|
||||
## Conformance suite
|
||||
|
||||
One suite (`tests/fabric_conformance/`) runs the same cases against every
|
||||
backend, driving the real `TransferEngine` in a single process using the
|
||||
server-free `P2PHANDSHAKE` metadata mode. A backend *conforms* iff no case fails
|
||||
and every *required* case passes. A case that exercises a feature a backend does
|
||||
not advertise is skipped rather than failed — conformance verifies *claimed*
|
||||
capabilities and never penalises an honest non-claim.
|
||||
|
||||
| case | what it proves | required |
|
||||
|---|---|---|
|
||||
| `capability` | `probe()` is self-consistent and stable across calls | yes |
|
||||
| `register` | a zero-length transfer is a clean no-op | yes |
|
||||
| `roundtrip` | byte-exact WRITE→READ across sizes from 1 B to 4 MiB | yes |
|
||||
| `batch` | a 16-request batch in one `submitTransfer`, byte-exact | yes |
|
||||
| `ordering` | for `ordered_write` backends: observing a later write implies all earlier writes are visible | no (skip) |
|
||||
| `concurrency` | 8 threads × 40 iterations on disjoint regions, byte-exact | no |
|
||||
|
||||
`conformance_runner` prints a backend × case matrix plus a JSON report;
|
||||
`fabric_conformance_test` wraps the same suite as a GoogleTest gate. Backends
|
||||
unreachable on the host (no hardware / no peer) are skipped, not failed, so the
|
||||
gate is green on any host while still asserting the contract wherever a backend
|
||||
can actually run.
|
||||
|
||||
The harness hides the two addressing conventions the engine already uses: a
|
||||
windowed backend (CXL / UALink) addresses by offset into a shared window, while
|
||||
RDMA / TCP address by absolute registered address. Each case is written once
|
||||
against a uniform `write()` / `read()` and therefore exercises every backend.
|
||||
|
||||
---
|
||||
|
||||
## Path selection
|
||||
|
||||
`PathSelector` (`include/fabric/path_selector.h`) classifies an endpoint pair and
|
||||
walks a priority ladder, skipping any kind that is absent or not `HEALTHY`, and
|
||||
emits one explainable log line per decision:
|
||||
|
||||
```
|
||||
same_node_gpu_gpu : NVLINK > PCIE > CXL/SHM > TCP
|
||||
cross_node_gpu_gpu: RDMA > TCP
|
||||
host_host : CXL/SHM > UB > UALINK > RDMA > NUMA > TCP
|
||||
memory_tier : CXL > NUMA > RDMA > TCP
|
||||
```
|
||||
|
||||
Because the selector queries `health()` at selection time, injecting a fault into
|
||||
a transport (or losing a link) takes effect live: a transfer that selected
|
||||
`nvlink_intra` falls to the next healthy rung and logs *what* it skipped and
|
||||
*why*. This is the "layered cost + backend replacement / self-healing" shape a
|
||||
heterogeneous deployment needs.
|
||||
|
||||
---
|
||||
|
||||
## Dynamic loading
|
||||
|
||||
`PluginLoader` (`include/fabric/plugin_loader.h`) loads a transport that ships as
|
||||
a `.so` exporting a small versioned C ABI (`MOONCAKE_DEFINE_TRANSPORT_PLUGIN`):
|
||||
it `dlopen`s the object, checks the ABI version, instantiates the transport, and
|
||||
runs `probe()` / `health()`. A backend that cannot initialise on the host is
|
||||
skipped, never fatal, and the same conformance suite runs over plugin-loaded
|
||||
backends.
|
||||
|
||||
---
|
||||
|
||||
## UALink backend
|
||||
|
||||
UALink (Ultra Accelerator Link) has no upstream backend, so it is honest green
|
||||
field. `UaLinkTransport` reuses the shared-window data-plane shape of the CXL
|
||||
transport but routes every chunk through a `UALinkSimulator` that *enforces* the
|
||||
headline UALink semantic — ordered delivery (release publish + monotonic
|
||||
sequence) — and exposes a switch topology. The capability is flagged
|
||||
`emulated`; the conformance `ordering` case verifies the guarantee. When real
|
||||
silicon and a driver appear, only the chunk-move primitive changes; the
|
||||
semantics and the tests stay identical.
|
||||
|
||||
---
|
||||
|
||||
## Mapping to the existing tree
|
||||
|
||||
| existing | this layer | change |
|
||||
|---|---|---|
|
||||
| `Transport` base (`transport.h`) | `Transport` + `probe()` / `health()` | two non-pure virtuals with defaults |
|
||||
| `CxlTransport` (window + memcpy) | `+probe()/health()` + host-window fallback | exercisable without `/dev/dax` |
|
||||
| `RdmaTransport`, `TcpTransport` | `+probe()` (and RDMA `health()`) | report real capability |
|
||||
| `MultiTransport::installTransport` | `+ualink` branch (`USE_UALINK`) | one new backend |
|
||||
| Store `storage_backend` eviction | `TieredStore` + CXL tier | CXL added as a tier |
|
||||
|
||||
Each piece is independently reviewable and revertible.
|
||||
|
|
@ -73,7 +73,6 @@ option(USE_UBSHMEM "option for using ascend npu with shmem" OFF)
|
|||
option(USE_ASCEND_HETEROGENEOUS "option for transferring between ascend npu and gpu" OFF)
|
||||
option(USE_MNNVL "option for using Multi-Node NVLink transport" OFF)
|
||||
option(USE_CXL "option for using CXL protocol" OFF)
|
||||
option(USE_UALINK "option for using UALink protocol transport" OFF)
|
||||
option(USE_EFA "option for using AWS EFA transport" OFF)
|
||||
option(USE_UB "option for using UB protocol transport" OFF)
|
||||
|
||||
|
|
@ -269,11 +268,6 @@ if (USE_CXL)
|
|||
message(STATUS "CXL support is enabled")
|
||||
endif()
|
||||
|
||||
if (USE_UALINK)
|
||||
add_compile_definitions(USE_UALINK)
|
||||
message(STATUS "UALink support is enabled")
|
||||
endif()
|
||||
|
||||
if (USE_TCP)
|
||||
add_compile_definitions(USE_TCP)
|
||||
endif()
|
||||
|
|
|
|||
|
|
@ -21,15 +21,9 @@ target_link_libraries(
|
|||
gflags::gflags glog::glog pthread)
|
||||
|
||||
# Add allocation strategy benchmark executable
|
||||
# This benchmark tests AllocationStrategy performance with configurable
|
||||
# This benchmark tests AllocationStrategy performance with configurable
|
||||
# segment counts, allocation sizes, replica counts, and workload patterns
|
||||
add_executable(allocation_strategy_bench allocation_strategy_bench.cpp)
|
||||
target_link_libraries(
|
||||
allocation_strategy_bench PRIVATE mooncake_store cachelib_memory_allocator
|
||||
gflags::gflags glog::glog pthread)
|
||||
|
||||
# CXL/NUMA memory tiering benchmark: compares a DRAM+NVMe store against a
|
||||
# DRAM+CXL(NUMA-remote)+NVMe store on a Zipfian KVCache-like trace.
|
||||
add_executable(tier_bench tier_bench.cpp)
|
||||
target_link_libraries(tier_bench PRIVATE mooncake_store gflags::gflags
|
||||
glog::glog pthread)
|
||||
|
|
|
|||
|
|
@ -1,286 +0,0 @@
|
|||
// Copyright 2024 KVCache.AI
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
// Measure the value of the CXL/NUMA "L2.5" tier on a KVCache-like workload. For
|
||||
// a sweep of local-DRAM sizes the same Zipfian access trace is run through two
|
||||
// stores:
|
||||
// baseline : DRAM + NVMe
|
||||
// proposed : DRAM + CXL(emulated NUMA-remote) + NVMe
|
||||
// and the tool reports, per DRAM size, where reads were served, the NVMe read
|
||||
// count, and the average GET latency. Latencies are measured against real
|
||||
// backing storage (malloc / numa_alloc_onnode / O_DIRECT file), so the
|
||||
// comparison is honest. Two results fall out: at equal DRAM the CXL tier cuts
|
||||
// NVMe reads and latency; iso-latency, the proposed store needs less DRAM.
|
||||
|
||||
#include <gflags/gflags.h>
|
||||
|
||||
#include <algorithm>
|
||||
#include <cmath>
|
||||
#include <cstdint>
|
||||
#include <cstdio>
|
||||
#include <cstring>
|
||||
#include <fstream>
|
||||
#include <random>
|
||||
#include <string>
|
||||
#include <vector>
|
||||
|
||||
#include "tier/memory_tier.h"
|
||||
|
||||
using namespace mooncake::store;
|
||||
|
||||
DEFINE_string(csv, "", "write per-config results to this CSV path");
|
||||
DEFINE_string(nvme_path, "/tmp/mooncake_tier_nvme.bin",
|
||||
"backing file for the NVMe tier (opened O_DIRECT)");
|
||||
DEFINE_uint64(working_set, 4000, "working set in blocks");
|
||||
DEFINE_uint64(trace, 40000, "access trace length");
|
||||
DEFINE_double(zipf, 1.10, "Zipf skew parameter");
|
||||
DEFINE_uint64(cxl_blocks, 1024,
|
||||
"CXL tier size in blocks for the proposed store");
|
||||
|
||||
namespace {
|
||||
|
||||
constexpr uint64_t kBlock = 64 * 1024; // 64 KiB KV page
|
||||
|
||||
void fillBlock(std::vector<uint8_t> &buf, uint64_t id) {
|
||||
uint64_t x = id * 0x9E3779B97F4A7C15ull + 1;
|
||||
for (size_t i = 0; i < buf.size(); i += 8) {
|
||||
x ^= x >> 30;
|
||||
x *= 0xBF58476D1CE4E5B9ull;
|
||||
x ^= x >> 27;
|
||||
std::memcpy(buf.data() + i, &x, std::min<size_t>(8, buf.size() - i));
|
||||
}
|
||||
std::memcpy(buf.data(), &id, sizeof(id)); // tag for verification
|
||||
}
|
||||
|
||||
std::vector<uint64_t> makeZipfTrace(uint64_t W, uint64_t T, double s,
|
||||
uint32_t seed) {
|
||||
std::vector<double> cdf(W);
|
||||
double norm = 0;
|
||||
for (uint64_t i = 0; i < W; ++i) norm += 1.0 / std::pow(i + 1, s);
|
||||
double acc = 0;
|
||||
for (uint64_t i = 0; i < W; ++i) {
|
||||
acc += (1.0 / std::pow(i + 1, s)) / norm;
|
||||
cdf[i] = acc;
|
||||
}
|
||||
std::mt19937_64 rng(seed);
|
||||
std::uniform_real_distribution<double> U(0.0, 1.0);
|
||||
std::vector<uint64_t> trace(T);
|
||||
for (uint64_t t = 0; t < T; ++t) {
|
||||
double u = U(rng);
|
||||
uint64_t lo = 0, hi = W - 1;
|
||||
while (lo < hi) {
|
||||
uint64_t mid = (lo + hi) / 2;
|
||||
if (cdf[mid] < u)
|
||||
lo = mid + 1;
|
||||
else
|
||||
hi = mid;
|
||||
}
|
||||
trace[t] = lo;
|
||||
}
|
||||
return trace;
|
||||
}
|
||||
|
||||
struct Result {
|
||||
std::string config;
|
||||
uint64_t dram_blocks, cxl_blocks, nvme_blocks;
|
||||
StoreStats st;
|
||||
double dram_ns = 0, cxl_ns = 0, nvme_ns = 0;
|
||||
};
|
||||
|
||||
Result runConfig(const std::string &config, uint64_t dram_blocks,
|
||||
uint64_t cxl_blocks, uint64_t W,
|
||||
const std::vector<uint64_t> &trace,
|
||||
const std::string &nvme_path) {
|
||||
std::vector<std::shared_ptr<MemoryTier>> tiers;
|
||||
tiers.push_back(makeDramTier(kBlock, dram_blocks));
|
||||
if (cxl_blocks > 0) tiers.push_back(makeCxlTier(kBlock, cxl_blocks));
|
||||
tiers.push_back(makeNvmeTier(kBlock, W, nvme_path)); // whole working set
|
||||
for (auto &t : tiers)
|
||||
if (!t->ready())
|
||||
std::fprintf(stderr, "tier %s not ready\n", t->name().c_str());
|
||||
|
||||
TieredStore store(std::move(tiers), /*promote_on_hit=*/true);
|
||||
|
||||
std::vector<uint8_t> buf(kBlock), out(kBlock);
|
||||
for (uint64_t id = 0; id < W; ++id) {
|
||||
fillBlock(buf, id);
|
||||
store.put(id, buf.data());
|
||||
}
|
||||
|
||||
// Warm up (let hot blocks migrate up) without counting.
|
||||
uint64_t warm = trace.size() / 2;
|
||||
for (uint64_t t = 0; t < warm; ++t) store.get(trace[t], out.data());
|
||||
store.resetStats();
|
||||
|
||||
uint64_t bad = 0;
|
||||
for (uint64_t t = warm; t < trace.size(); ++t) {
|
||||
if (store.get(trace[t], out.data())) {
|
||||
uint64_t tag;
|
||||
std::memcpy(&tag, out.data(), sizeof(tag));
|
||||
if (tag != trace[t]) ++bad;
|
||||
} else {
|
||||
++bad;
|
||||
}
|
||||
}
|
||||
if (bad)
|
||||
std::fprintf(stderr, "[WARN] %s: %lu verification mismatches\n",
|
||||
config.c_str(), (unsigned long)bad);
|
||||
|
||||
Result r;
|
||||
r.config = config;
|
||||
r.dram_blocks = dram_blocks;
|
||||
r.cxl_blocks = cxl_blocks;
|
||||
r.nvme_blocks = W;
|
||||
r.st = store.stats();
|
||||
for (auto &tier : store.tiers()) {
|
||||
double ns = tier->stats().avgReadNs();
|
||||
if (tier->kind() == TierKind::DRAM)
|
||||
r.dram_ns = ns;
|
||||
else if (tier->kind() == TierKind::CXL)
|
||||
r.cxl_ns = ns;
|
||||
else if (tier->kind() == TierKind::NVME)
|
||||
r.nvme_ns = ns;
|
||||
}
|
||||
return r;
|
||||
}
|
||||
|
||||
} // namespace
|
||||
|
||||
int main(int argc, char **argv) {
|
||||
gflags::ParseCommandLineFlags(&argc, &argv, true);
|
||||
uint64_t W = FLAGS_working_set;
|
||||
uint64_t T = FLAGS_trace;
|
||||
double skew = FLAGS_zipf;
|
||||
uint64_t cxl_cap = FLAGS_cxl_blocks;
|
||||
std::vector<uint64_t> dram_sweep = {64, 128, 256, 512, 1024};
|
||||
|
||||
auto trace = makeZipfTrace(W, T, skew, 1234);
|
||||
|
||||
std::printf("=== Mooncake Store -- CXL/NUMA Tiering Benchmark ===\n");
|
||||
std::printf(
|
||||
"block=%lu KiB working_set=%lu blocks (%.0f MiB) trace=%lu "
|
||||
"zipf-s=%.2f\n",
|
||||
(unsigned long)(kBlock / 1024), (unsigned long)W,
|
||||
(double)W * kBlock / (1024.0 * 1024.0), (unsigned long)T, skew);
|
||||
std::printf("proposed adds a CXL tier of %lu blocks (%.0f MiB)\n\n",
|
||||
(unsigned long)cxl_cap,
|
||||
(double)cxl_cap * kBlock / (1024.0 * 1024.0));
|
||||
|
||||
std::ofstream os;
|
||||
bool have_csv = !FLAGS_csv.empty();
|
||||
if (have_csv) {
|
||||
os.open(FLAGS_csv);
|
||||
os << "config,dram_blocks,cxl_blocks,nvme_blocks,block_bytes,"
|
||||
"measured_gets,dram_hits,cxl_hits,nvme_hits,nvme_reads,avg_get_"
|
||||
"us,"
|
||||
"dram_ns,cxl_ns,nvme_ns,promotions,demotions\n";
|
||||
}
|
||||
|
||||
auto emit = [&](const Result &r) {
|
||||
if (!have_csv) return;
|
||||
const auto &s = r.st;
|
||||
os << r.config << "," << r.dram_blocks << "," << r.cxl_blocks << ","
|
||||
<< r.nvme_blocks << "," << kBlock << "," << (s.gets - s.get_miss)
|
||||
<< "," << s.hits_dram << "," << s.hits_cxl << "," << s.hits_nvme
|
||||
<< "," << s.nvme_reads << "," << s.avgGetUs() << "," << r.dram_ns
|
||||
<< "," << r.cxl_ns << "," << r.nvme_ns << "," << s.promotions << ","
|
||||
<< s.demotions << "\n";
|
||||
};
|
||||
|
||||
std::printf("%-9s %5s %5s | %8s %8s %8s | %10s %10s\n", "config", "dram",
|
||||
"cxl", "dramHit", "cxlHit", "nvmeRd", "avgGET(us)", "nvmeRd%");
|
||||
std::printf(
|
||||
"------------------------------------------------------------------"
|
||||
"--------\n");
|
||||
|
||||
std::vector<Result> base, prop;
|
||||
for (uint64_t d : dram_sweep) {
|
||||
Result rb = runConfig("baseline", d, 0, W, trace, FLAGS_nvme_path);
|
||||
Result rp =
|
||||
runConfig("proposed", d, cxl_cap, W, trace, FLAGS_nvme_path);
|
||||
base.push_back(rb);
|
||||
prop.push_back(rp);
|
||||
emit(rb);
|
||||
emit(rp);
|
||||
auto pct = [](uint64_t a, uint64_t tot) {
|
||||
return tot ? 100.0 * a / tot : 0.0;
|
||||
};
|
||||
uint64_t gb = rb.st.gets - rb.st.get_miss;
|
||||
uint64_t gp = rp.st.gets - rp.st.get_miss;
|
||||
std::printf("%-9s %5lu %5lu | %8lu %8lu %8lu | %10.3f %9.1f%%\n",
|
||||
"baseline", (unsigned long)d, 0UL,
|
||||
(unsigned long)rb.st.hits_dram, 0UL,
|
||||
(unsigned long)rb.st.nvme_reads, rb.st.avgGetUs(),
|
||||
pct(rb.st.nvme_reads, gb));
|
||||
std::printf("%-9s %5lu %5lu | %8lu %8lu %8lu | %10.3f %9.1f%%\n",
|
||||
"proposed", (unsigned long)d, (unsigned long)cxl_cap,
|
||||
(unsigned long)rp.st.hits_dram,
|
||||
(unsigned long)rp.st.hits_cxl,
|
||||
(unsigned long)rp.st.nvme_reads, rp.st.avgGetUs(),
|
||||
pct(rp.st.nvme_reads, gp));
|
||||
}
|
||||
|
||||
// Headline at the equal-DRAM point with the largest NVMe-read reduction
|
||||
// (the robust win; average GET latency is dominated by NVMe's heavy tail
|
||||
// and is reported only as a secondary number).
|
||||
size_t best = 0;
|
||||
double best_drop = -1;
|
||||
for (size_t i = 0; i < base.size(); ++i) {
|
||||
double drop =
|
||||
base[i].st.nvme_reads
|
||||
? 100.0 * (base[i].st.nvme_reads - prop[i].st.nvme_reads) /
|
||||
base[i].st.nvme_reads
|
||||
: 0.0;
|
||||
if (drop > best_drop) {
|
||||
best_drop = drop;
|
||||
best = i;
|
||||
}
|
||||
}
|
||||
const Result &b = base[best];
|
||||
const Result &p = prop[best];
|
||||
double nvme_drop =
|
||||
b.st.nvme_reads
|
||||
? 100.0 * (b.st.nvme_reads - p.st.nvme_reads) / b.st.nvme_reads
|
||||
: 0.0;
|
||||
double lat_drop =
|
||||
b.st.avgGetUs() > 0
|
||||
? 100.0 * (b.st.avgGetUs() - p.st.avgGetUs()) / b.st.avgGetUs()
|
||||
: 0.0;
|
||||
std::printf("\n--- Headline (equal DRAM = %lu blocks) ---\n",
|
||||
(unsigned long)dram_sweep[best]);
|
||||
std::printf(
|
||||
" per-tier read latency: DRAM=%.0f ns CXL=%.0f ns NVMe=%.0f ns\n",
|
||||
p.dram_ns, p.cxl_ns, p.nvme_ns);
|
||||
std::printf(" CXL tier cut NVMe reads by %.1f%%\n", nvme_drop);
|
||||
if (lat_drop > 0)
|
||||
std::printf(" average GET latency also fell %.1f%%\n", lat_drop);
|
||||
|
||||
double target = base.back().st.avgGetUs();
|
||||
long need = -1;
|
||||
for (size_t i = 0; i < prop.size(); ++i)
|
||||
if (prop[i].st.avgGetUs() <= target) {
|
||||
need = (long)dram_sweep[i];
|
||||
break;
|
||||
}
|
||||
if (need >= 0 && need < (long)dram_sweep.back()) {
|
||||
double saving =
|
||||
100.0 * (1.0 - (double)need / (double)dram_sweep.back());
|
||||
std::printf(
|
||||
" iso-latency: proposed matches baseline@%lu-block latency with "
|
||||
"only %ld DRAM blocks -> %.0f%% less DRAM\n",
|
||||
(unsigned long)dram_sweep.back(), need, saving);
|
||||
}
|
||||
if (have_csv) std::printf("\nCSV -> %s\n", FLAGS_csv.c_str());
|
||||
return 0;
|
||||
}
|
||||
|
|
@ -1,173 +0,0 @@
|
|||
// Copyright 2024 KVCache.AI
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
#ifndef MOONCAKE_STORE_MEMORY_TIER_H_
|
||||
#define MOONCAKE_STORE_MEMORY_TIER_H_
|
||||
|
||||
#include <cstdint>
|
||||
#include <list>
|
||||
#include <memory>
|
||||
#include <string>
|
||||
#include <unordered_map>
|
||||
#include <vector>
|
||||
|
||||
namespace mooncake {
|
||||
namespace store {
|
||||
|
||||
// Memory tiering that makes CXL / NUMA-remote DRAM a first-class "L2.5" storage
|
||||
// tier between local DRAM and NVMe:
|
||||
//
|
||||
// L1 : GPU HBM (optional, not modelled in this CPU-side tier
|
||||
// set) L2 : local DRAM (fast, scarce) L2.5 : CXL / NUMA-remote
|
||||
// (this module) L3 : NVMe (SSD) (slow, abundant) L4 : remote
|
||||
// store (future)
|
||||
//
|
||||
// Each tier is a fixed-block pool over real backing storage so the latency a
|
||||
// benchmark observes is measured, not modelled:
|
||||
// DramTier -> local malloc arena
|
||||
// NumaEmulatedTier -> numa_alloc_onnode(far node), the CXL analog
|
||||
// CxlTier -> /dev/dax mmap when present, else NumaEmulatedTier
|
||||
// NvmeTier -> a file opened O_DIRECT so reads touch the device
|
||||
//
|
||||
// TieredStore keeps a single copy of each block, evicts the LRU victim down a
|
||||
// tier under pressure, and promotes a block to the fastest tier on a hit -- the
|
||||
// same shape as the store's local_hot_cache + storage_backend eviction, with
|
||||
// CXL inserted as the L2.5 tier.
|
||||
|
||||
enum class TierKind { HBM, DRAM, CXL, NVME, REMOTE };
|
||||
const char *toString(TierKind kind);
|
||||
|
||||
struct TierStats {
|
||||
uint64_t puts = 0, gets = 0, erases = 0;
|
||||
uint64_t bytes_written = 0, bytes_read = 0;
|
||||
double read_latency_ns_sum = 0;
|
||||
uint64_t read_count = 0;
|
||||
double avgReadNs() const {
|
||||
return read_count ? read_latency_ns_sum / read_count : 0.0;
|
||||
}
|
||||
};
|
||||
|
||||
// Fixed-block pool with measured read / write latency. The base handles slot
|
||||
// allocation, the id->slot map and stats; backends implement the storage.
|
||||
class MemoryTier {
|
||||
public:
|
||||
MemoryTier(TierKind kind, std::string name, uint64_t block_size,
|
||||
uint64_t num_blocks, bool emulated);
|
||||
virtual ~MemoryTier();
|
||||
|
||||
TierKind kind() const { return kind_; }
|
||||
const std::string &name() const { return name_; }
|
||||
bool emulated() const { return emulated_; }
|
||||
uint64_t blockSize() const { return block_size_; }
|
||||
uint64_t numBlocks() const { return num_blocks_; }
|
||||
uint64_t capacityBytes() const { return block_size_ * num_blocks_; }
|
||||
uint64_t usedBlocks() const { return slot_of_.size(); }
|
||||
uint64_t usedBytes() const { return usedBlocks() * block_size_; }
|
||||
bool full() const { return free_slots_.empty(); }
|
||||
bool contains(uint64_t id) const { return slot_of_.count(id) != 0; }
|
||||
const TierStats &stats() const { return stats_; }
|
||||
|
||||
bool put(uint64_t id, const void *data); // false if full
|
||||
bool get(uint64_t id, void *out); // false if absent; times the read
|
||||
bool peekRead(uint64_t id, void *out); // untimed read (for migration)
|
||||
bool erase(uint64_t id);
|
||||
|
||||
bool ready() const { return ready_; }
|
||||
|
||||
protected:
|
||||
virtual bool backingInit() = 0;
|
||||
virtual void backingFree() = 0;
|
||||
virtual void writeBlock(uint64_t slot, const void *data) = 0;
|
||||
virtual void readBlock(uint64_t slot, void *out) = 0;
|
||||
|
||||
void initSlots(); // call from a derived ctor after backingInit() succeeds
|
||||
|
||||
TierKind kind_;
|
||||
std::string name_;
|
||||
uint64_t block_size_;
|
||||
uint64_t num_blocks_;
|
||||
bool emulated_;
|
||||
bool ready_ = false;
|
||||
|
||||
std::vector<uint64_t> free_slots_;
|
||||
std::unordered_map<uint64_t, uint64_t> slot_of_; // id -> slot
|
||||
TierStats stats_;
|
||||
};
|
||||
|
||||
std::shared_ptr<MemoryTier> makeDramTier(uint64_t block_size,
|
||||
uint64_t num_blocks);
|
||||
// CXL: real /dev/dax window if present, else honest NUMA-remote emulation.
|
||||
std::shared_ptr<MemoryTier> makeCxlTier(uint64_t block_size,
|
||||
uint64_t num_blocks);
|
||||
std::shared_ptr<MemoryTier> makeNumaEmulatedTier(uint64_t block_size,
|
||||
uint64_t num_blocks,
|
||||
int far_node = -1);
|
||||
std::shared_ptr<MemoryTier> makeNvmeTier(uint64_t block_size,
|
||||
uint64_t num_blocks,
|
||||
const std::string &path);
|
||||
|
||||
struct StoreStats {
|
||||
uint64_t puts = 0, gets = 0, get_miss = 0;
|
||||
uint64_t hits_dram = 0, hits_cxl = 0, hits_nvme = 0, hits_other = 0;
|
||||
uint64_t promotions = 0, demotions = 0;
|
||||
double total_get_latency_ns = 0;
|
||||
uint64_t peak_dram_bytes = 0;
|
||||
uint64_t nvme_reads = 0;
|
||||
|
||||
double avgGetUs() const {
|
||||
return gets ? total_get_latency_ns / gets / 1000.0 : 0.0;
|
||||
}
|
||||
uint64_t hitsFor(TierKind kind) const;
|
||||
};
|
||||
|
||||
// A multi-tier store over the tiers above. Tiers are ordered fastest to
|
||||
// slowest; the slowest must be able to hold the whole working set.
|
||||
class TieredStore {
|
||||
public:
|
||||
explicit TieredStore(std::vector<std::shared_ptr<MemoryTier>> tiers,
|
||||
bool promote_on_hit = true);
|
||||
|
||||
void put(uint64_t id, const void *data);
|
||||
bool get(uint64_t id, void *out);
|
||||
|
||||
const StoreStats &stats() const { return stats_; }
|
||||
void resetStats() {
|
||||
stats_ = StoreStats{};
|
||||
} // keep data/LRU, zero counters
|
||||
const std::vector<std::shared_ptr<MemoryTier>> &tiers() const {
|
||||
return tiers_;
|
||||
}
|
||||
|
||||
private:
|
||||
int tierIndexHolding(uint64_t id) const; // -1 if absent
|
||||
bool insertInto(int idx, uint64_t id, const void *data); // evicts downward
|
||||
void lruPushFront(int idx, uint64_t id);
|
||||
void lruRemove(int idx, uint64_t id);
|
||||
void lruTouch(int idx, uint64_t id);
|
||||
void recordPeakDram();
|
||||
|
||||
std::vector<std::shared_ptr<MemoryTier>> tiers_;
|
||||
bool promote_on_hit_;
|
||||
int dram_index_ = -1;
|
||||
std::vector<std::list<uint64_t>> lru_; // per tier, front = most recent
|
||||
std::vector<std::unordered_map<uint64_t, std::list<uint64_t>::iterator>>
|
||||
lru_pos_;
|
||||
StoreStats stats_;
|
||||
std::vector<std::vector<uint8_t>> evict_scratch_; // one buffer per tier
|
||||
};
|
||||
|
||||
} // namespace store
|
||||
} // namespace mooncake
|
||||
|
||||
#endif // MOONCAKE_STORE_MEMORY_TIER_H_
|
||||
|
|
@ -54,25 +54,10 @@ set(MOONCAKE_STORE_SOURCES
|
|||
hot_standby_service.cpp
|
||||
standby_state_machine.cpp
|
||||
ha_metric_manager.cpp
|
||||
tier/memory_tier.cpp
|
||||
tier/cxl_tier.cpp
|
||||
tier/numa_emulated_tier.cpp
|
||||
tier/nvme_tier.cpp
|
||||
store_c.cpp)
|
||||
|
||||
set(EXTRA_LIBS "")
|
||||
|
||||
# The CXL/NUMA memory tier uses libnuma to place its emulated tier on a remote
|
||||
# NUMA node; when libnuma is absent the tier falls back to local DRAM.
|
||||
find_library(MOONCAKE_TIER_NUMA_LIB numa)
|
||||
if(MOONCAKE_TIER_NUMA_LIB)
|
||||
add_compile_definitions(MOONCAKE_TIER_HAVE_NUMA)
|
||||
list(APPEND EXTRA_LIBS ${MOONCAKE_TIER_NUMA_LIB})
|
||||
message(STATUS "Mooncake Store memory tier: libnuma found, NUMA-remote tier enabled")
|
||||
else()
|
||||
message(STATUS "Mooncake Store memory tier: libnuma not found, tier falls back to local DRAM")
|
||||
endif()
|
||||
|
||||
# Find AWS SDK
|
||||
find_package(AWSSDK QUIET COMPONENTS s3)
|
||||
if(AWSSDK_FOUND)
|
||||
|
|
|
|||
|
|
@ -1,101 +0,0 @@
|
|||
// Copyright 2024 KVCache.AI
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
// CxlTier selector: use a real CXL device (/dev/daxN.M mmap) when present,
|
||||
// otherwise transparently fall back to the NUMA-remote emulation. The data
|
||||
// plane is identical -- load/store into a mapped window -- so only the source
|
||||
// of the mapping differs.
|
||||
|
||||
#include <fcntl.h>
|
||||
#include <sys/mman.h>
|
||||
#include <sys/stat.h>
|
||||
#include <unistd.h>
|
||||
|
||||
#include <cstdlib>
|
||||
#include <cstring>
|
||||
#include <string>
|
||||
|
||||
#include "tier/memory_tier.h"
|
||||
|
||||
namespace mooncake {
|
||||
namespace store {
|
||||
namespace {
|
||||
|
||||
// Real CXL path: mmap a /dev/dax device. Compiled always, used only when the
|
||||
// device node exists, so the production path is exercised the moment hardware
|
||||
// appears with no code change.
|
||||
class DaxTier final : public MemoryTier {
|
||||
public:
|
||||
DaxTier(uint64_t bs, uint64_t nb, std::string dev)
|
||||
: MemoryTier(TierKind::CXL, "CXL(/dev/dax)", bs, nb,
|
||||
/*emulated=*/false),
|
||||
dev_(std::move(dev)) {
|
||||
if (backingInit()) initSlots();
|
||||
}
|
||||
~DaxTier() override { backingFree(); }
|
||||
|
||||
protected:
|
||||
bool backingInit() override {
|
||||
fd_ = ::open(dev_.c_str(), O_RDWR);
|
||||
if (fd_ < 0) return false;
|
||||
arena_ = static_cast<uint8_t *>(mmap(nullptr, capacityBytes(),
|
||||
PROT_READ | PROT_WRITE, MAP_SHARED,
|
||||
fd_, 0));
|
||||
if (arena_ == MAP_FAILED) {
|
||||
arena_ = nullptr;
|
||||
::close(fd_);
|
||||
fd_ = -1;
|
||||
return false;
|
||||
}
|
||||
name_ = "CXL(" + dev_ + ")";
|
||||
return true;
|
||||
}
|
||||
void backingFree() override {
|
||||
if (arena_) munmap(arena_, capacityBytes());
|
||||
if (fd_ >= 0) ::close(fd_);
|
||||
arena_ = nullptr;
|
||||
fd_ = -1;
|
||||
}
|
||||
void writeBlock(uint64_t slot, const void *data) override {
|
||||
std::memcpy(arena_ + slot * blockSize(), data, blockSize());
|
||||
}
|
||||
void readBlock(uint64_t slot, void *out) override {
|
||||
std::memcpy(out, arena_ + slot * blockSize(), blockSize());
|
||||
}
|
||||
|
||||
private:
|
||||
std::string dev_;
|
||||
int fd_ = -1;
|
||||
uint8_t *arena_ = nullptr;
|
||||
};
|
||||
|
||||
bool deviceExists(const char *path) {
|
||||
struct stat st;
|
||||
return ::stat(path, &st) == 0;
|
||||
}
|
||||
|
||||
} // namespace
|
||||
|
||||
std::shared_ptr<MemoryTier> makeCxlTier(uint64_t bs, uint64_t nb) {
|
||||
for (const char *dev : {"/dev/dax0.0", "/dev/dax1.0"}) {
|
||||
if (deviceExists(dev)) {
|
||||
auto t = std::make_shared<DaxTier>(bs, nb, dev);
|
||||
if (t->ready()) return t;
|
||||
}
|
||||
}
|
||||
return makeNumaEmulatedTier(bs, nb, -1);
|
||||
}
|
||||
|
||||
} // namespace store
|
||||
} // namespace mooncake
|
||||
|
|
@ -1,272 +0,0 @@
|
|||
// Copyright 2024 KVCache.AI
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
#include "tier/memory_tier.h"
|
||||
|
||||
#include <chrono>
|
||||
#include <cstdlib>
|
||||
#include <cstring>
|
||||
|
||||
namespace mooncake {
|
||||
namespace store {
|
||||
|
||||
const char *toString(TierKind kind) {
|
||||
switch (kind) {
|
||||
case TierKind::HBM:
|
||||
return "HBM";
|
||||
case TierKind::DRAM:
|
||||
return "DRAM";
|
||||
case TierKind::CXL:
|
||||
return "CXL";
|
||||
case TierKind::NVME:
|
||||
return "NVMe";
|
||||
case TierKind::REMOTE:
|
||||
return "remote";
|
||||
}
|
||||
return "?";
|
||||
}
|
||||
|
||||
uint64_t StoreStats::hitsFor(TierKind kind) const {
|
||||
switch (kind) {
|
||||
case TierKind::DRAM:
|
||||
return hits_dram;
|
||||
case TierKind::CXL:
|
||||
return hits_cxl;
|
||||
case TierKind::NVME:
|
||||
return hits_nvme;
|
||||
default:
|
||||
return hits_other;
|
||||
}
|
||||
}
|
||||
|
||||
MemoryTier::MemoryTier(TierKind kind, std::string name, uint64_t block_size,
|
||||
uint64_t num_blocks, bool emulated)
|
||||
: kind_(kind),
|
||||
name_(std::move(name)),
|
||||
block_size_(block_size),
|
||||
num_blocks_(num_blocks),
|
||||
emulated_(emulated) {}
|
||||
|
||||
MemoryTier::~MemoryTier() = default;
|
||||
|
||||
void MemoryTier::initSlots() {
|
||||
free_slots_.reserve(num_blocks_);
|
||||
for (uint64_t i = 0; i < num_blocks_; ++i)
|
||||
free_slots_.push_back(num_blocks_ - 1 - i); // pop_back gives 0,1,2,...
|
||||
ready_ = true;
|
||||
}
|
||||
|
||||
bool MemoryTier::put(uint64_t id, const void *data) {
|
||||
auto it = slot_of_.find(id);
|
||||
if (it != slot_of_.end()) { // overwrite in place
|
||||
writeBlock(it->second, data);
|
||||
stats_.puts++;
|
||||
stats_.bytes_written += block_size_;
|
||||
return true;
|
||||
}
|
||||
if (free_slots_.empty()) return false;
|
||||
uint64_t slot = free_slots_.back();
|
||||
free_slots_.pop_back();
|
||||
slot_of_[id] = slot;
|
||||
writeBlock(slot, data);
|
||||
stats_.puts++;
|
||||
stats_.bytes_written += block_size_;
|
||||
return true;
|
||||
}
|
||||
|
||||
bool MemoryTier::get(uint64_t id, void *out) {
|
||||
auto it = slot_of_.find(id);
|
||||
if (it == slot_of_.end()) return false;
|
||||
auto t0 = std::chrono::steady_clock::now();
|
||||
readBlock(it->second, out);
|
||||
auto t1 = std::chrono::steady_clock::now();
|
||||
stats_.read_latency_ns_sum +=
|
||||
std::chrono::duration<double, std::nano>(t1 - t0).count();
|
||||
stats_.read_count++;
|
||||
stats_.gets++;
|
||||
stats_.bytes_read += block_size_;
|
||||
return true;
|
||||
}
|
||||
|
||||
bool MemoryTier::peekRead(uint64_t id, void *out) {
|
||||
auto it = slot_of_.find(id);
|
||||
if (it == slot_of_.end()) return false;
|
||||
readBlock(it->second, out);
|
||||
return true;
|
||||
}
|
||||
|
||||
bool MemoryTier::erase(uint64_t id) {
|
||||
auto it = slot_of_.find(id);
|
||||
if (it == slot_of_.end()) return false;
|
||||
free_slots_.push_back(it->second);
|
||||
slot_of_.erase(it);
|
||||
stats_.erases++;
|
||||
return true;
|
||||
}
|
||||
|
||||
namespace {
|
||||
class DramTier final : public MemoryTier {
|
||||
public:
|
||||
DramTier(uint64_t bs, uint64_t nb)
|
||||
: MemoryTier(TierKind::DRAM, "DRAM(local)", bs, nb,
|
||||
/*emulated=*/false) {
|
||||
if (backingInit()) initSlots();
|
||||
}
|
||||
~DramTier() override { backingFree(); }
|
||||
|
||||
protected:
|
||||
bool backingInit() override {
|
||||
arena_ = static_cast<uint8_t *>(std::malloc(capacityBytes()));
|
||||
if (arena_) std::memset(arena_, 0, capacityBytes());
|
||||
return arena_ != nullptr;
|
||||
}
|
||||
void backingFree() override {
|
||||
std::free(arena_);
|
||||
arena_ = nullptr;
|
||||
}
|
||||
void writeBlock(uint64_t slot, const void *data) override {
|
||||
std::memcpy(arena_ + slot * blockSize(), data, blockSize());
|
||||
}
|
||||
void readBlock(uint64_t slot, void *out) override {
|
||||
std::memcpy(out, arena_ + slot * blockSize(), blockSize());
|
||||
}
|
||||
|
||||
private:
|
||||
uint8_t *arena_ = nullptr;
|
||||
};
|
||||
} // namespace
|
||||
|
||||
std::shared_ptr<MemoryTier> makeDramTier(uint64_t bs, uint64_t nb) {
|
||||
return std::make_shared<DramTier>(bs, nb);
|
||||
}
|
||||
|
||||
TieredStore::TieredStore(std::vector<std::shared_ptr<MemoryTier>> tiers,
|
||||
bool promote_on_hit)
|
||||
: tiers_(std::move(tiers)), promote_on_hit_(promote_on_hit) {
|
||||
lru_.resize(tiers_.size());
|
||||
lru_pos_.resize(tiers_.size());
|
||||
evict_scratch_.resize(tiers_.size());
|
||||
uint64_t bs = tiers_.empty() ? 0 : tiers_[0]->blockSize();
|
||||
for (size_t i = 0; i < tiers_.size(); ++i) {
|
||||
evict_scratch_[i].resize(bs);
|
||||
if (tiers_[i]->kind() == TierKind::DRAM && dram_index_ < 0)
|
||||
dram_index_ = static_cast<int>(i);
|
||||
}
|
||||
}
|
||||
|
||||
int TieredStore::tierIndexHolding(uint64_t id) const {
|
||||
for (size_t i = 0; i < tiers_.size(); ++i)
|
||||
if (tiers_[i]->contains(id)) return static_cast<int>(i);
|
||||
return -1;
|
||||
}
|
||||
|
||||
void TieredStore::lruPushFront(int idx, uint64_t id) {
|
||||
lru_[idx].push_front(id);
|
||||
lru_pos_[idx][id] = lru_[idx].begin();
|
||||
}
|
||||
void TieredStore::lruRemove(int idx, uint64_t id) {
|
||||
auto it = lru_pos_[idx].find(id);
|
||||
if (it == lru_pos_[idx].end()) return;
|
||||
lru_[idx].erase(it->second);
|
||||
lru_pos_[idx].erase(it);
|
||||
}
|
||||
void TieredStore::lruTouch(int idx, uint64_t id) {
|
||||
lruRemove(idx, id);
|
||||
lruPushFront(idx, id);
|
||||
}
|
||||
|
||||
bool TieredStore::insertInto(int idx, uint64_t id, const void *data) {
|
||||
if (idx < 0 || idx >= static_cast<int>(tiers_.size())) return false;
|
||||
auto &tier = tiers_[idx];
|
||||
if (tier->full()) {
|
||||
// Evict the LRU victim downward to make room.
|
||||
if (!lru_[idx].empty()) {
|
||||
uint64_t victim = lru_[idx].back();
|
||||
tier->peekRead(victim, evict_scratch_[idx].data());
|
||||
tier->erase(victim);
|
||||
lruRemove(idx, victim);
|
||||
if (insertInto(idx + 1, victim, evict_scratch_[idx].data()))
|
||||
stats_.demotions++;
|
||||
}
|
||||
}
|
||||
if (!tier->put(id, data)) return false;
|
||||
lruPushFront(idx, id);
|
||||
return true;
|
||||
}
|
||||
|
||||
void TieredStore::recordPeakDram() {
|
||||
if (dram_index_ >= 0) {
|
||||
uint64_t u = tiers_[dram_index_]->usedBytes();
|
||||
if (u > stats_.peak_dram_bytes) stats_.peak_dram_bytes = u;
|
||||
}
|
||||
}
|
||||
|
||||
void TieredStore::put(uint64_t id, const void *data) {
|
||||
stats_.puts++;
|
||||
int cur = tierIndexHolding(id);
|
||||
if (cur >= 0) { // already present: refresh in place
|
||||
tiers_[cur]->put(id, data);
|
||||
lruTouch(cur, id);
|
||||
recordPeakDram();
|
||||
return;
|
||||
}
|
||||
insertInto(0, id, data);
|
||||
recordPeakDram();
|
||||
}
|
||||
|
||||
bool TieredStore::get(uint64_t id, void *out) {
|
||||
stats_.gets++;
|
||||
int idx = tierIndexHolding(id);
|
||||
if (idx < 0) {
|
||||
stats_.get_miss++;
|
||||
return false;
|
||||
}
|
||||
auto &tier = tiers_[idx];
|
||||
|
||||
auto t0 = std::chrono::steady_clock::now();
|
||||
tier->get(id, out);
|
||||
auto t1 = std::chrono::steady_clock::now();
|
||||
stats_.total_get_latency_ns +=
|
||||
std::chrono::duration<double, std::nano>(t1 - t0).count();
|
||||
|
||||
switch (tier->kind()) {
|
||||
case TierKind::DRAM:
|
||||
stats_.hits_dram++;
|
||||
break;
|
||||
case TierKind::CXL:
|
||||
stats_.hits_cxl++;
|
||||
break;
|
||||
case TierKind::NVME:
|
||||
stats_.hits_nvme++;
|
||||
stats_.nvme_reads++;
|
||||
break;
|
||||
default:
|
||||
stats_.hits_other++;
|
||||
break;
|
||||
}
|
||||
|
||||
if (promote_on_hit_ && idx > 0) {
|
||||
tier->erase(id);
|
||||
lruRemove(idx, id);
|
||||
insertInto(0, id, out); // `out` already holds the block data
|
||||
stats_.promotions++;
|
||||
} else {
|
||||
lruTouch(idx, id);
|
||||
}
|
||||
recordPeakDram();
|
||||
return true;
|
||||
}
|
||||
|
||||
} // namespace store
|
||||
} // namespace mooncake
|
||||
|
|
@ -1,113 +0,0 @@
|
|||
// Copyright 2024 KVCache.AI
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
// The CXL "L2.5" tier emulated honestly with NUMA-remote DRAM: kind() is CXL
|
||||
// and emulated() is true, and the name records which NUMA node backs it.
|
||||
// numa_alloc_onnode(far_node) yields memory whose latency sits between local
|
||||
// DRAM and NVMe -- the same architectural slot a CXL.mem pool occupies. The
|
||||
// data plane (load/store into the mapped arena) is identical to the real device
|
||||
// path, which is why the emulation is faithful.
|
||||
|
||||
#include <cstdlib>
|
||||
#include <cstring>
|
||||
#include <string>
|
||||
|
||||
#include "tier/memory_tier.h"
|
||||
|
||||
#ifdef MOONCAKE_TIER_HAVE_NUMA
|
||||
#include <numa.h>
|
||||
#endif
|
||||
|
||||
namespace mooncake {
|
||||
namespace store {
|
||||
namespace {
|
||||
|
||||
#ifdef MOONCAKE_TIER_HAVE_NUMA
|
||||
// Pick a NUMA node other than the one the calling thread runs on, so the arena
|
||||
// is genuinely remote. Falls back to the highest-numbered node.
|
||||
int pickFarNode() {
|
||||
if (numa_available() < 0) return -1;
|
||||
int max_node = numa_max_node();
|
||||
if (max_node < 1) return -1;
|
||||
int local = numa_node_of_cpu(sched_getcpu());
|
||||
for (int n = max_node; n >= 0; --n)
|
||||
if (n != local) return n;
|
||||
return max_node;
|
||||
}
|
||||
#endif
|
||||
|
||||
class NumaEmulatedTier final : public MemoryTier {
|
||||
public:
|
||||
NumaEmulatedTier(uint64_t bs, uint64_t nb, int far_node)
|
||||
: MemoryTier(TierKind::CXL, "CXL(emulated)", bs, nb, /*emulated=*/true),
|
||||
far_node_(far_node) {
|
||||
if (backingInit()) initSlots();
|
||||
}
|
||||
~NumaEmulatedTier() override { backingFree(); }
|
||||
|
||||
protected:
|
||||
bool backingInit() override {
|
||||
#ifdef MOONCAKE_TIER_HAVE_NUMA
|
||||
if (far_node_ < 0) far_node_ = pickFarNode();
|
||||
if (numa_available() >= 0 && far_node_ >= 0) {
|
||||
arena_ = static_cast<uint8_t *>(numa_alloc_onnode(
|
||||
static_cast<size_t>(capacityBytes()), far_node_));
|
||||
if (arena_) {
|
||||
numa_backed_ = true;
|
||||
name_ = "CXL(emulated via NUMA node " +
|
||||
std::to_string(far_node_) + ")";
|
||||
}
|
||||
}
|
||||
#endif
|
||||
if (!arena_) { // portable fallback
|
||||
arena_ = static_cast<uint8_t *>(std::malloc(capacityBytes()));
|
||||
name_ = "CXL(emulated, malloc fallback)";
|
||||
}
|
||||
if (arena_) std::memset(arena_, 0, capacityBytes());
|
||||
return arena_ != nullptr;
|
||||
}
|
||||
void backingFree() override {
|
||||
if (!arena_) return;
|
||||
#ifdef MOONCAKE_TIER_HAVE_NUMA
|
||||
if (numa_backed_) {
|
||||
numa_free(arena_, capacityBytes());
|
||||
arena_ = nullptr;
|
||||
return;
|
||||
}
|
||||
#endif
|
||||
std::free(arena_);
|
||||
arena_ = nullptr;
|
||||
}
|
||||
void writeBlock(uint64_t slot, const void *data) override {
|
||||
std::memcpy(arena_ + slot * blockSize(), data, blockSize());
|
||||
}
|
||||
void readBlock(uint64_t slot, void *out) override {
|
||||
std::memcpy(out, arena_ + slot * blockSize(), blockSize());
|
||||
}
|
||||
|
||||
private:
|
||||
int far_node_;
|
||||
uint8_t *arena_ = nullptr;
|
||||
bool numa_backed_ = false;
|
||||
};
|
||||
|
||||
} // namespace
|
||||
|
||||
std::shared_ptr<MemoryTier> makeNumaEmulatedTier(uint64_t bs, uint64_t nb,
|
||||
int far_node) {
|
||||
return std::make_shared<NumaEmulatedTier>(bs, nb, far_node);
|
||||
}
|
||||
|
||||
} // namespace store
|
||||
} // namespace mooncake
|
||||
|
|
@ -1,117 +0,0 @@
|
|||
// Copyright 2024 KVCache.AI
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
// NvmeTier: a fixed-block pool backed by a real file, opened O_DIRECT so reads
|
||||
// actually touch the device rather than the page cache (otherwise the NVMe vs
|
||||
// CXL/DRAM latency comparison would be dishonest -- the cache would make NVMe
|
||||
// look as fast as DRAM). Falls back to buffered IO + POSIX_FADV_DONTNEED when
|
||||
// O_DIRECT is unavailable, and records which in its name.
|
||||
|
||||
#ifndef _GNU_SOURCE
|
||||
#define _GNU_SOURCE
|
||||
#endif
|
||||
#include <fcntl.h>
|
||||
#include <unistd.h>
|
||||
|
||||
#include <cstdlib>
|
||||
#include <cstring>
|
||||
#include <string>
|
||||
|
||||
#include "tier/memory_tier.h"
|
||||
|
||||
namespace mooncake {
|
||||
namespace store {
|
||||
namespace {
|
||||
|
||||
class NvmeTier final : public MemoryTier {
|
||||
public:
|
||||
NvmeTier(uint64_t bs, uint64_t nb, std::string path)
|
||||
: MemoryTier(TierKind::NVME, "NVMe", bs, nb, /*emulated=*/false),
|
||||
path_(std::move(path)) {
|
||||
if (backingInit()) initSlots();
|
||||
}
|
||||
~NvmeTier() override { backingFree(); }
|
||||
|
||||
protected:
|
||||
bool backingInit() override {
|
||||
int base = O_RDWR | O_CREAT | O_TRUNC;
|
||||
fd_ = ::open(path_.c_str(), base | O_DIRECT, 0644);
|
||||
if (fd_ >= 0) {
|
||||
direct_ = true;
|
||||
} else {
|
||||
fd_ = ::open(path_.c_str(), base, 0644);
|
||||
}
|
||||
if (fd_ < 0) return false;
|
||||
if (::ftruncate(fd_, static_cast<off_t>(capacityBytes())) != 0) {
|
||||
::close(fd_);
|
||||
fd_ = -1;
|
||||
return false;
|
||||
}
|
||||
// O_DIRECT needs a buffer aligned to the logical block size.
|
||||
if (posix_memalign(reinterpret_cast<void **>(&bounce_), 4096,
|
||||
blockSize()) != 0)
|
||||
bounce_ = nullptr;
|
||||
name_ = direct_ ? "NVMe(O_DIRECT)" : "NVMe(buffered)";
|
||||
return bounce_ != nullptr;
|
||||
}
|
||||
void backingFree() override {
|
||||
if (fd_ >= 0) {
|
||||
::close(fd_);
|
||||
::unlink(path_.c_str());
|
||||
fd_ = -1;
|
||||
}
|
||||
std::free(bounce_);
|
||||
bounce_ = nullptr;
|
||||
}
|
||||
void writeBlock(uint64_t slot, const void *data) override {
|
||||
std::memcpy(bounce_, data, blockSize());
|
||||
off_t off = static_cast<off_t>(slot * blockSize());
|
||||
size_t done = 0;
|
||||
while (done < blockSize()) {
|
||||
ssize_t k =
|
||||
::pwrite(fd_, bounce_ + done, blockSize() - done, off + done);
|
||||
if (k <= 0) break;
|
||||
done += static_cast<size_t>(k);
|
||||
}
|
||||
}
|
||||
void readBlock(uint64_t slot, void *out) override {
|
||||
off_t off = static_cast<off_t>(slot * blockSize());
|
||||
size_t done = 0;
|
||||
while (done < blockSize()) {
|
||||
ssize_t k =
|
||||
::pread(fd_, bounce_ + done, blockSize() - done, off + done);
|
||||
if (k <= 0) break;
|
||||
done += static_cast<size_t>(k);
|
||||
}
|
||||
if (!direct_) // discourage the page cache from hiding disk latency
|
||||
::posix_fadvise(fd_, off, blockSize(), POSIX_FADV_DONTNEED);
|
||||
std::memcpy(out, bounce_, blockSize());
|
||||
}
|
||||
|
||||
private:
|
||||
std::string path_;
|
||||
int fd_ = -1;
|
||||
bool direct_ = false;
|
||||
uint8_t *bounce_ = nullptr;
|
||||
};
|
||||
|
||||
} // namespace
|
||||
|
||||
std::shared_ptr<MemoryTier> makeNvmeTier(uint64_t bs, uint64_t nb,
|
||||
const std::string &path) {
|
||||
return std::make_shared<NvmeTier>(bs, nb, path);
|
||||
}
|
||||
|
||||
} // namespace store
|
||||
} // namespace mooncake
|
||||
|
|
@ -79,7 +79,6 @@ add_store_test(task_executor_test task_executor_test.cpp)
|
|||
add_store_test(task_integration_test task_integration_test.cpp)
|
||||
add_store_test(dummy_client_get_buffer_test dummy_client_get_buffer_test.cpp)
|
||||
add_store_test(health_check_test health_check_test.cpp)
|
||||
add_store_test(memory_tier_test memory_tier_test.cpp)
|
||||
add_subdirectory(e2e)
|
||||
|
||||
add_executable(high_availability_test ha/leadership/high_availability_test.cpp)
|
||||
|
|
|
|||
|
|
@ -1,111 +0,0 @@
|
|||
// Copyright 2024 KVCache.AI
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
#include "tier/memory_tier.h"
|
||||
|
||||
#include <gtest/gtest.h>
|
||||
|
||||
#include <cstdint>
|
||||
#include <cstring>
|
||||
#include <vector>
|
||||
|
||||
using namespace mooncake::store;
|
||||
|
||||
namespace {
|
||||
|
||||
constexpr uint64_t BS = 4096;
|
||||
|
||||
void fill(std::vector<uint8_t> &b, uint64_t id) {
|
||||
for (size_t i = 0; i < b.size(); ++i) b[i] = (uint8_t)(id * 7 + i);
|
||||
std::memcpy(b.data(), &id, sizeof(id));
|
||||
}
|
||||
bool verify(const std::vector<uint8_t> &b, uint64_t id) {
|
||||
std::vector<uint8_t> ref(b.size());
|
||||
fill(ref, id);
|
||||
return std::memcmp(b.data(), ref.data(), b.size()) == 0;
|
||||
}
|
||||
|
||||
std::string tmpPath(const char *name) {
|
||||
const char *dir = std::getenv("TMPDIR");
|
||||
return std::string(dir ? dir : "/tmp") + "/mooncake_" + name;
|
||||
}
|
||||
|
||||
} // namespace
|
||||
|
||||
TEST(MemoryTier, DramRoundtripAndCapacity) {
|
||||
auto dram = makeDramTier(BS, 4);
|
||||
ASSERT_TRUE(dram->ready());
|
||||
std::vector<uint8_t> b(BS), o(BS);
|
||||
for (uint64_t i = 0; i < 4; ++i) {
|
||||
fill(b, i);
|
||||
EXPECT_TRUE(dram->put(i, b.data()));
|
||||
}
|
||||
fill(b, 99);
|
||||
EXPECT_FALSE(dram->put(99, b.data())) << "put beyond capacity must fail";
|
||||
for (uint64_t i = 0; i < 4; ++i) {
|
||||
EXPECT_TRUE(dram->get(i, o.data()));
|
||||
EXPECT_TRUE(verify(o, i));
|
||||
}
|
||||
EXPECT_FALSE(dram->get(99, o.data()));
|
||||
}
|
||||
|
||||
TEST(MemoryTier, CxlEmulatedAndNvmeBackends) {
|
||||
auto cxl = makeCxlTier(BS, 4);
|
||||
ASSERT_TRUE(cxl->ready());
|
||||
EXPECT_EQ(cxl->kind(), TierKind::CXL);
|
||||
EXPECT_TRUE(cxl->emulated()); // no /dev/dax on most hosts
|
||||
|
||||
auto nvme = makeNvmeTier(BS, 4, tmpPath("tier_nvme.bin"));
|
||||
ASSERT_TRUE(nvme->ready());
|
||||
std::vector<uint8_t> b(BS), o(BS);
|
||||
fill(b, 7);
|
||||
EXPECT_TRUE(nvme->put(7, b.data()));
|
||||
EXPECT_TRUE(nvme->get(7, o.data()));
|
||||
EXPECT_TRUE(verify(o, 7));
|
||||
}
|
||||
|
||||
TEST(TieredStore, IntegrityPromotionSingleCopy) {
|
||||
const uint64_t W = 64;
|
||||
std::vector<std::shared_ptr<MemoryTier>> tiers = {
|
||||
makeDramTier(BS, 8), makeCxlTier(BS, 16),
|
||||
makeNvmeTier(BS, W, tmpPath("tier_store.bin"))};
|
||||
TieredStore store(std::move(tiers), /*promote_on_hit=*/true);
|
||||
|
||||
std::vector<uint8_t> b(BS), o(BS);
|
||||
for (uint64_t i = 0; i < W; ++i) {
|
||||
fill(b, i);
|
||||
store.put(i, b.data());
|
||||
}
|
||||
|
||||
bool all_ok = true;
|
||||
for (uint64_t i = 0; i < W; ++i)
|
||||
if (!store.get(i, o.data()) || !verify(o, i)) all_ok = false;
|
||||
EXPECT_TRUE(all_ok) << "all blocks must read back byte-exact across tiers";
|
||||
|
||||
// A freshly read cold block should be promoted into DRAM (tier 0).
|
||||
uint64_t cold = 3;
|
||||
store.get(cold, o.data());
|
||||
EXPECT_TRUE(store.tiers()[0]->contains(cold));
|
||||
|
||||
// Single-copy invariant: each live id lives in exactly one tier.
|
||||
bool single = true;
|
||||
for (uint64_t i = 0; i < W; ++i) {
|
||||
int count = 0;
|
||||
for (auto &t : store.tiers())
|
||||
if (t->contains(i)) ++count;
|
||||
if (count != 1) single = false;
|
||||
}
|
||||
EXPECT_TRUE(single);
|
||||
EXPECT_GT(store.stats().promotions, 0u);
|
||||
}
|
||||
|
|
@ -1,45 +0,0 @@
|
|||
# Memory fabric benchmarks
|
||||
|
||||
Drivers for the capability / routing layer. They are standalone scripts and
|
||||
binaries; none are required to build the Transfer Engine.
|
||||
|
||||
## One-shot demo
|
||||
|
||||
`hetero/run_demo.sh` runs the whole story end to end — conformance over every
|
||||
backend, the probe table, the routing / fault-fallback demo, the CXL/NUMA tier
|
||||
benchmark, and the GPU NVLink benchmark — then renders figures:
|
||||
|
||||
```bash
|
||||
./hetero/run_demo.sh [BUILD_DIR] # BUILD_DIR defaults to ./build
|
||||
# artifacts + PNGs land in ./fabric-results (override with MC_FABRIC_RESULTS)
|
||||
```
|
||||
|
||||
`hetero/plot.py` renders the figures from the collected CSV/JSON and skips any
|
||||
figure whose source is missing (so a partial run still plots what it has).
|
||||
|
||||
## `hetero/gpu_bench.py`
|
||||
|
||||
Measures device-to-device bandwidth (NVLink / PCIe peer access) versus a
|
||||
pinned-host staging round trip on two GPUs — the concrete payoff the
|
||||
`PathSelector` captures by preferring `nvlink_intra` over `host_staging_tcp`.
|
||||
Requires `torch` with CUDA and ≥2 GPUs; skips gracefully otherwise.
|
||||
|
||||
```bash
|
||||
python3 hetero/gpu_bench.py # GPUs 4/5 by default
|
||||
MC_CUDA_DEV=0 MC_CUDA_PEER=1 python3 hetero/gpu_bench.py
|
||||
MC_GPU_BENCH_CSV=/tmp/gpu_bench.csv python3 hetero/gpu_bench.py
|
||||
```
|
||||
|
||||
## Capability and conformance (built with the unit tests)
|
||||
|
||||
```bash
|
||||
./build/mooncake-transfer-engine/tests/fabric_cli probe # capability table
|
||||
./build/mooncake-transfer-engine/tests/fabric_cli select # routing + fault demo
|
||||
./build/mooncake-transfer-engine/tests/conformance_runner --json /tmp/conformance.json
|
||||
```
|
||||
|
||||
## Memory tier (Mooncake Store)
|
||||
|
||||
```bash
|
||||
./build/mooncake-store/benchmarks/tier_bench --csv /tmp/tier_bench.csv
|
||||
```
|
||||
|
|
@ -1,115 +0,0 @@
|
|||
#!/usr/bin/env python3
|
||||
# Copyright 2024 KVCache.AI
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
"""Measure the concrete payoff the capability-driven path selector captures by
|
||||
preferring NVLink peer access over the host-staging fallback.
|
||||
|
||||
On a host with >=2 NVLink-connected GPUs this reports, per transfer size, the
|
||||
device-to-device bandwidth (NVLink / PCIe P2P) versus a pinned-host staging
|
||||
round trip -- the exact NVLINK > ... > HOST_STAGING ladder the selector walks.
|
||||
Requires torch with CUDA; degrades gracefully (skips) when neither is present so
|
||||
it is safe to run anywhere. Targets idle GPUs (default 4/5, override with
|
||||
MC_CUDA_DEV / MC_CUDA_PEER) and uses modest buffers so it does not disturb other
|
||||
users of the machine."""
|
||||
|
||||
import csv
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
|
||||
|
||||
def main():
|
||||
out_csv = os.environ.get("MC_GPU_BENCH_CSV", "")
|
||||
try:
|
||||
import torch
|
||||
except Exception:
|
||||
print("torch not available -> skipping GPU benchmark.")
|
||||
return 0
|
||||
if not torch.cuda.is_available() or torch.cuda.device_count() < 2:
|
||||
print("need >=2 CUDA GPUs -> skipping GPU benchmark.")
|
||||
return 0
|
||||
|
||||
n = torch.cuda.device_count()
|
||||
a = min(int(os.environ.get("MC_CUDA_DEV", 4)), n - 1)
|
||||
b = min(int(os.environ.get("MC_CUDA_PEER", 5)), n - 1)
|
||||
if a == b:
|
||||
b = (a + 1) % n
|
||||
p2p = torch.cuda.can_device_access_peer(a, b)
|
||||
print("=== Heterogeneous GPU -- NVLink P2P vs Host-Staging ===")
|
||||
print(f"GPU{a} <-> GPU{b} peer-access(NVLink/PCIe P2P)={p2p}\n")
|
||||
|
||||
dev_a, dev_b = f"cuda:{a}", f"cuda:{b}"
|
||||
sizes_mb = [1, 4, 16, 64, 256]
|
||||
iters = 30
|
||||
|
||||
def timed(fn):
|
||||
# Wall-clock with a full sync of both devices around the loop. CUDA
|
||||
# events are unreliable here because a cross-device copy_ does not run on
|
||||
# the recording device's stream, which inflates event timings.
|
||||
for _ in range(5): # warm up and let clocks ramp
|
||||
fn()
|
||||
torch.cuda.synchronize(a)
|
||||
torch.cuda.synchronize(b)
|
||||
t0 = time.perf_counter()
|
||||
for _ in range(iters):
|
||||
fn()
|
||||
torch.cuda.synchronize(a)
|
||||
torch.cuda.synchronize(b)
|
||||
t1 = time.perf_counter()
|
||||
return (t1 - t0) / iters * 1e3 # ms per op
|
||||
|
||||
rows = []
|
||||
print(f'{"size":>7}{"NVLink GB/s":>14}{"staging GB/s":>15}{"speedup":>10}')
|
||||
print("-" * 46)
|
||||
for mb in sizes_mb:
|
||||
nelem = mb * 1024 * 1024 // 4
|
||||
xa = torch.randn(nelem, device=dev_a)
|
||||
xb = torch.empty(nelem, device=dev_b)
|
||||
host = torch.empty(nelem, pin_memory=True)
|
||||
nbytes = nelem * 4
|
||||
|
||||
def nvlink():
|
||||
xb.copy_(xa) # device-to-device (P2P over NVLink if available)
|
||||
|
||||
def staging():
|
||||
host.copy_(xa) # device -> pinned host
|
||||
xb.copy_(host) # pinned host -> device
|
||||
|
||||
ms_nv = timed(nvlink)
|
||||
ms_st = timed(staging)
|
||||
gbps_nv = nbytes / (ms_nv / 1e3) / 1e9
|
||||
gbps_st = nbytes / (ms_st / 1e3) / 1e9
|
||||
speedup = gbps_nv / gbps_st if gbps_st > 0 else 0
|
||||
rows.append((nbytes, gbps_nv, gbps_st, speedup))
|
||||
print(f'{str(mb)+"M":>7}{gbps_nv:>14.1f}{gbps_st:>15.1f}{speedup:>9.2f}x')
|
||||
|
||||
del xa, xb, host
|
||||
torch.cuda.empty_cache()
|
||||
|
||||
if out_csv:
|
||||
with open(out_csv, "w", newline="") as f:
|
||||
w = csv.writer(f)
|
||||
w.writerow(["size_bytes", "nvlink_gbps", "host_staging_gbps",
|
||||
"speedup"])
|
||||
for r in rows:
|
||||
w.writerow([r[0], f"{r[1]:.3f}", f"{r[2]:.3f}", f"{r[3]:.3f}"])
|
||||
print(f"\nCSV -> {out_csv}")
|
||||
print("\nThis is the payoff the path selector captures by preferring "
|
||||
"nvlink_intra over host_staging_tcp.")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
|
|
@ -1,193 +0,0 @@
|
|||
#!/usr/bin/env python3
|
||||
# Copyright 2024 KVCache.AI
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
"""Render figures from the CSV / JSON artifacts produced by the fabric tools.
|
||||
|
||||
Reads from a results directory (default: ./fabric-results, override with
|
||||
MC_FABRIC_RESULTS) and writes PNGs there. Uses the non-interactive Agg backend
|
||||
and skips any figure whose source artifact is missing, so it is safe to run
|
||||
after a partial demo."""
|
||||
|
||||
import csv
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
|
||||
RESULTS = os.environ.get("MC_FABRIC_RESULTS", os.path.abspath("fabric-results"))
|
||||
|
||||
|
||||
def _read_csv(path):
|
||||
return list(csv.DictReader(open(path))) if os.path.exists(path) else None
|
||||
|
||||
|
||||
def _have_matplotlib():
|
||||
try:
|
||||
import matplotlib # noqa: F401
|
||||
|
||||
return True
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
def plot_capabilities(plt):
|
||||
path = os.path.join(RESULTS, "conformance.json")
|
||||
if not os.path.exists(path):
|
||||
return
|
||||
data = json.load(open(path))
|
||||
names, bws, cols = [], [], []
|
||||
for entry in data:
|
||||
cap = entry.get("capability", {})
|
||||
if cap.get("max_bandwidth_mbps", 0) == 0:
|
||||
continue
|
||||
names.append(cap.get("name", entry.get("backend", "?")))
|
||||
bws.append(cap["max_bandwidth_mbps"])
|
||||
cols.append("#d62728" if cap.get("emulated") else "#2ca02c")
|
||||
if not names:
|
||||
return
|
||||
plt.figure(figsize=(8, 5))
|
||||
plt.bar(range(len(names)), bws, color=cols)
|
||||
plt.yscale("log")
|
||||
plt.xticks(range(len(names)), names, rotation=30, ha="right")
|
||||
plt.ylabel("probed bandwidth (MB/s, log)")
|
||||
plt.title("Probed per-backend bandwidth\n"
|
||||
"(green = real hardware, red = honestly emulated)")
|
||||
plt.grid(True, axis="y", which="both", alpha=0.3)
|
||||
plt.tight_layout()
|
||||
plt.savefig(os.path.join(RESULTS, "capabilities.png"), dpi=130)
|
||||
plt.close()
|
||||
print(" capabilities.png")
|
||||
|
||||
|
||||
def plot_conformance(plt):
|
||||
path = os.path.join(RESULTS, "conformance.json")
|
||||
if not os.path.exists(path):
|
||||
return
|
||||
data = json.load(open(path))
|
||||
cases = ["capability", "register", "roundtrip", "batch", "ordering",
|
||||
"concurrency"]
|
||||
backends = [e["backend"] for e in data]
|
||||
grid = []
|
||||
for e in data:
|
||||
by_name = {c["name"]: c["outcome"] for c in e.get("cases", [])}
|
||||
grid.append([by_name.get(c, "skip") for c in cases])
|
||||
if not backends:
|
||||
return
|
||||
code = {"PASS": 2, "SKIP": 1, "skip": 1, "FAIL": 0}
|
||||
import numpy as np
|
||||
|
||||
mat = np.array([[code.get(o, 1) for o in row] for row in grid])
|
||||
from matplotlib.colors import ListedColormap
|
||||
|
||||
cmap = ListedColormap(["#d62728", "#dddddd", "#2ca02c"])
|
||||
plt.figure(figsize=(8, 0.7 * len(backends) + 2))
|
||||
plt.imshow(mat, cmap=cmap, vmin=0, vmax=2, aspect="auto")
|
||||
plt.xticks(range(len(cases)), cases, rotation=30, ha="right")
|
||||
plt.yticks(range(len(backends)), backends)
|
||||
for i in range(len(backends)):
|
||||
for j in range(len(cases)):
|
||||
plt.text(j, i, ["FAIL", "skip", "PASS"][mat[i, j]],
|
||||
ha="center", va="center", fontsize=8)
|
||||
plt.title("Cross-backend conformance matrix\n"
|
||||
"one suite, every backend (green = PASS)")
|
||||
plt.tight_layout()
|
||||
plt.savefig(os.path.join(RESULTS, "conformance.png"), dpi=130)
|
||||
plt.close()
|
||||
print(" conformance.png")
|
||||
|
||||
|
||||
def plot_tier(plt):
|
||||
rows = _read_csv(os.path.join(RESULTS, "tier_bench.csv"))
|
||||
if not rows:
|
||||
return
|
||||
base = sorted([r for r in rows if r["config"] == "baseline"],
|
||||
key=lambda r: int(r["dram_blocks"]))
|
||||
prop = sorted([r for r in rows if r["config"] == "proposed"],
|
||||
key=lambda r: int(r["dram_blocks"]))
|
||||
d = [int(r["dram_blocks"]) for r in base]
|
||||
|
||||
plt.figure(figsize=(7, 5))
|
||||
plt.plot(d, [int(r["nvme_reads"]) for r in base], "-o",
|
||||
label="baseline (DRAM+NVMe)")
|
||||
plt.plot(d, [int(r["nvme_reads"]) for r in prop], "-s",
|
||||
label="proposed (DRAM+CXL+NVMe)")
|
||||
plt.xlabel("local DRAM tier size (blocks)")
|
||||
plt.ylabel("NVMe reads (measured window)")
|
||||
plt.title("CXL/NUMA tier absorbs warm reads -> fewer NVMe accesses")
|
||||
plt.grid(True, alpha=0.3)
|
||||
plt.legend()
|
||||
plt.tight_layout()
|
||||
plt.savefig(os.path.join(RESULTS, "tier_nvme.png"), dpi=130)
|
||||
plt.close()
|
||||
print(" tier_nvme.png")
|
||||
|
||||
plt.figure(figsize=(7, 5))
|
||||
plt.plot(d, [float(r["avg_get_us"]) for r in base], "-o",
|
||||
label="baseline (DRAM+NVMe)")
|
||||
plt.plot(d, [float(r["avg_get_us"]) for r in prop], "-s",
|
||||
label="proposed (DRAM+CXL+NVMe)")
|
||||
plt.xlabel("local DRAM tier size (blocks)")
|
||||
plt.ylabel("avg GET latency (us)")
|
||||
plt.title("CXL/NUMA tier lowers average GET latency at equal DRAM")
|
||||
plt.grid(True, alpha=0.3)
|
||||
plt.legend()
|
||||
plt.tight_layout()
|
||||
plt.savefig(os.path.join(RESULTS, "tier_latency.png"), dpi=130)
|
||||
plt.close()
|
||||
print(" tier_latency.png")
|
||||
|
||||
|
||||
def plot_gpu(plt):
|
||||
rows = _read_csv(os.path.join(RESULTS, "gpu_bench.csv"))
|
||||
if not rows:
|
||||
return
|
||||
xs = [int(r["size_bytes"]) / (1 << 20) for r in rows]
|
||||
nv = [float(r["nvlink_gbps"]) for r in rows]
|
||||
st = [float(r["host_staging_gbps"]) for r in rows]
|
||||
plt.figure(figsize=(7, 5))
|
||||
plt.plot(xs, nv, "-o", label="NVLink P2P (nvlink_intra)")
|
||||
plt.plot(xs, st, "-s", label="pinned host staging (fallback)")
|
||||
plt.xscale("log")
|
||||
plt.xlabel("transfer size (MiB)")
|
||||
plt.ylabel("bandwidth (GB/s)")
|
||||
plt.title("Real GPU-GPU: NVLink vs host-staging fallback")
|
||||
plt.grid(True, which="both", alpha=0.3)
|
||||
plt.legend()
|
||||
plt.tight_layout()
|
||||
plt.savefig(os.path.join(RESULTS, "gpu_bench.png"), dpi=130)
|
||||
plt.close()
|
||||
print(" gpu_bench.png")
|
||||
|
||||
|
||||
def main():
|
||||
if not _have_matplotlib():
|
||||
print("matplotlib not installed -> skipping plots "
|
||||
"(CSV/JSON artifacts are still produced).")
|
||||
return 0
|
||||
import matplotlib
|
||||
|
||||
matplotlib.use("Agg")
|
||||
import matplotlib.pyplot as plt
|
||||
|
||||
print(f"Rendering figures -> {RESULTS}/*.png")
|
||||
plot_capabilities(plt)
|
||||
plot_conformance(plt)
|
||||
plot_tier(plt)
|
||||
plot_gpu(plt)
|
||||
print("done.")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
|
|
@ -1,87 +0,0 @@
|
|||
#!/usr/bin/env bash
|
||||
# Copyright 2024 KVCache.AI
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
# One-shot driver for the memory-fabric demo: runs the conformance suite, the
|
||||
# routing / fault-fallback demo, the CXL/NUMA tier benchmark, and the GPU
|
||||
# NVLink benchmark, collects their CSV/JSON artifacts, and renders figures.
|
||||
#
|
||||
# Usage: run_demo.sh [BUILD_DIR]
|
||||
# BUILD_DIR defaults to the repository's ./build.
|
||||
# Artifacts and figures land in $MC_FABRIC_RESULTS (default: ./fabric-results).
|
||||
|
||||
set -u
|
||||
|
||||
HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
REPO_ROOT="$(cd "$HERE/../../.." && pwd)"
|
||||
BUILD_DIR="${1:-$REPO_ROOT/build}"
|
||||
RESULTS="${MC_FABRIC_RESULTS:-$REPO_ROOT/fabric-results}"
|
||||
export MC_FABRIC_RESULTS="$RESULTS"
|
||||
mkdir -p "$RESULTS"
|
||||
|
||||
TE_TESTS="$BUILD_DIR/mooncake-transfer-engine/tests"
|
||||
STORE_BENCH="$BUILD_DIR/mooncake-store/benchmarks"
|
||||
|
||||
section() { printf '\n========== %s ==========\n' "$1"; }
|
||||
have() { [ -x "$1" ]; }
|
||||
|
||||
section "1. Cross-backend conformance"
|
||||
if have "$TE_TESTS/conformance_runner"; then
|
||||
"$TE_TESTS/conformance_runner" --json "$RESULTS/conformance.json" \
|
||||
2>/dev/null | tee "$RESULTS/conformance.txt"
|
||||
else
|
||||
echo "conformance_runner not built (configure with -DBUILD_UNIT_TESTS=ON)"
|
||||
fi
|
||||
|
||||
section "2. Probed capabilities + routing / fault fallback"
|
||||
if have "$TE_TESTS/fabric_cli"; then
|
||||
{
|
||||
"$TE_TESTS/fabric_cli" probe
|
||||
echo
|
||||
"$TE_TESTS/fabric_cli" select
|
||||
echo
|
||||
"$TE_TESTS/fabric_cli" cost
|
||||
if [ -d "$BUILD_DIR/plugins" ]; then
|
||||
echo
|
||||
"$TE_TESTS/fabric_cli" plugins "$BUILD_DIR/plugins"
|
||||
fi
|
||||
} 2>/dev/null | tee "$RESULTS/fabric_cli.txt"
|
||||
else
|
||||
echo "fabric_cli not built"
|
||||
fi
|
||||
|
||||
section "3. Host fabric topology"
|
||||
python3 "$HERE/topology.py" 2>&1 | tee "$RESULTS/topology.txt"
|
||||
|
||||
section "4. CXL/NUMA memory tier benchmark"
|
||||
if have "$STORE_BENCH/tier_bench"; then
|
||||
"$STORE_BENCH/tier_bench" --csv "$RESULTS/tier_bench.csv" \
|
||||
--nvme_path "$RESULTS/.tier_nvme.bin" --trace 16000 \
|
||||
2>/dev/null | tee "$RESULTS/tier_bench.txt"
|
||||
rm -f "$RESULTS/.tier_nvme.bin"
|
||||
else
|
||||
echo "tier_bench not built (configure with -DWITH_STORE=ON)"
|
||||
fi
|
||||
|
||||
section "5. Real GPU NVLink vs host-staging"
|
||||
MC_GPU_BENCH_CSV="$RESULTS/gpu_bench.csv" \
|
||||
python3 "$HERE/gpu_bench.py" 2>&1 | tee "$RESULTS/gpu_bench.txt"
|
||||
|
||||
section "6. Rendering figures"
|
||||
python3 "$HERE/plot.py"
|
||||
|
||||
section "7. Assembling summary.html"
|
||||
python3 "$HERE/summary.py"
|
||||
|
||||
printf '\nArtifacts and figures in %s\n' "$RESULTS"
|
||||
|
|
@ -1,121 +0,0 @@
|
|||
#!/usr/bin/env python3
|
||||
# Copyright 2024 KVCache.AI
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
"""Assemble a single self-contained summary.html from the demo artifacts in the
|
||||
results directory (default ./fabric-results, override with MC_FABRIC_RESULTS):
|
||||
the conformance matrix, probe/routing/cost text, the topology view, and every
|
||||
PNG figure. Missing pieces are skipped so a partial run still produces a page."""
|
||||
|
||||
import html
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
|
||||
RESULTS = os.environ.get("MC_FABRIC_RESULTS", os.path.abspath("fabric-results"))
|
||||
|
||||
|
||||
def read(path):
|
||||
p = os.path.join(RESULTS, path)
|
||||
return open(p).read() if os.path.exists(p) else None
|
||||
|
||||
|
||||
def section(title, body):
|
||||
return f"<h2>{html.escape(title)}</h2>\n{body}\n"
|
||||
|
||||
|
||||
def pre(text):
|
||||
return f"<pre>{html.escape(text.rstrip())}</pre>" if text else ""
|
||||
|
||||
|
||||
def img(name, caption):
|
||||
if not os.path.exists(os.path.join(RESULTS, name)):
|
||||
return ""
|
||||
return (f'<figure><img src="{name}" style="max-width:760px;width:100%">'
|
||||
f"<figcaption>{html.escape(caption)}</figcaption></figure>")
|
||||
|
||||
|
||||
def conformance_table():
|
||||
data = read("conformance.json")
|
||||
if not data:
|
||||
return ""
|
||||
try:
|
||||
rows = json.loads(data)
|
||||
except Exception:
|
||||
return ""
|
||||
cases = ["capability", "register", "roundtrip", "batch", "ordering",
|
||||
"concurrency"]
|
||||
out = ['<table border="1" cellpadding="6" cellspacing="0">']
|
||||
out.append("<tr><th>backend</th>" +
|
||||
"".join(f"<th>{c}</th>" for c in cases) + "<th>verdict</th></tr>")
|
||||
for e in rows:
|
||||
by = {c["name"]: c["outcome"] for c in e.get("cases", [])}
|
||||
cells = []
|
||||
for c in cases:
|
||||
o = by.get(c, "skip")
|
||||
color = {"PASS": "#d9ead3", "FAIL": "#f4cccc"}.get(o, "#eeeeee")
|
||||
cells.append(f'<td style="background:{color}">{o}</td>')
|
||||
verdict = "CONFORMANT" if e.get("conformant") else "NON-CONFORMANT"
|
||||
vcolor = "#d9ead3" if e.get("conformant") else "#f4cccc"
|
||||
out.append(f"<tr><td><b>{html.escape(e['backend'])}</b></td>" +
|
||||
"".join(cells) +
|
||||
f'<td style="background:{vcolor}"><b>{verdict}</b></td></tr>')
|
||||
out.append("</table>")
|
||||
return "\n".join(out)
|
||||
|
||||
|
||||
def main():
|
||||
parts = [
|
||||
"<!doctype html><html><head><meta charset='utf-8'>",
|
||||
"<title>Mooncake Memory Fabric — demo summary</title>",
|
||||
"<style>body{font-family:system-ui,Arial,sans-serif;margin:32px;"
|
||||
"max-width:900px}h1{margin-bottom:0}pre{background:#f6f8fa;padding:12px;"
|
||||
"overflow:auto;font-size:13px}figure{margin:16px 0}"
|
||||
"figcaption{color:#555;font-size:13px}table{border-collapse:collapse}"
|
||||
"</style></head><body>",
|
||||
"<h1>Mooncake Memory Fabric</h1>",
|
||||
"<p>Capability-probe + conformance layer, capability-driven routing, a "
|
||||
"CXL/NUMA memory tier, and an honestly-emulated UALink backend — one "
|
||||
"demo, every backend.</p>",
|
||||
]
|
||||
|
||||
table = conformance_table()
|
||||
if table:
|
||||
parts.append(section("Cross-backend conformance (one suite, every "
|
||||
"backend)", table + img("conformance.png", "")))
|
||||
parts.append(section("Probed capabilities",
|
||||
img("capabilities.png",
|
||||
"green = real hardware, red = honestly emulated")))
|
||||
parts.append(section("Routing, fault fallback, and size-aware cost model",
|
||||
pre(read("fabric_cli.txt"))))
|
||||
parts.append(section("Host fabric topology",
|
||||
img("topology.png", "GPUs, NVLink mesh, NUMA, probed "
|
||||
"transports") + pre(read("topology.txt"))))
|
||||
parts.append(section("CXL/NUMA memory tier",
|
||||
img("tier_nvme.png",
|
||||
"CXL tier absorbs warm reads -> fewer NVMe reads") +
|
||||
img("tier_latency.png", "") + pre(read("tier_bench.txt"))))
|
||||
parts.append(section("Real NVLink vs host-staging",
|
||||
img("gpu_bench.png", "") + pre(read("gpu_bench.txt"))))
|
||||
parts.append("</body></html>")
|
||||
|
||||
out_path = os.path.join(RESULTS, "summary.html")
|
||||
with open(out_path, "w") as f:
|
||||
f.write("\n".join(p for p in parts if p))
|
||||
print(f"summary -> {out_path}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
|
|
@ -1,163 +0,0 @@
|
|||
#!/usr/bin/env python3
|
||||
# Copyright 2024 KVCache.AI
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
"""Render the host's memory-fabric topology: GPUs and their NVLink mesh, NICs,
|
||||
NUMA nodes, and the probed transport capabilities, as a text summary and a
|
||||
Graphviz DOT file.
|
||||
|
||||
Sources, each optional and degraded gracefully:
|
||||
- nvidia-smi topo -m (GPU<->GPU links, GPU<->NIC, NUMA affinity)
|
||||
- numactl --hardware (NUMA nodes)
|
||||
- the conformance.json probe report (per-backend capability), if present in
|
||||
MC_FABRIC_RESULTS.
|
||||
|
||||
Writes <results>/topology.dot (+ topology.png if Graphviz `dot` is installed)."""
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
RESULTS = os.environ.get("MC_FABRIC_RESULTS", os.path.abspath("fabric-results"))
|
||||
|
||||
|
||||
def run(cmd):
|
||||
try:
|
||||
return subprocess.run(cmd, capture_output=True, text=True,
|
||||
timeout=20).stdout
|
||||
except Exception:
|
||||
return ""
|
||||
|
||||
|
||||
def parse_nvidia_topo():
|
||||
"""Return (gpus, links, gpu_numa) from `nvidia-smi topo -m`."""
|
||||
out = run(["nvidia-smi", "topo", "-m"])
|
||||
if not out:
|
||||
return [], {}, {}
|
||||
lines = [l for l in out.splitlines() if l.strip()]
|
||||
header = None
|
||||
rows = {}
|
||||
gpu_numa = {}
|
||||
for line in lines:
|
||||
cells = re.split(r"\s{1,}", line.strip())
|
||||
if cells and cells[0].startswith("GPU") and header is None and \
|
||||
"GPU0" in line and cells[0] == "GPU0":
|
||||
pass
|
||||
if cells[0].startswith("GPU") and re.match(r"GPU\d+$", cells[0]):
|
||||
name = cells[0]
|
||||
rows[name] = cells[1:]
|
||||
if header is None and line.lstrip().startswith("GPU0"):
|
||||
header = re.split(r"\s{1,}", line.strip())
|
||||
gpus = sorted(rows.keys(), key=lambda g: int(g[3:]))
|
||||
links = {}
|
||||
for g, cells in rows.items():
|
||||
for j, gj in enumerate(gpus):
|
||||
if j < len(cells) and cells[j].startswith("NV"):
|
||||
a, b = sorted([g, gj])
|
||||
if a != b:
|
||||
links[(a, b)] = cells[j] # e.g. NV18
|
||||
# NUMA affinity column: look for a numeric NUMA id per GPU row.
|
||||
for g, cells in rows.items():
|
||||
for c in cells:
|
||||
if re.match(r"^\d+$", c):
|
||||
gpu_numa[g] = int(c)
|
||||
break
|
||||
return gpus, links, gpu_numa
|
||||
|
||||
|
||||
def parse_numa():
|
||||
out = run(["numactl", "--hardware"])
|
||||
m = re.search(r"available:\s*(\d+)\s*nodes", out)
|
||||
return int(m.group(1)) if m else 0
|
||||
|
||||
|
||||
def probed_caps():
|
||||
path = os.path.join(RESULTS, "conformance.json")
|
||||
if not os.path.exists(path):
|
||||
return []
|
||||
try:
|
||||
data = json.load(open(path))
|
||||
except Exception:
|
||||
return []
|
||||
return [e.get("capability", {}) for e in data]
|
||||
|
||||
|
||||
def main():
|
||||
os.makedirs(RESULTS, exist_ok=True)
|
||||
gpus, links, gpu_numa = parse_nvidia_topo()
|
||||
numa_nodes = parse_numa()
|
||||
caps = probed_caps()
|
||||
|
||||
print("=== Memory fabric topology ===")
|
||||
if gpus:
|
||||
nvlink = sorted({v for v in links.values()})
|
||||
print(f"GPUs: {len(gpus)} NVLink mesh edges: {len(links)} "
|
||||
f"({','.join(nvlink)})")
|
||||
else:
|
||||
print("GPUs: none detected (nvidia-smi unavailable)")
|
||||
if numa_nodes:
|
||||
print(f"NUMA nodes: {numa_nodes}")
|
||||
if caps:
|
||||
print("Probed transports:")
|
||||
for c in caps:
|
||||
tag = "EMULATED" if c.get("emulated") else "real"
|
||||
print(f" {c.get('name', '?'):12s} {c.get('kind', '?'):7s} {tag}")
|
||||
|
||||
# Graphviz DOT.
|
||||
dot = ["graph fabric {", ' rankdir=LR;', ' node [fontsize=10];']
|
||||
by_numa = {}
|
||||
for g in gpus:
|
||||
by_numa.setdefault(gpu_numa.get(g, 0), []).append(g)
|
||||
for node, members in sorted(by_numa.items()):
|
||||
dot.append(f' subgraph cluster_numa{node} {{')
|
||||
dot.append(f' label="NUMA node {node}"; color="#888888";')
|
||||
for g in members:
|
||||
dot.append(f' {g} [shape=box, style=filled, '
|
||||
f'fillcolor="#cfe8cf"];')
|
||||
dot.append(" }")
|
||||
for (a, b), kind in sorted(links.items()):
|
||||
dot.append(f' {a} -- {b} [color="#2ca02c", penwidth=2, '
|
||||
f'label="{kind}"];')
|
||||
# Fabric capability legend nodes.
|
||||
for c in caps:
|
||||
name = c.get("name", "?").replace(":", "_").replace(".", "_")
|
||||
color = "#f4cccc" if c.get("emulated") else "#d9ead3"
|
||||
label = f"{c.get('name','?')}\\n{c.get('kind','?')}"
|
||||
dot.append(f' fab_{name} [shape=ellipse, style=filled, '
|
||||
f'fillcolor="{color}", label="{label}"];')
|
||||
dot.append("}")
|
||||
|
||||
dot_path = os.path.join(RESULTS, "topology.dot")
|
||||
with open(dot_path, "w") as f:
|
||||
f.write("\n".join(dot) + "\n")
|
||||
print(f"\nDOT -> {dot_path}")
|
||||
|
||||
if shutil.which("dot"):
|
||||
png = os.path.join(RESULTS, "topology.png")
|
||||
try:
|
||||
subprocess.run(["dot", "-Tpng", dot_path, "-o", png], timeout=30,
|
||||
check=True)
|
||||
print(f"PNG -> {png}")
|
||||
except Exception as exc:
|
||||
print(f"(graphviz render skipped: {exc})")
|
||||
else:
|
||||
print("(install graphviz `dot` to render topology.png)")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
|
|
@ -1,110 +0,0 @@
|
|||
// Copyright 2024 KVCache.AI
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
#ifndef FABRIC_CAPABILITY_H_
|
||||
#define FABRIC_CAPABILITY_H_
|
||||
|
||||
#include <cstdint>
|
||||
#include <string>
|
||||
|
||||
namespace mooncake {
|
||||
namespace fabric {
|
||||
|
||||
// The physical / logical interconnect a transport rides on. Mirrors the set of
|
||||
// protocol strings dispatched by MultiTransport::selectTransport(), plus the
|
||||
// emulated host-memory kinds used to model cache-coherent fabrics where no
|
||||
// silicon is present on the host.
|
||||
enum class MemoryFabricKind : int {
|
||||
UNKNOWN = 0,
|
||||
TCP, // commodity sockets, the universal fallback
|
||||
RDMA, // InfiniBand / RoCE one-sided verbs
|
||||
NVLINK, // NVIDIA GPU<->GPU high-bandwidth peer access
|
||||
PCIE, // GPU<->GPU or GPU<->host over PCIe (P2P or staged)
|
||||
CXL, // Compute Express Link memory window (coherent load/store)
|
||||
UB, // Huawei Unified Bus (Kunpeng / Ascend)
|
||||
UALINK, // Ultra Accelerator Link
|
||||
SHM, // POSIX shared-memory window (host-local, CXL-like semantics)
|
||||
NUMA, // explicit NUMA-placed DRAM (models a CXL-attached memory tier)
|
||||
};
|
||||
|
||||
const char *toString(MemoryFabricKind kind);
|
||||
MemoryFabricKind fabricKindFromString(const std::string &name);
|
||||
|
||||
// Map a MultiTransport protocol string ("rdma", "tcp", "cxl", "ub", "nvlink",
|
||||
// "nvlink_intra", "ualink", ...) to its fabric kind. Used by the default
|
||||
// Transport::probe() so a backend that does not override probe() still reports
|
||||
// a sensible kind.
|
||||
MemoryFabricKind fabricKindFromProtocol(const std::string &protocol);
|
||||
|
||||
// Liveness of a backend at probe time. Generalises the intent of
|
||||
// TransferEngine::probePeerAliveByID() across every transport so the path
|
||||
// selector can consult a single uniform signal.
|
||||
enum class HealthStatus : int {
|
||||
UNINITIALIZED = 0, // backend object exists but install() has not run
|
||||
HEALTHY, // ready to move bytes
|
||||
DEGRADED, // works but a rail / peer is impaired
|
||||
UNREACHABLE, // cannot move bytes right now; selector must skip it
|
||||
};
|
||||
|
||||
const char *toString(HealthStatus status);
|
||||
|
||||
// Uniform, machine-readable capability descriptor returned by
|
||||
// Transport::probe(). Intentionally a flat POD so it can be serialised to JSON,
|
||||
// compared, and shipped across a wire for remote topology assembly.
|
||||
struct Capability {
|
||||
MemoryFabricKind kind = MemoryFabricKind::UNKNOWN;
|
||||
std::string name; // backend instance name, e.g. "cxl", "rdma:mlx5_0"
|
||||
|
||||
// Functional feature bits.
|
||||
bool supports_device_memory = false; // can register / move GPU memory
|
||||
bool supports_host_memory = true; // can register / move host memory
|
||||
bool supports_p2p = false; // direct peer access (no host bounce)
|
||||
bool supports_ordered_write = false; // writes to one peer land in order
|
||||
bool supports_atomic = false; // remote atomics (fetch-add / CAS)
|
||||
bool supports_zero_copy = false; // no intermediate staging buffer
|
||||
bool supports_multi_rail = false; // can stripe across >1 physical link
|
||||
bool remote_capable = false; // can cross a process / host boundary
|
||||
|
||||
// Honesty flag: when true this backend has no real silicon on this host and
|
||||
// its quantitative numbers describe an emulation explained in `notes`.
|
||||
bool emulated = false;
|
||||
|
||||
// Quantitative shape.
|
||||
uint64_t max_transfer_size = 0; // bytes; 0 == effectively unlimited
|
||||
uint64_t alignment = 1; // required address / length alignment
|
||||
uint64_t max_bandwidth_mbps = 0; // MB/s, calibrated by probe()
|
||||
uint64_t latency_ns = 0; // ns, small-message round-trip estimate
|
||||
|
||||
// Placement hints.
|
||||
int numa_node = -1; // affinity of this backend's memory
|
||||
int device_id = -1; // GPU / NIC ordinal, -1 if not applicable
|
||||
|
||||
std::string notes; // free-form, e.g. "emulated via NUMA node 1"
|
||||
|
||||
// Serialise to a single-line JSON object (no external json dependency, so
|
||||
// this stays usable from the conformance runner and CLI tools).
|
||||
std::string toJson() const;
|
||||
|
||||
// One-line human-readable summary for the CLI and logs.
|
||||
std::string toLine() const;
|
||||
|
||||
// Self-consistency check used by the conformance "capability" case: returns
|
||||
// "" when consistent, otherwise a human-readable reason for the violation.
|
||||
std::string checkConsistency() const;
|
||||
};
|
||||
|
||||
} // namespace fabric
|
||||
} // namespace mooncake
|
||||
|
||||
#endif // FABRIC_CAPABILITY_H_
|
||||
|
|
@ -1,118 +0,0 @@
|
|||
// Copyright 2024 KVCache.AI
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
#ifndef FABRIC_PATH_SELECTOR_H_
|
||||
#define FABRIC_PATH_SELECTOR_H_
|
||||
|
||||
#include <functional>
|
||||
#include <string>
|
||||
#include <vector>
|
||||
|
||||
#include "fabric/capability.h"
|
||||
#include "transport/transport.h"
|
||||
|
||||
namespace mooncake {
|
||||
namespace fabric {
|
||||
|
||||
// An endpoint a transfer can originate from or land on. The selector only needs
|
||||
// the few attributes that decide which interconnect class applies.
|
||||
struct Endpoint {
|
||||
enum class Kind { GPU, HOST, STORE };
|
||||
Kind kind = Kind::HOST;
|
||||
int device_id = -1; // GPU ordinal, if a GPU
|
||||
int numa_node = -1; // socket affinity
|
||||
std::string host = "node0"; // node identity (same vs cross node)
|
||||
std::string name; // human label, e.g. "gpu0", "store1"
|
||||
};
|
||||
|
||||
enum class PathClass {
|
||||
SAME_NODE_GPU_GPU,
|
||||
CROSS_NODE_GPU_GPU,
|
||||
HOST_HOST,
|
||||
MEMORY_TIER,
|
||||
};
|
||||
const char *toString(PathClass cls);
|
||||
|
||||
struct PathChoice {
|
||||
bool found = false;
|
||||
std::string transport_name;
|
||||
MemoryFabricKind kind = MemoryFabricKind::UNKNOWN;
|
||||
std::string label; // "nvlink_intra", "pinned_host_tcp", ...
|
||||
std::string reason; // why this one was chosen and what was skipped
|
||||
bool is_fallback = false;
|
||||
double estimated_cost_ns = 0; // modelled cost when a size is given
|
||||
};
|
||||
|
||||
// Turns the uniform capability + health view of every registered transport into
|
||||
// an automatic, explainable routing decision with health-based fallback.
|
||||
//
|
||||
// Two selection policies share the same registry and fallback semantics:
|
||||
// * the priority ladder per interconnect class (the default), e.g.
|
||||
// same_node_gpu_gpu : NVLINK > PCIE > CXL/SHM > TCP
|
||||
// cross_node_gpu_gpu: RDMA > TCP
|
||||
// host_host : CXL/SHM > UB > UALINK > RDMA > NUMA > TCP
|
||||
// memory_tier : CXL > NUMA > RDMA > TCP
|
||||
// * a calibrated cost model when a transfer size is given: among the healthy
|
||||
// candidates eligible for the class, pick the one with the lowest modelled
|
||||
// cost latency_ns + size / bandwidth, so the choice is size-aware
|
||||
// (small transfers favour low latency, large transfers favour bandwidth).
|
||||
class PathSelector {
|
||||
public:
|
||||
// Register a live transport; the selector reads its probed capability once
|
||||
// and queries health() at selection time so fault injection / link loss
|
||||
// takes effect immediately.
|
||||
void registerTransport(Transport &transport);
|
||||
|
||||
// Register a synthetic entry (for tests, or when no object exists yet).
|
||||
// The optional capability carries the modelled bandwidth/latency used by
|
||||
// the cost-based selector; without it the entry participates in the ladder
|
||||
// only.
|
||||
void registerKind(const std::string &name, MemoryFabricKind kind,
|
||||
std::function<HealthStatus()> health);
|
||||
void registerKind(const std::string &name, MemoryFabricKind kind,
|
||||
std::function<HealthStatus()> health,
|
||||
const Capability &cap);
|
||||
|
||||
static PathClass classify(const Endpoint &src, const Endpoint &dst);
|
||||
|
||||
// Priority-ladder selection (size-agnostic).
|
||||
PathChoice select(const Endpoint &src, const Endpoint &dst) const;
|
||||
|
||||
// Cost-model selection for a transfer of `size_bytes`. Falls back to the
|
||||
// ladder when no eligible candidate advertises a calibrated bandwidth.
|
||||
PathChoice selectForSize(const Endpoint &src, const Endpoint &dst,
|
||||
uint64_t size_bytes) const;
|
||||
|
||||
// Format the plan-style log line for a decision and return it.
|
||||
std::string logChoice(const Endpoint &src, const Endpoint &dst,
|
||||
const PathChoice &choice) const;
|
||||
|
||||
// The static priority ladder for a class (kinds, highest priority first).
|
||||
static std::vector<MemoryFabricKind> priorityFor(PathClass cls);
|
||||
|
||||
private:
|
||||
struct Entry {
|
||||
std::string name;
|
||||
MemoryFabricKind kind;
|
||||
std::function<HealthStatus()> health;
|
||||
Capability cap;
|
||||
bool has_cap = false;
|
||||
};
|
||||
std::vector<Entry> entries_;
|
||||
};
|
||||
|
||||
} // namespace fabric
|
||||
} // namespace mooncake
|
||||
|
||||
#endif // FABRIC_PATH_SELECTOR_H_
|
||||
|
|
@ -1,83 +0,0 @@
|
|||
// Copyright 2024 KVCache.AI
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
#ifndef FABRIC_PLUGIN_LOADER_H_
|
||||
#define FABRIC_PLUGIN_LOADER_H_
|
||||
|
||||
#include <memory>
|
||||
#include <string>
|
||||
#include <vector>
|
||||
|
||||
#include "transport/transport.h"
|
||||
|
||||
namespace mooncake {
|
||||
namespace fabric {
|
||||
|
||||
// Versioned C ABI for loading transport backends as shared objects at runtime.
|
||||
// A backend that cannot initialise on this host is skipped, never fatal -- the
|
||||
// "thin plugins / dynamic loading / standalone packaging" direction.
|
||||
constexpr int kPluginAbiVersion = 1;
|
||||
|
||||
extern "C" {
|
||||
typedef int (*plugin_abi_fn)();
|
||||
typedef const char *(*plugin_name_fn)();
|
||||
typedef Transport *(*plugin_create_fn)();
|
||||
typedef void (*plugin_destroy_fn)(Transport *);
|
||||
}
|
||||
|
||||
struct PluginInfo {
|
||||
std::string path;
|
||||
std::string name;
|
||||
int abi = 0;
|
||||
bool ok = false;
|
||||
std::string error;
|
||||
};
|
||||
|
||||
// Owns the dlopen handles; transports created from a loader must be destroyed
|
||||
// before the loader (declare the loader before the transports that use it).
|
||||
class PluginLoader {
|
||||
public:
|
||||
~PluginLoader();
|
||||
|
||||
// Load one .so. On success returns a Transport whose deleter calls the
|
||||
// plugin's destroy function; on failure returns nullptr and fills *info.
|
||||
std::unique_ptr<Transport, void (*)(Transport *)> loadFile(
|
||||
const std::string &path, PluginInfo *info);
|
||||
|
||||
// Load every matching transport plugin in a directory.
|
||||
std::vector<std::unique_ptr<Transport, void (*)(Transport *)>>
|
||||
loadDirectory(const std::string &dir, std::vector<PluginInfo> *infos);
|
||||
|
||||
private:
|
||||
std::vector<void *> handles_;
|
||||
};
|
||||
|
||||
} // namespace fabric
|
||||
} // namespace mooncake
|
||||
|
||||
// Emit the plugin C ABI for a transport. CREATE_EXPR must yield a
|
||||
// Transport* (heap-allocated; ownership transfers to the loader).
|
||||
#define MOONCAKE_DEFINE_TRANSPORT_PLUGIN(CREATE_EXPR, PLUGIN_NAME) \
|
||||
extern "C" int mooncake_plugin_abi_version() { \
|
||||
return mooncake::fabric::kPluginAbiVersion; \
|
||||
} \
|
||||
extern "C" const char *mooncake_plugin_name() { return PLUGIN_NAME; } \
|
||||
extern "C" mooncake::Transport *mooncake_plugin_create() { \
|
||||
return (CREATE_EXPR); \
|
||||
} \
|
||||
extern "C" void mooncake_plugin_destroy(mooncake::Transport *p) { \
|
||||
delete p; \
|
||||
}
|
||||
|
||||
#endif // FABRIC_PLUGIN_LOADER_H_
|
||||
|
|
@ -1,82 +0,0 @@
|
|||
// Copyright 2024 KVCache.AI
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
#ifndef FABRIC_PROBE_CALIBRATION_H_
|
||||
#define FABRIC_PROBE_CALIBRATION_H_
|
||||
|
||||
#include <algorithm>
|
||||
#include <chrono>
|
||||
#include <cstdint>
|
||||
#include <functional>
|
||||
#include <vector>
|
||||
|
||||
namespace mooncake {
|
||||
namespace fabric {
|
||||
|
||||
// A "move bytes" closure: perform one transfer of `nbytes` and return true on
|
||||
// success. Each backend wraps its own data path so probe() calibrates the same
|
||||
// honest way everywhere -- an emulated backend therefore reports the bandwidth
|
||||
// of its emulation, never an invented hardware figure.
|
||||
using MoveFn = std::function<bool(uint64_t nbytes)>;
|
||||
|
||||
struct CalibrationResult {
|
||||
uint64_t bandwidth_mbps = 0; // MB/s from the large-transfer loop
|
||||
uint64_t latency_ns = 0; // ns, median of tiny-transfer round trips
|
||||
bool ok = false;
|
||||
};
|
||||
|
||||
// Time `iters` large moves for bandwidth and `iters` tiny moves for latency.
|
||||
// Total work is kept small so probe() stays cheap to call repeatedly.
|
||||
inline CalibrationResult calibrate(const MoveFn &move,
|
||||
uint64_t large_bytes = (1u << 20),
|
||||
int iters = 16) {
|
||||
using clock = std::chrono::steady_clock;
|
||||
CalibrationResult result;
|
||||
|
||||
// Warm up once so the bandwidth loop measures steady state rather than
|
||||
// first-touch page faults or connection setup.
|
||||
if (!move(large_bytes)) return result;
|
||||
|
||||
auto t0 = clock::now();
|
||||
uint64_t moved = 0;
|
||||
for (int i = 0; i < iters; ++i) {
|
||||
if (!move(large_bytes)) return result;
|
||||
moved += large_bytes;
|
||||
}
|
||||
auto t1 = clock::now();
|
||||
double secs = std::chrono::duration<double>(t1 - t0).count();
|
||||
if (secs > 0) {
|
||||
double mb = static_cast<double>(moved) / (1024.0 * 1024.0);
|
||||
result.bandwidth_mbps = static_cast<uint64_t>(mb / secs);
|
||||
}
|
||||
|
||||
std::vector<double> samples;
|
||||
samples.reserve(iters);
|
||||
for (int i = 0; i < iters; ++i) {
|
||||
auto a = clock::now();
|
||||
if (!move(64)) return result;
|
||||
auto b = clock::now();
|
||||
samples.push_back(
|
||||
std::chrono::duration<double, std::nano>(b - a).count());
|
||||
}
|
||||
std::sort(samples.begin(), samples.end());
|
||||
result.latency_ns = static_cast<uint64_t>(samples[samples.size() / 2]);
|
||||
result.ok = true;
|
||||
return result;
|
||||
}
|
||||
|
||||
} // namespace fabric
|
||||
} // namespace mooncake
|
||||
|
||||
#endif // FABRIC_PROBE_CALIBRATION_H_
|
||||
|
|
@ -15,11 +15,8 @@
|
|||
#ifndef MULTI_TRANSPORT_H_
|
||||
#define MULTI_TRANSPORT_H_
|
||||
|
||||
#include <memory>
|
||||
#include <mutex>
|
||||
#include <unordered_map>
|
||||
|
||||
#include "fabric/path_selector.h"
|
||||
#include "transport/transport.h"
|
||||
|
||||
namespace mooncake {
|
||||
|
|
@ -85,12 +82,6 @@ class MultiTransport {
|
|||
std::map<std::string, std::shared_ptr<Transport>> transport_map_;
|
||||
RWSpinlock batch_desc_lock_;
|
||||
std::unordered_map<BatchID, std::shared_ptr<BatchDesc>> batch_desc_set_;
|
||||
|
||||
// Lazily built once per engine for MC_FABRIC_SELECTOR_DRY_RUN telemetry; it
|
||||
// caches this engine's transports, so its lifetime is bounded by the
|
||||
// engine.
|
||||
std::unique_ptr<fabric::PathSelector> dry_run_selector_;
|
||||
std::once_flag dry_run_once_;
|
||||
};
|
||||
} // namespace mooncake
|
||||
|
||||
|
|
|
|||
|
|
@ -50,14 +50,8 @@ class CxlTransport : public Transport {
|
|||
Status getTransferStatus(BatchID batch_id, size_t task_id,
|
||||
TransferStatus &status) override;
|
||||
|
||||
fabric::Capability probe() override;
|
||||
|
||||
fabric::HealthStatus health() override;
|
||||
|
||||
void *getCxlBaseAddr() { return cxl_base_addr; }
|
||||
|
||||
size_t getCxlDevSize() const { return cxl_dev_size; }
|
||||
|
||||
private:
|
||||
int install(std::string &local_server_name,
|
||||
std::shared_ptr<TransferMetadata> meta,
|
||||
|
|
@ -91,12 +85,9 @@ class CxlTransport : public Transport {
|
|||
bool validateMemoryBounds(void *dest, void *src, size_t size);
|
||||
|
||||
private:
|
||||
void *cxl_base_addr = nullptr;
|
||||
size_t cxl_dev_size = 0;
|
||||
char *cxl_dev_path = nullptr;
|
||||
// True when the window is host memory standing in for a CXL device (no
|
||||
// /dev/dax present). The data plane is identical; probe() flags it.
|
||||
bool cxl_emulated_ = false;
|
||||
void *cxl_base_addr;
|
||||
size_t cxl_dev_size;
|
||||
char *cxl_dev_path;
|
||||
};
|
||||
} // namespace mooncake
|
||||
|
||||
|
|
|
|||
|
|
@ -96,10 +96,6 @@ class RdmaTransport : public Transport {
|
|||
Status getTransferStatus(BatchID batch_id, size_t task_id,
|
||||
TransferStatus &status) override;
|
||||
|
||||
fabric::Capability probe() override;
|
||||
|
||||
fabric::HealthStatus health() override;
|
||||
|
||||
SegmentID getSegmentID(const std::string &segment_name);
|
||||
|
||||
private:
|
||||
|
|
|
|||
|
|
@ -78,8 +78,6 @@ class TcpTransport : public Transport {
|
|||
Status getTransferStatus(BatchID batch_id, size_t task_id,
|
||||
TransferStatus &status) override;
|
||||
|
||||
fabric::Capability probe() override;
|
||||
|
||||
private:
|
||||
int install(std::string &local_server_name,
|
||||
std::shared_ptr<TransferMetadata> meta,
|
||||
|
|
|
|||
|
|
@ -32,7 +32,6 @@
|
|||
#include <condition_variable>
|
||||
|
||||
#include "common/base/status.h"
|
||||
#include "fabric/capability.h"
|
||||
#include "transfer_metadata.h"
|
||||
|
||||
namespace mooncake {
|
||||
|
|
@ -375,28 +374,6 @@ class Transport {
|
|||
}
|
||||
virtual Status CheckStatus(SegmentID sid) { return Status::OK(); }
|
||||
|
||||
/// @brief Report a uniform, machine-readable description of what this
|
||||
/// backend can do and roughly how fast, on this host, right now. The
|
||||
/// default returns a minimal descriptor derived from getName(); backends
|
||||
/// override to fill feature bits and calibrated bandwidth / latency.
|
||||
virtual fabric::Capability probe() {
|
||||
fabric::Capability cap;
|
||||
cap.name = getName();
|
||||
cap.kind = fabric::fabricKindFromProtocol(cap.name);
|
||||
cap.supports_host_memory = true;
|
||||
cap.remote_capable = true;
|
||||
return cap;
|
||||
}
|
||||
|
||||
/// @brief Report backend liveness, consulted by the capability-driven path
|
||||
/// selector before a backend is chosen. Generalises the intent of
|
||||
/// probePeerAliveByID() to a single uniform signal. The default reports
|
||||
/// HEALTHY once a metadata handle has been installed.
|
||||
virtual fabric::HealthStatus health() {
|
||||
return metadata_ ? fabric::HealthStatus::HEALTHY
|
||||
: fabric::HealthStatus::UNINITIALIZED;
|
||||
}
|
||||
|
||||
protected:
|
||||
virtual int install(std::string &local_server_name,
|
||||
std::shared_ptr<TransferMetadata> meta,
|
||||
|
|
|
|||
|
|
@ -1,74 +0,0 @@
|
|||
// Copyright 2024 KVCache.AI
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
#ifndef UALINK_SIMULATOR_H_
|
||||
#define UALINK_SIMULATOR_H_
|
||||
|
||||
#include <atomic>
|
||||
#include <cstdint>
|
||||
#include <utility>
|
||||
#include <vector>
|
||||
|
||||
namespace mooncake {
|
||||
|
||||
// UALink (Ultra Accelerator Link) has no upstream backend; this is green field.
|
||||
// Rather than fabricate hardware numbers, the simulator defines and enforces
|
||||
// the semantics a real UALink transport must honour so the conformance suite
|
||||
// can verify them in software:
|
||||
// 1. Ordered delivery -- writes issued on one logical link become visible at
|
||||
// the peer in submission order (release publish + monotonic sequence).
|
||||
// 2. Switch-connected point-to-point -- a small set of accelerator ports is
|
||||
// discoverable via topology(), modelling a UALink switch fabric.
|
||||
// Everything runs over host memory and the owning transport flags the
|
||||
// capability as emulated. When real silicon appears, orderedWrite/orderedRead
|
||||
// are replaced by the driver's posted-write path while the semantics, and these
|
||||
// tests, stay identical.
|
||||
|
||||
struct UALinkPort {
|
||||
int id = 0;
|
||||
int numa_hint = -1;
|
||||
};
|
||||
|
||||
class UALinkSimulator {
|
||||
public:
|
||||
explicit UALinkSimulator(int num_ports = 4);
|
||||
|
||||
const std::vector<UALinkPort> &ports() const { return ports_; }
|
||||
|
||||
int numPorts() const { return static_cast<int>(ports_.size()); }
|
||||
|
||||
// Discover the fabric: undirected edges of a switch-connected full mesh
|
||||
// among the ports. Stand-in for the real topology query.
|
||||
std::vector<std::pair<int, int>> topology() const;
|
||||
|
||||
// Ordered posted-write: copy src->dst, publish with release ordering, and
|
||||
// return a monotonically increasing per-fabric sequence number. A reader
|
||||
// that observes a later write is guaranteed to observe all earlier ones.
|
||||
uint64_t orderedWrite(void *dst, const void *src, uint64_t n);
|
||||
|
||||
// Acquire-ordered read mirroring orderedWrite.
|
||||
void orderedRead(void *dst, const void *src, uint64_t n);
|
||||
|
||||
uint64_t deliveredCount() const {
|
||||
return seq_.load(std::memory_order_acquire);
|
||||
}
|
||||
|
||||
private:
|
||||
std::vector<UALinkPort> ports_;
|
||||
std::atomic<uint64_t> seq_{0};
|
||||
};
|
||||
|
||||
} // namespace mooncake
|
||||
|
||||
#endif // UALINK_SIMULATOR_H_
|
||||
|
|
@ -1,100 +0,0 @@
|
|||
// Copyright 2024 KVCache.AI
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
#ifndef UALINK_TRANSPORT_H_
|
||||
#define UALINK_TRANSPORT_H_
|
||||
|
||||
#include <cstddef>
|
||||
#include <memory>
|
||||
#include <string>
|
||||
#include <vector>
|
||||
|
||||
#include "transfer_metadata.h"
|
||||
#include "transport/transport.h"
|
||||
#include "transport/ualink_transport/ualink_simulator.h"
|
||||
|
||||
namespace mooncake {
|
||||
class TransferMetadata;
|
||||
|
||||
// Green-field UALink backend. It mirrors the data-plane shape of CxlTransport
|
||||
// (a shared host-memory window addressed by offset) but routes every chunk
|
||||
// through a UALinkSimulator so the defined ordered-delivery semantics are
|
||||
// actually exercised. The capability it reports is flagged emulated; the
|
||||
// conformance "ordering" case verifies the guarantee.
|
||||
class UaLinkTransport : public Transport {
|
||||
public:
|
||||
using BufferDesc = TransferMetadata::BufferDesc;
|
||||
using SegmentDesc = TransferMetadata::SegmentDesc;
|
||||
|
||||
public:
|
||||
UaLinkTransport();
|
||||
|
||||
~UaLinkTransport();
|
||||
|
||||
Status submitTransfer(BatchID batch_id,
|
||||
const std::vector<TransferRequest> &entries) override;
|
||||
|
||||
Status submitTransferTask(
|
||||
const std::vector<TransferTask *> &task_list) override;
|
||||
|
||||
Status getTransferStatus(BatchID batch_id, size_t task_id,
|
||||
TransferStatus &status) override;
|
||||
|
||||
fabric::Capability probe() override;
|
||||
|
||||
fabric::HealthStatus health() override;
|
||||
|
||||
UALinkSimulator &simulator() { return sim_; }
|
||||
|
||||
void *windowBase() const { return window_base_; }
|
||||
|
||||
size_t windowSize() const { return window_size_; }
|
||||
|
||||
private:
|
||||
int install(std::string &local_server_name,
|
||||
std::shared_ptr<TransferMetadata> meta,
|
||||
std::shared_ptr<Topology> topo) override;
|
||||
|
||||
int allocateLocalSegmentID();
|
||||
|
||||
int registerLocalMemory(void *addr, size_t length,
|
||||
const std::string &location, bool remote_accessible,
|
||||
bool update_metadata) override;
|
||||
|
||||
int unregisterLocalMemory(void *addr,
|
||||
bool update_metadata = false) override;
|
||||
|
||||
int registerLocalMemoryBatch(
|
||||
const std::vector<Transport::BufferEntry> &buffer_list,
|
||||
const std::string &location) override;
|
||||
|
||||
int unregisterLocalMemoryBatch(
|
||||
const std::vector<void *> &addr_list) override;
|
||||
|
||||
const char *getName() const override { return "ualink"; }
|
||||
|
||||
bool windowInit();
|
||||
|
||||
int moveChunk(void *dst, void *src, size_t size,
|
||||
TransferRequest::OpCode op);
|
||||
|
||||
private:
|
||||
void *window_base_ = nullptr;
|
||||
size_t window_size_ = 0;
|
||||
UALinkSimulator sim_;
|
||||
};
|
||||
|
||||
} // namespace mooncake
|
||||
|
||||
#endif // UALINK_TRANSPORT_H_
|
||||
|
|
@ -1,6 +1,5 @@
|
|||
file(GLOB ENGINE_SOURCES "*.cpp")
|
||||
add_subdirectory(common)
|
||||
add_subdirectory(fabric)
|
||||
add_subdirectory(transport)
|
||||
|
||||
# EFA library path is set globally via common.cmake (LIBFABRIC_LIB_DIR)
|
||||
|
|
@ -11,8 +10,7 @@ if(USE_HIP)
|
|||
hipify_files(ENGINE_SOURCES)
|
||||
endif()
|
||||
|
||||
add_library(transfer_engine ${ENGINE_SOURCES} $<TARGET_OBJECTS:transport>
|
||||
$<TARGET_OBJECTS:fabric>)
|
||||
add_library(transfer_engine ${ENGINE_SOURCES} $<TARGET_OBJECTS:transport>)
|
||||
if(BUILD_SHARED_LIBS)
|
||||
install(TARGETS transfer_engine DESTINATION lib)
|
||||
endif()
|
||||
|
|
|
|||
|
|
@ -1,4 +0,0 @@
|
|||
file(GLOB FABRIC_SOURCES "*.cpp")
|
||||
|
||||
add_library(fabric OBJECT ${FABRIC_SOURCES})
|
||||
target_link_libraries(fabric PRIVATE glog::glog ${CMAKE_DL_LIBS})
|
||||
|
|
@ -1,159 +0,0 @@
|
|||
// Copyright 2024 KVCache.AI
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
#include "fabric/capability.h"
|
||||
|
||||
#include <sstream>
|
||||
#include <unordered_map>
|
||||
|
||||
namespace mooncake {
|
||||
namespace fabric {
|
||||
|
||||
const char *toString(MemoryFabricKind kind) {
|
||||
switch (kind) {
|
||||
case MemoryFabricKind::TCP:
|
||||
return "TCP";
|
||||
case MemoryFabricKind::RDMA:
|
||||
return "RDMA";
|
||||
case MemoryFabricKind::NVLINK:
|
||||
return "NVLINK";
|
||||
case MemoryFabricKind::PCIE:
|
||||
return "PCIE";
|
||||
case MemoryFabricKind::CXL:
|
||||
return "CXL";
|
||||
case MemoryFabricKind::UB:
|
||||
return "UB";
|
||||
case MemoryFabricKind::UALINK:
|
||||
return "UALINK";
|
||||
case MemoryFabricKind::SHM:
|
||||
return "SHM";
|
||||
case MemoryFabricKind::NUMA:
|
||||
return "NUMA";
|
||||
case MemoryFabricKind::UNKNOWN:
|
||||
default:
|
||||
return "UNKNOWN";
|
||||
}
|
||||
}
|
||||
|
||||
MemoryFabricKind fabricKindFromString(const std::string &name) {
|
||||
static const std::unordered_map<std::string, MemoryFabricKind> table = {
|
||||
{"TCP", MemoryFabricKind::TCP}, {"RDMA", MemoryFabricKind::RDMA},
|
||||
{"NVLINK", MemoryFabricKind::NVLINK}, {"PCIE", MemoryFabricKind::PCIE},
|
||||
{"CXL", MemoryFabricKind::CXL}, {"UB", MemoryFabricKind::UB},
|
||||
{"UALINK", MemoryFabricKind::UALINK}, {"SHM", MemoryFabricKind::SHM},
|
||||
{"NUMA", MemoryFabricKind::NUMA},
|
||||
};
|
||||
auto it = table.find(name);
|
||||
return it == table.end() ? MemoryFabricKind::UNKNOWN : it->second;
|
||||
}
|
||||
|
||||
MemoryFabricKind fabricKindFromProtocol(const std::string &protocol) {
|
||||
static const std::unordered_map<std::string, MemoryFabricKind> table = {
|
||||
{"tcp", MemoryFabricKind::TCP},
|
||||
{"rdma", MemoryFabricKind::RDMA},
|
||||
{"efa", MemoryFabricKind::RDMA},
|
||||
{"barex", MemoryFabricKind::RDMA},
|
||||
{"nvlink", MemoryFabricKind::NVLINK},
|
||||
{"nvlink_intra", MemoryFabricKind::NVLINK},
|
||||
{"nvlink_intraNode", MemoryFabricKind::NVLINK},
|
||||
{"cxl", MemoryFabricKind::CXL},
|
||||
{"ub", MemoryFabricKind::UB},
|
||||
{"ubshmem", MemoryFabricKind::UB},
|
||||
{"ualink", MemoryFabricKind::UALINK},
|
||||
{"nvmeof", MemoryFabricKind::PCIE},
|
||||
};
|
||||
auto it = table.find(protocol);
|
||||
return it == table.end() ? MemoryFabricKind::UNKNOWN : it->second;
|
||||
}
|
||||
|
||||
const char *toString(HealthStatus status) {
|
||||
switch (status) {
|
||||
case HealthStatus::HEALTHY:
|
||||
return "HEALTHY";
|
||||
case HealthStatus::DEGRADED:
|
||||
return "DEGRADED";
|
||||
case HealthStatus::UNREACHABLE:
|
||||
return "UNREACHABLE";
|
||||
case HealthStatus::UNINITIALIZED:
|
||||
default:
|
||||
return "UNINITIALIZED";
|
||||
}
|
||||
}
|
||||
|
||||
namespace {
|
||||
const char *boolStr(bool value) { return value ? "true" : "false"; }
|
||||
} // namespace
|
||||
|
||||
std::string Capability::toJson() const {
|
||||
std::ostringstream o;
|
||||
o << "{"
|
||||
<< "\"kind\":\"" << toString(kind) << "\","
|
||||
<< "\"name\":\"" << name << "\","
|
||||
<< "\"supports_device_memory\":" << boolStr(supports_device_memory) << ","
|
||||
<< "\"supports_host_memory\":" << boolStr(supports_host_memory) << ","
|
||||
<< "\"supports_p2p\":" << boolStr(supports_p2p) << ","
|
||||
<< "\"supports_ordered_write\":" << boolStr(supports_ordered_write) << ","
|
||||
<< "\"supports_atomic\":" << boolStr(supports_atomic) << ","
|
||||
<< "\"supports_zero_copy\":" << boolStr(supports_zero_copy) << ","
|
||||
<< "\"supports_multi_rail\":" << boolStr(supports_multi_rail) << ","
|
||||
<< "\"remote_capable\":" << boolStr(remote_capable) << ","
|
||||
<< "\"emulated\":" << boolStr(emulated) << ","
|
||||
<< "\"max_transfer_size\":" << max_transfer_size << ","
|
||||
<< "\"alignment\":" << alignment << ","
|
||||
<< "\"max_bandwidth_mbps\":" << max_bandwidth_mbps << ","
|
||||
<< "\"latency_ns\":" << latency_ns << ","
|
||||
<< "\"numa_node\":" << numa_node << ","
|
||||
<< "\"device_id\":" << device_id << ","
|
||||
<< "\"notes\":\"" << notes << "\""
|
||||
<< "}";
|
||||
return o.str();
|
||||
}
|
||||
|
||||
std::string Capability::toLine() const {
|
||||
std::ostringstream o;
|
||||
o << name << " [" << toString(kind) << (emulated ? ",EMULATED" : "") << "]"
|
||||
<< " dev=" << boolStr(supports_device_memory)
|
||||
<< " p2p=" << boolStr(supports_p2p)
|
||||
<< " ordered=" << boolStr(supports_ordered_write)
|
||||
<< " zerocopy=" << boolStr(supports_zero_copy)
|
||||
<< " multirail=" << boolStr(supports_multi_rail)
|
||||
<< " remote=" << boolStr(remote_capable) << " bw=" << max_bandwidth_mbps
|
||||
<< "MB/s lat=" << latency_ns << "ns";
|
||||
if (numa_node >= 0) o << " numa=" << numa_node;
|
||||
if (device_id >= 0) o << " dev_id=" << device_id;
|
||||
if (!notes.empty()) o << " (" << notes << ")";
|
||||
return o.str();
|
||||
}
|
||||
|
||||
std::string Capability::checkConsistency() const {
|
||||
if (kind == MemoryFabricKind::UNKNOWN) return "kind is UNKNOWN";
|
||||
if (name.empty()) return "name is empty";
|
||||
if (!supports_host_memory && !supports_device_memory)
|
||||
return "supports neither host nor device memory";
|
||||
if (alignment == 0) return "alignment must be >= 1";
|
||||
if ((alignment & (alignment - 1)) != 0)
|
||||
return "alignment must be a power of two";
|
||||
// Zero-copy implies no host staging, which only makes sense over a shared
|
||||
// window or true peer access; a stream socket cannot claim it.
|
||||
if (supports_zero_copy && kind == MemoryFabricKind::TCP)
|
||||
return "TCP cannot be zero-copy";
|
||||
if (supports_p2p && !supports_device_memory && !supports_host_memory)
|
||||
return "p2p declared with no memory kind";
|
||||
if (emulated && notes.empty())
|
||||
return "emulated backend must explain itself in notes";
|
||||
return "";
|
||||
}
|
||||
|
||||
} // namespace fabric
|
||||
} // namespace mooncake
|
||||
|
|
@ -1,254 +0,0 @@
|
|||
// Copyright 2024 KVCache.AI
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
#include "fabric/path_selector.h"
|
||||
|
||||
#include <sstream>
|
||||
|
||||
namespace mooncake {
|
||||
namespace fabric {
|
||||
|
||||
const char *toString(PathClass cls) {
|
||||
switch (cls) {
|
||||
case PathClass::SAME_NODE_GPU_GPU:
|
||||
return "same_node_gpu_gpu";
|
||||
case PathClass::CROSS_NODE_GPU_GPU:
|
||||
return "cross_node_gpu_gpu";
|
||||
case PathClass::HOST_HOST:
|
||||
return "host_host";
|
||||
case PathClass::MEMORY_TIER:
|
||||
return "memory_tier";
|
||||
}
|
||||
return "?";
|
||||
}
|
||||
|
||||
void PathSelector::registerTransport(Transport &transport) {
|
||||
fabric::Capability cap = transport.probe();
|
||||
std::string name = cap.name.empty() ? toString(cap.kind) : cap.name;
|
||||
Transport *ptr = &transport;
|
||||
Entry e;
|
||||
e.name = name;
|
||||
e.kind = cap.kind;
|
||||
e.health = [ptr] { return ptr->health(); };
|
||||
e.cap = cap;
|
||||
e.has_cap = true;
|
||||
entries_.push_back(std::move(e));
|
||||
}
|
||||
|
||||
void PathSelector::registerKind(const std::string &name, MemoryFabricKind kind,
|
||||
std::function<HealthStatus()> health) {
|
||||
Entry e;
|
||||
e.name = name;
|
||||
e.kind = kind;
|
||||
e.health = std::move(health);
|
||||
e.has_cap = false;
|
||||
entries_.push_back(std::move(e));
|
||||
}
|
||||
|
||||
void PathSelector::registerKind(const std::string &name, MemoryFabricKind kind,
|
||||
std::function<HealthStatus()> health,
|
||||
const Capability &cap) {
|
||||
Entry e;
|
||||
e.name = name;
|
||||
e.kind = kind;
|
||||
e.health = std::move(health);
|
||||
e.cap = cap;
|
||||
e.has_cap = true;
|
||||
entries_.push_back(std::move(e));
|
||||
}
|
||||
|
||||
PathClass PathSelector::classify(const Endpoint &src, const Endpoint &dst) {
|
||||
const bool both_gpu =
|
||||
src.kind == Endpoint::Kind::GPU && dst.kind == Endpoint::Kind::GPU;
|
||||
if (both_gpu) {
|
||||
return (src.host == dst.host) ? PathClass::SAME_NODE_GPU_GPU
|
||||
: PathClass::CROSS_NODE_GPU_GPU;
|
||||
}
|
||||
if (src.kind == Endpoint::Kind::STORE || dst.kind == Endpoint::Kind::STORE)
|
||||
return PathClass::MEMORY_TIER;
|
||||
return PathClass::HOST_HOST;
|
||||
}
|
||||
|
||||
std::vector<MemoryFabricKind> PathSelector::priorityFor(PathClass cls) {
|
||||
using K = MemoryFabricKind;
|
||||
switch (cls) {
|
||||
case PathClass::SAME_NODE_GPU_GPU:
|
||||
return {K::NVLINK, K::PCIE, K::CXL, K::SHM, K::TCP};
|
||||
case PathClass::CROSS_NODE_GPU_GPU:
|
||||
return {K::RDMA, K::TCP};
|
||||
case PathClass::HOST_HOST:
|
||||
return {K::CXL, K::SHM, K::UB, K::UALINK, K::RDMA, K::NUMA, K::TCP};
|
||||
case PathClass::MEMORY_TIER:
|
||||
return {K::CXL, K::NUMA, K::RDMA, K::TCP};
|
||||
}
|
||||
return {K::TCP};
|
||||
}
|
||||
|
||||
namespace {
|
||||
// Human label for a (class, kind) pair, matching the demo log vocabulary.
|
||||
std::string labelFor(PathClass cls, MemoryFabricKind kind) {
|
||||
using K = MemoryFabricKind;
|
||||
if (cls == PathClass::SAME_NODE_GPU_GPU) {
|
||||
if (kind == K::NVLINK) return "nvlink_intra";
|
||||
if (kind == K::PCIE) return "pcie_p2p";
|
||||
if (kind == K::CXL || kind == K::SHM) return "shm_p2p";
|
||||
if (kind == K::TCP) return "host_staging_tcp";
|
||||
}
|
||||
if (cls == PathClass::CROSS_NODE_GPU_GPU) {
|
||||
if (kind == K::RDMA) return "gdr_rdma";
|
||||
if (kind == K::TCP) return "pinned_host_tcp";
|
||||
}
|
||||
if (cls == PathClass::MEMORY_TIER) {
|
||||
if (kind == K::CXL) return "cxl_window";
|
||||
if (kind == K::NUMA) return "numa_tier";
|
||||
if (kind == K::RDMA) return "rdma_store";
|
||||
if (kind == K::TCP) return "store_tcp";
|
||||
}
|
||||
if (cls == PathClass::HOST_HOST) {
|
||||
if (kind == K::CXL || kind == K::SHM) return "shm_window";
|
||||
if (kind == K::UB) return "ub_link";
|
||||
if (kind == K::UALINK) return "ualink";
|
||||
if (kind == K::RDMA) return "rdma_host";
|
||||
if (kind == K::NUMA) return "numa_window";
|
||||
if (kind == K::TCP) return "host_tcp";
|
||||
}
|
||||
return std::string(toString(kind));
|
||||
}
|
||||
} // namespace
|
||||
|
||||
PathChoice PathSelector::select(const Endpoint &src,
|
||||
const Endpoint &dst) const {
|
||||
PathClass cls = classify(src, dst);
|
||||
auto prio = priorityFor(cls);
|
||||
|
||||
PathChoice choice;
|
||||
std::vector<std::string> skipped;
|
||||
for (size_t rank = 0; rank < prio.size(); ++rank) {
|
||||
MemoryFabricKind want = prio[rank];
|
||||
const Entry *best = nullptr;
|
||||
bool present_but_unhealthy = false;
|
||||
for (auto &entry : entries_) {
|
||||
if (entry.kind != want) continue;
|
||||
if (entry.health() == HealthStatus::HEALTHY) {
|
||||
best = &entry;
|
||||
break;
|
||||
}
|
||||
present_but_unhealthy = true;
|
||||
}
|
||||
if (best) {
|
||||
choice.found = true;
|
||||
choice.transport_name = best->name;
|
||||
choice.kind = want;
|
||||
choice.label = labelFor(cls, want);
|
||||
choice.is_fallback = (rank > 0);
|
||||
std::ostringstream reason;
|
||||
reason << "selected " << toString(want) << " ('" << best->name
|
||||
<< "')";
|
||||
if (!skipped.empty()) {
|
||||
reason << " after skipping";
|
||||
for (auto &s : skipped) reason << " " << s;
|
||||
} else {
|
||||
reason << " (top priority for " << toString(cls) << ")";
|
||||
}
|
||||
choice.reason = reason.str();
|
||||
return choice;
|
||||
}
|
||||
if (present_but_unhealthy)
|
||||
skipped.push_back(std::string(toString(want)) + "(unhealthy)");
|
||||
else
|
||||
skipped.push_back(std::string(toString(want)) + "(absent)");
|
||||
}
|
||||
choice.reason = "no usable transport for " + std::string(toString(cls));
|
||||
return choice;
|
||||
}
|
||||
|
||||
PathChoice PathSelector::selectForSize(const Endpoint &src, const Endpoint &dst,
|
||||
uint64_t size_bytes) const {
|
||||
PathClass cls = classify(src, dst);
|
||||
auto prio = priorityFor(cls);
|
||||
// Only kinds eligible for this class compete, but the winner is the lowest
|
||||
// modelled cost rather than the highest ladder rank.
|
||||
auto rankOf = [&](MemoryFabricKind k) -> int {
|
||||
for (size_t i = 0; i < prio.size(); ++i)
|
||||
if (prio[i] == k) return static_cast<int>(i);
|
||||
return -1;
|
||||
};
|
||||
|
||||
const bool gpu_pair = (cls == PathClass::SAME_NODE_GPU_GPU ||
|
||||
cls == PathClass::CROSS_NODE_GPU_GPU);
|
||||
|
||||
const Entry *best = nullptr;
|
||||
double best_cost = 0;
|
||||
bool best_staged = false;
|
||||
int calibrated = 0;
|
||||
for (auto &entry : entries_) {
|
||||
if (rankOf(entry.kind) < 0) continue;
|
||||
if (entry.health() != HealthStatus::HEALTHY) continue;
|
||||
if (!entry.has_cap || entry.cap.max_bandwidth_mbps == 0) continue;
|
||||
// cost in nanoseconds: fixed latency + size / bandwidth.
|
||||
double bytes_per_ns =
|
||||
static_cast<double>(entry.cap.max_bandwidth_mbps) * 1024.0 *
|
||||
1024.0 / 1e9;
|
||||
double latency = static_cast<double>(entry.cap.latency_ns);
|
||||
// A GPU<->GPU transfer over a backend that cannot address device memory
|
||||
// must stage through host: the bytes cross the host bus twice (halved
|
||||
// effective bandwidth) and pick up a copy's worth of extra latency.
|
||||
bool staged = gpu_pair && !entry.cap.supports_device_memory;
|
||||
if (staged) {
|
||||
bytes_per_ns *= 0.5;
|
||||
latency += 5000; // pinned-host copy setup, ns
|
||||
}
|
||||
double cost = latency + static_cast<double>(size_bytes) / bytes_per_ns;
|
||||
++calibrated;
|
||||
if (!best || cost < best_cost) {
|
||||
best = &entry;
|
||||
best_cost = cost;
|
||||
best_staged = staged;
|
||||
}
|
||||
}
|
||||
|
||||
// No calibrated candidate eligible for this class: fall back to the ladder.
|
||||
if (!best) return select(src, dst);
|
||||
|
||||
PathChoice choice;
|
||||
choice.found = true;
|
||||
choice.transport_name = best->name;
|
||||
choice.kind = best->kind;
|
||||
choice.label = labelFor(cls, best->kind);
|
||||
if (best_staged) choice.label += "+stage";
|
||||
choice.estimated_cost_ns = best_cost;
|
||||
choice.is_fallback = (rankOf(best->kind) > 0);
|
||||
std::ostringstream reason;
|
||||
reason << "min-cost among " << calibrated << " calibrated candidate(s) for "
|
||||
<< toString(cls) << ": " << toString(best->kind) << " ('"
|
||||
<< best->name << "')" << (best_staged ? " [host-staged]" : "")
|
||||
<< " at " << static_cast<uint64_t>(best_cost) << " ns for "
|
||||
<< size_bytes << " B";
|
||||
choice.reason = reason.str();
|
||||
return choice;
|
||||
}
|
||||
|
||||
std::string PathSelector::logChoice(const Endpoint &src, const Endpoint &dst,
|
||||
const PathChoice &choice) const {
|
||||
std::ostringstream o;
|
||||
const char *tag = choice.is_fallback ? "fallback path" : "selected path";
|
||||
if (!choice.found) tag = "NO PATH";
|
||||
o << "[Mooncake TE] " << tag << ": " << src.name << " -> " << dst.name;
|
||||
if (choice.found) o << " via " << choice.label;
|
||||
o << " (" << choice.reason << ")";
|
||||
return o.str();
|
||||
}
|
||||
|
||||
} // namespace fabric
|
||||
} // namespace mooncake
|
||||
|
|
@ -1,111 +0,0 @@
|
|||
// Copyright 2024 KVCache.AI
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
#include "fabric/plugin_loader.h"
|
||||
|
||||
#include <dirent.h>
|
||||
#include <dlfcn.h>
|
||||
|
||||
#include <algorithm>
|
||||
|
||||
namespace mooncake {
|
||||
namespace fabric {
|
||||
|
||||
namespace {
|
||||
// Fallback deleter used only on the error path (never invoked for a live
|
||||
// transport, where the plugin's own destroy function is bound instead).
|
||||
void noopDelete(Transport *) {}
|
||||
|
||||
using TransportPtr = std::unique_ptr<Transport, void (*)(Transport *)>;
|
||||
} // namespace
|
||||
|
||||
PluginLoader::~PluginLoader() {
|
||||
// Closed in reverse load order. Callers must drop their Transport objects
|
||||
// before the loader (see the header note).
|
||||
for (auto it = handles_.rbegin(); it != handles_.rend(); ++it) {
|
||||
if (*it) dlclose(*it);
|
||||
}
|
||||
}
|
||||
|
||||
TransportPtr PluginLoader::loadFile(const std::string &path, PluginInfo *info) {
|
||||
PluginInfo local;
|
||||
PluginInfo &pi = info ? *info : local;
|
||||
pi.path = path;
|
||||
pi.ok = false;
|
||||
|
||||
dlerror(); // clear any stale error
|
||||
void *handle = dlopen(path.c_str(), RTLD_NOW | RTLD_LOCAL);
|
||||
if (!handle) {
|
||||
pi.error = std::string("dlopen: ") + (dlerror() ? dlerror() : "?");
|
||||
return TransportPtr(nullptr, noopDelete);
|
||||
}
|
||||
auto abi = reinterpret_cast<plugin_abi_fn>(
|
||||
dlsym(handle, "mooncake_plugin_abi_version"));
|
||||
auto name =
|
||||
reinterpret_cast<plugin_name_fn>(dlsym(handle, "mooncake_plugin_name"));
|
||||
auto create = reinterpret_cast<plugin_create_fn>(
|
||||
dlsym(handle, "mooncake_plugin_create"));
|
||||
auto destroy = reinterpret_cast<plugin_destroy_fn>(
|
||||
dlsym(handle, "mooncake_plugin_destroy"));
|
||||
if (!abi || !name || !create || !destroy) {
|
||||
pi.error = "missing plugin ABI symbols";
|
||||
dlclose(handle);
|
||||
return TransportPtr(nullptr, noopDelete);
|
||||
}
|
||||
pi.abi = abi();
|
||||
pi.name = name();
|
||||
if (pi.abi != kPluginAbiVersion) {
|
||||
pi.error = "ABI mismatch (plugin " + std::to_string(pi.abi) +
|
||||
" vs host " + std::to_string(kPluginAbiVersion) + ")";
|
||||
dlclose(handle);
|
||||
return TransportPtr(nullptr, noopDelete);
|
||||
}
|
||||
Transport *raw = create();
|
||||
if (!raw) {
|
||||
pi.error = "plugin create() returned null";
|
||||
dlclose(handle);
|
||||
return TransportPtr(nullptr, noopDelete);
|
||||
}
|
||||
handles_.push_back(handle); // kept alive until loader destruction
|
||||
pi.ok = true;
|
||||
return TransportPtr(raw, destroy);
|
||||
}
|
||||
|
||||
std::vector<TransportPtr> PluginLoader::loadDirectory(
|
||||
const std::string &dir, std::vector<PluginInfo> *infos) {
|
||||
std::vector<TransportPtr> out;
|
||||
DIR *d = opendir(dir.c_str());
|
||||
if (!d) return out;
|
||||
std::vector<std::string> files;
|
||||
struct dirent *ent;
|
||||
while ((ent = readdir(d)) != nullptr) {
|
||||
std::string n = ent->d_name;
|
||||
if (n.size() > 3 && n.substr(n.size() - 3) == ".so" &&
|
||||
n.find("plugin") != std::string::npos) {
|
||||
files.push_back(dir + "/" + n);
|
||||
}
|
||||
}
|
||||
closedir(d);
|
||||
std::sort(files.begin(), files.end());
|
||||
for (auto &f : files) {
|
||||
PluginInfo pi;
|
||||
auto t = loadFile(f, &pi);
|
||||
if (infos) infos->push_back(pi);
|
||||
if (t) out.push_back(std::move(t));
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
} // namespace fabric
|
||||
} // namespace mooncake
|
||||
|
|
@ -14,14 +14,10 @@
|
|||
|
||||
#include "multi_transport.h"
|
||||
#include <algorithm>
|
||||
#include <cstdio>
|
||||
#include <cstdlib>
|
||||
#include <mutex>
|
||||
#include <sstream>
|
||||
#include <string>
|
||||
|
||||
#include "config.h"
|
||||
#include "fabric/path_selector.h"
|
||||
#include "transport/rdma_transport/rdma_transport.h"
|
||||
#ifdef USE_BAREX
|
||||
#include "transport/barex_transport/barex_transport.h"
|
||||
|
|
@ -54,9 +50,6 @@
|
|||
#ifdef USE_CXL
|
||||
#include "transport/cxl_transport/cxl_transport.h"
|
||||
#endif
|
||||
#ifdef USE_UALINK
|
||||
#include "transport/ualink_transport/ualink_transport.h"
|
||||
#endif
|
||||
#ifdef USE_UBSHMEM
|
||||
#include "transport/ascend_transport/ubshmem_transport/ubshmem_transport.h"
|
||||
#endif
|
||||
|
|
@ -339,11 +332,6 @@ Transport* MultiTransport::installTransport(const std::string& proto,
|
|||
transport = new CxlTransport();
|
||||
}
|
||||
#endif
|
||||
#ifdef USE_UALINK
|
||||
else if (std::string(proto) == "ualink") {
|
||||
transport = new UaLinkTransport();
|
||||
}
|
||||
#endif
|
||||
#ifdef USE_UBSHMEM
|
||||
else if (std::string(proto) == "ubshmem") {
|
||||
transport = new UBShmemTransport();
|
||||
|
|
@ -424,44 +412,6 @@ Status MultiTransport::selectTransport(const TransferRequest& entry,
|
|||
" not installed");
|
||||
}
|
||||
transport = transport_map_[proto].get();
|
||||
|
||||
// Optional, non-invasive telemetry: when MC_FABRIC_SELECTOR_DRY_RUN is set,
|
||||
// log what the capability-driven PathSelector would recommend for this
|
||||
// transfer alongside the protocol actually used. Routing is unchanged.
|
||||
//
|
||||
// The selector is built once per engine and cached as a member: each
|
||||
// transport's capability is snapshotted a single time (probe() may
|
||||
// calibrate by touching a backend's window, so it must not run per
|
||||
// transfer) and only the side-effect-free health() is consulted on the hot
|
||||
// path. Caching it on the instance (rather than a process-static) bounds
|
||||
// the cached transport pointers' lifetime to this engine.
|
||||
static const bool kFabricDryRun =
|
||||
std::getenv("MC_FABRIC_SELECTOR_DRY_RUN") != nullptr;
|
||||
if (kFabricDryRun) {
|
||||
std::call_once(dry_run_once_, [this] {
|
||||
dry_run_selector_ = std::make_unique<fabric::PathSelector>();
|
||||
for (auto& kv : transport_map_) {
|
||||
Transport* t = kv.second.get();
|
||||
fabric::Capability cap = t->probe(); // snapshot once
|
||||
std::string name = cap.name.empty() ? kv.first : cap.name;
|
||||
dry_run_selector_->registerKind(
|
||||
name, cap.kind, [t] { return t->health(); }, cap);
|
||||
}
|
||||
});
|
||||
fabric::Endpoint local, remote;
|
||||
local.kind = fabric::Endpoint::Kind::HOST;
|
||||
local.name = "local";
|
||||
remote.kind = fabric::Endpoint::Kind::HOST;
|
||||
remote.host = "remote"; // cross-endpoint => not same-host
|
||||
remote.name = target_segment_desc->name;
|
||||
auto choice =
|
||||
dry_run_selector_->selectForSize(local, remote, entry.length);
|
||||
fprintf(stderr,
|
||||
"[fabric dry-run] actual_proto=%s selector_choice=%s (%s)\n",
|
||||
proto.c_str(),
|
||||
choice.found ? choice.transport_name.c_str() : "none",
|
||||
choice.reason.c_str());
|
||||
}
|
||||
return Status::OK();
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -30,11 +30,6 @@ if (USE_CXL)
|
|||
target_sources(transport PUBLIC $<TARGET_OBJECTS:cxl_transport>)
|
||||
endif()
|
||||
|
||||
if (USE_UALINK)
|
||||
add_subdirectory(ualink_transport)
|
||||
target_sources(transport PUBLIC $<TARGET_OBJECTS:ualink_transport>)
|
||||
endif()
|
||||
|
||||
if (USE_ASCEND_DIRECT)
|
||||
add_subdirectory(ascend_transport)
|
||||
elseif(USE_ASCEND)
|
||||
|
|
|
|||
|
|
@ -27,7 +27,6 @@
|
|||
#include <regex>
|
||||
|
||||
#include "common.h"
|
||||
#include "fabric/probe_calibration.h"
|
||||
#include "transfer_engine.h"
|
||||
#include "transfer_metadata.h"
|
||||
#include "transport/transport.h"
|
||||
|
|
@ -38,10 +37,6 @@
|
|||
|
||||
namespace mooncake {
|
||||
|
||||
// Default window size for the host-memory emulation used when no /dev/dax
|
||||
// device is configured. Overridable via MC_CXL_DEV_SIZE.
|
||||
static constexpr size_t kCxlEmulatedWindowSize = 256ull * 1024 * 1024;
|
||||
|
||||
CxlTransport::CxlTransport() {
|
||||
// cxl_dev_path = "/dev/dax0.0";
|
||||
// cxl_dev_size = 1024 * 1024 * 1024;
|
||||
|
|
@ -60,7 +55,7 @@ CxlTransport::~CxlTransport() {
|
|||
cxl_dev_size != 0) {
|
||||
munmap(cxl_base_addr, cxl_dev_size);
|
||||
}
|
||||
if (metadata_) metadata_->removeSegmentDesc(local_server_name_);
|
||||
metadata_->removeSegmentDesc(local_server_name_);
|
||||
}
|
||||
|
||||
size_t CxlTransport::cxlGetDeviceSize() {
|
||||
|
|
@ -172,36 +167,8 @@ bool CxlTransport::isAddressInCxlRange(void *addr) {
|
|||
}
|
||||
|
||||
int CxlTransport::cxlDevInit() {
|
||||
// No device node configured: stand up a host-memory window so the CXL data
|
||||
// plane (shared window + coherent memcpy) is exercisable and conformance
|
||||
// testable on hosts without CXL silicon. The window is anonymous shared
|
||||
// memory; the data plane below is identical to the /dev/dax path, which is
|
||||
// exactly why this emulation is faithful -- only the mapping source
|
||||
// differs.
|
||||
if (!cxl_dev_path) {
|
||||
const char *env_size = std::getenv("MC_CXL_DEV_SIZE");
|
||||
cxl_dev_size = kCxlEmulatedWindowSize;
|
||||
if (env_size) {
|
||||
char *end = nullptr;
|
||||
unsigned long long val = strtoull(env_size, &end, 10);
|
||||
if (end != env_size && *end == '\0' && val) cxl_dev_size = val;
|
||||
}
|
||||
void *ptr = mmap(nullptr, cxl_dev_size, PROT_READ | PROT_WRITE,
|
||||
MAP_SHARED | MAP_ANONYMOUS, -1, 0);
|
||||
if (ptr == MAP_FAILED) {
|
||||
LOG(ERROR) << "CxlTransport: cannot map emulated CXL window";
|
||||
return ERR_MEMORY;
|
||||
}
|
||||
cxl_base_addr = ptr;
|
||||
cxl_emulated_ = true;
|
||||
LOG(INFO) << "CxlTransport: no MC_CXL_DEV_PATH; using emulated host "
|
||||
"window of "
|
||||
<< cxl_dev_size << " bytes";
|
||||
return 0;
|
||||
}
|
||||
|
||||
if (!cxl_dev_size) {
|
||||
LOG(ERROR) << "CxlTransport: cxl_dev_size is zero.";
|
||||
if (!cxl_dev_path || !cxl_dev_size) {
|
||||
LOG(ERROR) << "CxlTransport: cxl_dev_path or cxl_dev_size is null.";
|
||||
return -1;
|
||||
}
|
||||
int fd = open(cxl_dev_path, O_RDWR);
|
||||
|
|
@ -261,7 +228,7 @@ int CxlTransport::allocateLocalSegmentID() {
|
|||
desc->protocol = "cxl";
|
||||
#endif
|
||||
desc->cxl_base_addr = (uint64_t)cxl_base_addr;
|
||||
desc->cxl_name = cxl_dev_path ? cxl_dev_path : "cxl-emulated";
|
||||
desc->cxl_name = cxl_dev_path;
|
||||
metadata_->addLocalSegment(LOCAL_SEGMENT_ID, local_server_name_,
|
||||
std::move(desc));
|
||||
return 0;
|
||||
|
|
@ -425,45 +392,4 @@ Status CxlTransport::submitTransferTask(
|
|||
return Status::OK();
|
||||
}
|
||||
|
||||
fabric::Capability CxlTransport::probe() {
|
||||
fabric::Capability cap;
|
||||
cap.kind = fabric::MemoryFabricKind::CXL;
|
||||
cap.name = "cxl";
|
||||
cap.supports_host_memory = true;
|
||||
cap.supports_p2p = true;
|
||||
cap.supports_ordered_write = true; // coherent window + release fence
|
||||
cap.supports_zero_copy = true; // load/store directly in the window
|
||||
cap.remote_capable = false; // one host's view of a shared window
|
||||
cap.alignment = 1;
|
||||
cap.max_transfer_size = cxl_dev_size;
|
||||
cap.emulated = cxl_emulated_;
|
||||
if (cxl_emulated_)
|
||||
cap.notes = "emulated via anonymous shared host memory (no /dev/dax)";
|
||||
else
|
||||
cap.notes =
|
||||
std::string("CXL device ") + (cxl_dev_path ? cxl_dev_path : "");
|
||||
|
||||
// Calibrate honestly against the actual window: copy a source buffer into
|
||||
// the front of the window and read it back, exactly as the data plane does.
|
||||
if (cxl_base_addr && cxl_dev_size >= (2u << 20)) {
|
||||
std::vector<uint8_t> src(1u << 20, 0xCD);
|
||||
auto move = [&](uint64_t n) -> bool {
|
||||
if (n > src.size()) n = src.size();
|
||||
return cxlMemcpy(cxl_base_addr, src.data(), n) == 0;
|
||||
};
|
||||
auto cal = fabric::calibrate(move);
|
||||
if (cal.ok) {
|
||||
cap.max_bandwidth_mbps = cal.bandwidth_mbps;
|
||||
cap.latency_ns = cal.latency_ns;
|
||||
}
|
||||
}
|
||||
return cap;
|
||||
}
|
||||
|
||||
fabric::HealthStatus CxlTransport::health() {
|
||||
if (!cxl_base_addr || cxl_base_addr == MAP_FAILED || cxl_dev_size == 0)
|
||||
return fabric::HealthStatus::UNREACHABLE;
|
||||
return fabric::HealthStatus::HEALTHY;
|
||||
}
|
||||
|
||||
} // namespace mooncake
|
||||
|
|
|
|||
|
|
@ -725,32 +725,4 @@ int RdmaTransport::selectDevice(SegmentDesc *desc, uint64_t offset,
|
|||
return selectDevice(desc, offset, length, "", buffer_id, device_id,
|
||||
retry_count);
|
||||
}
|
||||
|
||||
fabric::Capability RdmaTransport::probe() {
|
||||
fabric::Capability cap;
|
||||
cap.kind = fabric::MemoryFabricKind::RDMA;
|
||||
size_t rails = context_list_.size();
|
||||
if (local_topology_ && !local_topology_->getHcaList().empty()) {
|
||||
cap.name = "rdma:" + local_topology_->getHcaList().front();
|
||||
if (rails == 0) rails = local_topology_->getHcaList().size();
|
||||
} else {
|
||||
cap.name = "rdma";
|
||||
}
|
||||
cap.supports_host_memory = true;
|
||||
cap.supports_device_memory = true; // GPUDirect-capable MRs
|
||||
cap.supports_p2p = true; // one-sided RDMA READ/WRITE
|
||||
cap.supports_ordered_write = true;
|
||||
cap.supports_atomic = true; // verbs atomics
|
||||
cap.supports_zero_copy = true; // registered MR, no host bounce
|
||||
cap.supports_multi_rail = rails > 1;
|
||||
cap.remote_capable = true;
|
||||
cap.alignment = 1;
|
||||
return cap;
|
||||
}
|
||||
|
||||
fabric::HealthStatus RdmaTransport::health() {
|
||||
if (!metadata_) return fabric::HealthStatus::UNINITIALIZED;
|
||||
if (context_list_.empty()) return fabric::HealthStatus::UNREACHABLE;
|
||||
return fabric::HealthStatus::HEALTHY;
|
||||
}
|
||||
} // namespace mooncake
|
||||
|
|
|
|||
|
|
@ -268,10 +268,9 @@ struct ClientSession : public std::enable_shared_from_this<ClientSession> {
|
|||
*socket_, asio::buffer(&header_, sizeof(SessionHeader)),
|
||||
[this, self](const asio::error_code& ec, std::size_t len) {
|
||||
if (ec || len != sizeof(SessionHeader)) {
|
||||
LOG(ERROR)
|
||||
<< "ClientSession::writeHeader failed. Error: "
|
||||
<< ec.message() << " (value: " << ec.value() << ")"
|
||||
<< ", bytes written: " << len;
|
||||
LOG(ERROR) << "ClientSession::writeHeader failed. Error: "
|
||||
<< ec.message() << " (value: " << ec.value()
|
||||
<< ")" << ", bytes written: " << len;
|
||||
if (on_finalize_) on_finalize_(TransferStatusEnum::FAILED);
|
||||
session_mutex_.unlock();
|
||||
if (on_complete_) on_complete_();
|
||||
|
|
@ -955,16 +954,4 @@ void TcpTransport::startTransfer(Slice* slice) {
|
|||
}
|
||||
}
|
||||
|
||||
fabric::Capability TcpTransport::probe() {
|
||||
fabric::Capability cap;
|
||||
cap.kind = fabric::MemoryFabricKind::TCP;
|
||||
cap.name = "tcp";
|
||||
cap.supports_host_memory = true;
|
||||
cap.supports_ordered_write = true; // a single connection delivers in order
|
||||
cap.remote_capable = true; // crosses process / host boundaries
|
||||
cap.supports_zero_copy = false; // staged through socket buffers
|
||||
cap.alignment = 1;
|
||||
return cap;
|
||||
}
|
||||
|
||||
} // namespace mooncake
|
||||
|
|
|
|||
|
|
@ -1,4 +0,0 @@
|
|||
file(GLOB UALINK_SOURCES "*.cpp")
|
||||
|
||||
add_library(ualink_transport OBJECT ${UALINK_SOURCES})
|
||||
target_link_libraries(ualink_transport PRIVATE glog::glog)
|
||||
|
|
@ -1,55 +0,0 @@
|
|||
// Copyright 2024 KVCache.AI
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
#include "transport/ualink_transport/ualink_simulator.h"
|
||||
|
||||
#include <atomic>
|
||||
#include <cstring>
|
||||
|
||||
namespace mooncake {
|
||||
|
||||
UALinkSimulator::UALinkSimulator(int num_ports) {
|
||||
if (num_ports < 1) num_ports = 1;
|
||||
ports_.reserve(num_ports);
|
||||
for (int i = 0; i < num_ports; ++i) {
|
||||
UALinkPort port;
|
||||
port.id = i;
|
||||
port.numa_hint = i % 2; // alternate, modelling two accelerator islands
|
||||
ports_.push_back(port);
|
||||
}
|
||||
}
|
||||
|
||||
std::vector<std::pair<int, int>> UALinkSimulator::topology() const {
|
||||
// Switch-connected full mesh: every port reachable from every other.
|
||||
std::vector<std::pair<int, int>> edges;
|
||||
int n = numPorts();
|
||||
for (int a = 0; a < n; ++a)
|
||||
for (int b = a + 1; b < n; ++b) edges.emplace_back(a, b);
|
||||
return edges;
|
||||
}
|
||||
|
||||
uint64_t UALinkSimulator::orderedWrite(void *dst, const void *src, uint64_t n) {
|
||||
std::memcpy(dst, src, n);
|
||||
// Publish: everything written above is visible before the sequence
|
||||
// advances.
|
||||
std::atomic_thread_fence(std::memory_order_release);
|
||||
return seq_.fetch_add(1, std::memory_order_acq_rel) + 1;
|
||||
}
|
||||
|
||||
void UALinkSimulator::orderedRead(void *dst, const void *src, uint64_t n) {
|
||||
std::atomic_thread_fence(std::memory_order_acquire);
|
||||
std::memcpy(dst, src, n);
|
||||
}
|
||||
|
||||
} // namespace mooncake
|
||||
|
|
@ -1,305 +0,0 @@
|
|||
// Copyright 2024 KVCache.AI
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
#include "transport/ualink_transport/ualink_transport.h"
|
||||
|
||||
#include <glog/logging.h>
|
||||
#include <sys/mman.h>
|
||||
|
||||
#include <cassert>
|
||||
#include <cstdint>
|
||||
#include <cstring>
|
||||
|
||||
#include "common.h"
|
||||
#include "error.h"
|
||||
#include "fabric/probe_calibration.h"
|
||||
#include "transfer_metadata.h"
|
||||
#include "transport/transport.h"
|
||||
|
||||
namespace mooncake {
|
||||
|
||||
// Size of the host-memory window standing in for the UALink address space.
|
||||
// Overridable via MC_UALINK_WINDOW_SIZE.
|
||||
static constexpr size_t kUaLinkWindowSize = 256ull * 1024 * 1024;
|
||||
|
||||
UaLinkTransport::UaLinkTransport() : sim_(4) {}
|
||||
|
||||
UaLinkTransport::~UaLinkTransport() {
|
||||
if (window_base_ && window_base_ != MAP_FAILED && window_size_)
|
||||
munmap(window_base_, window_size_);
|
||||
if (metadata_) metadata_->removeSegmentDesc(local_server_name_);
|
||||
}
|
||||
|
||||
bool UaLinkTransport::windowInit() {
|
||||
window_size_ = kUaLinkWindowSize;
|
||||
const char *env_size = std::getenv("MC_UALINK_WINDOW_SIZE");
|
||||
if (env_size) {
|
||||
char *end = nullptr;
|
||||
unsigned long long val = strtoull(env_size, &end, 10);
|
||||
if (end != env_size && *end == '\0' && val) window_size_ = val;
|
||||
}
|
||||
void *ptr = mmap(nullptr, window_size_, PROT_READ | PROT_WRITE,
|
||||
MAP_SHARED | MAP_ANONYMOUS, -1, 0);
|
||||
if (ptr == MAP_FAILED) {
|
||||
window_base_ = nullptr;
|
||||
return false;
|
||||
}
|
||||
window_base_ = ptr;
|
||||
return true;
|
||||
}
|
||||
|
||||
int UaLinkTransport::install(std::string &local_server_name,
|
||||
std::shared_ptr<TransferMetadata> meta,
|
||||
std::shared_ptr<Topology> topo) {
|
||||
(void)topo;
|
||||
metadata_ = meta;
|
||||
local_server_name_ = local_server_name;
|
||||
|
||||
if (!windowInit()) {
|
||||
LOG(ERROR) << "UaLinkTransport: cannot map UALink window";
|
||||
return -1;
|
||||
}
|
||||
|
||||
if (allocateLocalSegmentID()) {
|
||||
LOG(ERROR) << "UaLinkTransport: cannot allocate local segment";
|
||||
return -1;
|
||||
}
|
||||
|
||||
if (metadata_->updateLocalSegmentDesc()) {
|
||||
LOG(ERROR) << "UaLinkTransport: cannot publish segments, check the "
|
||||
"availability of metadata storage";
|
||||
return -1;
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
int UaLinkTransport::allocateLocalSegmentID() {
|
||||
auto desc = metadata_->getSegmentDesc(local_server_name_);
|
||||
if (!desc) desc = std::make_shared<SegmentDesc>();
|
||||
desc->name = local_server_name_;
|
||||
#ifdef ENABLE_MULTI_PROTOCOL
|
||||
if (!desc->protocol.empty()) desc->protocol += ",";
|
||||
desc->protocol += "ualink";
|
||||
#else
|
||||
desc->protocol = "ualink";
|
||||
#endif
|
||||
desc->cxl_base_addr = (uint64_t)window_base_;
|
||||
desc->cxl_name = "ualink-emulated";
|
||||
metadata_->addLocalSegment(LOCAL_SEGMENT_ID, local_server_name_,
|
||||
std::move(desc));
|
||||
return 0;
|
||||
}
|
||||
|
||||
int UaLinkTransport::registerLocalMemory(void *addr, size_t length,
|
||||
const std::string &location,
|
||||
bool remote_accessible,
|
||||
bool update_metadata) {
|
||||
(void)location;
|
||||
(void)remote_accessible;
|
||||
BufferDesc buffer_desc;
|
||||
buffer_desc.name = local_server_name_;
|
||||
|
||||
uintptr_t base = reinterpret_cast<uintptr_t>(window_base_);
|
||||
uintptr_t end = base + window_size_;
|
||||
uintptr_t ptr = reinterpret_cast<uintptr_t>(addr);
|
||||
if (ptr < base || ptr >= end) {
|
||||
errno = EFAULT;
|
||||
return -1;
|
||||
}
|
||||
if (ptr + length > end || ptr + length < ptr) {
|
||||
errno = EOVERFLOW;
|
||||
return -1;
|
||||
}
|
||||
buffer_desc.offset = (uint64_t)addr - (uint64_t)window_base_;
|
||||
buffer_desc.length = length;
|
||||
#ifdef ENABLE_MULTI_PROTOCOL
|
||||
buffer_desc.protocol = "ualink";
|
||||
#endif
|
||||
return metadata_->addLocalMemoryBuffer(buffer_desc, update_metadata);
|
||||
}
|
||||
|
||||
int UaLinkTransport::unregisterLocalMemory(void *addr, bool update_metadata) {
|
||||
return metadata_->removeLocalMemoryBuffer(addr, update_metadata);
|
||||
}
|
||||
|
||||
int UaLinkTransport::registerLocalMemoryBatch(
|
||||
const std::vector<Transport::BufferEntry> &buffer_list,
|
||||
const std::string &location) {
|
||||
for (auto &buffer : buffer_list)
|
||||
registerLocalMemory(buffer.addr, buffer.length, location, true, false);
|
||||
return metadata_->updateLocalSegmentDesc();
|
||||
}
|
||||
|
||||
int UaLinkTransport::unregisterLocalMemoryBatch(
|
||||
const std::vector<void *> &addr_list) {
|
||||
for (auto &addr : addr_list) unregisterLocalMemory(addr, false);
|
||||
return metadata_->updateLocalSegmentDesc();
|
||||
}
|
||||
|
||||
int UaLinkTransport::moveChunk(void *dst, void *src, size_t size,
|
||||
TransferRequest::OpCode op) {
|
||||
if (!src || !dst) return -1;
|
||||
uintptr_t base = reinterpret_cast<uintptr_t>(window_base_);
|
||||
uintptr_t end = base + window_size_;
|
||||
// Whichever side lands inside the window must stay within bounds.
|
||||
for (void *p : {dst, src}) {
|
||||
uintptr_t v = reinterpret_cast<uintptr_t>(p);
|
||||
if (v >= base && v < end && v + size > end) return -1;
|
||||
}
|
||||
if (op == TransferRequest::READ)
|
||||
sim_.orderedRead(dst, src, size);
|
||||
else
|
||||
sim_.orderedWrite(dst, src, size);
|
||||
return 0;
|
||||
}
|
||||
|
||||
Status UaLinkTransport::getTransferStatus(BatchID batch_id, size_t task_id,
|
||||
TransferStatus &status) {
|
||||
auto &batch_desc = *((BatchDesc *)(batch_id));
|
||||
const size_t task_count = batch_desc.task_list.size();
|
||||
if (task_id >= task_count) {
|
||||
return Status::InvalidArgument(
|
||||
"UaLinkTransport::getTransferStatus invalid argument, batch id: " +
|
||||
std::to_string(batch_id));
|
||||
}
|
||||
auto &task = batch_desc.task_list[task_id];
|
||||
status.transferred_bytes = task.transferred_bytes;
|
||||
uint64_t success_slice_count = task.success_slice_count;
|
||||
uint64_t failed_slice_count = task.failed_slice_count;
|
||||
if (success_slice_count + failed_slice_count == task.slice_count) {
|
||||
status.s = failed_slice_count ? TransferStatusEnum::FAILED
|
||||
: TransferStatusEnum::COMPLETED;
|
||||
task.is_finished = true;
|
||||
} else {
|
||||
status.s = TransferStatusEnum::WAITING;
|
||||
}
|
||||
return Status::OK();
|
||||
}
|
||||
|
||||
Status UaLinkTransport::submitTransfer(
|
||||
BatchID batch_id, const std::vector<TransferRequest> &entries) {
|
||||
auto &batch_desc = *((BatchDesc *)(batch_id));
|
||||
if (batch_desc.task_list.size() + entries.size() > batch_desc.batch_size) {
|
||||
LOG(ERROR) << "UaLinkTransport: Exceed the limitation of current "
|
||||
"batch's capacity";
|
||||
return Status::InvalidArgument(
|
||||
"UaLinkTransport: Exceed the limitation of capacity, batch id: " +
|
||||
std::to_string(batch_id));
|
||||
}
|
||||
|
||||
size_t task_id = batch_desc.task_list.size();
|
||||
batch_desc.task_list.resize(task_id + entries.size());
|
||||
|
||||
for (auto &request : entries) {
|
||||
TransferTask &task = batch_desc.task_list[task_id];
|
||||
++task_id;
|
||||
task.total_bytes = request.length;
|
||||
Slice *slice = getSliceCache().allocate();
|
||||
slice->source_addr = (char *)request.source;
|
||||
slice->cxl.dest_addr = (char *)window_base_ + request.target_offset;
|
||||
slice->length = request.length;
|
||||
slice->opcode = request.opcode;
|
||||
slice->task = &task;
|
||||
slice->target_id = request.target_id;
|
||||
slice->status = Slice::PENDING;
|
||||
__sync_fetch_and_add(&task.slice_count, 1);
|
||||
int err;
|
||||
if (slice->opcode == TransferRequest::READ)
|
||||
err = moveChunk(slice->source_addr, (void *)slice->cxl.dest_addr,
|
||||
slice->length, slice->opcode);
|
||||
else
|
||||
err = moveChunk((void *)slice->cxl.dest_addr, slice->source_addr,
|
||||
slice->length, slice->opcode);
|
||||
if (err != 0)
|
||||
slice->markFailed();
|
||||
else
|
||||
slice->markSuccess();
|
||||
}
|
||||
return Status::OK();
|
||||
}
|
||||
|
||||
Status UaLinkTransport::submitTransferTask(
|
||||
const std::vector<TransferTask *> &task_list) {
|
||||
for (size_t index = 0; index < task_list.size(); ++index) {
|
||||
assert(task_list[index]);
|
||||
auto &task = *task_list[index];
|
||||
assert(task.request);
|
||||
auto &request = *task.request;
|
||||
task.total_bytes = request.length;
|
||||
|
||||
Slice *slice = getSliceCache().allocate();
|
||||
slice->source_addr = (char *)request.source;
|
||||
slice->cxl.dest_addr = (char *)window_base_ + request.target_offset;
|
||||
slice->length = request.length;
|
||||
slice->opcode = request.opcode;
|
||||
slice->task = &task;
|
||||
slice->target_id = request.target_id;
|
||||
slice->status = Slice::PENDING;
|
||||
task.slice_list.push_back(slice);
|
||||
__sync_fetch_and_add(&task.slice_count, 1);
|
||||
int err;
|
||||
if (slice->opcode == TransferRequest::READ)
|
||||
err = moveChunk(slice->source_addr, (void *)slice->cxl.dest_addr,
|
||||
slice->length, slice->opcode);
|
||||
else
|
||||
err = moveChunk((void *)slice->cxl.dest_addr, slice->source_addr,
|
||||
slice->length, slice->opcode);
|
||||
if (err != 0)
|
||||
slice->markFailed();
|
||||
else
|
||||
slice->markSuccess();
|
||||
}
|
||||
return Status::OK();
|
||||
}
|
||||
|
||||
fabric::Capability UaLinkTransport::probe() {
|
||||
fabric::Capability cap;
|
||||
cap.kind = fabric::MemoryFabricKind::UALINK;
|
||||
cap.name = "ualink";
|
||||
cap.supports_host_memory = true;
|
||||
cap.supports_device_memory = false;
|
||||
cap.supports_p2p = true;
|
||||
cap.supports_ordered_write = true; // enforced by the simulator
|
||||
cap.supports_zero_copy = true;
|
||||
cap.remote_capable = false;
|
||||
cap.alignment = 1;
|
||||
cap.max_transfer_size = window_size_;
|
||||
cap.emulated = true;
|
||||
cap.notes =
|
||||
"green-field UALink; ordered-delivery semantics enforced by "
|
||||
"software simulator over host memory (no silicon)";
|
||||
|
||||
if (window_base_ && window_size_ >= (2u << 20)) {
|
||||
std::vector<uint8_t> src(1u << 20, 0xAB);
|
||||
auto move = [&](uint64_t n) -> bool {
|
||||
if (n > src.size()) n = src.size();
|
||||
return moveChunk(window_base_, src.data(), n,
|
||||
TransferRequest::WRITE) == 0;
|
||||
};
|
||||
auto cal = fabric::calibrate(move);
|
||||
if (cal.ok) {
|
||||
cap.max_bandwidth_mbps = cal.bandwidth_mbps;
|
||||
cap.latency_ns = cal.latency_ns;
|
||||
}
|
||||
}
|
||||
return cap;
|
||||
}
|
||||
|
||||
fabric::HealthStatus UaLinkTransport::health() {
|
||||
if (!window_base_ || window_base_ == MAP_FAILED || window_size_ == 0)
|
||||
return fabric::HealthStatus::UNREACHABLE;
|
||||
return fabric::HealthStatus::HEALTHY;
|
||||
}
|
||||
|
||||
} // namespace mooncake
|
||||
|
|
@ -99,46 +99,6 @@ add_executable(common_test ${WORKSPACE}/common_test.cpp)
|
|||
target_link_libraries(common_test PUBLIC transfer_engine gtest gtest_main)
|
||||
add_test(NAME common_test COMMAND common_test)
|
||||
|
||||
# Memory-fabric cross-backend conformance: one suite, every backend.
|
||||
add_executable(conformance_runner
|
||||
${WORKSPACE}/fabric_conformance/conformance_runner.cpp
|
||||
${WORKSPACE}/fabric_conformance/conformance_suite.cpp
|
||||
${WORKSPACE}/fabric_conformance/conformance_cases.cpp)
|
||||
target_include_directories(conformance_runner
|
||||
PRIVATE ${WORKSPACE}/fabric_conformance)
|
||||
target_link_libraries(conformance_runner
|
||||
PUBLIC transfer_engine gflags::gflags glog::glog pthread)
|
||||
add_test(NAME conformance_runner COMMAND conformance_runner)
|
||||
|
||||
add_executable(fabric_conformance_test
|
||||
${WORKSPACE}/fabric_conformance/conformance_gtest.cpp
|
||||
${WORKSPACE}/fabric_conformance/conformance_suite.cpp
|
||||
${WORKSPACE}/fabric_conformance/conformance_cases.cpp)
|
||||
target_include_directories(fabric_conformance_test
|
||||
PRIVATE ${WORKSPACE}/fabric_conformance)
|
||||
target_link_libraries(fabric_conformance_test
|
||||
PUBLIC transfer_engine gtest gtest_main glog::glog pthread)
|
||||
add_test(NAME fabric_conformance_test COMMAND fabric_conformance_test)
|
||||
|
||||
add_executable(fabric_cli
|
||||
${WORKSPACE}/fabric_conformance/fabric_cli.cpp
|
||||
${WORKSPACE}/fabric_conformance/conformance_suite.cpp
|
||||
${WORKSPACE}/fabric_conformance/conformance_cases.cpp)
|
||||
target_include_directories(fabric_cli
|
||||
PRIVATE ${WORKSPACE}/fabric_conformance)
|
||||
target_link_libraries(fabric_cli
|
||||
PUBLIC transfer_engine gflags::gflags glog::glog pthread)
|
||||
|
||||
# Out-of-tree transport plugin: built as a standalone .so that is NOT linked
|
||||
# into the engine and is dlopened by PluginLoader at runtime
|
||||
# (fabric_cli plugins <dir>).
|
||||
add_library(shm_plugin MODULE
|
||||
${WORKSPACE}/fabric_conformance/sample_plugin/shm_plugin.cpp)
|
||||
target_link_libraries(shm_plugin PRIVATE transfer_engine)
|
||||
set_target_properties(shm_plugin PROPERTIES
|
||||
LIBRARY_OUTPUT_DIRECTORY "${CMAKE_BINARY_DIR}/plugins"
|
||||
PREFIX "")
|
||||
|
||||
if (USE_ASCEND_DIRECT)
|
||||
# AscendDirectTransport unit test with mock implementations
|
||||
# Mock implementations are included in the test file via anonymous namespace
|
||||
|
|
|
|||
|
|
@ -1,291 +0,0 @@
|
|||
// Copyright 2024 KVCache.AI
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
#include <atomic>
|
||||
#include <chrono>
|
||||
#include <cstdlib>
|
||||
#include <cstring>
|
||||
#include <thread>
|
||||
#include <vector>
|
||||
|
||||
#include "conformance_suite.h"
|
||||
|
||||
namespace mooncake {
|
||||
namespace conformance {
|
||||
|
||||
namespace {
|
||||
uint64_t alignUp(uint64_t v, uint64_t a) {
|
||||
return a <= 1 ? v : ((v + a - 1) / a) * a;
|
||||
}
|
||||
} // namespace
|
||||
|
||||
Outcome caseCapability(Harness &h, const fabric::Capability &cap,
|
||||
std::string &detail) {
|
||||
std::string why = cap.checkConsistency();
|
||||
if (!why.empty()) {
|
||||
detail = "inconsistent capability: " + why;
|
||||
return Outcome::FAIL;
|
||||
}
|
||||
// probe() must be stable in kind across calls.
|
||||
auto cap2 = h.transport()->probe();
|
||||
if (cap2.kind != cap.kind) {
|
||||
detail = "probe() kind not stable across calls";
|
||||
return Outcome::FAIL;
|
||||
}
|
||||
detail = cap.toLine();
|
||||
return Outcome::PASS;
|
||||
}
|
||||
|
||||
Outcome caseRegister(Harness &h, const fabric::Capability &,
|
||||
std::string &detail) {
|
||||
// A zero-length transfer must be a clean no-op, and a bounded transfer must
|
||||
// round-trip. (Memory registration itself is validated by the engine; here
|
||||
// we exercise the boundary behaviour the data path must honour.)
|
||||
uint8_t probe = 0;
|
||||
if (!h.write(&probe, 0, 0)) {
|
||||
detail = "zero-length transfer not handled as no-op";
|
||||
return Outcome::FAIL;
|
||||
}
|
||||
detail = "zero-length no-op handled";
|
||||
return Outcome::PASS;
|
||||
}
|
||||
|
||||
Outcome caseRoundtrip(Harness &h, const fabric::Capability &cap,
|
||||
std::string &detail) {
|
||||
const uint64_t window = h.windowBytes();
|
||||
const uint64_t cap_size = 4u * 1024 * 1024;
|
||||
void *src = std::malloc(cap_size);
|
||||
void *dst = std::malloc(cap_size);
|
||||
if (!src || !dst) {
|
||||
std::free(src);
|
||||
std::free(dst);
|
||||
detail = "malloc";
|
||||
return Outcome::FAIL;
|
||||
}
|
||||
h.registerLocal(src, cap_size);
|
||||
h.registerLocal(dst, cap_size);
|
||||
|
||||
const uint64_t sizes[] = {1, 7, 63, 64,
|
||||
4096, 65536, 1u << 20, (1u << 22) - 13};
|
||||
Outcome out = Outcome::PASS;
|
||||
for (uint64_t s : sizes) {
|
||||
if (s > cap_size || s + 4096 > window) continue;
|
||||
uint64_t off = alignUp(128, cap.alignment ? cap.alignment : 1);
|
||||
if (off + s > window) off = 0;
|
||||
fillPattern(src, s, s * 2654435761u + 7);
|
||||
std::memset(dst, 0, s);
|
||||
if (!h.write(src, off, s)) {
|
||||
detail = "write failed @size " + std::to_string(s);
|
||||
out = Outcome::FAIL;
|
||||
break;
|
||||
}
|
||||
if (!h.read(dst, off, s)) {
|
||||
detail = "read failed @size " + std::to_string(s);
|
||||
out = Outcome::FAIL;
|
||||
break;
|
||||
}
|
||||
if (std::memcmp(src, dst, s) != 0) {
|
||||
detail = "mismatch @size " + std::to_string(s);
|
||||
out = Outcome::FAIL;
|
||||
break;
|
||||
}
|
||||
}
|
||||
if (out == Outcome::PASS) detail = "byte-exact across sizes up to 4MiB";
|
||||
h.unregisterLocal(src);
|
||||
h.unregisterLocal(dst);
|
||||
std::free(src);
|
||||
std::free(dst);
|
||||
return out;
|
||||
}
|
||||
|
||||
Outcome caseBatch(Harness &h, const fabric::Capability &, std::string &detail) {
|
||||
const int K = 16;
|
||||
const uint64_t BLK = 64 * 1024;
|
||||
const uint64_t W = K * BLK;
|
||||
if (W > h.windowBytes()) {
|
||||
detail = "window too small for batch";
|
||||
return Outcome::SKIP;
|
||||
}
|
||||
void *src = std::malloc(W);
|
||||
void *dst = std::malloc(W);
|
||||
if (!src || !dst) {
|
||||
std::free(src);
|
||||
std::free(dst);
|
||||
detail = "malloc";
|
||||
return Outcome::FAIL;
|
||||
}
|
||||
h.registerLocal(src, W);
|
||||
h.registerLocal(dst, W);
|
||||
for (int i = 0; i < K; ++i)
|
||||
fillPattern(static_cast<char *>(src) + i * BLK, BLK, 100 + i);
|
||||
std::memset(dst, 0, W);
|
||||
|
||||
std::vector<Harness::Req> wreqs;
|
||||
for (int i = 0; i < K; ++i)
|
||||
wreqs.push_back({true, static_cast<char *>(src) + i * BLK,
|
||||
static_cast<uint64_t>(i) * BLK, BLK});
|
||||
bool wok = h.batch(wreqs);
|
||||
|
||||
std::vector<Harness::Req> rreqs;
|
||||
for (int i = 0; i < K; ++i)
|
||||
rreqs.push_back({false, static_cast<char *>(dst) + i * BLK,
|
||||
static_cast<uint64_t>(i) * BLK, BLK});
|
||||
bool rok = h.batch(rreqs);
|
||||
|
||||
Outcome out;
|
||||
if (!wok || !rok) {
|
||||
detail = "batch submit failed";
|
||||
out = Outcome::FAIL;
|
||||
} else if (std::memcmp(src, dst, W) != 0) {
|
||||
detail = "batch data mismatch";
|
||||
out = Outcome::FAIL;
|
||||
} else {
|
||||
detail = std::to_string(K) + " requests/batch, byte-exact";
|
||||
out = Outcome::PASS;
|
||||
}
|
||||
h.unregisterLocal(src);
|
||||
h.unregisterLocal(dst);
|
||||
std::free(src);
|
||||
std::free(dst);
|
||||
return out;
|
||||
}
|
||||
|
||||
Outcome caseOrdering(Harness &h, const fabric::Capability &cap,
|
||||
std::string &detail) {
|
||||
if (!cap.supports_ordered_write) {
|
||||
detail = "backend does not advertise ordered_write";
|
||||
return Outcome::SKIP;
|
||||
}
|
||||
const int N = 32;
|
||||
const uint64_t BLK = 4096;
|
||||
const uint64_t FLAG_OFF = N * BLK;
|
||||
const uint64_t W = FLAG_OFF + 64;
|
||||
if (W > h.windowBytes()) {
|
||||
detail = "window too small for ordering";
|
||||
return Outcome::SKIP;
|
||||
}
|
||||
|
||||
auto *prod_buf = static_cast<uint8_t *>(std::malloc(BLK));
|
||||
auto *flag_buf = static_cast<uint8_t *>(std::malloc(64));
|
||||
auto *cons_blk = static_cast<uint8_t *>(std::malloc(BLK));
|
||||
auto *cons_flag = static_cast<uint8_t *>(std::malloc(64));
|
||||
h.registerLocal(prod_buf, BLK);
|
||||
h.registerLocal(flag_buf, 64);
|
||||
h.registerLocal(cons_blk, BLK);
|
||||
h.registerLocal(cons_flag, 64);
|
||||
|
||||
std::atomic<int> mismatch{-1};
|
||||
std::thread producer([&] {
|
||||
for (int i = 0; i < N; ++i) {
|
||||
std::memset(prod_buf, (uint8_t)(i + 1), BLK);
|
||||
h.write(prod_buf, (uint64_t)i * BLK, BLK);
|
||||
}
|
||||
std::memset(flag_buf, 0, 64);
|
||||
flag_buf[0] = 0xAC; // publish "all blocks written" cookie last
|
||||
h.write(flag_buf, FLAG_OFF, 64);
|
||||
});
|
||||
std::thread consumer([&] {
|
||||
auto deadline =
|
||||
std::chrono::steady_clock::now() + std::chrono::seconds(10);
|
||||
for (;;) {
|
||||
std::memset(cons_flag, 0, 64);
|
||||
h.read(cons_flag, FLAG_OFF, 64);
|
||||
if (cons_flag[0] == 0xAC) break;
|
||||
if (std::chrono::steady_clock::now() > deadline) {
|
||||
mismatch.store(-2);
|
||||
return;
|
||||
}
|
||||
std::this_thread::yield();
|
||||
}
|
||||
for (int i = 0; i < N; ++i) {
|
||||
h.read(cons_blk, (uint64_t)i * BLK, BLK);
|
||||
for (uint64_t b = 0; b < BLK; ++b)
|
||||
if (cons_blk[b] != (uint8_t)(i + 1)) {
|
||||
mismatch.store(i);
|
||||
return;
|
||||
}
|
||||
}
|
||||
});
|
||||
producer.join();
|
||||
consumer.join();
|
||||
|
||||
Outcome out;
|
||||
if (mismatch.load() == -1) {
|
||||
out = Outcome::PASS;
|
||||
detail = "flag-after-data ordering held for 32 blocks";
|
||||
} else if (mismatch.load() == -2) {
|
||||
out = Outcome::FAIL;
|
||||
detail = "consumer timed out waiting for flag";
|
||||
} else {
|
||||
out = Outcome::FAIL;
|
||||
detail = "stale data at block " + std::to_string(mismatch.load());
|
||||
}
|
||||
h.unregisterLocal(prod_buf);
|
||||
h.unregisterLocal(flag_buf);
|
||||
h.unregisterLocal(cons_blk);
|
||||
h.unregisterLocal(cons_flag);
|
||||
std::free(prod_buf);
|
||||
std::free(flag_buf);
|
||||
std::free(cons_blk);
|
||||
std::free(cons_flag);
|
||||
return out;
|
||||
}
|
||||
|
||||
Outcome caseConcurrency(Harness &h, const fabric::Capability &,
|
||||
std::string &detail) {
|
||||
const int threads = 8;
|
||||
const int iters = 40;
|
||||
const uint64_t BLK = 64 * 1024;
|
||||
const uint64_t need = threads * BLK;
|
||||
if (need > h.windowBytes()) {
|
||||
detail = "window too small for concurrency";
|
||||
return Outcome::SKIP;
|
||||
}
|
||||
std::atomic<int> failures{0};
|
||||
std::vector<std::thread> pool;
|
||||
for (int t = 0; t < threads; ++t) {
|
||||
pool.emplace_back([&, t] {
|
||||
auto *src = static_cast<uint8_t *>(std::malloc(BLK));
|
||||
auto *dst = static_cast<uint8_t *>(std::malloc(BLK));
|
||||
h.registerLocal(src, BLK);
|
||||
h.registerLocal(dst, BLK);
|
||||
uint64_t off = (uint64_t)t * BLK; // disjoint region per thread
|
||||
for (int i = 0; i < iters; ++i) {
|
||||
fillPattern(src, BLK, (t * 1000 + i) * 2654435761u);
|
||||
std::memset(dst, 0, BLK);
|
||||
if (!h.write(src, off, BLK) || !h.read(dst, off, BLK) ||
|
||||
std::memcmp(src, dst, BLK) != 0) {
|
||||
failures.fetch_add(1);
|
||||
break;
|
||||
}
|
||||
}
|
||||
h.unregisterLocal(src);
|
||||
h.unregisterLocal(dst);
|
||||
std::free(src);
|
||||
std::free(dst);
|
||||
});
|
||||
}
|
||||
for (auto &th : pool) th.join();
|
||||
|
||||
if (failures.load() == 0) {
|
||||
detail = std::to_string(threads) + " threads x " +
|
||||
std::to_string(iters) + " iters, disjoint, byte-exact";
|
||||
return Outcome::PASS;
|
||||
}
|
||||
detail = std::to_string(failures.load()) + " thread(s) saw a mismatch";
|
||||
return Outcome::FAIL;
|
||||
}
|
||||
|
||||
} // namespace conformance
|
||||
} // namespace mooncake
|
||||
|
|
@ -1,81 +0,0 @@
|
|||
// Copyright 2024 KVCache.AI
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
// GoogleTest gate for the memory-fabric conformance suite. A backend that is
|
||||
// unreachable on the host (no hardware / no peer) is skipped, not failed -- so
|
||||
// the gate stays green on any host while still asserting that every backend
|
||||
// that *can* run honours the contract.
|
||||
|
||||
#include <glog/logging.h>
|
||||
#include <gtest/gtest.h>
|
||||
|
||||
#include <string>
|
||||
#include <vector>
|
||||
|
||||
#include "conformance_suite.h"
|
||||
|
||||
using namespace mooncake;
|
||||
using namespace mooncake::conformance;
|
||||
|
||||
namespace {
|
||||
|
||||
class FabricConformance : public ::testing::TestWithParam<std::string> {};
|
||||
|
||||
TEST_P(FabricConformance, BackendIsConformant) {
|
||||
const std::string proto = GetParam();
|
||||
BackendReport report = runConformance(proto);
|
||||
if (!report.loadable) {
|
||||
GTEST_SKIP() << proto
|
||||
<< " unreachable on this host: " << report.load_error;
|
||||
}
|
||||
for (const auto &c : report.cases) {
|
||||
if (c.required) {
|
||||
EXPECT_EQ(c.outcome, Outcome::PASS)
|
||||
<< proto << " required case '" << c.name
|
||||
<< "' did not pass: " << c.detail;
|
||||
} else {
|
||||
EXPECT_NE(c.outcome, Outcome::FAIL)
|
||||
<< proto << " optional case '" << c.name
|
||||
<< "' failed: " << c.detail;
|
||||
}
|
||||
}
|
||||
EXPECT_TRUE(report.conformant) << proto << " is non-conformant";
|
||||
}
|
||||
|
||||
std::vector<std::string> backends() {
|
||||
std::vector<std::string> v{"tcp"};
|
||||
#ifdef USE_CXL
|
||||
v.push_back("cxl");
|
||||
#endif
|
||||
#ifdef USE_UALINK
|
||||
v.push_back("ualink");
|
||||
#endif
|
||||
v.push_back("rdma");
|
||||
return v;
|
||||
}
|
||||
|
||||
INSTANTIATE_TEST_SUITE_P(AllBackends, FabricConformance,
|
||||
::testing::ValuesIn(backends()),
|
||||
[](const ::testing::TestParamInfo<std::string> &info) {
|
||||
return info.param;
|
||||
});
|
||||
|
||||
} // namespace
|
||||
|
||||
int main(int argc, char **argv) {
|
||||
::testing::InitGoogleTest(&argc, argv);
|
||||
google::InitGoogleLogging(argv[0]);
|
||||
FLAGS_minloglevel = 2;
|
||||
return RUN_ALL_TESTS();
|
||||
}
|
||||
|
|
@ -1,168 +0,0 @@
|
|||
// Copyright 2024 KVCache.AI
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
// Run the one conformance suite against every transport this build knows about
|
||||
// and print a backend x case matrix plus a machine-readable JSON report. This
|
||||
// is the single command that answers "is the memory fabric abstraction
|
||||
// implemented uniformly across all backends?".
|
||||
|
||||
#include <gflags/gflags.h>
|
||||
#include <glog/logging.h>
|
||||
|
||||
#include <cstdio>
|
||||
#include <fstream>
|
||||
#include <iostream>
|
||||
#include <string>
|
||||
#include <vector>
|
||||
|
||||
#include "conformance_suite.h"
|
||||
|
||||
using namespace mooncake;
|
||||
using namespace mooncake::conformance;
|
||||
|
||||
DEFINE_string(json, "", "write the JSON report to this path");
|
||||
DEFINE_string(backends, "", "comma-separated transports to run (default: all)");
|
||||
|
||||
namespace {
|
||||
const char *mark(Outcome o) {
|
||||
switch (o) {
|
||||
case Outcome::PASS:
|
||||
return "PASS";
|
||||
case Outcome::SKIP:
|
||||
return "skip";
|
||||
case Outcome::FAIL:
|
||||
return "FAIL";
|
||||
}
|
||||
return "?";
|
||||
}
|
||||
|
||||
const char *kCols[] = {"capab", "regis", "round", "batch", "order", "concu"};
|
||||
|
||||
std::vector<std::string> defaultBackends() {
|
||||
// Transports whose data plane is exercisable in a single process. RDMA is
|
||||
// included because the verbs path can self-loopback through a local HCA.
|
||||
std::vector<std::string> v;
|
||||
v.push_back("tcp");
|
||||
#ifdef USE_CXL
|
||||
v.push_back("cxl");
|
||||
#endif
|
||||
#ifdef USE_UALINK
|
||||
v.push_back("ualink");
|
||||
#endif
|
||||
v.push_back("rdma");
|
||||
return v;
|
||||
}
|
||||
|
||||
std::vector<std::string> split(const std::string &s) {
|
||||
std::vector<std::string> out;
|
||||
size_t start = 0;
|
||||
while (start <= s.size()) {
|
||||
size_t comma = s.find(',', start);
|
||||
if (comma == std::string::npos) {
|
||||
out.push_back(s.substr(start));
|
||||
break;
|
||||
}
|
||||
out.push_back(s.substr(start, comma - start));
|
||||
start = comma + 1;
|
||||
}
|
||||
return out;
|
||||
}
|
||||
} // namespace
|
||||
|
||||
int main(int argc, char **argv) {
|
||||
gflags::ParseCommandLineFlags(&argc, &argv, false);
|
||||
google::InitGoogleLogging(argv[0]);
|
||||
FLAGS_minloglevel = 1; // keep the matrix readable
|
||||
|
||||
std::vector<std::string> backends =
|
||||
FLAGS_backends.empty() ? defaultBackends() : split(FLAGS_backends);
|
||||
|
||||
std::cout
|
||||
<< "================================================================\n";
|
||||
std::cout << " Mooncake Memory Fabric -- Cross-Backend Conformance\n";
|
||||
std::cout
|
||||
<< "================================================================\n";
|
||||
|
||||
std::vector<BackendReport> reports;
|
||||
for (auto &proto : backends) {
|
||||
std::cout << "-- running: " << proto << " ..." << std::flush;
|
||||
BackendReport r = runConformance(proto);
|
||||
if (!r.loadable) {
|
||||
std::cout << " skipped (" << r.load_error << ")\n";
|
||||
} else {
|
||||
std::cout << " done (" << r.passed() << " pass / " << r.skipped()
|
||||
<< " skip / " << r.failed() << " fail)\n";
|
||||
}
|
||||
reports.push_back(r);
|
||||
}
|
||||
|
||||
std::cout << "\n--- Probed Capabilities "
|
||||
"-----------------------------------------\n";
|
||||
for (auto &r : reports) {
|
||||
if (!r.loadable) {
|
||||
std::cout << " " << r.backend << ": (" << r.load_error << ")\n";
|
||||
continue;
|
||||
}
|
||||
std::cout << " " << r.cap.toLine() << "\n";
|
||||
}
|
||||
|
||||
std::cout << "\n--- Conformance Matrix "
|
||||
"------------------------------------------\n";
|
||||
std::printf("%-12s", "backend");
|
||||
for (auto *c : kCols) std::printf(" %-5.5s", c);
|
||||
std::printf(" %s\n", "verdict");
|
||||
for (auto &r : reports) {
|
||||
std::printf("%-12s", r.backend.c_str());
|
||||
if (!r.loadable) {
|
||||
for (size_t i = 0; i < sizeof(kCols) / sizeof(kCols[0]); ++i)
|
||||
std::printf(" %-5s", "-");
|
||||
std::printf(" %s\n", "SKIP(no-hw)");
|
||||
continue;
|
||||
}
|
||||
for (auto &c : r.cases) std::printf(" %-5s", mark(c.outcome));
|
||||
std::printf(" %s\n", r.conformant ? "CONFORMANT" : "NON-CONFORMANT");
|
||||
}
|
||||
|
||||
std::cout << "\n--- Case Details "
|
||||
"------------------------------------------------\n";
|
||||
for (auto &r : reports) {
|
||||
if (!r.loadable) continue;
|
||||
std::cout << " [" << r.backend << "]\n";
|
||||
for (auto &c : r.cases)
|
||||
std::printf(" %-12s %-4s (%.2f ms) %s\n", c.name.c_str(),
|
||||
mark(c.outcome), c.ms, c.detail.c_str());
|
||||
}
|
||||
|
||||
if (!FLAGS_json.empty()) {
|
||||
std::ofstream os(FLAGS_json);
|
||||
os << "[";
|
||||
for (size_t i = 0; i < reports.size(); ++i) {
|
||||
os << reports[i].toJson();
|
||||
if (i + 1 < reports.size()) os << ",";
|
||||
}
|
||||
os << "]\n";
|
||||
std::cout << "\nJSON report -> " << FLAGS_json << "\n";
|
||||
}
|
||||
|
||||
int bad = 0, ran = 0;
|
||||
for (auto &r : reports) {
|
||||
if (r.loadable) {
|
||||
++ran;
|
||||
if (!r.conformant) ++bad;
|
||||
}
|
||||
}
|
||||
std::cout << "Exercised " << ran << " backend(s), " << (ran - bad)
|
||||
<< " conformant, " << bad << " non-conformant.\n";
|
||||
return bad == 0 ? 0 : 1;
|
||||
}
|
||||
|
|
@ -1,343 +0,0 @@
|
|||
// Copyright 2024 KVCache.AI
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
#include "conformance_suite.h"
|
||||
|
||||
#include <chrono>
|
||||
#include <cstdlib>
|
||||
#include <cstring>
|
||||
#include <sstream>
|
||||
#include <thread>
|
||||
|
||||
#include "common.h"
|
||||
#include "transport/cxl_transport/cxl_transport.h"
|
||||
#include "transport/ualink_transport/ualink_transport.h"
|
||||
|
||||
namespace mooncake {
|
||||
namespace conformance {
|
||||
|
||||
const char *toString(Outcome outcome) {
|
||||
switch (outcome) {
|
||||
case Outcome::PASS:
|
||||
return "PASS";
|
||||
case Outcome::SKIP:
|
||||
return "SKIP";
|
||||
case Outcome::FAIL:
|
||||
return "FAIL";
|
||||
}
|
||||
return "?";
|
||||
}
|
||||
|
||||
void fillPattern(void *buf, size_t n, uint64_t seed) {
|
||||
auto *p = static_cast<uint8_t *>(buf);
|
||||
uint64_t x = seed * 0x9E3779B97F4A7C15ull + 0x1234567;
|
||||
for (size_t i = 0; i < n; ++i) {
|
||||
x ^= x >> 30;
|
||||
x *= 0xBF58476D1CE4E5B9ull;
|
||||
x ^= x >> 27;
|
||||
p[i] = static_cast<uint8_t>((x >> ((i & 7) * 8)) ^ (i * 131));
|
||||
}
|
||||
}
|
||||
|
||||
bool checkPattern(const void *buf, size_t n, uint64_t seed) {
|
||||
std::vector<uint8_t> ref(n);
|
||||
fillPattern(ref.data(), n, seed);
|
||||
return std::memcmp(buf, ref.data(), n) == 0;
|
||||
}
|
||||
|
||||
namespace {
|
||||
// A locally generated server name and metadata mode that needs no external
|
||||
// metadata server (etcd / redis / http) -- the engine handshakes over a socket.
|
||||
// The RPC port encoded in the name must match the port passed to init() so a
|
||||
// self-loopback openSegment() can reach the local handshake daemon.
|
||||
uint16_t localRpcPort() {
|
||||
return static_cast<uint16_t>(20000 + (getpid() % 20000));
|
||||
}
|
||||
std::string localServerName() {
|
||||
return "127.0.0.1:" + std::to_string(localRpcPort());
|
||||
}
|
||||
} // namespace
|
||||
|
||||
Harness::Harness(const std::string &protocol, uint64_t window_bytes)
|
||||
: window_bytes_(window_bytes) {
|
||||
// RDMA needs topology auto-discovery to find a local RNIC; the others do
|
||||
// not and are cheaper to stand up without it.
|
||||
bool auto_discover = (protocol == "rdma");
|
||||
engine_ = std::make_unique<TransferEngine>(auto_discover);
|
||||
rpc_port_ = localRpcPort();
|
||||
std::string server = localServerName();
|
||||
if (engine_->init(P2PHANDSHAKE, server, "127.0.0.1", rpc_port_) != 0) {
|
||||
why_ = "engine init failed";
|
||||
return;
|
||||
}
|
||||
void *args[2] = {nullptr, nullptr};
|
||||
transport_ = engine_->installTransport(protocol, args);
|
||||
if (!transport_) {
|
||||
why_ = "installTransport(" + protocol + ") failed";
|
||||
return;
|
||||
}
|
||||
if (transport_->health() == fabric::HealthStatus::UNREACHABLE) {
|
||||
why_ = "transport unreachable on this host";
|
||||
return;
|
||||
}
|
||||
|
||||
// Windowed backends expose a shared window; address by offset into it.
|
||||
// Other backends register host memory and address by absolute address.
|
||||
if (auto *cxl = dynamic_cast<CxlTransport *>(transport_)) {
|
||||
window_ = cxl->getCxlBaseAddr();
|
||||
window_base_addr_ = reinterpret_cast<uint64_t>(window_);
|
||||
if (cxl->getCxlDevSize() < window_bytes_)
|
||||
window_bytes_ = cxl->getCxlDevSize();
|
||||
owns_window_ = false;
|
||||
windowed_ = true;
|
||||
segment_ = LOCAL_SEGMENT_ID;
|
||||
} else if (auto *ua = dynamic_cast<UaLinkTransport *>(transport_)) {
|
||||
window_ = ua->windowBase();
|
||||
window_base_addr_ = reinterpret_cast<uint64_t>(window_);
|
||||
if (ua->windowSize() < window_bytes_) window_bytes_ = ua->windowSize();
|
||||
owns_window_ = false;
|
||||
windowed_ = true;
|
||||
segment_ = LOCAL_SEGMENT_ID;
|
||||
} else {
|
||||
// Address-based backends (RDMA / TCP) are genuine two-endpoint
|
||||
// transports. Stand up a second "target" engine in the same process
|
||||
// that owns the window and publishes it; the initiator opens the
|
||||
// target's segment and transfers into it -- a real initiator->target
|
||||
// handshake, not a self-connect. In P2PHANDSHAKE mode the engine picks
|
||||
// its own RPC port, so the target's published name is read back rather
|
||||
// than assumed.
|
||||
target_engine_ = std::make_unique<TransferEngine>(auto_discover);
|
||||
std::string target_name = "127.0.0.1:" + std::to_string(rpc_port_ + 1);
|
||||
if (target_engine_->init(P2PHANDSHAKE, target_name, "127.0.0.1",
|
||||
rpc_port_ + 1) != 0) {
|
||||
why_ = "target engine init failed";
|
||||
return;
|
||||
}
|
||||
if (!target_engine_->installTransport(protocol, args)) {
|
||||
why_ = "target installTransport failed";
|
||||
return;
|
||||
}
|
||||
target_name = target_engine_->getLocalIpAndPort();
|
||||
window_ = std::malloc(window_bytes_);
|
||||
if (!window_) {
|
||||
why_ = "window malloc failed";
|
||||
return;
|
||||
}
|
||||
std::memset(window_, 0, window_bytes_);
|
||||
owns_window_ = true;
|
||||
windowed_ = false;
|
||||
if (target_engine_->registerLocalMemory(window_, window_bytes_,
|
||||
"cpu:0") != 0) {
|
||||
why_ = "registerLocalMemory(target window) failed";
|
||||
return;
|
||||
}
|
||||
// The target's handshake daemon may need a moment after install before
|
||||
// it accepts connections; retry the open a few times.
|
||||
for (int attempt = 0; attempt < 20; ++attempt) {
|
||||
segment_ = engine_->openSegment(target_name);
|
||||
if (segment_ != (SegmentID)-1) break;
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(100));
|
||||
}
|
||||
if (segment_ == (SegmentID)-1) {
|
||||
why_ = "openSegment(target) failed";
|
||||
return;
|
||||
}
|
||||
auto desc = engine_->getMetadata()->getSegmentDescByID(segment_);
|
||||
if (!desc || desc->buffers.empty()) {
|
||||
why_ = "target segment has no registered buffer";
|
||||
return;
|
||||
}
|
||||
window_base_addr_ = desc->buffers[0].addr;
|
||||
}
|
||||
ready_ = true;
|
||||
}
|
||||
|
||||
Harness::~Harness() {
|
||||
if (target_engine_ && owns_window_ && window_)
|
||||
target_engine_->unregisterLocalMemory(window_);
|
||||
if (owns_window_ && window_) std::free(window_);
|
||||
}
|
||||
|
||||
bool Harness::submitOne(bool is_write, void *buf, uint64_t off, uint64_t len) {
|
||||
return batch({{is_write, buf, off, len}});
|
||||
}
|
||||
|
||||
bool Harness::batch(const std::vector<Req> &reqs) {
|
||||
if (!ready_ || reqs.empty()) return reqs.empty();
|
||||
auto batch_id = engine_->allocateBatchID(reqs.size());
|
||||
if (batch_id == INVALID_BATCH_ID) return false;
|
||||
|
||||
std::vector<TransferRequest> entries;
|
||||
entries.reserve(reqs.size());
|
||||
for (auto &r : reqs) {
|
||||
TransferRequest e;
|
||||
e.opcode = r.is_write ? TransferRequest::WRITE : TransferRequest::READ;
|
||||
e.source = r.buf;
|
||||
e.target_id = segment_;
|
||||
// Windowed backends (CXL / UALink) take a window-relative offset; the
|
||||
// address-based backends (RDMA / TCP) take an absolute address.
|
||||
e.target_offset = windowed_ ? r.off : (window_base_addr_ + r.off);
|
||||
e.length = r.len;
|
||||
entries.push_back(e);
|
||||
}
|
||||
|
||||
Status s = engine_->submitTransfer(batch_id, entries);
|
||||
if (!s.ok()) {
|
||||
engine_->freeBatchID(batch_id);
|
||||
return false;
|
||||
}
|
||||
|
||||
bool all_ok = true;
|
||||
auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(20);
|
||||
for (size_t i = 0; i < entries.size(); ++i) {
|
||||
for (;;) {
|
||||
TransferStatus ts;
|
||||
Status gs = engine_->getTransferStatus(batch_id, i, ts);
|
||||
if (!gs.ok()) {
|
||||
all_ok = false;
|
||||
break;
|
||||
}
|
||||
if (ts.s == TransferStatusEnum::COMPLETED) break;
|
||||
if (ts.s == TransferStatusEnum::FAILED) {
|
||||
all_ok = false;
|
||||
break;
|
||||
}
|
||||
if (std::chrono::steady_clock::now() > deadline) {
|
||||
all_ok = false;
|
||||
break;
|
||||
}
|
||||
std::this_thread::yield();
|
||||
}
|
||||
}
|
||||
engine_->freeBatchID(batch_id);
|
||||
return all_ok;
|
||||
}
|
||||
|
||||
bool Harness::write(void *buf, uint64_t off, uint64_t len) {
|
||||
return submitOne(true, buf, off, len);
|
||||
}
|
||||
|
||||
bool Harness::read(void *buf, uint64_t off, uint64_t len) {
|
||||
return submitOne(false, buf, off, len);
|
||||
}
|
||||
|
||||
void Harness::registerLocal(void *buf, uint64_t len) {
|
||||
// Windowed backends memcpy from a plain host source, so registration is
|
||||
// only needed for the address-based (verbs) backends.
|
||||
if (!windowed_ && engine_ && buf && len)
|
||||
engine_->registerLocalMemory(buf, len, "cpu:0");
|
||||
}
|
||||
|
||||
void Harness::unregisterLocal(void *buf) {
|
||||
if (!windowed_ && engine_ && buf) engine_->unregisterLocalMemory(buf);
|
||||
}
|
||||
|
||||
int BackendReport::passed() const {
|
||||
int n = 0;
|
||||
for (auto &c : cases)
|
||||
if (c.outcome == Outcome::PASS) ++n;
|
||||
return n;
|
||||
}
|
||||
int BackendReport::skipped() const {
|
||||
int n = 0;
|
||||
for (auto &c : cases)
|
||||
if (c.outcome == Outcome::SKIP) ++n;
|
||||
return n;
|
||||
}
|
||||
int BackendReport::failed() const {
|
||||
int n = 0;
|
||||
for (auto &c : cases)
|
||||
if (c.outcome == Outcome::FAIL) ++n;
|
||||
return n;
|
||||
}
|
||||
|
||||
std::string BackendReport::toJson() const {
|
||||
std::ostringstream o;
|
||||
o << "{\"backend\":\"" << backend
|
||||
<< "\",\"conformant\":" << (conformant ? "true" : "false")
|
||||
<< ",\"loadable\":" << (loadable ? "true" : "false") << ",\"health\":\""
|
||||
<< fabric::toString(health) << "\",\"capability\":" << cap.toJson()
|
||||
<< ",\"cases\":[";
|
||||
for (size_t i = 0; i < cases.size(); ++i) {
|
||||
const auto &c = cases[i];
|
||||
o << "{\"name\":\"" << c.name << "\",\"outcome\":\""
|
||||
<< toString(c.outcome)
|
||||
<< "\",\"required\":" << (c.required ? "true" : "false")
|
||||
<< ",\"ms\":" << c.ms << ",\"detail\":\"";
|
||||
for (char ch : c.detail) {
|
||||
if (ch == '"' || ch == '\\') o << '\\';
|
||||
o << ch;
|
||||
}
|
||||
o << "\"}";
|
||||
if (i + 1 < cases.size()) o << ",";
|
||||
}
|
||||
o << "]}";
|
||||
return o.str();
|
||||
}
|
||||
|
||||
namespace {
|
||||
CaseResult timed(const std::string &name, bool required,
|
||||
Outcome (*fn)(Harness &, const fabric::Capability &,
|
||||
std::string &),
|
||||
Harness &h, const fabric::Capability &cap) {
|
||||
CaseResult r;
|
||||
r.name = name;
|
||||
r.required = required;
|
||||
std::string detail;
|
||||
auto t0 = std::chrono::steady_clock::now();
|
||||
r.outcome = fn(h, cap, detail);
|
||||
auto t1 = std::chrono::steady_clock::now();
|
||||
r.ms = std::chrono::duration<double, std::milli>(t1 - t0).count();
|
||||
r.detail = detail;
|
||||
return r;
|
||||
}
|
||||
} // namespace
|
||||
|
||||
BackendReport runConformance(const std::string &protocol) {
|
||||
BackendReport rep;
|
||||
rep.backend = protocol;
|
||||
|
||||
const uint64_t kWindow = 64ull * 1024 * 1024;
|
||||
Harness h(protocol, kWindow);
|
||||
if (!h.ready()) {
|
||||
rep.loadable = false;
|
||||
rep.load_error = h.why();
|
||||
if (h.transport()) {
|
||||
rep.cap = h.transport()->probe();
|
||||
rep.health = h.transport()->health();
|
||||
}
|
||||
return rep;
|
||||
}
|
||||
rep.cap = h.transport()->probe();
|
||||
rep.backend = rep.cap.name.empty() ? protocol : rep.cap.name;
|
||||
rep.health = h.transport()->health();
|
||||
|
||||
rep.cases.push_back(timed("capability", true, caseCapability, h, rep.cap));
|
||||
rep.cases.push_back(timed("register", true, caseRegister, h, rep.cap));
|
||||
rep.cases.push_back(timed("roundtrip", true, caseRoundtrip, h, rep.cap));
|
||||
rep.cases.push_back(timed("batch", true, caseBatch, h, rep.cap));
|
||||
rep.cases.push_back(timed("ordering", false, caseOrdering, h, rep.cap));
|
||||
rep.cases.push_back(
|
||||
timed("concurrency", false, caseConcurrency, h, rep.cap));
|
||||
|
||||
rep.conformant = true;
|
||||
for (auto &c : rep.cases) {
|
||||
if (c.outcome == Outcome::FAIL) rep.conformant = false;
|
||||
if (c.required && c.outcome != Outcome::PASS) rep.conformant = false;
|
||||
}
|
||||
return rep;
|
||||
}
|
||||
|
||||
} // namespace conformance
|
||||
} // namespace mooncake
|
||||
|
|
@ -1,141 +0,0 @@
|
|||
// Copyright 2024 KVCache.AI
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
#ifndef FABRIC_CONFORMANCE_SUITE_H_
|
||||
#define FABRIC_CONFORMANCE_SUITE_H_
|
||||
|
||||
#include <cstdint>
|
||||
#include <memory>
|
||||
#include <string>
|
||||
#include <vector>
|
||||
|
||||
#include "fabric/capability.h"
|
||||
#include "transfer_engine.h"
|
||||
#include "transport/transport.h"
|
||||
|
||||
namespace mooncake {
|
||||
namespace conformance {
|
||||
|
||||
enum class Outcome { PASS, SKIP, FAIL };
|
||||
const char *toString(Outcome outcome);
|
||||
|
||||
struct CaseResult {
|
||||
std::string name;
|
||||
Outcome outcome = Outcome::SKIP;
|
||||
std::string detail;
|
||||
double ms = 0.0;
|
||||
bool required = false;
|
||||
};
|
||||
|
||||
struct BackendReport {
|
||||
std::string backend;
|
||||
fabric::Capability cap;
|
||||
fabric::HealthStatus health = fabric::HealthStatus::UNINITIALIZED;
|
||||
std::vector<CaseResult> cases;
|
||||
bool conformant = false;
|
||||
bool loadable = true; // false if the transport could not be installed
|
||||
std::string load_error;
|
||||
|
||||
int passed() const;
|
||||
int skipped() const;
|
||||
int failed() const;
|
||||
std::string toJson() const;
|
||||
};
|
||||
|
||||
// Drives one transport (installed in a single-process TransferEngine using the
|
||||
// server-free P2PHANDSHAKE metadata mode) through a backend-agnostic data path.
|
||||
// It hides the two addressing conventions in the engine: windowed backends
|
||||
// (CXL / UALink) address by offset into a shared window, while RDMA / TCP
|
||||
// address by absolute registered address. Every case is written against the
|
||||
// uniform write()/read() here, so the same suite exercises every backend.
|
||||
class Harness {
|
||||
public:
|
||||
Harness(const std::string &protocol, uint64_t window_bytes);
|
||||
~Harness();
|
||||
|
||||
// True once the engine + transport + window are ready. If false, why() says
|
||||
// what went wrong (no hardware / install failed) so the runner can skip.
|
||||
bool ready() const { return ready_; }
|
||||
const std::string &why() const { return why_; }
|
||||
|
||||
Transport *transport() const { return transport_; }
|
||||
uint64_t windowBytes() const { return window_bytes_; }
|
||||
|
||||
// Synchronous WRITE / READ of `len` bytes between local `buf` and window
|
||||
// offset `off`. Returns true iff the batch COMPLETED.
|
||||
bool write(void *buf, uint64_t off, uint64_t len);
|
||||
bool read(void *buf, uint64_t off, uint64_t len);
|
||||
|
||||
// Submit a batch of (buf, off, len) WRITE or READ requests in one call;
|
||||
// returns true iff every request COMPLETED.
|
||||
struct Req {
|
||||
bool is_write;
|
||||
void *buf;
|
||||
uint64_t off;
|
||||
uint64_t len;
|
||||
};
|
||||
bool batch(const std::vector<Req> &reqs);
|
||||
|
||||
// Register / unregister a local source or destination buffer on the
|
||||
// initiator engine. Required by verbs-based backends (the source of a
|
||||
// one-sided RDMA op must be a registered MR); a no-op-equivalent for the
|
||||
// windowed and socket backends. Cases call these around their buffers.
|
||||
void registerLocal(void *buf, uint64_t len);
|
||||
void unregisterLocal(void *buf);
|
||||
|
||||
private:
|
||||
bool submitOne(bool is_write, void *buf, uint64_t off, uint64_t len);
|
||||
|
||||
std::unique_ptr<TransferEngine> engine_;
|
||||
// For address-based backends (RDMA / TCP) a second engine owns the target
|
||||
// window so the transfer is a real initiator->target handshake in process.
|
||||
std::unique_ptr<TransferEngine> target_engine_;
|
||||
Transport *transport_ = nullptr;
|
||||
SegmentID segment_ = 0;
|
||||
void *window_ = nullptr; // backing for offset addressing
|
||||
uint64_t window_base_addr_ = 0; // absolute base used to form target_offset
|
||||
uint64_t window_bytes_ = 0;
|
||||
bool owns_window_ = false;
|
||||
bool windowed_ = false; // true => target_offset is window-relative
|
||||
bool ready_ = false;
|
||||
uint16_t rpc_port_ = 0;
|
||||
std::string why_;
|
||||
};
|
||||
|
||||
// Deterministic, position-dependent pattern so partial / overlapping writes are
|
||||
// caught (a flat memset would hide offset bugs).
|
||||
void fillPattern(void *buf, size_t n, uint64_t seed);
|
||||
bool checkPattern(const void *buf, size_t n, uint64_t seed);
|
||||
|
||||
// The cases. Each returns PASS / SKIP / FAIL and fills `detail`.
|
||||
Outcome caseCapability(Harness &h, const fabric::Capability &cap,
|
||||
std::string &detail);
|
||||
Outcome caseRegister(Harness &h, const fabric::Capability &cap,
|
||||
std::string &detail);
|
||||
Outcome caseRoundtrip(Harness &h, const fabric::Capability &cap,
|
||||
std::string &detail);
|
||||
Outcome caseBatch(Harness &h, const fabric::Capability &cap,
|
||||
std::string &detail);
|
||||
Outcome caseOrdering(Harness &h, const fabric::Capability &cap,
|
||||
std::string &detail);
|
||||
Outcome caseConcurrency(Harness &h, const fabric::Capability &cap,
|
||||
std::string &detail);
|
||||
|
||||
// Install + probe + run every case against one transport protocol.
|
||||
BackendReport runConformance(const std::string &protocol);
|
||||
|
||||
} // namespace conformance
|
||||
} // namespace mooncake
|
||||
|
||||
#endif // FABRIC_CONFORMANCE_SUITE_H_
|
||||
|
|
@ -1,308 +0,0 @@
|
|||
// Copyright 2024 KVCache.AI
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
// fabric_cli -- diagnostics for the memory fabric capability / routing layer.
|
||||
// probe install each backend, print its probed Capability
|
||||
// select run the path selector on representative endpoint pairs and
|
||||
// show the fallback ladder, including a live fault-injection
|
||||
// demo
|
||||
// cost size-aware cost-model selection sweep
|
||||
// plugins DIR dlopen transport plugins from DIR and probe them
|
||||
// conformance run the conformance suite over every backend
|
||||
//
|
||||
// probe and conformance drive the real TransferEngine (server-free
|
||||
// P2PHANDSHAKE metadata), so the numbers come from the real data path.
|
||||
|
||||
#include <gflags/gflags.h>
|
||||
#include <glog/logging.h>
|
||||
|
||||
#include <cstdio>
|
||||
#include <functional>
|
||||
#include <string>
|
||||
#include <vector>
|
||||
|
||||
#include "fabric/plugin_loader.h"
|
||||
|
||||
#include "conformance_suite.h"
|
||||
#include "fabric/path_selector.h"
|
||||
|
||||
using namespace mooncake;
|
||||
using mooncake::fabric::Endpoint;
|
||||
|
||||
namespace {
|
||||
|
||||
std::vector<std::string> builtinBackends() {
|
||||
std::vector<std::string> v{"tcp"};
|
||||
#ifdef USE_CXL
|
||||
v.push_back("cxl");
|
||||
#endif
|
||||
#ifdef USE_UALINK
|
||||
v.push_back("ualink");
|
||||
#endif
|
||||
v.push_back("rdma");
|
||||
return v;
|
||||
}
|
||||
|
||||
int cmdProbe() {
|
||||
for (auto &proto : builtinBackends()) {
|
||||
conformance::Harness h(proto, 8ull * 1024 * 1024);
|
||||
if (!h.ready()) {
|
||||
std::printf(" %-12s UNREACHABLE on this host (%s)\n",
|
||||
proto.c_str(), h.why().c_str());
|
||||
continue;
|
||||
}
|
||||
std::printf(" %s\n", h.transport()->probe().toLine().c_str());
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
int cmdConformance() {
|
||||
int bad = 0, ran = 0;
|
||||
for (auto &proto : builtinBackends()) {
|
||||
auto r = conformance::runConformance(proto);
|
||||
if (!r.loadable) {
|
||||
std::printf(" %-12s : UNREACHABLE -> skipped (%s)\n",
|
||||
proto.c_str(), r.load_error.c_str());
|
||||
continue;
|
||||
}
|
||||
++ran;
|
||||
std::printf(" %-12s : %s (%d pass / %d skip / %d fail)\n",
|
||||
r.backend.c_str(),
|
||||
r.conformant ? "CONFORMANT" : "NON-CONFORMANT", r.passed(),
|
||||
r.skipped(), r.failed());
|
||||
if (!r.conformant) ++bad;
|
||||
}
|
||||
std::printf(" exercised %d backend(s), %d non-conformant\n", ran, bad);
|
||||
return bad == 0 ? 0 : 1;
|
||||
}
|
||||
|
||||
// For the routing demo the selector consults a kind+health table. We register
|
||||
// the kinds probed live on this host as HEALTHY, plus the kinds that a full
|
||||
// deployment would have (NVLINK, PCIE) as synthetic entries so the priority
|
||||
// ladder and fault-driven fallback are visible end to end. A small flag lets us
|
||||
// toggle the NVLINK entry's health to drive the fallback.
|
||||
struct DemoKind {
|
||||
fabric::MemoryFabricKind kind;
|
||||
std::string name;
|
||||
bool healthy = true;
|
||||
fabric::Capability cap; // modelled bandwidth/latency for the cost demo
|
||||
};
|
||||
|
||||
// Representative bandwidth/latency for a kind, used by the size-aware cost
|
||||
// model demo. NVLink/RDMA/TCP figures match the order of magnitude measured on
|
||||
// this class of host (NVLink ~350 GB/s, IB ~25 GB/s, TCP ~5 GB/s); the emulated
|
||||
// kinds report their host-window throughput.
|
||||
fabric::Capability modelCap(fabric::MemoryFabricKind kind,
|
||||
const std::string &name) {
|
||||
fabric::Capability c;
|
||||
c.kind = kind;
|
||||
c.name = name;
|
||||
using K = fabric::MemoryFabricKind;
|
||||
// Only NVLink and RDMA (GPUDirect) can address device memory directly; the
|
||||
// others must stage GPU<->GPU traffic through host memory.
|
||||
c.supports_device_memory = (kind == K::NVLINK || kind == K::RDMA);
|
||||
switch (kind) {
|
||||
case K::NVLINK:
|
||||
c.max_bandwidth_mbps = 350000;
|
||||
c.latency_ns = 8000;
|
||||
break;
|
||||
case K::CXL:
|
||||
c.max_bandwidth_mbps = 26000;
|
||||
c.latency_ns = 300;
|
||||
break;
|
||||
case K::RDMA:
|
||||
c.max_bandwidth_mbps = 25000;
|
||||
c.latency_ns = 2500;
|
||||
break;
|
||||
case K::UALINK:
|
||||
c.max_bandwidth_mbps = 28000;
|
||||
c.latency_ns = 250;
|
||||
break;
|
||||
case K::NUMA:
|
||||
c.max_bandwidth_mbps = 18000;
|
||||
c.latency_ns = 200;
|
||||
break;
|
||||
case K::TCP:
|
||||
c.max_bandwidth_mbps = 5000;
|
||||
c.latency_ns = 20000;
|
||||
break;
|
||||
default:
|
||||
c.max_bandwidth_mbps = 4000;
|
||||
c.latency_ns = 5000;
|
||||
break;
|
||||
}
|
||||
return c;
|
||||
}
|
||||
|
||||
std::vector<DemoKind> demoKinds() {
|
||||
std::vector<DemoKind> kinds;
|
||||
for (auto &proto : builtinBackends()) {
|
||||
conformance::Harness h(proto, 4ull * 1024 * 1024);
|
||||
if (h.ready()) {
|
||||
auto cap = h.transport()->probe();
|
||||
DemoKind d{cap.kind, proto, true, cap};
|
||||
// Backends that do not self-calibrate (TCP/RDMA) get a modelled cap
|
||||
// so the cost demo has numbers to work with.
|
||||
if (d.cap.max_bandwidth_mbps == 0)
|
||||
d.cap = modelCap(cap.kind, proto);
|
||||
kinds.push_back(d);
|
||||
}
|
||||
}
|
||||
// Synthetic high-end intra-node GPU link present in a full deployment.
|
||||
kinds.push_back({fabric::MemoryFabricKind::NVLINK, "nvlink", true,
|
||||
modelCap(fabric::MemoryFabricKind::NVLINK, "nvlink")});
|
||||
return kinds;
|
||||
}
|
||||
|
||||
int cmdSelect() {
|
||||
std::vector<DemoKind> kinds = demoKinds();
|
||||
|
||||
fabric::PathSelector sel;
|
||||
std::vector<bool> health(kinds.size(), true);
|
||||
for (size_t i = 0; i < kinds.size(); ++i) {
|
||||
size_t idx = i;
|
||||
sel.registerKind(
|
||||
kinds[i].name, kinds[i].kind,
|
||||
[&health, idx] {
|
||||
return health[idx] ? fabric::HealthStatus::HEALTHY
|
||||
: fabric::HealthStatus::UNREACHABLE;
|
||||
},
|
||||
kinds[i].cap);
|
||||
}
|
||||
|
||||
Endpoint gpu0{Endpoint::Kind::GPU, 0, 0, "node0", "gpu0"};
|
||||
Endpoint gpu3{Endpoint::Kind::GPU, 3, 0, "node0", "gpu3"};
|
||||
Endpoint store1{Endpoint::Kind::STORE, -1, 0, "node0", "store1"};
|
||||
Endpoint gpuA{Endpoint::Kind::GPU, 0, 0, "nodeA", "gpuA0"};
|
||||
Endpoint gpuB{Endpoint::Kind::GPU, 1, 0, "nodeB", "gpuB1"};
|
||||
Endpoint hostX{Endpoint::Kind::HOST, -1, 0, "node0", "hostmemX"};
|
||||
Endpoint hostY{Endpoint::Kind::HOST, -1, 1, "node0", "hostmemY"};
|
||||
|
||||
std::printf(
|
||||
"--- automatic path selection -----------------------------------\n");
|
||||
struct Pair {
|
||||
Endpoint a, b;
|
||||
};
|
||||
Pair pairs[] = {{gpu0, gpu3}, {gpu0, store1}, {gpuA, gpuB}, {hostX, hostY}};
|
||||
for (auto &p : pairs) {
|
||||
auto c = sel.select(p.a, p.b);
|
||||
std::printf("%s\n", sel.logChoice(p.a, p.b, c).c_str());
|
||||
}
|
||||
|
||||
std::printf(
|
||||
"\n--- fault-injection / fallback demo ----------------------------\n");
|
||||
int nvlink_idx = -1;
|
||||
for (size_t i = 0; i < kinds.size(); ++i)
|
||||
if (kinds[i].kind == fabric::MemoryFabricKind::NVLINK) nvlink_idx = i;
|
||||
if (nvlink_idx >= 0) {
|
||||
auto before = sel.select(gpu0, gpu3);
|
||||
std::printf("before: %s\n", sel.logChoice(gpu0, gpu3, before).c_str());
|
||||
std::printf(
|
||||
">> injecting fault into 'nvlink' (simulating link/peer loss)\n");
|
||||
health[nvlink_idx] = false;
|
||||
auto after = sel.select(gpu0, gpu3);
|
||||
std::printf("after: %s\n", sel.logChoice(gpu0, gpu3, after).c_str());
|
||||
health[nvlink_idx] = true;
|
||||
std::printf(">> fault cleared; 'nvlink' healthy again\n");
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
// Size-aware cost-model demo: for a host_host pair, sweep the transfer size and
|
||||
// show how the lowest-cost backend changes (small => low latency wins, large =>
|
||||
// high bandwidth wins).
|
||||
int cmdCost() {
|
||||
std::vector<DemoKind> kinds = demoKinds();
|
||||
fabric::PathSelector sel;
|
||||
for (auto &k : kinds)
|
||||
sel.registerKind(
|
||||
k.name, k.kind, [] { return fabric::HealthStatus::HEALTHY; },
|
||||
k.cap);
|
||||
|
||||
Endpoint gpu0{Endpoint::Kind::GPU, 0, 0, "node0", "gpu0"};
|
||||
Endpoint gpu3{Endpoint::Kind::GPU, 3, 0, "node0", "gpu3"};
|
||||
|
||||
std::printf(
|
||||
"--- size-aware cost-model selection (same-node GPU<->GPU) ------\n");
|
||||
std::printf("registered candidates and their model:\n");
|
||||
for (auto &k : kinds)
|
||||
std::printf(" %-10s bw=%8lu MB/s lat=%7lu ns\n", k.name.c_str(),
|
||||
(unsigned long)k.cap.max_bandwidth_mbps,
|
||||
(unsigned long)k.cap.latency_ns);
|
||||
std::printf("\n%-12s %-14s %-12s\n", "size", "chosen", "est cost");
|
||||
const uint64_t sizes[] = {64, 1024, 64ull * 1024,
|
||||
1ull << 20, 16ull << 20, 256ull << 20};
|
||||
for (uint64_t s : sizes) {
|
||||
auto c = sel.selectForSize(gpu0, gpu3, s);
|
||||
char szbuf[32];
|
||||
if (s < (1 << 20))
|
||||
std::snprintf(szbuf, sizeof(szbuf), "%lu B", (unsigned long)s);
|
||||
else
|
||||
std::snprintf(szbuf, sizeof(szbuf), "%lu MiB",
|
||||
(unsigned long)(s >> 20));
|
||||
std::printf("%-12s %-14s %.0f ns\n", szbuf,
|
||||
c.found ? c.label.c_str() : "-", c.estimated_cost_ns);
|
||||
}
|
||||
std::printf(
|
||||
"\nSmall transfers favour the lowest-latency fabric; large transfers "
|
||||
"favour the highest-bandwidth one -- the selector switches "
|
||||
"automatically.\n");
|
||||
return 0;
|
||||
}
|
||||
|
||||
void usage() {
|
||||
std::printf(
|
||||
"usage: fabric_cli <probe|select|cost|plugins [dir]|conformance>\n");
|
||||
}
|
||||
|
||||
// Load every transport plugin .so in a directory via the versioned C ABI, then
|
||||
// probe each one -- demonstrating that a backend can appear without relinking.
|
||||
int cmdPlugins(const std::string &dir) {
|
||||
fabric::PluginLoader loader; // declared first => outlives the transports
|
||||
std::vector<fabric::PluginInfo> infos;
|
||||
auto plugins = loader.loadDirectory(dir, &infos);
|
||||
std::printf("--- plugin loading from %s ---\n", dir.c_str());
|
||||
for (auto &pi : infos)
|
||||
std::printf(" %-44s %s%s\n", pi.path.c_str(),
|
||||
pi.ok ? "loaded" : "FAILED",
|
||||
pi.ok ? "" : (" : " + pi.error).c_str());
|
||||
if (plugins.empty()) {
|
||||
std::printf(" (no plugins found)\n");
|
||||
return 0;
|
||||
}
|
||||
std::printf("\n--- probing dynamically-loaded backends ---\n");
|
||||
for (auto &t : plugins) {
|
||||
auto cap = t->probe();
|
||||
std::printf(" %s health=%s\n", cap.toLine().c_str(),
|
||||
fabric::toString(t->health()));
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
} // namespace
|
||||
|
||||
int main(int argc, char **argv) {
|
||||
gflags::ParseCommandLineFlags(&argc, &argv, true);
|
||||
google::InitGoogleLogging(argv[0]);
|
||||
FLAGS_minloglevel = 2;
|
||||
std::string cmd = (argc > 1) ? argv[1] : "probe";
|
||||
if (cmd == "probe") return cmdProbe();
|
||||
if (cmd == "select") return cmdSelect();
|
||||
if (cmd == "cost") return cmdCost();
|
||||
if (cmd == "plugins")
|
||||
return cmdPlugins(argc > 2 ? argv[2] : "build/plugins");
|
||||
if (cmd == "conformance") return cmdConformance();
|
||||
usage();
|
||||
return 2;
|
||||
}
|
||||
|
|
@ -1,187 +0,0 @@
|
|||
// Copyright 2024 KVCache.AI
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
// A sample out-of-tree transport packaged as a shared object. It is built into
|
||||
// its own .so and is NOT linked into the Transfer Engine; PluginLoader dlopens
|
||||
// it at runtime, checks the ABI version, instantiates it, and runs probe() /
|
||||
// health() -- demonstrating that a backend can appear without relinking
|
||||
// Mooncake. The transport itself is a host-memory shared-window loopback (the
|
||||
// same data-plane shape as the CXL backend) so it is conformance-checkable.
|
||||
|
||||
#include <sys/mman.h>
|
||||
|
||||
#include <cstdint>
|
||||
#include <cstring>
|
||||
#include <string>
|
||||
|
||||
#include "fabric/plugin_loader.h"
|
||||
#include "transport/transport.h"
|
||||
|
||||
namespace mooncake {
|
||||
namespace {
|
||||
|
||||
class ShmPluginTransport : public Transport {
|
||||
public:
|
||||
ShmPluginTransport() {
|
||||
window_size_ = 64ull * 1024 * 1024;
|
||||
window_ = mmap(nullptr, window_size_, PROT_READ | PROT_WRITE,
|
||||
MAP_SHARED | MAP_ANONYMOUS, -1, 0);
|
||||
if (window_ == MAP_FAILED) window_ = nullptr;
|
||||
}
|
||||
~ShmPluginTransport() override {
|
||||
if (window_ && window_ != MAP_FAILED) munmap(window_, window_size_);
|
||||
if (metadata_) metadata_->removeSegmentDesc(local_server_name_);
|
||||
}
|
||||
|
||||
Status submitTransfer(
|
||||
BatchID batch_id,
|
||||
const std::vector<TransferRequest> &entries) override {
|
||||
auto &batch_desc = *((BatchDesc *)(batch_id));
|
||||
if (batch_desc.task_list.size() + entries.size() >
|
||||
batch_desc.batch_size) {
|
||||
return Status::InvalidArgument("shm-plugin: batch capacity");
|
||||
}
|
||||
size_t task_id = batch_desc.task_list.size();
|
||||
batch_desc.task_list.resize(task_id + entries.size());
|
||||
for (auto &request : entries) {
|
||||
TransferTask &task = batch_desc.task_list[task_id++];
|
||||
task.total_bytes = request.length;
|
||||
Slice *slice = getSliceCache().allocate();
|
||||
slice->source_addr = (char *)request.source;
|
||||
slice->cxl.dest_addr = (char *)window_ + request.target_offset;
|
||||
slice->length = request.length;
|
||||
slice->opcode = request.opcode;
|
||||
slice->task = &task;
|
||||
slice->target_id = request.target_id;
|
||||
slice->status = Slice::PENDING;
|
||||
__sync_fetch_and_add(&task.slice_count, 1);
|
||||
void *dst = (request.opcode == TransferRequest::READ)
|
||||
? slice->source_addr
|
||||
: (void *)slice->cxl.dest_addr;
|
||||
const void *src = (request.opcode == TransferRequest::READ)
|
||||
? (void *)slice->cxl.dest_addr
|
||||
: slice->source_addr;
|
||||
if (request.length) std::memcpy(dst, src, request.length);
|
||||
__sync_synchronize();
|
||||
slice->markSuccess();
|
||||
}
|
||||
return Status::OK();
|
||||
}
|
||||
|
||||
Status getTransferStatus(BatchID batch_id, size_t task_id,
|
||||
TransferStatus &status) override {
|
||||
auto &batch_desc = *((BatchDesc *)(batch_id));
|
||||
if (task_id >= batch_desc.task_list.size())
|
||||
return Status::InvalidArgument("shm-plugin: task id");
|
||||
auto &task = batch_desc.task_list[task_id];
|
||||
status.transferred_bytes = task.transferred_bytes;
|
||||
if (task.success_slice_count + task.failed_slice_count ==
|
||||
task.slice_count) {
|
||||
status.s = task.failed_slice_count ? TransferStatusEnum::FAILED
|
||||
: TransferStatusEnum::COMPLETED;
|
||||
task.is_finished = true;
|
||||
} else {
|
||||
status.s = TransferStatusEnum::WAITING;
|
||||
}
|
||||
return Status::OK();
|
||||
}
|
||||
|
||||
fabric::Capability probe() override {
|
||||
fabric::Capability cap;
|
||||
cap.kind = fabric::MemoryFabricKind::SHM;
|
||||
cap.name = "shm-plugin";
|
||||
cap.supports_host_memory = true;
|
||||
cap.supports_p2p = true;
|
||||
cap.supports_ordered_write = true;
|
||||
cap.supports_zero_copy = true;
|
||||
cap.alignment = 1;
|
||||
cap.max_transfer_size = window_size_;
|
||||
cap.emulated = true;
|
||||
cap.notes = "out-of-tree shared-memory loopback plugin (.so)";
|
||||
return cap;
|
||||
}
|
||||
|
||||
fabric::HealthStatus health() override {
|
||||
return window_ ? fabric::HealthStatus::HEALTHY
|
||||
: fabric::HealthStatus::UNREACHABLE;
|
||||
}
|
||||
|
||||
void *windowBase() const { return window_; }
|
||||
size_t windowSize() const { return window_size_; }
|
||||
|
||||
private:
|
||||
int install(std::string &local_server_name,
|
||||
std::shared_ptr<TransferMetadata> meta,
|
||||
std::shared_ptr<Topology> topo) override {
|
||||
(void)topo;
|
||||
metadata_ = meta;
|
||||
local_server_name_ = local_server_name;
|
||||
if (!window_) return -1;
|
||||
auto desc = metadata_->getSegmentDesc(local_server_name_);
|
||||
if (!desc) desc = std::make_shared<SegmentDesc>();
|
||||
desc->name = local_server_name_;
|
||||
desc->protocol = "shm-plugin";
|
||||
desc->cxl_base_addr = (uint64_t)window_;
|
||||
desc->cxl_name = "shm-plugin";
|
||||
metadata_->addLocalSegment(LOCAL_SEGMENT_ID, local_server_name_,
|
||||
std::move(desc));
|
||||
return metadata_->updateLocalSegmentDesc();
|
||||
}
|
||||
|
||||
int registerLocalMemory(void *addr, size_t length,
|
||||
const std::string &location, bool remote_accessible,
|
||||
bool update_metadata) override {
|
||||
(void)location;
|
||||
(void)remote_accessible;
|
||||
BufferDesc d;
|
||||
d.name = local_server_name_;
|
||||
uintptr_t base = (uintptr_t)window_;
|
||||
uintptr_t ptr = (uintptr_t)addr;
|
||||
if (ptr < base || ptr + length > base + window_size_) {
|
||||
errno = EFAULT;
|
||||
return -1;
|
||||
}
|
||||
d.offset = ptr - base;
|
||||
d.length = length;
|
||||
return metadata_->addLocalMemoryBuffer(d, update_metadata);
|
||||
}
|
||||
|
||||
int unregisterLocalMemory(void *addr, bool update_metadata) override {
|
||||
return metadata_->removeLocalMemoryBuffer(addr, update_metadata);
|
||||
}
|
||||
|
||||
int registerLocalMemoryBatch(const std::vector<BufferEntry> &buffer_list,
|
||||
const std::string &location) override {
|
||||
for (auto &b : buffer_list)
|
||||
registerLocalMemory(b.addr, b.length, location, true, false);
|
||||
return metadata_->updateLocalSegmentDesc();
|
||||
}
|
||||
|
||||
int unregisterLocalMemoryBatch(
|
||||
const std::vector<void *> &addr_list) override {
|
||||
for (auto &a : addr_list) unregisterLocalMemory(a, false);
|
||||
return metadata_->updateLocalSegmentDesc();
|
||||
}
|
||||
|
||||
const char *getName() const override { return "shm-plugin"; }
|
||||
|
||||
void *window_ = nullptr;
|
||||
size_t window_size_ = 0;
|
||||
};
|
||||
|
||||
} // namespace
|
||||
} // namespace mooncake
|
||||
|
||||
MOONCAKE_DEFINE_TRANSPORT_PLUGIN(new mooncake::ShmPluginTransport(),
|
||||
"shm-plugin")
|
||||
Loading…
Reference in New Issue