diff --git a/.github/workflows/documentation.yml b/.github/workflows/documentation.yml new file mode 100644 index 0000000..cad762a --- /dev/null +++ b/.github/workflows/documentation.yml @@ -0,0 +1,73 @@ +name: Documentation + +on: + push: + branches: [ "main" ] + paths: + - "docs/**" + - "README.md" + - "TaskFlow/README.md" + - "TaskFlow.Extensions.Time/README.md" + - "TaskFlow.Extensions.Microsoft.DependencyInjection/README.md" + - "TaskFlow.Extensions.Microsoft.Logging/README.md" + - ".github/workflows/documentation.yml" + pull_request: + branches: [ "main" ] + paths: + - "docs/**" + - "README.md" + - "TaskFlow/README.md" + - "TaskFlow.Extensions.Time/README.md" + - "TaskFlow.Extensions.Microsoft.DependencyInjection/README.md" + - "TaskFlow.Extensions.Microsoft.Logging/README.md" + - ".github/workflows/documentation.yml" + workflow_dispatch: + +permissions: + contents: read + +concurrency: + group: "pages" + cancel-in-progress: false + +jobs: + build: + runs-on: ubuntu-latest + + steps: + - name: Checkout + uses: actions/checkout@v6 + + - name: Configure GitHub Pages + if: github.event_name != 'pull_request' + uses: actions/configure-pages@v5 + + - name: Build the rendered site + uses: actions/jekyll-build-pages@v1 + with: + source: ./docs + destination: ./_site + + - name: Upload GitHub Pages artifact + if: github.event_name != 'pull_request' + uses: actions/upload-pages-artifact@v4 + with: + path: ./_site + + deploy: + if: github.event_name != 'pull_request' + needs: build + runs-on: ubuntu-latest + + permissions: + pages: write + id-token: write + + environment: + name: github-pages + url: ${{ steps.deployment.outputs.page_url }} + + steps: + - name: Deploy to GitHub Pages + id: deployment + uses: actions/deploy-pages@v4 diff --git a/Directory.Build.props b/Directory.Build.props index fb8bc8e..0cd321b 100644 --- a/Directory.Build.props +++ b/Directory.Build.props @@ -12,7 +12,7 @@ - 1.0.0-rc4 + 1.0.0-rc5 Volodymyr Dombrovskyi Copyright (c) 2023 Volodymyr Dombrovskyi https://github.com/dombrovsky/TaskFlow.git diff --git a/README.md b/README.md index dae7e6a..1099c84 100644 --- a/README.md +++ b/README.md @@ -1,68 +1,87 @@ -# TaskFlow +# TaskFlow for .NET -**TaskFlow** is a robust, high-performance, extensible, and thread-safe library for orchestrating and controlling the execution of asynchronous tasks in .NET. It provides advanced patterns for sequential task execution, resource management, and cancellation, making it ideal for scenarios where you need more than just `SemaphoreSlim` or basic Task chaining. +TaskFlow turns calls from many places into one owned FIFO lane of work. Every submission gets an awaitable result while the lane serializes execution and provides a clear lifetime boundary. [![NuGet](https://img.shields.io/nuget/v/TaskFlow.svg)](https://www.nuget.org/packages/TaskFlow/) +[![Build](https://github.com/dombrovsky/TaskFlow/actions/workflows/build.yml/badge.svg)](https://github.com/dombrovsky/TaskFlow/actions/workflows/build.yml) [![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](LICENSE) ---- +## Why TaskFlow? -## Key Features +Applications often need to accept work asynchronously while ensuring that only one operation touches a resource at a time and that operations retain their original order. Building that around a semaphore or task chain leaves ordering, per-call completion, cancellation, error observation, and shutdown ownership in application code. -- **Sequential Task Execution:** Guarantee that tasks are executed in the order they are enqueued, with no concurrency unless explicitly configured. -- **Thread Affinity:** Run tasks on a dedicated thread, the current thread, or the thread pool, with full control over execution context. -- **Robust Disposal:** Dispose/DisposeAsync will only complete after all enqueued tasks have finished, ensuring clean shutdowns. This makes it ideal for managing fire-and-forget tasks by binding their lifetime to a specific scope. -- **Cancellation Support:** All enqueued task functions are executed, even if canceled before execution, ensuring predictable execution order. -- **SynchronizationContext Awareness:** Async/await inside enqueued delegates will execute continuations on the same `TaskFlow` if a `SynchronizationContext` is captured. -- **Extensibility:** Extend `TaskFlowBase` to create custom task flow implementations or use extension methods and wrappers to enhance functionality, such as throttling, error handling, or scoped cancellation. -- **Clean Task Pipeline Definition:** Define task pipelines separately from execution logic using extension methods from `System.Threading.Tasks.Flow.Extensions`, enabling better segregation of responsibilities and cleaner code. -- **Dependency Injection Integration:** Extensions for `Microsoft.Extensions.DependencyInjection` for easy registration and scoping. +TaskFlow packages those concerns into a reusable execution lane. It is useful when you need to: ---- +- expose an asynchronous API over a synchronous or non-thread-safe resource; +- preserve event order when synchronous callbacks initiate work; +- own background work within a component or dependency-injection scope; +- cancel obsolete operations when a newer request arrives; +- add timeouts, throttling, logging, annotations, or error observation without changing the work itself; or +- run ordered work on the thread pool, a dedicated thread, a caller-owned thread, or a custom scheduler. -## When Should You Use TaskFlow? +One flow is one sequential lane. Create separate flows for work that should proceed independently. -TaskFlow is ideal for scenarios where you need: +## Features -- **Serialized access to a resource** (e.g., database, file, hardware) from multiple async operations. -- **Order-preserving task execution** (e.g., message processing, event handling). -- **Thread affinity** (e.g., UI thread, dedicated worker thread). -- **Graceful shutdown** with guaranteed completion of all in-flight work. -- **Advanced error handling and cancellation patterns.** -- **Fire-and-forget task lifetime management:** Bind fire-and-forget operations to a scope by disposing the `TaskFlow` instance, ensuring proper cleanup and resource management. -- **Segregation of responsibilities:** Use extension methods to define task pipelines separately from execution logic, improving maintainability and readability. +| Feature | What it provides | +|---|---| +| FIFO execution | Accepted operations start in submission order and do not overlap within one flow. | +| Per-operation tasks | Every caller can await its own result, exception, or cancellation. | +| Owned lifetime | A flow gives queued and running work an explicit component-level shutdown boundary. | +| Composable policies | Add cancellation scopes, latest-request-wins behavior, timeouts, leading-edge throttling, operation names, interception, and error observation. | +| Execution choices | Use the thread pool, a dedicated thread, the current thread, or another `TaskScheduler`. | +| Application integration | Register scoped or named flows and emit structured lifecycle logs. | +| Extensibility | Build scheduler decorators, adapters, interceptors, or custom `TaskFlowBase` implementations. | ---- +See the [extension reference](https://dombrovsky.github.io/TaskFlow/extensions/) and [execution models](https://dombrovsky.github.io/TaskFlow/execution-models/) for the available policies and implementations. -## Getting Started +## Serialize synchronous work for asynchronous callers -### Installation +```csharp +using System.Threading.Tasks.Flow; -Add the core package: -`dotnet add package TaskFlow` +public interface IDataStore +{ + void Save(Data data); +} -For dependency injection support: -`dotnet add package TaskFlow.Microsoft.Extensions.DependencyInjection` +public sealed class SerializedStore(IDataStore inner) : IAsyncDisposable +{ + private readonly TaskFlow _flow = new(); -For Microsoft.Extensions.Logging integration: -`dotnet add package TaskFlow.Microsoft.Extensions.Logging` + public Task SaveAsync(Data data) => + _flow.Enqueue(() => inner.Save(data)); -### Building from Source + public ValueTask DisposeAsync() => _flow.DisposeAsync(); +} +``` -Use a .NET 10 SDK to restore, build, and test the repository. The test projects execute against both `net8.0` and `net10.0`, so both runtimes must be installed. +Callers receive a task instead of blocking on `Save`. The wrapped synchronous method runs once at a time and in call order, regardless of how many callers submit work concurrently. -### Basic Usage -```csharp -using var taskFlow = new TaskFlow(); +## Packages -// Enqueue tasks for sequential execution -var task1 = taskFlow.Enqueue(() => Console.WriteLine("Task 1")); -var task2 = taskFlow.Enqueue(async () => await Task.Delay(100)); -``` ---- +| Package | Purpose | +|---|---| +| [`TaskFlow`](https://www.nuget.org/packages/TaskFlow/) | FIFO execution lanes, built-in execution models, and core scheduler policies. | +| [`TaskFlow.Extensions.Time`](https://www.nuget.org/packages/TaskFlow.Extensions.Time/) | Compatibility package for the `WithThrottle` time-based policy. | +| [`TaskFlow.Microsoft.Extensions.DependencyInjection`](https://www.nuget.org/packages/TaskFlow.Microsoft.Extensions.DependencyInjection/) | Scoped, named, and customizable TaskFlow registrations. | +| [`TaskFlow.Microsoft.Extensions.Logging`](https://www.nuget.org/packages/TaskFlow.Microsoft.Extensions.Logging/) | Structured operation-lifecycle logging through `Microsoft.Extensions.Logging`. | + +## Lifecycle essentials + +- Await returned tasks when their outcome belongs to the caller; intentionally discarded work remains bounded by the flow and can report failures inside the operation or through a decorator when needed. +- Prefer `await using` so asynchronous disposal can wait for the lane to finish. +- Cancellation is cooperative, and synchronous disposal has a timeout. +- Scheduler decorators do not own the underlying flow; dispose the original `ITaskFlow`. + +Read [Concepts and lifecycle](https://dombrovsky.github.io/TaskFlow/concepts-and-lifecycle/) and [Semantics and pitfalls](https://dombrovsky.github.io/TaskFlow/semantics-and-pitfalls/) for the full behavior contract. -## Extensions +## Documentation -## License +- [Getting started](https://dombrovsky.github.io/TaskFlow/getting-started/) +- [Recipes](https://dombrovsky.github.io/TaskFlow/recipes/) +- [Extensions](https://dombrovsky.github.io/TaskFlow/extensions/) +- [Dependency injection](https://dombrovsky.github.io/TaskFlow/dependency-injection/) +- [Troubleshooting](https://dombrovsky.github.io/TaskFlow/troubleshooting/) -This library is licensed under the [MIT License](LICENSE). +TaskFlow is available under the [MIT License](LICENSE). Contributions and problem reports are welcome through [GitHub issues](https://github.com/dombrovsky/TaskFlow/issues). diff --git a/TaskFlow.Extensions.Microsoft.DependencyInjection/README.md b/TaskFlow.Extensions.Microsoft.DependencyInjection/README.md index 487d00b..0c47533 100644 --- a/TaskFlow.Extensions.Microsoft.DependencyInjection/README.md +++ b/TaskFlow.Extensions.Microsoft.DependencyInjection/README.md @@ -1,18 +1,14 @@ # TaskFlow.Microsoft.Extensions.DependencyInjection -`TaskFlow.Microsoft.Extensions.DependencyInjection` integrates TaskFlow with the Microsoft dependency-injection container. It provides scoped schedulers, automatic lifetime management, named configurations, and factories for explicitly owned flows. +This package integrates TaskFlow with `Microsoft.Extensions.DependencyInjection`, providing scoped FIFO execution lanes, named configurations, decorator chains, and factories for explicitly owned flows. -## Installation +## Install ```shell dotnet add package TaskFlow.Microsoft.Extensions.DependencyInjection ``` -This package references the core `TaskFlow` package. - -## Basic usage - -Register a scoped TaskFlow execution lane: +## Scoped flow ```csharp using Microsoft.Extensions.DependencyInjection; @@ -21,18 +17,16 @@ using System.Threading.Tasks.Flow; services.AddTaskFlow(); ``` -Inject `ITaskScheduler` into a scoped service and enqueue work through it: +Inject `ITaskScheduler` into scoped consumers: ```csharp public sealed class ReportWriter { private readonly ITaskScheduler _scheduler; - private readonly IReportStore _store; - public ReportWriter(ITaskScheduler scheduler, IReportStore store) + public ReportWriter(ITaskScheduler scheduler) { _scheduler = scheduler; - _store = store; } public Task SaveAsync( @@ -40,48 +34,24 @@ public sealed class ReportWriter CancellationToken cancellationToken = default) { return _scheduler.Enqueue( - token => _store.SaveAsync(report, token), + token => PersistAsync(report, token), cancellationToken); } -} -``` - -The container creates one TaskFlow for each dependency-injection scope and disposes it when that scope ends. - -## Named flows and factories - -Register named options when different consumers need different flow configurations: - -```csharp -services.AddTaskFlow(); -services.AddTaskFlow( - "imports", - new TaskFlowOptions - { - SynchronousDisposeTimeout = TimeSpan.FromSeconds(30) - }); -``` -Create an explicitly owned named flow through `ITaskFlowFactory`: - -```csharp -await using ITaskFlow importFlow = factory.CreateTaskFlow("imports"); -await importFlow.Enqueue(token => ImportAsync(token)); + private static Task PersistAsync( + Report report, + CancellationToken token) => Task.CompletedTask; +} ``` -The advanced `AddTaskFlow` overload can also provide a custom base-flow factory, dynamically resolved options, and a scheduler-decorator chain. +The scope owns and disposes its underlying flow. Do not dispose an injected scheduler. -## Lifetime notes +Named registrations select independently configured sequential flows. When the caller needs ownership, create an `ITaskFlow` through `ITaskFlowFactory` and dispose it with `await using`. -- `ITaskScheduler` and `ITaskFlowInfo` are registered as scoped services. -- `ITaskFlowFactory` and the default factory are registered as singletons. -- `ITaskFlow` is not registered directly. Use the scoped scheduler for container-owned work or `ITaskFlowFactory` when the caller needs to own and dispose a flow. -- Do not dispose an injected scoped scheduler; the dependency-injection scope owns its underlying TaskFlow. +## Documentation -## Links +- [Dependency injection guide](https://dombrovsky.github.io/TaskFlow/dependency-injection/) +- [Concepts and lifecycle](https://dombrovsky.github.io/TaskFlow/concepts-and-lifecycle/) +- [Extension composition](https://dombrovsky.github.io/TaskFlow/extensions/) -- [TaskFlow repository](https://github.com/dombrovsky/TaskFlow) -- [Dependency-injection source](https://github.com/dombrovsky/TaskFlow/tree/main/TaskFlow.Extensions.Microsoft.DependencyInjection) -- [Core TaskFlow package documentation](https://github.com/dombrovsky/TaskFlow/blob/main/TaskFlow/README.md) -- [License](https://github.com/dombrovsky/TaskFlow/blob/main/LICENSE) -- [Issues and feedback](https://github.com/dombrovsky/TaskFlow/issues) +Source, license, and feedback are available in the [TaskFlow repository](https://github.com/dombrovsky/TaskFlow). diff --git a/TaskFlow.Extensions.Microsoft.DependencyInjection/TaskFlow.Extensions.Microsoft.DependencyInjection.csproj b/TaskFlow.Extensions.Microsoft.DependencyInjection/TaskFlow.Extensions.Microsoft.DependencyInjection.csproj index 2354392..ba03501 100644 --- a/TaskFlow.Extensions.Microsoft.DependencyInjection/TaskFlow.Extensions.Microsoft.DependencyInjection.csproj +++ b/TaskFlow.Extensions.Microsoft.DependencyInjection/TaskFlow.Extensions.Microsoft.DependencyInjection.csproj @@ -3,6 +3,8 @@ netstandard2.0;net8.0;net10.0 System.Threading.Tasks.Flow TaskFlow.Microsoft.Extensions.DependencyInjection + Microsoft dependency-injection integration for TaskFlow with scoped and named FIFO execution lanes and caller-owned factories. + async;task-scheduler;task-queue;concurrency;sequential;dependency-injection;microsoft-extensions diff --git a/TaskFlow.Extensions.Microsoft.Logging/README.md b/TaskFlow.Extensions.Microsoft.Logging/README.md index 6219aab..da26aa4 100644 --- a/TaskFlow.Extensions.Microsoft.Logging/README.md +++ b/TaskFlow.Extensions.Microsoft.Logging/README.md @@ -1,16 +1,14 @@ # TaskFlow.Microsoft.Extensions.Logging -`TaskFlow.Microsoft.Extensions.Logging` adds structured lifecycle logging to any TaskFlow `ITaskScheduler` through `Microsoft.Extensions.Logging`. +This package adds structured `Microsoft.Extensions.Logging` lifecycle events to any TaskFlow `ITaskScheduler`. -## Installation +## Install ```shell dotnet add package TaskFlow.Microsoft.Extensions.Logging ``` -This package references the core `TaskFlow` package. - -## Basic usage +## Log operation lifecycles ```csharp using Microsoft.Extensions.Logging; @@ -21,44 +19,25 @@ await using var flow = new TaskFlow(); ITaskScheduler operations = flow .WithLogging(logger, options => { - options.EnqueuedLogLevel = LogLevel.Debug; options.StartedLogLevel = LogLevel.Information; options.SucceededLogLevel = LogLevel.Information; options.FailedLogLevel = LogLevel.Error; - options.FinishedLogLevel = LogLevel.Debug; }) .WithOperationName("orders.persist"); await operations.Enqueue(token => PersistOrdersAsync(token)); ``` -Place `WithOperationName` outside `WithLogging`, as shown above, so the logging decorator can read the operation-name annotation. - -## Logged lifecycle - -The decorator can emit an event when an operation is: - -- enqueued; -- starting; -- requested to cancel through its enqueue token; -- completed successfully; -- failed or observed cancellation; and -- finished, regardless of outcome. +The decorator can emit enqueued, started, cancellation-requested, succeeded or failed, and finished events. Events include structured operation IDs, optional operation names, durations where applicable, and failure exceptions. -All event levels default to `LogLevel.Trace`. Set any corresponding `TaskFlowLoggingOptions` property to `LogLevel.None` to disable that event. +Place `WithOperationName` outside `WithLogging`, as shown, so logging can read the annotation. Cancellation logging reports a request rather than the final outcome. Failures remain observable through the task returned by `Enqueue`. -## Behavior and lifetime +The logging decorator does not own the underlying flow. -- The cancellation-request event reports a request, not the final outcome. The operation can still complete successfully if it does not observe cancellation. -- Failures are logged with their exception and still propagate through the task returned by `Enqueue`. -- Each logging wrapper assigns increasing operation IDs to the operations it observes. -- `WithLogging` returns a scheduler decorator and does not own the underlying flow. Dispose the original `ITaskFlow`. -- Decorator order is observable; add operation names and other annotations outside the logging wrapper when the logger should include them. +## Documentation -## Links +- [Observability extensions](https://dombrovsky.github.io/TaskFlow/extensions/observability/) +- [Extension composition](https://dombrovsky.github.io/TaskFlow/extensions/) +- [Semantics and pitfalls](https://dombrovsky.github.io/TaskFlow/semantics-and-pitfalls/) -- [TaskFlow repository](https://github.com/dombrovsky/TaskFlow) -- [Logging extension source](https://github.com/dombrovsky/TaskFlow/tree/main/TaskFlow.Extensions.Microsoft.Logging) -- [Core TaskFlow package documentation](https://github.com/dombrovsky/TaskFlow/blob/main/TaskFlow/README.md) -- [License](https://github.com/dombrovsky/TaskFlow/blob/main/LICENSE) -- [Issues and feedback](https://github.com/dombrovsky/TaskFlow/issues) +Source, license, and feedback are available in the [TaskFlow repository](https://github.com/dombrovsky/TaskFlow). diff --git a/TaskFlow.Extensions.Microsoft.Logging/TaskFlow.Extensions.Microsoft.Logging.csproj b/TaskFlow.Extensions.Microsoft.Logging/TaskFlow.Extensions.Microsoft.Logging.csproj index de125da..06ac637 100644 --- a/TaskFlow.Extensions.Microsoft.Logging/TaskFlow.Extensions.Microsoft.Logging.csproj +++ b/TaskFlow.Extensions.Microsoft.Logging/TaskFlow.Extensions.Microsoft.Logging.csproj @@ -3,6 +3,8 @@ netstandard2.0;net8.0;net10.0 System.Threading.Tasks.Flow TaskFlow.Microsoft.Extensions.Logging + Structured Microsoft.Extensions.Logging lifecycle events for TaskFlow operations, including names, timing, cancellation, and failures. + async;task-scheduler;task-queue;logging;observability;microsoft-extensions diff --git a/TaskFlow.Extensions.Time/README.md b/TaskFlow.Extensions.Time/README.md index dd31012..c2ae5a3 100644 --- a/TaskFlow.Extensions.Time/README.md +++ b/TaskFlow.Extensions.Time/README.md @@ -1,18 +1,14 @@ # TaskFlow.Extensions.Time -`TaskFlow.Extensions.Time` is only needed on older .NET runtimes when you want to use time-based extension methods such as `WithThrottle`; on newer runtimes, these extensions are included directly in the core `TaskFlow` package. +`TaskFlow.Extensions.Time` supplies `WithThrottle` to consumers using TaskFlow's `netstandard2.0` asset through `Microsoft.Bcl.TimeProvider`. .NET 8 and .NET 10 consumers receive the same API directly from the core `TaskFlow` package. -`WithThrottle` provides leading-edge throttling: it accepts the first operation and rejects later operations submitted during the configured interval. - -## Installation +## Install ```shell dotnet add package TaskFlow.Extensions.Time ``` -This package references the core `TaskFlow` package and uses `Microsoft.Bcl.TimeProvider`. - -## Basic usage +## Leading-edge admission throttling ```csharp using System.Threading.Tasks.Flow; @@ -20,34 +16,28 @@ using System.Threading.Tasks.Flow; await using var flow = new TaskFlow(); ITaskScheduler throttled = flow.WithThrottle(TimeSpan.FromSeconds(1)); -await throttled.Enqueue(() => SendUpdateAsync()); +await throttled.Enqueue(token => SendUpdateAsync(token)); try { - await throttled.Enqueue(() => SendUpdateAsync()); + await throttled.Enqueue(token => SendUpdateAsync(token)); } catch (OperationThrottledException) { - // The second operation was submitted inside the one-second interval. + // Rejected inside the one-second admission interval. } ``` -After the interval elapses, the next submitted operation is accepted and starts a new interval. +The first submission is admitted immediately. Later submissions inside the interval fail with `OperationThrottledException` without reaching the wrapped scheduler. Admission time is recorded before execution, so accepted operations consume the interval even if they later fail or observe cancellation. + +This is leading-edge throttling, not trailing-edge debounce: rejected work is not delayed, queued, or replaced. Pass a custom `TimeProvider` for deterministic tests. -## Behavior +The returned scheduler is a decorator and does not own the underlying flow. -- The first operation is accepted immediately. -- An operation submitted before the interval elapses fails with `OperationThrottledException` and is not forwarded to the wrapped scheduler. -- The interval is checked when `Enqueue` is called, so time spent waiting in the wrapped scheduler does not delay the start of the interval. -- The accepted timestamp is recorded before the operation executes. An accepted operation that later fails or observes cancellation still consumes the interval. -- This is leading-edge throttling, not trailing-edge debouncing: rejected work is not delayed, queued, or replaced with the latest request. -- Pass a custom `TimeProvider` to `WithThrottle` for deterministic time control in tests. -- The returned scheduler is a decorator and does not own the underlying flow. Dispose the original `ITaskFlow`. +## Documentation -## Links +- [Reliability extensions](https://dombrovsky.github.io/TaskFlow/extensions/reliability/) +- [Semantics and pitfalls](https://dombrovsky.github.io/TaskFlow/semantics-and-pitfalls/) +- [Compatibility](https://dombrovsky.github.io/TaskFlow/compatibility/) -- [TaskFlow repository](https://github.com/dombrovsky/TaskFlow) -- [Time extension source](https://github.com/dombrovsky/TaskFlow/tree/main/TaskFlow.Extensions.Time) -- [Core TaskFlow package documentation](https://github.com/dombrovsky/TaskFlow/blob/main/TaskFlow/README.md) -- [License](https://github.com/dombrovsky/TaskFlow/blob/main/LICENSE) -- [Issues and feedback](https://github.com/dombrovsky/TaskFlow/issues) +Source, license, and feedback are available in the [TaskFlow repository](https://github.com/dombrovsky/TaskFlow). diff --git a/TaskFlow.Extensions.Time/TaskFlow.Extensions.Time.csproj b/TaskFlow.Extensions.Time/TaskFlow.Extensions.Time.csproj index f3e8d02..d14de4d 100644 --- a/TaskFlow.Extensions.Time/TaskFlow.Extensions.Time.csproj +++ b/TaskFlow.Extensions.Time/TaskFlow.Extensions.Time.csproj @@ -3,6 +3,8 @@ netstandard2.0 System.Threading.Tasks.Flow TaskFlow.Extensions.Time + Leading-edge admission throttling for TaskFlow on netstandard2.0 with TimeProvider-based deterministic timing. + async;task-scheduler;task-queue;concurrency;throttling;time-provider;netstandard diff --git a/TaskFlow/README.md b/TaskFlow/README.md index c0a19c7..d6e1a7b 100644 --- a/TaskFlow/README.md +++ b/TaskFlow/README.md @@ -1,8 +1,8 @@ -# TaskFlow +# TaskFlow for .NET -`TaskFlow` provides an owned FIFO execution lane for asynchronous .NET work. Use it when operations must run one at a time, in submission order, while each caller still receives a task for its own result or failure. +`TaskFlow` provides owned FIFO execution lanes for asynchronous .NET work. Use it when operations must run one at a time in submission order while each caller retains a task for its own result, failure, or cancellation. -## Installation +## Install ```shell dotnet add package TaskFlow @@ -15,34 +15,33 @@ using System.Threading.Tasks.Flow; await using var flow = new TaskFlow(); -Task first = flow.Enqueue(async cancellationToken => -{ - await SaveAsync("first", cancellationToken); -}); - -Task second = flow.Enqueue(async cancellationToken => -{ - await SaveAsync("second", cancellationToken); -}); +Task first = flow.Enqueue(token => SaveAsync("first", token)); +Task second = flow.Enqueue(token => SaveAsync("second", token)); await Task.WhenAll(first, second); ``` -`second` does not begin until `first` finishes. Calls to `Enqueue` are thread-safe, so multiple callers can share the same flow to serialize access to a resource. +`second` starts only after `first` finishes. Calls to `Enqueue` are thread-safe, and one failed operation does not stop later queued operations. + +## Important behavior -## Behavior and lifetime +- Cancellation is cooperative. Built-in flows invoke an accepted queued delegate even if its token was canceled while waiting. +- `DisposeAsync` requests cancellation and waits for the lane to finish. Synchronous disposal is bounded by `TaskFlowOptions.SynchronousDisposeTimeout`. +- Observe every returned operation task; disposal does not surface individual operation failures. +- Timeout includes time spent waiting in the underlying queue. +- Scheduler decorators do not own the flow they wrap. Dispose the original `ITaskFlow`. -- Operations execute sequentially in FIFO order. -- Each returned task reports the result, cancellation, or exception of its own operation. One failed operation does not stop later queued operations from running. -- Cancellation is cooperative. A queued delegate is still invoked when its token has already been canceled, allowing queue progression while giving the delegate the canceled token. -- `DisposeAsync` requests cancellation and waits for queued work to finish. Synchronous disposal waits up to `TaskFlowOptions.SynchronousDisposeTimeout`. -- Scheduler decorators such as timeout, cancellation-scope, error-observation, interception, and cancel-previous wrappers do not own the underlying flow. Dispose the `ITaskFlow` that created the execution lane. +TaskFlow also provides dedicated-thread and caller-owned-thread flows, cancellation and timeout policies, leading-edge throttling, error observation, annotations, and interception. -TaskFlow also includes `DedicatedThreadTaskFlow` and `CurrentThreadTaskFlow` for work that requires thread affinity. +## Documentation -## Links +- [Getting started](https://dombrovsky.github.io/TaskFlow/getting-started/) +- [Concepts and lifecycle](https://dombrovsky.github.io/TaskFlow/concepts-and-lifecycle/) +- [Semantics and pitfalls](https://dombrovsky.github.io/TaskFlow/semantics-and-pitfalls/) +- [Execution models](https://dombrovsky.github.io/TaskFlow/execution-models/) +- [Recipes](https://dombrovsky.github.io/TaskFlow/recipes/) +- [Extensions](https://dombrovsky.github.io/TaskFlow/extensions/) +- [Compatibility](https://dombrovsky.github.io/TaskFlow/compatibility/) +- [Troubleshooting](https://dombrovsky.github.io/TaskFlow/troubleshooting/) -- [TaskFlow repository](https://github.com/dombrovsky/TaskFlow) -- [Source code](https://github.com/dombrovsky/TaskFlow/tree/main/TaskFlow) -- [License](https://github.com/dombrovsky/TaskFlow/blob/main/LICENSE) -- [Issues and feedback](https://github.com/dombrovsky/TaskFlow/issues) +Source, license, and feedback are available in the [TaskFlow repository](https://github.com/dombrovsky/TaskFlow). diff --git a/TaskFlow/TaskFlow.csproj b/TaskFlow/TaskFlow.csproj index 95bef9e..8900c53 100644 --- a/TaskFlow/TaskFlow.csproj +++ b/TaskFlow/TaskFlow.csproj @@ -3,6 +3,8 @@ netstandard2.0;net8.0;net10.0 System.Threading.Tasks.Flow TaskFlow + A composable .NET async scheduler for FIFO execution, lifetime-bound background work, cancellation, timeouts, logging, and thread affinity. + async;task-scheduler;task-queue;concurrency;sequential;serialization;cancellation;thread-affinity;background-tasks diff --git a/_config.yml b/_config.yml deleted file mode 100644 index 0f8cb67..0000000 --- a/_config.yml +++ /dev/null @@ -1,8 +0,0 @@ -theme: minima -plugins: - - jekyll-relative-links -relative_links: - enabled: true - collections: true -include: - - README.md diff --git a/docs/_config.yml b/docs/_config.yml new file mode 100644 index 0000000..ce39af1 --- /dev/null +++ b/docs/_config.yml @@ -0,0 +1,17 @@ +title: TaskFlow for .NET +description: FIFO async execution lanes with owned lifetime, cancellation, diagnostics, and thread affinity. +url: https://dombrovsky.github.io +baseurl: /TaskFlow +repository: dombrovsky/TaskFlow +theme: minima +plugins: + - jekyll-relative-links +relative_links: + enabled: true + collections: true +header_pages: + - getting-started.md + - recipes.md + - execution-models.md + - extensions/index.md + - dependency-injection.md diff --git a/docs/compatibility.md b/docs/compatibility.md new file mode 100644 index 0000000..e0bd686 --- /dev/null +++ b/docs/compatibility.md @@ -0,0 +1,48 @@ +--- +layout: page +title: Compatibility +permalink: /compatibility/ +--- + +# Compatibility + +## Package and framework matrix + +| Package | Target frameworks | Purpose | +|---|---|---| +| `TaskFlow` | `netstandard2.0`, `net8.0`, `net10.0` | Core flows, scheduler contracts, cancellation, timeout, error observation, annotations, and interception | +| `TaskFlow.Extensions.Time` | `netstandard2.0` | `WithThrottle` for older targets using `Microsoft.Bcl.TimeProvider` | +| `TaskFlow.Microsoft.Extensions.DependencyInjection` | `netstandard2.0`, `net8.0`, `net10.0` | Scoped and named flow registration and factories | +| `TaskFlow.Microsoft.Extensions.Logging` | `netstandard2.0`, `net8.0`, `net10.0` | Structured operation lifecycle logging | + +The core package includes `WithThrottle` when targeting .NET 8 or .NET 10. Its `netstandard2.0` asset omits that source because `TimeProvider` is not part of the target framework. Install `TaskFlow.Extensions.Time` in a `netstandard2.0` consumer to receive the same public API through `Microsoft.Bcl.TimeProvider`. + +## Language and runtime use + +The packages can be consumed from compatible target frameworks regardless of the repository's build SDK. Repository builds currently use the .NET 10 SDK, and the test projects execute against .NET 8 and .NET 10. + +`IAsyncDisposable` support for the `netstandard2.0` asset is supplied through `Microsoft.Bcl.AsyncInterfaces`. Prefer `await using` where the consuming language and runtime support it. + +## Package selection examples + +For a .NET 8 or .NET 10 application: + +```shell +dotnet add package TaskFlow +``` + +For a library targeting `netstandard2.0` that needs throttling: + +```shell +dotnet add package TaskFlow +dotnet add package TaskFlow.Extensions.Time +``` + +Add integration packages only when needed: + +```shell +dotnet add package TaskFlow.Microsoft.Extensions.DependencyInjection +dotnet add package TaskFlow.Microsoft.Extensions.Logging +``` + +All TaskFlow APIs use the `System.Threading.Tasks.Flow` namespace. Logging configuration also uses `Microsoft.Extensions.Logging`, and registration uses `Microsoft.Extensions.DependencyInjection`. diff --git a/docs/concepts-and-lifecycle.md b/docs/concepts-and-lifecycle.md new file mode 100644 index 0000000..05266ab --- /dev/null +++ b/docs/concepts-and-lifecycle.md @@ -0,0 +1,66 @@ +--- +layout: page +title: Concepts and lifecycle +permalink: /concepts-and-lifecycle/ +--- + +# Concepts and lifecycle + +## Scheduler and flow + +`ITaskScheduler` is the minimal submission contract. It accepts a delegate, optional state, and cancellation token, then returns a task for that operation. A custom implementation may choose any scheduling policy. + +`ITaskFlow` adds ownership to that contract through `IAsyncDisposable`, `IDisposable`, `ITaskFlowInfo`, and `Dispose(TimeSpan)`. Built-in flows own their execution lane; scheduler decorators such as `WithTimeout` and `OnError` do not. + +Keep both references when composing policies: + +```csharp +await using ITaskFlow flow = new TaskFlow(); +ITaskScheduler operations = flow + .WithTimeout(TimeSpan.FromSeconds(10)) + .OnError(exception => Console.Error.WriteLine(exception)); + +await operations.Enqueue(token => Task.Delay(25, token)); +``` + +Dispose `flow`, not `operations`. + +## FIFO and operation completion + +`TaskFlow`, `DedicatedThreadTaskFlow`, and `CurrentThreadTaskFlow` execute accepted operations one at a time in FIFO order. An operation begins after its predecessor finishes, including failure or cancellation. The task returned by `Enqueue` completes with that operation's own outcome. + +FIFO describes invocation order within one flow. It does not impose ordering across multiple flows, and it is not a universal guarantee of every `ITaskScheduler` implementation. + +## Cancellation sources + +A built-in flow links the caller's token with its disposal token. Decorators can add more sources: + +- `CreateCancellationScope` adds a component or request lifetime token. +- `CreateCancelPrevious` requests cancellation of older unfinished submissions. +- `WithTimeout` adds a timeout cancellation signal and reports a `TimeoutException` if the timer wins. + +Cancellation is a request. Delegates must observe the supplied token. Built-in flows still invoke a queued delegate after cancellation so the lane can advance deterministically. + +## Failure continuity + +An operation failure is exposed through its returned task. Built-in flows wait for that task, suppress its failure only in the internal predecessor chain, and then invoke the next delegate. `OnError` can observe matching failures, but it rethrows them to the returned operation task. + +## Disposal sequence + +`DisposeAsync`: + +1. prevents new submissions; +2. finishes pending initialization where applicable; +3. cancels the flow lifetime token; +4. waits for queued and active operations to finish; and +5. releases owned resources. + +`Dispose()` performs the same shutdown attempt but waits no longer than `TaskFlowOptions.SynchronousDisposeTimeout`. `Dispose(TimeSpan)` reports whether completion occurred within its explicit timeout. + +After disposal begins, new submissions throw `ObjectDisposedException`. Retain operation tasks if callers need their individual shutdown or failure outcomes; disposal intentionally does not surface every operation exception. + +## Synchronization context + +Thread-backed flows install a synchronization context that marshals captured continuations back to their execution thread. The standard `TaskFlow` schedules each operation through its configured `TaskScheduler`; code inside a delegate follows normal `await` and synchronization-context rules. + +See [Execution models](execution-models.md) for the differences among the built-in flows and `TaskFlowSchedulerAdapter`. diff --git a/docs/customization.md b/docs/customization.md new file mode 100644 index 0000000..aba22dd --- /dev/null +++ b/docs/customization.md @@ -0,0 +1,73 @@ +--- +layout: page +title: Customization +permalink: /customization/ +--- + +# Customization + +Use customization when a built-in FIFO flow, decorator chain, or .NET `TaskScheduler` adapter cannot express the required policy. Custom implementations are responsible for defining their own ordering, cancellation, and ownership guarantees. + +## Implement ITaskScheduler + +`ITaskScheduler` is the smallest extension point. This example adapts an immediate execution policy: + +```csharp +public sealed class InlineScheduler : ITaskScheduler +{ + public async Task Enqueue( + Func> taskFunc, + object? state, + CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(taskFunc); + return await taskFunc(state, cancellationToken); + } +} +``` + +This scheduler is not FIFO, owned, or disposable. Document such differences because the built-in-flow guarantees do not automatically apply to `ITaskScheduler` implementations. + +At minimum, a production scheduler should define: + +- whether submissions are FIFO, concurrent, prioritized, rejected, or coalesced; +- when and where delegates are invoked; +- what happens when cancellation precedes invocation; +- how delegate results, failures, and cancellation complete returned tasks; and +- whether it owns resources and how shutdown works. + +## Adapt a .NET TaskScheduler + +When an existing `TaskScheduler` already defines the execution environment, use `TaskFlowSchedulerAdapter`: + +```csharp +TaskScheduler dotnetScheduler = TaskScheduler.Default; +ITaskScheduler scheduler = new TaskFlowSchedulerAdapter(dotnetScheduler); + +await scheduler.Enqueue(token => Task.Delay(25, token)); +``` + +The adapter does not add FIFO serialization or lifetime ownership. + +## Derive from TaskFlowBase + +Derive from `TaskFlowBase` only when the implementation needs an owned lifecycle. A derived flow must: + +- make `Enqueue` thread-safe, validate delegates, call `CheckDisposed`, and return an operation task; +- link caller cancellation with `CompletionToken` where flow disposal should request cancellation; +- use `Starting()` and `Ready()` to report initialization state when startup is asynchronous; +- implement `GetInitializationTask()` and `GetCompletionTask()` accurately; +- preserve completion progress after individual operation failures; and +- release owned resources through the disposal hooks without losing observable operation outcomes. + +Use `ThisLock` for state coordinated with the base disposal state. Do not claim built-in FIFO or canceled-delegate behavior unless the custom implementation actually preserves it and tests it. + +## Interceptors and decorators + +Prefer an `ITaskScheduler` decorator when adding a cross-cutting policy without owning the lane. Forward the original state and cancellation token unless the policy intentionally transforms them. Preserve the returned task's result and failure unless replacement behavior is part of the documented contract. + +Use `ITaskSchedulerInterceptor` or `IAsyncTaskSchedulerInterceptor` for operation lifecycle callbacks. See [Observability extensions](extensions/observability.md) for callback ordering and exception replacement rules. + +## Testing custom implementations + +Exercise concurrent submissions, ordering, canceled-before-start work, disposal during startup and execution, delegate failures, callback failures, and operations that ignore cancellation. Run contract tests against every execution model whose guarantees the custom implementation claims. diff --git a/docs/dependency-injection.md b/docs/dependency-injection.md new file mode 100644 index 0000000..bf64346 --- /dev/null +++ b/docs/dependency-injection.md @@ -0,0 +1,156 @@ +--- +layout: page +title: Dependency injection +permalink: /dependency-injection/ +--- + +# Dependency injection + +Install the integration package: + +```shell +dotnet add package TaskFlow.Microsoft.Extensions.DependencyInjection +``` + +The package targets `netstandard2.0`, .NET 8, and .NET 10 and uses the `System.Threading.Tasks.Flow` namespace. + +## Scoped execution lane + +```csharp +using Microsoft.Extensions.DependencyInjection; +using System.Threading.Tasks.Flow; + +var services = new ServiceCollection(); +services.AddTaskFlow(); +services.AddScoped(); + +await using ServiceProvider provider = services.BuildServiceProvider(); +await using AsyncServiceScope scope = provider.CreateAsyncScope(); + +var writer = scope.ServiceProvider.GetRequiredService(); +await writer.SaveAsync(new Report()); + +public sealed class ReportWriter +{ + private readonly ITaskScheduler _scheduler; + + public ReportWriter(ITaskScheduler scheduler) + { + _scheduler = scheduler; + } + + public Task SaveAsync( + Report report, + CancellationToken cancellationToken = default) + { + return _scheduler.Enqueue( + token => PersistAsync(report, token), + cancellationToken); + } + + private static Task PersistAsync( + Report report, + CancellationToken token) => Task.CompletedTask; +} + +public sealed record Report; +``` + +`AddTaskFlow()` registers one scoped scheduler and flow information object. The dependency-injection scope owns the underlying flow and disposes it when the scope ends. Consumers should inject `ITaskScheduler` and must not dispose it. + +## Options + +Provide options when a scope needs a different .NET scheduler or synchronous-disposal timeout: + +```csharp +services.AddTaskFlow(new TaskFlowOptions +{ + TaskScheduler = TaskScheduler.Default, + SynchronousDisposeTimeout = TimeSpan.FromSeconds(15), +}); +``` + +Prefer asynchronous scope disposal. A finite synchronous timeout can return before noncooperative work finishes. + +## Named flows + +Named registrations hold independent configurations: + +```csharp +services.AddTaskFlow(); +services.AddTaskFlow( + "imports", + new TaskFlowOptions + { + SynchronousDisposeTimeout = TimeSpan.FromSeconds(30), + }); +``` + +A named flow is still sequential. Names select configurations; they do not create concurrency within a flow. + +Use `ITaskFlowFactory.CreateTaskFlow(name)` when the caller needs an explicitly owned named flow: + +```csharp +await using ITaskFlow imports = factory.CreateTaskFlow("imports"); +await imports.Enqueue(token => ImportAsync(token)); + +static Task ImportAsync(CancellationToken token) => Task.CompletedTask; +``` + +The returned `ITaskFlow` belongs to the caller and must be disposed. + +## Compose decorators at registration + +The advanced `AddTaskFlow` overload accepts `configureSchedulerChain`, which moves scheduler policy into the application's composition root. Consumers receive the configured `ITaskScheduler`; they do not need to construct, order, or retain references to its decorators. + +This is especially useful when a named TaskFlow configuration is exposed as a keyed DI service: + +```csharp +using Microsoft.Extensions.DependencyInjection; +using System.Threading.Tasks.Flow; + +services.AddTaskFlow(); + +services.AddTaskFlow( + name: "imports", + configureSchedulerChain: (scheduler, _) => scheduler + .WithOperationName("imports") + .WithTimeout(TimeSpan.FromSeconds(30))); + +services.AddKeyedScoped( + "imports", + (provider, _) => provider + .GetRequiredService() + .CreateTaskFlow("imports")); + +public sealed class ImportWorker( + [FromKeyedServices("imports")] ITaskScheduler scheduler) +{ + public Task RunAsync(CancellationToken cancellationToken) => + scheduler.Enqueue(ImportAsync, cancellationToken); + + private static Task ImportAsync(CancellationToken token) => + Task.CompletedTask; +} +``` + +The consumer knows only the service key and `ITaskScheduler`. The registration owns the operation name and timeout policies, so they can change without changing `ImportWorker`. + +`AddTaskFlow("imports", ...)` defines the named TaskFlow configuration; it does not by itself register a keyed `ITaskScheduler`. The explicit scoped bridge above asks `ITaskFlowFactory` to create the named flow. Because the DI container creates that scoped service, it also disposes the returned `ITaskFlow` ownership wrapper at the end of the scope. + +The same pipeline mechanism works for the ordinary unkeyed scoped scheduler by passing `name: null`. A chain can resolve services from the provided `IServiceProvider` and can compose cancellation, timeout, logging, interception, or application-specific wrappers. + +Decorator order is observable: each extension wraps the scheduler returned by the previous call, so the last extension is the outermost decorator. See [Semantics and pitfalls](semantics-and-pitfalls.md#decorator-order-changes-what-a-policy-sees). + +The factory keeps the root `ITaskFlow` separate from the decorated scheduler. Disposing a factory-created flow or a container-owned scoped registration therefore disposes the real owner rather than relying on its decorators to own it. + +## Advanced creation + +The same `AddTaskFlow` overload can also customize creation of the underlying `ITaskFlow` and resolve `TaskFlowOptions` from the service provider. Use those delegates when a named configuration needs a different execution model or options determined from other registered services. + +## Registered lifetimes + +- `ITaskScheduler` and `ITaskFlowInfo` are scoped. +- `ITaskFlowFactory` and the default factory are singletons. +- `ITaskFlow` is not registered directly for consumers. +- Factory-created flows are caller-owned; injected scoped schedulers are container-owned. diff --git a/docs/execution-models.md b/docs/execution-models.md new file mode 100644 index 0000000..272703f --- /dev/null +++ b/docs/execution-models.md @@ -0,0 +1,106 @@ +--- +layout: page +title: Execution models +permalink: /execution-models/ +--- + +# Execution models + +TaskFlow separates the scheduling contract from the lifetime-owning execution lane. Choose the smallest model that supplies the ordering and thread behavior your component actually needs. + +## Standard TaskFlow + +`TaskFlow` is the default choice. It serializes operations in FIFO order and schedules them through `TaskFlowOptions.TaskScheduler`, which defaults to `TaskScheduler.Default`. + +```csharp +await using var flow = new TaskFlow(); + +Task first = flow.Enqueue(token => ProcessAsync("first", token)); +Task second = flow.Enqueue(token => ProcessAsync("second", token)); + +await Task.WhenAll(first, second); + +static Task ProcessAsync(string value, CancellationToken token) => + Task.CompletedTask; +``` + +A custom `TaskScheduler` changes where TaskFlow schedules its chain; it does not remove the flow's one-operation-at-a-time guarantee. + +## DedicatedThreadTaskFlow + +Use `DedicatedThreadTaskFlow` when every operation and captured asynchronous continuation must run on one library-owned background thread. + +```csharp +await using var flow = new DedicatedThreadTaskFlow("device-io"); + +await flow.Enqueue(async token => +{ + int beforeAwait = Environment.CurrentManagedThreadId; + await Task.Delay(25, token); + int afterAwait = Environment.CurrentManagedThreadId; + + Console.WriteLine($"{beforeAwait} -> {afterAwait}"); +}); +``` + +The flow creates and owns the background thread. Its synchronization context returns captured continuations to that thread. Avoid blocking the thread on asynchronous operations that need the same context. + +## CurrentThreadTaskFlow + +`CurrentThreadTaskFlow` uses a thread supplied by the application. Calling `Run()` starts the processing loop and blocks that thread until disposal completes. + +```csharp +await using var flow = new CurrentThreadTaskFlow(); +var thread = new Thread(flow.Run) +{ + IsBackground = true, + Name = "externally-owned-lane", +}; + +thread.Start(); +await flow.Enqueue(token => Task.Delay(25, token)); +``` + +Use this model only when the application owns the thread and can dedicate it to the run loop. For a UI framework, its native dispatcher or synchronization-context scheduler is often the better integration point. + +## TaskFlowSchedulerAdapter + +`TaskFlowSchedulerAdapter` exposes an existing .NET `TaskScheduler` as an `ITaskScheduler`: + +```csharp +TaskScheduler dotnetScheduler = TaskScheduler.Default; +ITaskScheduler scheduler = new TaskFlowSchedulerAdapter(dotnetScheduler); + +await scheduler.Enqueue(token => Task.Delay(25, token)); +``` + +The adapter preserves the supplied scheduler's execution characteristics. It does not create an owned FIFO flow and is not disposable. Do not assume built-in-flow cancellation or ordering semantics when using it. + +## Multiple lanes and named flows + +Every built-in flow is independently sequential. Two flows can execute concurrently because neither waits for the other: + +```csharp +await using var imports = new TaskFlow(); +await using var exports = new TaskFlow(); + +await Task.WhenAll( + imports.Enqueue(token => ImportAsync(token)), + exports.Enqueue(token => ExportAsync(token))); + +static Task ImportAsync(CancellationToken token) => Task.CompletedTask; +static Task ExportAsync(CancellationToken token) => Task.CompletedTask; +``` + +Named dependency-injection registration selects independently configured flows; a name does not increase concurrency inside a flow. Use [Dependency injection](dependency-injection.md) for ownership and registration details. + +## Selection guide + +| Requirement | Choose | +|---|---| +| FIFO asynchronous work on the thread pool | `TaskFlow` | +| FIFO work and captured continuations on one owned thread | `DedicatedThreadTaskFlow` | +| FIFO work on an externally supplied dedicated thread | `CurrentThreadTaskFlow` | +| Adapt an existing .NET scheduling policy without ownership | `TaskFlowSchedulerAdapter` | +| Independent concurrent lanes | Multiple flow instances | +| A specialized queueing or concurrency policy | A custom `ITaskScheduler` or `TaskFlowBase` implementation | diff --git a/docs/extensions/cancellation.md b/docs/extensions/cancellation.md new file mode 100644 index 0000000..24afa23 --- /dev/null +++ b/docs/extensions/cancellation.md @@ -0,0 +1,76 @@ +--- +layout: page +title: Cancellation extensions +permalink: /extensions/cancellation/ +--- + +# Cancellation extensions + +Cancellation policies compose additional cancellation sources with the caller token and the underlying flow's disposal token. They request cancellation; they cannot force a delegate to stop. + +## Component cancellation scopes + +`CreateCancellationScope` links one shared token into every submission through the returned scheduler. + +```csharp +using var lifetimeSource = new CancellationTokenSource(); +await using var flow = new TaskFlow(); +ITaskScheduler componentOperations = + flow.CreateCancellationScope(lifetimeSource.Token); + +using var callerSource = new CancellationTokenSource(); +Task operation = componentOperations.Enqueue( + token => Task.Delay(TimeSpan.FromSeconds(30), token), + callerSource.Token); + +lifetimeSource.Cancel(); + +try +{ + await operation; +} +catch (OperationCanceledException) +{ + // The component lifetime requested cancellation. +} +``` + +Either the caller token or scope token can cancel the linked token. Disposing the underlying built-in flow adds its own cancellation request. + +## Cancel previous + +`CreateCancelPrevious` requests cancellation of every older unfinished submission when a new operation is enqueued. This includes queued operations and a currently executing operation. + +```csharp +await using var flow = new TaskFlow(); +ITaskScheduler latest = flow.CreateCancelPrevious(); + +Task first = latest.Enqueue(token => + Task.Delay(TimeSpan.FromSeconds(30), token)); + +Task second = latest.Enqueue(token => + Task.Delay(TimeSpan.FromMilliseconds(25), token)); + +try +{ + await first; +} +catch (OperationCanceledException) +{ + // The second submission canceled the first. +} + +await second; +``` + +On a built-in FIFO flow, an older queued delegate still runs when it reaches the front and receives the canceled token. A delegate that ignores cancellation can delay newer work. + +## Latest-request-wins pattern + +Place a cooperative delay at the start of each cancel-previous delegate. Rapid submissions cancel older delays; after a quiet period, the newest delegate proceeds. + +This is a useful latest-request-wins recipe, but it is not a dedicated trailing-edge debounce implementation. In particular, a newer submission also requests cancellation of already-started work. See [Recipes](../recipes.md#latest-request-wins) and [issue #21](https://github.com/dombrovsky/TaskFlow/issues/21). + +## Disposal and ownership + +Neither cancellation decorator is disposable. Dispose the flow that owns the execution lane. Await individual operation tasks when their outcomes belong to a caller; intentional fire-and-forget submissions can rely on flow disposal for their lifetime but need a separate error-reporting policy when failures matter. diff --git a/docs/extensions/index.md b/docs/extensions/index.md new file mode 100644 index 0000000..6b6d89b --- /dev/null +++ b/docs/extensions/index.md @@ -0,0 +1,50 @@ +--- +layout: page +title: Extensions +permalink: /extensions/ +--- + +# Extensions + +Extensions return `ITaskScheduler` decorators. They do not own or dispose the scheduler or flow they wrap. Keep the underlying `ITaskFlow` reference and dispose it at the owner boundary. + +| Extension | Package and frameworks | Effect | Details | +|---|---|---|---| +| `CreateCancelPrevious` | `TaskFlow`; all targets | Cancels older unfinished submissions | [Cancellation](cancellation.md) | +| `CreateCancellationScope` | `TaskFlow`; all targets | Links a shared lifetime token | [Cancellation](cancellation.md) | +| `WithTimeout` | `TaskFlow`; all targets | Adds a queue-and-execution timeout | [Reliability](reliability.md) | +| `WithThrottle` | `TaskFlow` on .NET 8/10; `TaskFlow.Extensions.Time` on `netstandard2.0` | Rejects submissions inside an admission interval | [Reliability](reliability.md) | +| `OnError` | `TaskFlow`; all targets | Observes matching failures and rethrows | [Reliability](reliability.md) | +| `WithOperationName` | `TaskFlow`; all targets | Adds an operation-name annotation | [Observability](observability.md) | +| `Intercept` | `TaskFlow`; all targets | Runs synchronous or asynchronous lifecycle callbacks | [Observability](observability.md) | +| `WithLogging` | `TaskFlow.Microsoft.Extensions.Logging`; all targets | Emits structured lifecycle events | [Observability](observability.md) | + +## Composition + +Decorator order is observable. The last extension call produces the outermost scheduler and sees a submission first. + +```csharp +await using var flow = new TaskFlow(); + +ITaskScheduler operations = flow + .WithLogging(logger) + .WithTimeout(TimeSpan.FromSeconds(10)) + .CreateCancellationScope(lifetimeToken) + .WithOperationName("orders.persist"); + +await operations.Enqueue(token => PersistAsync(token)); +``` + +Here the operation-name annotation travels inward to timeout and logging. The caller token, lifetime token, timeout signal, and flow-disposal token can all request cancellation. The delegate must cooperate with the token it receives. + +```csharp +static Task PersistAsync(CancellationToken token) => Task.CompletedTask; +``` + +## Common rules + +- Await or return an operation task when its outcome matters to the caller. For intentional fire-and-forget work, discard it explicitly and add failure reporting inside the operation or through a decorator when needed. +- Dispose the underlying flow, not its decorators. +- Treat cancellation as a request rather than proof that work stopped. +- Use named local functions for value-returning asynchronous delegates if overload resolution is ambiguous. +- Test policy ordering because moving one decorator can change which annotations or failures another decorator observes. diff --git a/docs/extensions/observability.md b/docs/extensions/observability.md new file mode 100644 index 0000000..967d5fa --- /dev/null +++ b/docs/extensions/observability.md @@ -0,0 +1,69 @@ +--- +layout: page +title: Observability extensions +permalink: /extensions/observability/ +--- + +# Observability extensions + +## Operation names + +`WithOperationName` attaches an `OperationNameAnnotation` to submissions. Place it outside consumers such as logging or timeout so they can read the annotation. + +```csharp +ITaskScheduler named = flow + .WithLogging(logger) + .WithOperationName("orders.persist"); + +await named.Enqueue(token => PersistAsync(token)); + +static Task PersistAsync(CancellationToken token) => Task.CompletedTask; +``` + +Decorator order is observable: `flow.WithOperationName(...).WithLogging(logger)` places logging outside the annotation and therefore does not provide that name to the logging wrapper. + +## Microsoft logging + +Install the integration package: + +```shell +dotnet add package TaskFlow.Microsoft.Extensions.Logging +``` + +```csharp +ITaskScheduler logged = flow + .WithLogging(logger, options => + { + options.EnqueuedLogLevel = LogLevel.Debug; + options.StartedLogLevel = LogLevel.Information; + options.SucceededLogLevel = LogLevel.Information; + options.FailedLogLevel = LogLevel.Error; + options.FinishedLogLevel = LogLevel.Debug; + }) + .WithOperationName("imports.run"); + +await logged.Enqueue(token => ImportAsync(token)); + +static Task ImportAsync(CancellationToken token) => Task.CompletedTask; +``` + +The decorator can emit enqueued, started, cancellation-requested, succeeded or failed, and finished events. Every level defaults to `Trace`; set a level to `LogLevel.None` to disable that event. Events include an increasing operation ID, optional name, duration where applicable, and the failure exception. + +The cancellation event reports a request, not the final outcome. A delegate can ignore cancellation and complete successfully. Logging never suppresses the operation exception. + +## Interception + +`Intercept` supports custom operation lifecycle behavior. A synchronous interceptor is a struct copied for each operation and implements callbacks in this order: + +1. `OnBefore`; +2. the operation; +3. either `OnSuccess` or `OnError`; and +4. `OnFinally`. + +An asynchronous interceptor uses `IAsyncTaskSchedulerInterceptor` to create one `IAsyncTaskInterceptor` per operation. Every returned `ValueTask` is awaited before the lifecycle advances. + +Callbacks run inside the selected scheduler context. A callback failure faults the returned operation task; error or finalization callback failures can replace the original operation failure. Use interception when this replacement behavior and lifecycle control are intentional. Prefer `OnError` or `WithLogging` for simpler observation. + +## Ownership + +Annotations, interception, and logging are scheduler decorators. They neither own nor dispose the underlying flow. Keep the original `ITaskFlow` and dispose it at the component boundary. diff --git a/docs/extensions/reliability.md b/docs/extensions/reliability.md new file mode 100644 index 0000000..edf75ea --- /dev/null +++ b/docs/extensions/reliability.md @@ -0,0 +1,94 @@ +--- +layout: page +title: Reliability extensions +permalink: /extensions/reliability/ +--- + +# Reliability extensions + +## Timeout + +`WithTimeout` applies one budget to queue waiting and delegate execution. It throws `TimeoutException` when the timer wins and requests cancellation of the underlying operation. + +```csharp +await using var flow = new TaskFlow(); +ITaskScheduler bounded = flow + .WithTimeout(TimeSpan.FromSeconds(2)) + .WithOperationName("catalog.refresh"); + +try +{ + await bounded.Enqueue(token => RefreshAsync(token)); +} +catch (TimeoutException exception) +{ + Console.Error.WriteLine(exception.Message); +} + +static Task RefreshAsync(CancellationToken token) => Task.CompletedTask; +``` + +The clock starts when `Enqueue` reaches the timeout wrapper, not when the delegate starts. On a busy built-in flow, the returned task can time out while waiting. The queued delegate is then invoked later with an already-canceled token. If it ignores cancellation, it can continue and hold the lane even though the caller already received `TimeoutException`. + +## Error observation + +`OnError` runs an action for matching exceptions and then rethrows so the returned task preserves failure. + +```csharp +await using var flow = new TaskFlow(); +ITaskScheduler observed = flow.OnError(exception => + Console.Error.WriteLine(exception.Message)); + +try +{ + await observed.Enqueue(token => FailAsync(token)); +} +catch (IOException) +{ + // The caller still receives the operation failure. +} + +static Task FailAsync(CancellationToken token) => + Task.FromException(new IOException("Storage unavailable")); +``` + +An optional filter can restrict which matching exceptions trigger the action. If the action itself throws, its exception replaces the operation exception. `OnError` does not suppress failures, retry work, or keep a failed background loop alive. + +## Leading-edge throttle + +`WithThrottle` admits the first submission immediately and rejects later submissions until the interval has elapsed. + +```csharp +await using var flow = new TaskFlow(); +ITaskScheduler throttled = + flow.WithThrottle(TimeSpan.FromSeconds(1)); + +await throttled.Enqueue(token => SendAsync(token)); + +try +{ + await throttled.Enqueue(token => SendAsync(token)); +} +catch (OperationThrottledException) +{ + // Rejected before reaching the underlying flow. +} + +static Task SendAsync(CancellationToken token) => Task.CompletedTask; +``` + +Admission is checked when the wrapper receives `Enqueue`. Accepted operations consume the interval even if they later fail or observe cancellation. An operation at the exact interval boundary is admitted. Rejected work is not delayed, queued, or replaced. + +`OperationThrottledException` derives from `OperationCanceledException`, so catch it first when the application needs to distinguish rejection from other cancellation. + +On .NET 8 and .NET 10, `WithThrottle` is in the core `TaskFlow` package. When a consumer resolves TaskFlow's `netstandard2.0` asset, install `TaskFlow.Extensions.Time`, which supplies the same API using `Microsoft.Bcl.TimeProvider`. + +## Choosing a policy + +| Need | Use | +|---|---| +| Bound queue wait plus execution time | `WithTimeout` | +| Report failures without changing successful behavior | `OnError` | +| Admit at most one submission per interval | `WithThrottle` | +| Cancel older unfinished operations | `CreateCancelPrevious` | +| Retry or suppress failures | Implement inside the delegate or use an external policy; TaskFlow has no built-in retry policy | diff --git a/docs/getting-started.md b/docs/getting-started.md new file mode 100644 index 0000000..9e17c5b --- /dev/null +++ b/docs/getting-started.md @@ -0,0 +1,85 @@ +--- +layout: page +title: Getting started +permalink: /getting-started/ +--- + +# Getting started + +Install the core package: + +```shell +dotnet add package TaskFlow +``` + +All public types use the `System.Threading.Tasks.Flow` namespace. + +## Create a FIFO execution lane + +```csharp +using System.Threading.Tasks.Flow; + +await using var flow = new TaskFlow(); + +Task first = flow.Enqueue(async token => +{ + await Task.Delay(25, token); + Console.WriteLine("first"); +}); + +Task second = flow.Enqueue(token => +{ + Console.WriteLine("second"); + return Task.CompletedTask; +}); + +await Task.WhenAll(first, second); +``` + +Calls to `Enqueue` are thread-safe. The standard `TaskFlow` invokes one operation at a time in submission order. Each call returns a task representing that operation's result, exception, or cancellation. + +## Choose whether to await an operation + +TaskFlow keeps the lane moving after an operation fails. Await or return the task when the operation's result, cancellation, or exception belongs to a caller: + +```csharp +Task write = flow.Enqueue(token => WriteAsync(token)); + +try +{ + await write; +} +catch (IOException exception) +{ + Console.Error.WriteLine(exception.Message); +} + +static Task WriteAsync(CancellationToken token) => Task.CompletedTask; +``` + +For component-owned fire-and-forget work, discard the task explicitly with `_ = flow.Enqueue(...)`. Disposing the flow still requests cancellation and waits for accepted work, but it does not surface an ignored operation's exception. Handle failures inside the operation or add an error-observation decorator when diagnostics are required. + +## Prefer asynchronous disposal + +Use `await using` when the owner has an asynchronous lifetime. `DisposeAsync` stops accepting work, requests cancellation through the operation tokens, and waits for the lane to finish. Cancellation remains cooperative: an operation that ignores its token can delay disposal indefinitely. + +Synchronous `Dispose()` waits only for `TaskFlowOptions.SynchronousDisposeTimeout`. It can return while noncooperative work continues. See [Concepts and lifecycle](concepts-and-lifecycle.md) and [Semantics and pitfalls](semantics-and-pitfalls.md) before using a flow to own long-running work. + +## Pass caller cancellation + +```csharp +using var cancellationSource = new CancellationTokenSource(TimeSpan.FromSeconds(5)); + +await flow.Enqueue( + token => Task.Delay(TimeSpan.FromSeconds(1), token), + cancellationSource.Token); +``` + +For built-in flows, cancellation does not remove a queued delegate. When it reaches the front of the lane, the delegate is invoked with an already-canceled token and decides cooperatively how to finish. + +## Next steps + +- Start from a complete application pattern in [Recipes](recipes.md). +- Add cancellation and reliability policies through [Extensions](extensions/index.md). +- Select a different thread or scheduler in [Execution models](execution-models.md). +- Register scoped or named lanes with [Dependency injection](dependency-injection.md). diff --git a/docs/index.md b/docs/index.md new file mode 100644 index 0000000..72456d6 --- /dev/null +++ b/docs/index.md @@ -0,0 +1,66 @@ +--- +layout: home +title: TaskFlow for .NET +permalink: / +--- + +TaskFlow is an owned FIFO execution lane for asynchronous .NET work. Each submitted operation gets its own result task while the lane serializes access, preserves submission order, and provides a clear shutdown boundary. + +Use TaskFlow to: + +- serialize calls to mutable or non-thread-safe resources; +- turn synchronous callbacks into ordered asynchronous processing; +- bind background work to a component or dependency-injection scope; +- cancel obsolete work when a newer operation arrives; +- compose timeout, cancellation, logging, annotations, and error observation; and +- run work on the thread pool, a dedicated thread, or a caller-owned thread. + +## Start by goal + +| Goal | Read | +|---|---| +| Create and dispose a first FIFO lane | [Getting started](getting-started.md) | +| Understand operation completion and ownership | [Concepts and lifecycle](concepts-and-lifecycle.md) | +| Avoid cancellation, timeout, and disposal surprises | [Semantics and pitfalls](semantics-and-pitfalls.md) | +| Choose the thread pool, a dedicated thread, or another scheduler | [Execution models](execution-models.md) | +| Apply TaskFlow to common application problems | [Recipes](recipes.md) | +| Add cancellation, reliability, and diagnostics policies | [Extensions](extensions/index.md) | +| Register scoped or named flows | [Dependency injection](dependency-injection.md) | +| Implement adapters or custom flows | [Customization](customization.md) | +| Check framework and package availability | [Compatibility](compatibility.md) | +| Diagnose common integration problems | [Troubleshooting](troubleshooting.md) | + +## Install + +```shell +dotnet add package TaskFlow +``` + +```csharp +using System.Threading.Tasks.Flow; + +await using var flow = new TaskFlow(); + +Task first = flow.Enqueue(async token => +{ + await Task.Delay(25, token); + Console.WriteLine("first"); +}); + +Task second = flow.Enqueue(token => +{ + Console.WriteLine("second"); + return Task.CompletedTask; +}); + +await Task.WhenAll(first, second); +``` + +`second` starts only after `first` completes. A failure in one returned task does not stop later queued operations. + +## Source and packages + +- [GitHub repository](https://github.com/dombrovsky/TaskFlow) +- [TaskFlow on NuGet](https://www.nuget.org/packages/TaskFlow/) +- [License](https://github.com/dombrovsky/TaskFlow/blob/main/LICENSE) +- [Issues and feedback](https://github.com/dombrovsky/TaskFlow/issues) diff --git a/docs/recipes.md b/docs/recipes.md new file mode 100644 index 0000000..a8179ed --- /dev/null +++ b/docs/recipes.md @@ -0,0 +1,248 @@ +--- +layout: page +title: Recipes +permalink: /recipes/ +--- + +# Recipes + +These examples use both caller-observed operation tasks and deliberate fire-and-forget submissions whose lifetime is owned by a flow. + +## Serialize a non-thread-safe resource + +```csharp +public sealed class SerializedReadingStore : IAsyncDisposable +{ + private readonly IReadingStore _inner; + private readonly TaskFlow _flow = new(); + + public SerializedReadingStore(IReadingStore inner) + { + _inner = inner; + } + + public Task AppendAsync( + Reading reading, + CancellationToken cancellationToken = default) + { + return _flow.Enqueue( + token => _inner.AppendAsync(reading, token), + cancellationToken); + } + + public ValueTask DisposeAsync() => _flow.DisposeAsync(); +} +``` + +All callers remain asynchronous while access to `_inner` stays FIFO and non-concurrent. Unlike a manually managed `SemaphoreSlim`, the flow also owns shutdown and returns a task for each queued operation. + +## Preserve order from synchronous events + +```csharp +public sealed class ReadingSubscriber : IAsyncDisposable +{ + private readonly IReadingSource _source; + private readonly IReadingSink _sink; + private readonly TaskFlow _flow = new(); + private readonly ITaskScheduler _events; + + public ReadingSubscriber(IReadingSource source, IReadingSink sink) + { + _source = source; + _sink = sink; + _events = _flow.OnError( + exception => Console.Error.WriteLine(exception)); + source.ReadingReceived += OnReadingReceived; + } + + private void OnReadingReceived(object? sender, ReadingEventArgs args) + { + Reading reading = args.Reading; + + _ = _events.Enqueue( + token => _sink.HandleAsync(reading, token)); + } + + public async ValueTask DisposeAsync() + { + _source.ReadingReceived -= OnReadingReceived; + await _flow.DisposeAsync(); + } +} +``` + +The synchronous event handler only captures the event data and enqueues the asynchronous work. It returns immediately without becoming `async void`, and TaskFlow preserves callback submission order. The operation tasks are intentionally discarded; `OnError` reports failures, while disposing `_flow` controls the lifetime of every accepted callback. Unsubscribe before disposal so no callback can submit during shutdown. + +## Avoid duplicate credential refreshes + +```csharp +public sealed class CachedTokenProvider : IAsyncDisposable +{ + private static readonly TimeSpan RefreshMargin = TimeSpan.FromMinutes(1); + + private readonly ITokenClient _client; + private readonly TaskFlow _flow = new(); + private AccessToken? _cached; + + public CachedTokenProvider(ITokenClient client) + { + _client = client; + } + + public async Task GetAsync( + CancellationToken cancellationToken = default) + { + AccessToken? cached = _cached; + if (IsUsable(cached)) + { + return cached!; + } + + return await _flow.Enqueue(RefreshAsync, cancellationToken); + + async Task RefreshAsync(CancellationToken token) + { + if (IsUsable(_cached)) + { + return _cached!; + } + + _cached = await _client.RequestAsync(token); + return _cached; + } + } + + public ValueTask DisposeAsync() => _flow.DisposeAsync(); + + private static bool IsUsable(AccessToken? token) => + token is not null && + token.ExpiresAt - RefreshMargin > DateTimeOffset.UtcNow; +} +``` + +The second check is essential because another caller may refresh the value while this request waits in the lane. The explicitly typed local function also avoids ambiguous `Task` and `ValueTask` overload selection. + +## Latest request wins + +```csharp +public sealed class SearchController : IAsyncDisposable +{ + private readonly ISearchClient _client; + private readonly IResultView _view; + private readonly TaskFlow _flow = new(); + private readonly ITaskScheduler _latestSearch; + + public SearchController(ISearchClient client, IResultView view) + { + _client = client; + _view = view; + _latestSearch = _flow.CreateCancelPrevious(); + } + + public Task SearchAsync( + string query, + CancellationToken cancellationToken = default) + { + return _latestSearch.Enqueue(async token => + { + await Task.Delay(TimeSpan.FromMilliseconds(250), token); + SearchResults results = await _client.SearchAsync(query, token); + await _view.ShowAsync(results, token); + }, cancellationToken); + } + + public ValueTask DisposeAsync() => _flow.DisposeAsync(); +} +``` + +Each submission cancels unfinished older submissions. The initial delay means rapid updates usually leave only the latest delegate past the delay, but this is still cooperative cancellation rather than a native trailing-edge debounce policy. + +## Own a recoverable background loop + +```csharp +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using System.Threading.Tasks.Flow; + +public sealed class InboxPump : IHostedService +{ + private readonly IInbox _inbox; + private readonly ILogger _logger; + private readonly TaskFlow _lifetime = new(); + + public InboxPump(IInbox inbox, ILogger logger) + { + _inbox = inbox; + _logger = logger; + } + + public Task StartAsync(CancellationToken cancellationToken) + { + cancellationToken.ThrowIfCancellationRequested(); + _ = _lifetime.Enqueue(RunAsync); + return Task.CompletedTask; + } + + public Task StopAsync(CancellationToken cancellationToken) => + _lifetime.DisposeAsync().AsTask().WaitAsync(cancellationToken); + + private async Task RunAsync(CancellationToken cancellationToken) + { + while (!cancellationToken.IsCancellationRequested) + { + await using (var iteration = new TaskFlow()) + { + ITaskScheduler work = iteration.OnError( + exception => _logger.LogWarning( + exception, + "Inbox iteration failed"), + _ => !cancellationToken.IsCancellationRequested); + + _ = work.Enqueue( + _inbox.ProcessAvailableAsync, + cancellationToken); + } + + await Task.Delay(TimeSpan.FromSeconds(10), cancellationToken); + } + } +} +``` + +`StartAsync` deliberately discards the outer operation task and returns promptly to the host. The outer flow owns the loop and supplies its shutdown token; `StopAsync` disposes that flow and waits within the host's shutdown budget. + +Each pass creates an inner flow, submits one fire-and-forget operation, and disposes the inner flow before delaying. Inner disposal waits for that iteration but does not propagate its operation failure into the outer loop, so the next iteration still runs. `OnError` reports failures before rethrowing them into the intentionally ignored per-iteration task. The delay belongs to the outer operation, so stopping the host cancels both the current iteration and the wait before the next one. + +## Compose operational policies + +```csharp +public sealed class ObservableDeviceClient : IAsyncDisposable +{ + private readonly TaskFlow _flow = new(); + private readonly ITaskScheduler _operations; + + public ObservableDeviceClient( + ILogger logger, + CancellationToken lifetimeToken) + { + _operations = _flow + .WithLogging(logger) + .WithTimeout(TimeSpan.FromSeconds(15)) + .OnError(exception => + logger.LogError(exception, "Device synchronization failed")) + .CreateCancellationScope(lifetimeToken) + .WithOperationName("device.sync"); + } + + public Task SynchronizeAsync( + Func operation, + CancellationToken cancellationToken = default) + { + return _operations.Enqueue(operation, cancellationToken); + } + + public ValueTask DisposeAsync() => _flow.DisposeAsync(); +} +``` + +The outer operation-name decorator passes its annotation through the chain so logging and timeout diagnostics can use it. Only `_flow` owns the lane and is disposed. diff --git a/docs/semantics-and-pitfalls.md b/docs/semantics-and-pitfalls.md new file mode 100644 index 0000000..4b7adcb --- /dev/null +++ b/docs/semantics-and-pitfalls.md @@ -0,0 +1,92 @@ +--- +layout: page +title: Semantics and pitfalls +permalink: /semantics-and-pitfalls/ +--- + +# Semantics and pitfalls + +## Canceled queued delegates are still invoked + +The built-in `TaskFlow`, `DedicatedThreadTaskFlow`, and `CurrentThreadTaskFlow` implementations preserve queue progression by invoking every accepted delegate after its predecessor finishes. If cancellation happened while the operation waited, the delegate receives an already-canceled token. + +```csharp +await using var flow = new TaskFlow(); +using var cancellationSource = new CancellationTokenSource(); + +Task blocker = flow.Enqueue(async token => +{ + await Task.Delay(50, token); +}); + +Task canceled = flow.Enqueue(token => +{ + token.ThrowIfCancellationRequested(); + return Task.CompletedTask; +}, cancellationSource.Token); + +cancellationSource.Cancel(); +await blocker; + +try +{ + await canceled; +} +catch (OperationCanceledException) +{ + // The delegate ran and observed its already-canceled token. +} +``` + +This is a guarantee of the built-in flows, not of arbitrary `ITaskScheduler` implementations or adapters. + +## Timeout includes queue waiting + +`WithTimeout` starts its timer when `Enqueue` reaches the wrapper. Time spent behind earlier operations consumes the same budget as delegate execution. The returned task can time out before the delegate reaches the front of a busy built-in flow; that delegate is later invoked with timeout cancellation already requested. + +The timeout is cooperative. A delegate that ignores the token can continue running after the caller receives `TimeoutException`, and the sequential lane cannot move to later work until that delegate actually finishes. + +## Synchronous disposal can return early + +`DisposeAsync` waits for full completion. Synchronous disposal is bounded by `TaskFlowOptions.SynchronousDisposeTimeout`; the default is infinite, but a configured finite value can expire while work continues. Use `Dispose(TimeSpan)` when the caller needs the Boolean completion result. + +Prefer asynchronous disposal for component-owned background work and make long-running delegates observe cancellation promptly. + +## Disposal owns lifetime, not operation outcomes + +Disposal waits for lane completion and suppresses operation failures internally. This makes deliberate fire-and-forget possible: a component can discard selected operation tasks and still use flow disposal to request cancellation and wait for accepted work. + +Disposal does not propagate an ignored operation's exception. Await or return the task when its outcome belongs to a caller. For intentionally discarded work, handle failures inside the delegate or use a decorator such as `OnError` to report them. `OnError` observes and rethrows, so its diagnostic side effect still runs even when the returned task is deliberately ignored. + +## Decorator order changes what a policy sees + +Decorators wrap from left to right. In this chain, `WithOperationName` is outermost, so its annotation reaches the logging decorator: + +```csharp +ITaskScheduler operations = flow + .WithLogging(logger) + .WithTimeout(TimeSpan.FromSeconds(10)) + .WithOperationName("orders.persist"); +``` + +Moving `WithOperationName` inside `WithLogging` prevents that logging wrapper from seeing the annotation. Cancellation and error wrappers can likewise observe different failures depending on their order. + +## Value-returning async lambdas can be ambiguous + +TaskFlow provides both `Task` and `ValueTask` convenience overloads. A directly supplied value-returning `async` lambda can produce compiler error `CS0121`. Give the return type explicitly with a named local function: + +```csharp +Task result = flow.Enqueue(LoadAsync); + +async Task LoadAsync(CancellationToken token) +{ + await Task.Delay(25, token); + return 42; +} +``` + +## Throttle, latest-wins, and debounce differ + +- `WithThrottle` is leading-edge admission throttling. It accepts the first submission and rejects later submissions during the interval with `OperationThrottledException`. +- `CreateCancelPrevious` requests cancellation of older unfinished work whenever newer work is submitted. Adding an initial delay creates a latest-request-wins pattern when delegates cooperate. +- Trailing-edge debounce waits for a quiet interval and eventually executes the latest submission. TaskFlow does not currently expose that policy; it is tracked in [issue #21](https://github.com/dombrovsky/TaskFlow/issues/21). diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md new file mode 100644 index 0000000..67197d0 --- /dev/null +++ b/docs/troubleshooting.md @@ -0,0 +1,71 @@ +--- +layout: page +title: Troubleshooting +permalink: /troubleshooting/ +--- + +# Troubleshooting + +## `CS0121` for a value-returning async lambda + +TaskFlow has convenience overloads for both `Func>` and `Func>`. Make the intended return type explicit with a named local function: + +```csharp +Task result = flow.Enqueue(LoadAsync); + +async Task LoadAsync(CancellationToken token) +{ + await Task.Delay(25, token); + return 42; +} +``` + +## An operation timed out before its delegate started + +`WithTimeout` measures from the wrapper's `Enqueue` call, including time spent waiting behind earlier work. Increase the budget, shorten preceding work, use separate lanes for independent resources, or put an execution-only timeout inside the delegate if that is the intended policy. + +On built-in flows, the timed-out queued delegate is later invoked with an already-canceled token. + +## Cancellation was requested but work continued + +Cancellation is cooperative. Ensure every long-running operation accepts and observes the supplied token, including delays and downstream I/O. A delegate that ignores cancellation can hold a sequential lane after caller cancellation, timeout, or disposal. + +## Disposal returned while work was still running + +Synchronous disposal uses `TaskFlowOptions.SynchronousDisposeTimeout`. A finite timeout allows it to return before noncooperative work finishes. Prefer `await flow.DisposeAsync()` when shutdown must wait for completion, and use `Dispose(TimeSpan)` when the caller needs to inspect the Boolean result. + +## A queued delegate ran after its caller canceled + +That is expected for `TaskFlow`, `DedicatedThreadTaskFlow`, and `CurrentThreadTaskFlow`. Accepted delegates remain in the lane and receive a linked token that may already be canceled. Check the token before side effects when canceled work should do nothing. + +## A background loop stopped after one failure + +`OnError` observes and rethrows; it does not retry or suppress. Catch recoverable exceptions inside each loop iteration, or give each iteration an inner TaskFlow whose disposal waits for that iteration while an `OnError` decorator reports its failure. See the [hosted background-loop recipe](recipes.md#own-a-recoverable-background-loop). + +## Operation names are missing from logs + +Place `WithOperationName` outside `WithLogging`: + +```csharp +ITaskScheduler operations = flow + .WithLogging(logger) + .WithOperationName("orders.persist"); +``` + +The last decorator is outermost and passes the annotation inward to logging. Reversing these calls makes logging see the operation before the name is attached. + +## `WithThrottle` is unavailable on an older target + +Install `TaskFlow.Extensions.Time`. The core `netstandard2.0` asset omits `WithThrottle`; the time extension package supplies it using `Microsoft.Bcl.TimeProvider` to consumers that resolve that asset. + +## A named flow is still sequential + +Names select independent dependency-injection configurations. They do not alter the one-operation-at-a-time behavior of a built-in flow. Use multiple named flows for independent lanes or implement a scheduler whose contract explicitly supports concurrency. + +## A decorator cannot be disposed + +Decorators return `ITaskScheduler` because they do not own the lane. Keep and dispose the original `ITaskFlow`, or allow the dependency-injection scope to dispose its internally owned flow. + +## Submissions throw `ObjectDisposedException` + +The owner has started shutting down the flow. Stop event sources and reject new component work before disposing the lane. In event-driven components, unsubscribe handlers before `DisposeAsync` so callbacks cannot race shutdown.