Ling-3.0-tiny-RKNN / src /portable_main.cpp
Sariel00's picture
Publish Ling-3.0-tiny RKNN engine and model
3fd1a35 verified
Raw History Blame Contribute Delete
56.3 kB
#include "ling3/chat_protocol.h"
#include "ling3/chat_sampling.h"
#include "ling3/flow_gate.h"
#include "ling3/context_options.h"
#include "ling3/decoder.h"
#include "ling3/model_package.h"
#include "ling3/rknn_backend.h"
#include "ling3/runtime_requirements.h"
#include "ling3/tokenizer.h"
#include "ling3/w4_linear.h"
#include "ling3/state_cache.h"
#include "mla_npu.h"
#include <arpa/inet.h>
#include <sys/ioctl.h>
#include <sys/resource.h>
#include <sys/socket.h>
#include <sys/utsname.h>
#include <fcntl.h>
#include <poll.h>
#include <sched.h>
#include <unistd.h>
#include <atomic>
#include <charconv>
#include <chrono>
#include <condition_variable>
#include <csignal>
#include <cstring>
#include <deque>
#include <filesystem>
#include <fstream>
#include <functional>
#include <iostream>
#include <mutex>
#include <numeric>
#include <random>
#include <sstream>
#include <thread>
namespace {
using namespace ling3;
using namespace ling3::chat;
using Clock = std::chrono::steady_clock;
volatile std::sig_atomic_t quitting = 0;
void OnSignal(int) { quitting = 1; }
double Ms(Clock::time_point a, Clock::time_point b = Clock::now()) {
return std::chrono::duration<double, std::milli>(b - a).count();
}
std::string Read(const std::filesystem::path & p) {
std::ifstream f(p, std::ios::binary);
return {std::istreambuf_iterator<char>(f), {}};
}
double MemoryMiB(std::string_view field, const char * file = "/proc/meminfo") {
std::istringstream s(Read(file));
std::string line;
while (std::getline(s, line)) if (line.starts_with(field)) {
std::istringstream value(line.substr(field.size()));
double kb = 0; value >> kb; return kb / 1024;
}
return 0;
}
void Configure() {
// A deployed engine always uses the measured FP32-state + NPU W4A8 path.
// Ignore ambient experiment settings; no external graph or tokenizer files.
for (auto key : {"LING3_PREFILL_W4A4", "LING3_EXPERT_ALL_CORES", "LING3_GDN_PREFILL_DIR",
"LING3_DISABLE_BATCH_INPUT_REUSE", "LING3_DISABLE_PARALLEL_GATHER"}) unsetenv(key);
for (auto key : {"LING3_PREWARM_EXPERTS", "LING3_EXPERT_BALANCED", "LING3_EXPERT_ZERO_COPY",
"LING3_GDN_CPU_PREFILL", "LING3_GDN_CPU_FP32_STATE", "LING3_GDN_CPU_DECODE",
"LING3_GDN_FULL_FP32", "LING3_MLA_SIMD", "LING3_VECTOR_MATH"}) setenv(key, "1", 1);
cpu_set_t mask; CPU_ZERO(&mask);
for (int i = 4; i < 8; ++i) CPU_SET(i, &mask);
if (sched_setaffinity(0, sizeof(mask), &mask) != 0)
throw std::runtime_error("cannot use A76 CPU4-7: " + std::string(std::strerror(errno)));
rlimit r {};
if (getrlimit(RLIMIT_NOFILE, &r) != 0) throw std::runtime_error("getrlimit failed");
r.rlim_cur = std::min<rlim_t>(262144, r.rlim_max);
if (setrlimit(RLIMIT_NOFILE, &r) != 0 || r.rlim_cur < 65536)
std::cerr << "resource_warning: file descriptor limit is below the qualified budget; continuing, actual opens may fail\n";
}
struct DrmVersion {
int major, minor, patch;
std::size_t name_len; char * name;
std::size_t date_len; char * date;
std::size_t desc_len; char * desc;
};
Json CheckDevice() {
#if !LING3_WITH_ICU
throw std::runtime_error("non-production build: exact ICU tokenizer support is required");
#endif
utsname u {}; uname(&u);
const auto compatible = Read("/proc/device-tree/compatible");
if (std::string(u.machine) != "aarch64" || compatible.find("rockchip,rk3588") == std::string::npos)
throw std::runtime_error("unsupported device: requires RK3588/RK3588S, 64-bit Linux");
const auto memory = MemoryMiB("MemTotal:");
if (memory < 14000)
std::cerr << "memory_warning: physical RAM " << memory
<< " MiB is below the previously tested 16 GB class; continuing, available memory may be insufficient\n";
Json report {{"soc", "RK3588"}, {"kernel", u.release}, {"memory_mib", memory},
{"minimum_rknpu_driver", "0.9.8"}, {"cpu_affinity", "4-7"}};
bool found = false;
std::error_code ec;
for (const auto & entry : std::filesystem::directory_iterator("/dev/dri", ec)) {
if (!entry.path().filename().string().starts_with("renderD")) continue;
const int fd = open(entry.path().c_str(), O_RDWR | O_CLOEXEC);
if (fd < 0) continue;
char name[128] {};
DrmVersion v {}; v.name = name; v.name_len = sizeof(name) - 1;
const int rc = ioctl(fd, _IOWR('d', 0x00, DrmVersion), &v);
close(fd);
if (rc != 0 || std::string(name) != "rknpu") continue;
report["npu_node"] = entry.path().string();
report["npu_driver"] = std::to_string(v.major) + "." + std::to_string(v.minor) + "." + std::to_string(v.patch);
// Conservative qualification floor, not a claim about every older SDK.
if (!QualifiedRknpuDriver(v.major, v.minor, v.patch))
std::cerr << "driver_warning: RKNPU " << report["npu_driver"].get<std::string>()
<< " is below qualified 0.9.8; continuing to real hardware capability checks\n";
found = true; break;
}
if (!found) throw std::runtime_error("no accessible RKNPU DRM device; check rknpu kernel driver and render/video group permissions (Panthor GPU is not NPU)");
Configure();
if (!RknnBackendAvailable()) throw std::runtime_error("this binary was built without RKNN");
// Exercise the actual native INT4 matmul path on each core, not just version text.
std::vector<std::byte> w(256 * 64 / 2, std::byte{0x11});
std::vector<float> scales(64, 1), input(256, 1), out(64);
std::vector<float> batch_input(128 * 256), batch_output(128 * 64);
std::vector<std::int32_t> correction(64, 8 * 256);
for (int core = 0; core < 3; ++core) {
DynamicW4Linear linear({256, 64, 1, {core}, 0}, w, scales, correction);
linear.Run(input, out);
for (auto x : out) if (!std::isfinite(x) || std::abs(x - 256) > 2)
throw std::runtime_error("native INT4 numerical probe failed on NPU core " + std::to_string(core));
// Check the FD-view capability required by shared batch A/C before
// allocating the full model. Large-to-small transitions catch stale IO.
float value = 1;
for (std::size_t rows : {128, 1, 3, 16, 2}) {
std::fill(batch_input.begin(), batch_input.end(), value);
linear.RunBatch(std::span<const float>(batch_input).first(rows * 256), rows,
std::span<float>(batch_output).first(rows * 64));
for (std::size_t i = 0; i < rows * 64; ++i)
if (!std::isfinite(batch_output[i]) || std::abs(batch_output[i] - 256 * value) > 2)
throw std::runtime_error("shared batch A/C FD-view probe failed on NPU core " + std::to_string(core));
value = -value;
}
}
report["w4a8_native_int4_three_core_probe"] = "passed";
report["shared_batch_ac_fd_view_probe"] = "passed";
report["mla_backend"] = std::getenv("LING3_MLA_BACKEND") ? std::getenv("LING3_MLA_BACKEND") : "auto";
{
MlaNpu attention;
std::vector<float> q(16*16*192,0), values(16*16*128);
std::vector<std::uint16_t> k(272*16*192,0), v(272*16*128,0x3e80); // BF16 0.25.
bool ok=attention.Run(q,k,v,16,272,values);
if(ok)for(float x:values)if(!std::isfinite(x)||std::abs(x-0.25F)>0.0001F)
throw std::runtime_error("MLA FP16 dynamic input probe returned incorrect values");
if(ok){
std::fill(v.begin(),v.end(),0xbf00); // BF16 -0.5; detect stale B after rebinding.
ok=attention.Run(q,k,v,16,272,values);
if(ok)for(float x:values)if(!std::isfinite(x)||std::abs(x+0.5F)>0.0001F)
throw std::runtime_error("MLA FP16 dynamic update probe returned incorrect values");
}
report["mla_fp16_dynamic_probe"] = ok ? "passed" : "CPU_attention_fallback_or_selected";
}
report["available_memory_mib"] = MemoryMiB("MemAvailable:");
std::cout << "device_check=" << report.dump() << std::endl;
return report;
}
struct Result { std::string text, reason = "length"; Json usage, metrics; };
class Engine {
ModelPackage package_;
std::size_t context_;
Tokenizer tokenizer_;
Decoder decoder_;
std::vector<float> logits_;
std::mutex mutex_;
std::ofstream log_;
FlowGate flow_;
std::atomic<std::uint64_t> evaluated_steps_ {0};
struct PrefixCache {
std::vector<std::uint32_t> tokens;
std::string user;
std::size_t aligned_position = 0;
std::shared_ptr<DecoderCheckpoint> aligned, exact;
std::shared_ptr<const DecoderState> aligned_state, exact_state;
std::vector<std::uint32_t> continuation_tokens;
std::shared_ptr<DecoderCheckpoint> continuation;
std::shared_ptr<const DecoderState> continuation_state;
bool generated_lineage = false;
std::vector<float> logits;
std::size_t bytes() const {
std::size_t n = tokens.size() * sizeof(std::uint32_t) + logits.size() * sizeof(float);
if (aligned) n += Decoder::CheckpointBytes(*aligned);
if (exact && exact != aligned) n += Decoder::CheckpointBytes(*exact);
if (aligned_state) n += aligned_state->bytes();
if (exact_state && exact_state != aligned_state) n += exact_state->bytes();
n += continuation_tokens.size() * sizeof(std::uint32_t);
if (continuation) n += Decoder::CheckpointBytes(*continuation);
if (continuation_state) n += continuation_state->bytes();
return n;
}
void ClearContinuation() {
continuation_tokens.clear(); continuation.reset(); continuation_state.reset();
}
} prefix_cache_;
std::string resident_key_; // Empty for the anonymous resident slot.
StateCache<PrefixCache> sessions_ {256ULL * 1024 * 1024};
static std::string SessionKey(const std::string & user, const std::string & session) {
return Json::array({user, session}).dump();
}
void ClearCacheUnlocked() { prefix_cache_ = {}; resident_key_.clear(); }
static Tokenizer MakeTokenizer(const ModelPackage & p) {
const auto & t = p.tensor("tokenizer");
return Tokenizer({t.data, static_cast<std::size_t>(t.entry->data_bytes)});
}
public:
double initialization_ms = 0, warmup_ms = 0;
Json WeightFormat() const {
if(package_.header().flags & ling3::kPackageMixedW4W8){
const auto& metadata=package_.tensor("precision.recipe");
auto recipe=Json::parse(reinterpret_cast<const char*>(metadata.data),
reinterpret_cast<const char*>(metadata.data)+metadata.entry->data_bytes);
const char* shared=std::getenv("LING3_BRIDGE_SHARED_STAGE");
return {{"source","self_contained_selective_w4_w8"},{"execution","mixed_W4A8_W8A8"},
{"w8_families",recipe.at("w8_families")},
{"calibrated_w4_families",recipe.at("calibrated_w4_families")},
{"linear_counts",recipe.at("linear_counts")},
{"shared_stage",!shared || std::string(shared)=="1"},{"experimental",true}};
}
const char* families=std::getenv("LING3_W8_FAMILIES");
const char* calibrated=std::getenv("LING3_CALIBRATED_W4_FAMILIES");
const char* shared=std::getenv("LING3_BRIDGE_SHARED_STAGE");
if((families && *families)||(calibrated && *calibrated)) return {{"source","original_W4_with_selective_overrides"},
{"execution","mixed_W4A8_W8A8"},{"w8_families",families?families:""},
{"calibrated_w4_families",calibrated?calibrated:""},
{"shared_stage",shared && std::string(shared)=="1"},{"experimental",true}};
if (!(package_.header().flags & 0x100)) return {{"source","custom_per_channel_w4"},{"execution","W4A8"}};
const std::string mode=std::getenv("LING3_OFFICIAL_EXECUTION")?std::getenv("LING3_OFFICIAL_EXECUTION"):"w8";
return {{"source","official_int4_group32_with_high_precision_layers"},
{"execution",mode=="w8"?"W8A8_bridge_requantized":"FP16_bridge"},
{"experimental",true}};
}
explicit Engine(const std::string & model, const std::string & log, std::size_t context)
: package_(model), context_(context), tokenizer_(MakeTokenizer(package_)), decoder_(package_, context),
logits_(package_.header().vocab_size), log_(log, std::ios::app) {
if (!log_) throw std::runtime_error("cannot open metrics log: " + log);
for (std::size_t rows = 1; rows <= 128; rows *= 2) decoder_.PrepareBatch(rows);
}
bool active() const { return flow_.Active(); }
bool Cancel(const std::string & id = {}) { return flow_.Cancel(id); }
Json FlowStatus() const {
auto status = flow_.Status();
status["evaluated_steps"] = evaluated_steps_.load();
return status;
}
void Ack(const std::string & id, std::uint64_t sequence) { flow_.Ack(id, sequence); }
void Pause(const std::string & id) { flow_.Pause(id); }
void Resume(const std::string & id) { flow_.Resume(id); }
void ClearCache(const std::string & session = {}, const std::string & user = {}) {
std::unique_lock lock(mutex_, std::try_to_lock);
if (!lock.owns_lock()) throw Error(429, "engine_busy", "cannot clear cache during inference");
if (session.empty()) { sessions_.Clear(); ClearCacheUnlocked(); decoder_.Reset(); }
else {
const auto key = SessionKey(user, session);
sessions_.Erase(key);
if (resident_key_ == key) { ClearCacheUnlocked(); decoder_.Reset(); }
}
}
void SetCacheBudget(std::size_t bytes) { sessions_ = StateCache<PrefixCache>(bytes); }
Json CacheStatus() {
std::unique_lock lock(mutex_, std::try_to_lock);
if (!lock.owns_lock()) throw Error(429, "engine_busy", "cache status requires an idle engine");
return {{"sessions", sessions_.size()}, {"snapshot_bytes", sessions_.bytes()},
{"budget_bytes", sessions_.limit()}, {"evictions", sessions_.evictions()},
{"default_prefix_bytes", resident_key_.empty() ? prefix_cache_.bytes() : 0},
{"resident_cache_bytes", prefix_cache_.bytes()}, {"resident_named", !resident_key_.empty()},
{"persistent", false}};
}
void ForkCache(const std::string & user, const std::string & source, const std::string & target) {
std::unique_lock lock(mutex_, std::try_to_lock);
if (!lock.owns_lock()) throw Error(429, "engine_busy", "cannot fork during inference");
const auto key = SessionKey(user, source);
const auto * entry = sessions_.Get(key);
if (!entry) throw Error(404, "session_not_found", "source cache is not resident");
const auto copy = *entry;
if (!sessions_.Put(SessionKey(user, target), copy, copy.bytes()))
throw Error(409, "cache_budget_exceeded", "snapshot does not fit cache budget");
if (resident_key_ == SessionKey(user, target)) { ClearCacheUnlocked(); decoder_.Reset(); }
}
std::size_t context() const { return context_; }
Json WeightMemory() const {
return {{"source_weight_discard_advised_mib", package_.discarded_weight_bytes() / 1048576.0},
{"source_weight_pagecache_resident_mib", package_.resident_linear_weight_bytes() / 1048576.0},
{"source_weight_cache_advice_failures", package_.cache_advice_failures()}};
}
void Warmup() {
const auto start = Clock::now();
auto r = Parse({{"model", kModel}, {"messages", Json::array({{{"role", "user"}, {"content", "你好"}}})}, {"max_tokens", 2}});
// Warm exactly 128 input positions, independent of tokenizer text length.
const auto ids = tokenizer_.Encode(r.prompt);
std::vector<std::uint32_t> tokens(128);
for (std::size_t i = 0; i < tokens.size(); ++i) tokens[i] = ids[i % ids.size()];
decoder_.EvalBatch(tokens, logits_);
decoder_.Eval(std::max_element(logits_.begin(), logits_.end()) - logits_.begin(), logits_);
decoder_.Reset();
warmup_ms = Ms(start);
}
Result Run(const Request & r, const std::string & id,
const std::function<bool()> & begin,
const std::function<bool(std::string_view)> & emit,
const std::function<bool()> & connected) {
const auto start = Clock::now();
std::unique_lock lock(mutex_, std::try_to_lock);
if (!lock.owns_lock()) throw Error(429, "engine_busy", "one inference at a time; retry after the active request finishes");
rusage faults_before {}; getrusage(RUSAGE_SELF, &faults_before);
const auto attention_before = decoder_.AttentionStats();
const auto tokens = tokenizer_.Encode(r.prompt);
const auto output_budget = OutputBudget(tokens.size(), context(), r.max_tokens);
const bool named_session = !r.session_id.empty();
const auto session_key = SessionKey(r.cache_user, r.session_id);
// The active named session, like the anonymous slot, retains lightweight
// checkpoints even when a full historical snapshot cannot fit the budget.
if (named_session) {
if (resident_key_ != session_key) {
const auto * saved = r.cache_prompt ? sessions_.Get(session_key) : nullptr;
prefix_cache_ = saved ? *saved : PrefixCache {};
}
if (!r.cache_prompt) sessions_.Erase(session_key);
} else if (!resident_key_.empty()) ClearCacheUnlocked();
// A strict request must never inherit the numerical lineage of an
// incremental decode path, even through an exact prompt cache hit.
if (!r.reuse_generated_state && prefix_cache_.generated_lineage) ClearCacheUnlocked();
const std::size_t aligned_position = (tokens.size()/128)*128;
const auto state_cost = [&](std::size_t position) {
return 29638656ULL + decoder_.StateSignature().size() + position*61440ULL;
};
const std::size_t snapshot_cost = state_cost(tokens.size()) +
((aligned_position && aligned_position != tokens.size()) ? state_cost(aligned_position) : 0) +
tokens.size()*sizeof(std::uint32_t) + logits_.size()*sizeof(float);
const bool cache_named = named_session && r.cache_prompt && snapshot_cost <= sessions_.limit();
if (output_budget < r.max_tokens)
std::cerr << "request_info: output_budget=" << output_budget
<< " remaining_context=" << context() - tokens.size()
<< "; generation may continue until EOS, stop, cancellation or context capacity\n";
flow_.Begin(id, r.flow_ack);
evaluated_steps_ = 0;
struct Guard {
Engine & e; bool preserve = false;
~Guard() {
if (!preserve) { e.ClearCacheUnlocked(); e.decoder_.Reset(); }
e.flow_.End();
}
} guard {*this};
const auto cache_begin = Clock::now();
std::size_t cached_tokens = 0;
bool exact_hit = false;
std::size_t generated_cached_tokens = 0;
std::string cache_status = r.cache_prompt ? "miss" : "disabled";
if (r.cache_prompt && prefix_cache_.user == r.cache_user && (prefix_cache_.exact || prefix_cache_.exact_state) &&
prefix_cache_.tokens == tokens) {
cached_tokens = prefix_cache_.exact ? decoder_.RestoreCheckpoint(*prefix_cache_.exact)
: decoder_.RestoreState(*prefix_cache_.exact_state);
if (!prefix_cache_.exact) prefix_cache_.exact = decoder_.SaveCheckpoint();
logits_ = prefix_cache_.logits;
exact_hit = true;
cache_status = "exact_prompt_hit";
} else if (r.cache_prompt && r.reuse_generated_state && prefix_cache_.user == r.cache_user &&
(prefix_cache_.continuation || prefix_cache_.continuation_state) &&
tokens.size() > prefix_cache_.continuation_tokens.size() &&
std::equal(prefix_cache_.continuation_tokens.begin(), prefix_cache_.continuation_tokens.end(), tokens.begin())) {
cached_tokens = prefix_cache_.continuation ? decoder_.RestoreCheckpoint(*prefix_cache_.continuation)
: decoder_.RestoreState(*prefix_cache_.continuation_state);
generated_cached_tokens = cached_tokens - prefix_cache_.tokens.size();
prefix_cache_.generated_lineage = true;
prefix_cache_.exact.reset(); prefix_cache_.exact_state.reset(); prefix_cache_.logits.clear();
cache_status = "generated_prefix_hit";
} else if (r.cache_prompt && r.reuse_generated_state && prefix_cache_.user == r.cache_user &&
(prefix_cache_.exact || prefix_cache_.exact_state) && tokens.size() > prefix_cache_.tokens.size() &&
std::equal(prefix_cache_.tokens.begin(), prefix_cache_.tokens.end(), tokens.begin())) {
cached_tokens = prefix_cache_.exact ? decoder_.RestoreCheckpoint(*prefix_cache_.exact)
: decoder_.RestoreState(*prefix_cache_.exact_state);
prefix_cache_.generated_lineage = true;
prefix_cache_.exact.reset(); prefix_cache_.exact_state.reset(); prefix_cache_.logits.clear();
cache_status = "exact_prefix_hit";
} else if (r.cache_prompt && prefix_cache_.user == r.cache_user && (prefix_cache_.aligned || prefix_cache_.aligned_state) &&
prefix_cache_.aligned_position > 0 && tokens.size() > prefix_cache_.aligned_position &&
std::equal(prefix_cache_.tokens.begin(),
prefix_cache_.tokens.begin() + prefix_cache_.aligned_position, tokens.begin())) {
cached_tokens = prefix_cache_.aligned ? decoder_.RestoreCheckpoint(*prefix_cache_.aligned)
: decoder_.RestoreState(*prefix_cache_.aligned_state);
prefix_cache_.exact.reset();
prefix_cache_.exact_state.reset();
prefix_cache_.logits.clear();
cache_status = "prefix_hit";
} else {
ClearCacheUnlocked(); decoder_.Reset();
}
prefix_cache_.ClearContinuation();
resident_key_ = named_session ? session_key : "";
const double cache_restore_ms = Ms(cache_begin);
double cache_save_ms = 0;
const auto aligned_target = (tokens.size() / 128) * 128;
if (r.cache_prompt && !exact_hit) {
prefix_cache_.tokens = tokens;
prefix_cache_.user = r.cache_user;
}
Result result;
std::size_t generated = 0, eval_count = 0, prompt_evaluated = 0;
double decode_eval_ms = 0, prefill_eval_ms = 0, first_ms = 0;
auto first = Clock::now(), last = first;
auto alive = [&] { return !quitting && flow_.Alive() && connected(); };
auto aborted = [&] { result.reason = "canceled"; flow_.Cancel(id); };
bool began = begin();
const auto prefill_start = Clock::now();
auto progress_at = prefill_start;
if (began) for (std::size_t offset = cached_tokens; offset < tokens.size();) {
if (!alive() || !flow_.Wait([&] { return !quitting && connected(); })) { aborted(); break; }
// Generated prefixes can end between block boundaries. Preserve
// the aligned checkpoint even when the first resumed block is short.
auto block = std::span<const std::uint32_t>(tokens).subspan(offset,
std::min<std::size_t>(128 - offset % 128, tokens.size() - offset));
const bool final = offset + block.size() == tokens.size();
if (block.size() == 1) prefill_eval_ms += decoder_.Eval(block[0], logits_).total_ms;
else prefill_eval_ms += (final ? decoder_.EvalBatch(block, logits_) : decoder_.EvalBatchState(block)).total_ms;
++evaluated_steps_;
offset += block.size(); prompt_evaluated = offset - cached_tokens;
if (r.cache_prompt && offset == aligned_target) {
const auto saved_at = Clock::now();
prefix_cache_.aligned = decoder_.SaveCheckpoint();
prefix_cache_.aligned_state = cache_named ? decoder_.SaveState() : nullptr;
prefix_cache_.aligned_position = offset;
cache_save_ms += Ms(saved_at);
}
const auto now = Clock::now();
if (Ms(progress_at, now) >= 10000) {
const Json progress {{"id", id}, {"timestamp", std::time(nullptr)},
{"prompt_tokens", tokens.size()}, {"prompt_evaluated_tokens", prompt_evaluated},
{"cached_tokens", cached_tokens}, {"processed_position", offset},
{"elapsed_ms", Ms(start, now)}, {"prefill_eval_ms", prefill_eval_ms},
{"rss_mib", MemoryMiB("VmRSS:", "/proc/self/status")},
{"peak_rss_mib", MemoryMiB("VmHWM:", "/proc/self/status")},
{"swap_mib", MemoryMiB("VmSwap:", "/proc/self/status")},
{"available_memory_mib", MemoryMiB("MemAvailable:")}};
std::cerr << "prefill_progress=" << progress.dump() << std::endl;
progress_at = now;
}
} else aborted();
if (result.reason != "canceled" && r.cache_prompt && !exact_hit) {
const auto saved_at = Clock::now();
prefix_cache_.exact = aligned_target == tokens.size() ? prefix_cache_.aligned : decoder_.SaveCheckpoint();
prefix_cache_.exact_state = cache_named ? (aligned_target == tokens.size()
? prefix_cache_.aligned_state : decoder_.SaveState()) : nullptr;
prefix_cache_.logits = logits_;
cache_save_ms += Ms(saved_at);
}
const auto prefill_end = Clock::now();
const bool retain_generated = r.cache_prompt && r.reuse_generated_state;
std::vector<std::uint32_t> evaluated_tokens = retain_generated ? tokens : std::vector<std::uint32_t>{};
TextFilter filter(r.stops);
std::mt19937 random(r.seed);
std::vector<bool> seen(logits_.size(), false);
for (auto token : tokens) if (token < seen.size()) seen[token] = true;
double flow_wait_ms = 0;
std::size_t flow_wait_count = 0;
auto output = [&](std::string text) {
result.text += text;
if (!text.empty()) {
flow_.Prepare();
if (!emit(text)) { aborted(); return false; }
if (r.flow_ack) {
const auto wait_start = Clock::now();
const bool resumed = flow_.Wait([&] { return !quitting && connected(); });
flow_wait_ms += Ms(wait_start); ++flow_wait_count;
if (!resumed) { aborted(); return false; }
}
}
return true;
};
if (result.reason != "canceled") for (std::size_t i = 0; i < output_budget; ++i) {
if (!alive() || !flow_.Wait([&] { return !quitting && connected(); })) { aborted(); break; }
const auto token = SampleToken(logits_, r, seen, random);
seen[token] = true;
last = Clock::now();
if (!generated) { first = last; first_ms = Ms(start, first); }
++generated; // includes EOS and stop tokens, like completion usage.
if (token == package_.header().eos_token) { result.reason = "stop"; break; }
if (!output(filter.Push(tokenizer_.Piece(token)))) break;
if (filter.stopped) { result.reason = "stop"; break; }
if (i + 1 < output_budget) {
if (!flow_.Wait([&] { return !quitting && connected(); })) { aborted(); break; }
decode_eval_ms += decoder_.Eval(token, logits_).total_ms; ++eval_count;
if (retain_generated) evaluated_tokens.push_back(token);
++evaluated_steps_;
}
}
if (result.reason != "canceled") output(filter.Push({}, true));
if (result.reason != "canceled" && !alive()) aborted();
bool continuation_stored = false, continuation_skipped_budget = false;
double continuation_save_ms = 0;
if (result.reason != "canceled" && retain_generated && eval_count) {
const auto cost = state_cost(decoder_.position()) + evaluated_tokens.size()*sizeof(std::uint32_t);
{
const auto save_begin = Clock::now();
// Do not evaluate the final sampled token just for caching.
// The stored prefix contains only tokens actually consumed.
prefix_cache_.continuation = decoder_.SaveCheckpoint();
// Historical backup cost excludes borrowed resident checkpoints.
auto backup = prefix_cache_;
backup.aligned.reset(); backup.exact.reset(); backup.continuation.reset();
if (named_session) {
if (!cache_named || cost > sessions_.limit() || backup.bytes() > sessions_.limit() - cost)
continuation_skipped_budget = true;
else prefix_cache_.continuation_state = decoder_.SaveState();
}
prefix_cache_.continuation_tokens = std::move(evaluated_tokens);
continuation_stored = true;
continuation_save_ms = Ms(save_begin);
}
}
if (result.reason != "canceled" && !alive()) aborted();
if (result.reason != "canceled" && !flow_.TryCommit()) aborted();
if (result.reason == "canceled") { prefix_cache_.ClearContinuation(); continuation_stored = false; }
result.usage = {{"prompt_tokens", tokens.size()}, {"completion_tokens", generated},
{"total_tokens", tokens.size() + generated}};
if (r.cache_prompt) result.usage["prompt_tokens_details"] = {{"cached_tokens", cached_tokens}};
const auto decode_ms = generated > 1 ? Ms(first, last) : 0;
result.metrics = {{"id", id}, {"model", kModel}, {"timestamp", std::time(nullptr)},
{"finish_reason", result.reason}, {"usage", result.usage},
{"ttft_ms", generated ? Json(first_ms) : Json(nullptr)}, {"prefill_ms", Ms(prefill_start, prefill_end)},
{"prefill_eval_ms", prefill_eval_ms}, {"prompt_evaluated_tokens", prompt_evaluated},
{"prefill_tokens_per_second", prefill_eval_ms > 0 ? 1000 * prompt_evaluated / prefill_eval_ms : 0},
{"decode_ms", decode_ms}, {"decode_eval_ms", decode_eval_ms}, {"decode_eval_count", eval_count},
{"decode_tokens_per_second", decode_ms > 0 ? 1000 * (generated - 1) / decode_ms : 0},
{"total_ms", Ms(start)}, {"rss_mib", MemoryMiB("VmRSS:", "/proc/self/status")},
{"peak_rss_mib", MemoryMiB("VmHWM:", "/proc/self/status")},
{"queue_ms", 0}, {"cached_tokens", cached_tokens}, {"warmed", true},
{"cache_status", cache_status}, {"cache_restore_ms", cache_restore_ms},
{"cache_save_ms", cache_save_ms}, {"prefix_cache_mib", prefix_cache_.bytes() / 1048576.0}};
result.metrics["context_length"] = context_;
result.metrics["reuse_generated_state"] = r.reuse_generated_state;
result.metrics["generated_state_lineage"] = prefix_cache_.generated_lineage;
result.metrics["generated_cached_tokens"] = generated_cached_tokens;
result.metrics["continuation_stored"] = continuation_stored;
result.metrics["continuation_skipped_budget"] = continuation_skipped_budget;
result.metrics["continuation_save_ms"] = continuation_save_ms;
result.metrics["continuation_tokens"] = continuation_stored ? prefix_cache_.continuation_tokens.size() : 0;
result.metrics["flow_wait_ms"] = flow_wait_ms;
result.metrics["flow_wait_count"] = flow_wait_count;
result.metrics["flow_control"] = r.flow_ack ? "ack" : "none";
result.metrics["sampling"] = {{"temperature", r.temperature}, {"top_p", r.top_p},
{"top_k", r.top_k}, {"repeat_penalty", r.repeat_penalty}, {"seed", r.seed},
{"enable_thinking", r.enable_thinking}};
if (named_session) {
bool stored = false;
if (result.reason != "canceled" && r.cache_prompt) {
if (cache_named && prefix_cache_.exact_state) {
auto backup = prefix_cache_;
backup.aligned.reset(); backup.exact.reset(); backup.continuation.reset();
if (!backup.continuation_state) backup.continuation_tokens.clear();
stored = sessions_.Put(session_key, backup, backup.bytes());
}
else sessions_.Erase(session_key);
}
result.metrics["session_id"] = r.session_id;
result.metrics["session_cache_stored"] = stored;
result.metrics["session_cache_skipped_budget"] = r.cache_prompt && !cache_named;
result.metrics["session_cache_bytes"] = sessions_.bytes();
result.metrics["session_cache_evictions"] = sessions_.evictions();
}
result.metrics["decode_compute_tokens_per_second"] = decode_eval_ms > 0 ? 1000 * eval_count / decode_eval_ms : 0;
result.metrics["requested_max_tokens"] = r.max_tokens == std::numeric_limits<std::size_t>::max()
? Json(nullptr) : Json(r.max_tokens);
result.metrics["effective_max_tokens"] = output_budget;
const bool context_reached = result.reason == "length" && generated == context() - tokens.size();
result.metrics["context_capacity_reached"] = context_reached;
if (context_reached)
std::cerr << "request_info: context_capacity_reached id=" << id
<< " capacity=" << context() << "; finished normally with reason=length\n";
const auto attention_after = decoder_.AttentionStats();
result.metrics["mla_npu_calls"] = attention_after.npu_calls - attention_before.npu_calls;
result.metrics["mla_cpu_calls"] = attention_after.cpu_calls - attention_before.cpu_calls;
result.metrics["mla_fallbacks"] = attention_after.fallbacks - attention_before.fallbacks;
result.metrics["mla_npu_ms"] = attention_after.npu_ms - attention_before.npu_ms;
rusage faults_after {}; getrusage(RUSAGE_SELF, &faults_after);
result.metrics["minor_page_faults"] = faults_after.ru_minflt - faults_before.ru_minflt;
result.metrics["major_page_faults"] = faults_after.ru_majflt - faults_before.ru_majflt;
log_ << result.metrics.dump() << '\n'; log_.flush();
std::cerr << "reply_metrics=" << result.metrics.dump() << std::endl;
guard.preserve = result.reason != "canceled" && r.cache_prompt;
return result;
}
};
bool Send(int fd, std::string_view data) {
while (!data.empty()) {
const auto n = send(fd, data.data(), data.size(), MSG_NOSIGNAL);
if (n < 0 && errno == EINTR) continue;
if (n <= 0) return false;
data.remove_prefix(n);
}
return true;
}
bool Connected(int fd) {
pollfd p {fd, POLLIN | POLLRDHUP, 0};
if (poll(&p, 1, 0) < 0) return false;
if (p.revents & (POLLERR | POLLHUP | POLLRDHUP | POLLNVAL)) return false;
return true;
}
void Response(int fd, int code, const Json & body) {
const auto s = body.dump();
Send(fd, "HTTP/1.1 " + std::to_string(code) + (code == 200 ? " OK" : " Error") +
"\r\nContent-Type: application/json; charset=utf-8\r\nConnection: close\r\nContent-Length: " +
std::to_string(s.size()) + "\r\n\r\n" + s);
}
struct Http { std::string method, path, body; };
Http ReadRequest(int fd) {
std::string data; char buf[4096]; std::size_t end;
const auto started = Clock::now();
auto receive = [&] {
if (Ms(started) > 10000 || quitting) throw Error(408, "request_timeout", "request read timed out");
const auto n = recv(fd, buf, sizeof(buf), 0);
if (n <= 0) throw Error(400, "invalid_http", "incomplete HTTP request");
data.append(buf, n);
};
while ((end = data.find("\r\n\r\n")) == std::string::npos) {
receive(); if (data.size() > 65536) throw Error(413, "too_large", "headers too large");
}
Http h;
std::istringstream headers(data.substr(0, end));
std::string line, version; std::getline(headers, line);
std::istringstream request_line(line); request_line >> h.method >> h.path >> version;
if (version != "HTTP/1.1" && version != "HTTP/1.0") throw Error(400, "invalid_http", "invalid HTTP request line");
std::size_t length = 0; bool have_length = false;
while (std::getline(headers, line)) {
if (!line.empty() && line.back() == '\r') line.pop_back();
auto colon = line.find(':');
if (colon == std::string::npos) throw Error(400, "invalid_http", "invalid header");
std::string name = line.substr(0, colon), value = line.substr(colon + 1);
std::transform(name.begin(), name.end(), name.begin(), [](unsigned char c) { return std::tolower(c); });
value.erase(0, value.find_first_not_of(" \t"));
if (name == "transfer-encoding") throw Error(400, "invalid_http", "request Transfer-Encoding unsupported; send Content-Length");
if (name == "content-length") {
auto [p, ec] = std::from_chars(value.data(), value.data() + value.size(), length);
if (have_length || ec != std::errc{} || p != value.data() + value.size())
throw Error(400, "invalid_http", "invalid/duplicate Content-Length");
have_length = true;
}
}
if (length > data.max_size()-end-4) throw Error(400,"invalid_http","Content-Length exceeds addressable string size");
if (length > 1024 * 1024)
std::cerr << "request_warning: body_bytes=" << length << " exceeds previous 1 MiB budget; continuing\n";
end += 4;
while (data.size() - end < length) receive();
h.body = data.substr(end, length);
return h;
}
std::string Id() {
static std::atomic_uint64_t count {0};
return "chatcmpl-" + std::to_string(std::time(nullptr)) + "-" + std::to_string(++count);
}
void Handle(int fd, Engine & engine, const Json & device) {
struct Guard { int fd; ~Guard() { close(fd); } } guard {fd};
timeval read_timeout {2, 0}, write_timeout {5, 0};
setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &read_timeout, sizeof(read_timeout));
setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &write_timeout, sizeof(write_timeout));
bool streaming_started = false;
auto error = [&](const Error & e) {
if (streaming_started) Send(fd, "data: " + ErrorBody(e).dump() + "\n\ndata: [DONE]\n\n");
else Response(fd, e.status, ErrorBody(e));
};
try {
const auto http = ReadRequest(fd);
if (http.method == "GET" && http.path == "/health") {
Response(fd, 200, {{"status", "ok"}, {"active", engine.active()}, {"device", device},
{"context_length", engine.context()}, {"initialization_ms", engine.initialization_ms},
{"warmup_ms", engine.warmup_ms}, {"weight_memory", engine.WeightMemory()},
{"weight_format",engine.WeightFormat()}, {"mindnano_integration_version", 5},
{"capabilities", {{"flow_control_ack", true}, {"pause_resume", true}, {"top_k", true},
{"repeat_penalty", true}, {"enable_thinking", true}, {"session_cache", true},
{"generated_state_reuse", true}, {"qa_cache_default", true}, {"resident_session_cache", true}}},
{"generation", engine.FlowStatus()}}); return;
}
auto cache_field = [](const Json & body, const char * field, bool required) {
if (!body.is_object() || (required && !body.contains(field)) ||
(body.contains(field) && !body[field].is_string()))
throw Error(400, "invalid_parameter", std::string(field)+" must be a string");
auto value = body.value(field, std::string{});
if (required && (value.empty() || value.size() > 256))
throw Error(400, "invalid_parameter", std::string(field)+" must contain 1..256 bytes");
return value;
};
auto cache_fields = [](const Json & body, std::initializer_list<std::string_view> allowed) {
if (!body.is_object()) throw Error(400, "invalid_parameter", "cache request must be an object");
for (auto it = body.begin(); it != body.end(); ++it)
if (std::find(allowed.begin(), allowed.end(), it.key()) == allowed.end())
throw Error(400, "invalid_parameter", "unsupported cache field: " + it.key());
};
if (http.method == "GET" && http.path == "/v1/cache/status") {
Response(fd, 200, engine.CacheStatus()); return;
}
if (http.method == "POST" && http.path == "/v1/cache/clear") {
const auto body = Json::parse(http.body);
cache_fields(body, {"user", "session_id"});
const auto user = cache_field(body, "user", false);
const auto session = cache_field(body, "session_id", body.contains("session_id"));
if (!user.empty() && session.empty())
throw Error(400, "invalid_parameter", "user-scoped clear requires session_id");
engine.ClearCache(session, user); Response(fd, 200, {{"cleared", true}}); return;
}
if (http.method == "POST" && http.path == "/v1/cache/fork") {
const auto body = Json::parse(http.body);
cache_fields(body, {"user", "source_session_id", "target_session_id"});
engine.ForkCache(cache_field(body, "user", false), cache_field(body, "source_session_id", true),
cache_field(body, "target_session_id", true));
Response(fd, 200, {{"forked", true}}); return;
}
if (http.method == "GET" && http.path == "/v1/models") {
Response(fd, 200, {{"object", "list"}, {"data", Json::array({{{"id", kModel}, {"object", "model"},
{"created", 0}, {"owned_by", "mindnano"}, {"context_length", engine.context()}}})}}); return;
}
if (http.method == "POST" && http.path == "/v1/cancel") {
const auto body = Json::parse(http.body);
const auto canceled = engine.Cancel(body.value("request_id", std::string{}));
Response(fd, 200, {{"cancel_requested", canceled}}); return;
}
if (http.method == "POST" && http.path == "/v1/flow/ack") {
const auto body = Json::parse(http.body);
if (!body.contains("sequence") || !body["sequence"].is_number_unsigned() ||
!body.contains("request_id") || !body["request_id"].is_string())
throw Error(400, "invalid_flow_control", "request_id and positive sequence are required");
engine.Ack(body["request_id"].get<std::string>(), body["sequence"].get<std::uint64_t>());
Response(fd, 200, {{"acknowledged", body["sequence"]}}); return;
}
if (http.method == "GET" && http.path == "/v1/generation/status") {
Response(fd, 200, engine.FlowStatus()); return;
}
if (http.method == "POST" && (http.path == "/v1/generation/pause" || http.path == "/v1/generation/resume")) {
const auto body = Json::parse(http.body);
const auto id = body.at("request_id").get<std::string>();
if (http.path == "/v1/generation/pause") engine.Pause(id);
else engine.Resume(id);
Response(fd, 200, engine.FlowStatus()); return;
}
if (http.method != "POST" || http.path != "/v1/chat/completions")
throw Error(404, "not_found", "endpoint not found");
const auto request = Parse(Json::parse(http.body));
const auto id = Id(); const auto created = std::time(nullptr);
auto chunk = [&](Json delta, Json finish = nullptr) {
Json out {{"id", id}, {"object", "chat.completion.chunk"}, {"created", created}, {"model", kModel},
{"choices", Json::array({{{"index", 0}, {"delta", delta}, {"finish_reason", finish}, {"logprobs", nullptr}}})}};
if (request.include_usage) out["usage"] = nullptr;
if (request.flow_ack && delta.contains("content") && delta["content"] != "")
out["mindnano_flow"] = {{"request_id", id}, {"sequence", engine.FlowStatus()["sequence"]}};
return Send(fd, "data: " + out.dump() + "\n\n");
};
const auto result = engine.Run(request, id, [&] {
if (!request.stream) return Connected(fd);
streaming_started = true;
return Send(fd, "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream; charset=utf-8\r\n"
"Cache-Control: no-cache\r\nX-Accel-Buffering: no\r\nConnection: close\r\n\r\n") &&
chunk({{"role", "assistant"}, {"content", ""}});
}, [&](std::string_view text) { return !request.stream || chunk({{"content", text}}); },
[&] { return Connected(fd); });
if (result.reason == "canceled") {
error(Error(409, "request_canceled", "generation canceled; partial text is not a completed reply")); return;
}
if (request.stream) {
chunk(Json::object(), result.reason);
Json final {{"id", id}, {"object", "chat.completion.chunk"}, {"created", created}, {"model", kModel},
{"choices", Json::array()}, {"mindnano_metrics", result.metrics}};
if (request.include_usage) final["usage"] = result.usage;
Send(fd, "data: " + final.dump() + "\n\ndata: [DONE]\n\n");
} else Response(fd, 200, {{"id", id}, {"object", "chat.completion"}, {"created", created}, {"model", kModel},
{"choices", Json::array({{{"index", 0}, {"message", {{"role", "assistant"}, {"content", result.text}}},
{"finish_reason", result.reason}, {"logprobs", nullptr}}})},
{"usage", result.usage}, {"mindnano_metrics", result.metrics}});
} catch (const Error & e) { error(e); }
catch (const Json::exception & e) { error(Error(400, "invalid_json", e.what())); }
catch (const std::exception & e) { error(Error(500, "inference_error", e.what())); }
}
void Console(Engine & engine) {
Json messages = Json::array();
bool reuse_generated = true;
std::cout << "\n输入中文即可对话。/reset 清空历史,/reuse on|off 切换生成状态复用,/quit 退出。\n> " << std::flush;
while (!quitting) {
pollfd p {STDIN_FILENO, POLLIN, 0};
if (poll(&p, 1, 100) <= 0) continue;
std::string line;
if (!std::getline(std::cin, line)) return;
if (line == "/quit") { quitting = 1; engine.Cancel(); return; }
if (line == "/reset") {
try { engine.ClearCache(); messages = Json::array(); }
catch (const std::exception & e) { std::cerr << "\nerror: " << e.what() << '\n'; }
}
else if (line == "/cancel") engine.Cancel();
else if (line == "/reuse on" || line == "/reuse off") {
reuse_generated = line == "/reuse on";
std::cout << "生成状态复用:" << (reuse_generated ? "开启(增量数值路径)" : "关闭(严格前缀路径)") << '\n';
}
else if (!line.empty()) {
auto next = messages;
next.push_back({{"role", "user"}, {"content", line}});
try {
auto r = Parse({{"model", kModel}, {"messages", next}, {"reuse_generated_state", reuse_generated}});
const auto result = engine.Run(r, Id(), [] { return true; }, [](auto text) {
std::cout << text << std::flush; return true;
}, [] { return true; });
if (result.reason != "canceled") {
next.push_back({{"role", "assistant"}, {"content", result.text}}); messages = next;
}
std::cout << "\n[TTFT " << result.metrics["ttft_ms"] << " ms; decode "
<< result.metrics["decode_tokens_per_second"] << " tokens/s; " << result.reason << "]\n";
} catch (const std::exception & e) { std::cerr << "\nerror: " << e.what() << '\n'; }
}
std::cout << "> " << std::flush;
}
}
} // namespace
int main(int argc, char ** argv) {
try {
std::string model = "ling3-tiny-w4.l3r", host = "127.0.0.1", log = "metrics.jsonl";
int port = 9091; bool check = false, console = isatty(STDIN_FILENO);
std::string mla_backend = "auto";
std::size_t context = 4096;
std::size_t session_cache_mib = 256;
for (int i = 1; i < argc; ++i) {
const std::string arg = argv[i];
if (arg == "--help") {
std::cout << "mindnano-infer [--model FILE] [--host IPv4] [--port 9091] [--log metrics.jsonl]\n"
" [--context 4K|8K|16K|32K|64K|128K|256K] [--list-contexts]\n"
" [--no-console] [--check] [--mla-backend auto|cpu|npu]\n"
" [--session-cache-mib 256] (0 disables historical snapshots, not resident KV)\n"
"RK3588 / tested RKNPU 0.9.8; qualification and memory estimates are advisory. Text-only OpenAI chat completions.\n"; return 0;
}
if (arg == "--check") { check = true; continue; }
if (arg == "--allow-experimental-context") { continue; } // Compatibility: no longer needed.
if (arg == "--list-contexts") {
for (auto n : kContexts) std::cout << n / 1024 << "K: estimated_model_rss_gib="
<< EstimateContextMiB(n) / 1024 << " suggested_available_memory_gib="
<< (EstimateContextMiB(n) + 1536) / 1024
<< (n > 131072 ? " experimental_extrapolation_beyond_native_128K" : "") << '\n';
return 0;
}
if (arg == "--no-console") { console = false; continue; }
if (i + 1 == argc) throw std::runtime_error("missing argument after " + arg);
if (arg == "--model") model = argv[++i];
else if (arg == "--context") context = ParseContext(argv[++i]);
else if (arg == "--mla-backend") mla_backend = argv[++i];
else if (arg == "--host") host = argv[++i];
else if (arg == "--log") log = argv[++i];
else if (arg == "--session-cache-mib") {
const std::string value = argv[++i];
const auto [p, ec] = std::from_chars(value.data(), value.data()+value.size(), session_cache_mib);
if (ec != std::errc{} || p != value.data()+value.size() ||
session_cache_mib > std::numeric_limits<std::size_t>::max()/(1024*1024))
throw std::invalid_argument("session cache budget must be a non-negative MiB integer");
}
else if (arg == "--port") {
std::string value = argv[++i]; auto [p, ec] = std::from_chars(value.data(), value.data() + value.size(), port);
if (ec != std::errc{} || p != value.data() + value.size() || port < 1 || port > 65535)
throw std::runtime_error("port must be 1..65535");
} else throw std::runtime_error("unknown option " + arg);
}
std::signal(SIGPIPE, SIG_IGN); std::signal(SIGINT, OnSignal); std::signal(SIGTERM, OnSignal);
if (mla_backend != "auto" && mla_backend != "cpu" && mla_backend != "npu")
throw std::invalid_argument("--mla-backend must be auto, cpu or npu");
setenv("LING3_MLA_BACKEND", mla_backend.c_str(), 1);
std::cerr << "mla_backend=" << mla_backend << " tile=512 shared_by_six_layers=1\n";
const auto device = CheckDevice();
if (check) return 0;
if (context > 131072)
std::cerr << "context_warning: 256K exceeds native 128K; continuing with unvalidated position extrapolation\n";
const auto required = EstimateContextMiB(context) + 1536;
const auto available = MemoryMiB("MemAvailable:");
const Json budget {{"context_length", context}, {"policy", "warn_only"},
{"estimate_basis", "legacy_before_memory_optimizations"},
{"estimated_model_rss_mib", EstimateContextMiB(context)},
{"reserve_mib", 1536}, {"suggested_available_mib", required},
{"available_memory_mib", available},
{"below_estimate", available < required}};
std::cerr << "memory_budget=" << budget.dump() << std::endl;
if (available < required)
std::cerr << "memory_warning: available RAM may be insufficient; legacy estimate including reserve is "
<< required << " MiB, available " << available
<< " MiB; continuing without a memory admission limit\n";
if (context > 4096) std::cerr << "notice: contexts above 4K have not passed long-context quality/performance qualification\n";
// Bind before expensive initialization; conflicting ports fail immediately.
const int server = socket(AF_INET, SOCK_STREAM | SOCK_CLOEXEC, 0);
if (server < 0) throw std::runtime_error("socket failed");
struct CloseServer { int fd; ~CloseServer() { close(fd); } } close_server {server};
int reuse = 1; setsockopt(server, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));
sockaddr_in addr {}; addr.sin_family = AF_INET; addr.sin_port = htons(port);
if (inet_pton(AF_INET, host.c_str(), &addr.sin_addr) != 1) throw std::runtime_error("host must be IPv4");
if (bind(server, reinterpret_cast<sockaddr *>(&addr), sizeof(addr)) != 0)
throw std::runtime_error("bind failed: " + std::string(std::strerror(errno)));
const auto start = Clock::now();
std::cout << "initializing_model=" << model << std::endl;
Engine engine(model, log, context);
engine.SetCacheBudget(session_cache_mib*1024*1024);
engine.initialization_ms = Ms(start);
std::cout << "weight_format=" << engine.WeightFormat().dump() << std::endl;
engine.Warmup();
std::cout << "weight_memory=" << engine.WeightMemory().dump() << std::endl;
if (listen(server, 16) != 0) throw std::runtime_error("listen failed");
std::cout << "service_ready=http://" << host << ':' << port << "/v1\n"
<< "initialization_ms=" << engine.initialization_ms << " warmup_ms=" << engine.warmup_ms
<< " context_length=" << engine.context() << " metrics_log=" << log << std::endl;
std::mutex mutex; std::condition_variable cv; std::deque<int> queue;
bool stop_workers = false;
std::vector<std::thread> workers;
for (int i = 0; i < 4; ++i) workers.emplace_back([&] {
while (true) {
int fd;
{ std::unique_lock lock(mutex); cv.wait(lock, [&] { return stop_workers || !queue.empty(); });
if (queue.empty()) return; fd = queue.front(); queue.pop_front(); }
Handle(fd, engine, device);
}
});
std::thread terminal;
if (console) terminal = std::thread([&] { Console(engine); });
while (!quitting) {
pollfd p {server, POLLIN, 0};
if (poll(&p, 1, 100) <= 0) continue;
const int fd = accept4(server, nullptr, nullptr, SOCK_CLOEXEC);
if (fd < 0) continue;
std::lock_guard lock(mutex);
if (queue.size() >= 16) { close(fd); continue; }
queue.push_back(fd); cv.notify_one();
}
engine.Cancel();
{ std::lock_guard lock(mutex); stop_workers = true; for (int fd : queue) close(fd); queue.clear(); }
cv.notify_all();
for (auto & worker : workers) worker.join();
if (terminal.joinable()) terminal.join();
return 0;
} catch (const std::exception & e) {
std::cerr << "startup_error: " << e.what() << std::endl; return 1;
}
}