ericm-db commented on code in PR #58367:
URL: https://github.com/apache/spark/pull/58367#discussion_r4223834312
##########
python/pyspark/sql/connect/local_server_pool.py:
##########
@@ -680,7 +849,143 @@ 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:
+ """Stop a verified abandoned launch, then withdraw its bookkeeping. A
launch group
+ whose owner cannot be verified is kept only until the recovery
deadline."""
+ 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)
+ now = time.time()
+ if created is not None and created <= now:
+ age = now - created
+ else:
+ # Malformed or future timestamps are expired immediately, but the
file's age
+ # still advances their give-up clock if inspection keeps failing.
+ try:
+ file_age = now - os.path.getmtime(path)
+ except OSError:
+ file_age = self._RETIRE_GIVE_UP_AFTER_SECONDS + 1
+ if file_age < 0:
+ file_age = self._RETIRE_GIVE_UP_AFTER_SECONDS + 1
+ age = self._LAUNCH_TIMEOUT_SECONDS + 1 + file_age
Review Comment:
Fixed in 56b3129b655: future created timestamps fall back to file mtime,
with a future mtime clamped to age zero. Added the backward-clock-step
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)
Review Comment:
Fixed in 56b3129b655: the macOS zombie ps probe is confined to attendant
liveness checks, and the liveness/docstring descriptions now match. General
server and reuse probes no longer run ps on each call.
##########
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 56b3129b655: self-abort now accepts the required attendant argv
prefix plus --master and --fingerprint. Added a regression with the real
trailing flags and a negative test for mismatched argv.
--
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]
