ww/lib/bufio/stream.ww

// bufio — buffered I/O over [[io.stream]] (= `*io.vtable`). Subset of
// Hare's bufio:: surface (ref/hare/bufio/{scanner,stream}.ha). Project
// #94 fold-eFinal.
//
// VALUE-RETURN (Hare ref/hare/bufio/stream.ha:69 init, scanner.ha:92
// newscanner_buf): each constructor builds in a local `let r: T;`,
// field-assigns every slot, and `return r;`. The caller owns the
// returned struct (stack ownership, no-GC). For the buffered stream
// the caller passes `&b.vt` to the io dispatchers; the scanner is a
// plain read-ahead tokenizer (no embedded vtable) driven directly via
// scanbyte / scanbytes / scanline.
//
// SCOPE: this is the surface ww ships, NOT a port of Hare's full
// scanner. The features ww never implemented — multibyte-delim
// scanbytes / scanstring, readbyte / readtok / readline / readrune /
// unreadrune — are a separate future feature fold (#217). The ported
// set: the auto-grow newscanner, the fixed-buffer newscannerbuf,
// single-byte scanbytes, single-byte scanline, scanbyte, scanrune,
// finish; and the buffered-stream init / setflush / flush / unread /
// isbuffered.
//
// Cast workaround per #206-payoff (ken: KEEP the explicit casts; they
// are cgen-neutral and sidestep the #214 over-acceptance surface).
// `(&fn_name): *io.<role>` at each vtable store + the isbuffered
// fn-ptr-equality comparand — the #206 cast-drop is gated on #214.
//
// Mode discrimination (drew-deferred): Hare uses three vtable
// singletons (vtable_r / vtable_w / vtable_rw); ww's vtable always
// carries all three callbacks (a zero-length rbuf/wbuf degenerates the
// matching callback in-cb). Defer-handle (MANAGED_* ownership bits)
// also deferred — caller owns rbuf/wbuf/src; bclose flushes only and
// neither frees nor closes (close-propagation rides #5
// MANAGED_HANDLE, ref/hare/bufio/stream.ha:194). Both graduate with io
// fold-2 (#5).

package bufio;

import io;

// flushdefault — backing storage for the default flush byte-set
// ("\n"). Hare scopes it inside `init` as `static let
// flush_default = ['\n': u8]` (ref/hare/bufio/stream.ha:75); ww
// has no function-scope statics, so it lives at module scope.
let flushdefault: [1]u8 = [10u8];

// rt_abort — terminate on a precondition violation. Used by
// [[unread]] for the "buf fits in front of rbuf" assertion that
// Hare expresses with `assert`.
@symbol("rt_abort") fn rtabort(msg: str) void;

// stream — heap-free buffered read+write over an underlying io.stream.
// `vt` at offset 0 for the intrusive io.stream→*stream cast. `src` is
// an io.stream (the vtable-native underlying handle). Mirrors
// ref/hare/bufio/stream.ha:18.
//
// rbuf[rstart..rend] is pending read data; wbuf[0..wend] is pending
// write data. rbuf or wbuf may be zero-length: read-empty makes bread
// return io.eof immediately, write-empty makes bwrite a pass-through to
// src (no buffering).
export type stream = struct {
	vt:     io.vtable,
	src:    io.stream,
	rbuf:   []u8,
	rstart: i32,
	rend:   i32,
	wbuf:   []u8,
	wend:   i32,
	flush:  []u8,
};

