Skip to content

Commit 2fbb384

Browse files
committed
fix(bigtable): only sweep a tempdir when its spawner is provably gone
Change-Id: I2dcd04d457e90e628ebd17d69987174a58489379
1 parent d6db158 commit 2fbb384

2 files changed

Lines changed: 554 additions & 137 deletions

File tree

‎packages/google-cloud-bigtable/google/cloud/bigtable/data/_accelerator/_daemon.py‎

Lines changed: 185 additions & 87 deletions
Original file line numberDiff line numberDiff line change
@@ -64,10 +64,12 @@
6464
# different BIGTABLE_ACCELERATOR_STARTUP_TIMEOUT, so the sweep cannot rely on
6565
# its own env value when judging another process's tempdir.
6666
_STARTUP_DEADLINE_FILENAME = "startup_deadline"
67-
# PID of the Python process that spawned the daemon, written at start(). The
68-
# sweep sends signal 0 to this PID to check liveness: if the spawner is alive
69-
# the daemon is alive; if the spawner is dead (ESRCH) the daemon's stdin is
70-
# closed and it will exit (or has already exited).
67+
# Identity of the Python process that spawned the daemon, written at start()
68+
# as ``"<pid> <pid-namespace>"``. The sweep signals this PID to decide whether
69+
# the spawner is still alive: if it is, the daemon's stdin is open and the
70+
# daemon is (or is about to be) serving; if it is gone, the daemon's stdin is
71+
# closed and it has exited or is exiting. See ``_sweep_stale_tempdirs`` for
72+
# why the namespace is recorded next to the number.
7173
_PID_FILENAME = "daemon.pid"
7274

7375
# How long to wait for the daemon to start listening on its UDS before giving
@@ -84,11 +86,23 @@
8486
_SIGTERM_GRACE_SECONDS = 2.0
8587
_SIGKILL_GRACE_SECONDS = 2.0
8688

87-
# A freshly-created tempdir has not written its daemon.pid yet. Never sweep one
88-
# younger than this, so a daemon another client is mid-way through starting is
89-
# never mistaken for a stale leftover and reaped out from under it.
89+
# Grace period past a tempdir's startup deadline before the sweep will judge
90+
# it. A daemon that has not bound by its own deadline has failed, but give it
91+
# this much slack anyway so a daemon another client is mid-way through
92+
# starting is never mistaken for a stale leftover and reaped out from under it.
9093
_STALE_SWEEP_MIN_AGE_SECONDS = 60.0
9194

95+
# Written in place of the PID namespace on platforms that do not have one to
96+
# report (no /proc, e.g. macOS). Comparing "none" against "none" is a match --
97+
# where namespaces do not exist, PIDs cannot be ambiguous between them.
98+
_NO_PID_NAMESPACE = "-"
99+
100+
# How long a single connect() probe of a daemon's socket may take. A UDS
101+
# connect against a listening socket completes in the kernel without the
102+
# daemon doing anything, so this only ever elapses on a badly overloaded
103+
# host — where the honest answer is "I cannot tell", not "it is dead".
104+
_SOCKET_PROBE_TIMEOUT_SECONDS = 0.25
105+
92106
# Upper bound on how long a single sweep may spend walking candidates. The
93107
# sweep runs on a fire-and-forget thread, so nothing is waiting on it; this
94108
# only stops a pathological temp directory (thousands of leftovers, a slow
@@ -135,82 +149,148 @@ def _resolve_startup_timeout() -> float:
135149

136150

137151
def _socket_is_live(uds_path: str) -> bool:
138-
"""Return True if something is currently accepting connections on the UDS."""
152+
"""Return True if something is currently accepting connections on the UDS.
153+
154+
Used by ``_wait_until_ready`` to poll for the daemon coming up, so every
155+
failure simply answers "not yet" and the caller tries again. Note that
156+
this is a *measurement*, which is why the sweep does not use it to decide
157+
whether a daemon has died -- see ``_sweep_stale_tempdirs``.
158+
"""
139159
if not os.path.exists(uds_path):
140160
return False
141161
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as probe:
142-
probe.settimeout(0.25)
162+
probe.settimeout(_SOCKET_PROBE_TIMEOUT_SECONDS)
143163
try:
144164
probe.connect(uds_path)
145165
return True
146166
except OSError:
147167
return False
148168

149169

