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 f8038474c1b..4e23abe0e20 100644 --- a/extension/pybindings/README.md +++ b/extension/pybindings/README.md @@ -235,4 +235,29 @@ 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, where a +Windows console shows text that is not ASCII in its own code page. If Python +has replaced `sys.stderr`, for example with pytest's `capsys`, +`contextlib.redirect_stderr` or in Jupyter, the lines a binding call logs wait +until the call returns and then go to `sys.stderr`. They 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, +lines also wait once the process's stderr is a pipe that stays full for a tenth +of a second, so a call cannot hang on a pipe that a Python thread in the same +process empties, as under `wurlitzer.pipes()`. On macOS, a pipe that has not +grown past 16 KiB counts as full once it holds 512 bytes, because macOS reports +room in such a pipe that it may not have. A call keeps up to 64 KiB of waiting +lines; past that it keeps about the newest 32 to 64 KiB and says how many +earlier bytes it left out. A fatal line and the lines waiting before it go +straight to the process's stderr while it has room, since an abort usually +follows. Any other crash or hang of the process loses the lines still waiting. + +On Linux with Python before 3.13.8, a daemon thread never writes its lines +through Python, because at exit Python can stop such a thread in the middle of +that write, which crashes the process. Its lines go straight to the process's +stderr as far as there is room. + +Lines logged outside a call always go straight to the process's stderr: on +another thread, while an object is destroyed, or by other ExecuTorch modules +such as the LLM runner. In a Jupyter kernel on Linux or macOS, enough such +lines while a thread holds the GIL can fill the kernel's pipe and hang it. diff --git a/extension/pybindings/pybindings.cpp b/extension/pybindings/pybindings.cpp index 9889c9622a8..a6cd773aec0 100644 --- a/extension/pybindings/pybindings.cpp +++ b/extension/pybindings/pybindings.cpp @@ -12,7 +12,6 @@ #include #include #include -#include #include #include #include @@ -21,6 +20,11 @@ #include #include #include +#ifndef _WIN32 +#include +#include +#include +#endif #include #include @@ -139,9 +143,184 @@ 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 = 64 * 1024; +// Null outside a binding call, so lines logged outside any, such as on worker +// threads, are not kept and never linger. A plain pointer for the same reason +// as above. +thread_local PendingLogs* pending_logs = nullptr; + +void write_to_stderr(const char* data, size_t size) { + std::fwrite(data, 1, size, stderr); + std::fflush(stderr); +} + +// Whether sys.stderr is still the stream Python opened on file descriptor 2. +// Runs 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, which the reference keeps alive. +bool stderr_is_process_stderr() noexcept { + static PyObject* checked = nullptr; + static bool checked_on_fd_2 = false; + PyObject* const current = PySys_GetObject("stderr"); + if (current != nullptr && current == checked) { + return checked_on_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_on_fd_2 = on_fd_2; + Py_XSETREF(checked, current); + return on_fd_2; +} + +// 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 pipe. 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 without a wait +// that may never end. That happens with a full pipe that a Python thread in +// this process drains, as wurlitzer.pipes() sets up. Only a pipe is checked: a +// terminal's reader is another process, and macOS calls a terminal full long +// before a write to it would wait. +bool stderr_has_room(ET_UNUSED size_t size) { +#ifdef _WIN32 + return true; +#else + struct stat info; + if (fstat(2, &info) != 0 || !S_ISFIFO(info.st_mode)) { + 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. + if (info.st_blksize <= 16 * 1024) { + for (int waited_ms = 0; static_cast(info.st_size) + size > PIPE_BUF; + ++waited_ms) { + if (waited_ms == kStderrRoomWaitMs || usleep(1000) != 0 || + fstat(2, &info) != 0) { + return false; + } + } + return true; + } +#endif + pollfd stderr_fd{2, POLLOUT, 0}; + return poll(&stderr_fd, 1, kStderrRoomWaitMs) > 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 this call's runtime log were left out]\n"); + logs.dropped = 0; + } catch (const std::bad_alloc&) { + } +} + +// Writes the kept lines to fd 2, oldest first, while it has room, and keeps the +// rest. Whole lines only, since a pipe that a Python thread drains stays full +// while the GIL is held. +void write_kept_lines_while_room(PendingLogs& logs) noexcept { + size_t begin = 0; + while (begin < logs.lines.size()) { + const size_t end = logs.lines.find('\n', begin) + 1; + if (!stderr_has_room(end - begin)) { + break; + } + write_to_stderr(logs.lines.data() + begin, end - begin); + begin = end; + } + logs.lines.erase(0, begin); +} + +// Runs as the outermost binding call returns, 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; + } + if (logs.daemon) { + write_kept_lines_while_room(logs); + 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()); +} std::string with_runtime_errors(const char* message) { std::string result(message); @@ -149,24 +328,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 = logs_->daemon || 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,10 +5089,12 @@ 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. + // Its destructor may write to Python, so it needs the GIL and never follows a + // py::gil_scoped_release guard. auto call_guard = py::call_guard(); #if defined(EXECUTORCH_PYBINDINGS_CUDA) && !defined(USE_ATEN_LIB) @@ -5320,8 +5520,15 @@ PYBIND11_MODULE(EXECUTORCH_PYTHON_MODULE_NAME, m) { namespace { -// Forward runtime logs to Python without mutating std::cerr's process-wide -// stream buffer. +// 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. During a binding call it waits for the call +// to return instead when Python has replaced sys.stderr, except on a daemon +// thread, or once fd 2 is a pipe that stays full. Even then a fatal line and +// the lines waiting before it go out at once while fd 2 has room, since an +// abort usually follows. void emit_log_message( ET_UNUSED et_timestamp_t timestamp, et_pal_log_level_t level, @@ -5332,7 +5539,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 +5552,21 @@ 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 || + (pending_logs->direct && stderr_has_room(formatted_message.size()))) { + 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, or on a daemon + // thread to fd 2 as far as it has room. + 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..c5315887ae1 100644 --- a/extension/pybindings/test/test_pybindings.py +++ b/extension/pybindings/test/test_pybindings.py @@ -9,9 +9,11 @@ import ctypes import gc import os +import select import subprocess import sys import tempfile +import time import unittest import weakref from io import StringIO @@ -377,6 +379,508 @@ 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 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)] +program = to_edge_transform_and_lower( + torch.export.export(Add(), tuple(inputs)), partitioner=[XnnpackPartitioner()] +).to_executorch() +data = program.buffer +# XNNPACK logs a broken payload while it holds its workspace lock, which the +# global sharing mode shares between all methods. +schema = program.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)) +broken = portable_lib._load_program_from_buffer( + bytes(serialize_pte_binary(PTEFile(schema))) +) +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: + 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 "XNNCompiler::compileModel failed" in message, message +print("no deadlock") +""" + for mode in ("replaced", "process"): + completed = subprocess.run( + [sys.executable, "-c", script, mode], + capture_output=True, + text=True, + timeout=120, + ) + self.assertEqual( + completed.returncode, + 0, + f"{mode}:\nstdout:\n{completed.stdout}\nstderr:\n{completed.stderr}", + ) + 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 os +import sys +import threading +import time + +from executorch.extension.pybindings import portable_lib + + +class Sink: + writes = 0 + + def write(self, text): + Sink.writes += 1 + return len(text) + + def flush(self): + pass + + +def spin(): + while True: + try: + portable_lib._load_program_from_buffer(b"x" * 128) + except RuntimeError: + pass + + +if sys.argv[1] == "full": + # Nobody reads this pipe, so the lines that find it full are kept. + 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) +sys.stderr = Sink() +for _ in range(2): + threading.Thread(target=spin, daemon=True).start() +time.sleep(0.3) +print("writes through Python:", Sink.writes) +""" + for mode in ("replaced", "full"): + 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 through Python: 0", completed.stdout) + if mode == "replaced": + self.assertIn("Program identifier", completed.stderr) + + 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_pipe_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 fcntl and POSIX pipes") + + script = """ +import fcntl +import os +import sys +import threading +import time + +from executorch.extension.pybindings import portable_lib + +read_end, write_end = os.pipe() +# Fill the pipe, so that a write that waits for room needs the reader below. +flags = fcntl.fcntl(write_end, fcntl.F_GETFL) +fcntl.fcntl(write_end, fcntl.F_SETFL, flags | os.O_NONBLOCK) +for size in (4096, 512, 64, 1): + try: + while True: + os.write(write_end, b"x" * size) + except BlockingIOError: + pass +fcntl.fcntl(write_end, fcntl.F_SETFL, flags) +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, 1000)) + + +threading.Thread(target=drain, daemon=True).start() +try: + try: + portable_lib._load_program_from_buffer(b"x" * 128) + except RuntimeError: + pass + while b"Program identifier" not in drained: + time.sleep(0.01) +finally: + # Lets an error here reach the test's stderr rather than the pipe. + os.dup2(saved_stderr, 2) +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) + + 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(2000): + 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 this call's runtime log were left out\]$", + ) + kept = len(text) - len(lines[0]) - 1 + self.assertGreater(kept, 32 * 1024 - 1024) + self.assertLessEqual(kept, 64 * 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):