Signals and awaitables

Draft

Build awaitable coordination on one scheduler signal.

On this page · 2 sections

One scheduler signal

Signal.make() separates the firing capability from its one listener. Firing is idempotent and remembers an early signal.

Fire Before Await

test✓

  let (~signalsignal: Signal, ~listenerlistener: Signal.Listener) = Signal.makemake: pure fn!() -> (signal: Signal, listener: Signal.Listener)();
  Signal.firefire: pure fn!(signal: Signal) -> Unit(signalsignal: Signal);
  Signal.firefire: pure fn!(signal: Signal) -> Unit(signalsignal: Signal);
  await listenerlistener: Signal.Listener;

Firing schedules a suspended listener; it never resumes that fiber inline.

type SignalProbe = {
  mut events: Vector(String);
}

let recordrecord: fn(probe: Book.Concurrency.Signals.SignalProbe @ global, event: String @ global) -> Unit = fn(probeprobe: Book.Concurrency.Signals.SignalProbe, eventevent: String) {
  probe.events := Vector.pushpush: pure fn!('a)(items: Vector(String), item: String) -> Vector(String)(probeprobe: Book.Concurrency.Signals.SignalProbe.eventsevents: Vector(String), eventevent: String);
};

Fire Enqueues Listener

test✓

  let probeprobe: Book.Concurrency.Signals.SignalProbe: SignalProbe @ global = SignalProbe { events = []; };
  let (~signal: startedstarted: Signal, ~listener: startstart: Signal.Listener) = Signal.makemake: pure fn!() -> (signal: Signal, listener: Signal.Listener)();
  let (~signal: releaserelease: Signal, ~listener: releasedreleased: Signal.Listener) = Signal.makemake: pure fn!() -> (signal: Signal, listener: Signal.Listener)();
  let childchild: Async.Task(Unit) = spawn {
    Signal.firefire: pure fn!(signal: Signal) -> Unit(startedstarted: Signal);
    await releasedreleased: Signal.Listener;
    recordrecord: fn(probe: Book.Concurrency.Signals.SignalProbe @ global, event: String @ global) -> Unit(probeprobe: Book.Concurrency.Signals.SignalProbe, "child""child": String);
  };
  await startstart: Signal.Listener;
  Signal.firefire: pure fn!(signal: Signal) -> Unit(releaserelease: Signal);
  recordrecord: fn(probe: Book.Concurrency.Signals.SignalProbe @ global, event: String @ global) -> Unit(probeprobe: Book.Concurrency.Signals.SignalProbe, "parent""parent": String);
  await childchild: Async.Task(Unit);
  expect ✓probeprobe: Book.Concurrency.Signals.SignalProbe.eventsevents: Vector(String) == ["parent""parent": String, "child""child": String];

Any code a fiber runs can fire: a helper function, or a callback that another function calls.

let releaserelease: fn(signal: Signal) -> Unit = fn(signalsignal: Signal) {
  Signal.firefire: pure fn!(signal: Signal) -> Unit(signalsignal: Signal);
};

Fire From A Helper

test✓

  let (~signalsignal: Signal, ~listenerlistener: Signal.Listener) = Signal.makemake: pure fn!() -> (signal: Signal, listener: Signal.Listener)();
  let childchild: Async.Task(Unit) = spawn {
    releaserelease: fn(signal: Signal) -> Unit(signalsignal: Signal);
  };
  await listenerlistener: Signal.Listener;
  await childchild: Async.Task(Unit);

Fire From A Callback

test✓

  let (~signalsignal: Signal, ~listenerlistener: Signal.Listener) = Signal.makemake: pure fn!() -> (signal: Signal, listener: Signal.Listener)();
  let childchild: Async.Task(Unit) = spawn {
    Vector.iteriter: pure fn!('a)(items: Vector(Signal) @ 'm, ~f: fn(value: Signal @ 'm) -> Unit) -> Unit([signalsignal: Signal], f = (eacheach: Signal) => Signal.firefire: pure fn!(signal: Signal) -> Unit(eacheach: Signal));
  };
  await listenerlistener: Signal.Listener;
  await childchild: Async.Task(Unit);

Owner cancellation disposes a suspended listener before child cleanup. A later fire cannot resurrect the finished fiber.

Cancellation Disposes Listener

test✓

  let probeprobe: Book.Concurrency.Signals.SignalProbe: SignalProbe @ global = SignalProbe { events = []; };
  let (~signal: startedstarted: Signal, ~listener: startstart: Signal.Listener) = Signal.makemake: pure fn!() -> (signal: Signal, listener: Signal.Listener)();
  let (~signal: parkedparked: Signal, ~listener: parkpark: Signal.Listener) = Signal.makemake: pure fn!() -> (signal: Signal, listener: Signal.Listener)();
  let parentparent: Async.Task(Unit) = spawn {
    let _child_child: Async.Task(Unit) = spawn {
      defer recordrecord: fn(probe: Book.Concurrency.Signals.SignalProbe @ global, event: String @ global) -> Unit(probeprobe: Book.Concurrency.Signals.SignalProbe, "cleaned""cleaned": String);
      Signal.firefire: pure fn!(signal: Signal) -> Unit(startedstarted: Signal);
      await parkpark: Signal.Listener;
      recordrecord: fn(probe: Book.Concurrency.Signals.SignalProbe @ global, event: String @ global) -> Unit(probeprobe: Book.Concurrency.Signals.SignalProbe, "resumed""resumed": String);
    };
    await startstart: Signal.Listener;
  };
  await parentparent: Async.Task(Unit);
  Signal.firefire: pure fn!(signal: Signal) -> Unit(parkedparked: Signal);
  expect ✓probeprobe: Book.Concurrency.Signals.SignalProbe.eventsevents: Vector(String) == ["cleaned""cleaned": String];

Canonical await behavior

await resolves the canonical Await behavior of its operand. Associated output types let task handles, one-shot values, listeners, and application types share one native operation without sharing a representation.

type Immediate =
  | Ready(Int)

module behavior Await(Immediate) {
  type Output = Int

  waitwait: fn(Book.Concurrency.Signals.Immediate) -> Int = function {
    | Ready(valuevalue: Int) -> valuevalue: Int
  };
}

Behavior Await

test✓

  ✓(await ReadyReady: fn(Int) -> Book.Concurrency.Signals.Immediate(4242: Int)): Int;
  expect ✓await ReadyReady: fn(Int) -> Book.Concurrency.Signals.Immediate(4242: Int) == 4242: Int;

A block remains an ordinary expression. It runs in the current fiber, leaves its lexical scope, and then its tail awaitable is awaited.

Await Block Value

test✓

  let probeprobe: Book.Concurrency.Signals.SignalProbe: SignalProbe @ global = SignalProbe { events = []; };
  let (~signalsignal: Signal, ~listenerlistener: Signal.Listener) = Signal.makemake: pure fn!() -> (signal: Signal, listener: Signal.Listener)();
  let childchild: Async.Task(Unit) = spawn {
    await listenerlistener: Signal.Listener;
    recordrecord: fn(probe: Book.Concurrency.Signals.SignalProbe @ global, event: String @ global) -> Unit(probeprobe: Book.Concurrency.Signals.SignalProbe, "child""child": String);
  };
  await {
    defer recordrecord: fn(probe: Book.Concurrency.Signals.SignalProbe @ global, event: String @ global) -> Unit(probeprobe: Book.Concurrency.Signals.SignalProbe, "block""block": String);
    Signal.firefire: pure fn!(signal: Signal) -> Unit(signalsignal: Signal);
    childchild: Async.Task(Unit)
  };
  expect ✓probeprobe: Book.Concurrency.Signals.SignalProbe.eventsevents: Vector(String) == ["block""block": String, "child""child": String];