mirror of
https://github.com/Niko1221/Strata.git
synced 2026-11-07 06:15:58 -10:00
Start-up after another model: judge pinning by the emptiest NUMA node, say how much is file cache when RAM is short, opt-in STRATA_DROP_CACHE_ON_EXIT
This commit is contained in:
1 parent
cdbf458684
commit
8c80a3ba12
6 files changed
+180
No files matched your search
@@ -99,6 +99,17 @@ bool pin_step_fits(uint64_t available, uint64_t step, uint64_t reserve);
|
||||
/// Parses `key` ("MemFree") of a /proc/meminfo text; 0 when absent.
|
||||
uint64_t meminfo_bytes(const std::string& meminfo_text, const std::string& key);
|
||||
|
||||
/// 0.1.42 (NUMA report): the MemFree to judge the page-locked complement by. `global_free` is the whole host's MemFree;
|
||||
/// `node_free` each NUMA node's own (/sys/devices/system/node/node*/meminfo, empty or one entry on a single-node host).
|
||||
/// A dual-socket host whose second node is full of another model's file cache has plenty free in total but none where
|
||||
/// the driver may allocate: with two or more nodes the figure is the smaller of the total and (node count x the
|
||||
/// SMALLEST node's free), i.e. what an even spread would find. Only lowers the figure, so the pages are then taken in
|
||||
/// steps (which reclaim cache as they fault) instead of in one driver call; it never refuses and never caps.
|
||||
uint64_t numa_pin_free(uint64_t global_free, const std::vector<uint64_t>& node_free);
|
||||
/// The sentence appended to a "not enough RAM" start failure: how much of the RAM is file cache (Cached + Buffers of
|
||||
/// `meminfo_text`, mapped pages included) and what to do about it; "" when under 1 GiB of cache (nothing to say).
|
||||
std::string page_cache_hint(const std::string& meminfo_text);
|
||||
|
||||
/// Build compact offsets for experts absent from both the primary GPU cache and an optional second GPU tier.
|
||||
/// Kept CPU-only so selection and byte accounting can be tested without initializing a GPU.
|
||||
bool make_cache_complement_plan(
|
||||
|
||||
@@ -0,0 +1,70 @@
|
||||
"""Opt-in: give the model files' page cache back when the server stops (STRATA_DROP_CACHE_ON_EXIT=1).
|
||||
|
||||
Linux keeps a stopped model's GGUF/pack files in the page cache (MemAvailable counts it, `free` shows it as used), which
|
||||
helps a restart of the SAME model. On a host that then starts a different, large model (a dual-socket server whose second
|
||||
NUMA node holds the old model's cache) some owners prefer the pages back at once. Off by default. Only this server's own
|
||||
model files are advised, and only after the engine process has ended (its mappings are gone, so the pages can go).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def enabled(env=None) -> bool:
|
||||
env = os.environ if env is None else env
|
||||
return os.name != "nt" and env.get("STRATA_DROP_CACHE_ON_EXIT", "").strip().lower() in ("1", "true", "yes", "on")
|
||||
|
||||
|
||||
def model_files(paths) -> list[str]:
|
||||
"""The regular files named by `paths` (a file, or a directory walked recursively); missing ones are skipped."""
|
||||
out: list[str] = []
|
||||
for p in paths:
|
||||
if not p:
|
||||
continue
|
||||
q = Path(p)
|
||||
try:
|
||||
if q.is_file():
|
||||
out.append(str(q))
|
||||
elif q.is_dir():
|
||||
out.extend(str(f) for f in sorted(q.rglob("*")) if f.is_file())
|
||||
except OSError:
|
||||
continue
|
||||
return out
|
||||
|
||||
|
||||
def drop(paths, advise=None) -> int:
|
||||
"""posix_fadvise(DONTNEED) on every file of `paths`; returns how many files were advised. Never raises."""
|
||||
advise = advise or getattr(os, "posix_fadvise", None)
|
||||
if advise is None:
|
||||
return 0
|
||||
n = 0
|
||||
for f in model_files(paths):
|
||||
try:
|
||||
fd = os.open(f, os.O_RDONLY)
|
||||
except OSError:
|
||||
continue
|
||||
try:
|
||||
advise(fd, 0, 0, getattr(os, "POSIX_FADV_DONTNEED", 4))
|
||||
n += 1
|
||||
except OSError:
|
||||
pass
|
||||
finally:
|
||||
os.close(fd)
|
||||
return n
|
||||
|
||||
|
||||
def drop_after_exit(engine, env=None, advise=None) -> int:
|
||||
"""Called once the engine has been closed. 0 when not opted in or the engine still runs."""
|
||||
if not enabled(env):
|
||||
return 0
|
||||
proc = getattr(engine, "proc", None)
|
||||
if proc is not None and proc.poll() is None:
|
||||
return 0
|
||||
spawn = getattr(engine, "spawn", None)
|
||||
args = list(spawn[1]) if spawn else []
|
||||
wanted = {"--native", "--pack", "--mmproj", "--model"}
|
||||
paths = [v for k, v in zip(args, args[1:]) if k in wanted]
|
||||
n = drop(paths or [getattr(engine, "model_path", None)], advise)
|
||||
print(f"[strata] STRATA_DROP_CACHE_ON_EXIT: released the page cache of {n} model files", flush=True)
|
||||
return n
|
||||
@@ -6277,6 +6277,11 @@ def main() -> int:
|
||||
except KeyboardInterrupt: # a second Ctrl+C: don't wait for the engine to free its memory
|
||||
if getattr(engine, "proc", None):
|
||||
engine.proc.kill()
|
||||
try:
|
||||
from serve import cache_drop
|
||||
cache_drop.drop_after_exit(engine) # opt-in (STRATA_DROP_CACHE_ON_EXIT=1), default unchanged
|
||||
except Exception as e: # noqa: BLE001 - never block the exit
|
||||
print(f"[strata] could not release the model files' cache: {e}", flush=True)
|
||||
print("[strata] stopped", flush=True)
|
||||
return 0
|
||||
|
||||
|
||||
@@ -0,0 +1,39 @@
|
||||
import os, tempfile, unittest
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from serve import cache_drop
|
||||
|
||||
|
||||
class CacheDrop(unittest.TestCase):
|
||||
def test_default_off(self):
|
||||
self.assertFalse(cache_drop.enabled({}))
|
||||
self.assertFalse(cache_drop.enabled({"STRATA_DROP_CACHE_ON_EXIT": "0"}))
|
||||
|
||||
@unittest.skipIf(os.name == "nt", "POSIX only")
|
||||
def test_on(self):
|
||||
self.assertTrue(cache_drop.enabled({"STRATA_DROP_CACHE_ON_EXIT": "1"}))
|
||||
|
||||
def test_files_and_advise(self):
|
||||
with tempfile.TemporaryDirectory() as d:
|
||||
Path(d, "a.gguf").write_bytes(b"x")
|
||||
os.mkdir(os.path.join(d, "sub"))
|
||||
Path(d, "sub", "b.bin").write_bytes(b"y")
|
||||
self.assertEqual(len(cache_drop.model_files([d, "/no/such", None])), 2)
|
||||
seen = []
|
||||
self.assertEqual(cache_drop.drop([d], advise=lambda fd, o, n, how: seen.append(how)), 2)
|
||||
self.assertEqual(len(seen), 2)
|
||||
|
||||
def test_not_while_engine_runs_or_not_opted_in(self):
|
||||
with tempfile.TemporaryDirectory() as d:
|
||||
env = {"STRATA_DROP_CACHE_ON_EXIT": "1"}
|
||||
calls = []
|
||||
run = lambda *a: calls.append(1)
|
||||
alive = SimpleNamespace(proc=SimpleNamespace(poll=lambda: None), spawn=("e", ["--native", d]), model_path=d)
|
||||
self.assertEqual(cache_drop.drop_after_exit(alive, env, run), 0)
|
||||
gone = SimpleNamespace(proc=SimpleNamespace(poll=lambda: 0), spawn=("e", ["--native", d]), model_path=d)
|
||||
self.assertEqual(cache_drop.drop_after_exit(gone, {}, run), 0)
|
||||
self.assertEqual(calls, [])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -115,6 +115,25 @@ uint64_t meminfo_bytes(const std::string& meminfo_text, const std::string& key)
|
||||
return 0;
|
||||
}
|
||||
|
||||
uint64_t numa_pin_free(uint64_t global_free, const std::vector<uint64_t>& node_free) {
|
||||
if (node_free.size() < 2) return global_free;
|
||||
const uint64_t smallest = *std::min_element(node_free.begin(), node_free.end());
|
||||
return std::min<uint64_t>(global_free, smallest * (uint64_t) node_free.size());
|
||||
}
|
||||
|
||||
std::string page_cache_hint(const std::string& meminfo_text) {
|
||||
const uint64_t cache = meminfo_bytes(meminfo_text, "Cached") + meminfo_bytes(meminfo_text, "Buffers");
|
||||
if (cache < (1ull << 30)) return "";
|
||||
char msg[420];
|
||||
std::snprintf(msg, sizeof msg,
|
||||
"; %.1f GiB of the RAM is file cache (earlier runs' model files, which Linux gives back on demand - "
|
||||
"'free' showing little is normal). If the start still fails, free it with: "
|
||||
"sync; echo 1 | sudo tee /proc/sys/vm/drop_caches (or start strata serve with "
|
||||
"STRATA_DROP_CACHE_ON_EXIT=1 so it releases its model files when it stops)",
|
||||
(double) cache / 1073741824.0);
|
||||
return msg;
|
||||
}
|
||||
|
||||
bool make_cache_complement_plan(
|
||||
int64_t n_layers, int64_t n_expert, const std::vector<uint64_t>& layer_blob_bytes,
|
||||
const std::vector<std::pair<int32_t, int32_t>>& primary_gpu_pairs,
|
||||
@@ -423,6 +442,21 @@ bool free_memory_bytes(uint64_t& bytes) {
|
||||
std::stringstream text;
|
||||
text << info.rdbuf();
|
||||
bytes = detail::meminfo_bytes(text.str(), "MemFree");
|
||||
if (bytes == 0) return false;
|
||||
// 0.1.42: a multi-socket host's nodes each have their own free pages; judge by the emptiest one too
|
||||
std::vector<uint64_t> nodes;
|
||||
for (int n = 0; n < 64; ++n) {
|
||||
std::ifstream node("/sys/devices/system/node/node" + std::to_string(n) + "/meminfo");
|
||||
if (!node) break;
|
||||
std::string line;
|
||||
while (std::getline(node, line)) {
|
||||
std::istringstream f(line);
|
||||
std::string a, id, key;
|
||||
uint64_t kb = 0;
|
||||
if (f >> a >> id >> key >> kb && key == "MemFree:") { nodes.push_back(kb * 1024); break; }
|
||||
}
|
||||
}
|
||||
bytes = detail::numa_pin_free(bytes, nodes);
|
||||
return bytes > 0;
|
||||
#else
|
||||
bytes = 0;
|
||||
@@ -2656,6 +2690,13 @@ bool FileExpertSource::pin_cache_complement(
|
||||
(double) bytes / 1073741824.0, commit < physical ? "commit" : "RAM",
|
||||
(double) available / 1073741824.0, (double) headroom_bytes / 1073741824.0);
|
||||
err = message;
|
||||
#if defined(__linux__)
|
||||
if (std::ifstream mi("/proc/meminfo"); mi) {
|
||||
std::stringstream mt;
|
||||
mt << mi.rdbuf();
|
||||
err += detail::page_cache_hint(mt.str());
|
||||
}
|
||||
#endif
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -383,6 +383,20 @@ void test_pin_pacing() {
|
||||
require(meminfo_bytes(info, "MemFree") == 1284096ull * 1024, "MemFree was misread");
|
||||
require(meminfo_bytes(info, "MemAvailable") == 33463296ull * 1024, "MemAvailable was misread");
|
||||
require(meminfo_bytes(info, "Mem") == 0 && meminfo_bytes(info, "Shmem") == 0, "an absent key was not 0");
|
||||
// 0.1.42 (dual-socket report): node 1 full of another model's cache, node 0 empty of it
|
||||
require(numa_pin_free(100 * GiB, {}) == 100 * GiB && numa_pin_free(100 * GiB, {100 * GiB}) == 100 * GiB,
|
||||
"a single-node host's figure changed");
|
||||
require(numa_pin_free(90 * GiB, {88 * GiB, 2 * GiB}) == 4 * GiB, "an emptied node did not lower the figure");
|
||||
require(numa_pin_free(40 * GiB, {20 * GiB, 20 * GiB}) == 40 * GiB, "balanced nodes changed the figure");
|
||||
require(numa_pin_free(10 * GiB, {50 * GiB, 50 * GiB}) == 10 * GiB, "the total must still bound it");
|
||||
require(pin_depends_on_reclaim(numa_pin_free(90 * GiB, {88 * GiB, 2 * GiB}), 26 * GiB, 4 * GiB),
|
||||
"an unbalanced two-node host was not paced");
|
||||
require(page_cache_hint(info).find("drop_caches") != std::string::npos &&
|
||||
page_cache_hint(info).find("STRATA_DROP_CACHE_ON_EXIT") != std::string::npos &&
|
||||
page_cache_hint(info).find("52.0 GiB") != std::string::npos, "the cache hint is incomplete");
|
||||
require(page_cache_hint("MemFree: 100 kB
|
||||
Cached: 1000 kB
|
||||
").empty(), "a small cache was reported");
|
||||
}
|
||||
|
||||
void test_resident_memory_budget() {
|
||||
|
||||
Reference in new issue
Block a user