Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 23 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -419,3 +419,26 @@ This project's own code is licensed under the [MIT License](LICENSE).
Commercial use, modification, and redistribution are permitted, provided
the copyright and license notices are retained. Model weights and
third-party components remain subject to their respective licenses.

## Engine lifecycle and failure contract (0.2.1)

`Engine.check_health()` reads the native decode-loop lifecycle without submitting inference.
It returns `None` while the loop has not exited. It is a liveness snapshot, not a progress
heartbeat: a busy or hung native call is not declared dead merely for lack of output.

`EngineUnavailableError(EngineError)` means this instance's loop has definitively exited.
`EngineClosedError(EngineUnavailableError)` means the caller explicitly closed it.
Both reject future inference. Ordinary `EngineError` and `InvalidRequestError` concern
individual requests and do not imply permanent engine death. Errors preserve `error_type`,
`code`, and `details`; native loop errors include the first C++ exception type and message.
C++ has no portable traceback here; Python traceback and adapter exception chains remain available.

A catchable exception escaping the decode loop is recorded, and waiting readers stop on their
next polling interval. `abort`, access violations, and whole-process termination remain process
supervisor responsibilities. This library neither terminates the host nor rebuilds an engine.

Cancellation of `agenerate()` waits for submitted native work to finish; work still queued in
its executor skips inference after cancellation. Explicit `astream().aclose()` or cancellation
also drains its worker. Use `contextlib.aclosing` when stopping asynchronous iteration early.
Drain all requests before calling `Engine.close()`/`aclose()`; the engine belongs to its caller.
A fatal signal discovered during draining is preserved for batch error arbitration.
9 changes: 9 additions & 0 deletions docs/runtime-lifecycle-validation.md
Original file line number Diff line number Diff line change
Expand Up @@ -61,3 +61,12 @@ CPU 按需加载改动位于 Engine;上游独立 llama-server 仍执行自己
结构化结果、wheel 哈希及 ELF 数据在 `build/runtime-lifecycle-final/validation.json`;
完整测试日志为 `tests.log`,GPU 日志位于 `gpu-smoke/` 和 `gpu-engine.log`,重复配置日志为 `reconfigure.log`。
此前的根因堆栈、符号绑定和隔离实验保留在 `build/runtime-lifecycle-diagnosis/`。


## 0.2.1 故障状态契约补充

历史验证只覆盖正常 shutdown;本轮新增循环退出锁存、明确关闭异常和异步取消排空。
`tests/test_engine_lifecycle_state.py` 在真实 C++ 线程抛出异常,验证首次原因和终止状态;
`tests/test_engine_failure_contract.py` 验证公开 Python 错误及取消规则;实际模型验收脚本位于
MinerU 的 `scripts/acceptance/llama_cpp_lifecycle.py`。合成异常、真实引擎显式关闭与整个进程退出
必须分开记录,不能据此宣称已经修复所有原生崩溃根因。
6 changes: 3 additions & 3 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ build-backend = "scikit_build_core.build"

[project]
name = "mineru-llama-cpp"
version = "0.2.0"
version = "0.2.1"
description = "In-process llama.cpp VLM inference engine for MinerU"
readme = "README.md"
license = {file = "LICENSE"}
Expand Down Expand Up @@ -93,7 +93,7 @@ build = "cp311-* cp312-* cp313-* cp314-*"
skip = "*-musllinux_*"
build-frontend = "build"
test-requires = ["pytest", "pytest-asyncio", "pillow", "setuptools", "packaging"]
test-command = "python {project}/tools/wheel_smoke.py --without-external-gpu-runtime && python -m pytest {project}/tests/test_output_utf8.py {project}/tests/test_backend_policy.py {project}/tests/test_binding_contract.py {project}/tests/test_optional_backend_load.py {project}/tests/test_cpu_topology.py {project}/tests/test_runtime_loading.py {project}/tests/test_queue_lifecycle.py {project}/tests/test_wheel_policy.py -q"
test-command = "python {project}/tools/wheel_smoke.py --without-external-gpu-runtime && python -m pytest {project}/tests/test_output_utf8.py {project}/tests/test_backend_policy.py {project}/tests/test_binding_contract.py {project}/tests/test_optional_backend_load.py {project}/tests/test_cpu_topology.py {project}/tests/test_runtime_loading.py {project}/tests/test_queue_lifecycle.py {project}/tests/test_wheel_policy.py {project}/tests/test_engine_failure_contract.py -q"
manylinux-x86_64-image = "manylinux_2_28"
manylinux-aarch64-image = "manylinux_2_28"