170+
def _pid_namespace() -> str:
171+
"""Identify the PID namespace this process's PIDs are numbered in.
172+
173+
A PID is only meaningful inside the namespace that issued it, but the
174+
tempdir holding it can be shared across namespaces -- two containers with
175+
the same ``/tmp`` volume, or a host ``/tmp`` bind-mounted into one. The
176+
sweep records this alongside the PID so it can tell whether a number it
177+
reads is one it is entitled to interpret.
178+
179+
Returns ``_NO_PID_NAMESPACE`` where there is nothing to report, which
180+
compares equal to itself: on a platform with no PID namespaces, PIDs
181+
cannot be ambiguous between them.
182+
"""
183+
try:
184+
st = os.stat("/proc/self/ns/pid")
185+
except OSError:
186+
return _NO_PID_NAMESPACE
187+
return f"{st.st_dev}:{st.st_ino}"
188+
189+
190+
def _write_metadata_file(path: str, contents: str) -> None:
191+
"""Write ``contents`` to ``path`` so a reader never sees a partial file.
192+
193+
The sweep reads these files out of *other* processes' tempdirs, possibly
194+
while they are still being written. A half-written pid file would be
195+
unreadable or, worse, parse as some other process's PID, and a truncated
196+
deadline could parse as a plausible but wrong time. Writing to a sibling
197+
temp name and renaming into place makes the swap atomic: a reader sees
198+
either no file at all or the complete contents. A leftover ``.tmp`` from
199+
a process killed mid-write goes away with the tempdir.
200+
"""
201+
tmp_path = path + ".tmp"
202+
with open(tmp_path, "w") as _f:
203+
_f.write(contents)
204+
os.replace(tmp_path, path)
205+
206+
150207
def _sweep_stale_tempdirs() -> None:
151208
"""Best-effort reaping of leftover daemon tempdirs at startup.
152209
153-
A clean ``close()`` removes its own tempdir, but a crash or ``SIGKILL``
154-
leaves the socket and log behind. We reap those on the next start.
155-
156-
Rather than treating any ``bt-accel-*`` directory as ours, we key off the
157-
``daemon.log`` marker file we write inside each one: a directory is a
158-
candidate only if it actually contains that log. This avoids deleting an
159-
unrelated directory that merely shares the tempdir prefix. A candidate is
160-
removed only when the spawning Python process (tracked via ``daemon.pid``)
161-
is no longer alive *and* the tempdir is old enough that a daemon still
162-
mid-startup would already have bound or died. ``os.kill(pid, 0)`` is
163-
instant and kernel-guaranteed: the lock is implicitly released the moment
164-
the process exits for any reason, including SIGKILL.
165-
166-
All failures are swallowed — sweeping is opportunistic and must never break
167-
startup. The walk also gives up after ``_SWEEP_BUDGET_SECONDS`` so the
168-
thread cannot linger on a pathological temp directory; whatever it did not
169-
reach is simply left for the next start().
210+
A clean ``close()`` removes its own tempdir; a crash or ``SIGKILL`` leaves
211+
the socket and log behind. We tidy those up on the next start.
212+
213+
Best-effort in both directions, and that is the whole design:
214+
215+
* It never breaks startup. Every failure is swallowed, the walk abandons
216+
whatever it has not reached once ``_SWEEP_BUDGET_SECONDS`` is up, and it
217+
runs on a thread nobody waits on.
218+
* It never deletes without proof. The only door to ``rmtree`` is
219+
``ProcessLookupError`` from ``os.kill(pid, 0)`` -- the kernel stating no
220+
such process exists. Everything else (unreadable file, truncated write,
221+
``EPERM``, a PID from another namespace) means "could not establish
222+
death", and skips.
223+
224+
The asymmetry is deliberate: a missed leftover costs a few KB of /tmp
225+
until the next start() looks again, while deleting a live daemon's
226+
directory destroys the socket it is serving on with no way back. So two
227+
leaks are accepted on purpose -- a recycled PID reads as alive and pins
228+
its directory forever, and a pid file truncated mid-write is skipped
229+
forever. Both are cheap; neither is worth a heuristic that can delete a
230+
live daemon.
231+
232+
Mechanically: candidates are ``bt-accel-*`` directories containing our
233+
``daemon.log`` marker, so an unrelated directory sharing the prefix is
234+
never touched. The spawning Python process is the liveness signal --
235+
while it lives the daemon's stdin is open and the daemon is serving; when
236+
it dies the daemon sees EOF and exits. Each tempdir also records the PID
237+
namespace that numbered it (see :func:`_pid_namespace`), since a shared
238+
``/tmp`` can cross namespaces and a foreign PID would otherwise give a
239+
confident wrong answer; and its own startup deadline, which covers the
240+
window before the pid file exists and keeps the sweep correct whatever
241+
``BIGTABLE_ACCELERATOR_STARTUP_TIMEOUT`` the sweeping process has.
170242
"""
171243
if not hasattr(socket, "AF_UNIX"):
172244
return
173-
root = tempfile.gettempdir()
174-
try:
175-
logs = glob.glob(os.path.join(root, _TEMPDIR_PREFIX + "*", _LOG_FILENAME))
176-
except OSError:
177-
return
245+
# Start the clock before the glob: on the pathological temp directory this
246+
# budget exists for, enumerating it is itself the slow part.
178247
now = time.time()
179248
budget_ends = now + _SWEEP_BUDGET_SECONDS
180-
for log_path in logs:
181-
if time.time() >= budget_ends:
182-
return
183-
path = os.path.dirname(log_path)
184-
try:
185-
# Each tempdir records its own startup deadline so this sweep is
186-
# correct even when the sweeping process has a different
187-
# BIGTABLE_ACCELERATOR_STARTUP_TIMEOUT than the process that
188-
# started the daemon being evaluated.
189-
deadline_path = os.path.join(path, _STARTUP_DEADLINE_FILENAME)
190-
try:
191-
with open(deadline_path) as _f:
192-
deadline = float(_f.read())
193-
except (OSError, ValueError):
194-
# Older daemon without a deadline file: fall back to a simple
195-
# age heuristic relative to the log's mtime.
196-
deadline = os.path.getmtime(log_path) + _STALE_SWEEP_MIN_AGE_SECONDS
197-
if now < deadline + _STALE_SWEEP_MIN_AGE_SECONDS:
198-
continue
199-
pid_path = os.path.join(path, _PID_FILENAME)
249+
own_namespace = _pid_namespace()
250+
root = tempfile.gettempdir()
251+
try:
252+
logs = glob.iglob(os.path.join(root, _TEMPDIR_PREFIX + "*", _LOG_FILENAME))
253+
for log_path in logs:
254+
if time.time() >= budget_ends:
255+
return
256+
path = os.path.dirname(log_path)
200257
try:
201-
with open(pid_path) as _f:
202-
pid = int(_f.read().strip())
203-
os.kill(pid, 0)
204-
continue # spawning process alive → daemon alive → skip
205-
except ProcessLookupError:
206-
pass # ESRCH: spawner is dead → stdin closed → daemon exiting
207-
except (OSError, ValueError):
208-
# No PID file (tempdir predates this change) or EPERM (process
209-
# exists but unreadable) → assume alive and skip.
258+
deadline_path = os.path.join(path, _STARTUP_DEADLINE_FILENAME)
259+
try:
260+
with open(deadline_path) as _f:
261+
deadline = float(_f.read())
262+
except (OSError, ValueError):
263+
# Older tempdir without a deadline file, or one we could
264+
# not read: fall back to a simple age heuristic relative
265+
# to the log's mtime. This gate only decides *when* to
266+
# look; the pid check below still has to prove death.
267+
deadline = os.path.getmtime(log_path) + _STALE_SWEEP_MIN_AGE_SECONDS
268+
if now < deadline + _STALE_SWEEP_MIN_AGE_SECONDS:
269+
continue
270+
pid_path = os.path.join(path, _PID_FILENAME)
271+
try:
272+
with open(pid_path) as _f:
273+
fields = _f.read().split()
274+
pid = int(fields[0])
275+
if fields[1] != own_namespace:
276+
# Someone else's numbering. IndexError lands in the
277+
# same place: a tempdir written before the namespace
278+
# was recorded cannot be checked either.
279+
continue
280+
os.kill(pid, 0)
281+
continue # spawner alive → daemon alive → skip
282+
except ProcessLookupError:
283+
pass # ESRCH: spawner is dead → stdin closed → daemon gone
284+
except (OSError, ValueError, IndexError):
285+
# Missing, unreadable, empty or truncated pid file, or
286+
# EPERM from os.kill (the PID exists, it just is not ours
287+
# to signal). None of these is proof of death.
288+
continue
289+
except OSError:
210290
continue
211-
except OSError:
212-
continue
213-
shutil.rmtree(path, ignore_errors=True)
291+
shutil.rmtree(path, ignore_errors=True)
292+
except OSError:
293+
return
214294

