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.

Drop discards every change that arrives while a flight is in progress. The change runs no body and starts no flight, and nothing is replayed when the flight settles: its value or its error becomes the memo's, and the memo is current until a source changes again. The first change after the settle starts a flight as usual. The settled value can belong to inputs the graph has since left. The settle brings the memo's sources current: a memo source left stale by a discarded change recomputes then, and an async memo source starts its flight then, without waiting for a read. That source's own settle is a change after the memo's settle and runs the memo. Use it for a submit that ignores repeated clicks while the request is in flight, as R3's AwaitOperation.Drop and Rx's exhaustMap do.

DropWhilePending discards the changes that arrive while a flight is pending, as Drop does, and keeps a successful flight's value. A failed flight that discarded a change does not publish its error: the memo stays pending and runs once more against the current inputs, so a failure always belongs to the latest inputs. The replacing run starts at the memo's next read, and its previous value is the value published before the failed flight. A failed flight with no change during it settles to its error, and so does a replacing run that fails with no change during it. Use it where a stale success is acceptable and a stale failure is not, such as a refresh that a newer query may repair.

The following names describe related scheduling choices in R3 and SignalsDotnet. They are scheduling parallels, not interchangeable APIs or identical state semantics: stream operators and computed signals do not share Ranvier's demand-driven flights, Previous.Settled or propagating pending channel.

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, under FinishCurrent a trailing one, and under Drop and DropWhilePending none while a flight is in progress. For commands, C# has ReactiveCommand, whose CommandPolicy.Disable provides analogous command gating to AllowConcurrentExecutions = false; this is not an async-memo drop policy. See C#. Debounce and throttle are implemented as timed values. Read one from the async memo to control when requests start, and choose the flight policy separately to handle requests already running.

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, Drop and DropWhilePending 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, Drop and DropWhilePending 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 every policy except Queue, Settled is complete when the body runs. Under CancelPrevious and KeepLatest, 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 Drop, the next run receives the settled flight's value; under DropWhilePending, a replacing run receives the value published before the failed flight. 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.