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..5ed76f3f626 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 @@ -70,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) { \ @@ -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); @@ -149,24 +588,43 @@ 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 -// 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 +5349,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 +5798,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 +5830,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 +5843,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..b212abafdc8 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,799 @@ 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 RuntimeError: + 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_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):