215295

216296
class AcceleratorDaemon:
@@ -326,30 +406,15 @@ def start(self) -> None:
326406
# scales with the number of leftovers, and start() (which the async
327407
# client calls synchronously, on the event loop thread) must not wait
328408
# on it. The sweep bounds its own runtime; see _SWEEP_BUDGET_SECONDS.
329-
threading.Thread(target=_sweep_stale_tempdirs, daemon=True).start()
409+
# Even spawning the thread is best-effort: a process already at its
410+
# thread ceiling should still get a working client, not an exception
411+
# from the one step of startup that exists purely to tidy up /tmp.
412+
try:
413+
threading.Thread(target=_sweep_stale_tempdirs, daemon=True).start()
414+
except RuntimeError:
415+
pass
330416
self._tempdir = tempfile.mkdtemp(prefix=_TEMPDIR_PREFIX)
331417
self._uds_path = os.path.join(self._tempdir, _SOCKET_FILENAME)
332-
# Record the metadata the sweep needs to judge this tempdir later: the
333-
# absolute time by which startup must complete (read per-directory, so
334-
# the sweep is right regardless of what
335-
# BIGTABLE_ACCELERATOR_STARTUP_TIMEOUT the *sweeping* process has) and
336-
# the spawning PID. A failure here means the tempdir is unwritable, so
337-
# the log file and daemon would fail next anyway: clean up and report
338-
# it the same way a failed spawn does, rather than leaving behind a
339-
# directory that has no daemon.log and so would never be swept.
340-
try:
341-
with open(
342-
os.path.join(self._tempdir, _STARTUP_DEADLINE_FILENAME), "w"
343-
) as _f:
344-
_f.write(repr(time.time() + self._startup_timeout))
345-
with open(os.path.join(self._tempdir, _PID_FILENAME), "w") as _f:
346-
_f.write(str(os.getpid()))
347-
except OSError as exc:
348-
tempdir = self._tempdir # _cleanup_tempdir() clears the attribute.
349-
self._cleanup_tempdir()
350-
raise RuntimeError(
351-
f"Failed to write accelerator daemon metadata to {tempdir}: {exc}"
352-
) from exc
353418

