Oro Computer logooro
Inside
Silk
Standard library / std::stream

std::stream

This module provides a Web Streams-inspired API for byte streams designed to work well with Silk’s async/task model.

The core goals of the Supported forms are:

  • Ergonomic piping between producers and consumers.
  • Backpressure via bounded buffering.
  • Safe chunk ownership across tasks using an owned Bytes type (no borrowed slice lifetime hazards).

Runtime note (Supported forms):

  • ReadableStream.read() / WritableStream.write() are blocking OS-thread operations implemented with mutex/condvar primitives.

See also:

  • concurrency (tasks, yield, structured blocks)
  • sync (mutex/condvar/channel; same blocking baseline)

Exported API#

module std::stream;

import std::arrays;
import std::buffer;
import std::interfaces;
import std::memory;
import std::result;

export enum StreamErrorKind {
  OutOfMemory,
  InvalidInput,
  Closed,
  Full,
  Cancelled,
  Aborted,
  RuntimeFailed,
  Unknown,
}

export error StreamFailed {
  code: int,
}

impl StreamFailed {
  public fn kind (self: &StreamFailed) -> StreamErrorKind;
}

// Owned, immutable byte chunks.
export struct Bytes {
  handle: u64,
}

export type BytesResult = std::result::Result(Bytes, std::memory::AllocFailed);

impl Bytes {
  public fn empty () -> Bytes;
  public fn from_handle (handle: u64) -> Bytes;
  public fn as_slice (self: &Bytes) -> std::arrays::ByteSlice;
  public fn copy_from (slice: std::arrays::ByteSlice) -> BytesResult;
  public fn copy_from_string (s: string) -> BytesResult;
  public fn take_from_buffer (mut buf: &std::buffer::BufferU8) -> BytesResult;
}

impl Bytes as std::interfaces::Len {
  public fn len (self: &Bytes) -> i64;
}

impl Bytes as std::interfaces::IsEmpty {
  public fn is_empty (self: &Bytes) -> bool;
}

impl Bytes as std::interfaces::Sized {
  public fn size (self: &Bytes) -> usize;
}

// Read outcomes.
export enum Read {
  Done,
  Pending,
  // Bytes are transferred as raw handles; wrap with `Bytes.from_handle(handle)`.
  Chunk(u64),
}

export type ReadResult = std::result::Result(Read, StreamFailed);

// Readable end.
export struct ReadableStream {
  handle: u64,
}

impl ReadableStream {
  public fn invalid () -> ReadableStream;
  public fn is_valid (self: &ReadableStream) -> bool;
  public fn cap (self: &ReadableStream) -> int;
  public fn len (self: &ReadableStream) -> int;
  public fn is_closed (self: &ReadableStream) -> bool;
  public fn read (self: &ReadableStream) -> ReadResult;
  public fn try_read (self: &ReadableStream) -> ReadResult;
  public fn cancel (self: &ReadableStream) -> void;
  public fn destroy (mut self: &ReadableStream) -> void;
}

// Writable end.
export struct WritableStream {
  handle: u64,
}

impl WritableStream {
  public fn invalid () -> WritableStream;
  public fn is_valid (self: &WritableStream) -> bool;
  public fn cap (self: &WritableStream) -> int;
  public fn len (self: &WritableStream) -> int;
  public fn is_closed (self: &WritableStream) -> bool;
  public fn write (self: &WritableStream, mut chunk: Bytes) -> StreamFailed?;
  public fn try_write (self: &WritableStream, mut chunk: Bytes) -> StreamFailed?;
  public fn close (self: &WritableStream) -> void;
  public fn abort (self: &WritableStream, err: StreamFailed) -> void;
  public fn destroy (mut self: &WritableStream) -> void;
}

// A paired in-memory stream (writable → readable).
export struct PassThroughStream {
  readable: ReadableStream,
  writable: WritableStream,
}

export type PassThroughResult = std::result::Result(PassThroughStream, StreamFailed);

impl PassThroughStream {
  public fn init_default () -> PassThroughResult;
  public fn init (cap: int) -> PassThroughResult;
  public fn take_readable (mut self: &PassThroughStream) -> ReadableStream;
  public fn take_writable (mut self: &PassThroughStream) -> WritableStream;
}

// Transformer output.
export type TransformBytesResult = std::result::Result(Bytes, StreamFailed);

// A paired transform stage.
export struct TransformStream {
  readable: ReadableStream,
  writable: WritableStream,
  transform_readable: ReadableStream,
  transform_writable: WritableStream,
}

export type TransformResult = std::result::Result(TransformStream, StreamFailed);

impl TransformStream {
  public fn init_default () -> TransformResult;
  public fn init (cap: int) -> TransformResult;
  public fn init_with_caps (cap_in: int, cap_out: int) -> TransformResult;
  public fn take_readable (mut self: &TransformStream) -> ReadableStream;
  public fn take_writable (mut self: &TransformStream) -> WritableStream;
  public fn take_transform_readable (mut self: &TransformStream) -> ReadableStream;
  public fn take_transform_writable (mut self: &TransformStream) -> WritableStream;
}

