Wait for I/O with a deadline

A worker can own or borrow a File like any other value. To keep it from blocking forever on a pipe that never delivers, wait with std.ready: every wait has a deadline, and a cancellable wait checks its token at least every 10 milliseconds.

import std.bytes
import std.io
import std.process
from std.io import FileStream
from std.process import Stdio
from std.ready import WaitError, read_before
from std.sync import CancellationToken
from std.time import Deadline, deadline_after, milliseconds, monotonic_now, sleep

def first_bytes(stream: &mut FileStream, deadline: &Deadline, stop: &CancellationToken) -> Result[str, std.bytes.BytesError | AllocError]:
    mut buffer = std.bytes.with_capacity(16)?
    buffer.resize(16, 0u8)?
    text = match read_before(stream, &mut buffer, 0, 16, deadline, stop):
        case Ok(count):
            str.from(count)?
        case Err(WaitError.TimedOut):
            "timed out"
        case Err(WaitError.Cancelled):
            "cancelled"
        case Err(WaitError.Stream(_)):
            "failed"
    Ok(text)

type PipeWaitError = str | RangeError | std.bytes.BytesError

def main() -> Result[(), SpawnError[()] | IoError | AllocError | PipeWaitError]:
    mut command = std.process.shell("sleep 10")?
    command.stdout(Stdio.Piped)
    mut child = command.spawn()?
    pipe: Result[File, str] = child.take_stdout().ok_or("missing piped stdout")
    mut quiet = std.io.from_file(pipe?)
    never = CancellationToken()?
    soon = deadline_after(&monotonic_now(), &milliseconds(20))?
    with spawn(first_bytes, &mut quiet, &soon, &never)? as task:
        print(task.join()?)?
    stop = CancellationToken()?
    later = deadline_after(&monotonic_now(), &milliseconds(60000))?
    observer = stop.share()
    with spawn(first_bytes, &mut quiet, &later, &observer)? as task:
        sleep(&milliseconds(20))
        stop.cancel()
        print(task.join()?)?
    child.kill()?
    child.wait()?
    Ok(())

timed out
cancelled

The child writes nothing, so the first worker’s 20-millisecond deadline passes. The second worker has a minute, but the parent cancels it, and the cancellation wakes its wait at once. Either way the worker returns normally and the scope joins it; the pipe and buffer are unchanged.

A worker may not change the working directory, because every thread resolves relative paths against it:

import std.env
import std.path

def worker() -> Result[(), AllocError | IoError]:
    std.env.set_current_dir(&std.path.from_text("/")?)

def main() -> Result[(), SpawnError[()]]:
    with spawn(worker).drop_job()?:
        pass
    Ok(())

changing the working directory affects every thread's relative paths

Change it on the main thread before starting workers, or give a child its own directory with Command.current_dir. See native effects on worker threads for the rules each native operation follows.