Async memos

pumpUntil is this page's helper: it calls graph.Pump () until the condition holds.

let pumpUntil (g: Graph) (cond: unit -> bool) (timeout: TimeSpan) =
    let sw = Diagnostics.Stopwatch.StartNew ()

    while not (cond ()) do
        if sw.Elapsed > timeout then
            failwithf "condition not met within %O" timeout

        g.Pump () |> ignore
        Threading.Thread.Sleep 1

The same boundary applies to creation. createAsync's purity check and createAsyncWith's scope cover the body up to its first await that suspends. A node created in a continuation, onCleanup included, attaches to the owner current when the continuation runs, usually the graph's root, and lives until that owner is disposed.

A body that throws before its first await, and a flight whose task faults, both settle the memo as Failed with the original exception. A flight whose task is cancelled by its own IO, such as an HttpClient timeout, settles as Failed with the OperationCanceledException. A cancellation of a superseded flight publishes nothing.

let failGraph = newGraph ()

let refused =
    failGraph.Run (fun () -> createAsync (fun _ _ -> failwith "no connection" : Task<int>))

let faulting = TaskCompletionSource<int> ()
let faulted = failGraph.Run (fun () -> createAsync (fun _ _ -> faulting.Task))
faulted.TryValue |> ignore
faulting.SetException (TimeoutException "timed out")
pumpUntil failGraph (fun () -> not (faulted.Status.HasFlag Status.Pending)) (TimeSpan.FromSeconds 5.)

let show (reading: Reading<'T>) =
    match reading with
    | Ready value -> sprintf "Ready %A" value
    | Pending -> "Pending"
    | Failed error -> sprintf "Failed %s: %s" (error.GetType().Name) error.Message

show refused.TryValue, show faulted.TryValue

("Failed Exception: no connection", "Failed TimeoutException: timed out")

CancelPrevious, KeepLatest and Queue start a new flight as soon as a changed memo is read, and decide only the fate of the flights already in progress. The memo observes every flight's exception, including a superseded flight's and one raised after the memo or its graph was disposed, so TaskScheduler.UnobservedTaskException receives none of them.

FinishCurrent lets the flight in progress finish. A change during it runs no body and starts no flight, and the memo stays pending. When the flight settles, the memo runs once more against the current inputs, however many changes arrived: its value becomes the Peek value and the trailing run's previous value, and a failure is discarded. The trailing run starts at the memo's next read, so a memo read by an effect starts it as soon as the flight settles. A flight with no change during it applies as under KeepLatest. Use it for work that must not be abandoned halfway, such as a save, where the settled value must still match the latest inputs.

The same policies under the names other .NET libraries use. A name appears only where its behaviour matches exactly:

FinishCurrent is closest to SignalsDotnet's ScheduleNext and R3's ThrottleFirstLast. Both publish the finished run's result; FinishCurrent keeps the memo pending until the trailing run settles, so a published value always belongs to the current inputs.

Ranvier does not implement these policies:

CommunityToolkit's AsyncRelayCommand has no async-memo counterpart. AllowConcurrentExecutions is a gate on CanExecute for a command with no result: false reports the command as not executable while it runs, and true lets executions overlap. An async memo has no CanExecute: every change starts a new flight, or under FinishCurrent a trailing one. For commands, C# has ReactiveCommand, whose CommandPolicy.Disable matches AllowConcurrentExecutions = false; see C#. Debounce and throttle are not implemented either, as a policy or as a combinator.

Cancellation costs one CancellationTokenSource per flight under CancelPrevious: each launch cancels and disposes the superseded flight's source and allocates the next. Under KeepLatest and Queue, overlapping flights share one source. Under FinishCurrent flights never overlap, and each flight allocates one source, disposed without being cancelled when the flight settles. A change during a flight allocates nothing. It is disposed, without being cancelled, once every flight that holds it has settled, and the next flight allocates another; disposing the memo cancels it. A body that ignores its token pays that allocation and a Cancel with no registered callbacks. A body that registers on the token, as HttpClient does, also pays for the callbacks the Cancel runs. The token belongs to the async memo alone: signals, memos and effects carry none, and their reads take no token.

A registration on the token lives as long as its source. use _ = token.Register ..., and every API that takes the token and completes (Task.Delay, HttpClient, SemaphoreSlim.WaitAsync, cancellableTask binds), releases its registration when the operation ends. A registration left undisposed is released with the source: under CancelPrevious at the next launch, under FinishCurrent when its flight settles, under KeepLatest and Queue once the flights sharing the source have all settled, so a stream of overlapping flights keeps each one's registration until the stream goes quiet. A flight that completes only on cancellation stays in progress under KeepLatest and Queue until the memo is disposed, together with everything its task and registrations hold; use CancelPrevious for such bodies. Work that outlives its flight and keeps the token sees a disposed source once the source is released: it is not cancelled when the memo is disposed, and token.WaitHandle throws ObjectDisposedException. FlightPolicyBenchmarks in the suspension bench compares a launch under CancelPrevious with one under KeepLatest, and FlightBenchmarks compares every policy, with and without changes during a flight.

Under Queue each outcome is readable as soon as it is applied, while later flights are still in progress: a new run makes the memo pending until the next outcome is applied. A body that throws before its first await fails in its turn, after the flights started before it. While the newest run waits on a pending source, an earlier flight's value becomes the Peek value and the memo stays pending.

Read every input, then await previous.Settled. Under CancelPrevious, KeepLatest and FinishCurrent, Settled is complete when the body runs. Under the first two, only the newest flight's result becomes a previous value; under FinishCurrent, a trailing run receives the value of the flight it waited for, or the value before it when that flight failed. Under Queue, Settled completes when the flight started before this one is applied, so the flights fold in start order. An await on it suspends, and reads after it are untracked. If the earlier flight fails or is dropped, Settled returns the value published before it. Disposing the memo completes a pending Settled with the value last published.

Settled creates its task when first read: one completed task, shared by every read until the memo next publishes, or under Queue, one pending task for a flight that waits on an earlier one. The async scenarios in the counter bench run within 0.1 % of their earlier instruction counts, with the same allocation.

The second page completes first. Which lists does the effect see, and in what order?

let queueGraph =
    new Graph (
        { GraphOptions.Default with
            Dispatcher = Some (ManualDispatcher () :> IGraphDispatcher)
            FlightPolicy = FlightPolicy.Queue }
    )

let page = queueGraph.Run (fun () -> createSignal 1)
let replies = Array.init 2 (fun _ -> TaskCompletionSource<string list> ())

let feed =
    queueGraph.Run (fun () ->
        createAsync (fun previous _ ->
            let n = page.Value // tracked: read before the first await

            task {
                let! items = replies[n - 1].Task
                let! earlier = previous.Settled
                return ValueOption.defaultValue [] earlier @ items
            }))

let pages = ResizeArray<string list> ()
queueGraph.Run (fun () -> createEffect (fun () -> pages.Add feed.Value))

page.Value <- 2
replies[1].SetResult [ "c"; "d" ]
replies[0].SetResult [ "a"; "b" ]
pumpUntil queueGraph (fun () -> pages.Count = 2) (TimeSpan.FromSeconds 5.)
List.ofSeq pages

[["a"; "b"]; ["a"; "b"; "c"; "d"]]

A disposed async memo reads as Failed with an ObjectDisposedException. Its readers are woken once to see the failure.