ericm-db commented on code in PR #58367:
URL: https://github.com/apache/spark/pull/58367#discussion_r4213148510
##########
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 a98fa7722f3: a dead-leader group signal now requires a recent valid
pending record and a live recorded Connect server in that group. Otherwise
recovery waits until the launch deadline, then withdraws without signalling.
Added the recycled leaderless-group regression.
##########
python/pyspark/sql/connect/local_server_pool.py:
##########
@@ -516,14 +553,21 @@ def remove_member_dir(self, uid: str) -> None:
class ServerPool:
"""Claims, reaps, and retires members of one pool directory."""
+ # A pending marker older than this belongs to a launch that hung. A launch
can spend the
+ # maximum in both the script and readiness phases; leave another minute
for setup and
+ # scheduling so a slow but healthy launch is never stopped by the janitor.
+ _LAUNCH_TIMEOUT_SECONDS = ServerLauncher._MAX_STARTUP_SECONDS + 60
# A retired server still alive after the grace period is hard-killed. With
a process handle,
# tracking is removed only once it is gone, replaced, or successfully
signalled. PID-less
# malformed state uses the give-up age as its bounded recovery window.
_RETIRE_KILL_AFTER_SECONDS = 30
_RETIRE_GIVE_UP_AFTER_SECONDS = 600
+ # Preserve a failed launch's logs for diagnosis before collecting its
unreferenced directory.
+ _MEMBER_DIR_GC_AGE_SECONDS = 24 * 3600
Review Comment:
Fixed in a98fa7722f3: abort retirements retain their member directory for
the existing 24-hour log grace period, including the pid-file/dead-server case.
Added a two-pass regression.
##########
python/pyspark/sql/connect/local_server_pool.py:
##########
@@ -680,7 +814,89 @@ def reap(self, uid: str) -> bool:
states = self._directory.states(uid)
if had_retired and "retired" in states:
self._reap_retired(uid, states["retired"])
- return not self._directory.states(uid)
+
+ remaining = self._directory.states(uid)
+ if set(remaining) == {"member"}:
+ # Nothing references the member directory anymore. The age gate
keeps the logs
+ # of a freshly failed launch around long enough to be looked at.
+ try:
+ expired = (
+ time.time() - os.path.getmtime(remaining["member"])
+ > self._MEMBER_DIR_GC_AGE_SECONDS
+ )
+ except FileNotFoundError:
+ expired = True
+ if expired:
+ self._directory.remove_member_dir(uid)
+ remaining = self._directory.states(uid)
+ return not remaining
+
+ def _reap_pending(self, uid: str, path: str) -> None:
+ """A launch whose attendant died or hung: kill the attendant and
whatever server
+ spark-daemon.sh may have recorded for it, and withdraw the launch's
bookkeeping so
+ refills stop counting it."""
+ data = self._directory.read_json(path)
+ pending = PendingState.from_data(data)
+ parsed_pid = pending.attendant_pid if pending is not None else None
+ created = pending.created if pending is not None else None
+ if pending is None and data is not None:
+ # Preserve independently valid lifecycle fields when another field
is corrupt.
+ parsed_pid = PendingState.attendant_pid_from_data(data)
+ created = PendingState.created_from_data(data)
+ age = time.time() - created if created is not None else
self._LAUNCH_TIMEOUT_SECONDS + 1
+ attendant_pid = parsed_pid if parsed_pid is not None else -1
+ attendant_alive = _pid_alive(attendant_pid)
+ if not attendant_alive:
+ # Recovered fields from malformed state guide cleanup, but cannot
authorize a
+ # process-group signal because the complete pending record was not
validated.
+ if pending is not None and not self._signal_attendant_group(
+ attendant_pid, signal.SIGKILL, leader_may_be_dead=True
+ ):
+ # Keep the only launch-group handle when signalling failed but
descendants
+ # remain. If the pid was reused by a live process, withdraw
the stale record
+ # without signalling it.
+ if not _pid_alive(attendant_pid) and
self._attendant_group_alive(attendant_pid):
+ return
+ self.abort_launch(uid)
+ elif age > self._LAUNCH_TIMEOUT_SECONDS:
+ is_attendant = self._is_pool_attendant(attendant_pid, uid)
+ if is_attendant is None:
+ return
+ if is_attendant and not
self._signal_attendant_group(attendant_pid, signal.SIGKILL):
+ # Keep the record when an attendant that still appears live
could not be
+ # stopped, or when its leader exited during the attempt but
children remain.
+ if _pid_alive(attendant_pid) or
self._attendant_group_alive(attendant_pid):
+ return
+ self.abort_launch(uid)
+
+ def abort_launch(self, uid: str) -> None:
+ """Withdraw a failed launch and retire any server it started before
failing."""
+ states = self._directory.states(uid)
+ pending_path = states.get("pending")
+ server_path = states.get("server")
+ if "retired" in states:
+ # A previous abort can die after retiring the server but before
removing the pending
+ # marker. The retired record owns the server's process-generation
identity; never
+ # replace it with the weaker attendant record on the recovery pass.
+ if pending_path is not None:
+ self._directory.remove(pending_path)
+ self._directory.remove(self._directory.conf_path(uid))
+ return
+ if server_path is not None:
+ data = self._directory.read_json(server_path)
+ server_pid, process_start_id = self._recover_server_handle(uid,
data)
+ else:
+ server_pid = self._recorded_daemon_pid(uid)
+ process_start_id = None
+ retirement_source = server_path or pending_path
+ retired_source = False
+ if server_pid is not None and retirement_source is not None:
+ # Keep shutdown state so a half-started JVM that ignores SIGTERM
is escalated.
+ self._retire(retirement_source, server_pid, process_start_id)
Review Comment:
Fixed in a98fa7722f3: on self-abort, the current attendant kills its
pid-file server only after confirming the server remains in its own process
group. Removed the escalation claim for pid-only retirements and added a
regression.
##########
python/pyspark/sql/connect/local_server.py:
##########
@@ -132,7 +128,17 @@ def _is_local_connect_server(pid: int) -> Optional[bool]:
)
except (OSError, subprocess.SubprocessError):
return None
- return result.returncode == 0 and _SERVER_CLASS in result.stdout
+ return result.stdout if result.returncode == 0 else ""
Review Comment:
Fixed in a98fa7722f3: a non-zero ps exit now reports unknown while the PID
is alive. Pending recovery rechecks liveness after inspection and takes the
dead-attendant path if it exited; tests cover both.
##########
python/pyspark/sql/connect/local_server_pool.py:
##########
@@ -680,7 +814,89 @@ def reap(self, uid: str) -> bool:
states = self._directory.states(uid)
if had_retired and "retired" in states:
self._reap_retired(uid, states["retired"])
- return not self._directory.states(uid)
+
+ remaining = self._directory.states(uid)
+ if set(remaining) == {"member"}:
+ # Nothing references the member directory anymore. The age gate
keeps the logs
+ # of a freshly failed launch around long enough to be looked at.
+ try:
+ expired = (
+ time.time() - os.path.getmtime(remaining["member"])
+ > self._MEMBER_DIR_GC_AGE_SECONDS
+ )
+ except FileNotFoundError:
+ expired = True
+ if expired:
+ self._directory.remove_member_dir(uid)
+ remaining = self._directory.states(uid)
+ return not remaining
+
+ def _reap_pending(self, uid: str, path: str) -> None:
+ """A launch whose attendant died or hung: kill the attendant and
whatever server
+ spark-daemon.sh may have recorded for it, and withdraw the launch's
bookkeeping so
+ refills stop counting it."""
+ data = self._directory.read_json(path)
+ pending = PendingState.from_data(data)
+ parsed_pid = pending.attendant_pid if pending is not None else None
+ created = pending.created if pending is not None else None
+ if pending is None and data is not None:
+ # Preserve independently valid lifecycle fields when another field
is corrupt.
+ parsed_pid = PendingState.attendant_pid_from_data(data)
+ created = PendingState.created_from_data(data)
+ age = time.time() - created if created is not None else
self._LAUNCH_TIMEOUT_SECONDS + 1
+ attendant_pid = parsed_pid if parsed_pid is not None else -1
+ attendant_alive = _pid_alive(attendant_pid)
+ if not attendant_alive:
+ # Recovered fields from malformed state guide cleanup, but cannot
authorize a
+ # process-group signal because the complete pending record was not
validated.
+ if pending is not None and not self._signal_attendant_group(
+ attendant_pid, signal.SIGKILL, leader_may_be_dead=True
+ ):
+ # Keep the only launch-group handle when signalling failed but
descendants
+ # remain. If the pid was reused by a live process, withdraw
the stale record
+ # without signalling it.
+ if not _pid_alive(attendant_pid) and
self._attendant_group_alive(attendant_pid):
+ return
+ self.abort_launch(uid)
+ elif age > self._LAUNCH_TIMEOUT_SECONDS:
+ is_attendant = self._is_pool_attendant(attendant_pid, uid)
+ if is_attendant is None:
+ return
+ if is_attendant and not
self._signal_attendant_group(attendant_pid, signal.SIGKILL):
+ # Keep the record when an attendant that still appears live
could not be
+ # stopped, or when its leader exited during the attempt but
children remain.
+ if _pid_alive(attendant_pid) or
self._attendant_group_alive(attendant_pid):
+ return
Review Comment:
Fixed in a98fa7722f3: pending keep paths now have give-up deadlines, and a
verified nonleader attendant gets a direct SIGKILL fallback. Added
failed-signal and nonleader regressions.
--
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]
