Skip to content

[SPARK-51966][PYTHON] Replace select.select() with select.poll() when running on POSIX os #53306

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wjszlachta-man wants to merge 7 commits into from
Closed

[SPARK-51966][PYTHON] Replace select.select() with select.poll() when running on POSIX os #53306

Changes from 4 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
88e18d4
[SPARK-51966][PYTHON] replace select.select() with select.poll() on p…
wjszlachta-man Apr 30, 2025
e4d08b0
[SPARK-51966][PYTHON] add env var PYSPARK_FORCE_SELECT to fallback to…
wjszlachta-man Dec 4, 2025
6da8aad
Merge branch 'master' into spark-51966-replace-select-with-poll-on-posix
wjszlachta-man Dec 9, 2025
4ff1538
[SPARK-51966][PYTHON] reflect changes in python/pyspark/daemon.py
wjszlachta-man Dec 9, 2025
d04b73a
[SPARK-51966][PYTHON] handle POLLHUP, POLLERR and POLLNVAL when using…
wjszlachta-man Dec 10, 2025
30505f9
[SPARK-51966][PYTHON] add type hints for mypy
wjszlachta-man Dec 10, 2025
879db0e
[SPARK-51966][PYTHON] use list for both select() and poll() for consi…
wjszlachta-man Dec 15, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 17 additions & 2 deletions python/pyspark/accumulators.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -266,12 +266,27 @@ def handle(self) -> None:
auth_token = self.server.auth_token # type: ignore[attr-defined]

def poll(func: Callable[[], bool]) -> None:
poller = None
if os.name == "posix":
# On posix systems use poll to avoid problems with file descriptor
# numbers above 1024.
poller = select.poll()
poller.register(self.rfile, select.POLLIN)

while not self.server.server_shutdown: # type: ignore[attr-defined]
# Poll every 1 second for new data -- don't block in case of shutdown.
r, _, _ = select.select([self.rfile], [], [], 1)
if self.rfile in r and func():
if poller is not None:
# Unlike select, poll timeout is in millis. Rule out error events.
r = [fd for fd, event in poller.poll(1000) if event & select.POLLIN]
else:
# If poll is not available, use select.
r, _, _ = select.select([self.rfile.fileno()], [], [], 1)
if self.rfile.fileno() in r and func():
break

if poller is not None:
poller.unregister(self.rfile)

def accum_updates() -> bool:
num_updates = read_int(self.rfile)
for _ in range(num_updates):
Expand Down