// init — wire a buffered stream over an underlying io.stream with
// caller-supplied read/write buffers. Both rbuf and wbuf may be empty;
// bread / bwrite degenerate (see [[stream]]). The flush byte-set
// defaults to "\n" (line-buffered writes); [[setflush]] swaps it.
//
// Returns the stream BY VALUE; the caller passes `&b.vt` to the io
// dispatchers. Mirrors ref/hare/bufio/stream.ha:69.
export fn newstream(src: io.stream, rbuf: []u8, wbuf: []u8) stream = {
	let r: stream;
	r.vt.reader = (&bread):  *io.reader;
	r.vt.writer = (&bwrite): *io.writer;
	r.vt.closer = (&bclose): *io.closer;
	r.src    = src;
	// rstart=rbuf.len, rend=rbuf.len: pre-read unread budget = rbuf.len
	// (Hare bufio/stream.ha:101).
	r.rstart = rbuf.len;
	r.rend   = rbuf.len;
	r.rbuf   = rbuf;
	r.wbuf   = wbuf;
	r.wend   = 0;
	r.flush  = flushdefault[0:1];
	return r;
};

fn bufiomove(dst: *u8, src: *u8, n: i32) void = {
	if (n <= 0 || dst == src) { return; };
	let i: i32 = 0;
	if ((dst: uintptr) > (src: uintptr)) {
		i = n;
		for (i > 0) {
			i -= 1;
			dst[i] = src[i];
		};
	} else {
		for (i < n) {
			dst[i] = src[i];
			i += 1;
		};
	};
};

fn flushpending(b: *stream, off: i32) void = {
	if (off == 0) { return; };
	let remain: i32 = b.wend - off;
	bufiomove(b.wbuf.ptr, b.wbuf.ptr + (off: u64), remain);
	b.wend = remain;
};

// setflush — install a new flush byte-set. Any byte from `bs` appearing
// in a write payload triggers an automatic flush after the write copies
// into wbuf. Mirrors ref/hare/bufio/stream.ha:128.
export fn setflush(b: *stream, bs: []u8) void = {
	b.flush = bs;
};

// flush — drain any pending wbuf data to src. Public API AND the
// internal drain called by bwrite / bclose. A 0-byte non-error write
// means the sink made no progress (memio.fixed full) — surfaced as a
// nomem-carried io.error rather than spinning. Accepted prefixes are removed
// from wbuf before any failure is returned, so a retry cannot duplicate them.
// Mirrors ref/hare/bufio/stream.ha.
export fn flush(b: *stream) (void | io.error) = {
	if (b.wend == 0) { return; };
	let off: i32 = 0;
	for (off < b.wend) {
		let r: (size | io.error) = io.write(b.src, b.wbuf[off:b.wend]);
		match (r) {
		case let n: size => {
			assert(n <= (b.wend - off): size,
				"bufio.flush: writer returned an oversized count");
			if (n == 0: size) {
				flushpending(b, off);
				let nm: nomem;
				let e: io.error = nm;
				return e;
			};
			off += n: i32;
		};
		case let e: io.error => {
			flushpending(b, off);
			return e;
		};
		};
	};
	b.wend = 0;
	return;
};

// unread — push `buf` back into the read buffer so the next bread
// returns it first. The bytes must fit in front of the pending region
// (rstart >= buf.len); Hare aborts on overflow, ww does the same via
// rtabort. Mirrors ref/hare/bufio/stream.ha:164.
export fn unread(b: *stream, buf: []u8) void = {
	if (b.rstart < buf.len) {
		rtabort("bufio.unread: more data than rbuf has room for");
	};
	bufiomove(b.rbuf.ptr + ((b.rstart - buf.len): u64), buf.ptr,
		buf.len);
	b.rstart -= buf.len;
};

// isbuffered — true when `s` was returned by [[init]]. Hare's
// discriminator is callback identity (ref/hare/bufio/stream.ha:179).
// Either reader or writer matching is sufficient.
export fn isbuffered(s: io.stream) bool = {
	match (s.reader) {
	case let r: *io.reader => {
		if (r == (&bread): *io.reader) { return true; };
	};
	case void => { };
	};
	match (s.writer) {
	case let w: *io.writer => {
		if (w == (&bwrite): *io.writer) { return true; };
	};
	case void => { };
	};
	return false;
};

