diff --git a/mooncake-transfer-engine/include/fabric/capability.h b/mooncake-transfer-engine/include/fabric/capability.h new file mode 100644 index 00000000..bc9cc85a --- /dev/null +++ b/mooncake-transfer-engine/include/fabric/capability.h @@ -0,0 +1,110 @@ +// 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 +#include + +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_ diff --git a/mooncake-transfer-engine/include/fabric/probe_calibration.h b/mooncake-transfer-engine/include/fabric/probe_calibration.h new file mode 100644 index 00000000..2bd6ea13 --- /dev/null +++ b/mooncake-transfer-engine/include/fabric/probe_calibration.h @@ -0,0 +1,82 @@ +// 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 +#include +#include +#include +#include + +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; + +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(t1 - t0).count(); + if (secs > 0) { + double mb = static_cast(moved) / (1024.0 * 1024.0); + result.bandwidth_mbps = static_cast(mb / secs); + } + + std::vector 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(b - a).count()); + } + std::sort(samples.begin(), samples.end()); + result.latency_ns = static_cast(samples[samples.size() / 2]); + result.ok = true; + return result; +} + +} // namespace fabric +} // namespace mooncake + +#endif // FABRIC_PROBE_CALIBRATION_H_ diff --git a/mooncake-transfer-engine/include/transport/transport.h b/mooncake-transfer-engine/include/transport/transport.h index e3a22c7e..a4247e88 100644 --- a/mooncake-transfer-engine/include/transport/transport.h +++ b/mooncake-transfer-engine/include/transport/transport.h @@ -32,6 +32,7 @@ #include #include "common/base/status.h" +#include "fabric/capability.h" #include "transfer_metadata.h" namespace mooncake { @@ -374,6 +375,28 @@ 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 meta, diff --git a/mooncake-transfer-engine/src/CMakeLists.txt b/mooncake-transfer-engine/src/CMakeLists.txt index 95590dc6..12291685 100644 --- a/mooncake-transfer-engine/src/CMakeLists.txt +++ b/mooncake-transfer-engine/src/CMakeLists.txt @@ -1,5 +1,6 @@ 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) @@ -10,7 +11,8 @@ if(USE_HIP) hipify_files(ENGINE_SOURCES) endif() -add_library(transfer_engine ${ENGINE_SOURCES} $) +add_library(transfer_engine ${ENGINE_SOURCES} $ + $) if(BUILD_SHARED_LIBS) install(TARGETS transfer_engine DESTINATION lib) endif() diff --git a/mooncake-transfer-engine/src/fabric/CMakeLists.txt b/mooncake-transfer-engine/src/fabric/CMakeLists.txt new file mode 100644 index 00000000..e09d4ebd --- /dev/null +++ b/mooncake-transfer-engine/src/fabric/CMakeLists.txt @@ -0,0 +1,4 @@ +file(GLOB FABRIC_SOURCES "*.cpp") + +add_library(fabric OBJECT ${FABRIC_SOURCES}) +target_link_libraries(fabric PRIVATE glog::glog ${CMAKE_DL_LIBS}) diff --git a/mooncake-transfer-engine/src/fabric/capability.cpp b/mooncake-transfer-engine/src/fabric/capability.cpp new file mode 100644 index 00000000..5349f85c --- /dev/null +++ b/mooncake-transfer-engine/src/fabric/capability.cpp @@ -0,0 +1,158 @@ +// 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 +#include + +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 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 table = { + {"tcp", MemoryFabricKind::TCP}, + {"rdma", MemoryFabricKind::RDMA}, + {"efa", MemoryFabricKind::RDMA}, + {"barex", MemoryFabricKind::RDMA}, + {"nvlink", MemoryFabricKind::NVLINK}, + {"nvlink_intra", 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