Skip to main content

MEP-45 research note 09, Streams and agents

Author: research pass for MEP-45. Date: 2026-05-22 (GMT+7).

Mochi has first-class stream<T> values, stream definitions that declare event types, on handlers that subscribe to events, and agent records with intent methods that wrap a mailbox-style actor model. This note covers their lowering.

1. Surface recap​

From examples/v0.4/stream.mochi and examples/v0.5/agent.mochi and the language docs:

stream prices {
  symbol: string
  price: float
  at: time
}

on prices as p where p.symbol == "AAPL" {
  print("AAPL @ ", p.price)
}

agent trader {
  position: int = 0

  intent buy(symbol: string, qty: int) {
    position = position + qty
  }

  intent sell(symbol: string, qty: int) {
    position = position - qty
  }

  on prices as p {
    if p.price > 200 {
      sell(p.symbol, 1)
    }
  }
}

emit prices { symbol: "AAPL", price: 210.5, at: now() }

2. Stream lowering​

A stream<T> is a multi-producer multi-consumer broadcast channel plus an immutable schema descriptor. The C representation:

typedef struct mochi_stream__T {
    pthread_mutex_t       lock;        // or platform equivalent
    pthread_cond_t        not_empty;
    mochi_list__T         ring;        // bounded ring buffer
    size_t                head;
    size_t                cap;
    mochi_list__sub__T    subs;        // registered subscribers
    atomic_uint           flags;       // closed?
} mochi_stream__T;

Operations:

mochi_stream__T *mochi_stream__T_new(size_t cap);
void             mochi_stream__T_emit(mochi_stream__T *s, T v);
mochi_sub__T    *mochi_stream__T_subscribe(mochi_stream__T *s, void *env,
                                            void (*cb)(void *env, T v));
void             mochi_stream__T_unsubscribe(mochi_sub__T *sub);
void             mochi_stream__T_close(mochi_stream__T *s);

emit synchronously fans out to every subscriber on a worker thread from the scheduler pool (not the emitter's thread, so that emit is non-blocking from the caller's perspective). Backpressure: if the ring is full, emit blocks the caller until a slot frees.

Optionally, a stream can be configured lossy (stream prices @lossy {...}); under backpressure the oldest entry is dropped. Not in v1.

3. on handler lowering​

A free-standing on STREAM as x where PRED { BODY } lowers to:

static void mochi_on__0(void *env, struct pkg_Price p) {
    if (!(/* PRED */)) return;
    /* BODY */
}

/* At module init: */
mochi_stream__Price_subscribe(prices, NULL, mochi_on__0);

Module init is generated by the codegen: each module produces a pkg__init(void) function that wires up its on handlers, registers its agents, and runs its top-level let bindings.

The order is: literal bindings -> agent registrations -> on subscriptions. The reason: on handlers may fire as soon as subscribed, so any state they touch must be initialised first.

4. Agent lowering​

An agent record lowers to a record type plus a per-agent mailbox:

struct pkg_trader {
    mochi_int position;
    mochi_mailbox *mbox;
};

/* Intents lower to typed messages: */
typedef enum {
    PKG_TRADER_MSG__buy,
    PKG_TRADER_MSG__sell,
} pkg_trader_msg_tag;

typedef struct {
    pkg_trader_msg_tag tag;
    union {
        struct { mochi_str symbol; mochi_int qty; } buy;
        struct { mochi_str symbol; mochi_int qty; } sell;
    } u;
} pkg_trader_msg;

The agent's "run loop" runs on a dedicated lightweight fiber from the scheduler:

static void pkg_trader__loop(struct pkg_trader *self) {
    for (;;) {
        pkg_trader_msg m;
        if (!mochi_mailbox_recv(self->mbox, &m, sizeof m)) break; // closed
        switch (m.tag) {
        case PKG_TRADER_MSG__buy:
            self->position = self->position + m.u.buy.qty;
            break;
        case PKG_TRADER_MSG__sell:
            self->position = self->position - m.u.sell.qty;
            break;
        }
    }
}

Invoking t.buy("AAPL", 10) from any thread translates to a send on the mailbox; the agent processes it in order. This matches the Erlang/Akka actor model. The intent call returns immediately (fire- and-forget); a intent fn variant that returns a value is a synchronous request/reply, implemented as a one-shot reply channel.

