Class AsyncConcurrencyStrategy<TInput, TState>
The base class of a MapAsync scheduling strategy, which is the admission
and the sequence of a stream of async requests. A strategy answers two
questions, and each answer is data.
Admit(TState, AsyncQueuedItem<TInput>, IReadOnlyList<AsyncTrackedItem<TInput>>) answers "which items start now for this new tracked
value?" and OnCompleted(TState, AsyncQueuedItem<TInput>, AsyncCompletion, IReadOnlyList<AsyncTrackedItem<TInput>>) answers "which items start next for
this outcome, and does the pipeline publish it?". A strategy answers only
from a TState that it controls. A strategy instance
holds no state of its own, because each value that changes is in
TState. CreateState() makes one
instance of that state for each MapAsync call. Thus, more than one MapAsync
call can use the same strategy instance safely, at the same time. The
execution engine (see
MapAsyncImpl<TInput, TResult, TStrategyInput>(Stream<TInput>, StreamSink<TResult>, StreamSink<Exception>, MapAsyncOperation<TInput, TResult>, AsyncConcurrencyStrategyBase<TStrategyInput>, Func<TInput, TStrategyInput>, Stream<UnitInternal>, Stream<IReadOnlyCollection<TInput>>, bool)
) holds the TState of each call, and it is the only
code that gives that state back to the strategy. Admit(TState, AsyncQueuedItem<TInput>, IReadOnlyList<AsyncTrackedItem<TInput>>) and
OnCompleted(TState, AsyncQueuedItem<TInput>, AsyncCompletion, IReadOnlyList<AsyncTrackedItem<TInput>>) cannot use the result sink, the error sink, or a
Task, and cannot start a Task. The two methods give a description of the
necessary operations, and the execution engine does them. The one
imperative operation is
Cancel(), which cancels
an item that the strategy controls. It publishes nothing and starts
nothing. It goes to the cancellation path of an external cancelAll stream,
thus the item ends through OnCompleted(TState, AsyncQueuedItem<TInput>, AsyncCompletion, IReadOnlyList<AsyncTrackedItem<TInput>>) as each other item
does. The identity of an item at its end is the item that the strategy got
in Admit(TState, AsyncQueuedItem<TInput>, IReadOnlyList<AsyncTrackedItem<TInput>>). There is no second handle. Keep the item in
TState to identify it again after this.
[PublicAPI]
public abstract class AsyncConcurrencyStrategy<TInput, TState> : AsyncConcurrencyStrategyBase<TInput>Type Parameters
TInputTState
- Inheritance
-
AsyncConcurrencyStrategyBase<TInput>AsyncConcurrencyStrategy<TInput, TState>
- Derived
- Inherited Members
- Extension Methods
Methods
- Admit(TState, AsyncQueuedItem<TInput>, IReadOnlyList<AsyncTrackedItem<TInput>>)
Gives the items to start now for a new admitted value. The pipeline tracks that value with the Queued status. See AsyncMapBase.AsyncQueuedItem<TInput>. The engine always calls this in a SodaFlow transaction. See SodaFlow.Async.AsyncMapExecutionManager<TInput, TResult, TStrategyInput> for the cause that makes a change to
statehere safe with no explicit lock. Return an AsyncMapBase.AsyncToStart<TInput> withincomingto start it immediately. Without that, the item keeps the Queued status. To start a Queued item after this, keepincominginstate, and not only its Value. That keeps its identity and its cancellation during the wait, and it is the value that OnCompleted(TState, AsyncQueuedItem<TInput>, AsyncCompletion, IReadOnlyList<AsyncTrackedItem<TInput>>) gives back.trackedgives the same items, thus a strategy that reads only the queue of the pipeline can keep no queue of its own.
- CreateState()
Makes a new scheduling state for one MapAsync call. The engine calls this one time for each call. See MapAsyncImpl<TInput, TResult, TStrategyInput>(Stream<TInput>, StreamSink<TResult>, StreamSink<Exception>, MapAsyncOperation<TInput, TResult>, AsyncConcurrencyStrategyBase<TStrategyInput>, Func<TInput, TStrategyInput>, Stream<UnitInternal>, Stream<IReadOnlyCollection<TInput>>, bool) Thus, two pipelines with the same strategy instance never read the state of the other, also when the two run at the same time.
- OnCanceled(TState, IReadOnlyList<AsyncQueuedItem<TInput>>, IReadOnlyList<AsyncTrackedItem<TInput>>)
Gives the tracked items to start because a cancellation ended one item or more. There is no publish decision here: a canceled item sends nothing to the results and nothing to the errors. A cancellation is always an expected end with no message. Its source is external code, a strategy that replaces its own previous run, or a Queued item that a cancellation removes before its turn.
The engine calls this one time for the cancellations of one transaction. One send of a cancellation stream can end each Queued item and each Running item whose operation observes its token, all at one instant. One call gives one decision over the queue that holds no item of those ends. Thus, a strategy does not start an item that is about to end.
A strategy that holds a reference to its current run clears that reference here. Otherwise, the last value stays in memory after the pipeline becomes empty.
- OnCompleted(TState, AsyncQueuedItem<TInput>, AsyncCompletion, IReadOnlyList<AsyncTrackedItem<TInput>>)
Gives two answers for one item whose operation returned or threw: if the pipeline publishes that outcome, and which tracked items start. For example, the next Queued item can start because of that end. The engine always calls this in a SodaFlow transaction, as it calls Admit(TState, AsyncQueuedItem<TInput>, IReadOnlyList<AsyncTrackedItem<TInput>>). Each AsyncMapBase.AsyncToStart<TInput> from this method must contain an AsyncMapBase.AsyncQueuedItem<TInput> from a previous Admit(TState, AsyncQueuedItem<TInput>, IReadOnlyList<AsyncTrackedItem<TInput>>) call. There is no path for a value with no admission.
One item, and never two. A stream carries one value for each transaction, and the edit of the queue for this end is in the transaction that sends its result. Thus, one such end is the most that a transaction can hold. A cancellation is the one end that comes in a group, and it goes to OnCanceled(TState, IReadOnlyList<AsyncQueuedItem<TInput>>, IReadOnlyList<AsyncTrackedItem<TInput>>).