// bread — buffered read. Recover stream via the intrusive cast; top up
// rbuf from src via io.read whenever the pending region holds fewer
// bytes than requested AND rbuf has spare capacity (Hare's short-read
// refill, ref/hare/bufio/stream.ha:209). The pending bytes shift to the
// front, then io.read fills the tail; EOF is fatal only when nothing is
// pending. Mirrors ref/hare/bufio/stream.ha:205 (stream_read).
fn bread(s: io.stream, buf: []u8) (size | io.eof | io.error) = {
	let b: *stream = s: *stream;
	// Degenerate read-disabled mode (zero-length rbuf, stream.ww:73-75);
	// ww's contract returns eof where Hare (always given a buffer) would
	// serve 0 — a separate concern from the top-up below.
	if (b.rbuf.len == 0) {
		let e: io.eof; return e;
	};
	let avail: i32 = b.rend - b.rstart;
	if (avail < buf.len && avail < b.rbuf.len) {
		bufiomove(b.rbuf.ptr, b.rbuf.ptr + (b.rstart: u64), avail);
		b.rstart = 0;
		b.rend = avail;
		let r: (size | io.eof | io.error) = io.read(b.src, b.rbuf[b.rend:b.rbuf.len]);
		match (r) {
		case let n: size => { b.rend += n: i32; };
		case io.eof => {
			if (avail == 0) { let e: io.eof; return e; };
		};
		case let e: io.error => return e;
		};
	};
	avail = b.rend - b.rstart;
	let n: i32 = buf.len;
	if (avail < n) { n = avail; };
	bufiomove(buf.ptr, b.rbuf.ptr + (b.rstart: u64), n);
	b.rstart += n;
	return n: size;
};

// bwrite — buffered write. wbuf-empty passes through to src; otherwise
// default-flush scan + per-batch copy + post-write conditional flush.
// Labeled-break inlined (ww has no labeled break). Mirrors
// ref/hare/bufio/stream.ha (stream_write).
fn bwrite(s: io.stream, buf: []u8) (size | io.error) = {
	let b: *stream = s: *stream;
	if (b.wbuf.len == 0) {
		return io.write(b.src, buf);
	};
	let doflush: bool = false;
	if (b.flush.len != 0) {
		let i: i32 = 0;
		for (i < buf.len) {
			let j: i32 = 0;
			for (j < b.flush.len) {
				if (buf[i] == b.flush[j]) {
					doflush = true;
					i = buf.len;
					j = b.flush.len;
				};
				j += 1;
			};
			i += 1;
		};
	};
	let z: i32 = 0;
	for (z < buf.len) {
		let avail: i32 = b.wbuf.len - b.wend;
		if (avail == 0) {
			let fr: (void | io.error) = flush(b);
			match (fr) {
			case void => { };
			case let e: io.error => return e;
			};
			avail = b.wbuf.len;
		};
		let n: i32 = buf.len - z;
		if (avail < n) { n = avail; };
		bufiomove(b.wbuf.ptr + (b.wend: u64),
			buf.ptr + (z: u64), n);
		b.wend += n;
		z += n;
	};
	if (doflush) {
		let fr: (void | io.error) = flush(b);
		match (fr) {
		case void => { };
		case let e: io.error => return e;
		};
	};
	return buf.len: size;
};

// bclose — flush pending wbuf only; the underlying src is NOT closed.
// Hare's close_buffered closes src solely under flag::MANAGED_HANDLE
// (ref/hare/bufio/stream.ha:194); init's default flag::NONE
// (stream.ha:73) flushes and leaves src to its owner. Close-propagation
// rides the io fold-2 MANAGED_HANDLE machinery (#5), per the ownership
// header (stream.ww:46-50) — caller owns src, bclose frees/closes nothing.
fn bclose(s: io.stream) (void | io.error) = {
	let b: *stream = s: *stream;
	return flush(b);
};