4.1 Agent on handlers​

When an agent body contains on STREAM as p { BODY }, the subscription's callback enqueues a synthetic message onto the agent's mailbox; the agent thread processes it. This means stream callbacks do not race with intent handlers on the same agent. The agent's loop is the single-writer for its state.

4.2 Agent fields are private​

The lowering treats agent fields as accessible only from inside the agent body (intent methods and on handlers). The codegen rejects t.position = 5 outside the agent. Reads t.position from outside go through a generated atomic getter that snapshots the field under a lock, since the agent loop may write at any time.

5. The scheduler​

Architecture: M:N work-stealing scheduler.

  • M worker threads, M = mochi_runtime_workers() (default: number of hardware threads).
  • Each worker has a local FIFO deque of runnable fibers.
  • Idle workers steal from peers (Chase-Lev deque).
  • Blocking syscalls (file I/O, sleep) execute on an overflow pool of threads created on demand, freeing the worker.

Fiber backend: minicoro (note 04 §2.1).

Reference: Go runtime scheduler (Pike, Cox); Tokio worker pool (Lerche). We do not implement Go's preemption (signal-based) in v1; tight loops that never await or block can monopolise a worker. The MEP body should note this trade.

5.1 Fiber lifecycle​

  • spawn creates a new fiber with a 32 KB initial stack (grows on page fault if the OS allows; otherwise a fixed 1 MB).
  • A fiber holds its captured env on the heap.
  • On exit, the fiber's stack and env are released to the allocator.

5.2 Scheduling fairness​

We use a per-worker run queue with a global overflow queue checked every N steps to prevent starvation. N defaults to 61 (a prime, to break resonance with periodic patterns).

6. Time and ticks​

The runtime advances mochi_time from the OS monotonic clock. There is no virtual time in v1. Tests that need deterministic time can inject a mochi_time_provider via --test-clock=fixed:2026-05-22T00:00:00Z.

7. emit and back-pressure​

emit stream { ... } synchronously inserts into the stream's ring. If the ring is full, the caller blocks (this is the v1 default; it preserves the "no dropped events" property). A @lossy annotation on the stream changes this to drop-oldest.

8. Stream from query​

from p in prices ... select ... against a stream produces another stream, not a list, when the result type is stream<T>. The implementation:

mochi_stream__Out *out = mochi_stream__Out_new(16);
mochi_stream__Price_subscribe(prices, out, fwd_cb);

where fwd_cb runs the predicates and projections and emits to out. This is the "compose streams" pattern.

limit n on a stream completes the result stream after n items and unsubscribes from the source.

9. Channels​

The language docs surface channels indirectly through the stream API. The runtime exposes channels too:

mochi_chan__T *mochi_chan__T_new(size_t cap);
bool           mochi_chan__T_send(mochi_chan__T *c, T v);   // blocks if full
bool           mochi_chan__T_recv(mochi_chan__T *c, T *v);  // blocks if empty
void           mochi_chan__T_close(mochi_chan__T *c);

The difference from streams: a channel is point-to-point (one consumer per message), while a stream broadcasts to all subscribers. Internally, a stream is a list of channels + a fanout fiber.

10. Shutdown​

On main's return:

  1. Close all streams (mochi_stream__T_close) so subscribers see end-of-stream.
  2. Join all agent fibers (mochi_mailbox_close then wait).
  3. Drain pending I/O.
  4. Run any defer blocks in reverse order.
  5. Free GC roots and exit.

Step ordering is documented in the MEP body so user code can rely on it (in particular: print from agent shutdown handlers is flushed).

11. Open questions​

  1. Whether emit is synchronous or asynchronous from the caller's perspective when the ring has space. (Spec is ambiguous; v1 will be sync.)
  2. Whether agent state can be inspected from outside (atomic snapshot) or only via intent methods.
  3. Whether on STREAM inside agents and at module top-level have the same execution-thread semantics.
  4. Whether stream subscriptions are weak (auto-unsubscribe on subscriber GC) or strong (manual unsubscribe).
  5. Whether the scheduler should preempt CPU-bound fibers (Go-style signal preemption) in v1. Current answer: no.