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

wjszlachta-man · 2025-12-03T11:21:11Z

What changes were proposed in this pull request?

On glibc based Linux systems select() can monitor only file descriptor numbers that are less than FD_SETSIZE (1024).

This is an unreasonably low limit for many modern applications.

This PR replaces select.select() with select.poll() when running on POSIX os.

Why are the changes needed?

When running via pyspark we frequently observe:

Exception occurred during processing of request from ('127.0.0.1', 46334)
Traceback (most recent call last):
  File "/usr/lib/python3.11/socketserver.py", line 317, in _handle_request_noblock
    self.process_request(request, client_address)
  File "/usr/lib/python3.11/socketserver.py", line 348, in process_request
    self.finish_request(request, client_address)
  File "/usr/lib/python3.11/socketserver.py", line 361, in finish_request
    self.RequestHandlerClass(request, client_address, self)
  File "/usr/lib/python3.11/socketserver.py", line 755, in __init__
    self.handle()
  File "/usr/lib/python3.11/site-packages/pyspark/accumulators.py", line 293, in handle
    poll(authenticate_and_accum_updates)
  File "/usr/lib/python3.11/site-packages/pyspark/accumulators.py", line 266, in poll
    r, _, _ = select.select([self.rfile], [], [], 1)
              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
ValueError: filedescriptor out of range in select()

On POSIX systems poll() should be used instead of select().

Does this PR introduce any user-facing change?

No

How was this patch tested?

