dtenedor commented on code in PR #58367:
URL: https://github.com/apache/spark/pull/58367#discussion_r4222723722
##########
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:
This makes a backward clock step kill a healthy launch, which the previous
head didn't do. A `created` in the future is treated as malformed, and the
fallback then treats the file's future mtime as very old (`file_age =
_RETIRE_GIVE_UP_AFTER_SECONDS + 1`). So a record written just before a small
step back looks 902 seconds old, and a live, verified attendant gets its whole
group killed. Could only a malformed `created` count as expired? For a future
one, fall back to the file's mtime, and treat a future mtime as age 0. This
test fails at this head:
```python
def test_reap_keeps_live_launch_after_backward_clock_step(self) -> None:
attendant, launch_child_pid = self._attendant_with_launch_child("c10c")
path = self._directory.pending_path("c10c")
future = time.time() + 5
self._write_state(
path, {"attendant_pid": attendant.pid, "created": future,
"fingerprint": "fp"}
)
os.utime(path, (future, future))
with self._directory:
self.assertFalse(self._pool.reap("c10c"))
self.assertIsNone(attendant.poll())
self.assertTrue(_pid_alive(launch_child_pid))
```
--
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]
