diff --git a/py4j-python/src/py4j/java_gateway.py b/py4j-python/src/py4j/java_gateway.py index 72f0eac2..a3f99bae 100644 --- a/py4j-python/src/py4j/java_gateway.py +++ b/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.") diff --git a/py4j-python/src/py4j/tests/java_callback_test.py b/py4j-python/src/py4j/tests/java_callback_test.py index a2f7b179..85a6508c 100644 --- a/py4j-python/src/py4j/tests/java_callback_test.py +++ b/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() diff --git a/py4j-web/changelog.rst b/py4j-web/changelog.rst index 1f9e39f2..7ea55bc9 100644 --- a/py4j-web/changelog.rst +++ b/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 diff --git a/py4j-web/faq.rst b/py4j-web/faq.rst index f866c6ab..c48354a5 100644 --- a/py4j-web/faq.rst +++ b/py4j-web/faq.rst @@ -214,6 +214,22 @@ default on most systems. To secure Py4J applications, refer to the :ref:`TLS And Authentication ` 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? ----------------------------------