From 27248700c1a51370decd67592add3a96213dfbc9 Mon Sep 17 00:00:00 2001 From: Anthony Shoumikhin Date: Sun, 1 Nov 2026 18:07:27 -0800 Subject: [PATCH 1/2] Python runtime: stop a backend's log from deadlocking 2 threads The Python bindings wrote each runtime log line through sys.stderr.write as it was logged, whenever the logging thread held the GIL (Python's global interpreter lock). A backend can log while it holds its own lock, as XNNPACK does when a load fails, and a Python write can let another thread take the GIL. If that thread then waits for the backend's lock, for example to execute or to destroy a method, the first thread never gets the GIL back and both hang. The log function no longer calls Python. While sys.stderr is still the stream Python opened on the process's stderr, each line goes straight there through C stdio, so a terminal or a log file sees it at once, even if the process later crashes. When Python has replaced sys.stderr, as pytest's capsys, contextlib.redirect_stderr and Jupyter do, most lines still reach it. A binding call keeps its lines and writes them to sys.stderr as it returns, when no backend lock is held. A line that Python's main thread logs outside any binding call, for example while the LLM runner loads, is written by that thread before it runs more Python code, so it still lands in a short redirect. It logged the line from native code, so it is not inside its own sys.stderr.write. A line from any other thread goes to a thread that the bindings start for it. A binding call's lines then follow the lines its thread handed to that thread, so one thread's lines keep their order. Python's main thread may be in the middle of a sys.stderr.write whose stream holds a lock, and a write nested there would wait for that lock forever. Outside a call, only a thread that holds the GIL can tell where sys.stderr goes, by comparing it with the stream checked last, a check that runs no Python code. A thread that does not hold the GIL writes to the process's stderr, as before, so a notebook with ipykernel's fd capture off does not show its lines. Lines from a thread that holds the GIL do reach such a notebook. On Linux and macOS, a line also waits once the process's stderr is a pipe or socket that stays full for 100 ms, so a thread that holds the GIL does not hang on it while a Python thread in the same process drains it. The rest of a binding call's lines then wait for the call to return. A line logged outside a call does not wait: the bindings' thread writes it without the GIL or any lock that Python's exit needs. While that thread writes, a binding call's lines wait for the call to return, since it can take the room a check just saw. A terminal is written at once, since its reader is another process. The process can still hang if another writer fills the pipe right after the check for room, or if a Python thread in the same process reads a terminal that is its stderr. macOS reports room in a small pipe that it may not have, so there a pipe that has not grown past 16 KiB takes a line only once it fits in 512 bytes beside what the pipe holds. Up to 1 MiB of lines wait at a time, keeping the newest and noting how much was left out. A fatal line goes out at once while there is room, with the lines its call kept, since an abort usually follows. If writing to sys.stderr fails, the lines go to the process's stderr instead, and a Ctrl-C raised in that write is raised again once the call returns. Once Python starts to exit, the lines still waiting are written, and later lines logged outside a call go to the process's stderr while it has room. Lines for the process's stderr are left out instead while the bindings' thread is still writing to it. On Linux with Python before 3.13.8, a daemon thread never writes its lines through Python. Those versions can end such a thread at exit in the middle of that write, and the unwinding then releases Python objects without the GIL and crashes the process. The old log function crashed this way too. The bindings' thread writes the daemon thread's lines instead. New tests cover the hang, also when the LLM runner logs under XNNPACK's lock outside any call, a sys.stderr whose write runs Python code while it holds a lock, daemon threads that log while Python exits, lines logged outside a call with sys.stderr replaced before or after import or left alone, a daemon thread's lines with a replaced sys.stderr, a failing write, a fatal line and the line before it, a call that dies in a busy terminal, a full pipe or socket drained by a thread, for a call, for the LLM runner and for both at once, and a long call's newest lines. The docs example for registering a PAL (platform abstraction layer) now uses the right namespace and compiles on its own, and the docs say why the bindings' log function never calls Python. --- .../using-executorch-runtime-integration.md | 50 +- extension/pybindings/README.md | 43 +- extension/pybindings/pybindings.cpp | 549 +++++++++++- extension/pybindings/test/BUCK | 2 + extension/pybindings/test/test_pybindings.py | 805 ++++++++++++++++++ 5 files changed, 1408 insertions(+), 41 deletions(-) diff --git a/docs/source/using-executorch-runtime-integration.md b/docs/source/using-executorch-runtime-integration.md index cac59affdff..0bf94fa4f5f 100644 --- a/docs/source/using-executorch-runtime-integration.md +++ b/docs/source/using-executorch-runtime-integration.md @@ -36,33 +36,45 @@ To register a custom PAL implementation, take the following steps: - Create an instance of the [PalImpl](https://github.com/pytorch/executorch/blob/7b39a0ce63bfb5124d4d29cfb6c8af85a3c580ba/runtime/platform/platform.h#L163) struct. - Set one or more fields to custom PAL function implementations. Leave fields as null to use the default platform implementation. - The PalImpl struct provides a [create](https://github.com/pytorch/executorch/blob/7b39a0ce63bfb5124d4d29cfb6c8af85a3c580ba/runtime/platform/platform.h#L168) method for this purpose. -- Call `executorch::platform::register_pal(pal_impl)` to register the implementation. +- Call `executorch::runtime::register_pal(pal_impl)` to register the implementation. - This can be done from as as a global constructor, as in the example below. -Here is a complete example from [pybindings.cpp](https://github.com/pytorch/executorch/blob/7b39a0ce63bfb5124d4d29cfb6c8af85a3c580ba/extension/pybindings/pybindings.cpp#L1178), where logs are redirected to show up properly in a Python notebook environment. +Here is a complete example that writes each log line to stderr. ```cpp +#include + +#include + namespace { - void emit_log_message( - et_timestamp_t timestamp, - et_pal_log_level_t level, - const char* filename, - ET_UNUSED const char* function, - size_t line, - const char* message, - ET_UNUSED size_t length) { - std::cerr << "[" << filename << ":" << line << "] " << message << std::endl; - } - - runtime::PalImpl build_pal() { - return runtime::PalImpl::create(emit_log_message, __FILE__); - } - - // Update PAL to redirect logs. - ET_UNUSED bool registration_result = runtime::register_pal(build_pal()); + +void emit_log_message( + ET_UNUSED et_timestamp_t timestamp, + ET_UNUSED et_pal_log_level_t level, + const char* filename, + ET_UNUSED const char* function, + size_t line, + const char* message, + size_t length) { + std::fprintf( + stderr, "[%s:%zu] %.*s\n", filename, line, static_cast(length), message); + std::fflush(stderr); } + +ET_UNUSED const bool registered = executorch::runtime::register_pal( + executorch::runtime::PalImpl::create(emit_log_message, __FILE__)); + +} // namespace ``` +The Python bindings register their PAL the same way, in +[pybindings.cpp](https://github.com/pytorch/executorch/blob/main/extension/pybindings/pybindings.cpp). +A backend can log while it holds a lock, and a log function that writes to +Python at that moment can deadlock with another thread that waits for that lock, +so their log function never calls Python. Their +[README](https://github.com/pytorch/executorch/blob/main/extension/pybindings/README.md#note) +says where their lines go. + ### Weak Symbol Override ExecuTorch also provides a link-time method to override the PAL using weak symbols. This method is primarily maintained for backwards compatibility. diff --git a/extension/pybindings/README.md b/extension/pybindings/README.md index 352c7a6f93a..634560b6b0c 100644 --- a/extension/pybindings/README.md +++ b/extension/pybindings/README.md @@ -234,4 +234,45 @@ CMAKE_ARGS="-DEXECUTORCH_BUILD_VULKAN=ON" ./install_executorch.sh This class is currently empty and serves as a placeholder for future methods and attributes. - `verify_result_with_bundled_expected_output(method_name: str, testset_idx: int, rtol: float = 1e-5, atol: float = 1e-8)`: Verify result with bundled expected output. ## Note -Runtime logs are forwarded to Python stderr. +Runtime logs go straight to the process's stderr as they are logged. On Windows, +a console shows text that is not ASCII in its own code page, so such text can +look wrong there. If Python has replaced `sys.stderr`, for example with pytest's +`capsys`, `contextlib.redirect_stderr` or in a Jupyter notebook, the lines go to +`sys.stderr` instead, a little later. The lines a binding call logs wait until +the call returns. A line logged outside any binding call, for example by the LLM +runner, goes to `sys.stderr` before the next Python code runs if Python's main +thread logged it. A line from another thread goes there a moment later, from a +thread that the bindings start for this. Outside a call, only a thread that +holds the GIL can tell where `sys.stderr` goes, so a line from a thread that +does not hold it, such as one the LLM runner logs while it generates, goes to +the process's stderr. The lines wait rather than go to Python at once because a +backend can log while it holds a lock, and writing to Python at that moment +could deadlock with another thread. + +On Linux and macOS, a line also waits once the process's stderr is a pipe or +socket that stays full for 100 ms, so a thread that holds the GIL does not hang +on it while a Python thread in the same process empties it, as under +`wurlitzer.pipes()`. The rest of a binding call's lines then wait for the call +to return. A thread that logs outside a call does not wait for room itself. It +hands the line to the bindings' thread, which writes it without the GIL. While +it writes, a binding call's lines wait for the call to return. A terminal is +written at once, since its reader is normally another process. The process can +still hang if another writer fills the process's stderr right after the check +for room, or if a Python thread in the same process reads a terminal that is the +process's stderr. On macOS, a pipe that has not grown past 16 KiB takes a line +only once the line fits in 512 bytes beside what the pipe holds, because macOS +reports room in such a pipe that it may not have. + +Up to 1 MiB of lines wait at a time; past that, about the newest 512 KiB to +1 MiB are kept, with a note of how many earlier bytes were left out. A fatal +line goes straight to the process's stderr while it has room, with the lines its +binding call kept, since an abort usually follows. Any crash or hang of the +process, including that abort, or an `os._exit()` loses the lines still waiting. +Once Python starts to exit, the lines still waiting are written, and later lines +logged outside a call go to the process's stderr while it has room. Lines for +the process's stderr are left out instead if the bindings' thread is still +writing to it then. + +On Linux with Python before 3.13.8, a daemon thread never writes through Python, +because at exit Python can stop such a thread in the middle of that write, which +crashes the process. The bindings' thread writes its lines instead. diff --git a/extension/pybindings/pybindings.cpp b/extension/pybindings/pybindings.cpp index 9889c9622a8..4f4efa26cf4 100644 --- a/extension/pybindings/pybindings.cpp +++ b/extension/pybindings/pybindings.cpp @@ -8,19 +8,29 @@ #include #include +#include #include +#include #include #include #include -#include #include #include #include #include #include +#include #include #include #include +#ifndef _WIN32 +#include +#include +#include +#endif +#ifdef __APPLE__ +#include +#endif #include #include @@ -139,9 +149,438 @@ namespace { constexpr size_t kRuntimeErrorsCapacity = 1024; thread_local char runtime_errors[kRuntimeErrorsCapacity]; thread_local size_t runtime_errors_size = 0; -// How many binding calls are running on this thread. Lines logged outside -// any, such as on worker threads, are not kept, so they never linger. -thread_local size_t runtime_error_scopes = 0; +// The lines kept for the current binding call; emit_log_message says when a +// line is kept. Past the capacity the oldest are counted rather than kept, +// since the newest are the closest to a failure. +struct PendingLogs final { + bool direct = false; + bool daemon = false; + std::string lines; + size_t dropped = 0; +}; +constexpr size_t kPendingLogsCapacity = 1024 * 1024; +// Null outside a binding call, so a line logged outside any is never kept here, +// where no call would write it out. A plain pointer for the same reason as +// above. +thread_local PendingLogs* pending_logs = nullptr; + +// Lines that Python's stderr must show but that were logged where Python cannot +// be called: outside any binding call, or on a daemon thread. Python's main +// thread writes the lines it logged itself at its next chance. A thread of +// their own writes the others, since the main thread may be in the middle of a +// sys.stderr.write that holds a lock. Never freed, so that a line logged during +// exit still finds it. +struct DeferredLogs final { + std::mutex mutex; + std::condition_variable writer_changed; + PendingLogs logs; + PendingLogs main_thread_logs; + bool main_thread_write_scheduled = false; + // Numbers the writer's batches, so that a thread can tell whether the lines + // it handed over are written. + uint64_t batch = 1; + uint64_t batches_written = 0; + bool writer_running = false; + // A new thread state crashes a Python that has finished exiting, so exit + // waits while the writer takes the GIL with one. + bool writer_entering = false; + bool exiting = false; +}; + +DeferredLogs*& deferred_logs_state() { + static DeferredLogs* logs = new DeferredLogs(); + return logs; +} + +DeferredLogs& deferred_logs() { + return *deferred_logs_state(); +} + +// The only thread that runs Python's pending calls. +unsigned long main_thread_id = 0; + +// The writer's batch that holds the newest line this thread handed over. +thread_local uint64_t last_deferred_batch = 0; + +void defer_lines(const std::string& lines) noexcept; +bool deferred_lines_wait(); +bool own_deferred_lines_wait(); + +void write_to_stderr(const char* data, size_t size) { + std::fwrite(data, 1, size, stderr); + std::fflush(stderr); +} + +// Writes to fd 2 without C stdio's lock on stderr, so that a write that waits +// for room holds up no thread that flushes stderr, as Python does as it exits. +void write_to_fd_2(const char* data, size_t size) { +#ifdef _WIN32 + write_to_stderr(data, size); +#else + while (size > 0) { + const ssize_t written = ::write(2, data, size); + if (written < 0 && errno == EINTR) { + continue; + } + if (written <= 0) { + return; + } + data += written; + size -= static_cast(written); + } +#endif +} + +// The last stream whose descriptor stderr_is_process_stderr() checked, kept +// alive by this reference, and whether that descriptor is 2. A stream that +// replaced sys.stderr is never kept here. +PyObject* checked_stderr = nullptr; +bool checked_stderr_is_fd_2 = false; + +// Whether sys.stderr is still the stream Python opened on file descriptor 2. +// Runs at import and as a binding call starts, with the GIL held and no lock +// held. Asking a stream for its descriptor costs more than a small binding +// call, so the answer is kept for the last stream checked. +bool stderr_is_process_stderr() noexcept { + PyObject* const current = PySys_GetObject("stderr"); + if (current != nullptr && current == checked_stderr) { + return checked_stderr_is_fd_2; + } + if (current == nullptr || current != PySys_GetObject("__stderr__")) { + return false; + } + Py_INCREF(current); + const bool on_fd_2 = PyObject_AsFileDescriptor(current) == 2; + PyErr_Clear(); + checked_stderr_is_fd_2 = on_fd_2; + Py_XSETREF(checked_stderr, current); + return on_fd_2; +} + +// Whether sys.stderr is still the stream that stderr_is_process_stderr() found +// on fd 2. Runs no Python code, so a thread that logs can ask it. +bool sys_stderr_is_checked_fd_2() noexcept { + return PySys_GetObject("stderr") == checked_stderr && checked_stderr_is_fd_2; +} + +// Whether this thread holds the GIL. PyGILState_Check says yes on every thread +// once a subinterpreter exists, so this compares thread states instead. +bool holds_gil() noexcept { + if (!Py_IsInitialized()) { + return false; + } + PyThreadState* const state = PyGILState_GetThisThreadState(); + return state != nullptr && state == py::detail::get_thread_state_unchecked(); +} + +// Whether this is a daemon thread that a write through Python could crash at +// exit. Before 3.13.8, Python ends a daemon thread with pthread_exit if it then +// waits for the GIL, and on Linux the unwinding runs the C++ destructors on its +// stack, which touch Python objects without the GIL. So such a thread never +// writes its lines through Python. The flag cannot change once the thread runs. +bool on_daemon_thread() { +#ifdef __linux__ + thread_local const bool daemon = Py_Version < 0x030D0800 && + py::module_::import("threading") + .attr("current_thread")() + .attr("daemon") + .cast(); + return daemon; +#else + return false; +#endif +} + +// How long a line waits for room in a full stderr. A reader in another process +// usually catches up within it, while a Python thread in this process never +// does, since it needs the GIL. +constexpr int kStderrRoomWaitMs = 100; + +// Whether a line of `size` bytes can go to the process's stderr, waiting up to +// `wait_ms` for room, so that a thread that holds the GIL never waits on a full +// pipe or socket that a Python thread in this process reads, as +// wurlitzer.pipes() sets up. A terminal is not checked: its reader is another +// process, and a line that waits for it must not be lost to a crash. +bool stderr_has_room(ET_UNUSED size_t size, int wait_ms) { +#ifdef _WIN32 + return true; +#else + if (isatty(2)) { + return true; + } +#ifdef __APPLE__ + // macOS reports room as if a pipe held 16 KiB, but a pipe made while pipe + // memory runs low holds as little as 512 bytes, and a write to it then waits. + // A pipe that grew past 16 KiB reports its real size. Until then, a line goes + // in only once it fits in 512 bytes beside what the pipe holds. + struct stat info; + if (fstat(2, &info) == 0 && S_ISFIFO(info.st_mode) && + info.st_blksize <= 16 * 1024) { + const auto deadline = + std::chrono::steady_clock::now() + std::chrono::milliseconds(wait_ms); + while (static_cast(info.st_size) + size > PIPE_BUF) { + if (std::chrono::steady_clock::now() >= deadline || usleep(1000) != 0 || + fstat(2, &info) != 0) { + return false; + } + } + return true; + } +#endif + pollfd stderr_fd{2, POLLOUT, 0}; + return poll(&stderr_fd, 1, wait_ms) > 0; +#endif +} + +// Drops the oldest half of the kept lines at once when they outgrow the +// capacity, so keeping a line stays cheap. +void keep_line(PendingLogs& logs, const std::string& line) { + logs.lines += line; + if (logs.lines.size() > kPendingLogsCapacity) { + const size_t cut = + logs.lines.find( + '\n', logs.lines.size() - kPendingLogsCapacity / 2 - 1) + + 1; + logs.dropped += cut; + logs.lines.erase(0, cut); + } +} + +void note_left_out_bytes(PendingLogs& logs) noexcept { + if (logs.dropped == 0) { + return; + } + try { + logs.lines.insert( + 0, + "[" + std::to_string(logs.dropped) + + " earlier bytes of the runtime log were left out]\n"); + logs.dropped = 0; + } catch (const std::bad_alloc&) { + } +} + +// Writes lines to fd 2, oldest first, while it has room, and returns how many +// bytes it wrote. Whole lines only, since a pipe that a Python thread drains +// stays full while the GIL is held. +size_t write_lines_while_room(const std::string& lines) noexcept { + size_t begin = 0; + while (begin < lines.size()) { + const size_t end = lines.find('\n', begin) + 1; + if (!stderr_has_room(end - begin, kStderrRoomWaitMs)) { + break; + } + write_to_stderr(lines.data() + begin, end - begin); + begin = end; + } + return begin; +} + +// Writes the kept lines to fd 2 while it has room, and keeps the rest. +void write_kept_lines_while_room(PendingLogs& logs) noexcept { + logs.lines.erase(0, write_lines_while_room(logs.lines)); +} + +// Runs as the outermost binding call returns, or at exit, with the GIL held and +// no backend lock or object lock held. An exception may be unwinding, so it +// never throws. +void write_logs_to_python(PendingLogs& logs) noexcept { + note_left_out_bytes(logs); + if (logs.lines.empty()) { + return; + } + // A call's lines go after the lines this thread handed to the writer thread. + if (logs.daemon || own_deferred_lines_wait()) { + defer_lines(logs.lines); + return; + } + try { + // A line that is not valid UTF-8, such as one that prints raw bytes from a + // model file, must not keep the rest from Python. + const auto text = py::reinterpret_steal(PyUnicode_DecodeUTF8( + logs.lines.data(), logs.lines.size(), "backslashreplace")); + if (!text) { + throw py::error_already_set(); + } + py::module_::import("sys").attr("stderr").attr("write")(text); + return; + } catch (py::error_already_set& error) { + // Python runs a Ctrl-C's handler in the first Python code after it, which + // can be this write. Raise it again once the call has returned. + if (error.matches(PyExc_KeyboardInterrupt)) { + PyErr_SetInterrupt(); + } + } catch (...) { + } + // A write that was cut short may have shown some lines already, but + // repeating them is better than losing them. stderr can be a pipe that a + // Python thread drains, as in Jupyter, so the GIL is released. + py::gil_scoped_release release; + write_to_stderr(logs.lines.data(), logs.lines.size()); +} + +// Runs on Python's main thread between two bytecodes, with the GIL held. +int write_main_thread_logs(void*) noexcept { + PendingLogs logs; + { + std::lock_guard lock(deferred_logs().mutex); + deferred_logs().main_thread_write_scheduled = false; + std::swap(logs, deferred_logs().main_thread_logs); + } + write_logs_to_python(logs); + return 0; +} + +// Writes through Python's C API only, and is not noexcept, since Python may end +// this thread while it waits for the GIL during exit, and the unwinding must +// neither touch a Python object nor stop at a noexcept frame. +void write_lines_through_python(const PendingLogs& logs) { + PyObject* const text = PyUnicode_DecodeUTF8( + logs.lines.data(), logs.lines.size(), "backslashreplace"); + PyObject* const stream = PySys_GetObject("stderr"); + PyObject* const result = text == nullptr || stream == nullptr + ? nullptr + : PyObject_CallMethod(stream, "write", "O", text); + const bool written = result != nullptr; + Py_XDECREF(result); + Py_XDECREF(text); + if (!written) { + PyErr_Clear(); + PyThreadState* const state = PyEval_SaveThread(); + write_to_fd_2(logs.lines.data(), logs.lines.size()); + PyEval_RestoreThread(state); + } +} + +// Runs on a thread of its own until no line waits. +void write_deferred_logs() { + DeferredLogs& deferred = deferred_logs(); + for (;;) { + PendingLogs logs; + uint64_t batch = 0; + { + std::lock_guard lock(deferred.mutex); + if (deferred.exiting || deferred.logs.lines.empty()) { + deferred.writer_running = false; + deferred.writer_changed.notify_all(); + return; + } + std::swap(logs, deferred.logs); + batch = deferred.batch++; + deferred.writer_entering = true; + } + const PyGILState_STATE gil = PyGILState_Ensure(); + { + std::lock_guard lock(deferred.mutex); + deferred.writer_entering = false; + deferred.writer_changed.notify_all(); + } + note_left_out_bytes(logs); + if (stderr_is_process_stderr()) { + // Waits for room without the GIL or any lock of Python's, so that a full + // fd 2 never holds up Python's exit. + PyThreadState* const state = PyEval_SaveThread(); + write_to_fd_2(logs.lines.data(), logs.lines.size()); + PyEval_RestoreThread(state); + } else { + write_lines_through_python(logs); + } + PyGILState_Release(gil); + std::lock_guard lock(deferred.mutex); + deferred.batches_written = batch; + } +} + +// Runs with the GIL held. +void defer_lines(const std::string& lines) noexcept { + DeferredLogs& deferred = deferred_logs(); + std::lock_guard lock(deferred.mutex); + if (deferred.exiting) { + // Python's main thread may need the GIL to finish exiting, so the lines go + // out only while fd 2 has room, and not at all while a writer that is still + // writing can take that room first. + if (!deferred.writer_running) { + write_lines_while_room(lines); + } + return; + } + try { + // The main thread logged these lines from native code, which is inside a + // sys.stderr.write only if that write runs such code. So it can write them + // before its next bytecode, while a short redirect of sys.stderr still + // lasts. Its earlier lines stay ahead. + if (PyThread_get_thread_ident() == main_thread_id && + (deferred.main_thread_write_scheduled || + !sys_stderr_is_checked_fd_2())) { + keep_line(deferred.main_thread_logs, lines); + if (!deferred.main_thread_write_scheduled) { + deferred.main_thread_write_scheduled = + Py_AddPendingCall(&write_main_thread_logs, nullptr) == 0; + } + return; + } + keep_line(deferred.logs, lines); + last_deferred_batch = deferred.batch; + if (!deferred.writer_running) { + std::thread(write_deferred_logs).detach(); + deferred.writer_running = true; + } + } catch (const std::exception&) { + // Lines that no thread could take wait for the exit hook. + } +} + +// Whether lines wait for the writer thread or for Python's main thread. While +// the writer writes to fd 2, the room that a check sees can be gone. +bool deferred_lines_wait() { + DeferredLogs& deferred = deferred_logs(); + std::lock_guard lock(deferred.mutex); + return (deferred.writer_running || deferred.main_thread_write_scheduled) && + !deferred.exiting; +} + +bool own_deferred_lines_wait() { + DeferredLogs& deferred = deferred_logs(); + std::lock_guard lock(deferred.mutex); + return last_deferred_batch > deferred.batches_written && !deferred.exiting; +} + +// Python stops other threads once exit begins, and later its state is gone, so +// the waiting lines go out now and later ones go to fd 2 while it has room. A +// writer that is taking the GIL gets it first, and one that is writing gets a +// moment to finish, so that its stream is free for Python's last flush. +void write_deferred_logs_at_exit() noexcept { + DeferredLogs& deferred = deferred_logs(); + PendingLogs main_thread_logs; + PendingLogs logs; + bool writer_running = false; + { + py::gil_scoped_release release; + std::unique_lock lock(deferred.mutex); + deferred.exiting = true; + std::swap(main_thread_logs, deferred.main_thread_logs); + std::swap(logs, deferred.logs); + deferred.writer_changed.wait( + lock, [&deferred] { return !deferred.writer_entering; }); + deferred.writer_changed.wait_for( + lock, std::chrono::milliseconds(kStderrRoomWaitMs), [&deferred] { + return !deferred.writer_running; + }); + writer_running = deferred.writer_running; + } + for (PendingLogs* waiting : {&main_thread_logs, &logs}) { + if (stderr_is_process_stderr()) { + // No Python thread may be left to read a full fd 2. A writer still + // writing means fd 2 stayed full, and it could take any room seen here. + if (!writer_running) { + note_left_out_bytes(*waiting); + write_kept_lines_while_room(*waiting); + } + } else { + write_logs_to_python(*waiting); + } + } +} std::string with_runtime_errors(const char* message) { std::string result(message); @@ -153,20 +592,31 @@ std::string with_runtime_errors(const char* message) { } // Starts every binding call with no runtime errors recorded, and gives an -// outer call its own back when a nested call returns. +// outer call its own back when a nested call returns. The outermost call owns +// the lines kept during it and writes them out as it returns. They live on the +// heap, so a pointer to them stays valid if a callback switches greenlets. struct RuntimeErrorScope final { RuntimeErrorScope() : outer_(runtime_errors, runtime_errors_size) { runtime_errors_size = 0; - ++runtime_error_scopes; + if (pending_logs == nullptr) { + logs_ = std::make_unique(); + logs_->daemon = on_daemon_thread(); + logs_->direct = stderr_is_process_stderr(); + pending_logs = logs_.get(); + } } ~RuntimeErrorScope() { - --runtime_error_scopes; std::memcpy(runtime_errors, outer_.data(), outer_.size()); runtime_errors_size = outer_.size(); + if (logs_ != nullptr) { + pending_logs = nullptr; + write_logs_to_python(*logs_); + } } private: std::string outer_; + std::unique_ptr logs_; }; #ifndef USE_ATEN_LIB @@ -4891,11 +5341,31 @@ py::dict get_option(const std::string& backend_name, const py::dict& options) { } // namespace PYBIND11_MODULE(EXECUTORCH_PYTHON_MODULE_NAME, m) { - // Gives each call its own record of runtime errors for its exceptions. - // Runtime logs reach Python through emit_log_message below rather than + // Gives each call its own record of runtime errors for its exceptions, and + // decides where the call's runtime logs go. This replaces // scoped_ostream_redirect, which mutates process-wide stream buffers and is // not thread-safe when Python operations inside a binding release the GIL. + // It calls Python as it starts and may write to Python as it ends, so it + // needs the GIL and never follows a py::gil_scoped_release guard. auto call_guard = py::call_guard(); + // A line logged outside any call compares sys.stderr with the stream checked + // last, so the first check cannot wait for the first call. + stderr_is_process_stderr(); + main_thread_id = py::module_::import("threading") + .attr("main_thread")() + .attr("ident") + .cast(); + py::module_::import("atexit").attr("register")( + py::cpp_function([]() { write_deferred_logs_at_exit(); })); +#ifndef _WIN32 + // A child of fork() has no writer thread, the waiting lines are the + // parent's, and the thread that forked is its main thread. + pthread_atfork(nullptr, nullptr, [] { + deferred_logs_state() = new DeferredLogs(); + main_thread_id = PyThread_get_thread_ident(); + last_deferred_batch = 0; + }); +#endif #if defined(EXECUTORCH_PYBINDINGS_CUDA) && !defined(USE_ATEN_LIB) // Back executorch.runtime.cuda, which keeps the runtime free of PyTorch. @@ -5320,8 +5790,28 @@ PYBIND11_MODULE(EXECUTORCH_PYTHON_MODULE_NAME, m) { namespace { -// Forward runtime logs to Python without mutating std::cerr's process-wide -// stream buffer. +// Whether a line logged outside any binding call waits. Only a thread that +// holds the GIL can see where sys.stderr goes. Such a thread hands the line +// over when sys.stderr goes elsewhere, when fd 2 has no room, since the GIL it +// holds may be what a thread draining it needs, or when earlier lines still +// wait. Any other thread writes to fd 2. +bool line_outside_a_call_waits(size_t size) { + return holds_gil() && + (!sys_stderr_is_checked_fd_2() || deferred_lines_wait() || + !stderr_has_room(size, 0)); +} + +// Records the runtime's errors for exceptions and routes each line without +// calling Python: a backend can log while it holds its own lock, and a Python +// write there can hand the GIL to a thread that then waits for that lock. A +// line goes to the process's stderr through C stdio, since another module can +// redirect std::cerr into Python. When Python has replaced sys.stderr, as a +// notebook does, a line logged during a binding call waits for the call to +// return, and a line logged outside any call by a thread that holds the GIL +// waits for Python's main thread, if it logged the line, or for the writer +// thread. Such lines also wait once fd 2 is a pipe or socket that stays full. +// Even then a fatal line goes out at once while fd 2 has room, with the lines +// its call kept, since an abort usually follows. void emit_log_message( ET_UNUSED et_timestamp_t timestamp, et_pal_log_level_t level, @@ -5332,7 +5822,7 @@ void emit_log_message( size_t length) { const std::string formatted_message = "[" + std::string(filename) + ":" + std::to_string(line) + "] " + std::string(message, length) + "\n"; - if (level == kError && runtime_error_scopes != 0) { + if (level == kError && pending_logs != nullptr) { // Keeps the first lines, which usually name the cause. const char separator[] = "; "; const size_t prefix = runtime_errors_size == 0 ? 0 : sizeof(separator) - 1; @@ -5345,17 +5835,34 @@ void emit_log_message( runtime_errors_size += prefix + copied; } } - if (Py_IsInitialized() && PyGILState_Check()) { - try { - py::gil_scoped_acquire gil; - py::module_::import("sys").attr("stderr").attr("write")( - formatted_message); - return; - } catch (py::error_already_set& error) { - error.clear(); + if (pending_logs == nullptr) { + const size_t size = formatted_message.size(); + if (level == kFatal + ? holds_gil() && !stderr_has_room(size, kStderrRoomWaitMs) + : line_outside_a_call_waits(size)) { + defer_lines(formatted_message); + } else { + write_to_stderr(formatted_message.data(), size); } + return; + } + // While the writer thread writes to fd 2, it can take the room seen here + // before this write. + if (pending_logs->direct && !deferred_lines_wait() && + stderr_has_room(formatted_message.size(), kStderrRoomWaitMs)) { + write_to_stderr(formatted_message.data(), formatted_message.size()); + return; + } + // The rest of the call's lines wait for it to return. Then they go through + // Python, which lets go of the GIL while it waits for room. A daemon thread + // leaves them to the writer thread. + pending_logs->direct = false; + keep_line(*pending_logs, formatted_message); + if (level == kFatal) { + // An abort usually follows, and the waiting lines tell what led to it. + note_left_out_bytes(*pending_logs); + write_kept_lines_while_room(*pending_logs); } - std::cerr << formatted_message << std::flush; } runtime::PalImpl build_pal() { diff --git a/extension/pybindings/test/BUCK b/extension/pybindings/test/BUCK index a06a66c4a29..0602bfc2de2 100644 --- a/extension/pybindings/test/BUCK +++ b/extension/pybindings/test/BUCK @@ -40,6 +40,7 @@ fbcode_target( preload_deps = ["//executorch/kernels/quantized:aot_lib"], deps = [ ":make_test", + "//executorch/backends/xnnpack/partition:xnnpack_partitioner", "//executorch/exir:schema", "//executorch/exir/backend/test:device_util", "//executorch/extension/pybindings:portable_lib", @@ -76,6 +77,7 @@ fbcode_target( srcs = ["test_pybindings.py"], deps = [ ":make_test", + "//executorch/backends/xnnpack/partition:xnnpack_partitioner", "//executorch/exir:schema", "//executorch/exir/backend/test:device_util", ], diff --git a/extension/pybindings/test/test_pybindings.py b/extension/pybindings/test/test_pybindings.py index 1b661995501..baf5ac312c6 100644 --- a/extension/pybindings/test/test_pybindings.py +++ b/extension/pybindings/test/test_pybindings.py @@ -8,10 +8,13 @@ import ctypes import gc +import json import os +import select import subprocess import sys import tempfile +import time import unittest import weakref from io import StringIO @@ -52,6 +55,22 @@ def __dlpack_device__(self): return self.array.__dlpack_device__() +def write_llm_tokenizer(directory: str) -> str: + """Writes a tokenizer that the LLM runner loads without printing anything, + so that its only lines come through the runtime, outside any call of these + bindings, while it finds that the model next to it is missing.""" + path = os.path.join(directory, "tokenizer.json") + with open(path, "w") as file: + json.dump( + { + "added_tokens": [], + "model": {"type": "BPE", "vocab": {"a": 0, "b": 1}, "merges": []}, + }, + file, + ) + return path + + class PybindingsTest(unittest.TestCase): def setUp(self): # Will test both portable and aten @@ -377,6 +396,792 @@ def run(executor, separate_calls, locked=True): ) self.assertIn("all calls finished", completed.stdout) + def test_backend_logs_do_not_deadlock_other_threads(self): + if self.kernel_mode != "portable": + self.skipTest("the script below uses the portable bindings") + if "XnnpackBackend" not in self.runtime._get_registered_backend_names(): + self.skipTest("needs the XNNPACK backend") + + script = """ +import os +import sys +import threading +import time +from concurrent.futures import ThreadPoolExecutor + +import torch +from executorch.backends.xnnpack.partition.xnnpack_partitioner import ( + XnnpackPartitioner, +) +from executorch.exir import to_edge_transform_and_lower +from executorch.exir._serialize._program import PTEFile, serialize_pte_binary +from executorch.exir.schema import ( + BackendDelegateDataReference, + BackendDelegateInlineData, + DataLocation, +) +from executorch.extension.pybindings import portable_lib + + +class Add(torch.nn.Module): + def forward(self, x, y): + return x + y + + +inputs = [torch.ones(2, 2), torch.ones(2, 2)] +exported = torch.export.export(Add(), tuple(inputs)) + + +def broken_program(methods): + # XNNPACK logs a broken payload while it holds its workspace lock, which + # the global sharing mode shares between all methods. + schema = to_edge_transform_and_lower( + methods, partitioner=[XnnpackPartitioner()] + ).to_executorch().executorch_program + schema.execution_plan[0].delegates[0].processed = BackendDelegateDataReference( + location=DataLocation.INLINE, index=len(schema.backend_delegate_data) + ) + schema.backend_delegate_data.append(BackendDelegateInlineData(data=b"\\x01" * 64)) + return bytes(serialize_pte_binary(PTEFile(schema))) + + +data = to_edge_transform_and_lower( + exported, partitioner=[XnnpackPartitioner()] +).to_executorch().buffer +if sys.argv[1] == "LLM runner": + try: + from executorch.extension.llm.runner import TextLLMRunner + except (ImportError, RuntimeError): + # The package raises RuntimeError when its extension is not built. + print("no LLM runner") + sys.exit() + # The LLM runner reads this method as it starts, outside any call of these + # bindings, on a thread that holds the GIL. + model = os.path.join(os.path.dirname(sys.argv[2]), "broken.pte") + with open(model, "wb") as file: + file.write(broken_program({"get_max_seq_len": exported})) +else: + broken = portable_lib._load_program_from_buffer(broken_program(exported)) +portable_lib._set_option("XnnpackBackend", {"workspace_sharing_mode": 2}) +method = portable_lib._load_program_from_buffer(data).load_method("forward") + +running = threading.Event() +done = threading.Event() + + +def run(): + while not done.is_set(): + method(inputs) + running.set() + + +class SlowStderr: + # A real write releases the GIL for its system call. Keep that window open + # so the other thread surely runs inside it. + def write(self, text): + time.sleep(0.5) + return len(text) + + def flush(self): + pass + + def fileno(self): + return 2 + + +message = "" +process_stderr = sys.__stderr__ +with ThreadPoolExecutor(max_workers=1) as pool: + worker = pool.submit(run) + while not running.wait(timeout=0.1): + if worker.done(): + worker.result() + sys.stderr = SlowStderr() + if sys.argv[1] == "process": + # Stands in for the stream Python opened on fd 2, the default case. + sys.__stderr__ = sys.stderr + try: + if sys.argv[1] == "LLM runner": + TextLLMRunner(model, sys.argv[2]) + else: + broken.load_method("forward") + except RuntimeError as error: + message = str(error) + finally: + sys.stderr = sys.__stderr__ = process_stderr + done.set() + worker.result() +# XNNPACK logged this while it held its workspace lock. +assert sys.argv[1] == "LLM runner" or "XNNCompiler::compileModel failed" in message, message +print("no deadlock") +""" + with tempfile.TemporaryDirectory() as directory: + tokenizer = write_llm_tokenizer(directory) + for mode in ("replaced", "process", "LLM runner"): + completed = subprocess.run( + [sys.executable, "-c", script, mode, tokenizer], + capture_output=True, + text=True, + timeout=120, + ) + self.assertEqual( + completed.returncode, + 0, + f"{mode}:\nstdout:\n{completed.stdout}\nstderr:\n{completed.stderr}", + ) + if "no LLM runner" not in completed.stdout: + self.assertIn("no deadlock", completed.stdout) + + def test_daemon_threads_that_log_do_not_crash_at_exit(self): + if self.kernel_mode != "portable": + self.skipTest("the script below uses the portable bindings") + if not sys.platform.startswith("linux") or sys.version_info >= (3, 13, 8): + self.skipTest("the crash needs Linux and Python before 3.13.8") + + script = """ +import atexit +import os +import sys +import threading +import time + +# Registered before the import, so it runs after the bindings' own exit hook, +# and the daemon threads log while it sleeps. +atexit.register(time.sleep, 0.1) + +from executorch.extension.pybindings import portable_lib + + +# The daemon threads below. Python calls any thread it did not start a daemon. +spinners = set() + + +class Sink: + daemon_writes = 0 + text = [] + + def write(self, text): + if threading.get_ident() in spinners: + Sink.daemon_writes += 1 + Sink.text.append(text) + return len(text) + + def flush(self): + pass + + +def spin(): + spinners.add(threading.get_ident()) + while True: + try: + portable_lib._load_program_from_buffer(b"x" * 128) + except RuntimeError: + pass + + +if "full" in sys.argv[1]: + # Nobody reads this pipe, so the lines that find it full are dropped. + read_end, write_end = os.pipe() + os.dup2(write_end, 2) + os.set_blocking(2, False) + try: + while True: + os.write(2, b"x" * 4096) + except BlockingIOError: + pass + os.set_blocking(2, True) +if "replaced" in sys.argv[1]: + sys.stderr = Sink() +for _ in range(2): + threading.Thread(target=spin, daemon=True).start() +time.sleep(0.3) +print("writes from daemon threads:", Sink.daemon_writes) +print("lines through Python:", "".join(Sink.text).count("Program identifier")) +""" + for mode in ("replaced", "full", "full and replaced"): + completed = subprocess.run( + [sys.executable, "-c", script, mode], + capture_output=True, + text=True, + timeout=120, + ) + self.assertEqual( + completed.returncode, 0, f"{mode}: {completed.stderr[-2000:]}" + ) + self.assertIn("writes from daemon threads: 0", completed.stdout) + if "replaced" in mode: + # Another thread wrote them, as a notebook shows only Python's + # stderr. + self.assertNotIn("lines through Python: 0", completed.stdout) + if mode == "replaced": + # The lines logged after the bindings' exit hook go to fd 2. + self.assertIn("Program identifier", completed.stderr) + + def test_runtime_logs_outside_a_call_reach_sys_stderr(self): + if self.kernel_mode != "portable": + self.skipTest("the script below uses the portable bindings") + + script = """ +import contextlib +import io +import os +import sys +import threading +import time + +captured = io.StringIO() +if sys.argv[1] == "replaced before import": + # As in a notebook, which replaces sys.stderr before any import. + sys.stderr = captured +from executorch.extension.pybindings import portable_lib + +try: + from executorch.extension.llm.runner import TextLLMRunner +except (ImportError, RuntimeError): + # The package raises RuntimeError when its extension is not built. + print("no LLM runner") + sys.exit() +missing_program = os.path.join(os.path.dirname(sys.argv[2]), "missing.pte") + +if sys.argv[1] == "replaced, another thread": + sys.stderr = captured + + def log_twice(): + try: + TextLLMRunner(missing_program, sys.argv[2]) + except RuntimeError: + pass + try: + portable_lib._load_program_from_buffer(b"x" * 128) + except RuntimeError: + pass + + worker = threading.Thread(target=log_twice) + worker.start() + worker.join() + # A thread of the bindings writes the LLM runner's lines a moment later, and + # the call's line must follow them. + deadline = time.monotonic() + 10 + while ( + "Program identifier" not in captured.getvalue() + and time.monotonic() < deadline + ): + time.sleep(0.01) + sys.stderr = sys.__stderr__ + text = captured.getvalue() + first = text.find("[llm_runner_helper.cpp:") + print("in order:", 0 <= first < text.find("Program identifier")) + sys.exit() + +with ( + contextlib.redirect_stderr(captured) + if sys.argv[1] == "replaced after import" + else contextlib.nullcontext() +): + try: + TextLLMRunner(missing_program, sys.argv[2]) + except RuntimeError: + if sys.argv[1] == "process stderr": + # Ends the process before Python could write lines that waited. + os._exit(0) + # Python's main thread writes the lines it logged before its next bytecode, + # so they reach a redirect that ends right after the call. +sys.stderr = sys.__stderr__ +print("lines through Python:", captured.getvalue().count("[llm_runner_helper.cpp:")) +""" + with tempfile.TemporaryDirectory() as directory: + tokenizer = write_llm_tokenizer(directory) + for mode in ( + "replaced before import", + "replaced after import", + "replaced, another thread", + "process stderr", + ): + completed = subprocess.run( + [sys.executable, "-c", script, mode, tokenizer], + capture_output=True, + text=True, + timeout=120, + ) + self.assertEqual( + completed.returncode, 0, f"{mode}: {completed.stderr[-2000:]}" + ) + if "no LLM runner" in completed.stdout: + self.skipTest("the LLM runner bindings are not built") + if mode == "process stderr": + self.assertIn("[llm_runner_helper.cpp:", completed.stderr) + elif mode == "replaced, another thread": + self.assertIn("in order: True", completed.stdout) + else: + self.assertIn("lines through Python:", completed.stdout, mode) + self.assertNotIn("lines through Python: 0", completed.stdout, mode) + + def test_runtime_logs_when_writing_to_stderr_fails(self): + if self.kernel_mode != "portable": + self.skipTest("the script below uses the portable bindings") + + script = """ +import signal +import sys + +from executorch.extension.pybindings import portable_lib + +# Python installs no SIGINT handler when SIGINT starts ignored, as in a +# background job, and then a Ctrl-C cannot be raised again. +signal.signal(signal.SIGINT, signal.default_int_handler) + + +class FailingStderr: + def __init__(self, error): + self.error = error + + def write(self, text): + raise self.error + + def flush(self): + pass + + +def load_broken_program(): + try: + portable_lib._load_program_from_buffer(b"x" * 128) + except RuntimeError: + pass + + +# A call made while sys.stderr is the process's own goes first, so that +# replacing sys.stderr afterwards must still be noticed. +load_broken_program() + +# A write that fails sends the lines to the process's stderr instead. +sys.stderr = FailingStderr(ValueError("closed")) +try: + load_broken_program() +finally: + sys.stderr = sys.__stderr__ + +# An interrupt raised in the write still reaches the caller. +interrupted = False +sys.stderr = FailingStderr(KeyboardInterrupt()) +try: + load_broken_program() + for _ in range(1000): + pass +except KeyboardInterrupt: + interrupted = True +finally: + sys.stderr = sys.__stderr__ +assert interrupted +print("done") +""" + completed = subprocess.run( + [sys.executable, "-c", script], + capture_output=True, + text=True, + timeout=120, + ) + + self.assertEqual( + completed.returncode, + 0, + f"stdout:\n{completed.stdout}\nstderr:\n{completed.stderr}", + ) + self.assertIn("done", completed.stdout) + # The first load writes straight to the process's stderr. Neither + # write showed the other loads' lines, so they reach it too. + self.assertEqual( + completed.stderr.count("Program identifier 'xxxx'"), 3, completed.stderr + ) + + def test_fatal_runtime_log_reaches_process_stderr(self): + if self.kernel_mode != "portable": + self.skipTest("the script below uses the portable bindings") + if sys.platform == "win32": + self.skipTest("an abort can start Windows Error Reporting") + + exported_program, _ = create_program(ModuleAddSingleInput()) + script = """ +import io +import sys + +import torch +from executorch.extension.pybindings import portable_lib + +# Logging the first output to a debug buffer this small fails, and the runtime +# reports that with a fatal line before it aborts. +module = portable_lib._load_for_executorch( + sys.argv[1], enable_etdump=True, debug_buffer_size=8 +) +sys.stderr = io.StringIO() +module.forward([torch.ones(2, 2)]) +""" + with tempfile.TemporaryDirectory() as directory: + path = os.path.join(directory, "add.pte") + with open(path, "wb") as file: + file.write(exported_program.buffer) + completed = subprocess.run( + [sys.executable, "-c", script, path], + capture_output=True, + text=True, + timeout=120, + ) + + self.assertNotEqual(completed.returncode, 0, completed.stderr) + # The abort skips the end of the call, where kept lines go to the + # replaced sys.stderr, so the fatal line went out at once, after the + # line the call kept before it. + kept = completed.stderr.find("Ran out of space to store intermediate outputs") + fatal = completed.stderr.find("failed to write tensor to debug buffer") + self.assertNotEqual(kept, -1, completed.stderr) + self.assertGreater(fatal, kept, completed.stderr) + + def test_runtime_logs_reach_process_stderr_at_once(self): + if self.kernel_mode != "portable": + self.skipTest("the script below uses the portable bindings") + if sys.platform == "win32": + self.skipTest("uses a pseudo-terminal") + + exported_program, _ = create_program(ModuleAddSingleInput()) + script = """ +import os +import sys + +import numpy as np +from executorch.extension.pybindings import portable_lib + + +class DiesDuringTheCall: + # Converting this input fills the terminal, logs from inside the call, and + # ends the process before the call returns. + def __dlpack__(self, stream=None): + os.set_blocking(2, False) + try: + while True: + os.write(2, b"." * 64) + except BlockingIOError: + pass + os.set_blocking(2, True) + print("filled", flush=True) + try: + portable_lib._load_program_from_buffer(b"x" * 128) + except RuntimeError: + pass + os.write(2, b"still in the call\\n") + os._exit(3) + + def __dlpack_device__(self): + return np.ones((2, 2), dtype=np.float32).__dlpack_device__() + + +with open(sys.argv[1], "rb") as file: + module = portable_lib._load_for_executorch_from_buffer(file.read()) +module(DiesDuringTheCall()) +""" + try: + terminal, child_end = os.openpty() + except OSError as error: + self.skipTest(f"no pseudo-terminal: {error}") + with tempfile.TemporaryDirectory() as directory: + path = os.path.join(directory, "add.pte") + with open(path, "wb") as file: + file.write(exported_program.buffer) + child = subprocess.Popen( + [sys.executable, "-c", script, path], + stdout=subprocess.PIPE, + stderr=child_end, + text=True, + ) + os.close(child_end) + # Like a busy terminal, start reading a little after the child has + # filled it and logged. + filled = ( + child.stdout.readline() + if select.select([child.stdout], [], [], 120)[0] + else "" + ) + time.sleep(0.5) + output = b"" + while select.select([terminal], [], [], 120)[0]: + try: + chunk = os.read(terminal, 65536) + except OSError: # Linux reports the closed other end as EIO. + break + if not chunk: + break + output += chunk + child.kill() + child.wait() + child.stdout.close() + os.close(terminal) + + text = output.decode(errors="replace") + self.assertEqual(filled, "filled\n", text) + self.assertEqual(child.returncode, 3, text) + # sys.stderr is the process's own stderr there, so the line went out as + # it was logged rather than with the call that never returned. + self.assertIn("Program identifier 'xxxx'", text) + self.assertLess( + text.index("Program identifier 'xxxx'"), text.index("still in the call") + ) + + def test_full_stderr_drained_by_a_thread_does_not_block_a_call(self): + if self.kernel_mode != "portable": + self.skipTest("the script below uses the portable bindings") + if sys.platform == "win32": + self.skipTest("uses POSIX pipes and sockets as fd 2") + + script = """ +import os +import socket +import sys +import threading +import time + +from executorch.extension.pybindings import portable_lib + +if "LLM runner" in sys.argv[1]: + try: + from executorch.extension.llm.runner import TextLLMRunner + except (ImportError, RuntimeError): + # The package raises RuntimeError when its extension is not built. + print("no LLM runner") + sys.exit() +missing_program = os.path.join(os.path.dirname(sys.argv[3]), "missing.pte") +if sys.argv[2] == "socket": + read_socket, write_socket = socket.socketpair() + read_end, write_end = read_socket.fileno(), write_socket.fileno() +else: + read_end, write_end = os.pipe() +# Fill it, so that a write that waits for room needs the reader below. +os.set_blocking(write_end, False) +for size in (4096, 512, 64, 1): + try: + while True: + os.write(write_end, b"x" * size) + except BlockingIOError: + pass +os.set_blocking(write_end, True) +saved_stderr = os.dup(2) +os.dup2(write_end, 2) +drained = bytearray() + + +def drain(): + # Like a capture thread, this needs the GIL after every read. It starts + # after the call below has begun. + time.sleep(0.2) + while True: + drained.extend(os.read(read_end, 4096)) + + +threading.Thread(target=drain, daemon=True).start() +try: + if sys.argv[1] == "LLM runner": + # It holds the GIL and logs outside any call of these bindings. + try: + TextLLMRunner(missing_program, sys.argv[3]) + except RuntimeError: + pass + expected = b"[llm_runner_helper.cpp:" + elif sys.argv[1] == "LLM runner and binding calls": + # The LLM runner's lines wait for a thread of the bindings, which writes + # them to fd 2 while the calls check it for room. + stop = threading.Event() + + def build_runners(): + while not stop.is_set(): + try: + TextLLMRunner(missing_program, sys.argv[3]) + except RuntimeError: + pass + + builder = threading.Thread(target=build_runners) + builder.start() + end = time.monotonic() + 10 + while time.monotonic() < end: + try: + portable_lib._load_program_from_buffer(b"x" * 128) + except RuntimeError: + pass + stop.set() + builder.join() + expected = b"Program identifier" + else: + try: + portable_lib._load_program_from_buffer(b"x" * 128) + except RuntimeError: + pass + expected = b"Program identifier" + while expected not in drained: + time.sleep(0.01) +finally: + # Lets an error here reach the test's stderr rather than fd 2. + os.dup2(saved_stderr, 2) +print("done") +""" + with tempfile.TemporaryDirectory() as directory: + tokenizer = write_llm_tokenizer(directory) + for kind in ("pipe", "socket"): + for mode in ( + "binding call", + "LLM runner", + "LLM runner and binding calls", + ): + completed = subprocess.run( + [sys.executable, "-c", script, mode, kind, tokenizer], + capture_output=True, + text=True, + timeout=120, + ) + self.assertEqual( + completed.returncode, + 0, + f"{mode}, {kind}:\nstdout:\n{completed.stdout}\n" + f"stderr:\n{completed.stderr}", + ) + if "no LLM runner" not in completed.stdout: + self.assertIn("done", completed.stdout, f"{mode}, {kind}") + + def test_deferred_runtime_logs_do_not_nest_in_a_stderr_write(self): + if self.kernel_mode != "portable": + self.skipTest("the script below uses the portable bindings") + + script = """ +import os +import queue +import sys +import threading +import time + +from executorch.extension.pybindings import portable_lib + +if sys.argv[1] == "LLM runner": + try: + from executorch.extension.llm.runner import TextLLMRunner + except (ImportError, RuntimeError): + # The package raises RuntimeError when its extension is not built. + print("no LLM runner") + sys.exit() + expected = "[llm_runner_helper.cpp:" +else: + expected = "Program identifier" + + +class QueuedStderr: + # Like a logging queue, its write runs Python code while it holds a lock. + def __init__(self): + self.texts = queue.Queue() + + def write(self, text): + self.texts.put(text) + return len(text) + + def flush(self): + pass + + def text(self): + return "".join(self.texts.queue) + + +def log(): + while not done.is_set(): + if sys.argv[1] == "LLM runner": + # It holds the GIL and logs outside any call of these bindings. + try: + TextLLMRunner( + os.path.join(os.path.dirname(sys.argv[2]), "missing.pte"), + sys.argv[2], + ) + except RuntimeError: + pass + else: + try: + portable_lib._load_program_from_buffer(b"x" * 128) + except RuntimeError: + pass + started.set() + + +done, started = threading.Event(), threading.Event() +stream = QueuedStderr() +sys.stderr = stream +thread = threading.Thread(target=log, daemon=sys.argv[1] == "daemon thread") +thread.start() +started.wait(60) +# The other thread's lines must never be written from inside these writes. +deadline = time.monotonic() + 2 +while time.monotonic() < deadline: + print("main thread", file=sys.stderr) +done.set() +thread.join() +deadline = time.monotonic() + 10 +while expected not in stream.text() and time.monotonic() < deadline: + time.sleep(0.01) +sys.stderr = sys.__stderr__ +print("lines through Python:", stream.text().count(expected)) +""" + with tempfile.TemporaryDirectory() as directory: + tokenizer = write_llm_tokenizer(directory) + for mode in ("LLM runner", "daemon thread"): + completed = subprocess.run( + [sys.executable, "-c", script, mode, tokenizer], + capture_output=True, + text=True, + timeout=120, + ) + self.assertEqual( + completed.returncode, 0, f"{mode}: {completed.stderr[-2000:]}" + ) + if "no LLM runner" not in completed.stdout: + self.assertIn("lines through Python:", completed.stdout, mode) + self.assertNotIn("lines through Python: 0", completed.stdout, mode) + + def test_long_call_keeps_its_newest_runtime_logs(self): + if self.kernel_mode != "portable": + self.skipTest("only the portable bindings take DLPack inputs") + + exported_program, _ = create_program(ModuleAddSingleInput()) + module = self.load_fn(exported_program.buffer) + runtime = self.runtime + values = np.ones((2, 2), dtype=np.float32) + + class LogsDuringTheCall: + # Converting this input logs far more than one call keeps, ending + # with a line that is not valid UTF-8. + def __dlpack__(self, stream=None): + for _ in range(20000): + try: + runtime._load_program_from_buffer(b"x" * 128) + except RuntimeError: + pass + try: + runtime._load_program_from_buffer(b"\xff" * 128) + except Exception: # Its error message is not valid UTF-8 either. + pass + return values.__dlpack__(stream=stream) + + def __dlpack_device__(self): + return values.__dlpack_device__() + + captured = StringIO() + saved_stderr = sys.stderr + sys.stderr = captured + try: + module(LogsDuringTheCall()) + finally: + sys.stderr = saved_stderr + + text = captured.getvalue() + lines = text.splitlines() + self.assertRegex( + lines[0], + r"^\[\d+ earlier bytes of the runtime log were left out\]$", + ) + kept = len(text) - len(lines[0]) - 1 + self.assertGreater(kept, 512 * 1024 - 1024) + self.assertLessEqual(kept, 1024 * 1024) + self.assertIn("Program identifier '\\xff\\xff\\xff\\xff'", lines[-1]) + def test_bfloat16_output_preserves_dtype(self): class ModuleAddBFloat16(ModuleAdd): def get_inputs(self): From 7a5fb48bbcd7a45a44e30a568d2fd3b59332a9bc Mon Sep 17 00:00:00 2001 From: Anthony Shoumikhin Date: Mon, 2 Nov 2026 04:06:03 -0800 Subject: [PATCH 2/2] Python runtime: raise RuntimeError, not UnicodeDecodeError, for a corrupt program The bindings add the runtime's error lines to the exceptions they raise. Some of those lines carry raw bytes from the input, such as 4 bytes near the start of a file that is not a program, which the runtime prints as its identifier. The message itself is cut to fit 128 bytes, which can split a character of a path. Python rejects an exception message that is not valid UTF-8, so such a load raised UnicodeDecodeError instead of RuntimeError. The whole message is now decoded with backslashreplace, so those bytes show as escapes. The long-call test, which loads such a file, now expects RuntimeError, and a new test loads paths whose message is cut inside a character. --- extension/pybindings/pybindings.cpp | 12 ++++++++++-- extension/pybindings/test/test_pybindings.py | 9 ++++++++- 2 files changed, 18 insertions(+), 3 deletions(-) diff --git a/extension/pybindings/pybindings.cpp b/extension/pybindings/pybindings.cpp index 4f4efa26cf4..5ed76f3f626 100644 --- a/extension/pybindings/pybindings.cpp +++ b/extension/pybindings/pybindings.cpp @@ -80,7 +80,7 @@ #endif /// Throws a runtime_error with the provided message if `error` is not `Ok`, -/// followed by the errors the runtime logged during this call. +/// followed by the errors the runtime logged during this call. Needs the GIL. #define THROW_IF_ERROR(error, message, ...) \ ({ \ if ((error) != Error::Ok) { \ @@ -588,7 +588,15 @@ std::string with_runtime_errors(const char* message) { result.append(": ").append(runtime_errors, runtime_errors_size); runtime_errors_size = 0; } - return result; + // Python rejects an exception message that is not valid UTF-8. The message is + // cut at a fixed length, which can split a character of a path, and the + // runtime can log raw bytes, such as a file's identifier. + const auto text = py::reinterpret_steal( + PyUnicode_DecodeUTF8(result.data(), result.size(), "backslashreplace")); + if (!text) { + throw py::error_already_set(); + } + return text.cast(); } // Starts every binding call with no runtime errors recorded, and gives an diff --git a/extension/pybindings/test/test_pybindings.py b/extension/pybindings/test/test_pybindings.py index baf5ac312c6..b212abafdc8 100644 --- a/extension/pybindings/test/test_pybindings.py +++ b/extension/pybindings/test/test_pybindings.py @@ -1156,7 +1156,7 @@ def __dlpack__(self, stream=None): pass try: runtime._load_program_from_buffer(b"\xff" * 128) - except Exception: # Its error message is not valid UTF-8 either. + except RuntimeError: pass return values.__dlpack__(stream=stream) @@ -1182,6 +1182,13 @@ def __dlpack_device__(self): self.assertLessEqual(kept, 1024 * 1024) self.assertIn("Program identifier '\\xff\\xff\\xff\\xff'", lines[-1]) + def test_error_message_cut_in_a_character_raises_runtime_error(self): + # An error message is cut at a fixed length. With these paths, the cut + # falls inside a character for at least one of them. + for prefix in ("", "a", "ab"): + with self.assertRaises(RuntimeError): + self.runtime._load_for_executorch(prefix + "\u4e2d" * 60 + ".pte") + def test_bfloat16_output_preserves_dtype(self): class ModuleAddBFloat16(ModuleAdd): def get_inputs(self):