From 25bdcf6b7dab052a650537571d21432ddaa73e01 Mon Sep 17 00:00:00 2001 From: myhloli Date: Fri, 13 Nov 2026 18:58:16 +0800 Subject: [PATCH] fix(engine): preserve terminal state and drain cancelled llama.cpp calls --- README.md | 23 +++ docs/runtime-lifecycle-validation.md | 9 + pyproject.toml | 6 +- src/cpp/binding.cpp | 49 +++++- src/cpp/engine_core.cpp | 20 ++- src/cpp/engine_core.h | 8 +- src/cpp/engine_lifecycle.h | 55 ++++++ src/mineru_llama_cpp/__init__.py | 4 + src/mineru_llama_cpp/_mineru_llama_cpp.pyi | 3 + src/mineru_llama_cpp/engine.py | 127 +++++++++++--- src/mineru_llama_cpp/exceptions.py | 21 +++ tests/native/engine_lifecycle_test.cpp | 40 +++++ tests/native/queue_lifecycle_test.cpp | 32 ++++ tests/test_engine_failure_contract.py | 195 +++++++++++++++++++++ tests/test_engine_lifecycle_state.py | 42 +++++ tests/test_queue_lifecycle.py | 12 +- 16 files changed, 605 insertions(+), 41 deletions(-) create mode 100644 src/cpp/engine_lifecycle.h create mode 100644 tests/native/engine_lifecycle_test.cpp create mode 100644 tests/test_engine_failure_contract.py create mode 100644 tests/test_engine_lifecycle_state.py diff --git a/README.md b/README.md index 9504a9a..04c24d7 100644 --- a/README.md +++ b/README.md @@ -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. diff --git a/docs/runtime-lifecycle-validation.md b/docs/runtime-lifecycle-validation.md index e458f29..db55c39 100644 --- a/docs/runtime-lifecycle-validation.md +++ b/docs/runtime-lifecycle-validation.md @@ -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`。合成异常、真实引擎显式关闭与整个进程退出 +必须分开记录，不能据此宣称已经修复所有原生崩溃根因。 diff --git a/pyproject.toml b/pyproject.toml index 8240a34..8879b1b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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"} @@ -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" @@ -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" diff --git a/src/cpp/binding.cpp b/src/cpp/binding.cpp index 798a56e..a6426a8 100644 --- a/src/cpp/binding.cpp +++ b/src/cpp/binding.cpp @@ -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()); } // 接管字典值的新引用，写入失败时也释放引用。 @@ -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()); } @@ -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())); @@ -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"); @@ -259,6 +292,7 @@ PyObject * core_stream(CoreObject * self, PyObject * args, PyObject * kwargs) { auto body = parse_body(args, kwargs); std::unique_ptr handle; try { handle = std::make_unique(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(PyType_GetModuleState(Py_TYPE(reinterpret_cast(self)))); if (!state) throw PythonError{}; @@ -312,6 +346,7 @@ PyObject * stream_next(StreamObject * self) { } PyMethodDef core_methods[] = { + {"check_health", reinterpret_cast(core_check_health), METH_NOARGS, "检查真实后台循环状态。"}, {"generate", reinterpret_cast(core_generate), METH_VARARGS | METH_KEYWORDS, "同步生成并返回统计。"}, {"generate_stream", reinterpret_cast(core_stream), METH_VARARGS | METH_KEYWORDS, "创建流式 reader。"}, {nullptr, nullptr, 0, nullptr} diff --git a/src/cpp/engine_core.cpp b/src/cpp/engine_core.cpp index 995c145..2cc345e 100644 --- a/src/cpp/engine_core.cpp +++ b/src/cpp/engine_core.cpp @@ -357,12 +357,13 @@ EngineCore::EngineCore(const std::string & model_path, const std::string & mmpro throw; } meta_ = std::make_unique(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(); } @@ -370,6 +371,7 @@ EngineCore::~EngineCore() { } EngineCore::GenerateResult EngineCore::generate(const std::string & body_json) { + check_health(); // 无效 JSON 的 common_json_error 继续在绑定层映射为请求异常。 json body = json::parse(body_json); @@ -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"; @@ -421,8 +425,8 @@ EngineCore::GenerateResult EngineCore::generate(const std::string & body_json) { return result; } -EngineCore::StreamHandle::StreamHandle(std::unique_ptr rd) - : rd_(std::move(rd)) {} +EngineCore::StreamHandle::StreamHandle(std::unique_ptr rd, const EngineLifecycle & lifecycle) + : rd_(std::move(rd)), lifecycle_(lifecycle) {} EngineCore::Chunk EngineCore::StreamHandle::next_chunk() { Chunk chunk; @@ -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"; @@ -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 @@ -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 身份及生命周期引用随迭代器保留。 } diff --git a/src/cpp/engine_core.h b/src/cpp/engine_core.h index f2c4dca..ecd9ba1 100644 --- a/src/cpp/engine_core.h +++ b/src/cpp/engine_core.h @@ -5,6 +5,7 @@ #include "server-context.h" #include "server-queue.h" #include "server-task.h" +#include "engine_lifecycle.h" #include #include @@ -86,7 +87,7 @@ class EngineCore { // rd_ to a by-value server_response_reader member. class StreamHandle { public: - explicit StreamHandle(std::unique_ptr rd); + explicit StreamHandle(std::unique_ptr 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 @@ -96,6 +97,7 @@ class EngineCore { private: std::unique_ptr rd_; + const EngineLifecycle & lifecycle_; bool done_ = false; }; @@ -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 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 diff --git a/src/cpp/engine_lifecycle.h b/src/cpp/engine_lifecycle.h new file mode 100644 index 0000000..bcf87e8 --- /dev/null +++ b/src/cpp/engine_lifecycle.h @@ -0,0 +1,55 @@ +#pragma once +#include +#include +#include +#include +#include + +// 仅用于已确认退出的引擎，不能由普通服务端请求错误构造。 +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 void run(Function function) noexcept { + std::exception_ptr failure; + try { function(); } catch (...) { failure = std::current_exception(); } + std::lock_guard lock(mutex_); + if (!stopped_) failure_ = failure; + stopped_ = true; + } + + // 关闭只停止接单，不覆盖已经记录的首次致命原因。 + void begin_close() { + std::lock_guard lock(mutex_); + closing_ = true; + } + + // 响应 reader 用此状态退出等待，不终止上游结果队列或宿主进程。 + bool stopped() const { + std::lock_guard lock(mutex_); + return stopped_ || closing_; + } + + // 快速状态快照只确认循环是否已退出，不将推理繁忙或无心跳视为死亡。 + void check_health() const { + std::lock_guard 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; +}; diff --git a/src/mineru_llama_cpp/__init__.py b/src/mineru_llama_cpp/__init__.py index a910610..180646b 100644 --- a/src/mineru_llama_cpp/__init__.py +++ b/src/mineru_llama_cpp/__init__.py @@ -36,6 +36,8 @@ from .exceptions import ( ContextExceededError, EngineError, + EngineClosedError, + EngineUnavailableError, InvalidRequestError, MineruLlamaCppError, ) @@ -67,6 +69,8 @@ "InvalidRequestError", "ContextExceededError", "EngineError", + "EngineClosedError", + "EngineUnavailableError", "Message", "Messages", "ContentPart", diff --git a/src/mineru_llama_cpp/_mineru_llama_cpp.pyi b/src/mineru_llama_cpp/_mineru_llama_cpp.pyi index d76529c..14d24c3 100644 --- a/src/mineru_llama_cpp/_mineru_llama_cpp.pyi +++ b/src/mineru_llama_cpp/_mineru_llama_cpp.pyi @@ -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: diff --git a/src/mineru_llama_cpp/engine.py b/src/mineru_llama_cpp/engine.py index e499ffd..d173ff3 100644 --- a/src/mineru_llama_cpp/engine.py +++ b/src/mineru_llama_cpp/engine.py @@ -7,9 +7,10 @@ import threading from pathlib import Path from types import TracebackType -from typing import AsyncIterator, Iterator +from typing import AsyncIterator, Callable, Iterator, TypeVar from ._mineru_llama_cpp import _EngineCore +from .exceptions import EngineClosedError, EngineUnavailableError from .sampling import SamplingParams from .types import GenerateChunk, GenerateResult, GenerationTimings, Messages from .verbosity import LOG_LEVEL_WARN @@ -43,6 +44,46 @@ def _timings_from_dict(d: dict | None) -> GenerationTimings | None: return GenerationTimings(**d) +_Result = TypeVar("_Result") + + +async def _wait_until_complete( + work: asyncio.Future[_Result], cancelled: threading.Event | None = None +) -> _Result: + """取消后排空已提交线程，明确死亡优先上报，其他情形保留调用方取消。""" + try: + await asyncio.wait([work]) + return work.result() + except asyncio.CancelledError: + if cancelled is not None: + cancelled.set() + while not work.done(): + try: + await asyncio.wait([work]) + except asyncio.CancelledError: + continue + if not work.cancelled(): + error = work.exception() + if isinstance(error, EngineUnavailableError): + raise error + raise + + +async def _run_in_thread(function: Callable[[], _Result]) -> _Result: + """未开始的工作取消后跳过执行；已开始的工作等待结束再交还控制权。""" + cancelled = threading.Event() + + def run() -> _Result | None: + """在线程开始时检查取消，避免排队任务在取消后启动迟到推理。""" + if cancelled.is_set(): + return None + return function() + + return await _wait_until_complete( + asyncio.get_running_loop().run_in_executor(None, run), cancelled + ) + + class Engine: def __init__( self, @@ -95,7 +136,9 @@ def __init__( # cryptic assert deep inside llama.cpp) rather than a clear error here. # n_ctx_seq == 0 is valid and means "use the model's training context". if n_ctx_seq < 0: - raise ValueError(f"n_ctx_seq must be >= 0 (0 = model training context), got {n_ctx_seq}") + raise ValueError( + f"n_ctx_seq must be >= 0 (0 = model training context), got {n_ctx_seq}" + ) if n_parallel < 1: raise ValueError(f"n_parallel must be >= 1, got {n_parallel}") # Round a nonzero n_ctx_seq up to a multiple of 256 so it survives @@ -106,7 +149,13 @@ def __init__( if n_ctx_seq > 0 and n_ctx_seq % 256 != 0: n_ctx_seq = ((n_ctx_seq + 255) // 256) * 256 self._core = _EngineCore( - str(model), str(mmproj), n_ctx_seq, n_gpu_layers, n_parallel, verbosity, n_threads + str(model), + str(mmproj), + n_ctx_seq, + n_gpu_layers, + n_parallel, + verbosity, + n_threads, ) # The engine is constructed with special=true (see engine_core.cpp), # which is needed to keep MinerU's structured tokens (<|box_start|> @@ -119,8 +168,14 @@ def __init__( # find_stopping_strings()/erase() in process_token(). self._eos_token_str: str = self._core.eos_token_str - def _build_body(self, messages: Messages, sampling_params: SamplingParams | None, stream: bool) -> str: - body: dict = {"model": "mineru-llama-cpp", "messages": messages, "stream": stream} + def _build_body( + self, messages: Messages, sampling_params: SamplingParams | None, stream: bool + ) -> str: + body: dict = { + "model": "mineru-llama-cpp", + "messages": messages, + "stream": stream, + } if sampling_params is not None: body.update(sampling_params.to_json_fields()) # Default, not an override -- see _VALID_UNICODE_GRAMMAR's @@ -140,8 +195,9 @@ def generate( ) -> GenerateResult: """Blocking call. Raises InvalidRequestError/ContextExceededError/ EngineError on failure (see exceptions.py).""" + core = self._require_core() body = self._build_body(messages, sampling_params, stream=False) - d = self._core.generate(body) + d = core.generate(body) return GenerateResult( content=d["content"], finish_reason=d["finish_reason"], @@ -157,8 +213,7 @@ async def agenerate( ) -> GenerateResult: """Async version of generate(), bridged via run_in_executor — does not block the event loop.""" - loop = asyncio.get_event_loop() - return await loop.run_in_executor(None, self.generate, messages, sampling_params) + return await _run_in_thread(lambda: self.generate(messages, sampling_params)) # --- streaming --- @@ -170,8 +225,9 @@ def stream( """Synchronous generator, yields one GenerateChunk per token. The final chunk has finish_reason set (all others have it as None) — check `chunk.finish_reason is not None` to detect the end.""" + core = self._require_core() body = self._build_body(messages, sampling_params, stream=True) - core_iter = self._core.generate_stream(body) + core_iter = core.generate_stream(body) for c in core_iter: yield GenerateChunk( delta=c["delta"], @@ -198,33 +254,63 @@ async def astream( abandoning this generator before it's exhausted) does not cancel the underlying generation — the background thread runs to completion regardless, occupying a slot until the model finishes. + Explicit aclose() or cancellation waits for that work to drain. """ loop = asyncio.get_event_loop() queue: asyncio.Queue = asyncio.Queue() _SENTINEL = object() + abandoned = threading.Event() + work: asyncio.Future[None] = loop.create_future() + def _run() -> None: + """在工作线程生成，消费者退出后仅排空而不继续缓存文本。""" + fatal: EngineUnavailableError | None = None try: for chunk in self.stream(messages, sampling_params): - loop.call_soon_threadsafe(queue.put_nowait, chunk) + if not abandoned.is_set(): + loop.call_soon_threadsafe(queue.put_nowait, chunk) except Exception as exc: # noqa: BLE001 - relayed to the consumer, not swallowed loop.call_soon_threadsafe(queue.put_nowait, exc) + if isinstance(exc, EngineUnavailableError): + fatal = exc finally: loop.call_soon_threadsafe(queue.put_nowait, _SENTINEL) + if fatal is not None: + loop.call_soon_threadsafe(work.set_exception, fatal) + else: + loop.call_soon_threadsafe(work.set_result, None) + # 保留独立流线程，避免同步流占满共享 executor 后阻塞其他异步调用。 thread = threading.Thread(target=_run, daemon=True) thread.start() - - while True: - item = await queue.get() - if item is _SENTINEL: - break - if isinstance(item, Exception): - raise item - yield item + try: + while True: + item = await queue.get() + if item is _SENTINEL: + break + if isinstance(item, Exception): + raise item + yield item + finally: + abandoned.set() + await _wait_until_complete(work) # --- lifecycle --- + def _require_core(self) -> _EngineCore: + """在读取或生成前拒绝已关闭实例，并持有局部引用直至本次调用排空。""" + with self._close_lock: + if self._closed: + raise EngineClosedError( + "engine is closed", error_type="engine_closed", code="engine_closed" + ) + return self._core + + def check_health(self) -> None: + """查询原生循环终止状态；活着但卡住的循环不伪称已死亡，不发送推理任务。""" + self._require_core().check_health() + def close(self) -> None: """Explicit shutdown: terminate + join the background loop thread, free the llama backend. Idempotent and thread-safe (safe to call @@ -239,13 +325,14 @@ def close(self) -> None: if self._closed: return self._closed = True - del self._core # drops the last reference -> EngineCore's C++ destructor runs now + # drops the last reference -> EngineCore's C++ destructor runs now + del self._core async def aclose(self) -> None: """Async version of close(), offloaded to a thread so it doesn't block the event loop while the loop thread joins.""" loop = asyncio.get_event_loop() - await loop.run_in_executor(None, self.close) + await _wait_until_complete(loop.run_in_executor(None, self.close)) def __enter__(self) -> Engine: return self diff --git a/src/mineru_llama_cpp/exceptions.py b/src/mineru_llama_cpp/exceptions.py index c108a37..094b2ef 100644 --- a/src/mineru_llama_cpp/exceptions.py +++ b/src/mineru_llama_cpp/exceptions.py @@ -10,6 +10,19 @@ class MineruLlamaCppError(Exception): """Base class for all exceptions raised by this library.""" + def __init__( + self, + *args: object, + error_type: str | None = None, + code: int | str | None = None, + details: dict | None = None, + ) -> None: + """保留原始类型、错误码和完整服务响应，避免上层只能按消息猜测分类。""" + super().__init__(*args) + self.error_type = error_type + self.code = code + self.details = dict(details or {}) + class InvalidRequestError(MineruLlamaCppError): """The request itself was invalid: malformed JSON, bad image data, etc. @@ -43,3 +56,11 @@ class EngineError(MineruLlamaCppError): not_supported_error, unavailable_error) or to an unrecognized/missing "type" field. """ + + +class EngineUnavailableError(EngineError): + """已确认后台循环退出，此实例永久不可复用；普通 EngineError 不代表死亡。""" + + +class EngineClosedError(EngineUnavailableError): + """调用方已关闭实例；后续请求须使用显式新建的引擎。""" diff --git a/tests/native/engine_lifecycle_test.cpp b/tests/native/engine_lifecycle_test.cpp new file mode 100644 index 0000000..d2f3820 --- /dev/null +++ b/tests/native/engine_lifecycle_test.cpp @@ -0,0 +1,40 @@ +#include "engine_lifecycle.h" +#include +#include +#include + +// 验证后台退出锁存、原始异常诊断、并发读取及正常关闭。 +int main() { + EngineLifecycle state; + state.check_health(); + std::thread loop([&] { state.run([] { throw std::logic_error("first failure"); }); }); + loop.join(); + assert(state.stopped()); + state.run([] { throw std::runtime_error("second failure"); }); + std::vector readers; + for (int i = 0; i < 8; ++i) readers.emplace_back([&] { + try { state.check_health(); assert(false); } + catch (const EngineTerminated & error) { + assert(std::string(error.what()) == "first failure"); + assert(error.native_type == typeid(std::logic_error).name()); + } + }); + for (auto & reader : readers) reader.join(); + state.begin_close(); + try { state.check_health(); assert(false); } + catch (const EngineTerminated & error) { assert(std::string(error.what()) == "first failure"); } + EngineLifecycle exited; + exited.run([] {}); + assert(exited.stopped()); + try { exited.check_health(); assert(false); } + catch (const EngineTerminated & error) { assert(std::string(error.what()).find("exited") != std::string::npos); } + EngineLifecycle closing; + closing.begin_close(); + closing.run([] {}); + try { closing.check_health(); assert(false); } + catch (const EngineTerminated & error) { assert(std::string(error.what()).find("closing") != std::string::npos); } + EngineLifecycle unknown; + unknown.run([] { throw 42; }); + try { unknown.check_health(); assert(false); } + catch (const EngineTerminated & error) { assert(error.native_type == "unknown"); } +} diff --git a/tests/native/queue_lifecycle_test.cpp b/tests/native/queue_lifecycle_test.cpp index 026dd09..14463de 100644 --- a/tests/native/queue_lifecycle_test.cpp +++ b/tests/native/queue_lifecycle_test.cpp @@ -1,5 +1,7 @@ #include "server-queue.h" +#include "engine_lifecycle.h" #include +#include #include #include @@ -7,6 +9,36 @@ int main(int argc, char ** argv) { assert(argc == 2); const std::string scenario = argv[1]; + if (scenario == "fatal-reader") { + // 使用真实上游响应队列验证：后台循环抛错后，所有等待者通过状态轮询退出。 + server_queue queue; + server_response responses; + EngineLifecycle lifecycle; + queue.on_new_task([](server_task &&, bool) { return true; }); + queue.on_update_slots([] { throw std::logic_error("decode loop failure"); }); + std::vector> readers; + for (int i = 0; i < 4; ++i) { + auto reader = std::make_unique(queue, responses, 1); + server_task task(SERVER_TASK_TYPE_COMPLETION); + task.id = reader->get_new_id(); + reader->post_task(std::move(task)); + readers.emplace_back(std::move(reader)); + } + std::vector> waiting; + for (auto & reader : readers) waiting.emplace_back(std::async(std::launch::async, [&lifecycle, ptr = reader.get()] { + auto result = ptr->next([&] { return lifecycle.stopped(); }); + assert(!result); + try { lifecycle.check_health(); assert(false); } + catch (const EngineTerminated & error) { assert(std::string(error.what()) == "decode loop failure"); } + })); + std::thread loop([&] { lifecycle.run([&] { queue.start_loop(); }); }); + loop.join(); + for (auto & future : waiting) { + assert(future.wait_for(std::chrono::seconds(5)) == std::future_status::ready); + future.get(); + } + return 0; + } const int repeats = scenario == "concurrent" ? 100 : 1; for (int iteration = 0; iteration < repeats; ++iteration) { server_queue queue; diff --git a/tests/test_engine_failure_contract.py b/tests/test_engine_failure_contract.py new file mode 100644 index 0000000..769b71b --- /dev/null +++ b/tests/test_engine_failure_contract.py @@ -0,0 +1,195 @@ +"""验证实际公开异常类、关闭拒绝及异步线程排空契约。""" + +import asyncio +import threading +from concurrent.futures import ThreadPoolExecutor +from types import SimpleNamespace + +import pytest + +from mineru_llama_cpp import ( + Engine, + EngineClosedError, + EngineError, + EngineUnavailableError, + InvalidRequestError, +) + + +@pytest.fixture +def empty_engine(monkeypatch: pytest.MonkeyPatch) -> Engine: + """只替换模型核心，保留生产 Engine 的公共边界和异步实现。""" + import mineru_llama_cpp.engine as module + + monkeypatch.setattr( + module, "_EngineCore", lambda *args: SimpleNamespace(eos_token_str="") + ) + return Engine("unused", "unused") + + +def test_structured_error_hierarchy() -> None: + """普通服务错误不代表实例死亡，明确关闭同时属于致命及服务错误。""" + ordinary = EngineError( + "busy", code=503, error_type="unavailable_error", details={"param": "slot"} + ) + assert not isinstance(ordinary, EngineUnavailableError) + assert ordinary.code == 503 and ordinary.details == {"param": "slot"} + assert not issubclass(InvalidRequestError, EngineError) + assert issubclass(EngineClosedError, EngineUnavailableError) + + +def test_closed_engine_rejects_every_entry(empty_engine: Engine) -> None: + """显式关闭后健康、同步及生成器入口均返回精确错误，重复关闭安全。""" + empty_engine.close() + empty_engine.close() + for call in [ + empty_engine.check_health, + lambda: empty_engine.generate([]), + lambda: next(empty_engine.stream([])), + ]: + with pytest.raises(EngineClosedError) as captured: + call() + assert captured.value.code == "engine_closed" + + +@pytest.mark.asyncio +async def test_closed_async_entries(empty_engine: Engine) -> None: + """异步与同步使用同一关闭边界。""" + empty_engine.close() + with pytest.raises(EngineClosedError): + await empty_engine.agenerate([]) + with pytest.raises(EngineClosedError): + await anext(empty_engine.astream([])) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("fatal", [False, True]) +async def test_cancel_drains_generation(empty_engine: Engine, fatal: bool) -> None: + """重复取消不越过线程排空，排空期间的确认死亡仍能上报批次收集器。""" + started, release, ended = threading.Event(), threading.Event(), threading.Event() + error = EngineUnavailableError("loop stopped", code="engine_dead") + + def generate(*args): + """模拟已进入原生推理的线程，在测试许可后退出。""" + started.set() + try: + assert release.wait(5) + if fatal: + raise error + return None + finally: + ended.set() + + empty_engine.generate = generate + task = asyncio.create_task(empty_engine.agenerate([])) + await asyncio.to_thread(started.wait, 5) + task.cancel() + await asyncio.sleep(0) + task.cancel() + await asyncio.sleep(0) + assert not task.done() and not ended.is_set() + release.set() + with pytest.raises(EngineUnavailableError if fatal else asyncio.CancelledError): + await task + assert ended.is_set() + + +def test_cancelled_queue_never_starts_inference(empty_engine: Engine) -> None: + """独立循环中占满线程池，验证排队期间取消不会迟到调用引擎。""" + started, release = threading.Event(), threading.Event() + calls = [] + empty_engine.generate = lambda *args: calls.append(args) + + def occupy() -> None: + """占用唯一线程，直到取消已生效。""" + started.set() + assert release.wait(5) + + async def scenario() -> None: + """在独立默认线程池上控制提交与取消顺序。""" + loop = asyncio.get_running_loop() + loop.set_default_executor(ThreadPoolExecutor(max_workers=1)) + busy = loop.run_in_executor(None, occupy) + while not started.is_set(): + await asyncio.sleep(0.001) + task = asyncio.create_task(empty_engine.agenerate([])) + await asyncio.sleep(0) + task.cancel() + await asyncio.sleep(0) + release.set() + with pytest.raises(asyncio.CancelledError): + await task + await busy + + asyncio.run(scenario()) + assert calls == [] + + +@pytest.mark.asyncio +async def test_astream_close_drains_worker(empty_engine: Engine) -> None: + """显式关闭异步生成器时等待同步流排空，不能提前释放模型。""" + release, ended = threading.Event(), threading.Event() + + def stream(*args): + """首片输出后阻塞，模拟仍占用模型的原生流。""" + yield "first" + assert release.wait(5) + ended.set() + + empty_engine.stream = stream + stream_iter = empty_engine.astream([]) + assert await anext(stream_iter) == "first" + closing = asyncio.create_task(stream_iter.aclose()) + await asyncio.sleep(0) + assert not closing.done() + release.set() + await closing + assert ended.is_set() + + +@pytest.mark.asyncio +async def test_astream_fatal_during_close_is_preserved(empty_engine: Engine) -> None: + """流消费者停止后才到达的致命异常仍要交给宿主，不能留在无人读取的队列。""" + release = threading.Event() + original = EngineUnavailableError("loop failed while draining", code="engine_dead") + + def stream(*args): + """在首片输出之后、消费者关闭期间模拟真实类型的终止故障。""" + yield "first" + assert release.wait(5) + raise original + + empty_engine.stream = stream + iterator = empty_engine.astream([]) + assert await anext(iterator) == "first" + closing = asyncio.create_task(iterator.aclose()) + await asyncio.sleep(0) + release.set() + with pytest.raises(EngineUnavailableError) as raised: + await closing + assert raised.value is original + + +@pytest.mark.asyncio +async def test_aclose_cancellation_waits_for_release(empty_engine: Engine) -> None: + """取消异步关闭不能让调用方误判原生资源已经释放。""" + started, release, ended = threading.Event(), threading.Event(), threading.Event() + + def close() -> None: + """用可控释放过程模拟等待原生后台线程退出。""" + started.set() + assert release.wait(5) + ended.set() + + empty_engine.close = close + closing = asyncio.create_task(empty_engine.aclose()) + assert await asyncio.to_thread(started.wait, 5) + closing.cancel() + await asyncio.sleep(0) + closing.cancel() + await asyncio.sleep(0) + assert not closing.done() + release.set() + with pytest.raises(asyncio.CancelledError): + await closing + assert ended.is_set() diff --git a/tests/test_engine_lifecycle_state.py b/tests/test_engine_lifecycle_state.py new file mode 100644 index 0000000..e9a12f7 --- /dev/null +++ b/tests/test_engine_lifecycle_state.py @@ -0,0 +1,42 @@ +"""不加载模型，验证生产后台循环的状态与诊断契约。""" + +import os +import shlex +import subprocess +import sys +from pathlib import Path + + +def test_native_loop_failure_state(tmp_path: Path) -> None: + """编译生产状态组件，覆盖真正线程抛错和意外正常退出。""" + root = Path(__file__).resolve().parents[1] + executable = tmp_path / ( + "lifecycle.exe" if sys.platform == "win32" else "lifecycle" + ) + command = [ + *shlex.split(os.environ.get("CXX", "c++")), + "-std=c++17", + "-pthread", + "-UNDEBUG", + ] + command += [ + "-I", + str(root / "src/cpp"), + str(root / "tests/native/engine_lifecycle_test.cpp"), + "-o", + str(executable), + ] + if sys.platform == "win32": + command = [ + "cl", + "/nologo", + "/std:c++17", + "/EHsc", + "/utf-8", + "/UNDEBUG", + f"/I{root / 'src/cpp'}", + str(root / "tests/native/engine_lifecycle_test.cpp"), + f"/Fe:{executable}", + ] + subprocess.run(command, check=True, cwd=tmp_path) + subprocess.run([str(executable)], check=True, timeout=10) diff --git a/tests/test_queue_lifecycle.py b/tests/test_queue_lifecycle.py index 8261f78..58b9c5a 100644 --- a/tests/test_queue_lifecycle.py +++ b/tests/test_queue_lifecycle.py @@ -12,9 +12,9 @@ @pytest.fixture(scope="session") def queue_probe(tmp_path_factory: pytest.TempPathFactory) -> Path: - """链接已安装的 Linux wheel，构建队列生命周期探针。""" - if sys.platform != "linux": - pytest.skip("This native probe links the Linux wheel's C++ server symbols") + """链接已安装 wheel 的真实服务器队列，构建生命周期探针。""" + if sys.platform not in {"linux", "darwin"}: + pytest.skip("This probe requires exported ELF/Mach-O C++ server symbols") import mineru_llama_cpp root = Path(__file__).resolve().parents[1] @@ -22,6 +22,7 @@ def queue_probe(tmp_path_factory: pytest.TempPathFactory) -> Path: libraries = Path(mineru_llama_cpp.__file__).resolve().parent / "lib" executable = tmp_path_factory.mktemp("queue_lifecycle") / "probe" command = [*shlex.split(os.environ.get("CXX", "c++")), "-std=c++17", "-pthread", "-UNDEBUG"] + command += ["-I", str(root / "src/cpp")] for directory in ["tools/server", "common", "include", "ggml/include", "tools/mtmd", "vendor"]: command += ["-I", str(upstream / directory)] command += [ @@ -38,11 +39,14 @@ def queue_probe(tmp_path_factory: pytest.TempPathFactory) -> Path: "-o", str(executable), ] + if sys.platform == "darwin": + command = [str(libraries / (arg[3:].replace(".so.0", ".0.dylib").replace(".so", ".dylib"))) + if arg.startswith("-l:") else arg for arg in command] subprocess.run(command, check=True) return executable -@pytest.mark.parametrize("scenario", ["early-stop", "running-stop", "concurrent"]) +@pytest.mark.parametrize("scenario", ["early-stop", "running-stop", "concurrent", "fatal-reader"]) def test_queue_termination_is_terminal(queue_probe: Path, scenario: str) -> None: # Test-only deadlock guard; this probe never loads a model or runs inference. """验证停止请求不会被延迟启动或并发启动覆盖。"""