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

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.

Author: wjszlachta-man

python/pyspark/accumulators.py

@@ -266,11 +266,29 @@ def handle(self) -> None:
         auth_token = self.server.auth_token  # type: ignore[attr-defined]
 
         def poll(func: Callable[[], bool]) -> None:
-            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():
-                    break
+            rlist = [self.rfile.fileno()]
+            poller = None
+            try:
+                if os.name == "posix":
+                    # On posix systems use poll to avoid problems with file descriptor numbers
+                    # above 1024.
+                    poller = select.poll()
+                    for fd in rlist:
+                        poller.register(fd, 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.
+                    if poller is not None:
+                        r = [fd for fd, event in poller.poll(1) if event & select.POLLIN]
+                    else:
+                        # If poll is not available, use select.
+                        r, _, _ = select.select(rlist, [], [], 1)
+                    if self.rfile.fileno() in r and func():
+                        break
+            finally:
+                if poller is not None:
+                    for fd in rlist:
+                        poller.unregister(fd)
 
         def accum_updates() -> bool:
             num_updates = read_int(self.rfile)

python/pyspark/daemon.py

@@ -144,10 +144,22 @@ def handle_sigterm(*args):
     reuse = os.environ.get("SPARK_REUSE_WORKER")
 
     # Initialization complete
+    rlist = [0, listen_sock.fileno()]
+    poller = None
     try:
+        if os.name == "posix":
+            # On posix systems use poll to avoid problems with file descriptor numbers above 1024.
+            poller = select.poll()
+            for fd in rlist:
+                poller.register(fd, select.POLLIN)
+
         while True:
             try:
-                ready_fds = select.select([0, listen_sock], [], [], 1)[0]
+                if poller is not None:
+                    ready_fds = [fd for fd, event in poller.poll(1)]
+                else:
+                    # If poll is not available, use select.
+                    ready_fds = select.select(rlist, [], [], 1)[0]
             except select.error as ex:
                 if ex[0] == EINTR:
                     continue
@@ -165,7 +177,7 @@ def handle_sigterm(*args):
                 except OSError:
                     pass  # process already died
 
-            if listen_sock in ready_fds:
+            if listen_sock.fileno() in ready_fds:
                 try:
                     sock, _ = listen_sock.accept()
                 except OSError as e:
@@ -238,6 +250,10 @@ def handle_sigterm(*args):
                     sock.close()
 
     finally:
+        if poller is not None:
+            for fd in rlist:
+                poller.unregister(fd)
+
         shutdown(1)