dtenedor commented on code in PR #58367:
URL: https://github.com/apache/spark/pull/58367#discussion_r4222721727
##########
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:
Thanks, the ownership check is the right fix, and the recycled-group case no
longer kills anything. One problem with the extra age condition, though:
`_reap_dead_attendant` only signals a verified group while `age <=
_LAUNCH_TIMEOUT_SECONDS`. In #57687 only an acquire runs the full janitor, so a
background refill whose attendant dies mid-boot is usually first seen after
that window. `abort_launch` then retires the server by pid only, and nothing
ever stops it. The previous head killed the group at any age.
The same condition makes the new hand-off from the timed-out branch a no-op:
that branch only runs when `age > _LAUNCH_TIMEOUT_SECONDS`, so if the attendant
exits during inspection, `_reap_dead_attendant` can only withdraw the record.
Since the ownership check already makes the kill safe at any age, I'd keep
the age limit only for groups that can't be verified. This test fails at this
head, because the launch child survives:
```python
def test_reap_kills_verified_launch_group_after_launch_timeout(self) -> None:
attendant, launch_child_pid = self._attendant_with_launch_child("fade")
self._write_daemon_pid("fade", launch_child_pid)
self._write_state(
self._directory.pending_path("fade"),
{
"attendant_pid": attendant.pid,
"created": time.time() - ServerPool._LAUNCH_TIMEOUT_SECONDS - 1,
"fingerprint": "fp",
},
)
attendant.kill()
self.assertTrue(_wait_proc_dead(attendant))
with self._directory:
self._pool.reap("fade")
self.assertTrue(_wait_pid_dead(launch_child_pid))
```
With the limit moved to the unverified path, this passes and the existing
tests still do.
##########
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:
Thanks for fixing the comment. The new self-abort kill doesn't fire with the
real command line, though. `_stop_own_launch` requires `sys.argv[1:]` to equal
`_attendant_command(...)[3:]` exactly, but #57687's spawner appends `--master`
and `--fingerprint`, and will still need to after it adopts the builder. If
`test_own_attendant_stops_server_without_a_start_id` patches `sys.argv` to that
tail plus `"--master", "local[*]", "--fingerprint", "fp"`, the server survives.
A prefix match would fix it, but an explicit flag from the attendant's own
abort path, e.g. `abort_launch(uid, own_launch=True)`, seems simpler than
inspecting `sys.argv`.
##########
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)
Review Comment:
Thanks, this works. Two small follow-ups: off Linux, every `_pid_alive` call
now runs `ps` with a 5-second timeout. That includes each candidate in
`claim()` while the lock is held, the 0.25-second poll in `_await_ready`, and
the reuse path, so it may be worth confining the zombie probe to the attendant
checks in `_reap_pending`. The `_pid_alive` docstring also still says only
Linux zombies count as terminated.
--
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]
