Download src/portable_main.cpp from Sariel00/Ling-3.0-tiny-RKNN: direct link, hf CLI and curl.
- Browser
- Download file 56.3 kB
-
https://huggingface.co/Sariel00/Ling-3.0-tiny-RKNN/resolve/main/src/portable_main.cpp
- Command line
-
hf download hf://Sariel00/Ling-3.0-tiny-RKNN/src/portable_main.cpp
-
curl -L -o portable_main.cpp https://huggingface.co/Sariel00/Ling-3.0-tiny-RKNN/resolve/main/src/portable_main.cpp
56.3 kB
| 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() { | |
| throw std::runtime_error("non-production build: exact ICU tokenizer support is required"); | |
| 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; | |
| } | |
| } | |