155 lines
6.6 KiB
Python
155 lines
6.6 KiB
Python
"""Invocation cgroups inside a dedicated, delegated USER service.
|
|
|
|
This owns ordinary descendants, including setsid/double forks. It is deliberately
|
|
not a sandbox against the runtime UID: that UID can ask its user manager to own
|
|
persistent services/Podman scopes. Those are outside invocation cancellation.
|
|
Never point this class at the manager/authority groups or the whole user slice.
|
|
"""
|
|
import os
|
|
import resource
|
|
from pathlib import Path
|
|
import uuid
|
|
|
|
|
|
def require(value, message):
|
|
if not value:
|
|
raise ValueError(message)
|
|
|
|
|
|
def current_group(pid='self'):
|
|
rows = Path(f'/proc/{pid}/cgroup').read_text().splitlines()
|
|
require(len(rows) == 1 and rows[0].startswith('0::/'), 'unified cgroup required')
|
|
return Path('/sys/fs/cgroup') / rows[0][4:]
|
|
|
|
|
|
class Invocation:
|
|
RETIRE_STEPS = 64
|
|
RETIRE_DEPTH = 64
|
|
# One process-wide traversal, not one depth-sized FD stack per job.
|
|
walker = None
|
|
|
|
def __init__(self, path):
|
|
self.path = path
|
|
# Pin this reservation, never traverse a replacement pathname or sibling.
|
|
self.fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC)
|
|
self.walk = []
|
|
|
|
def enter(self):
|
|
self.write_control('cgroup.procs', str(os.getpid()))
|
|
require(current_group() == self.path, 'invocation cgroup entry failed')
|
|
|
|
def write_control(self, name, value):
|
|
fd = os.open(name, os.O_WRONLY | os.O_NOFOLLOW | os.O_CLOEXEC, dir_fd=self.fd)
|
|
with os.fdopen(fd, 'w') as stream:
|
|
stream.write(value)
|
|
|
|
def kill(self):
|
|
self.write_control('cgroup.kill', '1')
|
|
|
|
def empty(self):
|
|
fd = os.open('cgroup.events', os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC, dir_fd=self.fd)
|
|
with os.fdopen(fd) as stream:
|
|
events = dict(row.split() for row in stream.read().splitlines())
|
|
return events['populated'] == '0'
|
|
|
|
def release_walk(self):
|
|
"""Allocation-free pressure recovery; retain the reservation itself."""
|
|
for fd, name, entries in self.walk:
|
|
entries.close()
|
|
if name is not None:
|
|
os.close(fd)
|
|
self.walk.clear()
|
|
if Invocation.walker is self:
|
|
Invocation.walker = None
|
|
|
|
def close(self):
|
|
"""Bounded, resumable postorder retirement of this empty subtree only.
|
|
|
|
False means more ticks are required. Errors retain the root capability
|
|
and reset the traversal for retry; no populated/unreadable tree is lost.
|
|
cgroupfs has no symlinks, but dirfd/no-follow also prevents following one
|
|
in a malformed fixture or an unexpectedly replaced reservation.
|
|
"""
|
|
try:
|
|
require(self.empty(), 'invocation retirement unresolved')
|
|
if Invocation.walker not in (None, self):
|
|
return False
|
|
Invocation.walker = self
|
|
if not self.walk:
|
|
# scandir(fd) duplicates the descriptor but shares its directory
|
|
# offset. A failed traversal must start over, not skip a subtree.
|
|
os.lseek(self.fd, 0, os.SEEK_SET)
|
|
self.walk.append((self.fd, None, os.scandir(self.fd)))
|
|
for _ in range(self.RETIRE_STEPS):
|
|
fd, name, entries = self.walk[-1]
|
|
entry = next(entries, None)
|
|
if entry is not None:
|
|
if entry.is_dir(follow_symlinks=False):
|
|
require(len(self.walk) < self.RETIRE_DEPTH, 'invocation retirement depth bound')
|
|
child = os.open(entry.name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC,
|
|
dir_fd=fd)
|
|
try:
|
|
scan = os.scandir(child)
|
|
except BaseException:
|
|
os.close(child)
|
|
raise
|
|
self.walk.append((child, entry.name, scan))
|
|
continue
|
|
entries.close()
|
|
if name is None:
|
|
before, after = os.fstat(self.fd), self.path.lstat()
|
|
require((before.st_dev, before.st_ino) == (after.st_dev, after.st_ino),
|
|
'invocation reservation replaced')
|
|
self.path.rmdir()
|
|
self.walk.pop()
|
|
os.close(self.fd)
|
|
self.fd = None
|
|
Invocation.walker = None
|
|
return True
|
|
# rmdir is the kernel's final empty/no-children barrier. Do not
|
|
# unlink files, recurse outside this dirfd, or use rmtree.
|
|
parent = self.walk[-2][0]
|
|
before, after = os.fstat(fd), os.stat(name, dir_fd=parent, follow_symlinks=False)
|
|
require((before.st_dev, before.st_ino) == (after.st_dev, after.st_ino),
|
|
'invocation descendant replaced')
|
|
os.rmdir(name, dir_fd=parent)
|
|
self.walk.pop()
|
|
os.close(fd)
|
|
return False
|
|
except (OSError, ValueError):
|
|
self.release_walk()
|
|
raise
|
|
|
|
|
|
class Groups:
|
|
def __init__(self):
|
|
# At most 32 jobs (root + connection), 32 pending admissions (four
|
|
# rights each), one 128-FD walk, and 128 spare observation/control FDs.
|
|
# Bind this reserve to the actual inherited soft limit; do not raise it.
|
|
soft, _ = resource.getrlimit(resource.RLIMIT_NOFILE)
|
|
require(soft == resource.RLIM_INFINITY or
|
|
soft >= len(os.listdir('/proc/self/fd')) + 512,
|
|
'runtime descriptor reserve unavailable')
|
|
self.root = current_group()
|
|
require(self.root.name == 'archipelago-runtime-worker.service',
|
|
'worker must run in its dedicated USER unit')
|
|
require('/user' in str(self.root) or 'runtime-user' in str(self.root),
|
|
'runtime user hierarchy required')
|
|
require((self.root / 'cgroup.kill').is_file(), 'cgroup.kill unavailable')
|
|
# No adoption of a previous worker's jobs after crash. systemd must first
|
|
# retire its service cgroup. Refuse stale leaves, including empty ones.
|
|
require(not any(p.is_dir() for p in self.root.iterdir()), 'prior worker groups remain')
|
|
control = self.root / 'owner'
|
|
control.mkdir()
|
|
(control / 'cgroup.procs').write_text(str(os.getpid()))
|
|
self.control = control
|
|
|
|
def reserve(self):
|
|
path = self.root / ('job-' + uuid.uuid4().hex)
|
|
path.mkdir()
|
|
try:
|
|
return Invocation(path)
|
|
except BaseException:
|
|
path.rmdir() # No child was launched into this empty reservation.
|
|
raise
|