abkslmandClaude Fable 5
47ba27da34
582cf9eaf clamped every host child (agent CLIs, builds, tests, probes) to
.utility to protect main-thread responsiveness. Measured on a loaded M2 Max
(load avg ~30, sibling session builds running), that clamp makes an 8-way
parallel child workload 3.2x slower (~1.0s -> ~3.1s) and single-threaded
commands ~1.6x slower on average (up to 2.5x), with the child at kernel
priority 20 vs 31 — the same mechanism that pushed every GitRunner call past
its 120s wall-clock timeout. Default-tier children still lose to the
user-interactive main thread under contention, so the clamp buys no
responsiveness; .userInitiated measured no better than default, so no
per-command priority plumbing is warranted.
ProcessStallMonitor needs no threshold change: even a maximally starved
utility child still received 86% of one core (43x the 0.02 resource-quiet
threshold), so slow-but-honest work was never at risk of a false
"Command Stalled" verdict — the clamp's only real cost was throughput.
Co-Authored-By: Claude Fable 5 <[email protected]>
655 lines
32 KiB
Swift
655 lines
32 KiB
Swift
import Foundation
|
||
|
||
#if canImport(Darwin)
|
||
import Darwin
|
||
#elseif canImport(Musl)
|
||
import Musl
|
||
#elseif canImport(Glibc)
|
||
import Glibc
|
||
#endif
|
||
|
||
// MARK: - ProcessHost contract (ADAPTERS §0.1)
|
||
|
||
public enum StdinMode: Sendable {
|
||
case pipe
|
||
case closed
|
||
}
|
||
|
||
public struct ProcessSpec: Sendable {
|
||
public let executable: String
|
||
public let args: [String]
|
||
public let cwd: String
|
||
public let env: [String: String]
|
||
public let stdinMode: StdinMode
|
||
/// Keys to strip from the *inherited* app environment before the child launches. A launched
|
||
/// child normally inherits Nucleic's own process environment; this lets a caller guarantee a
|
||
/// credential the app happened to inherit (e.g. an ambient `ANTHROPIC_API_KEY`) never reaches a
|
||
/// subprocess. `env` still wins over the inherited value; removal applies after that merge.
|
||
public let removeFromEnvironment: Set<String>
|
||
|
||
public init(
|
||
executable: String, args: [String], cwd: String,
|
||
env: [String: String] = [:], stdinMode: StdinMode = .pipe,
|
||
removeFromEnvironment: Set<String> = []
|
||
) {
|
||
self.executable = executable
|
||
self.args = args
|
||
self.cwd = cwd
|
||
self.env = env
|
||
self.stdinMode = stdinMode
|
||
self.removeFromEnvironment = removeFromEnvironment
|
||
}
|
||
}
|
||
|
||
public protocol ProcessHandle: Sendable {
|
||
/// NDJSON-friendly: one line per element, newline stripped, partial trailing
|
||
/// bytes held across pipe reads.
|
||
var stdoutLines: AsyncThrowingStream<Data, Error> { get }
|
||
var stderrLines: AsyncThrowingStream<Data, Error> { get }
|
||
var processID: Int32 { get }
|
||
func writeLine(_ data: Data) throws
|
||
func closeStdin()
|
||
func sendSignal(_ sig: Int32)
|
||
func wait() async -> Int32
|
||
/// Finish the stdout/stderr line streams immediately, independent of the process's exit, so a
|
||
/// consumer blocked on `for await stdoutLines` is released. Only meaningful when the streams can
|
||
/// outlive a killed process (the containerized handle, whose stream closure rides a guest RPC
|
||
/// that can stall); the host pipe handle EOFs reliably on exit, so its default is a no-op.
|
||
func forceCloseStreams()
|
||
|
||
/// Monotonic uptime-nanoseconds (`DispatchTime`) of the most recent stdout/stderr chunk this
|
||
/// process produced — or of launch, before any output. Drives ``ProcessStallMonitor``: a handle
|
||
/// whose output has gone silent is a hang candidate. Default: "just now", so a handle that
|
||
/// doesn't track activity never looks stalled.
|
||
var lastActivityNanos: UInt64 { get }
|
||
|
||
/// Whether ``processID`` is a real host pid whose CPU time can be sampled from this machine (a
|
||
/// host ``ChildProcess``). False for guest/VM handles, whose `processID` is a guest pid or a
|
||
/// socket fd — for those ``ProcessStallMonitor`` falls back to the output-idle signal alone.
|
||
var supportsHostCPUSampling: Bool { get }
|
||
}
|
||
|
||
extension ProcessHandle {
|
||
/// Default: nothing to force. The host `ChildProcess` reads OS pipes that EOF the moment the
|
||
/// process dies, so the line streams always finish on their own. ``ContainerizedProcessHandle``
|
||
/// overrides this because its streams close only when a guest wait RPC resolves.
|
||
public func forceCloseStreams() {}
|
||
|
||
/// Default: treat the handle as perpetually active, so a handle that opts out of activity
|
||
/// tracking is never mistaken for a hang. The concrete host/container/VM handles override this.
|
||
public var lastActivityNanos: UInt64 { DispatchTime.now().uptimeNanoseconds }
|
||
|
||
/// Default: not host-CPU-samplable. Only the host ``ChildProcess`` overrides this to `true`.
|
||
public var supportsHostCPUSampling: Bool { false }
|
||
}
|
||
|
||
/// A thread-safe "last time output was seen" marker for one process, sampled by
|
||
/// ``ProcessStallMonitor`` to tell a busy-but-quiet command from a wedged one. Backed by
|
||
/// `DispatchTime` uptime nanos (monotonic — immune to wall-clock jumps). Bumped from the
|
||
/// stdout/stderr chunk sinks of each ``ProcessHandle`` implementation.
|
||
final class ActivityClock: @unchecked Sendable {
|
||
private let lock = NSLock()
|
||
private var nanos: UInt64 = DispatchTime.now().uptimeNanoseconds
|
||
func bump() {
|
||
let now = DispatchTime.now().uptimeNanoseconds
|
||
lock.lock(); nanos = now; lock.unlock()
|
||
}
|
||
var lastNanos: UInt64 { lock.lock(); defer { lock.unlock() }; return nanos }
|
||
}
|
||
|
||
/// One-shot claim guard: the first caller to `claim()` wins, the rest get `false`. Lets two racing
|
||
/// tasks (exit vs. timeout) resume a single `CheckedContinuation` exactly once.
|
||
private final class ResumeOnceFlag: @unchecked Sendable {
|
||
private let lock = NSLock()
|
||
private var claimed = false
|
||
func claim() -> Bool {
|
||
lock.lock(); defer { lock.unlock() }
|
||
if claimed { return false }
|
||
claimed = true
|
||
return true
|
||
}
|
||
}
|
||
|
||
extension ProcessHandle {
|
||
/// Await the process's exit, but give up after `nanos`. Returns `true` iff it exited in time.
|
||
/// On timeout the background wait is left running (harmless — it resolves once the process
|
||
/// finally dies, e.g. after a follow-up SIGKILL), so this never blocks longer than `nanos`.
|
||
/// Used to bound every Stop/shutdown so a wedged agent can't hang the actor on `wait()`.
|
||
func waitForExit(within nanos: UInt64) async -> Bool {
|
||
let once = ResumeOnceFlag()
|
||
return await withCheckedContinuation { (cont: CheckedContinuation<Bool, Never>) in
|
||
Task { _ = await self.wait(); if once.claim() { cont.resume(returning: true) } }
|
||
Task {
|
||
try? await Task.sleep(nanoseconds: nanos)
|
||
if once.claim() { cont.resume(returning: false) }
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Stop the process for good and return once it has actually exited: SIGTERM, then — if it
|
||
/// hasn't gone within `graceNanos` — SIGKILL. For a containerized handle the signals reach the
|
||
/// agent's whole process group. Bounding the wait is what lets a stopped/ended session settle
|
||
/// instead of spinning forever on an agent that ignores the polite signal.
|
||
func terminate(graceNanos: UInt64 = 3 * 1_000_000_000) async {
|
||
sendSignal(SIGTERM)
|
||
if await waitForExit(within: graceNanos) { return }
|
||
sendSignal(SIGKILL)
|
||
// SIGKILL is uncatchable, so the process exits and `wait()` resolves; still bound it so a
|
||
// broken wait can never hang teardown.
|
||
if await waitForExit(within: graceNanos) { return }
|
||
// Killed, but `wait()` still hasn't resolved — the guest wait/stdio RPC is wedged. Finish the
|
||
// line streams ourselves so a consumer blocked on `stdoutLines` is released and teardown ends.
|
||
forceCloseStreams()
|
||
}
|
||
|
||
/// Like ``terminate(graceNanos:)`` but reaps the process's ENTIRE descendant tree, not just the
|
||
/// direct child. This is what a host `host_exec` build/test command needs: the login shell we
|
||
/// spawn (`/bin/zsh -lc …`) forks a deep tree — `make → swift → swiftpm-testing-helper →
|
||
/// *.xctest → fake-* / sh`. Signaling only the shell's pid (all ``sendSignal`` can do) orphans
|
||
/// that tree; the orphans keep holding the shared trunk's SwiftPM `.build` lock and wedge every
|
||
/// later build/test until they're killed by hand — the exact "host processes need to be
|
||
/// terminated to continue" failure this exists to prevent.
|
||
///
|
||
/// `Foundation.Process` puts the child in the app's OWN process group, so a `killpg` would kill
|
||
/// the app too; there's no supported way to make the child a group leader. A snapshot tree-walk is
|
||
/// the portable equivalent: capture the descendants BEFORE any signal (once the shell exits its
|
||
/// children reparent to launchd/init and the ancestry link is lost, but pids stay valid across
|
||
/// reparenting), then escalate SIGTERM → SIGKILL over the captured set. Accurate for the idle
|
||
/// (hung/aborted) trees this cleans up — nothing is forking new grandchildren at that point. On a
|
||
/// platform where the process table can't be read, it degrades to ``terminate(graceNanos:)``.
|
||
func terminateTree(graceNanos: UInt64 = 3 * 1_000_000_000) async {
|
||
let root = processID
|
||
guard root > 0 else { await terminate(graceNanos: graceNanos); return }
|
||
// Snapshot first — root + every transitive descendant — while the links still exist.
|
||
let tree = [root] + hostDescendantPIDs(of: root)
|
||
for pid in tree where pid > 0 { kill(pid, SIGTERM) }
|
||
if await waitForExit(within: graceNanos) {
|
||
// Root is gone; sweep any descendant that outlived it (a build child mid-shutdown).
|
||
for pid in tree where pid > 0 && pid != root { kill(pid, SIGKILL) }
|
||
return
|
||
}
|
||
// Root ignored SIGTERM — SIGKILL the whole captured set (uncatchable).
|
||
for pid in tree where pid > 0 { kill(pid, SIGKILL) }
|
||
if await waitForExit(within: graceNanos) {
|
||
for pid in tree where pid > 0 && pid != root { kill(pid, SIGKILL) }
|
||
return
|
||
}
|
||
// Even the kill didn't settle `wait()` — release any consumer blocked on the line streams.
|
||
forceCloseStreams()
|
||
}
|
||
|
||
/// Best-effort escalation for an interactive Stop: SIGINT now (let the agent end its turn
|
||
/// cleanly / emit an interrupted result), then SIGKILL the process group if it's still alive
|
||
/// after `graceNanos`. Detached so the caller's `interrupt()` stays non-blocking; the run loop
|
||
/// settles when the process finally exits. A responsive agent exits on SIGINT and the escalation
|
||
/// no-ops; a wedged one is force-killed so Stop always takes effect.
|
||
func interruptThenKill(graceNanos: UInt64 = 5 * 1_000_000_000) {
|
||
sendSignal(SIGINT)
|
||
Task { [self] in
|
||
if await waitForExit(within: graceNanos) { return }
|
||
sendSignal(SIGKILL)
|
||
// If even the kill doesn't settle `wait()` (wedged guest RPC), release any consumer
|
||
// blocked on the line streams so Stop always takes effect.
|
||
if await waitForExit(within: graceNanos) { return }
|
||
forceCloseStreams()
|
||
}
|
||
}
|
||
|
||
/// Safety net behind a *cooperative* cancel (e.g. an `turn/interrupt` / `session/cancel` RPC the
|
||
/// agent might ignore): force-SIGKILL the process group if it's still alive after `graceNanos`,
|
||
/// without sending an upfront signal that could disrupt a clean cancel in flight. Detached and a
|
||
/// no-op if the process exits on its own — so a well-behaved agent stops gracefully and a wedged
|
||
/// one is still guaranteed to stop.
|
||
func killGroupAfter(graceNanos: UInt64) {
|
||
Task { [self] in
|
||
if await waitForExit(within: graceNanos) { return }
|
||
sendSignal(SIGKILL)
|
||
if await waitForExit(within: graceNanos) { return }
|
||
forceCloseStreams()
|
||
}
|
||
}
|
||
}
|
||
|
||
// MARK: - Executable resolution (shared by every host spawn site)
|
||
|
||
/// What to hand `Process` in order to run `executable` with `args`, resolving a bare name the way
|
||
/// the platform can. `nil` means the name is not on `path` — Windows only.
|
||
///
|
||
/// This exists because there are three host spawn sites (``ChildProcess``, `GitRunner.run`,
|
||
/// `GitRunner.runRedirected`) and they must not each invent their own answer. They did, briefly:
|
||
/// item 9 taught `ChildProcess` to search PATH on Windows and left `GitRunner`'s two copies
|
||
/// pointing at `/usr/bin/env`, so every `git` call on Windows failed with "The file doesn't
|
||
/// exist" — 13 Carbon tests, one missed call site.
|
||
///
|
||
/// - POSIX keeps delegating to `/usr/bin/env`, which does the PATH search itself (and reports its
|
||
/// own exit 127 when it fails), so this never returns nil there.
|
||
/// - Windows has no `env`, and `Process` resolves a relative `executableURL` against the *current
|
||
/// directory* — a bare `git` would try to spawn `<cwd>\git`. So the search happens here, over
|
||
/// PATH × PATHEXT.
|
||
func resolvedSpawnTarget(
|
||
executable: String, args: [String], path: String?
|
||
) -> (executableURL: URL, arguments: [String])? {
|
||
#if os(Windows)
|
||
// Either separator counts as "already a path": a Windows absolute path (`C:\…\git.exe`)
|
||
// contains no forward slash, and a caller may well hand us either spelling.
|
||
if executable.contains("/") || executable.contains("\\") {
|
||
return (URL(fileURLWithPath: executable), args)
|
||
}
|
||
guard let resolved = windowsResolveExecutable(executable, path: path) else { return nil }
|
||
return (URL(fileURLWithPath: resolved), args)
|
||
#else
|
||
_ = path
|
||
if executable.contains("/") {
|
||
return (URL(fileURLWithPath: executable), args)
|
||
}
|
||
return (URL(fileURLWithPath: "/usr/bin/env"), [executable] + args)
|
||
#endif
|
||
}
|
||
|
||
/// Spawns and monitors child processes with line-buffered stdio.
|
||
public actor ProcessHost {
|
||
public init() {}
|
||
|
||
public func launch(_ spec: ProcessSpec) throws -> any ProcessHandle {
|
||
try ChildProcess(spec: spec)
|
||
}
|
||
}
|
||
|
||
// MARK: - Implementation
|
||
|
||
/// All mutable state is confined: Process control calls (signal/terminate) are
|
||
/// thread-safe, stdin writes are serialized behind a lock, and the line streams
|
||
/// are produced by dedicated reader tasks.
|
||
final class ChildProcess: ProcessHandle, @unchecked Sendable {
|
||
private let process: Process
|
||
private let stdinHandle: FileHandle?
|
||
private let stdinLock = NSLock()
|
||
private var stdinClosed = false
|
||
#if os(Windows)
|
||
/// The child's Job Object — assigned right after launch, released when this handle dies
|
||
/// (which kills any survivor). Nil only if the job couldn't be created, in which case
|
||
/// termination degrades to the single-pid path POSIX uses.
|
||
private var job: WindowsJobObject?
|
||
#endif
|
||
|
||
let stdoutLines: AsyncThrowingStream<Data, Error>
|
||
let stderrLines: AsyncThrowingStream<Data, Error>
|
||
private let exitTask: Task<Int32, Never>
|
||
/// Bumped on every stdout/stderr chunk (both pipes share it) so ``ProcessStallMonitor`` can tell
|
||
/// how long this command has been silent.
|
||
private let activity = ActivityClock()
|
||
|
||
var processID: Int32 { process.processIdentifier }
|
||
var lastActivityNanos: UInt64 { activity.lastNanos }
|
||
/// A host `ChildProcess`'s `processID` is a real host pid, so its (and its descendants') CPU time
|
||
/// can be sampled — the second stall signal alongside output-idle.
|
||
var supportsHostCPUSampling: Bool { true }
|
||
|
||
init(spec: ProcessSpec) throws {
|
||
let process = Process()
|
||
// Deliberately NO `qualityOfService` clamp here (it was `.utility` briefly). On Apple
|
||
// Silicon, Utility children are efficiency-core-biased and priority-20 (vs 31 default):
|
||
// measured on a loaded M2 Max, that made an 8-way parallel child workload 3.2× slower and
|
||
// single commands up to 2.5× slower — session builds/tests are exactly that workload, and
|
||
// the same clamp on GitRunner children pushed every git call past its wall-clock timeout.
|
||
// Default-tier children already lose to the user-interactive main thread under contention,
|
||
// so UI responsiveness does not need this.
|
||
// Resolve against the SAME widened PATH the child will receive (below) — otherwise a `git`
|
||
// that only the augmented PATH can see would be declared missing here.
|
||
let searchPath = LoginShellPATH.augment(
|
||
ProcessInfo.processInfo.environment.merging(spec.env) { _, new in new })["PATH"]
|
||
guard let target = resolvedSpawnTarget(
|
||
executable: spec.executable, args: spec.args, path: searchPath)
|
||
else {
|
||
throw BackendError.spawnFailed("\(spec.executable): not found on PATH")
|
||
}
|
||
process.executableURL = target.executableURL
|
||
process.arguments = target.arguments
|
||
process.currentDirectoryURL = URL(fileURLWithPath: spec.cwd)
|
||
// Widen PATH to the user's login-shell PATH (+ well-known install dirs) before spawning.
|
||
// A GUI-launched app inherits launchd's minimal PATH, which omits where `claude`/`codex`/
|
||
// `grok` install — so resolving a bare name via `/usr/bin/env` would fail with
|
||
// `env: <name>: No such file or directory` (exit 127). See ``LoginShellPATH``.
|
||
var childEnv = ProcessInfo.processInfo.environment.merging(spec.env) { _, new in new }
|
||
// Underlay the user's login-shell env delta (Homebrew/version-manager/toolchain exports)
|
||
// beneath the composed env — the non-PATH half of what `zsh -lc` used to re-source per
|
||
// command. Additive only (explicit spec/app values and the Managed Git pins always win),
|
||
// gated by `nucleic.host.loginEnvSnapshot`. See ``LoginShellEnv`` (docs/NASH.md §7.7.2).
|
||
// The spec's removals run AFTER the underlay so the snapshot can never re-introduce a key
|
||
// the spawn site asked to drop (its own credential filter already ran at snapshot time).
|
||
childEnv = LoginShellEnv.inject(into: childEnv)
|
||
for key in spec.removeFromEnvironment { childEnv.removeValue(forKey: key) }
|
||
process.environment = LoginShellPATH.augment(childEnv)
|
||
|
||
let stdoutPipe = Pipe()
|
||
let stderrPipe = Pipe()
|
||
process.standardOutput = stdoutPipe
|
||
process.standardError = stderrPipe
|
||
|
||
switch spec.stdinMode {
|
||
case .pipe:
|
||
let stdinPipe = Pipe()
|
||
process.standardInput = stdinPipe
|
||
stdinHandle = stdinPipe.fileHandleForWriting
|
||
case .closed:
|
||
process.standardInput = FileHandle.nullDevice
|
||
stdinHandle = nil
|
||
}
|
||
|
||
let exitStream = AsyncStream<Int32> { continuation in
|
||
process.terminationHandler = { p in
|
||
continuation.yield(p.terminationStatus)
|
||
continuation.finish()
|
||
}
|
||
}
|
||
exitTask = Task {
|
||
for await code in exitStream { return code }
|
||
return -1
|
||
}
|
||
|
||
stdoutLines = Self.lineStream(from: stdoutPipe.fileHandleForReading, activity: activity)
|
||
stderrLines = Self.lineStream(from: stderrPipe.fileHandleForReading, activity: activity)
|
||
|
||
do {
|
||
try process.run()
|
||
} catch {
|
||
throw BackendError.spawnFailed("\(spec.executable): \(error.localizedDescription)")
|
||
}
|
||
self.process = process
|
||
#if os(Windows)
|
||
// Bind the child (and everything it goes on to spawn) to a Job Object, so SIGKILL can
|
||
// reap the whole tree and so nothing survives this handle being released — the Windows
|
||
// stand-in for the POSIX process-group semantics the shared code assumes. See
|
||
// `Windows/ProcessTree.swift`.
|
||
job = WindowsJobObject(attaching: process.processIdentifier)
|
||
#endif
|
||
}
|
||
|
||
private static func lineStream(
|
||
from handle: FileHandle, activity: ActivityClock
|
||
) -> AsyncThrowingStream<Data, Error> {
|
||
AsyncThrowingStream { continuation in
|
||
// Chunked reads via `readabilityHandler` rather than `FileHandle.bytes`:
|
||
// the byte-by-byte `AsyncBytes` sequence terminates early / drops data on
|
||
// large bursty CLI output, truncating a big tool-result line and losing
|
||
// everything the agent streamed after it (the final answer + result event).
|
||
let splitter = LineSplitter()
|
||
handle.readabilityHandler = { fh in
|
||
let chunk = fh.availableData
|
||
if chunk.isEmpty { // EOF
|
||
fh.readabilityHandler = nil
|
||
if let rest = splitter.drain() { continuation.yield(rest) }
|
||
continuation.finish()
|
||
return
|
||
}
|
||
// Any output — even a chunk with no complete line yet — is a sign of life for the
|
||
// stall monitor.
|
||
activity.bump()
|
||
for line in splitter.push(chunk) { continuation.yield(line) }
|
||
}
|
||
continuation.onTermination = { _ in handle.readabilityHandler = nil }
|
||
}
|
||
}
|
||
|
||
func writeLine(_ data: Data) throws {
|
||
stdinLock.lock()
|
||
defer { stdinLock.unlock() }
|
||
guard let stdinHandle, !stdinClosed else { throw BackendError.notRunning }
|
||
var framed = data
|
||
framed.append(0x0A)
|
||
try stdinHandle.write(contentsOf: framed)
|
||
}
|
||
|
||
func closeStdin() {
|
||
stdinLock.lock()
|
||
defer { stdinLock.unlock() }
|
||
guard !stdinClosed else { return }
|
||
stdinClosed = true
|
||
try? stdinHandle?.close()
|
||
}
|
||
|
||
func sendSignal(_ sig: Int32) {
|
||
let pid = process.processIdentifier
|
||
guard pid > 0 else { return }
|
||
#if os(Windows)
|
||
// A tree-wide kill is the whole point of the job; fall through to the single-pid
|
||
// `TerminateProcess` only when there is no job.
|
||
if sig == SIGKILL, let job {
|
||
job.terminate()
|
||
return
|
||
}
|
||
#endif
|
||
kill(pid, sig)
|
||
}
|
||
|
||
func wait() async -> Int32 {
|
||
await exitTask.value
|
||
}
|
||
}
|
||
|
||
/// Accumulates raw stdout/stderr bytes and splits them into newline-delimited lines,
|
||
/// holding any partial trailing line across chunk boundaries. `readabilityHandler`
|
||
/// fires serially, but it's locked anyway so a final drain can't race the last chunk.
|
||
///
|
||
/// Module-internal (not file-private) so the containerization adapter
|
||
/// (`ContainerizedProcessHandle`) can reuse the exact same NDJSON line framing the host
|
||
/// `ChildProcess` uses — guest stdio arrives as raw byte chunks over vsock and must be split
|
||
/// identically for the backend decode loop to behave the same in-container as on the host.
|
||
final class LineSplitter: @unchecked Sendable {
|
||
private let lock = NSLock()
|
||
private var buffer = Data()
|
||
/// Number of bytes at the front of `buffer` already inspected for a newline. A native
|
||
/// stream-json event can be several megabytes when a Write tool carries a large file, and
|
||
/// pipe reads deliver that one line in many small chunks. Restarting the search at byte zero
|
||
/// for every chunk makes framing quadratic in the tool input's length and can wedge the decode
|
||
/// loop even though no bytes were lost.
|
||
private var scannedCount = 0
|
||
|
||
/// Append a chunk; return every complete line it produced (newline stripped, CR-trimmed).
|
||
func push(_ chunk: Data) -> [Data] {
|
||
lock.lock(); defer { lock.unlock() }
|
||
buffer.append(chunk)
|
||
var lines: [Data] = []
|
||
var start = buffer.startIndex
|
||
// Everything before `scannedCount` was proven newline-free by an earlier push. Begin at
|
||
// the first newly appended byte; after finding a newline, scan the remainder normally.
|
||
var searchStart = buffer.index(
|
||
buffer.startIndex, offsetBy: min(scannedCount, buffer.count))
|
||
while searchStart < buffer.endIndex,
|
||
let newline = buffer[searchStart...].firstIndex(of: 0x0A) {
|
||
var line = Data(buffer[start..<newline])
|
||
if line.last == 0x0D { line.removeLast() }
|
||
lines.append(line)
|
||
start = buffer.index(after: newline)
|
||
searchStart = start
|
||
}
|
||
// Rebase the unconsumed remainder so indices stay zero-based.
|
||
buffer = start == buffer.startIndex ? buffer : Data(buffer[start...])
|
||
// The entire remainder has now been searched. The next push only needs to inspect bytes
|
||
// appended after it, making total scan work linear in the stream size.
|
||
scannedCount = buffer.count
|
||
return lines
|
||
}
|
||
|
||
/// Any trailing bytes with no terminating newline (flushed at EOF).
|
||
func drain() -> Data? {
|
||
lock.lock(); defer { lock.unlock() }
|
||
guard !buffer.isEmpty else { return nil }
|
||
defer {
|
||
buffer.removeAll()
|
||
scannedCount = 0
|
||
}
|
||
return buffer
|
||
}
|
||
}
|
||
|
||
// MARK: - Process-tree enumeration (host)
|
||
|
||
/// The transitive descendant pids of `root`, from a single snapshot of the host process table.
|
||
/// Returns just the descendants (NOT `root`). Empty when `root` has no children or the table can't
|
||
/// be read — callers degrade to signaling `root` alone. Best-effort by design: a hung build tree is
|
||
/// idle, so one snapshot captures it faithfully; a live-forking tree could miss a just-spawned
|
||
/// grandchild. Used by ``ProcessHandle/terminateTree(graceNanos:)``.
|
||
func hostDescendantPIDs(of root: pid_t) -> [pid_t] {
|
||
let table = hostProcessTable()
|
||
guard !table.isEmpty else { return [] }
|
||
var childrenByParent: [pid_t: [pid_t]] = [:]
|
||
for entry in table { childrenByParent[entry.ppid, default: []].append(entry.pid) }
|
||
var result: [pid_t] = []
|
||
var visited: Set<pid_t> = [root] // guards against pid-reuse cycles in the snapshot
|
||
var queue = childrenByParent[root] ?? []
|
||
var i = 0
|
||
while i < queue.count {
|
||
let pid = queue[i]; i += 1
|
||
guard visited.insert(pid).inserted else { continue }
|
||
result.append(pid)
|
||
if let kids = childrenByParent[pid] { queue.append(contentsOf: kids) }
|
||
}
|
||
return result
|
||
}
|
||
|
||
#if canImport(Darwin)
|
||
/// (pid, ppid) for every process, via `sysctl(KERN_PROC_ALL)`. Over-allocates a little slack so the
|
||
/// table growing between the size probe and the fetch doesn't ENOMEM; on any error returns [].
|
||
private func hostProcessTable() -> [(pid: pid_t, ppid: pid_t)] {
|
||
var mib: [Int32] = [CTL_KERN, KERN_PROC, KERN_PROC_ALL, 0]
|
||
var size = 0
|
||
if sysctl(&mib, u_int(mib.count), nil, &size, nil, 0) != 0 || size == 0 { return [] }
|
||
let stride = MemoryLayout<kinfo_proc>.stride
|
||
let capacity = size / stride + 32 // slack for processes spawned since the probe
|
||
var procs = [kinfo_proc](repeating: kinfo_proc(), count: capacity)
|
||
var byteLen = capacity * stride
|
||
let rc = procs.withUnsafeMutableBytes { buf in
|
||
sysctl(&mib, u_int(mib.count), buf.baseAddress, &byteLen, nil, 0)
|
||
}
|
||
if rc != 0 { return [] }
|
||
let count = byteLen / stride
|
||
return (0..<count).map { i in
|
||
(pid: procs[i].kp_proc.p_pid, ppid: procs[i].kp_eproc.e_ppid)
|
||
}
|
||
}
|
||
#elseif os(Linux)
|
||
/// (pid, ppid) for every process, from `/proc/<pid>/stat`. The `comm` field (field 2) can contain
|
||
/// spaces and parentheses, so ppid is read as the second whitespace token AFTER the final ')'.
|
||
private func hostProcessTable() -> [(pid: pid_t, ppid: pid_t)] {
|
||
guard let entries = try? FileManager.default.contentsOfDirectory(atPath: "/proc") else { return [] }
|
||
var table: [(pid: pid_t, ppid: pid_t)] = []
|
||
for entry in entries {
|
||
guard let pid = pid_t(entry),
|
||
let stat = try? String(contentsOfFile: "/proc/\(entry)/stat", encoding: .utf8),
|
||
let close = stat.lastIndex(of: ")")
|
||
else { continue }
|
||
let fields = stat[stat.index(after: close)...]
|
||
.split(separator: " ", omittingEmptySubsequences: true)
|
||
// fields[0] = state, fields[1] = ppid
|
||
guard fields.count >= 2, let ppid = pid_t(fields[1]) else { continue }
|
||
table.append((pid: pid, ppid: ppid))
|
||
}
|
||
return table
|
||
}
|
||
#elseif os(Windows)
|
||
/// (pid, ppid) for every process, from a Toolhelp snapshot — see `Windows/ProcessTree.swift`.
|
||
private func hostProcessTable() -> [(pid: pid_t, ppid: pid_t)] { windowsProcessTable() }
|
||
#else
|
||
private func hostProcessTable() -> [(pid: pid_t, ppid: pid_t)] { [] }
|
||
#endif
|
||
|
||
// MARK: - Process-tree CPU sampling (host)
|
||
|
||
/// Cumulative CPU time (user + system) in nanoseconds for `root` plus every transitive descendant,
|
||
/// from a single process-table snapshot. `0` when the tree can't be read or `root` is invalid.
|
||
/// ``ProcessStallMonitor`` diffs consecutive samples to get a "% of one core" over the interval —
|
||
/// the signal that tells a wedged (idle-CPU) command apart from one that's quietly working hard.
|
||
/// Reuses ``hostDescendantPIDs(of:)`` so it covers the whole `zsh → make → swift → xctest` fork
|
||
/// tree, not just the shell.
|
||
func hostProcessTreeCPUNanos(of root: pid_t) -> UInt64 {
|
||
guard root > 0 else { return 0 }
|
||
var total = hostProcessCPUNanos(root)
|
||
for pid in hostDescendantPIDs(of: root) where pid > 0 { total &+= hostProcessCPUNanos(pid) }
|
||
return total
|
||
}
|
||
|
||
#if canImport(Darwin)
|
||
/// One process's cumulative user+system CPU nanoseconds via `proc_pid_rusage`. `0` on any error
|
||
/// (dead pid, permission) — a dead child just contributes nothing to the tree total.
|
||
private func hostProcessCPUNanos(_ pid: pid_t) -> UInt64 {
|
||
var info = rusage_info_v4()
|
||
let rc = withUnsafeMutablePointer(to: &info) { ptr in
|
||
ptr.withMemoryRebound(to: rusage_info_t?.self, capacity: 1) { reb in
|
||
proc_pid_rusage(pid, RUSAGE_INFO_V4, reb)
|
||
}
|
||
}
|
||
guard rc == 0 else { return 0 }
|
||
// `ri_user_time` / `ri_system_time` are already in nanoseconds.
|
||
return info.ri_user_time &+ info.ri_system_time
|
||
}
|
||
#elseif os(Linux)
|
||
/// One process's cumulative user+system CPU nanoseconds from `/proc/<pid>/stat` (utime+stime, in
|
||
/// clock ticks → nanos via `_SC_CLK_TCK`). `0` on any error.
|
||
private func hostProcessCPUNanos(_ pid: pid_t) -> UInt64 {
|
||
guard let stat = try? String(contentsOfFile: "/proc/\(pid)/stat", encoding: .utf8),
|
||
let close = stat.lastIndex(of: ")")
|
||
else { return 0 }
|
||
// After the final ')', the first token is field 3 (state); utime is field 14 → index 11,
|
||
// stime is field 15 → index 12 (mirrors the ppid-at-index-1 arithmetic in `hostProcessTable`).
|
||
let fields = stat[stat.index(after: close)...]
|
||
.split(separator: " ", omittingEmptySubsequences: true)
|
||
guard fields.count > 12, let utime = UInt64(fields[11]), let stime = UInt64(fields[12])
|
||
else { return 0 }
|
||
let hz = UInt64(max(sysconf(Int32(_SC_CLK_TCK)), 1))
|
||
return (utime &+ stime) &* 1_000_000_000 / hz
|
||
}
|
||
#elseif os(Windows)
|
||
/// One process's cumulative user+kernel CPU nanoseconds via `GetProcessTimes`.
|
||
private func hostProcessCPUNanos(_ pid: pid_t) -> UInt64 { windowsProcessCPUNanos(pid) }
|
||
#else
|
||
private func hostProcessCPUNanos(_ pid: pid_t) -> UInt64 { 0 }
|
||
#endif
|
||
|
||
// MARK: - Process-tree disk-I/O sampling (host)
|
||
|
||
/// Cumulative disk-I/O bytes (read + written) for `root` plus every transitive descendant, from a
|
||
/// single process-table snapshot. `0` when the tree can't be read or `root` is invalid.
|
||
/// ``ProcessStallMonitor`` diffs consecutive samples to tell a wedged command apart from one that is
|
||
/// quietly *moving data* at near-zero CPU — a `cctl images push` streaming blobs to a registry, a
|
||
/// slow `cp` — the exact case CPU sampling alone can't exonerate. Reuses ``hostDescendantPIDs(of:)``
|
||
/// so it covers the whole fork tree, matching the CPU sampler.
|
||
func hostProcessTreeIOBytes(of root: pid_t) -> UInt64 {
|
||
guard root > 0 else { return 0 }
|
||
var total = hostProcessIOBytes(root)
|
||
for pid in hostDescendantPIDs(of: root) where pid > 0 { total &+= hostProcessIOBytes(pid) }
|
||
return total
|
||
}
|
||
|
||
#if canImport(Darwin)
|
||
/// One process's cumulative disk bytes (read + written) via `proc_pid_rusage` — the same call the CPU
|
||
/// sampler already makes. `0` on any error (dead pid, permission).
|
||
private func hostProcessIOBytes(_ pid: pid_t) -> UInt64 {
|
||
var info = rusage_info_v4()
|
||
let rc = withUnsafeMutablePointer(to: &info) { ptr in
|
||
ptr.withMemoryRebound(to: rusage_info_t?.self, capacity: 1) { reb in
|
||
proc_pid_rusage(pid, RUSAGE_INFO_V4, reb)
|
||
}
|
||
}
|
||
guard rc == 0 else { return 0 }
|
||
return info.ri_diskio_bytesread &+ info.ri_diskio_byteswritten
|
||
}
|
||
#elseif os(Linux)
|
||
/// One process's cumulative I/O bytes from `/proc/<pid>/io` (`rchar` + `wchar` — the bytes the
|
||
/// process caused to be read/written, the closest analogue to Darwin's disk counter). `0` on any
|
||
/// error (the file is readable only by the process owner).
|
||
private func hostProcessIOBytes(_ pid: pid_t) -> UInt64 {
|
||
guard let io = try? String(contentsOfFile: "/proc/\(pid)/io", encoding: .utf8) else { return 0 }
|
||
var total: UInt64 = 0
|
||
for line in io.split(separator: "\n") where line.hasPrefix("rchar:") || line.hasPrefix("wchar:") {
|
||
total &+= line.split(separator: " ").last.flatMap { UInt64($0) } ?? 0
|
||
}
|
||
return total
|
||
}
|
||
#elseif os(Windows)
|
||
/// One process's cumulative I/O bytes via `GetProcessIoCounters`.
|
||
private func hostProcessIOBytes(_ pid: pid_t) -> UInt64 { windowsProcessIOBytes(pid) }
|
||
#else
|
||
private func hostProcessIOBytes(_ pid: pid_t) -> UInt64 { 0 }
|
||
#endif
|