// Pipe a readable into a writable until done.
export fn pipe_to (mut src: ReadableStream, mut dst: WritableStream) -> std::result::Result(int, StreamFailed);

// Pipe until done or until aborted.
export fn pipe_to_abortable (
  mut src: ReadableStream,
  mut dst: WritableStream,
  sig: std::abort_controller::AbortSignalBorrow?
) -> std::result::Result(int, StreamFailed);

Semantics#

Backpressure#

Each stream has a bounded in-memory queue. The cap is expressed in chunks (Bytes values), not bytes:

  • PassThroughStream.init(cap) / TransformStream.init(cap) require cap > 0 (otherwise they return Err(StreamFailed) with kind() == InvalidInput).
  • WritableStream.write() blocks while the queue is full.
  • ReadableStream.read() blocks while the queue is empty (until closed or errored).

Close vs cancel vs abort#

  • WritableStream.close():
  • graceful end-of-stream,
  • readers drain remaining buffered chunks and then observe Read::Done.
  • ReadableStream.cancel():
  • marks the stream cancelled,
  • discards buffered chunks,
  • causes writers to fail with a Cancelled error.
  • WritableStream.abort(err):
  • marks the stream aborted with err,
  • discards buffered chunks,
  • causes readers to return Err(err) from read / try_read.

Transform streams#

TransformStream models a Web Streams-style transform stage.

In Silk currently, std::stream does not attach a transformer callback internally. Instead, TransformStream exposes two bounded pipes and expects you to run the transform loop in a task:

  • input pipe: producers write to writable, transformer reads from transform_readable,
  • output pipe: transformer writes to transform_writable, consumers read from readable.

This design composes naturally with task-based structured concurrency: run the transform loop in a task and rely on backpressure to bound memory.

Usage patterns#

Producer → consumer (tasks)#

import std::stream;

task fn producer (w: std::stream::WritableStream) -> int { ... }
task fn consumer (r: std::stream::ReadableStream) -> int { ... }

async fn main () -> int {
  task {
    match (std::stream::PassThroughStream.init_default()) {
      Ok(stream) => {
        let mut pt: std::stream::PassThroughStream = stream;
        let w = pt.take_writable();
        let r = pt.take_readable();

        let hp = producer(w);
        let hc = consumer(r);

        let rp: int = yield hp;
        let rc: int = yield hc;
        if rp != 0 { return rp; }
        if rc != 0 { return rc; }
        return 0;
      },
      Err(_) => {
        return 1;
      },
    }
  }
}

Transform stage (tasks)#

TransformStream is a pair of pipes intended to be driven by a transformer task. Typical wiring:

  • producer writes to take_writable(),
  • transformer reads from take_transform_readable() and writes to take_transform_writable(),
  • consumer reads from take_readable().

File I/O adapters (std::fs)#

std::fs provides task-based helpers for piping files into/out of streams:

import std::fs;
import std::fs::stream;
import std::stream;

async fn main () -> int {
  task {
    match (std::stream::PassThroughStream.init(2)) {
      Ok(stream) => {
        let mut pt: std::stream::PassThroughStream = stream;
        let w = pt.take_writable();
        let r = pt.take_readable();

        let hr = std::fs::stream::pipe_file_to_stream("input.txt", w, 4096);
        let hw = std::fs::stream::pipe_stream_to_file(r, "output.txt", 420);

        let rr: std::stream::PipeResult = yield hr;
        let rw: std::stream::PipeResult = yield hw;
        match (rr) {
          Err(_) => { return 2; },
        }
        match (rw) {
          Err(_) => { return 3; },
        }
        return 0;
      },
      Err(_) => {
        return 1;
      },
    }
  }
}

Piping#

pipe_to is a structured copy loop:

  • closes dst when src ends,
  • aborts/cancels on error.

To make piping cooperatively cancellable, use pipe_to_abortable with an std::abort_controller::AbortSignalBorrow. In the Supported forms, aborts are observed between read/write steps; they do not yet interrupt a blocking ReadableStream.read() call. 

import std::stream;

fn run_pipeline (src: std::stream::ReadableStream, dst: std::stream::WritableStream) -> int {
  let r = std::stream::pipe_to(src, dst);
  return match (r) {
    Ok(_) => 0,
    Err(_) => 1,
  };
}

Notes#

  • The Supported forms uses blocking primitives; it is intended to become suspension-friendly once the async runtime exists.
  • Drop semantics are designed to avoid leaked pipes:
  • dropping ReadableStream cancels the stream (writers start failing),
  • dropping WritableStream closes the stream (readers observe Read::Done after draining).
  • PassThroughStream.take_readable / take_writable exist to make ownership transfer ergonomic in the Supported forms (moving out of struct fields is limited).

Source repository · Edit this page · View Markdown

Previousstd::websocket
Nextstd::crypto