Skip to content

Allow multiple writes to a Stream in a single Cmd.batch - #132

Open
dkoontz wants to merge 2 commits into
gren-lang:mainfrom
dkoontz:stream-simultaneous-writes
Open

Allow multiple writes to a Stream in a single Cmd.batch#132
dkoontz wants to merge 2 commits into
gren-lang:mainfrom
dkoontz:stream-simultaneous-writes

Conversation

@dkoontz

@dkoontz dkoontz commented Jul 31, 2026

Copy link
Copy Markdown
Contributor

Allow multiple writes in a single Cmd.batch without encountering Locked errors

Depends on #133

Problem

When two or more write/enqueue/close operations on the same WritableStream were issued in the same tick (e.g. via Cmd.batch or Task.concurrent), the second would fail with a spurious Locked error. Each operation checked stream.locked and called getWriter() synchronously, but only released the lock inside a later microtask (writer.ready.then(...)). The second getWriter() therefore ran against a still-held lock. This made it unsafe to do things like write in multiple places to stdout or to a log file stream.

Fix

Serialize all writer-acquiring operations on a given stream through a per-stream FIFO promise chain, so each operation only acquires its writer after the previous operation's queued promise has either fulfilled or rejected.

src/Gren/Kernel/Stream.js

  • _Stream_writeChains (WeakMap): one FIFO chain per WritableStream. Keyed by the stream object, so independent streams don't block each other and dead streams are GC'd.
  • _Stream_runChained(stream, work): appends work to the chain and stores run.then(noop, noop) back into the WeakMap so the next operation's prev is always fulfilled. Without these recovery handlers, a single rejected work would propagate down the chain and cause every subsequent prev.then(work) to skip its work entirely (a fulfillment handler never fires on a rejected promise), permanently breaking the stream after one failure.
  • _Stream_rejectLocked() + _Stream_reportRun(...): the stream.locked check now runs inside the queued work (not synchronously at the top), so it reflects the real lock state at execution time. A { __grenStreamLocked: true } sentinel is mapped back to Stream.Locked; everything else maps to Stream.Cancelled.
  • _Stream_toUint8Array: factored out the repeated DataViewUint8Array conversion.
  • _Stream_write, _Stream_enqueue, and _Stream_closeWritable were all rewritten on top of these helpers. write still awaits the writer.write() promise (backpressure-aware); enqueue still drops it (fire-and-forget into the buffer) respecting the semantics specified in src/Stream.gren.

Prerequisite commit (cc140eb)

_Stream_closeWritable was fixed to await writer.close() before releasing the lock and to surface close errors as Cancelled, instead of firing-and-forgetting. This is required for the serialization work to behave correctly and is included in the PR.

Tests

New integration_tests/src/Test/Stream.gren (registered in Main.gren) covers:

  • Concurrency — two writes / two enqueues / write+enqueue issued in the same tick all succeed (the regression this PR fixes).
  • Ordering — write-then-close succeeds; close-then-write fails with Cancelled (close semantics unchanged).
  • Genuine Locked — writes against a writable held by an active pipeThrough or pipeTo still correctly report Locked, confirming the FIFO chain didn't swallow real lock conflicts.

Risk / Notes

  • Behavior is unchanged for single-operation usage and for genuinely locked streams; the change only affects the interleaving of concurrent operations on one stream.
  • enqueue continues to resolve before the sink has processed the chunk (per its doc), so write errors on the dropped writer.write() promise still won't flow through _Stream_reportRun but that's pre-existing documented behavior.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant