ericm-db commented on code in PR #58367:
URL: https://github.com/apache/spark/pull/58367#discussion_r4223832614
##########
python/pyspark/sql/connect/local_server_pool.py:
##########
@@ -613,20 +657,88 @@ def _signal_server(cls, pid: int, process_start_id: str,
sig: int) -> bool:
def _idle_timeout(cls) -> int:
"""Seconds an unclaimed member may sit before it is retired.
- Zero or a negative value disables idle retirement. Read the
environment on each pass so
- every reaper uses the same source of truth.
+ Zero or a negative value disables idle retirement. Read the
environment wherever
+ reaping runs so clients and attendants use the same source of truth.
"""
try:
return int(os.environ["SPARK_LOCAL_CONNECT_POOL_IDLE_TIMEOUT"])
except (KeyError, ValueError):
return cls._DEFAULT_IDLE_TIMEOUT_SECONDS
+ @classmethod
+ def _is_pool_attendant(cls, pid: int, uid: str) -> Optional[bool]:
+ """Whether ``pid`` is still the pool attendant recorded for ``uid``.
+
+ Returns ``None`` when the process cannot be inspected. A stale pending
record can
+ outlive its attendant long enough for the pid to be reused, so
liveness alone is not
+ sufficient before a janitor signals it.
+ """
+ command = _process_command(pid)
+ if command is None:
+ return None
+ args = command.split()
+ try:
+ module_index = args.index(cls._ATTENDANT_MODULE)
+ uid_index = args.index("--uid")
+ except ValueError:
+ return False
+ return (
+ module_index > 0
+ and args[module_index - 1] == "-m"
+ and "--attend" in args
+ and uid_index + 1 < len(args)
+ and args[uid_index + 1] == uid
+ )
+
+ @staticmethod
+ def _attendant_group_alive(pgid: int) -> bool:
+ """Whether a recorded attendant process group still has any members."""
+ if pgid <= 0 or pgid == os.getpgrp():
+ return False
+ try:
+ os.killpg(pgid, 0)
+ return True
+ except (ProcessLookupError, OverflowError):
+ return False
+ except OSError:
+ # As with _pid_alive, an existing group we cannot signal still
counts as alive.
+ return True
+
+ @staticmethod
+ def _signal_attendant_group(pid: int, sig: int, *, leader_may_be_dead:
bool = False) -> bool:
+ """Signal a detached attendant and the launch subprocesses in its
process group.
+
+ A process group survives its leader while any child remains, and its
id cannot be
+ recycled during that time. When the recorded leader is already gone,
signal the still
+ owned group id directly; ``killpg`` then fails harmlessly if the group
is empty.
+ """
+ if pid <= 0 or pid == os.getpgrp():
+ return False
+ try:
+ try:
+ if os.getpgid(pid) != pid:
+ return False
+ if leader_may_be_dead and _pid_alive(pid):
+ # The caller observed a dead leader, but this pid now
belongs to a live
+ # process. It was reused between checks, so do not signal
its group.
+ return False
+ except ProcessLookupError:
+ if not leader_may_be_dead:
+ return False
+ os.killpg(pid, sig)
Review Comment:
Fixed in 56b3129b655: a verified Connect server in the recorded launch group
now permits cleanup at any record age. Unverified groups still time out. Added
a regression for a verified orphan found after the launch timeout.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
