Wait on callback pipes with poll

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)
 

c_src/py_nif.h

@@ -64,7 +64,7 @@
 #define NEED_DLOPEN_GLOBAL 1
 #endif
 
-#include <sys/select.h>
+#include <poll.h>
 /** @} */
 
 /* ============================================================================
@@ -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;

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;
     }

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},

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

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'