sycl: port 0.1.39 (re-migration with dpct 2025.3, 3-way merge per docs/INTEL.md)

The SYCL port of PR #423 tracked 0.1.38; 13 of its migrated copies were behind
0.1.39's sources. Re-migrated 0.1.38 and 0.1.39 with sycl/tools/migrate.sh,
normalized, merged with sycl/tools/merge_upstream.py:

- clean merges: expert_cache, expert_source, native_dense, pinned, elementwise,
  kv_stream, iq_multi_parity, prefill, verify.hpp; q8_1_finite.hpp new (#606)
- verify.cpp: the #649 trace kept with the port's queue handling
  (STRATA_VERIFY_TRACE; the HIP-only coherent/doorbell switches stay HIP-only)
- generate.cpp: the port's --prefill-borrow beside --vram-elastic/--vram-segment-mib
- iq_kernels / native_mmvq: the port's kernels kept, #606's finite q8_1 scale
  and sum applied by hand (dpct's conflicts there were kernel-name hashes)
- expert_cache: --vram-elastic (#533) is CUDA-only on the port too (no driver
  VMM on Level Zero through dpct); it reports so instead of failing to build
- version 0.1.39

Compiled here with icpx 2026.1.1 (SPIR-V, no Intel GPU on this PC): strata
builds. Not run on an Arc.
This commit is contained in:
Niko1221 committed 2026-10-04 00:08:15 +02:00
1 parent ef18ede418
commit 5ffcce4382
15 files changed
+1666 -134

No files matched your search

+1 -1
View File
@@ -10,7 +10,7 @@
#
# Device code is JIT-compiled from SPIR-V unless STRATA_SYCL_AOT names a device (bmg-g31 for the Arc Pro B70).
cmake_minimum_required(VERSION 3.24)
project(strata_sycl VERSION 0.1.38 LANGUAGES C CXX) # the upstream engine version this port tracks
project(strata_sycl VERSION 0.1.39 LANGUAGES C CXX) # the upstream engine version this port tracks
set(STRATA_ROOT ${CMAKE_CURRENT_SOURCE_DIR}/..)
set(CMAKE_CXX_STANDARD 20)
set(CMAKE_CXX_STANDARD_REQUIRED ON)
+9
View File
@@ -61,6 +61,9 @@ public:
/// The watchdog's view of the window in flight (issue #31): the layer, the GPU's sequence, the flags.
void diag(std::FILE* f) const;
/// #649 (STRATA_VERIFY_TRACE=1, opt-in): the last windows' host-side handshake events and the GPU's breadcrumbs
/// (in mapped memory, so they read while the GPU hangs). Nothing when the trace is off.
void trace_dump(std::FILE* f) const;
/// #267: raise every flag the window's spin kernels wait on past any ring (UINT32_MAX), so a window the GPU
/// cannot finish drains instead of staying resident, then wait up to `timeout_ms` for its streams. For the
@@ -193,6 +196,12 @@ private:
int pending_commit_ = 0; ///< n_keep of a launched, unfinished commit (0: none)
std::chrono::steady_clock::time_point pending_commit_t0_{};
bool record_window(int T, dpct::queue_ptr cs, std::string &err);
// #649: STRATA_VERIFY_TRACE=1 - a host event ring (trace_ev) and GPU breadcrumbs: the profiler's stamp points,
// per layer and token group, written to mapped memory (null when the trace is off)
void trace_ev(const char* what, int64_t step, int64_t layer, int64_t aux) const;
unsigned long long* trace_h_ = nullptr;
unsigned long long* trace_m_ = nullptr;
size_t trace_n_ = 0;
static constexpr int kProfPer = 33; // stamps per layer (32 left the hc-read second
// half's up-stamp at slot 32 = the next layer's slot 0: D8)
bool prof_on_ = false;
@@ -0,0 +1,41 @@
// include/strata/kernels/q8_1_finite.hpp - #606: the q8_1 activation block's scale and sum, kept finite.
//
// A q8_1 block stores its scale d = amax / 127 and the sum of its 32 raw activations as fp16. A sum past 65504 (one
// massive-activation dimension among 32 values is enough) rounds to inf, and d does past amax = 8.3M; the dot
// products then read inf * 0 = NaN, and a NaN in the residual ends as one token repeated (ggml-org/llama.cpp#23606
// is the same defect). Clamped to the largest finite half, such a block gives an approximate product instead of NaN,
// and the values are held to the int8 range they then exceed. Every block that was finite before is stored bit for
// bit as before: values in (65504, 65520) round to 65504 anyway, and a finite block's |x / d| is at most 127. A NaN
// stays NaN (the comparisons are false for it), so STRATA_DBG_NAN still sees one.
#pragma once
#define DPCT_PROFILING_ENABLED
#include <sycl/sycl.hpp>
#include <dpct/dpct.hpp>
namespace strata::kernels {
constexpr float kHalfMax = 65504.0f;
__dpct_inline__ float q8_1_finite(const float v) {
return sycl::fabs(v) > kHalfMax ? sycl::copysign((float)kHalfMax, (float)v)
: v;
}
/// The block's quantized value of `xi` for the (finite) scale `d`; `amax` = 0 is an all-zero block.
__dpct_inline__ int8_t q8_1_quant(const float xi, const float d,
const float amax) {
if (amax == 0.0f) return 0;
const float q = sycl::round(xi / d);
return (int8_t) (q > 127.0f ? 127.0f : q < -127.0f ? -127.0f : q);
}
__dpct_inline__ sycl::half2 q8_1_ds(const float d, const float sum) {
return sycl::half2(
sycl::vec<float, 1>(q8_1_finite(d))
.convert<sycl::half, sycl::rounding_mode::automatic>()[0],
sycl::vec<float, 1>(q8_1_finite(sum))
.convert<sycl::half, sycl::rounding_mode::automatic>()[0]);
}
} // namespace strata::kernels
+316 -4
View File
@@ -1,10 +1,17 @@
// src/core/expert_cache.cpp - R4's slot storage and residency table. Read the header first.
#define DPCT_COMPAT_RT_VERSION 12080
// src/core/expert_cache.cpp - R4's slot storage and residency table. Read the
// header first.
#define DPCT_PROFILING_ENABLED
#include <sycl/sycl.hpp>
#include <dpct/dpct.hpp>
#include "strata/sycl_queue.hpp"
#include "strata/core/expert_cache.hpp"
#if !defined(STRATA_USE_HIP)
// #533: the virtual memory management types (the functions come through the
// runtime's entry points)
#endif
#include <algorithm>
#include <cstdio>
#include <filesystem>
@@ -144,6 +151,303 @@ bool write_expert_profile(const std::string& path, int64_t n_layers, int64_t n_e
ExpertCache::~ExpertCache() { close(); }
// ---- #533: the segmented arena (--vram-elastic). The driver API's virtual memory functions, looked up through the
// runtime (no link against the driver library): one address range for the whole arena, backed by physical segments,
// and the tail's segments unmapped / mapped again later. Nothing here runs unless a segment size was set.
#if 0 // SYCL port: no driver virtual memory management (--vram-elastic is CUDA-only, #533)
namespace {
struct Vmm {
int(CUDAAPI *device_get)(int *, int) = nullptr;
int(CUDAAPI *attribute)(int *, CUdevice_attribute, int) = nullptr;
int(CUDAAPI *granularity)(
size_t *, const dpct::experimental::mem_prop *,
sycl::ext::oneapi::experimental::granularity_mode) = nullptr;
int(CUDAAPI *reserve)(dpct::device_ptr *, size_t, size_t, dpct::device_ptr,
unsigned long long) = nullptr;
int(CUDAAPI *address_free)(dpct::device_ptr, size_t) = nullptr;
int(CUDAAPI *create)(dpct::experimental::physical_mem_ptr *, size_t,
const dpct::experimental::mem_prop *,
unsigned long long) = nullptr;
int(CUDAAPI *release)(dpct::experimental::physical_mem_ptr) = nullptr;
int(CUDAAPI *map)(dpct::device_ptr, size_t, size_t,
dpct::experimental::physical_mem_ptr,
unsigned long long) = nullptr;
int(CUDAAPI *unmap)(dpct::device_ptr, size_t) = nullptr;
int(CUDAAPI *set_access)(dpct::device_ptr, size_t,
const dpct::experimental::mem_access_desc *,
size_t) = nullptr;
bool ok = false;
};
template <class F> bool entry(const char *name, F &f) try {
void* p = nullptr;
cudaDriverEntryPointQueryResult q{};
#if DPCT_COMPAT_RT_VERSION >= 12050
/*
DPCT1007: Migration of cudaGetDriverEntryPointByVersion is not
supported.
*/
const dpct::err0 e = cudaGetDriverEntryPointByVersion(
name, &p, 12000, cudaEnableDefault, &q);
#else
const cudaError_t e = cudaGetDriverEntryPoint(name, &p, cudaEnableDefault, &q);
#endif
if (e != 0 || q != cudaDriverEntryPointSuccess || p == nullptr) {
/*
DPCT1010: SYCL uses exceptions to report errors and does not use the
error codes. The cudaGetLastError function call was replaced with 0. You
need to rewrite this code.
*/
(void)0;
return false;
}
f = reinterpret_cast<F>(p);
return true;
}
catch (sycl::exception const &exc) {
std::cerr << exc.what() << "Exception caught at file:" << __FILE__
<< ", line:" << __LINE__ << std::endl;
std::exit(1);
}
const Vmm& vmm() {
static const Vmm v = [] {
Vmm x;
x.ok = entry("cuDeviceGet", x.device_get) && entry("cuDeviceGetAttribute", x.attribute) &&
entry("cuMemGetAllocationGranularity", x.granularity) && entry("cuMemAddressReserve", x.reserve) &&
entry("cuMemAddressFree", x.address_free) && entry("cuMemCreate", x.create) &&
entry("cuMemRelease", x.release) && entry("cuMemMap", x.map) && entry("cuMemUnmap", x.unmap) &&
entry("cuMemSetAccess", x.set_access);
return x;
}();
return v;
}
dpct::experimental::mem_prop device_prop(int dev) {
dpct::experimental::mem_prop prop{};
prop.type = 0;
prop.location.type = 1;
prop.location.id = dev;
return prop;
}
// one segment: a physical allocation mapped at `va`, readable and writable by this device
bool map_segment(const Vmm &v, int dev, dpct::device_ptr va, size_t bytes,
unsigned long long &handle) {
const dpct::experimental::mem_prop prop = device_prop(dev);
dpct::experimental::physical_mem_ptr h = 0;
if (v.create(&h, bytes, &prop, 0) != 0) return false;
if (v.map(va, bytes, 0, h, 0) != 0) {
v.release(h);
return false;
}
dpct::experimental::mem_access_desc access{};
access.location.type = 1;
access.location.id = dev;
access.flags =
sycl::ext::oneapi::experimental::address_access_mode::read_write;
if (v.set_access(va, bytes, &access, 1) != 0) {
v.unmap(va, bytes);
v.release(h);
return false;
}
handle = (unsigned long long) h;
return true;
}
} // namespace
#endif
bool ExpertCache::open_segmented(uint64_t want, std::string &err) try {
#if 1 // SYCL port: --vram-elastic is CUDA-only (#533)
(void) want;
err = "ExpertCache: --vram-elastic (a segmented expert cache) is CUDA-only for now";
return false;
#else
const Vmm& v = vmm();
int dev = 0, supported = 0;
int cu = 0;
if (!v.ok || DPCT_CHECK_ERROR(dev = dpct::get_current_device_id()) != 0 ||
v.device_get(&cu, dev) != 0 ||
v.attribute(&supported,
CU_DEVICE_ATTRIBUTE_VIRTUAL_MEMORY_MANAGEMENT_SUPPORTED,
cu) != 0 ||
!supported) {
err = "ExpertCache: --vram-elastic needs the driver's virtual memory management, which this GPU or driver "
"does not offer";
return false;
}
const dpct::experimental::mem_prop prop = device_prop(dev);
size_t gran = 0;
if (v.granularity(
&gran, &prop,
sycl::ext::oneapi::experimental::granularity_mode::recommended) !=
0 ||
gran == 0) {
err = "ExpertCache: cannot read the driver's allocation granularity";
return false;
}
const uint64_t g = (uint64_t) gran;
const uint64_t total = (want + g - 1) / g * g;
seg_ = (int64_t) (((uint64_t) seg_req_ + g - 1) / g * g);
dpct::device_ptr va = 0;
if (v.reserve(&va, (size_t)total, 0, 0, 0) != 0) {
err = "ExpertCache: cannot reserve the address range of the segmented expert cache";
return false;
}
base_ = reinterpret_cast<uint8_t*>(va);
reserved_ = total;
for (uint64_t at = 0; at < total; at += (uint64_t) seg_) {
segs_.push_back(0);
seg_size_.push_back((int64_t) std::min<uint64_t>((uint64_t) seg_, total - at));
}
for (size_t i = 0; i < segs_.size(); ++i) {
if (!map_segment(v, dev,
va + (dpct::device_ptr)((uint64_t)i * (uint64_t)seg_),
(size_t)seg_size_[i], segs_[i])) {
char buf[200];
std::snprintf(buf, sizeof buf, "ExpertCache: cudaMalloc failed: segment %zu of %zu (%.2f GiB) of the "
"segmented cache could not be allocated", i + 1, segs_.size(),
(double) seg_size_[i] / 1073741824.0);
err = buf; // "cudaMalloc failed": the auto cache's smaller-retry path reads it as an allocation failure
release_segmented();
return false;
}
mapped_segs_ = (int64_t) i + 1;
}
return true;
#endif
}
catch (sycl::exception const &exc) {
std::cerr << exc.what() << "Exception caught at file:" << __FILE__
<< ", line:" << __LINE__ << std::endl;
std::exit(1);
}
void ExpertCache::release_segmented() {
#if 0 // SYCL port: no segmented arena (#533)
const Vmm& v = vmm();
if (base_ != nullptr) dpct::get_current_device().queues_wait_and_throw();
const dpct::device_ptr va = reinterpret_cast<dpct::device_ptr>(base_);
for (size_t i = 0; i < segs_.size(); ++i)
if (segs_[i] != 0) {
v.unmap(va + (dpct::device_ptr)((uint64_t)i * (uint64_t)seg_),
(size_t)seg_size_[i]);
v.release((dpct::experimental::physical_mem_ptr)segs_[i]);
}
if (base_ != nullptr && reserved_ > 0) v.address_free(va, (size_t) reserved_);
#endif
segs_.clear();
seg_size_.clear();
mapped_segs_ = 0;
reserved_ = 0;
base_ = nullptr;
}
int64_t ExpertCache::mapped_bytes() const {
if (segs_.empty()) return base_ != nullptr ? full_bytes() : 0;
int64_t b = 0;
for (int64_t i = 0; i < mapped_segs_; ++i) b += seg_size_[(size_t) i];
return b;
}
int64_t ExpertCache::slots_within(int64_t bytes) const {
if (bytes >= full_bytes()) return slots_;
if (bytes <= 0) return 0;
if (off_.empty()) return blob_ > 0 ? bytes / blob_ : 0;
// off_[i + 1] is slot i's end: count the slots whose end is at most `bytes`
return (int64_t) (std::upper_bound(off_.begin() + 1, off_.end(), (uint64_t) bytes) - (off_.begin() + 1));
}
bool ExpertCache::shrink(int64_t keep_bytes, std::string &err) try {
if (segs_.empty()) {
err = "the expert cache is not segmented (the engine needs --vram-elastic)";
return false;
}
#if 1 // SYCL port: --vram-elastic is CUDA-only (#533)
(void) keep_bytes;
return false;
#else
const Vmm& v = vmm();
int64_t keep = 0, at = 0; // the segments [0, keep) hold the first keep_bytes
while (keep < (int64_t) segs_.size() && at < keep_bytes) at += seg_size_[(size_t) keep++];
if (keep >= mapped_segs_) return true;
if (DPCT_CHECK_ERROR(dpct::get_current_device().queues_wait_and_throw()) !=
0) {
/*
DPCT1009: SYCL reports errors using exceptions and does not use
error codes. Please replace the "get_error_string_dummy(...)" with a
real error-handling function.
*/
/*
DPCT1010: SYCL uses exceptions to report errors and does not use the
error codes. The cudaGetLastError function call was replaced with 0. You
need to rewrite this code.
*/
err = std::string("the device failed before the cache shrank: ") +
dpct::get_error_string_dummy(0);
return false;
}
const dpct::device_ptr va = reinterpret_cast<dpct::device_ptr>(base_);
for (int64_t i = mapped_segs_ - 1; i >= keep; --i) {
const dpct::device_ptr p =
va + (dpct::device_ptr)((uint64_t)i * (uint64_t)seg_);
if (v.unmap(p, (size_t)seg_size_[(size_t)i]) != 0 ||
v.release((dpct::experimental::physical_mem_ptr)segs_[(size_t)i]) !=
0) {
err = "the driver refused to release an expert-cache segment";
live_slots_ = slots_within(mapped_bytes());
return false;
}
segs_[(size_t) i] = 0;
mapped_segs_ = i;
}
live_slots_ = slots_within(mapped_bytes());
return true;
#endif
}
catch (sycl::exception const &exc) {
std::cerr << exc.what() << "Exception caught at file:" << __FILE__
<< ", line:" << __LINE__ << std::endl;
std::exit(1);
}
bool ExpertCache::grow(int64_t want_bytes, std::string& err) {
if (segs_.empty()) {
err = "the expert cache is not segmented (the engine needs --vram-elastic)";
return false;
}
#if 1 // SYCL port: --vram-elastic is CUDA-only (#533)
(void) want_bytes;
return false;
#else
const Vmm& v = vmm();
int dev = 0;
dev = dpct::get_current_device_id();
const dpct::device_ptr va = reinterpret_cast<dpct::device_ptr>(base_);
int64_t at = mapped_bytes();
while (mapped_segs_ < (int64_t) segs_.size() && at + seg_size_[(size_t) mapped_segs_] <= want_bytes) {
const size_t i = (size_t) mapped_segs_;
if (!map_segment(v, dev,
va + (dpct::device_ptr)((uint64_t)i * (uint64_t)seg_),
(size_t)seg_size_[i], segs_[i])) {
/*
DPCT1010: SYCL uses exceptions to report errors and does not use
the error codes. The cudaGetLastError function call was replaced
with 0. You need to rewrite this code.
*/
(void)0;
err = "the driver has no VRAM for another expert-cache segment";
live_slots_ = slots_within(mapped_bytes());
return false;
}
at += seg_size_[i];
++mapped_segs_;
}
live_slots_ = slots_within(mapped_bytes());
return true;
#endif
}
#if defined(STRATA_USE_HIP)
bool ExpertCache::ensure_blocking_staging(std::size_t bytes, std::string& err) {
if (bytes <= blocking_staging_bytes_) return true;
@@ -201,8 +505,11 @@ bool ExpertCache::open(int64_t n_slots, int64_t n_layers, int64_t n_expert,
}
}
if (DPCT_CHECK_ERROR(base_ = (uint8_t *)sycl::malloc_device(
(size_t)want, dpct::get_in_order_queue())) != 0) {
if (seg_req_ > 0) { // #533: --vram-elastic: physical segments behind one address range (zeroed below)
if (!open_segmented(want, err)) return false;
} else if (DPCT_CHECK_ERROR(
base_ = (uint8_t *)sycl::malloc_device(
(size_t)want, dpct::get_in_order_queue())) != 0) {
base_ = nullptr;
char buf[256];
std::snprintf(
@@ -233,6 +540,7 @@ bool ExpertCache::open(int64_t n_slots, int64_t n_layers, int64_t n_expert,
residency_.assign((size_t) (n_layers * n_expert), kNotResident);
slots_ = n_slots;
live_slots_ = n_slots;
n_layers_ = n_layers;
n_expert_ = n_expert;
blob_ = blob_bytes;
@@ -274,6 +582,7 @@ bool ExpertCache::open_sized(const std::vector<int64_t>& slot_bytes, int64_t n_l
// one allocation of the summed size, through the uniform path's checks: n "slots" of 1 byte
if (!open((int64_t) off.back(), n_layers, n_expert, 1, err)) return false;
slots_ = (int64_t) slot_bytes.size();
live_slots_ = slots_;
blob_ = mx;
off_ = std::move(off);
#if defined(STRATA_USE_HIP)
@@ -300,12 +609,15 @@ void ExpertCache::close() {
blocking_staging_bytes_ = 0;
#endif
off_.clear();
if (base_ != nullptr) {
if (!segs_.empty()) {
release_segmented();
} else if (base_ != nullptr) {
sycl::free(base_, dpct::get_in_order_queue());
base_ = nullptr;
}
residency_.clear();
slots_ = 0;
live_slots_ = 0;
n_layers_ = 0;
n_expert_ = 0;
blob_ = 0;
+277 -51
View File
@@ -219,18 +219,25 @@ bool read_cgroup_memory_stat(const std::filesystem::path& path, uint64_t current
}
#endif
bool available_memory_bytes(uint64_t& bytes) {
} // namespace
namespace detail {
bool host_available_memory(HostMemory& m, const std::string& meminfo, const std::string& self_cgroup,
const std::string& cgroup_root) {
m = HostMemory{};
#if defined(_WIN32)
(void) meminfo; (void) self_cgroup; (void) cgroup_root;
MEMORYSTATUSEX status{};
status.dwLength = sizeof(status);
if (!GlobalMemoryStatusEx(&status)) return false;
bytes = (uint64_t) status.ullAvailPhys;
return bytes > 0;
m.available = (uint64_t) status.ullAvailPhys;
return m.available > 0;
#elif defined(__linux__)
// MemAvailable includes reclaimable page cache, unlike _SC_AVPHYS_PAGES.
std::ifstream info("/proc/meminfo");
std::ifstream info(meminfo);
std::string line;
bytes = 0;
uint64_t bytes = 0;
while (std::getline(info, line)) {
std::istringstream fields(line);
std::string key, unit;
@@ -239,17 +246,26 @@ bool available_memory_bytes(uint64_t& bytes) {
value <= std::numeric_limits<uint64_t>::max() / 1024) bytes = value * 1024;
}
if (bytes == 0) return false;
// Account for the tightest cgroup-v2 ancestor limit when its normal mount is visible.
// Account for the tightest cgroup ancestor limit when its normal mount is visible.
// This is a point-in-time guard, not a reservation against concurrent allocations.
std::ifstream groups("/proc/self/cgroup");
std::ifstream groups(self_cgroup);
if (!groups) return false;
bool resolved_v2 = false;
const std::filesystem::path root(cgroup_root);
bool v2 = false;
std::string v1_path; // the memory controller's group (cgroup v1), when there is no v2 line
while (std::getline(groups, line)) {
if (line.rfind("0::/", 0) != 0) continue;
const std::filesystem::path root("/sys/fs/cgroup");
if (line.rfind("0::/", 0) != 0) {
// v1: "N:controller[,controller]:/path"
const size_t c1 = line.find(':'), c2 = c1 == std::string::npos ? c1 : line.find(':', c1 + 1);
if (c2 == std::string::npos) continue;
std::istringstream ctl(line.substr(c1 + 1, c2 - c1 - 1));
for (std::string c; std::getline(ctl, c, ',');)
if (c == "memory") v1_path = line.substr(c2 + 1);
continue;
}
auto path = (root / line.substr(4)).lexically_normal();
if (path.string().rfind(root.string(), 0) != 0 || !std::filesystem::is_directory(path)) return false;
resolved_v2 = true;
v2 = true;
while (path.string().rfind(root.string(), 0) == 0) {
std::ifstream limit_file(path / "memory.max"), current_file(path / "memory.current");
std::string limit;
@@ -263,28 +279,59 @@ bool available_memory_bytes(uint64_t& bytes) {
size_t consumed = 0;
const uint64_t cap = std::stoull(limit, &consumed);
if (consumed != limit.size()) return false;
detail::CgroupMemoryStat stat;
CgroupMemoryStat stat;
if (!read_cgroup_memory_stat(path, current, stat)) return false;
uint64_t cgroup_available = 0;
if (!detail::cgroup_available_bytes(cap, stat, cgroup_available)) return false;
if (!cgroup_available_bytes(cap, stat, cgroup_available)) return false;
bytes = std::min(bytes, cgroup_available);
m.cgroup_limit = std::min(m.cgroup_limit, cap);
} catch (...) { return false; }
}
if (path == root) break;
path = path.parent_path();
}
}
return resolved_v2;
if (!v2 && !v1_path.empty()) {
// #633: cgroup v1 (older Docker hosts, RHEL 7/8): the memory controller's group and its ancestors. An
// unlimited group says a number near 2^63 (rounded to its page size); a group whose files are not
// visible (no mount in this namespace) is skipped - MemAvailable alone, as with no cgroup at all.
const std::filesystem::path mroot = root / "memory";
auto path = (mroot / v1_path.substr(v1_path.rfind('/', 0) == 0 ? 1 : 0)).lexically_normal();
while (path.string().rfind(mroot.string(), 0) == 0) {
std::ifstream limit_file(path / "memory.limit_in_bytes"), usage_file(path / "memory.usage_in_bytes");
uint64_t cap = 0, usage = 0;
if (limit_file >> cap && usage_file >> usage && cap < (1ull << 62)) {
bytes = std::min(bytes, usage < cap ? cap - usage : 0);
m.cgroup_limit = std::min(m.cgroup_limit, cap);
}
if (path == mroot) break;
path = path.parent_path();
}
}
m.available = bytes;
return true;
#else
(void) meminfo; (void) self_cgroup; (void) cgroup_root;
const long pages = sysconf(_SC_AVPHYS_PAGES);
const long page_bytes = sysconf(_SC_PAGESIZE);
if (pages <= 0 || page_bytes <= 0 ||
(uint64_t) pages > std::numeric_limits<uint64_t>::max() / (uint64_t) page_bytes) return false;
bytes = (uint64_t) pages * (uint64_t) page_bytes;
return bytes > 0;
m.available = (uint64_t) pages * (uint64_t) page_bytes;
return m.available > 0;
#endif
}
} // namespace detail
namespace {
bool available_memory_bytes(uint64_t& bytes) {
detail::HostMemory m;
if (!detail::host_available_memory(m)) return false;
bytes = m.available;
return bytes > 0;
}
} // namespace
// ================================ THE FILE-BACKED SOURCE ================================
@@ -882,13 +929,14 @@ void FileExpertSource::fill_many(const std::vector<Fill>& todo) {
for (auto& t : th) t.join();
}
bool FileExpertSource::set_unbuffered(uint64_t ram_bytes, std::string& why) {
uint64_t FileExpertSource::expert_bytes() const {
uint64_t total = 0;
for (uint64_t b : layer_blob_bytes_) total += b * (uint64_t) n_expert_;
return total;
}
bool FileExpertSource::open_direct(std::string& why) {
#if defined(_WIN32)
if (base_ == nullptr || paths_.empty() || !direct_.empty()) {
why = !direct_.empty() ? "already unbuffered" : "no expert files open";
return !direct_.empty();
}
if (!experts_unbuffered(paths_, ram_bytes, why, /*cache_counts=*/false)) return false;
for (const std::string& path : paths_) {
const int wide = MultiByteToWideChar(CP_UTF8, 0, path.c_str(), -1, nullptr, 0);
std::vector<wchar_t> w((size_t) (wide > 0 ? wide : 1), L'\0');
@@ -909,6 +957,24 @@ bool FileExpertSource::set_unbuffered(uint64_t ram_bytes, std::string& why) {
for (uint64_t b : layer_blob_bytes_) stage_blob_ = std::max(stage_blob_, b);
}
return true;
#else
(void) why;
return false;
#endif
}
bool FileExpertSource::set_unbuffered(uint64_t ram_bytes, std::string& why) {
#if defined(_WIN32)
if (base_ == nullptr || paths_.empty() || !direct_.empty()) {
why = !direct_.empty() ? "already unbuffered" : "no expert files open";
return !direct_.empty();
}
// #577: the file tier reads the experts outside the RAM copy, not every byte of the shards (dense weights, the
// PLE table), and the copy holds at most every expert - the budget asked for can be more than that
const uint64_t experts = expert_bytes();
const uint64_t arena = std::min(ram_bytes, experts);
if (!experts_unbuffered(paths_, arena, why, /*cache_counts=*/false, experts - arena)) return false;
return open_direct(why);
#else
(void) ram_bytes;
why = "through the file cache (not Windows)";
@@ -916,6 +982,39 @@ bool FileExpertSource::set_unbuffered(uint64_t ram_bytes, std::string& why) {
#endif
}
bool FileExpertSource::recheck_unbuffered(std::string& why) {
#if defined(_WIN32)
if (const char* env = std::getenv("STRATA_UNBUFFERED_LOAD"); env != nullptr && env[0] != '\0') {
why = std::string("STRATA_UNBUFFERED_LOAD=") + env;
return !direct_.empty();
}
if (base_ == nullptr || paths_.empty()) {
why = "no expert files open";
return false;
}
// #577: the RAM copy is built (and already out of the available RAM), so what the file cache would have to keep
// is exactly the experts outside it - the GPU cache's (refilled after a prompt borrowed their slots) and the
// ones neither holds
const uint64_t experts = expert_bytes();
const uint64_t read = experts > complement_bytes_ ? experts - complement_bytes_ : 0;
std::string w;
const bool ub = experts_unbuffered(paths_, 0, w, /*cache_counts=*/false, read);
char head[96];
std::snprintf(head, sizeof head, "re-checked with the RAM copy built (%.2f GiB): ",
(double) complement_bytes_ / 1073741824.0);
why = head + w;
if (ub == !direct_.empty()) return ub;
if (ub) return open_direct(why);
// through the file cache after all: the mapped reads take over (staged() stays true for the GGUF in place)
for (void* d : direct_) CloseHandle((HANDLE) d);
direct_.clear();
return false;
#else
why = "through the file cache (not Windows)";
return false;
#endif
}
bool FileExpertSource::read_direct(const Fill* fills, size_t n) const {
#if defined(_WIN32)
// NTFS runs the unbuffered reads of a file one at a time while the file is mapped or cached anywhere (the engine
@@ -1241,9 +1340,9 @@ bool FileExpertSource::pin_cache_complement(
err = std::string("FileExpertSource: GPU expert cache is not ready: ") +
dpct::get_error_string_dummy(sync);
/*
DPCT1010: SYCL uses exceptions to report errors and does not use the
error codes. The cudaGetLastError function call was replaced with 0. You
need to rewrite this code.
DPCT1010: SYCL uses exceptions to report errors and does not use
the error codes. The cudaGetLastError function call was replaced with 0.
You need to rewrite this code.
*/
(void)0;
return false;
@@ -1437,9 +1536,9 @@ bool FileExpertSource::pin_cache_complement(
note = std::string("no device alias (") +
dpct::get_error_string_dummy(aliased) + ")";
/*
DPCT1010: SYCL uses exceptions to report errors and does
not use the error codes. The cudaGetLastError function call
was replaced with 0. You need to rewrite this code.
DPCT1010: SYCL uses exceptions to report errors and
does not use the error codes. The cudaGetLastError function
call was replaced with 0. You need to rewrite this code.
*/
(void)0;
(void)DPCT_CHECK_ERROR(
@@ -1458,8 +1557,8 @@ bool FileExpertSource::pin_cache_complement(
note = std::string("page-locking refused (") +
dpct::get_error_string_dummy(allocated) + ")";
/*
DPCT1010: SYCL uses exceptions to report errors and does not
use the error codes. The cudaGetLastError function call was
DPCT1010: SYCL uses exceptions to report errors and does
not use the error codes. The cudaGetLastError function call was
replaced with 0. You need to rewrite this code.
*/
(void)0;
@@ -1524,11 +1623,11 @@ bool FileExpertSource::pin_cache_complement(
*/
(void)0;
/*
DPCT1027: The call to cudaHostUnregister was
replaced with 0 because SYCL currently does not
support registering of existing host memory for
use by device. Use USM to allocate memory for
use by host and device.
DPCT1027: The call to cudaHostUnregister
was replaced with 0 because SYCL currently does
not support registering of existing host memory
for use by device. Use USM to allocate memory
for use by host and device.
*/
(void)0;
}
@@ -1733,9 +1832,9 @@ bool FileExpertSource::reserve_exchanges(int64_t n, std::string &err) try {
xstage_pinned_ = true;
} else {
/*
DPCT1010: SYCL uses exceptions to report errors and does not use the
error codes. The cudaGetLastError function call was replaced with 0. You
need to rewrite this code.
DPCT1010: SYCL uses exceptions to report errors and does not use
the error codes. The cudaGetLastError function call was replaced with 0.
You need to rewrite this code.
*/
(void)0;
p = std::malloc(total);
@@ -2152,7 +2251,8 @@ void expert_pool_dispatch_multi(ExpertDispatch& d, const float* x_f, const int32
}
if (kind[i] >= 0) { // CUDA0, PCIe, or a remote/peer result staged into this row below
if (kind[i] == 0) ++d.cache_hits;
else if (kind[i] == 2 && d.peer != nullptr) ++d.peer_entries;
else ++d.offload_entries; // #588: PCIe or another GPU
if (kind[i] == 2 && d.peer != nullptr) ++d.peer_entries;
// multi-GPU: a direct peer launch is writing this row right now - zeroing it would race it
if (!(kind[i] == 2 && d.peer != nullptr && d.peer->launched_direct()))
std::memset(row, 0, (size_t) H * sizeof(float));
@@ -2686,10 +2786,9 @@ LoadStats load_experts_gguf(const std::string& gguf, uint8_t* dst, const strata:
ArenaExpertSource::~ArenaExpertSource() { close(); }
bool ArenaExpertSource::open(const std::string &pack_dir, int64_t n_layers,
int64_t n_expert, int threads, std::string &err,
uint64_t max_pinned_bytes,
const std::string &shared_arena_file) try {
bool ArenaExpertSource::open(const std::string& pack_dir, int64_t n_layers, int64_t n_expert, int threads,
std::string& err, uint64_t max_pinned_bytes,
const std::string& shared_arena_file) {
close();
const std::string path = pack_dir + "/experts.bin";
// plan v0.3 P6: the layout (canonical, or a native pack's per-layer blobs) was loaded by the driver
@@ -2714,7 +2813,7 @@ bool ArenaExpertSource::open(const std::string &pack_dir, int64_t n_layers,
std::ifstream f(path, std::ios::binary | std::ios::ate);
if (!f) { err = "ArenaExpertSource: cannot open " + path; return false; }
const uint64_t got = (uint64_t) f.tellg();
if (got != want) {
if (got != want && got != want + (uint64_t) blob) { // + one blob: STRATA_ARENA_MMAP's padded file
char buf[400];
std::snprintf(buf, sizeof buf,
"ArenaExpertSource: %s is %llu B but %lld layers x %lld experts (blobs up to %lld B) "
@@ -2726,6 +2825,76 @@ bool ArenaExpertSource::open(const std::string &pack_dir, int64_t n_layers,
}
}
// STRATA_ARENA_MMAP=1 (a small-RAM machine whose GPUs hold most experts): the arena is the pack's experts.bin
// mapped READ-ONLY - not locked, not pinned. The page cache keeps what the CPU pool and the prompt path
// actually read (the experts no VRAM cache holds) and gives back the rest under memory pressure; a page
// dropped is re-read from the file, so a result never depends on what is resident. The GPUs then get no
// mapped alias: run with --pcie-frac 0. The first start writes experts.bin (arena layout, padded by one blob
// so a whole-slot copy may start at any expert), later ones map it.
#if defined(_WIN32)
static const bool arena_mmap = false; // POSIX mmap/madvise: Linux only for now
#else
static const bool arena_mmap = [] { const char* v = std::getenv("STRATA_ARENA_MMAP"); return v && v[0] == '1'; }();
if (arena_mmap) {
const uint64_t file_bytes = want + (uint64_t) blob;
uint64_t have = 0;
{
std::ifstream f(path, std::ios::binary | std::ios::ate);
if (f) have = (uint64_t) f.tellg();
}
if (have == file_bytes) {
const int fd = ::open(path.c_str(), O_RDONLY);
void* v = fd >= 0 ? mmap(nullptr, (size_t) file_bytes, PROT_READ, MAP_SHARED, fd, 0) : MAP_FAILED;
if (fd >= 0) ::close(fd);
if (v == MAP_FAILED) { err = "ArenaExpertSource: mmap of " + path + " failed"; return false; }
map_ = v;
map_bytes_ = file_bytes;
base_ = (const uint8_t*) v;
pinned_bytes_ = 0;
dev_slice_.clear();
slice_bytes_ = 0;
blobs_ = n_layers * n_expert;
n_expert_ = n_expert;
reads_ = 0;
note_ = "mapped read-only from " + path + " (STRATA_ARENA_MMAP: not locked, not pinned)";
gib_per_s_ = 0.0;
load_seconds_ = load_read_s_ = load_copy_s_ = 0.0;
return true;
}
}
#endif
// #633: THE RAM BEFORE THE ALLOCATION. On Linux the arena is an anonymous mapping that succeeds whatever the host
// has; its pages are committed as the load writes them, so a container whose memory limit is below the arena was
// killed by the OOM killer part-way through the load, without a message. A hard cgroup limit below it is
// certain to end that way: refused, with the numbers. Less RAM available than the arena (another program, an
// engine still exiting) is only a warning - the OS may make room - per the recommend-not-force rule.
ram_warning_.clear();
// (a shared arena, --shared-expert-arena, may already be in RAM for another engine: not checked)
if (detail::HostMemory hm; shared_arena_file.empty() && detail::host_available_memory(hm)) {
const uint64_t need = want + (uint64_t) blob;
const double gib = 1073741824.0;
char buf[512];
if (hm.cgroup_limit < need) {
std::snprintf(buf, sizeof buf,
"ArenaExpertSource: the expert arena needs %.2f GiB of RAM but this process's memory limit "
"(cgroup memory.max / memory.limit_in_bytes) is %.2f GiB: it would be killed while loading. "
"Raise the container's limit, or run with less RAM: --mmap-experts with "
"--resident-budget-gib N keeps only the hottest experts in RAM",
(double) need / gib, (double) hm.cgroup_limit / gib);
err = buf;
return false;
}
if (hm.available < need) {
std::snprintf(buf, sizeof buf,
"the expert arena needs %.2f GiB of RAM but %.2f GiB is available (%.2f GiB short): the load "
"may swap or be stopped by the OS. Close other programs (or wait for an engine that is "
"still exiting), or run with less RAM: --mmap-experts with --resident-budget-gib N",
(double) need / gib, (double) hm.available / gib, (double) (need - hm.available) / gib);
ram_warning_ = buf;
}
}
uint64_t pack_hash = 0;
if (!shared_arena_file.empty() &&
!shared_arena_pack_hash(pack_dir, path, gguf_, lay, pack_hash, err)) return false;
@@ -2784,6 +2953,30 @@ bool ArenaExpertSource::open(const std::string &pack_dir, int64_t n_layers,
err = "ArenaExpertSource: the load read " + std::to_string(st.bytes) + " B of " + std::to_string(want);
return false;
}
// the first start: write experts.bin for the mapped starts after this one - only when the drive has room for it
// and 2 GiB more (a full drive fails other writes too); otherwise this start says so and runs pinned as before
std::error_code space_ec;
const uint64_t free_disk = arena_mmap && from_gguf ? (uint64_t) std::filesystem::space(
std::filesystem::path(path).parent_path(), space_ec).available : 0;
const bool room = !space_ec && free_disk >= want + (uint64_t) blob + (2ull << 30);
if (arena_mmap && from_gguf && !room)
std::fprintf(stderr, "strata generate: STRATA_ARENA_MMAP: NOT writing %s - %.1f GiB free on that drive, it needs "
"%.1f GiB plus 2 GiB to spare; this start keeps the arena in RAM\n", path.c_str(),
(double) free_disk / 1073741824.0, (double) (want + (uint64_t) blob) / 1073741824.0);
if (arena_mmap && from_gguf && room) {
const std::string tmp = path + ".tmp";
std::FILE* f = std::fopen(tmp.c_str(), "wb");
bool ok = f != nullptr && std::fwrite(a->data(), 1, (size_t) want, f) == (size_t) want;
if (ok) {
std::vector<uint8_t> pad((size_t) blob, 0);
ok = std::fwrite(pad.data(), 1, pad.size(), f) == pad.size();
}
if (f) ok = std::fclose(f) == 0 && ok;
if (ok) ok = std::rename(tmp.c_str(), path.c_str()) == 0;
if (!ok) std::remove(tmp.c_str());
std::fprintf(stderr, "strata generate: STRATA_ARENA_MMAP: %s %s (the next start maps it)\n",
ok ? "wrote" : "could NOT write", path.c_str());
}
arena_ = a;
base_ = a->data();
pinned_bytes_ = a->registered_bytes;
@@ -2796,8 +2989,8 @@ bool ArenaExpertSource::open(const std::string &pack_dir, int64_t n_layers,
void* d = nullptr;
if (DPCT_CHECK_ERROR(d = (void *)(void *)(base_ + off)) != 0) {
/*
DPCT1010: SYCL uses exceptions to report errors and does not
use the error codes. The cudaGetLastError function call was
DPCT1010: SYCL uses exceptions to report errors and does
not use the error codes. The cudaGetLastError function call was
replaced with 0. You need to rewrite this code.
*/
(void)0;
@@ -2817,13 +3010,13 @@ bool ArenaExpertSource::open(const std::string &pack_dir, int64_t n_layers,
load_copy_s_ = st.copy_seconds;
return true;
}
catch (sycl::exception const &exc) {
std::cerr << exc.what() << "Exception caught at file:" << __FILE__
<< ", line:" << __LINE__ << std::endl;
std::exit(1);
}
void ArenaExpertSource::close() {
#if !defined(_WIN32)
if (map_ != nullptr) munmap(map_, (size_t) map_bytes_);
#endif
map_ = nullptr;
map_bytes_ = 0;
if (arena_ != nullptr) {
delete (PinnedArena*) arena_;
arena_ = nullptr;
@@ -2833,6 +3026,39 @@ void ArenaExpertSource::close() {
n_expert_ = 0;
}
void ArenaExpertSource::prefetch(int64_t layer, int64_t expert) {
#if defined(_WIN32)
(void) layer; (void) expert;
#else
if (map_ == nullptr || layer < 0 || expert < 0 || expert >= n_expert_) return;
const auto& lay = strata::kernels::cpu::expert_layout();
const uint64_t off = lay.blob_offset(layer, expert) & ~(uint64_t) 4095;
const uint64_t end = lay.blob_offset(layer, expert) + lay.blob_bytes(layer);
madvise((uint8_t*) map_ + off, (size_t) (end - off), MADV_WILLNEED);
#endif
}
// Only the pages wholly inside the blob: a page shared with a neighbour the CPU may still read stays. MADV_DONTNEED
// unmaps them from this process; they stay in the page cache as clean, unmapped pages - the first the kernel takes
// back under pressure, and a minor fault away when the CPU needs the expert again (an adaptive swap evicts it).
// Dropping them from the cache too (POSIX_FADV_DONTNEED) made every eviction a re-read from the disk: ~200 major
// faults a second in decode, +8 ms a window. The memory the arena used to hold was never the cache itself but
// ROCclr's pin-in-place locks on it (see main) - locked pages are the ones the kernel cannot take back.
uint64_t ArenaExpertSource::release(int64_t layer, int64_t expert) {
#if defined(_WIN32)
(void) layer; (void) expert;
return 0;
#else
if (map_ == nullptr || layer < 0 || expert < 0 || expert >= n_expert_) return 0;
const auto& lay = strata::kernels::cpu::expert_layout();
const uint64_t off = (lay.blob_offset(layer, expert) + 4095) & ~(uint64_t) 4095;
const uint64_t end = (lay.blob_offset(layer, expert) + lay.blob_bytes(layer)) & ~(uint64_t) 4095;
if (end <= off) return 0;
if (madvise((uint8_t*) map_ + off, (size_t) (end - off), MADV_DONTNEED) != 0) return 0;
return end - off;
#endif
}
bool ArenaExpertSource::pinned(int64_t layer, int64_t expert) const {
if (base_ == nullptr || layer < 0 || expert < 0 || expert >= n_expert_) return false;
const auto& lay = strata::kernels::cpu::expert_layout();
+9
View File
@@ -8,6 +8,7 @@
#include <algorithm>
#include <climits>
#include <cstdlib>
#include <exception>
#include <limits>
#include <memory>
@@ -15,6 +16,12 @@
namespace strata::core {
namespace {
int g_layer_lb = -1, g_layer_le = -1; // set_layer_range; -1: every layer
bool in_range(const std::string& name) {
if (g_layer_lb < 0 || name.rfind("blk.", 0) != 0) return true;
const int l = std::atoi(name.c_str() + 4);
return l >= g_layer_lb && l < g_layer_le;
}
bool eligible(const strata::TensorInfo& tensor, bool include_ple_key) {
const auto& name = tensor.name;
if (name.rfind("blk.", 0) != 0) return false;
@@ -60,6 +67,7 @@ bool NativeDense::served_names(const std::vector<std::string>& shards, bool incl
}
}
void NativeDense::set_layer_range(int lb, int le) { g_layer_lb = lb; g_layer_le = le; }
bool NativeDense::keep_unquantized_ple_key(const std::string& pack_dir, std::set<std::string>& skip,
std::string& err) {
const std::string key = "blk.1.ple_key.weight";
@@ -148,6 +156,7 @@ bool NativeDense::load(const std::vector<std::string> &shards,
}
for (const auto& tensor : gguf.tensors()) {
if (!eligible(tensor, include_ple_key)) continue;
if (!in_range(tensor.name) && tensor.name.find("ple") == std::string::npos) continue;
if (!seen.insert(tensor.name).second) {
err = "native dense: duplicate tensor " + tensor.name; return false;
}
+13 -7
View File
@@ -571,7 +571,7 @@ LoadStats load_experts_direct(const std::string& path, uint8_t* dst, const std::
}
bool experts_unbuffered(const std::vector<std::string>& files, uint64_t arena_bytes, std::string& why,
bool cache_counts) {
bool cache_counts, uint64_t read_bytes) {
const char* env = std::getenv("STRATA_UNBUFFERED_LOAD");
if (env != nullptr && env[0] != '\0') {
why = std::string("STRATA_UNBUFFERED_LOAD=") + env;
@@ -619,15 +619,21 @@ bool experts_unbuffered(const std::vector<std::string>& files, uint64_t arena_by
GlobalMemoryStatusEx(&ms);
const uint64_t avail = ms.ullAvailPhys;
// what the cache could keep beside the arena (~4 GiB for everything else)
const uint64_t room = avail > arena_bytes + (4ull << 30) ? avail - arena_bytes - (4ull << 30) : 0;
const bool keepable = room >= total_bytes;
char msg[200];
std::snprintf(msg, sizeof msg, "%d of %d probe reads from the file cache; %.1f GiB available, %.1f GiB of files",
fast, n, (double) avail / (1ull << 30), (double) total_bytes / (1ull << 30));
const uint64_t need = read_bytes == kAllFileBytes ? total_bytes : read_bytes;
const bool keepable = strata::platform::file_cache_keeps(avail, arena_bytes, need);
char msg[256];
if (read_bytes == kAllFileBytes)
std::snprintf(msg, sizeof msg, "%d of %d probe reads from the file cache; %.1f GiB available, %.1f GiB of files",
fast, n, (double) avail / (1ull << 30), (double) total_bytes / (1ull << 30));
else
std::snprintf(msg, sizeof msg, "%.1f GiB available, %.1f GiB of it still to be taken by the RAM copy, %.1f GiB "
"of experts read from the files: the file cache %s keep them",
(double) avail / (1ull << 30), (double) arena_bytes / (1ull << 30), (double) need / (1ull << 30),
keepable ? "can" : "cannot");
why = msg;
return (!cached || !cache_counts) && !keepable;
#else
(void) files; (void) arena_bytes; (void) cache_counts;
(void) files; (void) arena_bytes; (void) cache_counts; (void) read_bytes;
why = "buffered (not Windows)";
return false;
#endif
+134 -1
View File
@@ -71,6 +71,19 @@ struct Bump {
}
};
// #649 (HIP, opt-in A/Bs for the gfx1030 verify timeouts; CUDA never reads them)
// STRATA_VERIFY_COHERENT=1 the handshake words and rows in explicitly coherent (fine-grained) host memory
// STRATA_DOORBELL_STORE=1 the GPU stores each step's ring instead of read-modify-writing it over PCIe
bool env_on(const char* name) {
const char* e = std::getenv(name);
return e != nullptr && e[0] != 0 && e[0] != '0';
}
#if defined(STRATA_USE_HIP)
const bool g_coherent = env_on("STRATA_VERIFY_COHERENT");
const bool g_doorbell_store = env_on("STRATA_DOORBELL_STORE");
#endif
const bool g_trace = env_on("STRATA_VERIFY_TRACE");
bool mapped(size_t bytes, void **h, void **d) try {
/*
DPCT1048: The original value cudaHostAllocMapped is not meaningful in the
@@ -136,6 +149,20 @@ void release_live_verifiers(std::FILE* f) {
}
if (f != nullptr) std::fflush(f);
}
// #649: the host side of STRATA_VERIFY_TRACE - a ring of the last kTraceN handshake events of every verifier
struct TraceEv {
int64_t t_ns;
const void* who;
int64_t window, step, layer, aux;
uint32_t seq, flag, a, b;
const char* what;
};
constexpr uint64_t kTraceN = 4096;
TraceEv g_trace_ring[kTraceN];
std::atomic<uint64_t> g_trace_next{0};
int64_t trace_now_ns() {
return std::chrono::duration_cast<std::chrono::nanoseconds>(Clock::now().time_since_epoch()).count();
}
std::string released_note(bool drained) {
return drained ? "; its GPU waits were released and the GPU finished (#267)"
: "; its GPU waits were released but the GPU did not finish within 5 s (#267)";
@@ -150,6 +177,7 @@ const int64_t g_test_stall = [] {
bool Verifier::release_gpu_waits(int timeout_ms) try {
released_.store(true);
trace_ev("RELEASE", -1, -1, timeout_ms);
// the words the spin kernels read (wait_flag_ge, wait_flag_ge_or) are mapped host memory, so a store here
// reaches them with no API call; UINT32_MAX is past every ring. (E-6's skip words are device memory, but
// wait_flag_ge_or also returns on its flag.) A host function raising flag B later only raises.
@@ -163,15 +191,78 @@ bool Verifier::release_gpu_waits(int timeout_ms) try {
for (dpct::queue_ptr q : {cs_, copy_}) {
if (q == nullptr) continue;
while (!q->ext_oneapi_empty()) {
if (ms_since(t0) > timeout_ms) return false;
if (ms_since(t0) > timeout_ms) {
trace_ev("RELEASE-NOT-DRAINED", -1, -1, (int64_t) ms_since(t0));
return false;
}
std::this_thread::sleep_for(std::chrono::milliseconds(1));
}
}
trace_ev("RELEASE-DRAINED", -1, -1, (int64_t) ms_since(t0)); // aux: ms the GPU took to finish once released
return true;
} catch (...) {
return false; // the runtime may already be tearing down (the watchdog's path ends the engine)
}
void Verifier::trace_ev(const char* what, int64_t step, int64_t layer, int64_t aux) const {
if (!g_trace) return;
auto rd = [](const uint32_t* p) { return p ? *(const volatile uint32_t*) p : 0u; };
TraceEv& e = g_trace_ring[g_trace_next.fetch_add(1) % kTraceN];
e = {trace_now_ns(), this, windows, step, layer, aux, rd(h_seq_), rd(h_flag_), rd(h_flagA_), rd(h_flagB_), what};
}
void Verifier::trace_dump(std::FILE* f) const {
if (!g_trace || f == nullptr) return;
static const char* names[kProfPer] = {"pre", "hc-read0", "qkv gemv", "conv", "ab", "z", "rec", "kv-idx",
"k/v rope", "kv append", "q", "scores+topk", "kv-resolve", "attention",
"gate", "", "out-proj", "router+ring", "shared", "waitA", "VRAM hits", "waitB",
"PCIe grp", "waitCPU", "combine", "", "head", "", "", "", "", "", ""};
const uint64_t end = g_trace_next.load();
const uint64_t n = end < 160 ? end : 160;
std::fprintf(f, "strata verify trace (#649): the last %llu handshake events (host clock, us before the last; "
"words: seq = GPU rang, flag = CPU served, A = plan, B = copies)\n", (unsigned long long) n);
const int64_t t_last = n ? g_trace_ring[(end - 1) % kTraceN].t_ns : 0;
for (uint64_t i = end - n; i < end; ++i) {
const TraceEv& e = g_trace_ring[i % kTraceN];
if (e.what == nullptr) continue;
std::fprintf(f, " %10.1f %-20s v%p window %lld step %lld layer %lld aux %lld | seq %u flag %u A %u B %u\n",
(double) (t_last - e.t_ns) / 1000.0, e.what, e.who, (long long) e.window, (long long) e.step,
(long long) e.layer, (long long) e.aux, e.seq, e.flag, e.a, e.b);
}
if (trace_h_ == nullptr || g_ == nullptr) return;
// the breadcrumbs: every stamp point the GPU passed in this window (ns of the GPU clock after the window's first)
unsigned long long t0 = ~0ull;
for (size_t i = 0; i < trace_n_; ++i) {
const unsigned long long t = *(const volatile unsigned long long*) (trace_h_ + i);
if (t != 0 && t < t0) t0 = t;
}
if (t0 == ~0ull) {
std::fprintf(f, "strata verify trace: no GPU breadcrumb in this window (the GPU never started it)\n");
return;
}
int64_t last_l = -1;
for (int64_t l = 0; l <= g_->n_layers; ++l)
for (int i = 0; i < kProfPer; ++i)
for (int grp = 0; grp < 2; ++grp)
if (trace_h_[(size_t) ((l * kProfPer + i) * 2 + grp)] != 0) last_l = l;
std::fprintf(f, "strata verify trace: GPU breadcrumbs (us after the window's first), the GPU got as far as layer "
"%lld:\n", (long long) last_l);
for (int64_t l = std::max<int64_t>(0, last_l - 2); l <= std::min<int64_t>(g_->n_layers, last_l + 1); ++l)
for (int grp = 0; grp < 2; ++grp) {
std::string line;
char b[64];
for (int i = 0; i < kProfPer; ++i) {
const unsigned long long t = trace_h_[(size_t) ((l * kProfPer + i) * 2 + grp)];
if (t == 0) continue;
std::snprintf(b, sizeof b, " %s(%d)=%.1f", names[i], i, (double) (t - t0) / 1000.0);
line += b;
}
if (!line.empty())
std::fprintf(f, " layer %lld group %d:%s\n", (long long) l, grp, line.c_str());
}
std::fflush(f);
}
void Verifier::diag(std::FILE* f) const {
auto rd = [](const uint32_t* p) { return p ? *(const volatile uint32_t*) p : 0u; };
// #251: outside a verify stage these are the LAST window's numbers (it finished), not the stalled work's
@@ -180,6 +271,7 @@ void Verifier::diag(std::FILE* f) const {
std::fprintf(f, " verify window%s: %d tokens at position %lld, host at layer step %u; the GPU rang %u; flags: "
"served %u, plan (A) %u, copies (B) %u\n", current ? "" : " (last window, not the current stage)",
last_t_, (long long) last_pos0_, cur_layer_ + 1, rd(h_seq_), rd(h_flag_), rd(h_flagA_), rd(h_flagB_));
trace_dump(f); // #649: STRATA_VERIFY_TRACE=1 only
}
Verifier::~Verifier() try {
@@ -357,6 +449,22 @@ bool Verifier::init(const WeightTable &wt, const ModelGeometry &g,
return false;
}
dpct::get_in_order_queue().memset(arena_, 0, count.used).wait();
if (g_trace && trace_h_ == nullptr) { // #649: the breadcrumbs, mapped so they read while the GPU hangs
trace_n_ = (size_t) (g.n_layers + 1) * kProfPer * 2;
if (!mapped(trace_n_ * 8, (void**) &trace_h_, (void**) &trace_m_)) {
trace_h_ = trace_m_ = nullptr;
trace_n_ = 0;
}
#if defined(STRATA_USE_HIP)
std::fprintf(stderr, "strata verify trace (#649): on; coherent words %s, doorbell %s, HIP_HOST_COHERENT=%s "
"HSA_ENABLE_SDMA=%s GPU_MAX_HW_QUEUES=%s\n", g_coherent ? "explicit" : "default",
g_doorbell_store ? "stored" : "incremented", std::getenv("HIP_HOST_COHERENT") ? std::getenv("HIP_HOST_COHERENT") : "-",
std::getenv("HSA_ENABLE_SDMA") ? std::getenv("HSA_ENABLE_SDMA") : "-",
std::getenv("GPU_MAX_HW_QUEUES") ? std::getenv("GPU_MAX_HW_QUEUES") : "-");
#else
std::fprintf(stderr, "strata verify trace (#649): on\n");
#endif
}
prof_on_ = std::getenv("STRATA_VERIFY_PROFILE") != nullptr;
if (prof_on_) {
const size_t np = (size_t) g.n_layers * kProfPer + 4;
@@ -473,6 +581,7 @@ bool Verifier::record_window(int T, dpct::queue_ptr cs, std::string &err) {
// a host-side wait + clock (there is no %globaltimer here), so the existing stage profiler reports ms per stage.
static const bool eager = std::getenv("STRATA_VERIFY_EAGER") != nullptr;
auto stamp = [&](int64_t l, int i, int grp) {
if (trace_m_ != nullptr) gpu_stamp(trace_m_, (int) ((l * kProfPer + i) * 2 + grp), cs); // #649
if (!prof_on_ || grp != 0) return;
if (eager) { cs->wait(); prof_h_[(size_t) (l * kProfPer + i)] = (unsigned long long) std::chrono::steady_clock::now().time_since_epoch().count(); }
else gpu_stamp(prof_, (int) (l * kProfPer + i), cs);
@@ -781,6 +890,12 @@ bool Verifier::record_window(int T, dpct::queue_ptr cs, std::string &err) {
hits_.cache_base, slot_off_d_, (long long) hits_.blob,
plan_ + (size_t) grp * (size_t) (plan_i32_ + 16), (long long) max_t_ * K, skip_ + grp,
(uint32_t) ((l - lb_) * G + grp + 1), cs);
#if defined(STRATA_USE_HIP)
if (g_doorbell_store) // #649 A/B: the step's ring stored, not incremented over PCIe
doorbell_publish_value(xm, ids_ + tb * K, w_ + tb * K, (int64_t) n * N, (int64_t) n * K, m_x_ + tb * N,
m_ids_ + tb * K, m_w_ + tb * K, m_seq_, (uint32_t) ((l - lb_) * G + grp + 1), cs);
else
#endif
doorbell_publish(xm, ids_ + tb * K, w_ + tb * K, (int64_t) n * N, (int64_t) n * K, m_x_ + tb * N,
m_ids_ + tb * K, m_w_ + tb * K, m_seq_, cs);
stamp(l, 17, grp);
@@ -1203,7 +1318,9 @@ bool Verifier::run(int T, const int32_t *tokens, int64_t pos0, PoolMultiFn pool,
*(volatile uint32_t*) h_flag_ = 0;
*(volatile uint32_t*) h_flagA_ = 0;
*(volatile uint32_t*) h_flagB_ = 0;
if (trace_h_ != nullptr) std::memset(trace_h_, 0, trace_n_ * 8); // #649: this window's breadcrumbs only
std::atomic_thread_fence(std::memory_order_seq_cst);
trace_ev("WINDOW", -1, -1, pos0 * 16 + T);
last_t_ = T;
last_pos0_ = pos0;
for (int t = 0; t < T; ++t) last_tokens_[t] = tokens[t];
@@ -1225,6 +1342,7 @@ bool Verifier::run(int T, const int32_t *tokens, int64_t pos0, PoolMultiFn pool,
/*
DPCT1000: Error handling if-stmt was detected but could not be rewritten.
*/
trace_ev("LAUNCHED", -1, -1, (int64_t) le);
if (le != 0) {
err =
std::string("verify: launch: ") + dpct::get_error_string_dummy(le);
@@ -1261,6 +1379,8 @@ bool Verifier::run(int T, const int32_t *tokens, int64_t pos0, PoolMultiFn pool,
const dpct::err0 q =
DPCT_CHECK_ERROR(((cs_)->ext_oneapi_empty()));
if (q != 1 && *seq < want) {
trace_ev("NEVER-RANG", k, l, (int64_t) q);
trace_dump(stderr);
err = "verify: layer " + std::to_string(l) +
" never rang (" +
/*
@@ -1278,11 +1398,20 @@ bool Verifier::run(int T, const int32_t *tokens, int64_t pos0, PoolMultiFn pool,
}
if (now - a > std::chrono::seconds(20)) {
// #267: the caller ends the engine; no spin kernel may outlive it
trace_ev("TIMEOUT", k, l, (int64_t) !cs_->ext_oneapi_empty()); // SYCL port: 1 = still running
if (g_trace) {
diag(stderr); // the words and the breadcrumbs before the release ...
const bool drained = release_gpu_waits(5000);
trace_dump(stderr); // ... and after it: did the GPU move once its waits were raised?
err = "verify: timed out at layer " + std::to_string(l) + released_note(drained);
return false;
}
err = "verify: timed out at layer " + std::to_string(l) + released_note(release_gpu_waits(5000));
return false;
}
}
const Clock::time_point b = Clock::now();
if (g_trace) trace_ev("RANG", k, l, (int64_t) std::chrono::duration_cast<std::chrono::microseconds>(b - a).count());
VDBG("layer %lld rang\n", (long long) l);
cur_layer_ = want - 1;
set_plan_slot(grp);
@@ -1292,6 +1421,8 @@ bool Verifier::run(int T, const int32_t *tokens, int64_t pos0, PoolMultiFn pool,
pool(user, h_x_ + (size_t) tb * g.n_embd, h_ids_ + (size_t) tb * ss.k, n, ss.k,
h_ymiss_ + (size_t) tb * ss.k * g.n_embd, l);
VDBG("layer %lld served\n", (long long) l);
if (g_trace) trace_ev(*(volatile uint32_t*) h_flagA_ == want ? "SERVED" : "SERVED-NO-PLAN-YET", k, l,
(int64_t) ms_since(b)); // aux: ms the CPU experts took
progress_tick();
std::atomic_thread_fence(std::memory_order_seq_cst);
_mm_sfence();
@@ -1313,7 +1444,9 @@ bool Verifier::run(int T, const int32_t *tokens, int64_t pos0, PoolMultiFn pool,
// #267: a window the GPU never finishes (a spin kernel that never sees its flag) holds the host here; the stall
// watchdog then releases every verifier's GPU waits (release_live_verifiers) before it ends the engine, so no
// spin kernel outlives the process. The wait itself stays a blocking sync (a polling wait cost decode upstream).
trace_ev("SYNC", -1, -1, 0);
const dpct::err0 se = DPCT_CHECK_ERROR(cs_->wait());
trace_ev("SYNCED", -1, -1, (int64_t) se);
if (std::getenv("STRATA_VERIFY_DEBUG") != nullptr) {
const Clock::time_point t_done = Clock::now();
std::fprintf(stderr, "verify dbg: T=%d window: since previous window %.1f ms, staging %.1f ms, gpu %.1f ms\n", T,
+66
View File
@@ -600,6 +600,72 @@ __dpct_inline__ void doorbell_publish_kernel(const float *__restrict__ x,
}
}
// #649 (HIP, STRATA_DOORBELL_STORE=1): the same publish, but the ring is STORED (the step's own number, known at
// capture) instead of read-modify-written over PCIe - one store, no read of host memory from the GPU.
__dpct_inline__ void doorbell_publish_value_kernel(
const float *__restrict__ x, const int32_t *__restrict__ ids,
const float *__restrict__ w, int n, int k, float *x_out, int32_t *ids_out,
float *w_out, uint32_t *seq, uint32_t value) {
auto item_ct1 = sycl::ext::oneapi::this_work_item::get_nd_item<3>();
#pragma unroll
for (int i = item_ct1.get_local_id(2); i < n;
i += item_ct1.get_local_range(2)) x_out[i] = x[i];
if ((int)item_ct1.get_local_id(2) < k) {
ids_out[item_ct1.get_local_id(2)] = ids[item_ct1.get_local_id(2)];
w_out[item_ct1.get_local_id(2)] = w[item_ct1.get_local_id(2)];
}
/*
DPCT1078: Consider replacing memory_order::acq_rel with
memory_order::seq_cst for correctness if strong memory order restrictions
are needed.
*/
sycl::atomic_fence(sycl::memory_order::acq_rel, sycl::memory_scope::system);
item_ct1.barrier(sycl::access::fence_space::local_space);
if (item_ct1.get_local_id(2) == 0) {
/*
DPCT1078: Consider replacing memory_order::acq_rel with
memory_order::seq_cst for correctness if strong memory order
restrictions are needed.
*/
sycl::atomic_fence(sycl::memory_order::acq_rel,
sycl::memory_scope::system);
*(volatile uint32_t*) seq = value;
/*
DPCT1078: Consider replacing memory_order::acq_rel with
memory_order::seq_cst for correctness if strong memory order
restrictions are needed.
*/
sycl::atomic_fence(sycl::memory_order::acq_rel,
sycl::memory_scope::system);
}
}
void doorbell_publish_value(const float* x, const int32_t* ids, const float* weights, int64_t n, int64_t k,
float* x_out, int32_t* ids_out, float* weights_out, uint32_t* d_seq, uint32_t value,
void* stream) {
if (k > 1024) { std::fprintf(stderr, "doorbell_publish: k too large\n"); std::exit(1); }
/*
DPCT1049: The work-group size passed to the SYCL kernel may exceed the
limit. To get the device limit, query info::device::max_work_group_size.
Adjust the work-group size if needed.
*/
{
auto exp_props = sycl::ext::oneapi::experimental::properties{
sycl::ext::oneapi::experimental::use_root_sync};
strata::q_of(stream)
->parallel_for<
dpct_kernel_name<class doorbell_publish_value_kernel_9bdb22>>(
sycl::nd_range<3>(sycl::range(1, 1, 1024),
sycl::range(1, 1, 1024)),
exp_props, [=](sycl::nd_item<3> item_ct1) {
doorbell_publish_value_kernel(x, ids, weights, (int)n,
(int)k, x_out, ids_out,
weights_out, d_seq, value);
});
}
}
void doorbell_publish(const float* x, const int32_t* ids, const float* weights, int64_t n, int64_t k, float* x_out,
int32_t* ids_out, float* weights_out, uint32_t* d_seq, void* stream) {
if (k > 1024) { std::fprintf(stderr, "doorbell_publish: k too large\n"); std::exit(1); }
+4 -3
View File
@@ -11,6 +11,7 @@
#include "strata/sycl_queue.hpp"
#include "strata/kernels/iq_kernels.hpp"
#include "strata/kernels/dp4a.hpp"
#include "strata/kernels/q8_1_finite.hpp"
#define GGML_COMMON_DECL_SYCL
#define GGML_COMMON_IMPL_SYCL
@@ -1564,11 +1565,11 @@ __dpct_inline__ void q8_1_store(const float xi, block_q8_1 *__restrict__ y,
0xffffffffu, sycl::ext::oneapi::this_work_item::get_sub_group(),
sum, o);
}
const float d = amax / 127.0f;
const int8_t q = amax == 0.0f ? 0 : sycl::round(xi / d);
const float d = q8_1_finite(amax / 127.0f); // #606: q8_1_finite.hpp - the same bits for every finite block
const int8_t q = q8_1_quant(xi, d, amax);
const long long ib = i / 32, iqs = i % 32;
y[ib].qs[iqs] = q;
if (iqs == 0) y[ib].ds = sycl::half2(d, sum);
if (iqs == 0) y[ib].ds = q8_1_ds(d, sum);
}
__dpct_inline__ void quantize_q8_1_kernel(const float *__restrict__ x,
+23
View File
@@ -444,6 +444,29 @@ catch (sycl::exception const &exc) {
std::exit(1);
}
void kv_unstage_to_host(const QsaAttnPools &stage, const KvHostPools &host,
int fmt, int64_t b0, int64_t b1, const QsaShapes &s,
void *stream) try {
if (b1 <= b0) return;
const Runs r = runs_of(stage, host, fmt, s); // src: the host copy, dst: the staging pool (identity layout both)
for (int a = 0; a < r.n; ++a)
/*
DPCT1124: cudaMemcpyAsync is migrated to asynchronous memcpy API.
While the origin API might be synchronous, it depends on the type of
operand memory, so you may need to call wait() on event return by memcpy
API to ensure synchronization behavior.
*/
if (DPCT_CHECK_ERROR(strata::q_of(stream)->memcpy(
(void *)(r.src[a] + b0 * r.len[a]), r.dst[a] + b0 * r.len[a],
(size_t)((b1 - b0) * r.len[a]))) != 0)
check("unstage");
}
catch (sycl::exception const &exc) {
std::cerr << exc.what() << "Exception caught at file:" << __FILE__
<< ", line:" << __LINE__ << std::endl;
std::exit(1);
}
KvStreamCounters kv_stream_counters(const KvStreamMap &m) try {
int32_t c[kKvCtlInts] = {};
KvStreamCounters r;
+4 -3
View File
@@ -30,6 +30,7 @@
#include "strata/sycl_queue.hpp"
#include "strata/kernels/native_mmvq.hpp"
#include "strata/kernels/dp4a.hpp"
#include "strata/kernels/q8_1_finite.hpp"
#include "strata/kernels/iq_kernels.hpp"
#include <cstdint>
@@ -190,10 +191,10 @@ __dpct_inline__ void native_quantize_q8_1_kernel(const float *__restrict__ x,
const float xi = x[i];
const float amax = warp_max(sycl::fabs(xi));
const float sum = warp_sum(xi);
const float d = amax / 127.0f;
const int8_t q = amax == 0.0f ? 0 : sycl::round(xi / d);
const float d = q8_1_finite(amax / 127.0f); // #606: q8_1_finite.hpp - the same bits for every finite block
const int8_t q = q8_1_quant(xi, d, amax);
y[i / Q8K].qs[i % Q8K] = q;
if (i % Q8K == 0) y[i / Q8K].ds = sycl::half2(d, sum);
if (i % Q8K == 0) y[i / Q8K].ds = q8_1_ds(d, sum);
}
// Exact pinned vec_dot_q5_K_q8_1_impl_vmmq expression and integer dot order.
+78
View File
@@ -19,6 +19,7 @@
#include <dpct/dpct.hpp>
#include "strata/kernels/f16_bits.hpp"
#include "strata/kernels/iq_kernels.hpp"
#include "strata/kernels/native_mmvq.hpp"
#include <algorithm>
#include <cmath>
@@ -506,6 +507,82 @@ void bench(dpct::queue_ptr s, std::mt19937 &rng) {
}
}
// #606: both q8_1 activation quantizers (quantize_q8_1_rows and native_quantize_q8_1) against a host transcription
// of the 0.1.38 formula: every finite block bit for bit; a block whose sum (or scale) overflowed fp16 is now stored
// finite (the largest half, its sign), with its values in the int8 range.
void check_q8_1_finite(dpct::queue_ptr s, std::mt19937 &rng) {
const int n = 2560, rows = 3, nb = n / 32;
std::vector<float> x = random_x((size_t) rows * n, rng);
for (int i = 0; i < 32; ++i) x[(size_t) 5 * 32 + i] = 3000.0f; // sum 96,000: overflowed
for (int i = 0; i < 32; ++i) x[(size_t) 9 * 32 + i] = i == 3 ? -70000.0f : 0.5f; // sum -69,984.5
for (int i = 0; i < 32; ++i) x[(size_t) 13 * 32 + i] = 2000.0f + i; // sum 64,496: finite, kept
x[(size_t) n + 40] = 2.0e7f; // amax / 127 overflowed too
float* dx = dalloc<float>(x.size());
/*
DPCT1114: cudaMemcpy is migrated to asynchronization memcpy, assuming
in the original code the source host memory is pageable memory. If the
memory is not pageable, call wait() on event return by memcpy API to ensure
synchronization behavior.
*/
ck(DPCT_CHECK_ERROR(
(dpct::get_current_device().queues_wait_and_throw(), dpct::get_in_order_queue()).memcpy(dx, x.data(), x.size() * 4).wait()),
"x");
uint8_t* dq = dalloc<uint8_t>((size_t) rows * nb * 36);
for (int path = 0; path < 2; ++path) {
if (path == 0) k::quantize_q8_1_rows(dx, rows, n, dq, s);
else k::native_quantize_q8_1(dx, dq, n, rows, s);
ck(DPCT_CHECK_ERROR(s->wait()), "quantize");
std::vector<uint8_t> got((size_t) rows * nb * 36);
ck(DPCT_CHECK_ERROR((dpct::get_current_device().queues_wait_and_throw(), dpct::get_in_order_queue())
.memcpy(got.data(), dq, got.size())
.wait()),
"q");
int same = 0, clamped = 0, bad = 0;
for (int b = 0; b < rows * nb; ++b) {
const float* v = x.data() + (size_t) b * 32;
float lane[32], amax = 0.0f;
for (int i = 0; i < 32; ++i) { lane[i] = v[i]; amax = std::max(amax, std::fabs(v[i])); }
for (int o = 16; o > 0; o >>= 1) { // the warp's butterfly: every lane ends with the same sum
float nx[32];
for (int i = 0; i < 32; ++i) nx[i] = lane[i] + lane[i ^ o];
std::memcpy(lane, nx, sizeof lane);
}
const float d = amax / 127.0f, sum = lane[0];
const uint8_t* g = got.data() + (size_t) b * 36;
uint16_t gd, gs;
std::memcpy(&gd, g, 2);
std::memcpy(&gs, g + 2, 2);
const uint16_t od = k::f16_from_f32(d), os = k::f16_from_f32(sum);
const bool finite_before = (od & 0x7c00) != 0x7c00 && (os & 0x7c00) != 0x7c00;
if (finite_before) {
uint8_t want[36];
std::memcpy(want, &od, 2);
std::memcpy(want + 2, &os, 2);
for (int i = 0; i < 32; ++i) // CUDA's roundf: half away from zero, as std::round
want[4 + i] = (uint8_t) (int8_t) (amax == 0.0f ? 0.0f : std::round(v[i] / d));
// native_mmvq.cu is built with --use_fast_math (its divisions are approximate): there the sum must be
// the same bits, the scale within one fp16 step and each value within one step
bool near = path == 1 && gs == os && std::abs((int) gd - (int) od) <= 1;
for (int i = 0; near && i < 32; ++i) near = std::abs((int) (int8_t) g[4 + i] - (int) (int8_t) want[4 + i]) <= 1;
if (std::memcmp(want, g, 36) == 0 || near) ++same;
else { ++bad; std::printf(" q8_1 path %d block %d: changed although it was finite\n", path, b); }
} else {
const float fd = k::f32_from_f16(gd), fs = k::f32_from_f16(gs);
bool ok = std::isfinite(fd) && std::isfinite(fs) && fd > 0.0f &&
((os & 0x7c00) != 0x7c00 ? gs == os : std::fabs(fs) == 65504.0f && (fs < 0) == (sum < 0));
for (int i = 0; i < 32; ++i) ok = ok && (int8_t) g[4 + i] >= -127 && (int8_t) g[4 + i] <= 127;
if (ok) ++clamped;
else { ++bad; std::printf(" q8_1 path %d block %d: d %g sum %g not finite and clamped\n", path, b, fd, fs); }
}
}
std::printf("q8_1 finite (%s): %d finite blocks as before, %d overflowed blocks clamped finite%s\n",
path == 0 ? "quantize_q8_1_rows" : "native_quantize_q8_1", same, clamped, bad ? " FAIL" : "");
g_fail += bad + (clamped != 3);
}
sycl::free(dx, dpct::get_in_order_queue());
sycl::free(dq, dpct::get_in_order_queue());
}
} // namespace
int main(int argc, char** argv) {
@@ -522,6 +599,7 @@ int main(int argc, char** argv) {
for (int dt : {20, 42}) check_grouped(gu, dt, 2560, 640, s, rng); // the model's shape
check_grouped(gu, 23, 1024, 512, s, rng); // IQ4_XS down needs n_ff % 256 == 0
}
check_q8_1_finite(s, rng);
if (do_bench) bench(s, rng);
std::printf("iq_multi_parity: %d failures\n", g_fail);
dpct::get_current_device().destroy_queue(s);
+58 -2
View File
@@ -2113,8 +2113,41 @@ bool Prefill::run(const int64_t *tokens, int64_t n, int64_t pos0,
}
host_setup_ms += ms_since(tsetup);
bool normed = false; // F-2: the previous half's write already normed R for this half (grs, xn16)
// #579 #613 (opt-in diagnosis, STRATA_PF_STEP_SYNC=1): the compute and copy streams are waited for after each
// step named below, a step that took over 250 ms is logged, and a stall's report names the step it is in.
// Slower (a sync per step); the bytes are the same.
static const bool step_sync = [] { const char* e = std::getenv("STRATA_PF_STEP_SYNC"); return e && e[0] == '1'; }();
auto pf_step = [&](const char *what, int64_t layer) {
try {
if (!step_sync) return;
core::progress_at(what, layer, p0);
const auto ts = Clock::now();
const dpct::err0 a = DPCT_CHECK_ERROR(m.cs->wait()),
b = DPCT_CHECK_ERROR(m.copy->wait());
const double ms = ms_since(ts);
if (ms > 250.0 || a != 0 || b != 0)
std::fprintf(stderr,
"strata pf-step: chunk from token %lld, layer "
"%lld: %s took %.0f ms (%s / %s)\n",
/*
DPCT1009: SYCL reports errors using exceptions
and does not use error codes. Please replace the
"get_error_string_dummy(...)" with a real
error-handling function.
*/
(long long)p0, (long long)layer, what, ms,
dpct::get_error_string_dummy(a),
dpct::get_error_string_dummy(b));
}
catch (sycl::exception const &exc) {
std::cerr << exc.what() << "Exception caught at file:" << __FILE__
<< ", line:" << __LINE__ << std::endl;
std::exit(1);
}
};
for (int64_t l = LB; l < LE; ++l) {
core::progress_beat(); // the serve watchdog: a prompt chunk of 8192 tokens is still moving
if (l > LB) pf_step("reading the prompt (batched, step sync): the experts and the rest of layer", l - 1);
core::progress_at("reading the prompt (batched): layer", l, p0); // #251: a stall names layer and chunk
const core::LayerView v(*m.wt, l);
if (l == std::max<int64_t>(LB, 1) && !ple_land()) return false; // the PLE rows, read from layer 1 on
@@ -2244,7 +2277,20 @@ bool Prefill::run(const int64_t *tokens, int64_t n, int64_t pos0,
core::qsa_kv_format(st),
(p0 + s.page_size - 1) / s.page_size, s, m.cs);
pt.mark(kPfQsa, cs);
pf_step("reading the prompt (batched, step sync): the K/V staged from RAM at layer", l);
}
// #579 #613 (HIP, opt-in A/B, STRATA_KV_HOST_DMA=1): the append writes the staging pool and the
// resident slots only, and one DMA copies the chunk's blocks from the staging pool to the host copy
// - no kernel writes host memory over PCIe. The bytes every reader sees are the same (the staged
// first block is complete; past the chunk's last cell nothing is read until a later append writes
// it). CUDA: never.
#if defined(STRATA_USE_HIP)
static const bool kv_host_dma = [] { const char* e = std::getenv("STRATA_KV_HOST_DMA"); return e && e[0] == '1'; }();
#else
constexpr bool kv_host_dma = false;
#endif
const bool host_by_dma = staged && kv_host_dma;
const strata::kernels::KvHostPools* host_w = host_by_dma ? nullptr : &st.host;
if (st.kv_hybrid) { // K8V4: K INT8 unrotated, V rotated Q4_0 (only V and the output rotate)
strata::kernels::fwht256_inplace_cuda(m.Vc, T * 2, m.cs);
kv_append(m.Kc, m.Kc, T, p0, st.page_table, s.page_size, nullptr, nullptr,
@@ -2259,12 +2305,17 @@ bool Prefill::run(const int64_t *tokens, int64_t n, int64_t pos0,
}
if (st.kv_q4)
strata::kernels::kv_append_q4(st.k_q4, st.v_q4, st.page_table, p0, T, m.Kc, m.Vc, s, m.cs,
&st.host, staged ? &m.stage : nullptr);
host_w, staged ? &m.stage : nullptr);
else
kv_append(m.Kc, m.Vc, T, p0, st.page_table, s.page_size, st.kv_int8 ? nullptr : st.k_pool,
st.kv_int8 ? nullptr : st.v_pool, st.k_q, st.v_q, st.k_scale, st.v_scale, m.cs,
&st.host, staged ? &m.stage : nullptr);
host_w, staged ? &m.stage : nullptr);
if (host_by_dma)
strata::kernels::kv_unstage_to_host(pools_of(m.stage, m.ident_table), st.host,
core::qsa_kv_format(st), p0 / s.page_size,
(p0 + T + s.page_size - 1) / s.page_size, s, m.cs);
}
if (staged) pf_step("reading the prompt (batched, step sync): the K/V append at layer", l);
split_q(m.Qf, m.q, T, m.cs);
rms_rows(m.q, (const float*) wqn->data, T * 24, 256, 256, EPS, m.cs);
rope(m.q, T, 24, 256, 6144, p0, strata::kernels::rope_scaling(), m.cs);
@@ -2587,7 +2638,12 @@ bool Prefill::run(const int64_t *tokens, int64_t n, int64_t pos0,
*/
else m.cs->memcpy(m.ids_host.data(), m.ids,
(size_t)T * K * 4);
// #579: a stall here is the GPU (this layer's attention and router, or the previous layer's
// work), not the host: the watchdog's report says so (only its text changes)
core::progress_at("reading the prompt (batched): waiting for the GPU (attention, router) at layer",
l, p0);
m.cs->wait();
core::progress_at("reading the prompt (batched): layer", l, p0);
pt.fold();
if (pe.on) { // the peer's marks so far are done: the primary waited for its last rows
/*
+633 -62
View File
@@ -84,6 +84,8 @@ namespace strata::prefill { void set_nonresident_share(double share); } // SYC
#else
#include <unistd.h>
#include <cerrno>
#include <sys/stat.h>
#include <sys/sysmacros.h>
#endif
#include <array>
@@ -125,7 +127,116 @@ bool under_wddm() {
#endif
}
// #620 #486: " (N MiB of M MiB VRAM free on this GPU)" for an allocation's failure message (clears the error first)
std::string vram_free_note() try {
/*
DPCT1010: SYCL uses exceptions to report errors and does not use the
error codes. The cudaGetLastError function call was replaced with 0. You
need to rewrite this code.
*/
(void)0;
size_t free_b = 0, total_b = 0;
/*
DPCT1106: 'cudaMemGetInfo' was migrated with the Intel extensions for
device information which may not be supported by all compilers or runtimes.
You may need to adjust the code.
*/
if (DPCT_CHECK_ERROR(
dpct::get_current_device().get_memory_info(free_b, total_b)) != 0) {
/*
DPCT1010: SYCL uses exceptions to report errors and does not use the
error codes. The cudaGetLastError function call was replaced with 0. You
need to rewrite this code.
*/
(void)0;
return {};
}
char buf[96];
std::snprintf(buf, sizeof buf, " (%llu MiB of %llu MiB VRAM free on this GPU)", (unsigned long long) (free_b >> 20),
(unsigned long long) (total_b >> 20));
return buf;
}
catch (sycl::exception const &exc) {
std::cerr << exc.what() << "Exception caught at file:" << __FILE__
<< ", line:" << __LINE__ << std::endl;
std::exit(1);
}
// perf-review D-4: the lent slots are refilled with queued copies and one wait; STRATA_REFILL_BLOCKING=1 waits on each
// A file-backed arena on AMD (STRATA_ARENA_MMAP; the engine raises ROCclr's pin-in-place threshold, see main): an
// adaptive swap copies one expert straight from the mapped file. Through ROCclr's staging that copy blocks the host
// (decode windows of 150+ ms); instead the batch's pages are locked here for the async copies and unlocked when they
// have landed (unpin_blobs), so the release that follows can hand them back. Neighbouring experts share a page and
// a copy from a range only PART of which is registered fails ("invalid argument"), so the page ranges are merged
// first and each merged span is registered once. A span that does not register is copied through the staging
// path. Portable: a layer split copies to either card.
// Whether pin_blobs locks anything (AMD on Linux with STRATA_ARENA_MMAP=1): the callers only gather the swaps' spans
// then - `blob()` is not free (it counts reads, and the file tier may assemble the blob), so the default runs skip it.
bool pin_blobs_on() {
#if (defined(STRATA_USE_HIP) || defined(STRATA_HIP_GFX906)) && !defined(_WIN32)
static const bool on = std::getenv("STRATA_ARENA_MMAP") && std::getenv("STRATA_ARENA_MMAP")[0] == '1';
return on;
#else
return false;
#endif
}
int pin_blobs(std::vector<std::pair<uintptr_t, uintptr_t>> r, std::vector<void*>& live) {
int failed = 0; // ranges left pageable (a copy from them still works, through the driver's staging)
#if (defined(STRATA_USE_HIP) || defined(STRATA_HIP_GFX906)) && !defined(_WIN32)
if (!pin_blobs_on() || r.empty()) return 0;
for (auto& [a, e] : r) { a &= ~(uintptr_t) 4095; e = (e + 4095) & ~(uintptr_t) 4095; }
std::sort(r.begin(), r.end());
size_t k = 0;
for (size_t i = 1; i < r.size(); ++i) {
if (r[i].first <= r[k].second) r[k].second = std::max(r[k].second, r[i].second);
else r[++k] = r[i];
}
r.resize(k + 1);
for (const auto& [a, e] : r) {
if (cudaHostRegister((void*) a, (size_t) (e - a), cudaHostRegisterReadOnly | cudaHostRegisterPortable) ==
cudaSuccess)
live.push_back((void*) a);
else {
(void) cudaGetLastError();
++failed;
}
}
#else
(void) r; (void) live;
#endif
return failed;
}
void unpin_blobs(std::vector<void*>& v) {
/*
DPCT1027: The call to cudaHostUnregister was replaced with 0 because
SYCL currently does not support registering of existing host memory for use
by device. Use USM to allocate memory for use by host and device.
*/
for (void *p : v)(void) 0;
/*
DPCT1010: SYCL uses exceptions to report errors and does not use the
error codes. The cudaGetLastError function call was replaced with 0. You
need to rewrite this code.
*/
(void)0;
v.clear();
}
// STRATA_RSS_TRACE=1: the process's mapped file pages (the arena) at a startup step - who reads the arena back
void rss_probe(const char* where) {
static const bool on = std::getenv("STRATA_RSS_TRACE") != nullptr;
if (!on) return;
FILE* f = std::fopen("/proc/self/status", "r");
if (!f) return;
char line[256];
long long kb = -1;
while (std::fgets(line, sizeof line, f))
if (std::strncmp(line, "RssFile:", 8) == 0) { kb = std::atoll(line + 8); break; }
std::fclose(f);
std::fprintf(stderr, "strata rss: %s: RssFile %.2f GiB\n", where, (double) kb / (1024.0 * 1024.0));
std::fflush(stderr);
}
bool refill_blocking() {
static const bool v = std::getenv("STRATA_REFILL_BLOCKING") != nullptr;
return v;
@@ -367,6 +478,10 @@ struct Options {
/// and the KV state, minus this reserve for the graphs, the hit scratch and the head.
int vram_reserve_mib = 700;
bool vram_reserve_given = false; ///< --vram-reserve-mib on the command line (#496: no smaller automatic reserve)
/// #533 (opt-in): the expert cache in physical segments (CUDA virtual memory management), so the serve loop's
/// `VRAM <reserve_mib>` command can give part of it back to other programs and take it again. Off: one cudaMalloc.
bool vram_elastic = false;
int64_t vram_segment_mib = 512;
/// Plan v0.3 P5: batched prompt processing in chunks of this many tokens (0 = the token path).
int64_t prefill_chunk = 0;
/// `--prefill auto`: the largest chunk (up to 8192) whose buffers the expert cache can lend. Every expert a chunk
@@ -440,6 +555,11 @@ struct Options {
/// --serve: the token that opens a chat turn (<|im_start|>). The last one in a prompt is where the chat's
/// history ends and the new assistant turn begins, which is the checkpoint the next request can reuse.
int64_t turn_token = 248045;
/// --serve (#458, opt-in): the role token of a short turn the server puts right before the new assistant turn
/// (the "system" of a trailing reasoning-effort turn). When the turn before the last <|im_start|> opens with it,
/// the checkpoint goes in front of that turn instead, so it holds the conversation and not the effort text.
/// -1 = off (the checkpoint at the last <|im_start|>, as before).
int64_t tail_role_token = -1;
/// --serve: a text part of the prompt of at most N tokens (a chat message, the assistant header, a short tool
/// result) goes through the verify windows, S tokens at a time, instead of the batched prompt path (0 = always
/// the batched path)
@@ -542,8 +662,13 @@ void usage() {
" --conversation-cache-mib N --serve: RAM budget for parked conversations (default 0 = off)\n"
" --conversation-cache-slots N --serve: at most N parked conversations (default 4)\n"
" --conversation-cache-min-free-mib N --serve: physical RAM floor when parking (default 2560)\n"
" --vram-elastic --serve (#533, opt-in, NVIDIA): the expert cache in segments (--vram-segment-mib,\n"
" default 512), so the command `VRAM <reserve_mib>` (the server's POST /v1/vram) can\n"
" give VRAM back to other programs between requests and take it back later\n"
" --prompt-cache-every N --serve: also checkpoint every N fresh prompt tokens (default 16384, 0 = off)\n"
" --turn-token ID --serve: the token that opens a chat turn (default 248045, <|im_start|>)\n"
" --tail-role-token ID --serve: a turn of this role right before the last turn is left out of\n"
" the conversation checkpoint (a trailing effort turn; default -1 = off)\n"
" --short-read N --serve: read at most N fresh text tokens through the decode windows instead\n"
" of the batched prompt path (default 64, 0 = off)\n"
" --suffix-draft N prompt lookup: draft from an earlier repeat of the last N+ tokens of context\n"
@@ -820,6 +945,55 @@ std::string stage_text() {
return s;
}
#if !defined(_WIN32)
// #605: whether `path` is on a rotational disk (1), not (0) or unknown (-1), from sysfs; `dev` is the disk's name
int on_rotational_disk(const std::string& path, std::string& dev) {
struct stat st{};
if (::stat(path.c_str(), &st) != 0) return -1;
char sys[64];
std::snprintf(sys, sizeof sys, "/sys/dev/block/%u:%u", (unsigned) major(st.st_dev), (unsigned) minor(st.st_dev));
std::error_code ec;
std::filesystem::path p = std::filesystem::canonical(sys, ec);
for (int up = 0; !ec && up < 2; ++up, p = p.parent_path()) { // a partition has its disk's queue
std::ifstream q(p / "queue" / "rotational");
int r = -1;
if (q >> r) {
dev = p.filename().string();
return r;
}
}
return -1;
}
// #605: the engine's threads in uninterruptible sleep (state D - almost always waiting for the disk), and where
// (wchan): 16 threads in blk_io_schedule is a disk that cannot keep up, not a deadlock in the engine
void disk_wait_threads(std::FILE* f) {
std::map<std::string, int> where;
int waiting = 0, threads = 0;
std::error_code ec;
for (const auto& e : std::filesystem::directory_iterator("/proc/self/task", ec)) {
++threads;
std::ifstream st(e.path() / "stat");
std::string line;
std::getline(st, line);
const size_t rp = line.rfind(')');
if (rp == std::string::npos || rp + 2 >= line.size() || line[rp + 2] != 'D') continue;
++waiting;
std::ifstream w(e.path() / "wchan");
std::string wc;
std::getline(w, wc);
++where[wc.empty() || wc == "0" ? "?" : wc];
}
if (threads == 0) return;
std::string list;
for (const auto& [name, n] : where) list += (list.empty() ? " (" : ", ") + name + " x" + std::to_string(n);
if (!list.empty()) list += ")";
std::fprintf(f, " threads waiting on the disk (state D): %d of %d%s%s\n", waiting, threads, list.c_str(),
waiting > 0 ? " - the engine is waiting for the drive: a rotational or failing disk with --ple-io "
"direct (#605: --ple-io ram), or a drive too slow for the reads asked of it" : "");
}
#endif
void stall_report(std::FILE* f, uint64_t layers_during) {
strata::core::Progress& p = strata::core::progress();
std::fprintf(f, "strata serve: stall report (engine %s): stage \"%s\" for %lld s; %llu layers served since the "
@@ -833,6 +1007,9 @@ void stall_report(std::FILE* f, uint64_t layers_during) {
}
if (auto fn = strata::core::diag_pool_fn().load()) fn(f);
if (auto fn = strata::core::diag_verify_fn().load()) fn(f);
#if !defined(_WIN32)
disk_wait_threads(f);
#endif
const MemSample m = mem_sample();
std::fprintf(f, " memory: %llu MiB resident, %llu MiB %s, %llu MiB RAM available; %llu %s\n", m.rss_mib,
m.commit_mib,
@@ -1134,6 +1311,15 @@ int main(int argc, char **argv) try {
// pipe or a file is block-buffered, so a program that dies loses every line it had already printed - which
// turns "it crashed at step 7" into "it crashed somewhere", and the difference is a debugging session.
std::setvbuf(stdout, nullptr, _IONBF, 0);
#if (defined(STRATA_USE_HIP) || defined(STRATA_HIP_GFX906)) && !defined(_WIN32)
// AMD, a file-backed arena (STRATA_ARENA_MMAP): ROCclr copies a pageable source of 1 MiB or more by locking its
// pages in place (a GPU userptr), and keeps them - so every expert the VRAM fill copied from the mapped
// experts.bin stayed resident, the release of the VRAM-held ones was undone by the driver at once, and a
// 32 GB PC ran with ~0.3 GB free: the kernel swapped, and a turn's checkpoint took 25-80 s. Above this size
// ROCclr stages through its own pinned buffers instead; decode is unchanged (49.06 vs 49.16 ms/window).
if (std::getenv("STRATA_ARENA_MMAP") && std::getenv("STRATA_ARENA_MMAP")[0] == '1')
setenv("GPU_PINNED_MIN_XFER_SIZE", "1000000", 0); // MiB; an explicit setting wins
#endif
// Load every CUDA kernel when the context is created, before the expert cache takes the free VRAM. With the
// default lazy loading, a kernel first used mid-prompt (MMQ for IQ3_XXS at 64K+ on a 12 GB card) found no VRAM
// left for its code and the engine ended ("out of memory: cudaFuncSetAttribute"). Costs ~30 MB of VRAM.
@@ -1281,6 +1467,8 @@ int main(int argc, char **argv) try {
else if (a == "--no-split-rows") o.no_split_rows = true;
else if (a == "--no-prefill-borrow") { o.no_prefill_borrow = true; borrow_explicit = true; }
else if (a == "--prefill-borrow") { o.no_prefill_borrow = false; borrow_explicit = true; }
else if (a == "--vram-elastic") o.vram_elastic = true;
else if (a == "--vram-segment-mib") o.vram_segment_mib = std::atoll(next("--vram-segment-mib"));
else if (a == "--prefill-until") o.prefill_until = std::atoll(next("--prefill-until"));
else if (a == "--dump-final-r") o.dump_final_r = next("--dump-final-r");
else if (a == "--spec") o.spec = std::atoi(next("--spec"));
@@ -1318,6 +1506,7 @@ int main(int argc, char **argv) try {
else if (a == "--prompt-cache-every") o.prompt_cache_every = std::max(0LL, std::atoll(next("--prompt-cache-every")));
else if (a == "--prompt-cache-root") o.prompt_cache_root = std::max(0LL, std::atoll(next("--prompt-cache-root")));
else if (a == "--turn-token") o.turn_token = std::atoll(next("--turn-token"));
else if (a == "--tail-role-token") o.tail_role_token = std::atoll(next("--tail-role-token"));
else if (a == "--short-read") o.short_read = std::max(0LL, std::atoll(next("--short-read")));
else if (a == "--suffix-draft") o.suffix_draft = std::max(0, std::atoi(next("--suffix-draft")));
else if (a == "--mtp-max-t") o.mtp_max_t = std::max(0, std::atoi(next("--mtp-max-t")));
@@ -1477,7 +1666,9 @@ int main(int argc, char **argv) try {
}
if (!ok) {
std::fprintf(stderr, "strata generate: --layer-split K[,K2..]|auto needs --serve, rising K from 2, and one "
"distinct GPU per K in --split-device (1..%d; or 0 with one K: the same GPU)\n", n_dev - 1);
"distinct GPU per K in --split-device (1..%d; or 0 with one K: the same GPU). K is the "
"FIRST LAYER of each later GPU, not a count of layers per card: \"24,36,42\" for 4 "
"GPUs, not \"24,12,6,6\" (got \"%s\")\n", n_dev - 1, o.layer_split.c_str());
return 2;
}
}
@@ -2080,6 +2271,38 @@ int main(int argc, char **argv) try {
}
if (native_pack) skip.insert("token_embd.weight");
}
// Layer split with explicit split points: every GPU holds only the dense weights of ITS layers (the PLE tensors
// stay everywhere). Without this each card keeps a full copy (~3.4 GB for the Coder) that its stage never
// reads - VRAM the expert cache wants. Opt-in, STRATA_STAGE_TRIM=1, on HIP and CUDA alike: measured on 2x MI50
// (PR #639), and the default waits for its author's re-run of the release branch on his cards.
const std::set<std::string> skip_base = skip;
const bool stage_trim = multi_gpu && !split_auto && !split_at.empty() && [] {
const char* v = std::getenv("STRATA_STAGE_TRIM");
return v != nullptr && v[0] != 0 && std::string(v) != "0";
}();
// keep_routers: CUDA0 with --mmap-experts keeps every layer's router (ffn_gate_inp, ~2.5 MiB a layer) - the file
// tier's routing-aware prefetch (RouterLookahead, below) copies all of them from CUDA0's arena, and would
// otherwise turn itself off
auto add_foreign = [&](int64_t lb, int64_t le, std::set<std::string>& out, bool keep_routers) {
std::FILE* f = std::fopen((o.pack + "/index.txt").c_str(), "rb");
if (!f) return;
char line[1024], name[256];
while (std::fgets(line, sizeof line, f)) {
if (line[0] == '#' || std::sscanf(line, "%255s", name) != 1) continue;
const std::string n = name;
if (n.rfind("blk.", 0) != 0 || n.find("ple") != std::string::npos) continue;
if (keep_routers && n.ends_with(".ffn_gate_inp.weight")) continue;
const int64_t l = std::atoll(name + 4);
if (l < lb || l >= le) out.insert(n);
}
std::fclose(f);
};
if (stage_trim) {
add_foreign(0, split_at[0], skip, o.mmap_experts);
strata::core::NativeDense::set_layer_range(0, (int) split_at[0]);
std::fprintf(stderr, "strata generate: layer split: CUDA0 loads the dense weights of layers 0-%lld only\n",
(long long) split_at[0] - 1);
}
uint64_t pool_bytes = 0;
if (!strata::core::WeightTable::pool_bytes(o.pack, pool_bytes, err, skip.empty() ? nullptr : &skip)) {
std::fprintf(stderr, "strata generate: %s\n", err.c_str());
@@ -2301,6 +2524,18 @@ int main(int argc, char **argv) try {
std::fprintf(stderr, "strata generate: %s\n", err.c_str());
return 1;
}
#if !defined(_WIN32)
if (pio.mode == strata::kernels::PleIo::Direct) {
// #605: --ple-io direct's random reads (up to --ple-inflight at once) are for SSDs; a rotational disk
// can take longer than the stall watchdog's 60 s for one prompt chunk's rows. Recommended, not forced.
std::string dev;
if (!under_wddm() && on_rotational_disk(o.ple_gguf, dev) == 1) // (WSL's virtual disk says rotational)
std::fprintf(stderr, "strata generate: WARNING: the n-gram table (%s) is on a rotational disk (%s): "
"--ple-io direct reads it at random and can stall a prompt for minutes (#605). "
"--ple-io ram (it then needs RAM for the table) or the model on an SSD is "
"recommended\n", o.ple_gguf.c_str(), dev.c_str());
}
#endif
if (pio.lock)
std::fprintf(stderr, "strata generate: PLE table %s (--ple-io ram) in %.1f s\n",
ple_table.locked() ? "locked in RAM" : "loaded (not locked)",
@@ -2474,12 +2709,26 @@ int main(int argc, char **argv) try {
return 1;
}
const strata::core::OnDevice on(st.dev);
std::set<std::string> skip_s = skip_base;
uint64_t pool_s = pool_bytes;
if (stage_trim) {
const int64_t lb = split_at[i], le = i + 1 < split_at.size() ? split_at[i + 1] : g.n_layers;
add_foreign(lb, le, skip_s, false);
strata::core::NativeDense::set_layer_range((int) lb, (int) le);
if (!strata::core::WeightTable::pool_bytes(o.pack, pool_s, err, &skip_s)) {
std::fprintf(stderr, "strata generate: %s\n", err.c_str());
return 1;
}
std::fprintf(stderr, "strata generate: layer split: CUDA%d loads the dense weights of layers %lld-%lld only\n",
st.dev, (long long) lb, (long long) le - 1);
} else {
skip_s = skip;
}
void* arena_s = nullptr;
if (DPCT_CHECK_ERROR(arena_s = (void *)sycl::malloc_device(
pool_bytes, dpct::get_in_order_queue())) !=
0 ||
!st.wt.load(o.pack, arena_s, pool_bytes, err,
skip.empty() ? nullptr : &skip)) {
pool_s, dpct::get_in_order_queue())) != 0 ||
!st.wt.load(o.pack, arena_s, pool_s, err,
skip_s.empty() ? nullptr : &skip_s)) {
/*
DPCT1026: The call to cudaGetLastError was removed because this
functionality is redundant in SYCL.
@@ -2494,7 +2743,7 @@ int main(int argc, char **argv) try {
std::fprintf(stderr, "strata generate: layer split, CUDA%d weights: %s (%llu MiB needed, %llu MiB of %llu "
"MiB free on that card)\n", st.dev,
err.empty() ? "the weight arena does not fit" : err.c_str(),
(unsigned long long) (pool_bytes >> 20), (unsigned long long) (free_b >> 20),
(unsigned long long) (pool_s >> 20), (unsigned long long) (free_b >> 20),
(unsigned long long) (total_b >> 20));
return 1;
}
@@ -2871,8 +3120,9 @@ int main(int argc, char **argv) try {
const strata::core::WeightRef* wo_s = st.wt.find("output.weight");
if (wo_s == nullptr ||
(last && !o.native_head_gguf.empty() && !st.head.load(o.native_head_shards, g.n_embd, wo_s->ne1, err))) {
std::fprintf(stderr, "strata generate: layer split, CUDA%d head: %s\n", st.dev,
wo_s == nullptr ? "output.weight is missing" : err.c_str());
std::fprintf(stderr, "strata generate: layer split, CUDA%d head: %s%s\n", st.dev,
wo_s == nullptr ? "output.weight is missing" : err.c_str(),
wo_s == nullptr ? "" : vram_free_note().c_str());
return 1;
}
size_t fb = 0, tb = 0;
@@ -2901,7 +3151,39 @@ int main(int argc, char **argv) try {
static const strata::core::ModelGeometry draft_geometry{};
// with a layer split across GPUs the drafter reads the last stage's residual: it lives on that device
const strata::core::OnDevice on_mtp(last_st ? last_st->dev : -1);
if (!o.mtp.empty() && !mtp.load(o.mtp, draft_geometry, last_st ? last_st->ss : ss, o.spec, err, o.mtp_window)) { std::fprintf(stderr, "strata generate: %s\n", err.c_str()); return 1; }
if (!o.mtp.empty() && !mtp.load(o.mtp, draft_geometry, last_st ? last_st->ss : ss, o.spec, err, o.mtp_window)) { std::fprintf(stderr, "strata generate: %s%s\n", err.c_str(), vram_free_note().c_str()); return 1; }
}
// THE HEAD BEFORE THE CACHE, AND BEFORE THE ARENA. The expert cache takes what is free minus the reserve, so
// everything allocated after it comes out of the reserve. The native head (~0.5 GB with IQ3_S) was loaded after
// it and ate most of the 700 MiB: 128K IQ3_S ended with 30 MiB free, the driver paged, and a request stalled for
// good at its first verify window. Loaded first, the cache is sized around it. #620: and before the expert
// arena registers tens of GiB of host pages, like the drafter above - WDDM then refused the head's cudaMalloc on
// a 16 GB card with the desktop on the iGPU ("native head upload: out of memory" with GiBs free).
const strata::core::WeightRef* wo = wt.find("output.weight");
if (wo == nullptr) { std::fprintf(stderr, "strata generate: output.weight is missing\n"); return 1; }
const int64_t n_vocab = wo->ne1;
strata::core::NativeHead native_head;
if (!o.native_head_gguf.empty() && !multi_gpu) { // a layer split's head is on its last stage
if (!native_head.load(o.native_head_shards, g.n_embd, n_vocab, err)) {
std::fprintf(stderr, "strata generate: %s%s\n", err.c_str(), vram_free_note().c_str());
return 1;
}
std::fprintf(stderr, "strata generate: experimental native Q5_K head, %llu bytes\n",
(unsigned long long) native_head.weight_bytes());
}
std::vector<float> logits((size_t) n_vocab);
float* d_logits = nullptr;
if (DPCT_CHECK_ERROR(
d_logits = (float *)sycl::malloc_device(
(size_t)n_vocab * 4, dpct::get_in_order_queue())) != 0) {
/*
DPCT1010: SYCL uses exceptions to report errors and does not use
the error codes. The cudaGetLastError function call was replaced with 0.
You need to rewrite this code.
*/
(void)0;
std::fprintf(stderr, "strata generate: the logits buffer failed%s\n", vram_free_note().c_str());
return 1;
}
// Create the additional contexts after MTP has secured CUDA0 memory, but
// before the host arena maps its expert pages into their address spaces.
@@ -3078,6 +3360,8 @@ int main(int argc, char **argv) try {
std::fprintf(stderr, "strata generate: %s\n", err.c_str());
return 1;
}
if (!arena_src.ram_warning().empty()) // #633: said before the load's numbers, which it explains
std::fprintf(stderr, "strata generate: WARNING: %s\n", arena_src.ram_warning().c_str());
std::fprintf(stderr, "strata generate: expert arena: %s\n", arena_src.note().c_str());
std::fprintf(stderr, "strata generate: loaded %.2f GiB at %.2f GiB/s\n",
(double) strata::kernels::cpu::expert_layout().total / (1024.0 * 1024 * 1024),
@@ -3112,30 +3396,7 @@ int main(int argc, char **argv) try {
// numbers if the slots do not fit, instead of handing back a cache smaller than it was asked for.
mem_mark("the weights, the session and the drafter");
strata::core::ExpertCache xcache;
// THE HEAD BEFORE THE CACHE. The expert cache takes what is free minus the reserve, so everything allocated
// after it comes out of the reserve. The native head (~0.5 GB with IQ3_S) was loaded after it and ate most of
// the 700 MiB: 128K IQ3_S ended with 30 MiB free, the driver paged, and a request stalled for good at its first
// verify window. Loaded first, the cache is sized around it.
const strata::core::WeightRef* wo = wt.find("output.weight");
if (wo == nullptr) { std::fprintf(stderr, "strata generate: output.weight is missing\n"); return 1; }
const int64_t n_vocab = wo->ne1;
strata::core::NativeHead native_head;
if (!o.native_head_gguf.empty() && !multi_gpu) { // a layer split's head is on its last stage
if (!native_head.load(o.native_head_shards, g.n_embd, n_vocab, err)) {
std::fprintf(stderr, "strata generate: %s\n", err.c_str());
return 1;
}
std::fprintf(stderr, "strata generate: experimental native Q5_K head, %llu bytes\n",
(unsigned long long) native_head.weight_bytes());
}
std::vector<float> logits((size_t) n_vocab);
float* d_logits = nullptr;
if (DPCT_CHECK_ERROR(
d_logits = (float *)sycl::malloc_device(
(size_t)n_vocab * 4, dpct::get_in_order_queue())) != 0) {
std::fprintf(stderr, "strata generate: the logits buffer failed\n");
return 1;
}
// THE HEAD BEFORE THE CACHE: the native head and the logits are allocated above, before the expert arena (#620)
const bool auto_cache = o.expert_cache < 0;
bool reserve_adapted = false; // #496: the auto sizing lowered the reserve so a small card's cache fits
if (o.expert_cache < 0) {
@@ -3262,6 +3523,24 @@ int main(int argc, char **argv) try {
}
o.expert_cache = (int) sized_slots.size();
}
// #533: --vram-elastic: the cache in physical segments (the VRAM command resizes it between requests). One GPU,
// no helper caches, serve mode: anything else keeps the one cudaMalloc, said once.
if (o.vram_elastic) {
const char* why = !o.serve ? "it works between requests of --serve"
: multi_gpu ? "a layer split has a cache per GPU"
: std::any_of(o.expert_cache_remote.begin(), o.expert_cache_remote.end(),
[](int n) { return n > 0; }) ? "the helper caches on other GPUs"
: o.vram_segment_mib < 64 ? "--vram-segment-mib is below 64" : nullptr;
#if defined(STRATA_USE_HIP)
if (why == nullptr) why = "it is NVIDIA-only for now";
#endif
if (why != nullptr) {
std::fprintf(stderr, "strata generate: --vram-elastic is off: %s\n", why);
o.vram_elastic = false;
} else {
xcache.set_segment_bytes(o.vram_segment_mib << 20);
}
}
if (o.expert_cache > 0) {
// keep the first `keep_bytes` of the cache (the profile's hottest experts first); false when nothing is left
auto shrink_to = [&](int64_t keep_bytes) -> bool {
@@ -3562,6 +3841,17 @@ int main(int argc, char **argv) try {
// to hold back is the windows - and `free_b` has already lost the drafter.
const int64_t room = stage_room(st.dev, true, false);
const strata::core::OnDevice on(st.dev);
{
size_t fb = 0, tb = 0;
/*
DPCT1106: 'cudaMemGetInfo' was migrated with the Intel
extensions for device information which may not be supported by all
compilers or runtimes. You may need to adjust the code.
*/
dpct::get_current_device().get_memory_info(fb, tb);
std::fprintf(stderr, "strata generate: layer split, CUDA%d: %.2f GiB free of %.2f, room for experts %.2f GiB\n",
st.dev, (double) fb / 1073741824.0, (double) tb / 1073741824.0, (double) room / 1073741824.0);
}
std::vector<int64_t> sized;
int64_t used = 0;
for (const auto& pr : st.profile) {
@@ -4101,7 +4391,12 @@ int main(int argc, char **argv) try {
strata::core::doorbell_reset(db);
double mix = 0, ffn = 0, post = 0;
if (!strata::core::session_replay_stages(g, 0, 0, ss, gr, main_cs, mix, ffn, post, err)) {
std::fprintf(stderr, "strata generate: %s\n", err.c_str());
// #610: a native (IQ) pack captures no per-layer graphs; the serve path times its window's stages
std::fprintf(stderr, "strata generate: %s%s\n", err.c_str(),
native_pack ? " - a native (IQ) pack has no per-layer graphs to replay. Its decode window's GPU "
"stages are printed per request by the serve path instead: STRATA_VERIFY_PROFILE=1 "
"STRATA_DECODE_TIMING=1 with --serve (docs/DETAILS.md, \"Where a decode window's "
"time goes\")" : "");
return 1;
}
const int reps = 20;
@@ -4422,6 +4717,31 @@ int main(int argc, char **argv) try {
}
thits.d_res = d_res;
thits.n_expert = g.n_expert;
// a file-backed arena (STRATA_ARENA_MMAP): the experts no GPU holds are the ones the CPU pool and the
// prompt path will read - start reading them now instead of faulting them in 4 KB at a time mid-request
// ... and the ones a GPU holds are handed back first (STRATA_ARENA_RELEASE=0 keeps them): the fill read the
// whole arena, and left mapped and referenced it crowds every other allocation into swap
{
static const bool keep = std::getenv("STRATA_ARENA_RELEASE") && std::getenv("STRATA_ARENA_RELEASE")[0] == '0';
rss_probe("before the release");
uint64_t released = 0;
if (!keep)
for (size_t i = 0; i < host_res.size(); ++i)
if (host_res[i] != strata::core::kNotResident)
released += srcp->release((int64_t) i / g.n_expert, (int64_t) i % g.n_expert);
if (released > 0)
std::fprintf(stderr, "strata generate: expert arena: %.2f GiB of VRAM-held experts handed back to the OS\n",
(double) released / (1024.0 * 1024.0 * 1024.0));
rss_probe("after the release");
int64_t pf = 0;
for (size_t i = 0; i < host_res.size(); ++i)
if (host_res[i] == strata::core::kNotResident) {
srcp->prefetch((int64_t) i / g.n_expert, (int64_t) i % g.n_expert);
++pf;
}
(void) pf;
rss_probe("after the prefetch hints");
}
for (auto& st : stages) { // layer split across GPUs: the same table on every device
const strata::core::OnDevice on(st->dev);
if (DPCT_CHECK_ERROR(
@@ -4629,6 +4949,19 @@ int main(int argc, char **argv) try {
std::fprintf(stderr, "strata generate: CPU expert residency: %s\n", err.c_str());
return 1;
}
// #577: the unbuffered choice above was made before the RAM copy existed, from the budget asked for; now the
// copy is built, decide again from the RAM it really holds and the expert bytes outside it (on a 96 GB PC
// the file cache keeps those, and every refill after a prompt read the drive instead). Windows only: the
// unbuffered reads exist there alone
#if defined(_WIN32)
if (o.mmap_experts && o.resident_budget > 0) {
std::string why;
const bool was = src.unbuffered();
const bool ub = src.recheck_unbuffered(why);
std::fprintf(stderr, "strata generate: the file tier reads %s%s (%s)\n",
ub ? "unbuffered" : "through the file cache", ub == was ? "" : " (changed)", why.c_str());
}
#endif
}
if (o.serve) {
if (o.spec < 2 || o.mtp.empty() || o.prefill_chunk <= 0 ||
@@ -4866,6 +5199,7 @@ int main(int argc, char **argv) try {
} else {
std::fprintf(stderr, "strata serve: the prompt path allocates its own buffers (too few cache slots to borrow)\n");
}
rss_probe("the prompt path set up");
// layer split across GPUs: a prompt path per stage, each handing its chunk's rows to the next.
// THE CHUNK STEPS DOWN INSTEAD OF EXITING. A split's stage caches are sized after the arena is registered, and
// the prompt path's own buffers (no loan) are not priced into them: with the whole arena pinned (#253) a
@@ -5275,10 +5609,33 @@ int main(int argc, char **argv) try {
std::vector<int64_t> part_next(stages.size() + 1, INT64_MAX);
// a checkpoint of the state after `cur[0, L)`; false only when the copy itself failed. `parts`: the stages'
// states saved at L (a split's mid-prompt checkpoint); without, they are read now (everything is at L)
auto checkpoint_at = [&](int64_t L, std::vector<ConvCheckpoint> *parts =
nullptr) -> bool {
// #613: why the last checkpoint_at failed - a failed device sync is the GPU itself (a hang Windows then resets),
// not the checkpoint, and the error says so
std::string ckpt_why;
auto gpu_sync_ok = [&]() -> bool {
try {
if (o.prompt_cache <= 0 || L < 1) return true;
const dpct::err0 e = DPCT_CHECK_ERROR(
dpct::get_current_device().queues_wait_and_throw());
if (e == 0) return true;
/*
DPCT1009: SYCL reports errors using exceptions and does not use
error codes. Please replace the "get_error_string_dummy(...)" with a
real error-handling function.
*/
ckpt_why = std::string(": the GPU stopped responding before it (") +
dpct::get_error_string_dummy(e) +
") - a GPU hang; on Windows the driver is then reset";
return false;
}
catch (sycl::exception const &exc) {
std::cerr << exc.what() << "Exception caught at file:" << __FILE__
<< ", line:" << __LINE__ << std::endl;
std::exit(1);
}
};
auto checkpoint_at = [&](int64_t L, std::vector<ConvCheckpoint>* parts = nullptr) -> bool {
ckpt_why.clear();
if (o.prompt_cache <= 0 || L < 1) return true;
for (ConvCheckpoint& c : checks)
if ((int64_t) c.ids.size() == L) { c.used = ++check_clock; return true; }
ConvCheckpoint c;
@@ -5293,17 +5650,12 @@ int main(int argc, char **argv) try {
c.block_pos = std::move((*parts)[0].block_pos);
for (size_t i = 1; i < parts->size(); ++i) c.stage_parts.push_back(std::move((*parts)[i]));
} else {
if (DPCT_CHECK_ERROR(
dpct::get_current_device().queues_wait_and_throw()) !=
0 ||
!checkpoint_save(c, ss, g)) return false;
if (!gpu_sync_ok() || !checkpoint_save(c, ss, g)) return false;
for (auto& st : stages) { // a layer split's later stages: their sessions' part
const strata::core::OnDevice on(st->dev);
ConvCheckpoint part;
part.ids = c.ids;
if (DPCT_CHECK_ERROR(dpct::get_current_device()
.queues_wait_and_throw()) != 0 ||
!checkpoint_save(part, st->ss, g)) return false;
if (!gpu_sync_ok() || !checkpoint_save(part, st->ss, g)) return false;
c.stage_parts.push_back(std::move(part));
}
}
@@ -5318,12 +5670,6 @@ int main(int argc, char **argv) try {
checks.erase(checks.begin() + (std::ptrdiff_t) victim);
}
return true;
}
catch (sycl::exception const &exc) {
std::cerr << exc.what() << "Exception caught at file:" << __FILE__
<< ", line:" << __LINE__ << std::endl;
std::exit(1);
}
};
sp.on_chunk = [&](const float* R_rows, int64_t T, int64_t p0, std::string& e) -> bool {
std::vector<int32_t> nxt((size_t) T);
@@ -5359,7 +5705,7 @@ int main(int argc, char **argv) try {
} else {
saved = checkpoint_at(done);
}
if (!saved) { e = "saving a conversation checkpoint failed"; return false; }
if (!saved) { e = "saving a conversation checkpoint failed" + ckpt_why; return false; }
pp_next_check = done + o.prompt_cache_every;
}
return true;
@@ -5411,6 +5757,7 @@ int main(int argc, char **argv) try {
}
// plan v0.3 P6: swaps in flight - (residency index, slot) admitted when adapt_ev has completed
std::vector<std::pair<int32_t, int32_t>> pending;
std::vector<void*> pin_live; // the swaps' locked arena pages (pin_blob), unlocked once they have landed
dpct::event_ptr adapt_ev = nullptr;
adapt_ev = new sycl::event();
// a layer split's later stages keep a copy of the residency table on their devices, and swap on their own
@@ -5456,8 +5803,12 @@ int main(int argc, char **argv) try {
sycl::info::event_command_status::complete) return;
}
for (auto& st : stages) st->adapt_live = false;
unpin_blobs(pin_live);
src.commit_exchanges(); // the resident RAM mode: the evicted experts take their places in RAM
for (const auto& [i, slot] : pending) host_res[(size_t) i] = slot;
for (const auto& [i, slot] : pending) {
host_res[(size_t) i] = slot;
srcp->release((int64_t) i / g.n_expert, (int64_t) i % g.n_expert); // in VRAM now: RAM not needed
}
pending.clear();
res_upload();
}
@@ -5495,6 +5846,14 @@ int main(int argc, char **argv) try {
std::sort(swaps.begin(), swaps.end(), [](const Swap& a, const Swap& b) { return a.gain > b.gain; });
if ((int) swaps.size() > o.adapt_swaps) swaps.resize((size_t) o.adapt_swaps);
if (!resident_stage_swaps(src, xcache, host_res, g.n_expert, swaps, adapt_stream)) return false;
if (pin_blobs_on()) { // lock the batch's source pages (a file-backed arena on AMD; see pin_blobs)
std::vector<std::pair<uintptr_t, uintptr_t>> spans;
for (const Swap& s : swaps)
if (const uint8_t* b = srcp->blob(s.layer, s.in))
spans.emplace_back((uintptr_t) b,
(uintptr_t) b + strata::kernels::cpu::expert_layout().blob_bytes(s.layer));
pin_blobs(std::move(spans), pin_live);
}
bool main_live = false;
for (const Swap& s : swaps) {
const size_t in = (size_t) s.layer * g.n_expert + s.in, out = (size_t) s.layer * g.n_expert + s.out;
@@ -5517,11 +5876,31 @@ int main(int argc, char **argv) try {
: xcache.device_slot(slot),
b,
(size_t)strata::kernels::cpu::expert_layout()
.blob_bytes(s.layer))) != 0)
.blob_bytes(s.layer))) != 0) {
std::fprintf(stderr,
"strata serve: adaptive swap copy failed "
"(layer %d, slot %d, pinned %d): %s\n",
/*
DPCT1009: SYCL reports errors using
exceptions and does not use error codes. Please
replace the "get_error_string_dummy(...)" with
a real error-handling function.
*/
/*
DPCT1010: SYCL uses exceptions to report
errors and does not use the error codes. The
cudaGetLastError function call was replaced
with 0. You need to rewrite this code.
*/
(int)s.layer, (int)slot,
pin_live.empty() ? 0 : 1,
dpct::get_error_string_dummy(0));
return false;
}
if (gs) gs->adapt_live = true;
else main_live = true;
host_res[out] = strata::core::kNotResident; // evicted now: the CPU computes it meanwhile
srcp->prefetch(s.layer, s.out); // a file-backed arena released its pages: read them back ahead
pending.emplace_back((int32_t) in, slot); // resident once the copy has landed
}
if (!swaps.empty()) dpct::sync_barrier(adapt_ev, adapt_stream);
@@ -5643,6 +6022,7 @@ int main(int argc, char **argv) try {
const int64_t free_mib = (int64_t) (free_b >> 20);
if (free_mib >= 256) {
std::fprintf(stderr, "strata serve: %lld MiB of VRAM free with everything loaded\n", (long long) free_mib);
rss_probe("serving");
} else if (reserve_adapted) {
// #496: the reserve was already lowered to make the cache fit - a bigger one would leave it no room
std::fprintf(stderr, "strata serve: WARNING: %lld MiB of VRAM free with everything loaded - this card "
@@ -5684,7 +6064,7 @@ int main(int argc, char **argv) try {
"expert_slots_primary=%lld expert_cache_primary_mib=%lld spec=%d "
"mtp_max=%d lookup=%d vram_free_mib=%lld cvec=%s arena_mib=%lld pool_workers=%d pcie_frac=%.2f "
"spec_min_p=%.2f conversation_cache_mib=%lld conversation_cache_slots=%d "
"conversation_cache_min_free_mib=%lld engine=" STRATA_VERSION "\n",
"conversation_cache_min_free_mib=%lld tail_role_token=%lld vram_elastic=%d engine=" STRATA_VERSION "\n",
(long long) o.max_context, o.kv.c_str(),
(long long) (g.n_qsa_layers() > 0 && ss.qsa_states[ss.qsa_primary()].kv_mode == 1
? ss.qsa_states[ss.qsa_primary()].n_slots * 4 : 0),
@@ -5695,7 +6075,7 @@ int main(int argc, char **argv) try {
(long long) ((o.mmap_experts ? src.resident_bytes() : strata::kernels::cpu::expert_layout().total) >> 20),
pool.workers(), o.pcie_frac,
o.spec_min_p, (long long) o.conversation_cache_mib, o.conversation_cache_slots,
(long long) o.conversation_cache_min_free_mib);
(long long) o.conversation_cache_min_free_mib, (long long) o.tail_role_token, xcache.segmented() ? 1 : 0);
}
// issue #29: a request whose heartbeat (tokens, prompt chunks, verify windows) stops for this long is stuck on
// a flag nobody will raise - end the engine with where it was, so the server starts it again instead of the
@@ -5742,7 +6122,162 @@ int main(int argc, char **argv) try {
bool mrope_identity = true;
std::vector<float> img_rows;
std::vector<const float*> row_ptr;
// ---- #533: VRAM <reserve_mib>, between requests, only with --vram-elastic. It shrinks the expert cache
// until that much VRAM is free for other programs (a game, a CAD session), or grows it back towards its full
// size when more than that is free. Never on its own: only this command (the server's POST /v1/vram) moves
// it. A shrink gives back the cache's LAST segments: the experts in their slots become CPU misses, as any
// expert outside the cache is (no output changes beyond what a smaller cache decodes), the prompt path's
// loan moves down to the end of what is left (a smaller chunk when it no longer fits). A grow maps them
// again and puts each slot's expert back (or, when the adaptive tier already brought that one back, the
// most-routed missing expert of the same layer); the adaptive tier carries on from there.
std::vector<std::pair<int32_t, int32_t>> vram_evicted; // (residency index, slot) the shrinks took
const int64_t chunk_full = o.prefill_chunk; // the loan's chunk with the whole cache
auto relend = [&]() { // the prompt path's loan: the end of the slots left, its chunk as large as fits
if (pf_parts.empty() || pf_parts[0].first < 0) return;
PfPart& p = pf_parts[0];
const int64_t live = xcache.slots();
auto fits = [&](int64_t c) {
const int64_t k = part_slots(p, c);
return k > 0 && k + 128 <= live && !(o.prefill_auto && k * 100 > kAutoLendPct * live);
};
int64_t c = chunk_full;
while (c > 256 && !fits(c)) c = std::max<int64_t>(256, c / 2 / 256 * 256);
o.prefill_chunk = c;
p.first = (int32_t) std::max<int64_t>(0, live - part_slots(p, c));
p.first_now = -1; // laid out again at the next loan
};
auto vram_command = [&](const std::string &cmd,
std::string &e) -> bool {
// `VRAM` alone: the reserve the engine started with (--vram-reserve-mib)
try {
char *end = nullptr;
const bool bare = cmd.find_first_not_of(' ', 4) == std::string::npos;
const long long reserve = bare ? (long long) o.vram_reserve_mib : std::strtoll(cmd.c_str() + 4, &end, 10);
if (!bare && (end == cmd.c_str() + 4 || reserve < 0)) { e = "expected: VRAM [reserve_mib]"; return false; }
if (!xcache.segmented()) {
e = "VRAM needs an engine started with --vram-elastic (one NVIDIA GPU, --serve)";
return false;
}
if (src.complement_ready()) {
e = "VRAM: not with the resident low-RAM mode (the cache holds experts RAM does not)";
return false;
}
if (peer.valid()) { e = "VRAM: not with --peer-device"; return false; }
for (const PfPart& p : pf_parts)
if (!p.lent.empty()) { e = "VRAM: a prompt's loan is still out"; return false; }
if (!ver.wait_commit(e)) return false;
apply_pending(true); // the adaptive tier's swaps in flight land first
if (DPCT_CHECK_ERROR(
dpct::get_current_device().queues_wait_and_throw()) != 0) {
/*
DPCT1009: SYCL reports errors using exceptions and does not
use error codes. Please replace the
"get_error_string_dummy(...)" with a real error-handling
function.
*/
/*
DPCT1010: SYCL uses exceptions to report errors and does
not use the error codes. The cudaGetLastError function call was
replaced with 0. You need to rewrite this code.
*/
e = std::string("VRAM: ") + dpct::get_error_string_dummy(0);
return false;
}
const auto t0 = Clock::now();
size_t free_b = 0, total_b = 0;
/*
DPCT1106: 'cudaMemGetInfo' was migrated with the Intel
extensions for device information which may not be supported by all
compilers or runtimes. You may need to adjust the code.
*/
dpct::get_current_device().get_memory_info(free_b, total_b);
const int64_t want_free = (int64_t) reserve << 20, mapped = xcache.mapped_bytes();
const int64_t before = xcache.slots();
// what stays at least: the prompt path's smallest loan (256 tokens) and the 128 slots it must leave
const int64_t floor_slots = std::min<int64_t>(xcache.full_slots(),
128 + (pf_parts.empty() || pf_parts[0].first < 0 ? 0 : part_slots(pf_parts[0], 256)));
std::string note;
if ((int64_t) free_b < want_free) {
const int64_t floor_b = xcache.bytes_of(floor_slots);
int64_t keep = mapped - (want_free - (int64_t) free_b);
if (keep < floor_b) {
keep = floor_b;
note = " (the cache keeps its smallest size: the prompt path's buffers)";
}
if (!xcache.shrink(keep, e)) return false;
const int64_t live = xcache.slots();
for (size_t i = 0; i < host_res.size(); ++i)
if (host_res[i] >= live) {
vram_evicted.emplace_back((int32_t) i, host_res[i]);
host_res[i] = strata::core::kNotResident;
}
res_upload();
} else if (mapped < xcache.full_bytes()) {
std::string gerr;
if (!xcache.grow(mapped + ((int64_t) free_b - want_free), gerr)) note = " (" + gerr + ")";
const int64_t live = xcache.slots();
std::string ferr;
std::vector<std::pair<int32_t, int32_t>> keep_out;
for (const auto& [i, slot] : vram_evicted) {
if (slot >= live) { keep_out.emplace_back(i, slot); continue; }
const int64_t layer = i / g.n_expert;
int64_t pick = i;
if (host_res[(size_t) i] >= 0) { // back already (the adaptive tier): the layer's most-routed miss
pick = -1;
float best = -1.0f;
for (int64_t ex = 0; ex < g.n_expert; ++ex) {
const size_t j = (size_t) (layer * g.n_expert + ex);
if (host_res[j] >= 0) continue;
const float u = drive.d.usage.empty() ? 0.0f : drive.d.usage[j];
if (u > best) { best = u; pick = (int64_t) j; }
}
}
if (pick < 0) continue;
const uint8_t* b = srcp->blob(layer, pick % g.n_expert);
if (b == nullptr || !xcache.fill_slot_queued(slot, b, ferr,
(int64_t) strata::kernels::cpu::expert_layout().blob_bytes(layer))) {
e = "VRAM: refilling the cache failed: " + ferr;
return false;
}
host_res[(size_t) pick] = slot;
}
vram_evicted.swap(keep_out);
if (!xcache.sync_queued(ferr)) { e = "VRAM: " + ferr; return false; }
res_upload();
}
relend();
/*
DPCT1106: 'cudaMemGetInfo' was migrated with the Intel
extensions for device information which may not be supported by all
compilers or runtimes. You may need to adjust the code.
*/
dpct::get_current_device().get_memory_info(free_b, total_b);
std::fprintf(stderr, "strata serve: VRAM %lld MiB kept free: the expert cache %lld -> %lld of %lld slots "
"(%.2f of %.2f GiB), %lld MiB free now, prompt chunk %lld, in %.0f ms%s\n",
reserve, (long long) before, (long long) xcache.slots(), (long long) xcache.full_slots(),
(double) xcache.mapped_bytes() / 1073741824.0, (double) xcache.full_bytes() / 1073741824.0,
(long long) (free_b >> 20), (long long) o.prefill_chunk,
std::chrono::duration<double, std::milli>(Clock::now() - t0).count(), note.c_str());
std::printf("VRAM reserve_mib=%lld expert_slots=%lld expert_slots_full=%lld expert_cache_mib=%lld "
"expert_cache_full_mib=%lld vram_free_mib=%lld prompt_chunk=%lld\n", reserve,
(long long) xcache.slots(), (long long) xcache.full_slots(),
(long long) (xcache.mapped_bytes() >> 20), (long long) (xcache.full_bytes() >> 20),
(long long) (free_b >> 20), (long long) o.prefill_chunk);
return true;
}
catch (sycl::exception const &exc) {
std::cerr << exc.what() << "Exception caught at file:" << __FILE__
<< ", line:" << __LINE__ << std::endl;
std::exit(1);
}
};
while (next_line(line)) {
if (line.rfind("VRAM", 0) == 0) { // #533 (above): between requests, not a request
std::string verr;
if (!vram_command(line, verr)) std::printf("ERR %s\n", verr.c_str());
std::fflush(stdout);
continue;
}
// #477: every --expert-profile-save-every minutes, before the next request (at QUIT: after the loop)
if (!heat.empty() && line != "QUIT" && o.expert_profile_save_min > 0 &&
Clock::now() - profile_saved_at >= std::chrono::duration<double>(o.expert_profile_save_min * 60.0))
@@ -6319,6 +6854,15 @@ int main(int argc, char **argv) try {
if (o.prompt_cache > 0 && o.turn_token >= 0)
for (int64_t i = n - 1; i > resume; --i)
if (ids[(size_t) i] == o.turn_token) { turn_at = i; break; }
// #458 (opt-in): a short turn of --tail-role-token's role right before the new assistant turn (the
// server's trailing reasoning-effort turn) stays out of the checkpoint, so the next request - another
// effort, or the next turn of the chat - finds the conversation without it
if (turn_at > 0 && o.tail_role_token >= 0)
for (int64_t i = turn_at - 1; i > resume; --i)
if (ids[(size_t) i] == o.turn_token) {
if (ids[(size_t) i + 1] == o.tail_role_token) turn_at = i;
break;
}
// A prompt read from token 0 also stops at its FIRST turn boundary: the end of the system prompt (with
// the tools), which every new chat of the same client shares. That checkpoint becomes the chain's root,
// which the retention policy pins (conv_cache.hpp), so the next new chat reads only what comes after it.
@@ -6376,7 +6920,7 @@ int main(int argc, char **argv) try {
}
at = to;
if ((to == turn_at || to == root_at) && !checkpoint_at(to)) {
std::printf("ERR saving a conversation checkpoint failed\n");
std::printf("ERR saving a conversation checkpoint failed%s\n", ckpt_why.c_str());
return 1;
}
}
@@ -6426,6 +6970,7 @@ int main(int argc, char **argv) try {
const int64_t ram0 = src.ram_reads(), files0 = src.file_reads();
const uint64_t file_bytes0 = src.file_read_bytes();
const int64_t decode_look0 = drive.d.cache_hits + drive.d.cache_admitted + drive.d.cache_refused;
const int64_t offload0 = drive.d.offload_entries; // #588
if (cancelled) finish = "cancel";
while (!cancelled && produced_n < max_new) {
int T = S_mtp;
@@ -6715,6 +7260,7 @@ int main(int argc, char **argv) try {
}
const int64_t req_hits = drive.d.cache_hits - decode_hits0;
const int64_t req_look = (drive.d.cache_hits + drive.d.cache_admitted + drive.d.cache_refused) - decode_look0;
const int64_t req_offload = drive.d.offload_entries - offload0;
// #471: the prompt tokens this request read - all the fresh ones, or as far as the prompt pass got when a
// cancel stopped it part-way (a cancelled request used to be logged and counted as having read them all)
const int64_t fresh = n - resume;
@@ -6722,11 +7268,15 @@ int main(int argc, char **argv) try {
// DONE <generated> <prompt> <prompt ms> <decode ms> <finish> <drafts accepted> <drafts offered> <reused> [hits] [lookups]
// [RAM blobs] [file blobs] [file MB] (CS-T tiers; appended, so an older server reads the rest)
// [prompt tokens read] (#471: fewer than <prompt> - <reused> when a cancel stopped the read)
std::printf("DONE %lld %lld %.1f %.1f %s %lld %lld %lld %lld %lld %lld %lld %.1f %lld\n", (long long) produced_n,
// [offloaded] (#588: the decode's routed experts the GPU read over PCIe or another GPU computed;
// not in [lookups])
std::printf("DONE %lld %lld %.1f %.1f %s %lld %lld %lld %lld %lld %lld %lld %.1f %lld %lld\n",
(long long) produced_n,
(long long) n, prompt_ms, decode_ms, finish, (long long) draft_accepted, (long long) draft_offered,
(long long) resume, (long long) req_hits, (long long) req_look,
(long long) (src.ram_reads() - ram0), (long long) (src.file_reads() - files0),
(double) (src.file_read_bytes() - file_bytes0) / 1e6, (long long) read_n);
(double) (src.file_read_bytes() - file_bytes0) / 1e6, (long long) read_n,
(long long) req_offload);
std::fflush(stdout);
if (drive.routing != nullptr) std::fflush(drive.routing); // the routing trace survives a crash and is watchable mid-session
// "12288 of 98179" when cancelled mid-read (#471), the rate from what was read
@@ -6742,11 +7292,18 @@ int main(int argc, char **argv) try {
decode_ms > 0 ? 1000.0 * produced_n / decode_ms : 0.0, (long long) draft_accepted,
(long long) draft_offered, checks.size(), cancelled ? " (cancelled)" : "");
// the VRAM share of the experts the pool looked up while decoding; experts it sent over PCIe for the GPU
// to read (--pcie-frac) are in neither count
// to read (--pcie-frac) are in neither count - #588: so that share is said beside it (raising --pcie-frac
// raises the hit rate while the PCIe reads may make the decode slower)
if (req_look > 0) {
std::fprintf(stderr, "strata serve: decode expert cache hit rate: %.1f%% (%lld hits / %lld lookups)\n",
char off[160] = "";
if (req_offload > 0)
std::snprintf(off, sizeof off, "; %lld more read by the GPU over PCIe or from another GPU (%.1f%% of "
"all %lld routed)", (long long) req_offload,
100.0 * (double) req_offload / (double) (req_look + req_offload),
(long long) (req_look + req_offload));
std::fprintf(stderr, "strata serve: decode expert cache hit rate: %.1f%% (%lld hits / %lld lookups)%s\n",
100.0 * (double) req_hits / (double) req_look,
(long long) req_hits, (long long) req_look);
(long long) req_hits, (long long) req_look, off);
}
// the resident RAM mode, cumulative: experts read from experts.bin since the copy was made (what the plain
// mmap mode reads through the OS file cache, from the SSD when the RAM could not keep it)
@@ -7273,6 +7830,7 @@ int main(int argc, char **argv) try {
}
// plan v0.3 P6: swaps in flight - (residency index, slot) admitted when adapt_ev has completed
std::vector<std::pair<int32_t, int32_t>> pending;
std::vector<void*> pin_live; // the swaps' locked arena pages (pin_blob), unlocked once they have landed
dpct::event_ptr adapt_ev = nullptr;
adapt_ev = new sycl::event();
int64_t adapt_rounds = 0; // counted here: `rounds` is declared below the adapt lambda
@@ -7292,8 +7850,12 @@ int main(int argc, char **argv) try {
}
if (trace_pending)
std::fprintf(stderr, "strata: PENDING landed, %zu experts become resident\n", pending.size());
unpin_blobs(pin_live);
src.commit_exchanges(); // the resident RAM mode: the evicted experts take their places in RAM
for (const auto& [i, slot] : pending) host_res[(size_t) i] = slot;
for (const auto& [i, slot] : pending) {
host_res[(size_t) i] = slot;
srcp->release((int64_t) i / g.n_expert, (int64_t) i % g.n_expert); // in VRAM now: RAM not needed
}
pending.clear();
if (d_res != nullptr)
/*
@@ -7356,6 +7918,14 @@ int main(int argc, char **argv) try {
std::fprintf(stderr, "strata generate: an adaptive refill failed (copying evicted experts back)\n");
return false;
}
if (pin_blobs_on()) { // lock the batch's source pages (a file-backed arena on AMD; see pin_blobs)
std::vector<std::pair<uintptr_t, uintptr_t>> spans;
for (const Swap& s : swaps)
if (const uint8_t* b = srcp->blob(s.layer, s.in))
spans.emplace_back((uintptr_t) b,
(uintptr_t) b + strata::kernels::cpu::expert_layout().blob_bytes(s.layer));
pin_blobs(std::move(spans), pin_live);
}
for (const Swap& s : swaps) {
const size_t in = (size_t) s.layer * g.n_expert + s.in, out = (size_t) s.layer * g.n_expert + s.out;
const int32_t slot = host_res[out];
@@ -7377,6 +7947,7 @@ int main(int argc, char **argv) try {
return false;
}
host_res[out] = strata::core::kNotResident; // evicted now: the CPU computes it meanwhile
srcp->prefetch(s.layer, s.out); // a file-backed arena released its pages: read them back ahead
pending.emplace_back((int32_t) in, slot); // resident once the copy has landed
}
if (!swaps.empty()) dpct::sync_barrier(adapt_ev, adapt_stream);