Expand Down Expand Up @@ -121,4 +121,4 @@ inherit = { "environment" = "none", "before-all" = "none", "before-build" = "non
environment = { CMAKE_GENERATOR = "Ninja", VULKAN_SDK = "C:/VulkanSDK-ARM64", SKBUILD_CMAKE_ARGS = "-DCMAKE_C_COMPILER=clang-cl;-DCMAKE_CXX_COMPILER=clang-cl;-DGGML_OPENMP=OFF;-DGGML_VULKAN=ON;-DVulkan_LIBRARY=C:/VulkanSDK-ARM64/Lib/vulkan-1.lib;-DVulkan_GLSLC_EXECUTABLE=C:/VulkanSDK-ARM64/Bin/glslc.exe" }
before-all = "pip install ninja cmake && python {project}/tools/setup_windows_gpu.py --arch ARM64"
before-build = "python {project}/tools/clean_native_artifacts.py"
test-command = "python {project}/tools/windows_vulkan_smoke.py && python {project}/tools/wheel_smoke.py --without-external-gpu-runtime && python -m pytest {project}/tests/test_output_utf8.py {project}/tests/test_backend_policy.py {project}/tests/test_binding_contract.py {project}/tests/test_optional_backend_load.py {project}/tests/test_cpu_topology.py {project}/tests/test_runtime_loading.py {project}/tests/test_queue_lifecycle.py {project}/tests/test_wheel_policy.py -q"
test-command = "python {project}/tools/windows_vulkan_smoke.py && python {project}/tools/wheel_smoke.py --without-external-gpu-runtime && python -m pytest {project}/tests/test_output_utf8.py {project}/tests/test_backend_policy.py {project}/tests/test_binding_contract.py {project}/tests/test_optional_backend_load.py {project}/tests/test_cpu_topology.py {project}/tests/test_runtime_loading.py {project}/tests/test_queue_lifecycle.py {project}/tests/test_wheel_policy.py {project}/tests/test_engine_failure_contract.py -q"
49 changes: 42 additions & 7 deletions src/cpp/binding.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -109,24 +109,44 @@ int integer_argument(PyObject * object) {
}

// 保留生产异常分类,将服务端错误转换为 Python 异常。
[[noreturn]] void raise_mapped_error(const std::string & type, const std::string & message) {
[[noreturn]] void raise_mapped_error(const std::string & type, const std::string & message,
const std::string & details = "{}") {
OwnedRef module(checked(PyImport_ImportModule("mineru_llama_cpp.exceptions")));
const char * name = type == "exceed_context_size_error" ? "ContextExceededError" :
type == "invalid_request_error" ? "InvalidRequestError" : "EngineError";
type == "invalid_request_error" ? "InvalidRequestError" :
type == "engine_dead_error" ? "EngineUnavailableError" : "EngineError";
OwnedRef exception(checked(PyObject_GetAttrString(module.get(), name)));
PyErr_SetString(exception.get(), message.c_str());
OwnedRef json_module(checked(PyImport_ImportModule("json")));
OwnedRef diagnostic(checked(PyObject_CallMethod(json_module.get(), "loads", "s", details.c_str())));
OwnedRef args(checked(Py_BuildValue("(s)", message.c_str())));
OwnedRef kwargs(checked(Py_BuildValue("{s:s,s:O}", "error_type", type.c_str(), "details", diagnostic.get())));
PyObject * code = PyDict_GetItemString(diagnostic.get(), "code");
if (code && PyDict_SetItemString(kwargs.get(), "code", code) < 0) throw PythonError{};
OwnedRef value(checked(PyObject_Call(exception.get(), args.get(), kwargs.get())));
PyErr_SetObject(exception.get(), value.get());
throw PythonError{};
}

// 解析错误 JSON;无效 JSON 继续使用原字符串作为 EngineError 消息。
// 将原生首次致命异常映射为专用类型,保留原始 C++ 类型和消息。
[[noreturn]] void raise_engine_dead(const EngineTerminated & error) {
common_json details = {{"type", "engine_dead_error"}, {"code", "engine_dead"},
{"message", error.what()}, {"native_type", error.native_type}};
raise_mapped_error("engine_dead_error", error.what(), details.dump());
}

// 解析错误 JSON 并保留完整响应;格式损坏时仍保留原始诊断文本。
[[noreturn]] void raise_from_error_json(const std::string & text) {
std::string type = "server_error", message = text;
common_json details = {{"raw_response", text}};
try {
auto error = common_json::parse(text);
type = error.value("type", type);
message = error.value("message", text);
if (error.is_object()) {
type = error.value("type", type);
message = error.value("message", text);
details = error;
}
} catch (...) {}
raise_mapped_error(type, message);
raise_mapped_error(type, message, details.dump());
}

// 接管字典值的新引用,写入失败时也释放引用。
Expand Down Expand Up @@ -168,6 +188,9 @@ PyObject * decode_generated_content(const std::string & content, const std::stri
void translate_exception() {
try { throw; }
catch (const PythonError &) {}
catch (const EngineTerminated & error) {
try { raise_engine_dead(error); } catch (const PythonError &) {}
}
catch (const std::bad_alloc &) { PyErr_NoMemory(); }
catch (const std::invalid_argument & error) { PyErr_SetString(PyExc_ValueError, error.what()); }
catch (const std::out_of_range & error) { PyErr_SetString(PyExc_IndexError, error.what()); }
Expand Down Expand Up @@ -234,6 +257,7 @@ PyObject * core_generate(CoreObject * self, PyObject * args, PyObject * kwargs)
auto body = parse_body(args, kwargs);
EngineCore::GenerateResult result;
try { ReleasedGIL release; result = self->core->generate(body); }
catch (const EngineTerminated & error) { raise_engine_dead(error); }
catch (const std::exception & error) { raise_mapped_error("invalid_request_error", error.what()); }
if (result.is_error) raise_from_error_json(result.error_json);
OwnedRef output(checked(PyDict_New()));
Expand All @@ -246,6 +270,15 @@ PyObject * core_generate(CoreObject * self, PyObject * args, PyObject * kwargs)
} catch (...) { translate_exception(); return nullptr; }
}

// 提供无模型调用、无新增任务的真实循环状态探测。
PyObject * core_check_health(CoreObject * self, PyObject *) {
try {
require_core(self);
self->core->check_health();
return none();
} catch (...) { translate_exception(); return nullptr; }
}

// 只允许 generate_stream 创建迭代器,禁止无 reader 的空实例。
PyObject * stream_new(PyTypeObject *, PyObject *, PyObject *) {
PyErr_SetString(PyExc_TypeError, "_StreamIterator cannot be constructed directly");
Expand All @@ -259,6 +292,7 @@ PyObject * core_stream(CoreObject * self, PyObject * args, PyObject * kwargs) {
auto body = parse_body(args, kwargs);
std::unique_ptr<EngineCore::StreamHandle> handle;
try { handle = std::make_unique<EngineCore::StreamHandle>(self->core->generate_stream(body)); }
catch (const EngineTerminated & error) { raise_engine_dead(error); }
catch (const std::exception & error) { raise_mapped_error("invalid_request_error", error.what()); }
auto * state = static_cast<ModuleState *>(PyType_GetModuleState(Py_TYPE(reinterpret_cast<PyObject *>(self))));
if (!state) throw PythonError{};
Expand Down Expand Up @@ -312,6 +346,7 @@ PyObject * stream_next(StreamObject * self) {
}

PyMethodDef core_methods[] = {
{"check_health", reinterpret_cast<PyCFunction>(core_check_health), METH_NOARGS, "检查真实后台循环状态。"},
{"generate", reinterpret_cast<PyCFunction>(core_generate), METH_VARARGS | METH_KEYWORDS, "同步生成并返回统计。"},
{"generate_stream", reinterpret_cast<PyCFunction>(core_stream), METH_VARARGS | METH_KEYWORDS, "创建流式 reader。"},
{nullptr, nullptr, 0, nullptr}
Expand Down
20 changes: 14 additions & 6 deletions src/cpp/engine_core.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -357,19 +357,21 @@ EngineCore::EngineCore(const std::string & model_path, const std::string & mmpro
throw;
}
meta_ = std::make_unique<server_context_meta>(ctx_.get_meta());
loop_thread_ = std::thread([this] { ctx_.start_loop(); });
loop_thread_ = std::thread([this] { lifecycle_.run([this] { ctx_.start_loop(); }); });
started_ = true;
}

EngineCore::~EngineCore() {
if (started_) {
lifecycle_.begin_close();
ctx_.terminate();
if (loop_thread_.joinable()) loop_thread_.join();
}
llama_backend_free();
}

EngineCore::GenerateResult EngineCore::generate(const std::string & body_json) {
check_health();
// 无效 JSON 的 common_json_error 继续在绑定层映射为请求异常。
json body = json::parse(body_json);

Expand Down Expand Up @@ -397,10 +399,12 @@ EngineCore::GenerateResult EngineCore::generate(const std::string & body_json) {
task.cli_files = std::move(files);
}

check_health();
rd.post_task(std::move(task));

GenerateResult result;
auto r = rd.next([]() { return false; }); // blocks until the result arrives; no GIL concept at this layer
auto r = rd.next([this]() { return lifecycle_.stopped(); });
check_health();
if (!r) {
result.is_error = true;
result.error_json = R"json({"type":"server_error","message":"no result (stopped)"})json";
Expand All @@ -421,8 +425,8 @@ EngineCore::GenerateResult EngineCore::generate(const std::string & body_json) {
return result;
}

EngineCore::StreamHandle::StreamHandle(std::unique_ptr<server_response_reader> rd)
: rd_(std::move(rd)) {}
EngineCore::StreamHandle::StreamHandle(std::unique_ptr<server_response_reader> rd, const EngineLifecycle & lifecycle)
: rd_(std::move(rd)), lifecycle_(lifecycle) {}

EngineCore::Chunk EngineCore::StreamHandle::next_chunk() {
Chunk chunk;
Expand All @@ -435,7 +439,9 @@ EngineCore::Chunk EngineCore::StreamHandle::next_chunk() {
return chunk;
}

auto r = rd_->next([]() { return false; });
lifecycle_.check_health();
auto r = rd_->next([this]() { return lifecycle_.stopped(); });
lifecycle_.check_health();
if (!r) {
chunk.is_error = true;
chunk.error_json = R"json({"type":"server_error","message":"no result (stopped)"})json";
Expand Down Expand Up @@ -469,6 +475,7 @@ EngineCore::Chunk EngineCore::StreamHandle::next_chunk() {
}

EngineCore::StreamHandle EngineCore::generate_stream(const std::string & body_json) {
check_health();
json body = json::parse(body_json); // throws on malformed input; propagates to binding layer

// IMPORTANT: construct the ONE server_response_reader that will live for
Expand Down Expand Up @@ -509,6 +516,7 @@ EngineCore::StreamHandle EngineCore::generate_stream(const std::string & body_js
task.cli_files = std::move(files);
}

check_health();
reader->post_task(std::move(task));
return StreamHandle(std::move(reader)); // only the unique_ptr moves; the reader's identity never changes
return StreamHandle(std::move(reader), lifecycle_); // reader 身份及生命周期引用随迭代器保留。
}
8 changes: 7 additions & 1 deletion src/cpp/engine_core.h
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
#include "server-context.h"
#include "server-queue.h"
#include "server-task.h"
#include "engine_lifecycle.h"

#include <cstdint>
#include <memory>
Expand Down Expand Up @@ -86,7 +87,7 @@ class EngineCore {
// rd_ to a by-value server_response_reader member.
class StreamHandle {
public:
explicit StreamHandle(std::unique_ptr<server_response_reader> rd);
explicit StreamHandle(std::unique_ptr<server_response_reader> rd, const EngineLifecycle & lifecycle);
// Blocks until the next token (or the final chunk, or an error) is
// ready. May throw std::exception for pre-post errors, but in
// practice those all happen inside generate_stream() before a
Expand All @@ -96,6 +97,7 @@ class EngineCore {

private:
std::unique_ptr<server_response_reader> rd_;
const EngineLifecycle & lifecycle_;
bool done_ = false;
};

Expand All @@ -111,12 +113,16 @@ class EngineCore {
// token gets rendered as literal text instead of being silently dropped.
const std::string & eos_token_str() const { return meta_->eos_token_str; }

// 暴露真实后台循环状态;不读取或修改上游私有任务。
void check_health() const { lifecycle_.check_health(); }

private:
server_context ctx_;
common_params params_;
std::unique_ptr<server_context_meta> meta_;
std::thread loop_thread_;
bool started_ = false;
EngineLifecycle lifecycle_;

// Guards oaicompat_chat_params_parse() + eval_llama_cmpl_schema() +
// post_task() for both generate() and generate_stream(). Fixes the
Expand Down
55 changes: 55 additions & 0 deletions src/cpp/engine_lifecycle.h
Original file line number Diff line number Diff line change
@@ -0,0 +1,55 @@
#pragma once
#include <exception>
#include <mutex>
#include <stdexcept>
#include <string>
#include <typeinfo>

// 仅用于已确认退出的引擎,不能由普通服务端请求错误构造。
struct EngineTerminated : std::runtime_error {
std::string native_type;
// 保留首次原生异常的类型名与消息;C++ 不提供可移植的栈回溯。
EngineTerminated(const std::string & message, const std::string & type)
: std::runtime_error(message), native_type(type) {}
};

class EngineLifecycle {
public:
// 在后台线程边界捕获可恢复到宿主的异常;abort/访问违规仍交由进程管理器处理。
template<class Function> void run(Function function) noexcept {
std::exception_ptr failure;
try { function(); } catch (...) { failure = std::current_exception(); }
std::lock_guard<std::mutex> lock(mutex_);
if (!stopped_) failure_ = failure;
stopped_ = true;
}

// 关闭只停止接单,不覆盖已经记录的首次致命原因。
void begin_close() {
std::lock_guard<std::mutex> lock(mutex_);
closing_ = true;
}

// 响应 reader 用此状态退出等待,不终止上游结果队列或宿主进程。
bool stopped() const {
std::lock_guard<std::mutex> lock(mutex_);
return stopped_ || closing_;
}

// 快速状态快照只确认循环是否已退出,不将推理繁忙或无心跳视为死亡。
void check_health() const {
std::lock_guard<std::mutex> lock(mutex_);
if (failure_) {
try { std::rethrow_exception(failure_); }
catch (const std::exception & error) { throw EngineTerminated(error.what(), typeid(error).name()); }
catch (...) { throw EngineTerminated("unknown native loop exception", "unknown"); }
}
if (closing_) throw EngineTerminated("engine is closing", "");
if (stopped_) throw EngineTerminated("engine loop exited unexpectedly", "");
}
private:
mutable std::mutex mutex_;
std::exception_ptr failure_;
bool stopped_ = false;
bool closing_ = false;
};
4 changes: 4 additions & 0 deletions src/mineru_llama_cpp/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,8 @@
from .exceptions import (
ContextExceededError,
EngineError,
EngineClosedError,
EngineUnavailableError,
InvalidRequestError,
MineruLlamaCppError,
)
Expand Down Expand Up @@ -67,6 +69,8 @@
"InvalidRequestError",
"ContextExceededError",
"EngineError",
"EngineClosedError",
"EngineUnavailableError",
"Message",
"Messages",
"ContentPart",
Expand Down
3 changes: 3 additions & 0 deletions src/mineru_llama_cpp/_mineru_llama_cpp.pyi
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,9 @@ __all__: list[str] = list()
class _EngineCore:
def __init__(self, model_path: str, mmproj_path: str, n_ctx_seq: typing.SupportsInt | typing.SupportsIndex, n_gpu_layers: typing.SupportsInt | typing.SupportsIndex, n_parallel: typing.SupportsInt | typing.SupportsIndex, verbosity: typing.SupportsInt | typing.SupportsIndex, n_threads: typing.SupportsInt | typing.SupportsIndex) -> None:
...
def check_health(self) -> None:
"""查询原生循环状态,退出时抛出专用异常。"""
...
def generate(self, body: str) -> dict:
...
def generate_stream(self, body: str) -> _StreamIterator:
Expand Down
Loading
Loading