dtenedor commented on code in PR #58367:
URL: https://github.com/apache/spark/pull/58367#discussion_r4213003039
##########
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:
This can SIGKILL an unrelated process group. The invariant in the docstring
only holds while the original group never empties. Once the attendant and all
of its descendants are gone, the id can be reused by a process that becomes a
group leader and exits while its children keep running: the double-fork daemon
shape, or `setsid sh -c 'x &'`. The result is a leaderless group whose id
equals the recorded `attendant_pid`, so `getpgid` raises `ProcessLookupError`
and we reach `killpg`. Nothing bounds the record's age or boot, so unlike the
check-to-signal window, this state can persist indefinitely.
A stale record can easily outlive its pid. The janitor only runs on the next
acquire, and the pool lives under the temp dir: that survives a container
restart, where pids restart from 1, and on macOS `$TMPDIR` only expires files
after a few days, while pids wrap around 100k. When the janitor runs as root,
as is common in containers, any process is a candidate.
This test fails at this head, because the unrelated child is killed:
```python
def test_reap_does_not_signal_a_recycled_leaderless_group(self) -> None:
# A group leader that forks a long-lived child and exits; its group
outlives it.
leader = subprocess.Popen(
[
sys.executable,
"-c",
"import subprocess, sys\n"
"child = subprocess.Popen([sys.executable, '-c', 'import time;
time.sleep(60)'])\n"
"print(child.pid, flush=True)",
],
stdout=subprocess.PIPE,
text=True,
start_new_session=True,
)
self._launch_groups.append(leader.pid)
assert leader.stdout is not None
unrelated_child = int(leader.stdout.readline())
leader.wait()
# A stale but complete record whose attendant pid now names that
unrelated group.
self._write_state(
self._directory.pending_path("aaaa"),
{
"attendant_pid": leader.pid,
"created": time.time() - 7 * 24 * 3600,
"fingerprint": "fp",
},
)
with self._directory:
self._pool.reap("aaaa")
self.assertFalse(_wait_pid_dead(unrelated_child, timeout=2.0))
```
One option is to require positive evidence before signalling a leaderless
group: for example, that the server pid spark-daemon.sh recorded in
`member-<uid>` is alive, that `os.getpgid(daemon_pid) == attendant_pid`, and
that `_is_local_connect_server(daemon_pid)` is true. Without that evidence,
keep the record for at most the launch timeout (the pid file may still appear),
then withdraw it without signalling, as the malformed-record path already does.
A cheaper mitigation is to persist the attendant's `_process_start_id` (which
includes the boot id on Linux) in the pending record, and only signal a
leaderless group for current-boot records younger than a small bound.
##########
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:
Neither keep-the-record path has a give-up bound. If the attendant passes
`_is_pool_attendant` but isn't a group leader, `_signal_attendant_group`
returns `False`, and this keeps the record for as long as the attendant lives.
A hung attendant then holds a slot in #57687's `refill` count forever. The dead
branch above does the same while `killpg` gets EPERM, for example from another
user's group that reused the id. Falling back to `os.kill(attendant_pid,
SIGKILL)` for a verified non-leader, plus a give-up age like
`_RETIRE_GIVE_UP_AFTER_SECONDS`, would bound both.
##########
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:
This maps every non-zero `ps` exit to `""` ("gone"), not just the exit for a
missing pid. That's harmless for `_is_local_connect_server`, but
`_is_pool_attendant` turns it into `False`, and the timed-out branch of
`_reap_pending` then withdraws a live attendant's record without stopping it.
The attendant keeps running and later publishes into the withdrawn launch,
since #57687's `_boot` doesn't re-check the pending marker. Meanwhile `refill`
no longer counts it and spawns a replacement. The same branch also drops the
only group handle if the attendant exits between `_pid_alive` and `ps`.
Could this return `None` when `ps` fails but the pid is still alive? On
Linux, reading `/proc/<pid>/cmdline` would also give the exact argv without
splitting on whitespace. Then in `_reap_pending`, "not an attendant, and now
dead" could fall through to the dead-attendant handling instead of
`abort_launch`.
##########
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:
Following up on the "daemon-pid-only server can linger" note from my first
review: with only a daemon pid, `process_start_id` is `None`, so `_retire`
sends no SIGTERM, and `_reap_retired` keeps the pid until the process exits on
its own, without ever signalling it. (`test_reap_pending_of_dead_attendant`
pins this no-identity retirement.) So nothing here is escalated, and this
comment overstates what happens.
I said then that the group kill normally covers this case, but that kill
doesn't run when the attendant aborts its own launch, which is what
`MemberAttendant._boot` does in #57687. For example, if `_process_start_id`
returns `None` for a healthy, just-launched server (say `ps` fails or times out
on macOS), `_boot` raises and `abort_launch` records a pid-only retirement. The
attendant then exits, and the JVM keeps running with nothing left to stop it.
At least the comment should change. Better, the self-abort path could kill the
server it just started before withdrawing the launch.
##########
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:
Off Linux, `_pid_alive` counts zombies as alive, so an attendant that died
but hasn't been reaped by its spawning client never takes the dead-attendant
branch. After the timeout, `_is_pool_attendant` can't read a zombie's command
line, so the record is withdrawn without killing the launch group. This is
probably best handled where the attendant is spawned, by reaping it or
double-forking, but it's worth noting here.
##########
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:
This retention doesn't cover the most common launch failure. Once
spark-daemon.sh has written its pid file, `abort_launch` retires that pid even
if the server has already exited. On the next pass, `_reap_retired` finds it
dead and `_remove_retired` deletes `member-<uid>/`, including the server logs
and `attendant.log`. That's exactly the case where `_await_ready` reports "the
server exited during start-up; see logs under ...".
With a pending record and a pid file naming a dead pid, the first `reap`
leaves `{member, retired}` and the second leaves nothing. Only launches that
die before the pid file exists get the 24 hours. Maybe skip the retirement in
`abort_launch` when the recorded pid is already dead, and mark retirements that
come from aborted launches so `_remove_retired` leaves the directory to this
rule?
##########
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
_DEFAULT_IDLE_TIMEOUT_SECONDS = 1800
_PROCESS_INSPECTION_TIMEOUT_SECONDS = 5
_PROC_STAT_START_TIME_INDEX = 19
+ _ATTENDANT_MODULE = "pyspark.sql.connect.local_server_pool"
Review Comment:
Re the CLI-contract coupling from my last review: this constant helps, but
#57687's `MemberAttendant.spawn` still hard-codes the module string and builds
the argv independently, and the test fakes are a third copy. A drift such as
`-I -m` or `--uid=<uid>` would silently disable hung-launch cleanup. With the
`ps` behavior noted on `_process_command`, it would also withdraw the records
of attendants that keep running. Could this layer add one builder, e.g.
`_attendant_command(pool_dir, uid, ...)`, that the spawner and the tests both
use, plus a test that `_is_pool_attendant` accepts its output?
--
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]
