Bounded.pudu

Pudu144 lines5.6 KB

Copy pathGitHub ↗

1/** @Services.Process.Bounded.Backbone — runs one child within deadline and cap */2module PuduLangMcp.Services.Process.Bounded34import Std.Bytes as ByteSeq5import Std.Concurrent as Concurrent6import Std.Env as Env7import Std.Math as Math8import Std.Process as Process9import Std.Result as Result10import Std.Sync as Sync1112/** @Services.Process.Bounded.Finished — what a bounded child left behind */13export type Finished = { status: Int, output: Str, errors: Str, timedOut: Bool, truncated: Bool, millis: Int }1415/// How often a running child is looked at while it is waited for.16const POLL_MILLIS: Int = 251718/// How long the output readers may take to reach the end of their streams once the child is gone.19const DRAIN_GRACE_MILLIS: Int = 2502021/// How often an unfinished reader is looked at during the grace.22const DRAIN_POLL_MILLIS: Int = 22324/// What one reader has kept so far, and whether its stream has closed.25type Captured = { kept: Bytes, ended: Bool }2627/// The longest byte sequence of one UTF-8 character, less one.28const MAX_CONTINUATION: Int = 32930/// What a stream holding bytes that are not UTF-8 text is shown as.31const NOT_TEXT: Str = "[output that is not UTF-8 text]"3233/// Runs `program` to its end or its deadline and answers everything it wrote, within the cap.34export fn run(program: Str, arguments: &Array[Str], input: Str, directory: Str, millis: Int, capBytes: Int) -> Result[Finished, Str] {35  let started = Env.elapsedMilliseconds()36  let base = Process.launch(program, arguments)37  let launch = if directory.isEmpty() { base } else { base.inDirectory(directory) }38  let running = Process.begin(&launch) ?39  let output = Sync.cell(Captured{kept: ByteSeq.empty(), ended: false})40  let errors = Sync.cell(Captured{kept: ByteSeq.empty(), ended: false})41  let outputReader = capture(&running, Process.read, &output, capBytes)42  let errorReader = capture(&running, Process.readErrors, &errors, capBytes)43  if !input.isEmpty() {44    let _written = Process.writeText(&running, input)45  }46  let _closed = Process.closeInput(&running)47  let waited = waitFor(&running, started + millis)48  let drainDeadline = Env.elapsedMilliseconds() + DRAIN_GRACE_MILLIS49  let outputEnded = settled(outputReader, &output, drainDeadline)50  let errorsEnded = settled(errorReader, &errors, drainDeadline)51  let ending = waited ?52  let shownOutput = textWithin(&current(&output).kept, capBytes)53  let shownErrors = textWithin(&current(&errors).kept, capBytes)54  Ok(Finished {55      status: ending[0],56      output: shownOutput[0],57      errors: shownErrors[0],58      timedOut: ending[1],59      truncated: shownOutput[1] || shownErrors[1] || !outputEnded || !errorsEnded,60      millis: Env.elapsedMilliseconds() - started61    })62}6364/// At most `limit` bytes as text, cut on a character boundary, and whether anything was cut.65export fn textWithin(written: &Bytes, limit: Int) -> (Str, Bool) {66  let size = ByteSeq.length(written)67  let over = size > limit68  let kept = if over { ByteSeq.take(written, limit) } else { *written }69  let keptSize = ByteSeq.length(&kept)70  if keptSize == 0 { return ("", over) }71  var back = 072  let lastBack = Math.min(MAX_CONTINUATION, keptSize - 1)73  while back <= lastBack {74    match ByteSeq.toText(&ByteSeq.take(&kept, keptSize - back)) {75      case Ok(text) => { return (text, over || back > 0) }76      case Err(_) => { back = back + 1 }77    }78  }79  if over && keptSize <= MAX_CONTINUATION { ("", true) } else { (NOT_TEXT, over) }80}8182/// The child's exit status and whether it was stopped at the deadline.83fn waitFor(running: &Process.Started, deadline: Int) -> Result[(Int, Bool), Str] {84  loop {85    match Process.waitWithin(running, POLL_MILLIS) {86      case Err(problem) => {87        let _stopped = Process.stop(running)88        return Err(problem)89      }90      case Ok(Some(status)) => { return Ok((status, false)) }91      case Ok(None) => {92        if deadlineReached(Env.elapsedMilliseconds(), deadline) {93          let _stopped = Process.stop(running)94          return Ok((-1, true))95        }96      }97    }98  }99}100101/// Whether a child has reached its execution deadline.102export fn deadlineReached(now: Int, deadline: Int) -> Bool {103  now >= deadline104}105106/// Reads one stream to its end on its own thread, keeping at most one byte past `limit`.107fn capture(running: &Process.Started, next: fn(&Process.Started) -> Result[Option[Bytes], Str], into: &Sync.Cell[Captured], limit: Int) -> Result[Concurrent.Task, Concurrent.ConcurrentError] {108  let held = *running109  let target = *into110  Concurrent.start(fn() -> () {111      var kept = ByteSeq.empty()112      loop {113        match next(&held) {114          case Ok(Some(piece)) => {115            if ByteSeq.length(&kept) <= limit {116              kept = ByteSeq.take(&ByteSeq.concat(&kept, &piece), limit + 1)117              let _published = Sync.set(&target, Captured{kept: kept, ended: false})118            }119          }120          case Ok(None) => { break }121          case Err(_) => { break }122        }123      }124      let _stored = Sync.set(&target, Captured{kept: kept, ended: true})125    })126}127128/// Whether a reader's stream closed by `deadline`; a closed reader is joined, an open one is left.129fn settled(reader: Result[Concurrent.Task, Concurrent.ConcurrentError], into: &Sync.Cell[Captured], deadline: Int) -> Bool {130  while !current(into).ended && !deadlineReached(Env.elapsedMilliseconds(), deadline) {131    let _slept = Concurrent.sleep(DRAIN_POLL_MILLIS)132  }133  let ended = current(into).ended134  if ended {135    let _joined = Result.map(reader, |worker: Concurrent.Task| Concurrent.join(&worker))136  }137  ended138}139140/// What a reader has published; an unreadable cell reads as a stream still open, so waits stay bounded.141fn current(into: &Sync.Cell[Captured]) -> Captured {142  Result.unwrapOr(Sync.get(into), Captured{kept: ByteSeq.empty(), ended: false})143}144