Existing unit tests + we have been running this change (combined with py4j/py4j#560) on our YARN cluster (Linux) since April 2025.

Was this patch authored or co-authored using generative AI tooling?

No

…osix On glibc based Linux systems select() can monitor only file descriptor numbers that are less than FD_SETSIZE (1024). This is an unreasonably low limit for many modern applications.

wjszlachta-man · 2025-12-03T11:23:48Z

This is identical to #50774 (which was never reviewed and closed by the bot), but rebased against current master.

wjszlachta-man · 2025-12-03T11:27:32Z

@HyukjinKwon is that something you could maybe review (in combination with py4j/py4j#560)?

We needed to implement this change to allow us to run 1000+ executors without running into filedescriptor out of range in select() error.

HyukjinKwon · 2025-12-04T00:25:11Z

Can we have an environment variable to fallback?

… select.select()

wjszlachta-man · 2025-12-04T15:36:55Z

@HyukjinKwon you can now use PYSPARK_FORCE_SELECT to fallback to select.select().

wjszlachta-man · 2025-12-09T12:51:12Z

@HyukjinKwon @gaogaotiantian - I see you now merged #53388

Can you merge a similar change to python/pyspark/accumulators.py as per this PR?

wjszlachta-man · 2025-12-09T13:11:28Z

See traceback:

Exception occurred during processing of request from ('127.0.0.1', 46334)
Traceback (most recent call last):
  File "/usr/lib/python3.11/socketserver.py", line 317, in _handle_request_noblock
    self.process_request(request, client_address)
  File "/usr/lib/python3.11/socketserver.py", line 348, in process_request
    self.finish_request(request, client_address)
  File "/usr/lib/python3.11/socketserver.py", line 361, in finish_request
    self.RequestHandlerClass(request, client_address, self)
  File "/usr/lib/python3.11/socketserver.py", line 755, in __init__
    self.handle()
  File "/usr/lib/python3.11/site-packages/pyspark/accumulators.py", line 293, in handle
    poll(authenticate_and_accum_updates)
  File "/usr/lib/python3.11/site-packages/pyspark/accumulators.py", line 266, in poll
    r, _, _ = select.select([self.rfile], [], [], 1)
              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
ValueError: filedescriptor out of range in select()

gaogaotiantian · 2025-12-09T16:53:25Z

Could you make sure the CI pass? I think it’s the linter issue. Also it has some conflicts now.

wjszlachta-man · 2025-12-09T17:27:57Z

Sure - let me rebase... (although conflict is because of #53388 - not sure why duplicate?)

Can we have an environment variable to fallback?

Do you still want environment variable to fallback? I see you don't have one in #53388

gaogaotiantian · 2025-12-09T18:34:34Z

Sure - let me rebase... (although conflict is because of #53388 - not sure why duplicate?)

That would be my fault. I did not realize this PR exists when I fix the worker. Whether we should have an envvar fallback is a decision for @HyukjinKwon . Personally I think it's okay to just replace the old mechanism. We would need a whole config path for the fallback to work. With the heavily executed CI I think we can validate this local scope change pretty well.

wjszlachta-man · 2025-12-09T18:42:42Z

Just updated the branch - this should fix conflict and overall is similar to your changes in python/pyspark/daemon.py.

Removed the fallback - if we have one it should be for both daemon.py and accumulators.py (personally I think it's redundant and should always use poll() if available).

As per your PR I updated poll() timeout to 1000 - I missed it was in millis (unlike select()) in my original commit 👍

HyukjinKwon · 2025-12-09T22:54:09Z

Does this LGTM, @gaogaotiantian ?

gaogaotiantian · 2025-12-09T23:11:52Z

python/pyspark/accumulators.py

-                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]


Okay I took some time on this and I think my implementation lacks error checking as well which I will fix later. However, we need to deal with errors here. If the socket has some issues, we will busy loop here forever because poller.poll() does not raise an exception.

We should break on POLLHUP and raise an error on POLLERR and POLLNVAL. We also need to confirm that there's no other event set.

Ok - you are right - so we want to handle these:

POLLIN and POLLHUP we want to break - when using select() these both return "ready for reading" - if peer hangs up reads will return 0 bytes

POLLERR and POLLNVAL we want to raise - select() will already raise on error

… poll()

wjszlachta-man · 2025-12-10T13:41:20Z

@gaogaotiantian let me know what you think re. latest commit - it will check for errors in both accumulators.py and daemon.py.

I also removed try/catch around:

try:
    ready_fds = select.select([0, listen_sock], [], [], 1)[0]
except select.error as ex:
    if ex[0] == EINTR:
        continue
    else:
        raise

in daemon.py as this is old code and only needed for Python <3.5 (see: https://peps.python.org/pep-0475/ - from Python 3.5 onward select.select() will automatically retry system calls on EINTR).

Considering python_requires=">=3.9" as of Spark>=4, it should be safe to remove.

[SPARK-51966][PYTHON] replace select.select() with select.poll() on p…

88e18d4

…osix On glibc based Linux systems select() can monitor only file descriptor numbers that are less than FD_SETSIZE (1024). This is an unreasonably low limit for many modern applications.

github-actions bot added CORE PYTHON labels Dec 3, 2025

wjszlachta-man mentioned this pull request Dec 3, 2025

fixes 559 -- replace select.select() with select.poll() on posix py4j/py4j#560

Open

[SPARK-51966][PYTHON] add env var PYSPARK_FORCE_SELECT to fallback to…

e4d08b0

… select.select()

wjszlachta-man mentioned this pull request Dec 9, 2025

[SPARK-54640][PYTHON] Replace select.select with select.poll on UNIX #53388

Closed

wjszlachta-man added 2 commits December 9, 2025 17:40

Merge branch 'master' into spark-51966-replace-select-with-poll-on-posix

6da8aad

[SPARK-51966][PYTHON] reflect changes in python/pyspark/daemon.py

4ff1538

gaogaotiantian reviewed Dec 9, 2025

View reviewed changes

[SPARK-51966][PYTHON] handle POLLHUP, POLLERR and POLLNVAL when using…

d04b73a

… poll()

[SPARK-51966][PYTHON] add type hints for mypy

30505f9

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

Are you sure you want to change the base?

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

Uh oh!

Conversation

wjszlachta-man commented Dec 3, 2025

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Was this patch authored or co-authored using generative AI tooling?

Uh oh!

wjszlachta-man commented Dec 3, 2025

Uh oh!

wjszlachta-man commented Dec 3, 2025

Uh oh!

HyukjinKwon commented Dec 4, 2025

Uh oh!

wjszlachta-man commented Dec 4, 2025

Uh oh!

wjszlachta-man commented Dec 9, 2025

Uh oh!

wjszlachta-man commented Dec 9, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

gaogaotiantian commented Dec 9, 2025

Uh oh!

wjszlachta-man commented Dec 9, 2025

Uh oh!

gaogaotiantian commented Dec 9, 2025

Uh oh!

wjszlachta-man commented Dec 9, 2025

Uh oh!

HyukjinKwon commented Dec 9, 2025

Uh oh!

gaogaotiantian Dec 9, 2025

Choose a reason for hiding this comment

Uh oh!

wjszlachta-man Dec 10, 2025

Choose a reason for hiding this comment

Uh oh!

wjszlachta-man commented Dec 10, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

wjszlachta-man commented Dec 9, 2025 •

edited

Loading

wjszlachta-man commented Dec 10, 2025 •

edited

Loading