Replica: #560 select→poll (#559) + conflict fix + hardening + tests

py4j-python/src/py4j/java_gateway.py

@@ -58,6 +58,7 @@ def emit(self, record):
 DEFAULT_CALLBACK_SERVER_ACCEPT_TIMEOUT = 5
 PY4J_SKIP_COLLECTIONS = "PY4J_SKIP_COLLECTIONS"
 PY4J_TRUE = {"yes", "y", "t", "true"}
+PY4J_FORCE_SELECT = "PY4J_FORCE_SELECT"
 
 
 server_connection_stopped = Signal()
@@ -951,7 +952,12 @@ def __init__(
 
 class CallbackServerParameters(object):
     """Wrapper class that contains all parameters that can be passed to
-    configure a `CallbackServer`
+    configure a `CallbackServer`.
+
+    On POSIX systems the callback server's accept loop uses ``select.poll()``,
+    which (unlike ``select.select()``) is not bounded by ``FD_SETSIZE``; set
+    the ``PY4J_FORCE_SELECT`` environment variable to force the legacy
+    ``select()`` path. Windows always uses ``select()``.
     """
 
     def __init__(
@@ -2617,35 +2623,102 @@ def run(self):
                 self, server=self)
 
             read_list = [self.server_socket]
-            while not self.is_shutdown:
-                readable, writable, errored = select.select(
-                    read_list, [], [],
-                    self.callback_server_parameters.accept_timeout)
-
-                if self.is_shutdown:
-                    break
+            poller = None
+            try:
+                if (
+                    os.name == "posix"
+                    and hasattr(select, "poll")
+                    and os.getenv(
+                        PY4J_FORCE_SELECT, "").lower() not in PY4J_TRUE
+                ):
+                    # On posix systems use poll to avoid problems with file
+                    # descriptor numbers above 1024 (select is bounded by
+                    # FD_SETSIZE). Guarded by hasattr because a few posix
+                    # builds (e.g. Emscripten/WASM) ship select without poll.
+                    # Set PY4J_FORCE_SELECT to force the select path.
+                    poller = select.poll()
+                    for r in read_list:
+                        poller.register(r, select.POLLIN)
+
+                while not self.is_shutdown:
+                    if poller is not None:
+                        # poll's timeout is in milliseconds (select uses
+                        # seconds). A None accept_timeout means "block until
+                        # an event"; mirror select's handling of None rather
+                        # than crashing on ``1000 * None``.
+                        accept_timeout = \
+                            self.callback_server_parameters.accept_timeout
+                        poll_timeout = (
+                            None if accept_timeout is None
+                            else 1000 * accept_timeout)
+                        readable_fds = []
+                        for fd, event in poller.poll(poll_timeout):
+                            if event & select.POLLIN:
+                                # A connection is waiting to be accepted.
+                                readable_fds.append(fd)
+                            elif not self.is_shutdown:
+                                # POLLERR / POLLHUP / POLLNVAL without POLLIN
+                                # on the listening socket. The select-based
+                                # path passed an empty exceptional set -- it
+                                # ignored these conditions and kept waiting --
+                                # so mirror that here rather than tearing the
+                                # callback server down on a transient socket
+                                # error (a regression for long-lived servers,
+                                # the very case #559 targets). During shutdown
+                                # the socket was closed under us, so stay
+                                # silent. We deliberately do NOT accept() on a
+                                # POLLHUP-only event: accept() on a listening
+                                # socket with no pending connection blocks.
+                                logger.warning(
+                                    "Ignoring unexpected poll event %s on the "
+                                    "callback server listening socket (fd %s)",
+                                    event, fd)
+                        readable = [
+                            r for r in read_list if r.fileno() in readable_fds
+                        ]
+                    else:
+                        # If poll is not available, use select.
+                        readable, writable, errored = select.select(
+                            read_list, [], [],
+                            self.callback_server_parameters.accept_timeout)
+
+                    if self.is_shutdown:
+                        break
 
-                for s in readable:
-                    socket_instance, _ = self.server_socket.accept()
-                    disable_nagle(socket_instance)
-                    if self.callback_server_parameters.read_timeout:
-                        socket_instance.settimeout(
-                            self.callback_server_parameters.read_timeout)
-                    if self.ssl_context:
-                        socket_instance = self.ssl_context.wrap_socket(
-                            socket_instance, server_side=True)
-                    input = socket_instance.makefile("rb")
-                    connection = self._create_connection(
-                        socket_instance, input)
-                    with self.lock:
-                        if not self.is_shutdown:
-                            self.connections.add(connection)
-                            connection.start()
-                            server_connection_started.send(
-                                self, connection=connection)
-                        else:
-                            quiet_shutdown(connection.socket)
-                            quiet_close(connection.socket)
+                    for s in readable:
+                        socket_instance, _ = self.server_socket.accept()
+                        disable_nagle(socket_instance)
+                        if self.callback_server_parameters.read_timeout:
+                            socket_instance.settimeout(
+                                self.callback_server_parameters.read_timeout)
+                        if self.ssl_context:
+                            socket_instance = self.ssl_context.wrap_socket(
+                                socket_instance, server_side=True)
+                        input = socket_instance.makefile("rb")
+                        connection = self._create_connection(
+                            socket_instance, input)
+                        with self.lock:
+                            if not self.is_shutdown:
+                                self.connections.add(connection)
+                                connection.start()
+                                server_connection_started.send(
+                                    self, connection=connection)
+                            else:
+                                quiet_shutdown(connection.socket)
+                                quiet_close(connection.socket)
+            finally:
+                if poller is not None:
+                    for r in read_list:
+                        # During shutdown the socket may already be closed
+                        # (fileno() == -1); guard so unregister can't mask
+                        # the real exception on the way out of the loop.
+                        try:
+                            poller.unregister(r)
+                        except (KeyError, OSError, ValueError) as unreg_exc:
+                            logger.debug(
+                                "Ignoring poller.unregister error during "
+                                "callback server teardown: %s: %s",
+                                type(unreg_exc).__name__, str(unreg_exc))
         except Exception as e:
             if self.is_shutdown:
                 logger.info("Error while waiting for a connection.")

py4j-python/src/py4j/tests/java_callback_test.py

@@ -5,14 +5,17 @@
 """
 from contextlib import contextmanager
 from multiprocessing import Process
+import os
+import select
 import subprocess
 from threading import Thread
 from traceback import print_exc
 import unittest
+from unittest.mock import patch
 
 from py4j.java_gateway import (
     JavaGateway, PythonProxyPool, CallbackServerParameters,
-    set_default_callback_accept_timeout, is_instance_of)
+    set_default_callback_accept_timeout, is_instance_of, PY4J_FORCE_SELECT)
 from py4j.protocol import Py4JJavaError
 from py4j.tests.java_gateway_test import (
     PY4J_JAVA_PATH, safe_join, safe_shutdown, sleep, check_connection,
@@ -593,5 +596,139 @@ def testByteString(self):
             self.fail()
 
 
+class _WrappingPoller(object):
+    """Wraps a real ``select.poll`` object so a test can confirm the accept
+    loop actually CALLS ``poll()`` (not merely that a poller was created),
+    and optionally inject a synthetic event on the first ``poll()`` call to
+    exercise the error path. register/unregister delegate to the real poller.
+    """
+
+    def __init__(self, real, inject_first=None):
+        self._real = real
+        self._inject_first = inject_first
+        self.poll_calls = 0
+
+    def register(self, *args, **kwargs):
+        return self._real.register(*args, **kwargs)
+
+    def unregister(self, *args, **kwargs):
+        return self._real.unregister(*args, **kwargs)
+
+    def poll(self, *args, **kwargs):
+        self.poll_calls += 1
+        if self._inject_first is not None and self.poll_calls == 1:
+            return self._inject_first
+        return self._real.poll(*args, **kwargs)
+
+
+class CallbackServerPollSelectTest(unittest.TestCase):
+    """Covers the CallbackServer accept loop under both back-ends:
+
+    - poll (the posix default; issue #559 -- avoids select's
+      ``FD_SETSIZE=1024`` ceiling so callback sockets with high fd
+      numbers still work), and
+    - select (Windows, or forced on posix via ``PY4J_FORCE_SELECT``).
+
+    The poll path is already exercised implicitly by every callback test
+    on posix CI; these tests pin it explicitly and add the otherwise
+    uncovered select fallback and env-var escape hatch.
+    """
+
+    def setUp(self):
+        self.p = start_example_app_process()
+        self.gateway = None
+        sleep()
+
+    def tearDown(self):
+        if self.gateway is not None:
+            safe_shutdown(self)
+        self.p.join()
+        os.environ.pop(PY4J_FORCE_SELECT, None)
+        sleep()
+
+    def _assert_callback_works(self):
+        # A Java->Python callback forces the accept loop to accept a
+        # callback connection, so it must have gone through the active
+        # back-end (poll or select) to succeed.
+        example = self.gateway.entry_point.getNewExample()
+        self.assertEqual("This is Hello!", example.callHello(IHelloImpl()))
+
+    @unittest.skipIf(os.name != "posix", "poll is only used on posix")
+    def testAcceptLoopUsesPollOnPosix(self):
+        # White-box: the accept loop must actually CALL poll() (not merely
+        # construct a poller) on posix -- the fix for #559's FD_SETSIZE
+        # ceiling. A wrapping poller records poll() invocations. Because a
+        # successful callback requires the accept loop to have accepted the
+        # connection, poll_calls > 0 is guaranteed once the callback returns.
+        pollers = []
+        real_poll = select.poll
+
+        def make_poller():
+            wrapper = _WrappingPoller(real_poll())
+            pollers.append(wrapper)
+            return wrapper
+
+        with patch("py4j.java_gateway.select.poll", side_effect=make_poller):
+            self.gateway = JavaGateway(
+                callback_server_parameters=CallbackServerParameters())
+            sleep()
+            self._assert_callback_works()
+        self.assertTrue(pollers, "a poll object should have been created")
+        self.assertGreater(
+            pollers[0].poll_calls, 0,
+            "poller.poll() should drive the accept loop on posix")
+
+    @unittest.skipIf(os.name != "posix", "select is the only path off posix")
+    def testForceSelectSkipsPoll(self):
+        # PY4J_FORCE_SELECT must make the accept loop use select (not poll)
+        # even on posix; poll must never be constructed and select.select
+        # must actually back the loop (proven by the callback succeeding).
+        os.environ[PY4J_FORCE_SELECT] = "true"
+        with patch("py4j.java_gateway.select.poll",
+                   wraps=select.poll) as mock_poll, \
+                patch("py4j.java_gateway.select.select",
+                      wraps=select.select) as mock_select:
+            self.gateway = JavaGateway(
+                callback_server_parameters=CallbackServerParameters())
+            sleep()
+            self._assert_callback_works()
+        mock_poll.assert_not_called()
+        self.assertTrue(
+            mock_select.called,
+            "select.select() should back the accept loop when forced")
+
+    @unittest.skipIf(os.name != "posix", "poll is only used on posix")
+    def testTransientPollErrorDoesNotKillServer(self):
+        # A transient POLLERR on the listening socket must be logged and
+        # ignored (the old select path passed an empty exceptional set, i.e.
+        # ignored these), NOT raised -- otherwise the callback server would
+        # die on a transient error, a regression for the long-lived servers
+        # #559 targets. Inject a synthetic POLLERR on the first poll(), then
+        # confirm a callback still succeeds (the loop survived).
+        real_poll = select.poll
+
+        def make_poller():
+            return _WrappingPoller(
+                real_poll(), inject_first=[(999, select.POLLERR)])
+
+        with patch("py4j.java_gateway.select.poll", side_effect=make_poller):
+            self.gateway = JavaGateway(
+                callback_server_parameters=CallbackServerParameters())
+            sleep()
+            # Had the POLLERR raised, the accept loop would be dead and this
+            # callback would fail/hang instead of returning.
+            self._assert_callback_works()
+
+    def testCleanShutdownDrainsPool(self):
+        # A clean shutdown must drain the proxy pool without the poll-path
+        # teardown (poller.unregister in the finally) raising.
+        self.gateway = JavaGateway(
+            callback_server_parameters=CallbackServerParameters())
+        sleep()
+        self._assert_callback_works()
+        self.gateway.shutdown()
+        self.assertEqual(0, len(self.gateway.gateway_property.pool))
+
+
 if __name__ == "__main__":
     unittest.main()

py4j-web/changelog.rst

@@ -7,6 +7,16 @@ releases.
 Unreleased
 ----------
 
+- Python side: The callback server's accept loop now uses ``select.poll()``
+  instead of ``select.select()`` on POSIX systems. ``select()`` is bounded by
+  ``FD_SETSIZE`` (typically 1024), so a callback-server socket assigned a file
+  descriptor >= 1024 -- common in long-lived processes such as PySpark drivers
+  that have opened many descriptors -- raised ``ValueError: filedescriptor out
+  of range in select()``. ``poll()`` has no such limit. Windows, and any POSIX
+  build without ``select.poll``, continue to use ``select()``. Set the
+  ``PY4J_FORCE_SELECT`` environment variable (to ``yes``/``y``/``t``/``true``)
+  to force the legacy ``select()`` path on POSIX. Closes issue #559.
+
 - Python side: Memoize attribute lookups on ``JVMView``,
   ``JavaPackage``, and ``JavaClass`` via a per-instance bounded LRU
   cache (default 1024 entries). Repeated walks along chains like

py4j-web/faq.rst

@@ -214,6 +214,22 @@ default on most systems.
 To secure Py4J applications, refer to the :ref:`TLS And Authentication <security>`
 documentation.
 
+Why do I get "filedescriptor out of range in select()" with callbacks?
+----------------------------------------------------------------------
+
+This happens in long-lived processes (for example PySpark drivers) that have
+opened many file descriptors: the callback server's socket is assigned a file
+descriptor number greater than or equal to ``FD_SETSIZE`` (typically 1024),
+which the underlying ``select()`` call cannot represent. Since Py4J 0.11 the
+callback server uses ``select.poll()`` on POSIX systems, which has no such
+limit, so this should no longer occur.
+
+If you need the legacy behavior, set the ``PY4J_FORCE_SELECT`` environment
+variable before starting your application. This reintroduces the limit and is
+intended only as an escape hatch::
+
+  export PY4J_FORCE_SELECT=yes
+
 I found a bug, how do I report it?
 ----------------------------------