354419
# Redirect the daemon's stdout/stderr to a log file rather than
355420
# subprocess.PIPE. Nothing drains those pipes for the daemon's
@@ -363,13 +428,46 @@ def start(self) -> None:
363428
# The log file handle only needs to live long enough for Popen to dup
364429
# it into the child, so it stays local to start() rather than being an
365430
# attribute. Startup failures read the tail back from the path.
366-
argv = [self._binary_path, "--uds-path", self._uds_path, *self._cli_flags]
367431
log_file = None
432+
try:
433+
# Create the log before anything else can fail. It is the marker
434+
# the sweep keys on, so a crash from here on still leaves a
435+
# directory that a later sweep can recognise and reap.
436+
log_file = open(self._log_path, "wb")
437+
# Record the absolute time by which startup must complete. The
438+
# sweep reads this per-directory, so it judges correctly however
439+
# the *sweeping* process has BIGTABLE_ACCELERATOR_STARTUP_TIMEOUT
440+
# set. Written atomically, since a sweep in another process may
441+
# read it at any moment.
442+
_write_metadata_file(
443+
os.path.join(self._tempdir, _STARTUP_DEADLINE_FILENAME),
444+
repr(time.time() + self._startup_timeout),
445+
)
446+
# Record who owns this tempdir, as "<pid> <pid-namespace>". The
447+
# namespace qualifies the number: a sweep running where that PID
448+
# would mean something else entirely can see that and decline to
449+
# judge, rather than confidently signalling an unrelated process.
450+
_write_metadata_file(
451+
os.path.join(self._tempdir, _PID_FILENAME),
452+
f"{os.getpid()} {_pid_namespace()}",
453+
)
454+
except OSError as exc:
455+
if log_file is not None:
456+
try:
457+
log_file.close()
458+
except OSError:
459+
pass
460+
tempdir = self._tempdir # _cleanup_tempdir() clears the attribute.
461+
self._cleanup_tempdir()
462+
raise RuntimeError(
463+
f"Failed to write accelerator daemon metadata to {tempdir}: {exc}"
464+
) from exc
465+
466+
argv = [self._binary_path, "--uds-path", self._uds_path, *self._cli_flags]
368467
env = None
369468
if self._extra_env:
370469
env = {**os.environ, **self._extra_env}
371470
try:
372-
log_file = open(self._log_path, "wb")
373471
self._proc = subprocess.Popen(
374472
argv,
375473
stdin=subprocess.PIPE,

0 commit comments

Comments
 (0)