ericm-db commented on code in PR #58367:
URL: https://github.com/apache/spark/pull/58367#discussion_r4213149907
##########
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:
Added the shared attendant argv builder in a98fa7722f3, and the test fakes
now use it with a parser regression. The #57687 spawner needs to adopt this
builder when that stacked PR is rebased.
##########
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 a98fa7722f3: off-Linux liveness probes now check ps state and treat
zombies as dead. Added a macOS-path regression.
--
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]
