From 5f9e60339bf0768563dbbe23fac4fa5bb5af9309 Mon Sep 17 00:00:00 2001 From: Benoit Chesneau Date: Sat, 29 Aug 2026 17:39:55 +0200 Subject: [PATCH] Wait on callback pipes with poll, not select select() is undefined for a descriptor above FD_SETSIZE, so a VM with more than 1024 open files could not bring up a thread worker and every thread callback after that failed with "Failed to spawn thread handler". The ready-wait also releases the GIL, and the coordinator logs a failed ready signal instead of leaving Python to time out. --- CHANGELOG.md | 5 ++++ c_src/py_nif.h | 24 ++++++------------ c_src/py_thread_worker.c | 8 ++++-- src/py_thread_handler.erl | 16 ++++++++---- test/py_reentrant_SUITE.erl | 34 +++++++++++++++++++++++++ test/py_test_high_fds.py | 50 +++++++++++++++++++++++++++++++++++++ 6 files changed, 114 insertions(+), 23 deletions(-) create mode 100644 test/py_test_high_fds.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 437a85c..8983353 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -66,6 +66,11 @@ - `pthread_timedjoin_np` was called without `_GNU_SOURCE`, an implicit declaration on Linux that newer compilers reject. +- Callback pipes waited with `select()`, which is undefined for a file + descriptor above 1024: in a VM with many open files a thread callback + could time out with "Failed to spawn thread handler". The waits use + `poll()`, the handler ready-wait no longer holds the GIL, and + `py_thread_handler` logs a failed ready signal. ## 4.1.0 (2026-08-15) diff --git a/c_src/py_nif.h b/c_src/py_nif.h index 3b1fcf5..6832c40 100644 --- a/c_src/py_nif.h +++ b/c_src/py_nif.h @@ -64,7 +64,7 @@ #define NEED_DLOPEN_GLOBAL 1 #endif -#include +#include /** @} */ /* ============================================================================ @@ -1598,13 +1598,10 @@ static ssize_t read_with_timeout(int fd, void *buf, size_t count, int timeout_ms errno = ETIMEDOUT; return (ssize_t)got; } - struct timeval tv; - tv.tv_sec = remain_ms / 1000; - tv.tv_usec = (remain_ms % 1000) * 1000; - fd_set fds; - FD_ZERO(&fds); - FD_SET(fd, &fds); - int s = select(fd + 1, &fds, NULL, NULL, &tv); + /* poll, not select: select() is undefined for fd >= FD_SETSIZE + * (1024) and a VM with many open files gets pipe fds above it. */ + struct pollfd pfd = { .fd = fd, .events = POLLIN, .revents = 0 }; + int s = poll(&pfd, 1, (int)remain_ms); if (s < 0) { if (errno == EINTR) continue; return -1; @@ -1690,7 +1687,7 @@ typedef enum { * @brief Write exactly @p count bytes to a (typically non-blocking) fd * with a deadline. * - * Loops on partial writes / EINTR / EAGAIN. On EAGAIN, uses select() for + * Loops on partial writes / EINTR / EAGAIN. On EAGAIN, uses poll() for * write-readiness with the remaining deadline. Used by the thread-worker * write path to avoid pinning a dirty I/O scheduler thread on a stalled * Python reader. @@ -1741,13 +1738,8 @@ static write_result_t write_all_with_deadline(int fd, const void *buf, (deadline.tv_sec - now.tv_sec) * 1000L + (deadline.tv_nsec - now.tv_nsec) / 1000000L; if (remain_ms <= 0) return WRITE_TIMEOUT; - struct timeval tv; - tv.tv_sec = remain_ms / 1000; - tv.tv_usec = (remain_ms % 1000) * 1000; - fd_set fds; - FD_ZERO(&fds); - FD_SET(fd, &fds); - int s = select(fd + 1, NULL, &fds, NULL, &tv); + struct pollfd pfd = { .fd = fd, .events = POLLOUT, .revents = 0 }; + int s = poll(&pfd, 1, (int)remain_ms); if (s < 0) { if (errno == EINTR) continue; return WRITE_ERROR; diff --git a/c_src/py_thread_worker.c b/c_src/py_thread_worker.c index d8a3e37..31fde37 100644 --- a/c_src/py_thread_worker.c +++ b/c_src/py_thread_worker.c @@ -488,8 +488,12 @@ static int thread_worker_spawn_handler(thread_worker_t *tw) { * condition (Defect 5): both the byte count must match AND the * value must be zero. Short reads are not silently accepted. */ uint32_t response_len = 0; - ssize_t n = read_with_timeout(tw->response_pipe[0], &response_len, - sizeof(response_len), 10000); + ssize_t n; + /* The coordinator answers without Python; do not hold the GIL for it. */ + Py_BEGIN_ALLOW_THREADS + n = read_with_timeout(tw->response_pipe[0], &response_len, + sizeof(response_len), 10000); + Py_END_ALLOW_THREADS if (n != (ssize_t)sizeof(response_len) || response_len != 0) { return -1; } diff --git a/src/py_thread_handler.erl b/src/py_thread_handler.erl index 4455061..756e5e0 100644 --- a/src/py_thread_handler.erl +++ b/src/py_thread_handler.erl @@ -115,11 +115,17 @@ handle_info({thread_worker_spawn, WorkerId, WriteFd}, #state{handlers = Handlers HandlerPid = spawn_link(fun() -> handler_loop(WorkerId, WriteFd) end), %% Signal readiness to Python (write 0 length to indicate success) - py_nif:thread_worker_signal_ready(WriteFd), - - %% Store handler mapping - NewHandlers = Handlers#{WorkerId => {HandlerPid, WriteFd}}, - {noreply, State#state{handlers = NewHandlers}}; + case py_nif:thread_worker_signal_ready(WriteFd) of + ok -> + NewHandlers = Handlers#{WorkerId => {HandlerPid, WriteFd}}, + {noreply, State#state{handlers = NewHandlers}}; + {error, Reason} -> + %% The Python side times out and reports it; say why here + logger:error("py_thread_handler: ready signal for worker ~p (fd ~p) failed: ~p", + [WorkerId, WriteFd, Reason]), + HandlerPid ! shutdown, + {noreply, State} + end; %% Handle callback request from Python thread handle_info({thread_callback, WorkerId, CallbackId, FuncName, Args}, diff --git a/test/py_reentrant_SUITE.erl b/test/py_reentrant_SUITE.erl index 4eb6f18..77f447a 100644 --- a/test/py_reentrant_SUITE.erl +++ b/test/py_reentrant_SUITE.erl @@ -22,6 +22,7 @@ test_callback_with_complex_types/1, test_multiple_sequential_callbacks/1, test_call_from_non_worker_thread/1, + test_thread_callback_fd_above_fd_setsize/1, test_callback_with_try_except/1, test_async_call/1, test_callback_name_registry/1, @@ -38,6 +39,7 @@ all() -> test_callback_with_complex_types, test_multiple_sequential_callbacks, test_call_from_non_worker_thread, + test_thread_callback_fd_above_fd_setsize, test_callback_with_try_except, test_async_call, test_callback_name_registry, @@ -133,6 +135,12 @@ test_etf_decode_safe(_Config) -> %% Negative: many DISTINCT brand-new atoms wrapped in marker-shaped binaries %% must all come back verbatim, never decoded into atoms. + %% Warm up first so modules loaded on first use (base64, the callback + %% path) do not count as atoms minted by the round trips below. + Warm = etf_marker(novel_atom_etf("zzqx_etf_safe_warmup")), + py:register_function(etf_probe_novel, fun(_) -> Warm end), + {ok, Warm} = py:eval(<<"__import__('erlang').call('etf_probe_novel', [])">>), + assert_atom_absent("zzqx_etf_safe_warmup"), Before = erlang:system_info(atom_count), N = 50, lists:foreach( @@ -349,6 +357,32 @@ test_call_from_non_worker_thread(_Config) -> py:unregister_function(simple_add), ok. +%% @doc Thread callbacks must work when the response pipe lands on an fd +%% above FD_SETSIZE: select() is undefined there and used to make the +%% handler ready-wait time out after 10 s ("Failed to spawn thread handler"). +test_thread_callback_fd_above_fd_setsize(_Config) -> + py:register_function(high_fd_add, fun([A, B]) -> A + B end), + TestDir = filename:join(code:lib_dir(erlang_python), "test"), + ok = py:exec(iolist_to_binary(io_lib:format( + "import sys; sys.path.insert(0, '~s')", [TestDir]))), + try + case py:call(py_test_high_fds, prepare, [1200]) of + {ok, Last} when Last >= 1200 -> + ct:log("highest fd opened: ~p", [Last]), + N = 8, + {ok, Results} = py:call(py_test_high_fds, call_from_threads, [N]), + Expected = [I + 1 || I <- lists:seq(0, N - 1)], + Expected = Results; + {ok, -1} -> + {skip, "fd hard limit too low to open 1200 files"}; + Other -> + ct:fail({prepare_failed, Other}) + end + after + _ = py:call(py_test_high_fds, cleanup, []), + py:unregister_function(high_fd_add) + end. + %% @doc Test that erlang.call() works even when wrapped in try/except blocks. %% This simulates ASGI/WSGI middleware that catches all exceptions. %% The flag-based detection should work even when the SuspensionRequired diff --git a/test/py_test_high_fds.py b/test/py_test_high_fds.py new file mode 100644 index 0000000..ef386fe --- /dev/null +++ b/test/py_test_high_fds.py @@ -0,0 +1,50 @@ +"""Thread callbacks with pipe fds above FD_SETSIZE. + +Opens enough files to push every fd the runtime creates afterwards above +1024, then has several new threads call Erlang at once so fresh thread +workers (and their pipes) are created in that range. select() cannot +watch such fds; poll() can. +""" +import concurrent.futures +import os +import resource + +_kept = [] + + +def prepare(target=1200): + """Raise the fd soft limit and fill descriptors up to `target`. + + Returns the highest fd opened, or -1 if the hard limit is too low. + """ + soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE) + want = target + 256 + if hard != resource.RLIM_INFINITY and hard < want: + return -1 + if soft < want: + resource.setrlimit(resource.RLIMIT_NOFILE, (want, hard)) + last = -1 + while last < target: + fd = os.open(os.devnull, os.O_RDONLY) + _kept.append(fd) + last = fd + return last + + +def call_from_threads(n): + import erlang + with concurrent.futures.ThreadPoolExecutor(max_workers=n) as ex: + futures = [ex.submit(erlang.call, 'high_fd_add', i, 1) for i in range(n)] + results = [] + for f in futures: + try: + results.append(f.result()) + except Exception as exc: + results.append('error: %s' % exc) + return results + + +def cleanup(): + while _kept: + os.close(_kept.pop()) + return 'ok'