From 02c1faad809a3dbb9ca9e785dc5926e348e4d8ba Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 16:46:17 -0400 Subject: [PATCH] Name lock holders once when store acquisition waits --- src/agent_comms/active_route.py | 69 ++------------------------------ src/agent_comms/child_process.py | 65 ++++++++++++++++++++++++++++++ src/agent_comms/store_files.py | 7 ++++ 3 files changed, 75 insertions(+), 66 deletions(-) diff --git a/src/agent_comms/active_route.py b/src/agent_comms/active_route.py index 25a823214a..1584aec855 100644 --- a/src/agent_comms/active_route.py +++ b/src/agent_comms/active_route.py @@ -11,14 +11,14 @@ import os import time import uuid -from abc import ABC, abstractmethod +from abc import ABC from collections.abc import Iterator from contextlib import contextmanager from dataclasses import dataclass, replace from pathlib import Path from typing import TYPE_CHECKING, Annotated, ClassVar -from .child_process import Platform, ProcessIdentity +from .child_process import Platform from .coordination_errors import CoordinationError from .declared_family import DeclaredFamily from .errors import RelationViolationError @@ -264,69 +264,6 @@ def guard_default_route_write( os.close(directory) -class RouteLockHolder(DeclaredFamily, affix="LockHolder"): - """An observed Linux flock owner; diagnosis, not authority to signal it.""" - - @abstractmethod - def describe(self) -> str: ... - - -@dataclass(frozen=True, slots=True) -class IdentifiedLockHolder(RouteLockHolder): - pid: int - - def describe(self) -> str: - return f"PID {self.pid}: {self.detail()}" - - @abstractmethod - def detail(self) -> str: ... - - -@dataclass(frozen=True, slots=True) -class RunningLockHolder(IdentifiedLockHolder): - command: str - - def detail(self) -> str: - return self.command - - -class ExitedLockHolder(IdentifiedLockHolder): - def detail(self) -> str: - return "process exited" - - -class UnidentifiedLockHolder(RouteLockHolder): - def describe(self) -> str: - return "kernel did not identify the owner" - - -def route_lock_holders(directory: int) -> tuple[RouteLockHolder, ...]: - """Identify actual flock owners of this opened directory, not merely open FDs. - - An OFD lock with no reported PID, inaccessible /proc, or a concurrent exit - is unknown; absence from this snapshot never proves sole ownership of the lock. - """ - info = os.fstat(directory) - identity = f"{os.major(info.st_dev):02x}:{os.minor(info.st_dev):02x}:{info.st_ino}" - holders = [] - with open('/proc/locks', encoding='ascii') as locks: - for line in locks: - fields = line.split() - if len(fields) < 8 or fields[1] != 'FLOCK' or fields[5] != identity: - continue - pid = int(fields[4]) - if pid < 0: - holders.append(UnidentifiedLockHolder()) - continue - try: - command = " ".join(ProcessIdentity.capture(pid).arguments()) - except ProcessLookupError: - holders.append(ExitedLockHolder(pid)) - else: - holders.append(RunningLockHolder(pid, command)) - return tuple(holders) - - class RouteReadersHold(CoordinationError): """A client still reads the default route; a route change never waits for it.""" @@ -336,7 +273,7 @@ def acquire_exclusive_route_lock(directory: int) -> None: try: fcntl.flock(directory, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError as error: - detail = '; '.join(holder.describe() for holder in route_lock_holders(directory)) + detail = '; '.join(holder.describe() for holder in Platform.current().lock_holders(directory)) raise RouteReadersHold( f'default-route readers hold the route: ' f'{detail or "unknown (lock changed during inspection)"}; close them and rerun' diff --git a/src/agent_comms/child_process.py b/src/agent_comms/child_process.py index 60300325cb..eae8eba27a 100644 --- a/src/agent_comms/child_process.py +++ b/src/agent_comms/child_process.py @@ -315,6 +315,42 @@ def members(self, platform: PosixPlatform) -> tuple[ProcessIdentity, ...]: return (self,) if platform.matches(self) else () +class LockHolder(DeclaredFamily, affix="LockHolder"): + """An observed flock owner; diagnosis, not authority to signal it.""" + + @abstractmethod + def describe(self) -> str: ... + + +@dataclass(frozen=True, slots=True) +class IdentifiedLockHolder(LockHolder): + pid: int + + def describe(self) -> str: + return f"PID {self.pid}: {self.detail()}" + + @abstractmethod + def detail(self) -> str: ... + + +@dataclass(frozen=True, slots=True) +class RunningLockHolder(IdentifiedLockHolder): + command: str + + def detail(self) -> str: + return self.command + + +class ExitedLockHolder(IdentifiedLockHolder): + def detail(self) -> str: + return "process exited" + + +class UnidentifiedLockHolder(LockHolder): + def describe(self) -> str: + return "kernel did not identify the owner" + + class PidfdHandles: @staticmethod def open_pidfd(pid: int) -> int: @@ -429,6 +465,9 @@ def arguments(self, process: ProcessIdentity) -> tuple[str, ...]: ... @abstractmethod def environment(self, process: ProcessIdentity) -> dict[str, str]: ... + def lock_holders(self, descriptor: int) -> tuple[LockHolder, ...]: + return (UnidentifiedLockHolder(),) + def parent_identity(self, pid: int) -> ProcessIdentity: raise NotImplementedError('This platform does not expose exact parent identities') @@ -592,6 +631,32 @@ def environment(self, process: ProcessIdentity) -> dict[str, str]: values = (item.split(b"=", 1) for item in raw.split(b"\0") if b"=" in item) return {os.fsdecode(key): os.fsdecode(value) for key, value in values} + def lock_holders(self, descriptor: int) -> tuple[LockHolder, ...]: + """Identify actual flock owners of this opened file, not merely open FDs. + + An OFD lock with no reported PID, inaccessible /proc, or a concurrent exit + is unknown; absence never proves sole ownership of the lock. + """ + info = os.fstat(descriptor) + identity = f"{os.major(info.st_dev):02x}:{os.minor(info.st_dev):02x}:{info.st_ino}" + holders = [] + with open('/proc/locks', encoding='ascii') as locks: + for line in locks: + fields = line.split() + if len(fields) < 8 or fields[1] != 'FLOCK' or fields[5] != identity: + continue + pid = int(fields[4]) + if pid < 0: + holders.append(UnidentifiedLockHolder()) + continue + try: + command = " ".join(self.arguments(self.identity(pid))) + except ProcessLookupError: + holders.append(ExitedLockHolder(pid)) + else: + holders.append(RunningLockHolder(pid, command)) + return tuple(holders) + def parent_identity(self, pid: int) -> ProcessIdentity: return self.identity(self._stat(pid).parent) diff --git a/src/agent_comms/store_files.py b/src/agent_comms/store_files.py index 70df0b98d1..d58fff504a 100644 --- a/src/agent_comms/store_files.py +++ b/src/agent_comms/store_files.py @@ -7,6 +7,7 @@ import math import os import stat +import sys import tempfile import threading import time @@ -80,11 +81,17 @@ class StoreLockContention: def waits(self, descriptor: int, platform: Platform, *, mode: LockMode) -> Iterator[float]: """One physical acquisition algorithm; drivers own sync/async waiting.""" + reported = False while True: begun = time.monotonic() try: platform.try_store_lock(descriptor, mode=mode) except BlockingIOError: + if not reported: + print("Waiting on a store lock held by " + "; ".join( + holder.describe() for holder in platform.lock_holders(descriptor)), + file=sys.stderr, flush=True) + reported = True self.remaining -= time.monotonic() - begun if self.remaining <= 0: raise