Something went wrong. Try again.
small gleam coding and persistent agent daemon with a detachable cli and a canvas ui forked from okami.mom/albedo
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495"""Conservative release eligibility; unknown process resources keep the kernel."""
from __future__ import annotations
import asynciofrom collections.abc import Callableimport hashlibimport osimport threading
def environment_hash() -> bytes: digest = hashlib.sha256() for key, value in sorted(os.environ.items()): digest.update(os.fsencode(key)) digest.update(b"\0") digest.update(os.fsencode(value)) digest.update(b"\0") return digest.digest()
def descriptors() -> set[tuple[int, int, int, int]] | None: for directory in ("/proc/self/fd", "/dev/fd"): try: names = os.listdir(directory) except OSError: continue result = set() for name in names: if not name.isdecimal(): continue fd = int(name) try: stat = os.fstat(fd) except OSError: continue # listdir's own descriptor already closed result.add((fd, stat.st_dev, stat.st_ino, stat.st_mode)) return result return None
class Quiescence: def __init__( self, loop: asyncio.AbstractEventLoop, owns: list[Callable[[], bool]], internal: Callable[[], tuple[set[threading.Thread], set[int]]], monitor: Callable[[], None], ) -> None: self.loop = loop self.owns = owns self.internal = internal self.monitor = monitor # the periodic sampler's own callback self.environment = environment_hash() self.threads = set(threading.enumerate()) initial = descriptors() _, internal_fds = internal() self.descriptors = ( None if initial is None else {fd for fd in initial if fd[0] not in internal_fds} ) self.tasks = asyncio.all_tasks(loop) self.timers = set(getattr(loop, "_scheduled", ()))
def ready(self) -> bool: try: if environment_hash() != self.environment: return False internal_threads, internal_fds = self.internal() if set(threading.enumerate()) - self.threads - internal_threads: return False if asyncio.all_tasks(self.loop) - self.tasks: return False timers = getattr(self.loop, "_scheduled", None) if timers is None or any( timer not in self.timers and not timer.cancelled() and getattr(getattr(timer, "_callback", None), "__name__", "") not in ("refresh_work", "recheck_work") for timer in timers ): return False current = descriptors() if ( current is None or self.descriptors is None or {fd for fd in current if fd[0] not in internal_fds} - self.descriptors ): return False return not any(check() for check in self.owns) except Exception: return False