From b9f05dca088e23f88a82833595d3613f722291b8 Mon Sep 17 00:00:00 2001 From: Volodymyr Dombrovskyi <5788605+dombrovsky@users.noreply.github.com> Date: Sun, 16 Aug 2026 11:53:20 -0600 Subject: [PATCH 1/8] Rebuild TaskFlow documentation --- README.md | 174 +++++++++--- .../README.md | 64 ++--- ...sions.Microsoft.DependencyInjection.csproj | 2 + .../README.md | 43 +-- ...skFlow.Extensions.Microsoft.Logging.csproj | 2 + TaskFlow.Extensions.Time/README.md | 40 +-- .../TaskFlow.Extensions.Time.csproj | 2 + TaskFlow/README.md | 49 ++-- TaskFlow/TaskFlow.csproj | 2 + _config.yml | 8 - docs/_config.yml | 17 ++ docs/compatibility.md | 48 ++++ docs/concepts-and-lifecycle.md | 66 +++++ docs/customization.md | 73 +++++ docs/dependency-injection.md | 117 ++++++++ docs/execution-models.md | 106 +++++++ docs/extensions/cancellation.md | 76 +++++ docs/extensions/index.md | 50 ++++ docs/extensions/observability.md | 69 +++++ docs/extensions/reliability.md | 94 +++++++ docs/getting-started.md | 83 ++++++ docs/index.md | 66 +++++ docs/recipes.md | 260 ++++++++++++++++++ docs/semantics-and-pitfalls.md | 90 ++++++ docs/troubleshooting.md | 71 +++++ 25 files changed, 1493 insertions(+), 179 deletions(-) delete mode 100644 _config.yml create mode 100644 docs/_config.yml create mode 100644 docs/compatibility.md create mode 100644 docs/concepts-and-lifecycle.md create mode 100644 docs/customization.md create mode 100644 docs/dependency-injection.md create mode 100644 docs/execution-models.md create mode 100644 docs/extensions/cancellation.md create mode 100644 docs/extensions/index.md create mode 100644 docs/extensions/observability.md create mode 100644 docs/extensions/reliability.md create mode 100644 docs/getting-started.md create mode 100644 docs/index.md create mode 100644 docs/recipes.md create mode 100644 docs/semantics-and-pitfalls.md create mode 100644 docs/troubleshooting.md diff --git a/README.md b/README.md index dae7e6a..b7afa3c 100644 --- a/README.md +++ b/README.md @@ -1,68 +1,158 @@ -# 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 provides owned FIFO execution lanes for asynchronous .NET work, with composable cancellation, timeout, diagnostics, and thread-affinity policies. [![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) ---- +Use TaskFlow to: -## Key Features +- serialize asynchronous access to a mutable or non-thread-safe resource; +- preserve event order when synchronous callbacks start asynchronous work; +- bind background work to a component or dependency-injection scope; +- cancel obsolete work when a newer request arrives; +- compose cancellation, timeout, logging, annotations, and error observation; and +- run work on the thread pool, a dedicated thread, or a caller-owned thread. -- **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. +## Install ---- +```shell +dotnet add package TaskFlow +``` -## When Should You Use TaskFlow? +Optional integrations: -TaskFlow is ideal for scenarios where you need: +```shell +dotnet add package TaskFlow.Microsoft.Extensions.DependencyInjection +dotnet add package TaskFlow.Microsoft.Extensions.Logging +dotnet add package TaskFlow.Extensions.Time +``` -- **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. +`TaskFlow.Extensions.Time` is needed only by consumers that resolve TaskFlow's `netstandard2.0` asset and use `WithThrottle`; .NET 8 and .NET 10 receive that extension from the core package. ---- +## A FIFO execution lane -## Getting Started +```csharp +using System.Threading.Tasks.Flow; -### Installation +await using var flow = new TaskFlow(); -Add the core package: -`dotnet add package TaskFlow` +Task first = flow.Enqueue(async token => +{ + await Task.Delay(25, token); + Console.WriteLine("first"); +}); -For dependency injection support: -`dotnet add package TaskFlow.Microsoft.Extensions.DependencyInjection` +Task second = flow.Enqueue(token => +{ + Console.WriteLine("second"); + return Task.CompletedTask; +}); -For Microsoft.Extensions.Logging integration: -`dotnet add package TaskFlow.Microsoft.Extensions.Logging` +await Task.WhenAll(first, second); +``` -### Building from Source +`second` starts only after `first` finishes. Each call returns a task for that operation's result, exception, or cancellation. One failed operation does not stop later queued operations. -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. +## Serialize a resource -### Basic Usage ```csharp -using var taskFlow = new TaskFlow(); +public sealed class SerializedStore : IAsyncDisposable +{ + private readonly IDataStore _inner; + private readonly TaskFlow _flow = new(); + + public SerializedStore(IDataStore inner) + { + _inner = inner; + } + + public Task SaveAsync( + Data data, + CancellationToken cancellationToken = default) + { + return _flow.Enqueue( + token => _inner.SaveAsync(data, token), + cancellationToken); + } + + public ValueTask DisposeAsync() => _flow.DisposeAsync(); +} +``` + +Callers remain asynchronous while access to the wrapped resource stays ordered and non-concurrent. -// Enqueue tasks for sequential execution -var task1 = taskFlow.Enqueue(() => Console.WriteLine("Task 1")); -var task2 = taskFlow.Enqueue(async () => await Task.Delay(100)); +## Latest request wins + +```csharp +await using var flow = new TaskFlow(); +ITaskScheduler latest = flow.CreateCancelPrevious(); + +Task search = latest.Enqueue(async token => +{ + await Task.Delay(TimeSpan.FromMilliseconds(250), token); + await SearchAsync(token); +}); + +await search; ``` ---- -## Extensions +Every new submission requests cancellation of older unfinished work. The delay creates a latest-request-wins pattern when delegates cooperate with cancellation. + +## Features + +| Capability | API or implementation | Documentation | +|---|---|---| +| FIFO asynchronous execution | `TaskFlow` | [Concepts and lifecycle](https://dombrovsky.github.io/TaskFlow/concepts-and-lifecycle/) | +| Thread affinity | `DedicatedThreadTaskFlow`, `CurrentThreadTaskFlow` | [Execution models](https://dombrovsky.github.io/TaskFlow/execution-models/) | +| Latest request wins | `CreateCancelPrevious` | [Cancellation](https://dombrovsky.github.io/TaskFlow/extensions/cancellation/) | +| Component cancellation | `CreateCancellationScope` | [Cancellation](https://dombrovsky.github.io/TaskFlow/extensions/cancellation/) | +| Queue-and-execution timeout | `WithTimeout` | [Reliability](https://dombrovsky.github.io/TaskFlow/extensions/reliability/) | +| Leading-edge admission throttle | `WithThrottle` | [Reliability](https://dombrovsky.github.io/TaskFlow/extensions/reliability/) | +| Error observation | `OnError` | [Reliability](https://dombrovsky.github.io/TaskFlow/extensions/reliability/) | +| Structured lifecycle logging | `WithLogging` | [Observability](https://dombrovsky.github.io/TaskFlow/extensions/observability/) | +| Scoped and named registration | DI integration package | [Dependency injection](https://dombrovsky.github.io/TaskFlow/dependency-injection/) | + +## When to use something else + +| Primitive | Prefer it when | +|---|---| +| `lock` | The entire critical section is synchronous. | +| `SemaphoreSlim` | Mutual exclusion is enough and callers will manage acquisition, release, ordering, and lifetime. | +| `Channel` | The application is fundamentally a producer/consumer data stream. | +| `BackgroundService` | Work belongs to the application host lifetime rather than a smaller component. | +| TaskFlow | Each submission needs FIFO execution, its own result task, composable policies, and an owned shutdown boundary. | + +## Packages and frameworks + +| Package | Frameworks | +|---|---| +| `TaskFlow` | `netstandard2.0`, `net8.0`, `net10.0` | +| `TaskFlow.Extensions.Time` | `netstandard2.0` | +| `TaskFlow.Microsoft.Extensions.DependencyInjection` | `netstandard2.0`, `net8.0`, `net10.0` | +| `TaskFlow.Microsoft.Extensions.Logging` | `netstandard2.0`, `net8.0`, `net10.0` | + +See the [compatibility matrix](https://dombrovsky.github.io/TaskFlow/compatibility/) for feature-level availability. + +## Lifecycle essentials + +- Prefer `await using`; `DisposeAsync` requests cancellation and waits for the lane to finish. +- Cancellation is cooperative. Synchronous disposal can time out while noncooperative work continues. +- Observe every task returned by `Enqueue`, even when the flow owns the work's lifetime. +- Built-in flows invoke accepted queued delegates with canceled tokens instead of removing them from the lane. +- Scheduler decorators do not own the underlying flow; dispose the original `ITaskFlow`. + +Read [Semantics and pitfalls](https://dombrovsky.github.io/TaskFlow/semantics-and-pitfalls/) before using timeouts or owning long-running background work. + +## Documentation + +- [Documentation home](https://dombrovsky.github.io/TaskFlow/) +- [Getting started](https://dombrovsky.github.io/TaskFlow/getting-started/) +- [Recipes](https://dombrovsky.github.io/TaskFlow/recipes/) +- [Extensions](https://dombrovsky.github.io/TaskFlow/extensions/) +- [Troubleshooting](https://dombrovsky.github.io/TaskFlow/troubleshooting/) -## License +Build the repository with a .NET 10 SDK. Tests run against .NET 8 and .NET 10. -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..9b8aa60 --- /dev/null +++ b/docs/dependency-injection.md @@ -0,0 +1,117 @@ +--- +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. + +## Advanced registration + +The advanced `AddTaskFlow` overload accepts delegates for: + +- creating the underlying `ITaskFlow`; +- resolving named `TaskFlowOptions`; and +- composing an `ITaskScheduler` decorator chain. + +Use it to centralize cancellation, timeout, logging, or application-specific wrappers. Preserve the root flow separately from the decorated scheduler so the dependency-injection scope disposes the actual owner. + +## 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..684792c --- /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. During shutdown, continue observing returned operation tasks so expected cancellation and unexpected failures are distinguished deliberately. diff --git a/docs/extensions/index.md b/docs/extensions/index.md new file mode 100644 index 0000000..72348aa --- /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 + +- Always observe the task returned by `Enqueue`. +- 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..0598099 --- /dev/null +++ b/docs/getting-started.md @@ -0,0 +1,83 @@ +--- +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. + +## Await operation tasks + +TaskFlow keeps the lane moving after an operation fails, but it does not consume the failure on behalf of the caller. Await returned tasks, return them to a caller, retain them as a component completion signal, or deliberately observe them with application-specific telemetry. + +```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; +``` + +## 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..4ea8a03 --- /dev/null +++ b/docs/recipes.md @@ -0,0 +1,260 @@ +--- +layout: page +title: Recipes +permalink: /recipes/ +--- + +# Recipes + +These examples focus on ownership and observable operation tasks rather than treating queued work as untracked fire-and-forget work. + +## 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 IReadingSink _sink; + private readonly TaskFlow _flow = new(); + + public ReadingSubscriber(IReadingSource source, IReadingSink sink) + { + _sink = sink; + source.ReadingReceived += OnReadingReceived; + } + + private async void OnReadingReceived(object? sender, ReadingEventArgs args) + { + Reading reading = args.Reading; + + try + { + await _flow.Enqueue( + token => _sink.HandleAsync(reading, token)); + } + catch (OperationCanceledException) + { + // Expected when the subscriber is disposed. + } + catch (Exception exception) + { + Console.Error.WriteLine(exception); + } + } + + public ValueTask DisposeAsync() => _flow.DisposeAsync(); +} +``` + +The event handler copies the event data and calls `Enqueue` before its first suspension, so TaskFlow preserves callback submission order. Event handlers are the conventional exception to avoiding `async void`; this one catches every operation outcome locally. In production, unsubscribe the event before disposing the flow 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 +public sealed class InboxPump : IAsyncDisposable +{ + private readonly IInbox _inbox; + private readonly ILogger _logger; + private readonly TaskFlow _flow = new(); + private Task? _completion; + + public InboxPump(IInbox inbox, ILogger logger) + { + _inbox = inbox; + _logger = logger; + } + + public Task Completion => _completion ?? Task.CompletedTask; + + public void Start() + { + _completion ??= _flow.Enqueue(RunAsync); + } + + public async ValueTask DisposeAsync() + { + await _flow.DisposeAsync(); + + if (_completion is null) + { + return; + } + + try + { + await _completion; + } + catch (OperationCanceledException) + { + // Disposal canceled the loop: normal shutdown. + } + } + + private async Task RunAsync(CancellationToken cancellationToken) + { + while (true) + { + cancellationToken.ThrowIfCancellationRequested(); + + try + { + await _inbox.ProcessAvailableAsync(cancellationToken); + } + catch (TransientInboxException exception) + when (!cancellationToken.IsCancellationRequested) + { + _logger.LogWarning(exception, "Inbox iteration failed"); + } + + await Task.Delay(TimeSpan.FromSeconds(10), cancellationToken); + } + } +} +``` + +Recoverable failures are caught per iteration so the loop survives them. `Completion` exposes unexpected terminal failures, while disposal cancellation is handled as normal shutdown. An outer `OnError` decorator can report a terminal failure but cannot restart a failed loop. + +## 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..6fa9f28 --- /dev/null +++ b/docs/semantics-and-pitfalls.md @@ -0,0 +1,90 @@ +--- +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 is not operation-task observation + +Disposal waits for lane completion and suppresses operation failures internally. It does not replace awaiting or otherwise observing the task returned for each operation. Store a background loop's completion task so unexpected terminal failures remain visible. + +## 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..4e534bd --- /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. Retain the loop's returned task so unexpected terminal failures remain observable. See the [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. From 447e93071a7c7a393d4a553f03da1a2c112c4bf3 Mon Sep 17 00:00:00 2001 From: Volodymyr Dombrovskyi <5788605+dombrovsky@users.noreply.github.com> Date: Sun, 16 Aug 2026 11:55:35 -0600 Subject: [PATCH 2/8] version 1.0.0-rc5 --- Directory.Build.props | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 From 676de2ee0bbb40ddf26efe1c73d6fedb306a87ca Mon Sep 17 00:00:00 2001 From: Volodymyr Dombrovskyi <5788605+dombrovsky@users.noreply.github.com> Date: Sun, 16 Aug 2026 12:03:12 -0600 Subject: [PATCH 3/8] Automate documentation validation and deployment --- .github/workflows/documentation.yml | 88 ++++++ eng/validate-documentation.ps1 | 404 ++++++++++++++++++++++++++++ 2 files changed, 492 insertions(+) create mode 100644 .github/workflows/documentation.yml create mode 100644 eng/validate-documentation.ps1 diff --git a/.github/workflows/documentation.yml b/.github/workflows/documentation.yml new file mode 100644 index 0000000..05f0486 --- /dev/null +++ b/.github/workflows/documentation.yml @@ -0,0 +1,88 @@ +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" + - "eng/validate-documentation.ps1" + - ".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" + - "eng/validate-documentation.ps1" + - ".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: Setup .NET 8 + uses: actions/setup-dotnet@v5 + with: + dotnet-version: 8.0.x + + - name: Validate Markdown links and compile examples + shell: pwsh + run: ./eng/validate-documentation.ps1 + + - 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: Validate rendered links + shell: pwsh + run: ./eng/validate-documentation.ps1 -SiteRoot ./_site -SkipSnippetCompilation + + - 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/eng/validate-documentation.ps1 b/eng/validate-documentation.ps1 new file mode 100644 index 0000000..7764ae7 --- /dev/null +++ b/eng/validate-documentation.ps1 @@ -0,0 +1,404 @@ +param( + [string] $SiteRoot, + [switch] $SkipSnippetCompilation +) + +$ErrorActionPreference = 'Stop' +Set-StrictMode -Version Latest + +$repositoryRoot = (Resolve-Path (Join-Path $PSScriptRoot '..')).Path +$documentationFiles = @( + Get-Item (Join-Path $repositoryRoot 'README.md'), + (Join-Path $repositoryRoot 'TaskFlow/README.md'), + (Join-Path $repositoryRoot 'TaskFlow.Extensions.Time/README.md'), + (Join-Path $repositoryRoot 'TaskFlow.Extensions.Microsoft.DependencyInjection/README.md'), + (Join-Path $repositoryRoot 'TaskFlow.Extensions.Microsoft.Logging/README.md') +) + @(Get-ChildItem (Join-Path $repositoryRoot 'docs') -Recurse -Filter '*.md') + +function Get-MarkdownAnchors([string] $Path) +{ + $anchors = [System.Collections.Generic.HashSet[string]]::new([System.StringComparer]::OrdinalIgnoreCase) + $duplicates = @{} + + foreach ($line in Get-Content -LiteralPath $Path) + { + if ($line -notmatch '^#{1,6}\s+(.+?)\s*#*\s*$') { continue } + + $heading = $Matches[1] + $heading = $heading -replace '`', '' + $heading = $heading -replace '<[^>]+>', '' + $heading = $heading -replace '\[([^]]+)\]\([^)]+\)', '$1' + $anchor = $heading.ToLowerInvariant() + $anchor = $anchor -replace '[^\p{L}\p{Nd}\s-]', '' + $anchor = ($anchor -replace '\s+', '-' -replace '-+', '-').Trim('-') + + if ($duplicates.ContainsKey($anchor)) + { + $duplicates[$anchor]++ + $anchor = "$anchor-$($duplicates[$anchor])" + } + else + { + $duplicates[$anchor] = 0 + } + + [void]$anchors.Add($anchor) + } + + return $anchors +} + +function Test-MarkdownAnchor([string] $Path, [string] $Anchor, [System.Collections.Generic.List[string]] $Errors) +{ + if ([string]::IsNullOrWhiteSpace($Anchor)) { return } + + $decodedAnchor = [Uri]::UnescapeDataString($Anchor) + if (-not (Get-MarkdownAnchors $Path).Contains($decodedAnchor)) + { + $relativePath = [IO.Path]::GetRelativePath($repositoryRoot, $Path) + $Errors.Add("Missing anchor '#$decodedAnchor' in $relativePath") + } +} + +$permalinkFiles = @{} +foreach ($markdownFile in Get-ChildItem (Join-Path $repositoryRoot 'docs') -Recurse -Filter '*.md') +{ + $markdown = Get-Content -Raw -LiteralPath $markdownFile.FullName + if ($markdown -match '(?m)^permalink:\s*(\S+)\s*$') + { + $permalinkFiles[$Matches[1].TrimEnd('/') + '/'] = $markdownFile.FullName + } +} + +$linkErrors = [System.Collections.Generic.List[string]]::new() +$linkCount = 0 + +foreach ($markdownFile in $documentationFiles) +{ + $markdown = Get-Content -Raw -LiteralPath $markdownFile.FullName + $links = [regex]::Matches($markdown, '!?(?:\[[^]]*\])\((?[^)\s]+)(?:\s+["''][^"'']*["''])?\)') + + foreach ($link in $links) + { + $linkCount++ + $target = $link.Groups['target'].Value.Trim('<', '>') + if ($target -match '^(mailto:|tel:|javascript:|data:)') { continue } + + $pathAndFragment = $target -split '#', 2 + $targetPath = [Uri]::UnescapeDataString($pathAndFragment[0]) + $fragment = if ($pathAndFragment.Count -eq 2) { $pathAndFragment[1] } else { '' } + + if ($target -match '^https?://') + { + $uri = [Uri]$target + if ($uri.Host -ne 'dombrovsky.github.io' -or -not $uri.AbsolutePath.StartsWith('/TaskFlow', [StringComparison]::OrdinalIgnoreCase)) + { + continue + } + + $permalink = $uri.AbsolutePath.Substring('/TaskFlow'.Length).TrimEnd('/') + '/' + if (-not $permalinkFiles.ContainsKey($permalink)) + { + $relativePath = [IO.Path]::GetRelativePath($repositoryRoot, $markdownFile.FullName) + $linkErrors.Add("Missing canonical page '$($uri.AbsolutePath)' linked from $relativePath") + continue + } + + Test-MarkdownAnchor $permalinkFiles[$permalink] $uri.Fragment.TrimStart('#') $linkErrors + continue + } + + $linkedFile = if ([string]::IsNullOrWhiteSpace($targetPath)) + { + $markdownFile.FullName + } + else + { + [IO.Path]::GetFullPath((Join-Path $markdownFile.DirectoryName $targetPath)) + } + + if (-not (Test-Path -LiteralPath $linkedFile -PathType Leaf)) + { + $relativePath = [IO.Path]::GetRelativePath($repositoryRoot, $markdownFile.FullName) + $linkErrors.Add("Missing file '$targetPath' linked from $relativePath") + continue + } + + if ([IO.Path]::GetExtension($linkedFile) -eq '.md') + { + Test-MarkdownAnchor $linkedFile $fragment $linkErrors + } + } +} + +if ($linkErrors.Count -gt 0) +{ + $linkErrors | ForEach-Object { Write-Error $_ } + throw "Documentation link validation failed with $($linkErrors.Count) error(s)." +} + +Write-Output "Validated $linkCount Markdown links across $($documentationFiles.Count) files." + +if (-not [string]::IsNullOrWhiteSpace($SiteRoot)) +{ + $resolvedSiteRoot = (Resolve-Path $SiteRoot).Path + $renderErrors = [System.Collections.Generic.List[string]]::new() + + foreach ($permalink in $permalinkFiles.Keys) + { + $outputPath = Join-Path $resolvedSiteRoot ($permalink.Trim('/') -replace '/', [IO.Path]::DirectorySeparatorChar) + if ($permalink -eq '/') { $outputPath = $resolvedSiteRoot } + $outputFile = Join-Path $outputPath 'index.html' + + if (-not (Test-Path -LiteralPath $outputFile -PathType Leaf)) + { + $renderErrors.Add("Missing rendered page for permalink '$permalink': $outputFile") + } + } + + foreach ($htmlFile in Get-ChildItem $resolvedSiteRoot -Recurse -Filter '*.html') + { + $html = Get-Content -Raw -LiteralPath $htmlFile.FullName + foreach ($match in [regex]::Matches($html, '(?i)href=["''](?[^"'']+)["'']')) + { + $href = [Net.WebUtility]::HtmlDecode($match.Groups['href'].Value) + if ($href -match '^(mailto:|tel:|javascript:|data:)' -or $href.StartsWith('#')) { continue } + + $fragment = '' + $hrefParts = $href -split '#', 2 + $linkPath = $hrefParts[0] -replace '\?.*$', '' + if ($hrefParts.Count -eq 2) { $fragment = [Uri]::UnescapeDataString($hrefParts[1]) } + + if ($linkPath -match '^https?://') + { + $uri = [Uri]$linkPath + if ($uri.Host -ne 'dombrovsky.github.io' -or -not $uri.AbsolutePath.StartsWith('/TaskFlow', [StringComparison]::OrdinalIgnoreCase)) + { + continue + } + $linkPath = $uri.AbsolutePath + } + + if ($linkPath.StartsWith('/TaskFlow', [StringComparison]::OrdinalIgnoreCase)) + { + $linkPath = $linkPath.Substring('/TaskFlow'.Length) + } + + $decodedPath = [Uri]::UnescapeDataString($linkPath) + $candidate = if ($decodedPath.StartsWith('/')) + { + Join-Path $resolvedSiteRoot $decodedPath.TrimStart('/') + } + else + { + Join-Path $htmlFile.DirectoryName $decodedPath + } + + if ([string]::IsNullOrWhiteSpace($decodedPath)) { $candidate = $htmlFile.FullName } + $candidate = [IO.Path]::GetFullPath($candidate) + if ($decodedPath.EndsWith('/')) { $candidate = Join-Path $candidate 'index.html' } + if (-not [IO.Path]::HasExtension($candidate) -and -not (Test-Path -LiteralPath $candidate -PathType Leaf)) + { + $candidate = Join-Path $candidate 'index.html' + } + + if (-not (Test-Path -LiteralPath $candidate -PathType Leaf)) + { + $relativeHtml = [IO.Path]::GetRelativePath($resolvedSiteRoot, $htmlFile.FullName) + $renderErrors.Add("Missing rendered target '$href' linked from $relativeHtml") + continue + } + + if (-not [string]::IsNullOrWhiteSpace($fragment) -and [IO.Path]::GetExtension($candidate) -eq '.html') + { + $targetHtml = Get-Content -Raw -LiteralPath $candidate + $escapedFragment = [regex]::Escape($fragment) + $anchorPattern = '(?i)\bid=["'']{0}["'']' -f $escapedFragment + if ($targetHtml -notmatch $anchorPattern) + { + $relativeHtml = [IO.Path]::GetRelativePath($resolvedSiteRoot, $htmlFile.FullName) + $renderErrors.Add("Missing rendered anchor '#$fragment' for '$href' linked from $relativeHtml") + } + } + } + } + + if ($renderErrors.Count -gt 0) + { + $renderErrors | Sort-Object -Unique | ForEach-Object { Write-Error $_ } + throw "Rendered-site validation failed with $($renderErrors.Count) error(s)." + } + + Write-Output "Validated the rendered site in $resolvedSiteRoot." +} + +if ($SkipSnippetCompilation) { exit 0 } + +$generatedRoot = Join-Path $repositoryRoot 'obj/documentation-validation/generated' +if (Test-Path -LiteralPath $generatedRoot) +{ + Remove-Item -Recurse -Force -LiteralPath $generatedRoot +} +New-Item -ItemType Directory -Path $generatedRoot | Out-Null + +$project = @' + + + net8.0 + latest + enable + enable + false + false + false + false + false + + + + + + + + +'@ +Set-Content -LiteralPath (Join-Path $generatedRoot 'DocumentationExamples.csproj') -Value $project + +$support = @' +global using System; +global using System.IO; +global using System.Threading; +global using System.Threading.Tasks; +global using System.Threading.Tasks.Flow; +global using Microsoft.Extensions.DependencyInjection; +global using Microsoft.Extensions.Logging; +global using Microsoft.Extensions.Logging.Abstractions; + +public interface IDataStore { Task SaveAsync(Data data, CancellationToken token); } +public sealed record Data; +public interface IReadingStore { Task AppendAsync(Reading reading, CancellationToken token); } +public interface IReadingSink { Task HandleAsync(Reading reading, CancellationToken token); } +public interface IReadingSource { event EventHandler ReadingReceived; } +public sealed record Reading; +public sealed class ReadingEventArgs : EventArgs { public required Reading Reading { get; init; } } +public interface ITokenClient { Task RequestAsync(CancellationToken token); } +public sealed record AccessToken(DateTimeOffset ExpiresAt); +public interface ISearchClient { Task SearchAsync(string query, CancellationToken token); } +public interface IResultView { Task ShowAsync(SearchResults results, CancellationToken token); } +public sealed record SearchResults; +public interface IInbox { Task ProcessAvailableAsync(CancellationToken token); } +public sealed class TransientInboxException : Exception; +public sealed record Report; +'@ +Set-Content -LiteralPath (Join-Path $generatedRoot 'Support.cs') -Value $support + +$snippetNumber = 0 +$manifest = [System.Collections.Generic.List[string]]::new() + +foreach ($markdownFile in $documentationFiles) +{ + $markdown = Get-Content -Raw -LiteralPath $markdownFile.FullName + $matches = [regex]::Matches($markdown, '(?ms)^```csharp\s*\r?\n(.*?)^```\s*$') + + foreach ($match in $matches) + { + $snippetNumber++ + $namespace = "DocumentationSnippet$($snippetNumber.ToString('D2'))" + $code = $match.Groups[1].Value.Trim() + $lines = [System.Collections.Generic.List[string]]::new() + foreach ($line in ($code -split '\r?\n')) { $lines.Add($line) } + + $usings = [System.Collections.Generic.List[string]]::new() + while ($lines.Count -gt 0 -and ($lines[0] -match '^using\s+(?:static\s+)?[A-Za-z_][A-Za-z0-9_.]*\s*;\s*$' -or [string]::IsNullOrWhiteSpace($lines[0]))) + { + if (-not [string]::IsNullOrWhiteSpace($lines[0])) { $usings.Add($lines[0]) } + $lines.RemoveAt(0) + } + + $remaining = ($lines -join "`n").Trim() + $typeMatch = [regex]::Match($remaining, '(?m)^public\s+(?:sealed\s+|static\s+|abstract\s+)?(?:class|record|interface)\s+') + $statements = $remaining + $types = '' + if ($typeMatch.Success) + { + $statements = $remaining.Substring(0, $typeMatch.Index).Trim() + $types = $remaining.Substring($typeMatch.Index).Trim() + } + + $builder = [Text.StringBuilder]::new() + [void]$builder.AppendLine('#pragma warning disable') + foreach ($using in $usings) { [void]$builder.AppendLine($using) } + [void]$builder.AppendLine("namespace $namespace") + [void]$builder.AppendLine('{') + + if ($statements.Length -gt 0) + { + [void]$builder.AppendLine(' public static class Example') + [void]$builder.AppendLine(' {') + [void]$builder.AppendLine(' public static async Task RunAsync()') + [void]$builder.AppendLine(' {') + + if ($statements -match '\bflow\b' -and $statements -notmatch '(?:var|ITaskFlow|TaskFlow|CurrentThreadTaskFlow|DedicatedThreadTaskFlow)\s+flow\s*=') + { + [void]$builder.AppendLine(' await using var flow = new TaskFlow();') + } + if ($statements -match '\blogger\b' -and $statements -notmatch '(?:var|ILogger)\s+logger\s*=') + { + [void]$builder.AppendLine(' ILogger logger = NullLogger.Instance;') + } + if ($statements -match '\blifetimeToken\b' -and $statements -notmatch 'CancellationToken\s+lifetimeToken') + { + [void]$builder.AppendLine(' CancellationToken lifetimeToken = default;') + } + if ($statements -match '\bservices\b' -and $statements -notmatch '(?:var|IServiceCollection)\s+services\s*=') + { + [void]$builder.AppendLine(' IServiceCollection services = new ServiceCollection();') + } + if ($statements -match '\bfactory\b' -and $statements -notmatch 'ITaskFlowFactory\s+factory') + { + [void]$builder.AppendLine(' ITaskFlowFactory factory = null!;') + } + + foreach ($line in ($statements -split '\r?\n')) { [void]$builder.AppendLine(" $line") } + + $helpers = @{ + 'SaveAsync' = 'static Task SaveAsync(string value, CancellationToken token) => Task.CompletedTask;' + 'SearchAsync' = 'static Task SearchAsync(CancellationToken token) => Task.CompletedTask;' + 'SendUpdateAsync' = 'static Task SendUpdateAsync(CancellationToken token) => Task.CompletedTask;' + 'PersistOrdersAsync' = 'static Task PersistOrdersAsync(CancellationToken token) => Task.CompletedTask;' + 'PersistAsync' = 'static Task PersistAsync(CancellationToken token) => Task.CompletedTask;' + 'ImportAsync' = 'static Task ImportAsync(CancellationToken token) => Task.CompletedTask;' + 'ExportAsync' = 'static Task ExportAsync(CancellationToken token) => Task.CompletedTask;' + 'RefreshAsync' = 'static Task RefreshAsync(CancellationToken token) => Task.CompletedTask;' + 'ProcessAsync' = 'static Task ProcessAsync(string value, CancellationToken token) => Task.CompletedTask;' + } + foreach ($name in $helpers.Keys) + { + if ($statements -match "\b$name\s*\(" -and $statements -notmatch "(?:static\s+)?(?:async\s+)?Task(?:<[^>]+>)?\s+$name\s*\(") + { + [void]$builder.AppendLine(" $($helpers[$name])") + } + } + + [void]$builder.AppendLine(' }') + [void]$builder.AppendLine(' }') + } + + if ($types.Length -gt 0) + { + foreach ($line in ($types -split '\r?\n')) { [void]$builder.AppendLine(" $line") } + } + [void]$builder.AppendLine('}') + + $fileName = "Snippet$($snippetNumber.ToString('D2')).cs" + Set-Content -LiteralPath (Join-Path $generatedRoot $fileName) -Value $builder.ToString() + $relativePath = [IO.Path]::GetRelativePath($repositoryRoot, $markdownFile.FullName) + $manifest.Add("$fileName`t$relativePath") + } +} + +Set-Content -LiteralPath (Join-Path $generatedRoot 'manifest.txt') -Value $manifest +Write-Output "Generated $snippetNumber C# snippets from $($documentationFiles.Count) Markdown files." + +dotnet build (Join-Path $generatedRoot 'DocumentationExamples.csproj') --configuration Release +if ($LASTEXITCODE -ne 0) { exit $LASTEXITCODE } From 3ee81d5f8d8abfdc90ed658ac84280844eebae8e Mon Sep 17 00:00:00 2001 From: Volodymyr Dombrovskyi <5788605+dombrovsky@users.noreply.github.com> Date: Sun, 16 Aug 2026 12:08:53 -0600 Subject: [PATCH 4/8] Simplify Pages workflow --- .github/workflows/documentation.yml | 15 -- eng/validate-documentation.ps1 | 404 ---------------------------- 2 files changed, 419 deletions(-) delete mode 100644 eng/validate-documentation.ps1 diff --git a/.github/workflows/documentation.yml b/.github/workflows/documentation.yml index 05f0486..cad762a 100644 --- a/.github/workflows/documentation.yml +++ b/.github/workflows/documentation.yml @@ -10,7 +10,6 @@ on: - "TaskFlow.Extensions.Time/README.md" - "TaskFlow.Extensions.Microsoft.DependencyInjection/README.md" - "TaskFlow.Extensions.Microsoft.Logging/README.md" - - "eng/validate-documentation.ps1" - ".github/workflows/documentation.yml" pull_request: branches: [ "main" ] @@ -21,7 +20,6 @@ on: - "TaskFlow.Extensions.Time/README.md" - "TaskFlow.Extensions.Microsoft.DependencyInjection/README.md" - "TaskFlow.Extensions.Microsoft.Logging/README.md" - - "eng/validate-documentation.ps1" - ".github/workflows/documentation.yml" workflow_dispatch: @@ -40,15 +38,6 @@ jobs: - name: Checkout uses: actions/checkout@v6 - - name: Setup .NET 8 - uses: actions/setup-dotnet@v5 - with: - dotnet-version: 8.0.x - - - name: Validate Markdown links and compile examples - shell: pwsh - run: ./eng/validate-documentation.ps1 - - name: Configure GitHub Pages if: github.event_name != 'pull_request' uses: actions/configure-pages@v5 @@ -59,10 +48,6 @@ jobs: source: ./docs destination: ./_site - - name: Validate rendered links - shell: pwsh - run: ./eng/validate-documentation.ps1 -SiteRoot ./_site -SkipSnippetCompilation - - name: Upload GitHub Pages artifact if: github.event_name != 'pull_request' uses: actions/upload-pages-artifact@v4 diff --git a/eng/validate-documentation.ps1 b/eng/validate-documentation.ps1 deleted file mode 100644 index 7764ae7..0000000 --- a/eng/validate-documentation.ps1 +++ /dev/null @@ -1,404 +0,0 @@ -param( - [string] $SiteRoot, - [switch] $SkipSnippetCompilation -) - -$ErrorActionPreference = 'Stop' -Set-StrictMode -Version Latest - -$repositoryRoot = (Resolve-Path (Join-Path $PSScriptRoot '..')).Path -$documentationFiles = @( - Get-Item (Join-Path $repositoryRoot 'README.md'), - (Join-Path $repositoryRoot 'TaskFlow/README.md'), - (Join-Path $repositoryRoot 'TaskFlow.Extensions.Time/README.md'), - (Join-Path $repositoryRoot 'TaskFlow.Extensions.Microsoft.DependencyInjection/README.md'), - (Join-Path $repositoryRoot 'TaskFlow.Extensions.Microsoft.Logging/README.md') -) + @(Get-ChildItem (Join-Path $repositoryRoot 'docs') -Recurse -Filter '*.md') - -function Get-MarkdownAnchors([string] $Path) -{ - $anchors = [System.Collections.Generic.HashSet[string]]::new([System.StringComparer]::OrdinalIgnoreCase) - $duplicates = @{} - - foreach ($line in Get-Content -LiteralPath $Path) - { - if ($line -notmatch '^#{1,6}\s+(.+?)\s*#*\s*$') { continue } - - $heading = $Matches[1] - $heading = $heading -replace '`', '' - $heading = $heading -replace '<[^>]+>', '' - $heading = $heading -replace '\[([^]]+)\]\([^)]+\)', '$1' - $anchor = $heading.ToLowerInvariant() - $anchor = $anchor -replace '[^\p{L}\p{Nd}\s-]', '' - $anchor = ($anchor -replace '\s+', '-' -replace '-+', '-').Trim('-') - - if ($duplicates.ContainsKey($anchor)) - { - $duplicates[$anchor]++ - $anchor = "$anchor-$($duplicates[$anchor])" - } - else - { - $duplicates[$anchor] = 0 - } - - [void]$anchors.Add($anchor) - } - - return $anchors -} - -function Test-MarkdownAnchor([string] $Path, [string] $Anchor, [System.Collections.Generic.List[string]] $Errors) -{ - if ([string]::IsNullOrWhiteSpace($Anchor)) { return } - - $decodedAnchor = [Uri]::UnescapeDataString($Anchor) - if (-not (Get-MarkdownAnchors $Path).Contains($decodedAnchor)) - { - $relativePath = [IO.Path]::GetRelativePath($repositoryRoot, $Path) - $Errors.Add("Missing anchor '#$decodedAnchor' in $relativePath") - } -} - -$permalinkFiles = @{} -foreach ($markdownFile in Get-ChildItem (Join-Path $repositoryRoot 'docs') -Recurse -Filter '*.md') -{ - $markdown = Get-Content -Raw -LiteralPath $markdownFile.FullName - if ($markdown -match '(?m)^permalink:\s*(\S+)\s*$') - { - $permalinkFiles[$Matches[1].TrimEnd('/') + '/'] = $markdownFile.FullName - } -} - -$linkErrors = [System.Collections.Generic.List[string]]::new() -$linkCount = 0 - -foreach ($markdownFile in $documentationFiles) -{ - $markdown = Get-Content -Raw -LiteralPath $markdownFile.FullName - $links = [regex]::Matches($markdown, '!?(?:\[[^]]*\])\((?[^)\s]+)(?:\s+["''][^"'']*["''])?\)') - - foreach ($link in $links) - { - $linkCount++ - $target = $link.Groups['target'].Value.Trim('<', '>') - if ($target -match '^(mailto:|tel:|javascript:|data:)') { continue } - - $pathAndFragment = $target -split '#', 2 - $targetPath = [Uri]::UnescapeDataString($pathAndFragment[0]) - $fragment = if ($pathAndFragment.Count -eq 2) { $pathAndFragment[1] } else { '' } - - if ($target -match '^https?://') - { - $uri = [Uri]$target - if ($uri.Host -ne 'dombrovsky.github.io' -or -not $uri.AbsolutePath.StartsWith('/TaskFlow', [StringComparison]::OrdinalIgnoreCase)) - { - continue - } - - $permalink = $uri.AbsolutePath.Substring('/TaskFlow'.Length).TrimEnd('/') + '/' - if (-not $permalinkFiles.ContainsKey($permalink)) - { - $relativePath = [IO.Path]::GetRelativePath($repositoryRoot, $markdownFile.FullName) - $linkErrors.Add("Missing canonical page '$($uri.AbsolutePath)' linked from $relativePath") - continue - } - - Test-MarkdownAnchor $permalinkFiles[$permalink] $uri.Fragment.TrimStart('#') $linkErrors - continue - } - - $linkedFile = if ([string]::IsNullOrWhiteSpace($targetPath)) - { - $markdownFile.FullName - } - else - { - [IO.Path]::GetFullPath((Join-Path $markdownFile.DirectoryName $targetPath)) - } - - if (-not (Test-Path -LiteralPath $linkedFile -PathType Leaf)) - { - $relativePath = [IO.Path]::GetRelativePath($repositoryRoot, $markdownFile.FullName) - $linkErrors.Add("Missing file '$targetPath' linked from $relativePath") - continue - } - - if ([IO.Path]::GetExtension($linkedFile) -eq '.md') - { - Test-MarkdownAnchor $linkedFile $fragment $linkErrors - } - } -} - -if ($linkErrors.Count -gt 0) -{ - $linkErrors | ForEach-Object { Write-Error $_ } - throw "Documentation link validation failed with $($linkErrors.Count) error(s)." -} - -Write-Output "Validated $linkCount Markdown links across $($documentationFiles.Count) files." - -if (-not [string]::IsNullOrWhiteSpace($SiteRoot)) -{ - $resolvedSiteRoot = (Resolve-Path $SiteRoot).Path - $renderErrors = [System.Collections.Generic.List[string]]::new() - - foreach ($permalink in $permalinkFiles.Keys) - { - $outputPath = Join-Path $resolvedSiteRoot ($permalink.Trim('/') -replace '/', [IO.Path]::DirectorySeparatorChar) - if ($permalink -eq '/') { $outputPath = $resolvedSiteRoot } - $outputFile = Join-Path $outputPath 'index.html' - - if (-not (Test-Path -LiteralPath $outputFile -PathType Leaf)) - { - $renderErrors.Add("Missing rendered page for permalink '$permalink': $outputFile") - } - } - - foreach ($htmlFile in Get-ChildItem $resolvedSiteRoot -Recurse -Filter '*.html') - { - $html = Get-Content -Raw -LiteralPath $htmlFile.FullName - foreach ($match in [regex]::Matches($html, '(?i)href=["''](?[^"'']+)["'']')) - { - $href = [Net.WebUtility]::HtmlDecode($match.Groups['href'].Value) - if ($href -match '^(mailto:|tel:|javascript:|data:)' -or $href.StartsWith('#')) { continue } - - $fragment = '' - $hrefParts = $href -split '#', 2 - $linkPath = $hrefParts[0] -replace '\?.*$', '' - if ($hrefParts.Count -eq 2) { $fragment = [Uri]::UnescapeDataString($hrefParts[1]) } - - if ($linkPath -match '^https?://') - { - $uri = [Uri]$linkPath - if ($uri.Host -ne 'dombrovsky.github.io' -or -not $uri.AbsolutePath.StartsWith('/TaskFlow', [StringComparison]::OrdinalIgnoreCase)) - { - continue - } - $linkPath = $uri.AbsolutePath - } - - if ($linkPath.StartsWith('/TaskFlow', [StringComparison]::OrdinalIgnoreCase)) - { - $linkPath = $linkPath.Substring('/TaskFlow'.Length) - } - - $decodedPath = [Uri]::UnescapeDataString($linkPath) - $candidate = if ($decodedPath.StartsWith('/')) - { - Join-Path $resolvedSiteRoot $decodedPath.TrimStart('/') - } - else - { - Join-Path $htmlFile.DirectoryName $decodedPath - } - - if ([string]::IsNullOrWhiteSpace($decodedPath)) { $candidate = $htmlFile.FullName } - $candidate = [IO.Path]::GetFullPath($candidate) - if ($decodedPath.EndsWith('/')) { $candidate = Join-Path $candidate 'index.html' } - if (-not [IO.Path]::HasExtension($candidate) -and -not (Test-Path -LiteralPath $candidate -PathType Leaf)) - { - $candidate = Join-Path $candidate 'index.html' - } - - if (-not (Test-Path -LiteralPath $candidate -PathType Leaf)) - { - $relativeHtml = [IO.Path]::GetRelativePath($resolvedSiteRoot, $htmlFile.FullName) - $renderErrors.Add("Missing rendered target '$href' linked from $relativeHtml") - continue - } - - if (-not [string]::IsNullOrWhiteSpace($fragment) -and [IO.Path]::GetExtension($candidate) -eq '.html') - { - $targetHtml = Get-Content -Raw -LiteralPath $candidate - $escapedFragment = [regex]::Escape($fragment) - $anchorPattern = '(?i)\bid=["'']{0}["'']' -f $escapedFragment - if ($targetHtml -notmatch $anchorPattern) - { - $relativeHtml = [IO.Path]::GetRelativePath($resolvedSiteRoot, $htmlFile.FullName) - $renderErrors.Add("Missing rendered anchor '#$fragment' for '$href' linked from $relativeHtml") - } - } - } - } - - if ($renderErrors.Count -gt 0) - { - $renderErrors | Sort-Object -Unique | ForEach-Object { Write-Error $_ } - throw "Rendered-site validation failed with $($renderErrors.Count) error(s)." - } - - Write-Output "Validated the rendered site in $resolvedSiteRoot." -} - -if ($SkipSnippetCompilation) { exit 0 } - -$generatedRoot = Join-Path $repositoryRoot 'obj/documentation-validation/generated' -if (Test-Path -LiteralPath $generatedRoot) -{ - Remove-Item -Recurse -Force -LiteralPath $generatedRoot -} -New-Item -ItemType Directory -Path $generatedRoot | Out-Null - -$project = @' - - - net8.0 - latest - enable - enable - false - false - false - false - false - - - - - - - - -'@ -Set-Content -LiteralPath (Join-Path $generatedRoot 'DocumentationExamples.csproj') -Value $project - -$support = @' -global using System; -global using System.IO; -global using System.Threading; -global using System.Threading.Tasks; -global using System.Threading.Tasks.Flow; -global using Microsoft.Extensions.DependencyInjection; -global using Microsoft.Extensions.Logging; -global using Microsoft.Extensions.Logging.Abstractions; - -public interface IDataStore { Task SaveAsync(Data data, CancellationToken token); } -public sealed record Data; -public interface IReadingStore { Task AppendAsync(Reading reading, CancellationToken token); } -public interface IReadingSink { Task HandleAsync(Reading reading, CancellationToken token); } -public interface IReadingSource { event EventHandler ReadingReceived; } -public sealed record Reading; -public sealed class ReadingEventArgs : EventArgs { public required Reading Reading { get; init; } } -public interface ITokenClient { Task RequestAsync(CancellationToken token); } -public sealed record AccessToken(DateTimeOffset ExpiresAt); -public interface ISearchClient { Task SearchAsync(string query, CancellationToken token); } -public interface IResultView { Task ShowAsync(SearchResults results, CancellationToken token); } -public sealed record SearchResults; -public interface IInbox { Task ProcessAvailableAsync(CancellationToken token); } -public sealed class TransientInboxException : Exception; -public sealed record Report; -'@ -Set-Content -LiteralPath (Join-Path $generatedRoot 'Support.cs') -Value $support - -$snippetNumber = 0 -$manifest = [System.Collections.Generic.List[string]]::new() - -foreach ($markdownFile in $documentationFiles) -{ - $markdown = Get-Content -Raw -LiteralPath $markdownFile.FullName - $matches = [regex]::Matches($markdown, '(?ms)^```csharp\s*\r?\n(.*?)^```\s*$') - - foreach ($match in $matches) - { - $snippetNumber++ - $namespace = "DocumentationSnippet$($snippetNumber.ToString('D2'))" - $code = $match.Groups[1].Value.Trim() - $lines = [System.Collections.Generic.List[string]]::new() - foreach ($line in ($code -split '\r?\n')) { $lines.Add($line) } - - $usings = [System.Collections.Generic.List[string]]::new() - while ($lines.Count -gt 0 -and ($lines[0] -match '^using\s+(?:static\s+)?[A-Za-z_][A-Za-z0-9_.]*\s*;\s*$' -or [string]::IsNullOrWhiteSpace($lines[0]))) - { - if (-not [string]::IsNullOrWhiteSpace($lines[0])) { $usings.Add($lines[0]) } - $lines.RemoveAt(0) - } - - $remaining = ($lines -join "`n").Trim() - $typeMatch = [regex]::Match($remaining, '(?m)^public\s+(?:sealed\s+|static\s+|abstract\s+)?(?:class|record|interface)\s+') - $statements = $remaining - $types = '' - if ($typeMatch.Success) - { - $statements = $remaining.Substring(0, $typeMatch.Index).Trim() - $types = $remaining.Substring($typeMatch.Index).Trim() - } - - $builder = [Text.StringBuilder]::new() - [void]$builder.AppendLine('#pragma warning disable') - foreach ($using in $usings) { [void]$builder.AppendLine($using) } - [void]$builder.AppendLine("namespace $namespace") - [void]$builder.AppendLine('{') - - if ($statements.Length -gt 0) - { - [void]$builder.AppendLine(' public static class Example') - [void]$builder.AppendLine(' {') - [void]$builder.AppendLine(' public static async Task RunAsync()') - [void]$builder.AppendLine(' {') - - if ($statements -match '\bflow\b' -and $statements -notmatch '(?:var|ITaskFlow|TaskFlow|CurrentThreadTaskFlow|DedicatedThreadTaskFlow)\s+flow\s*=') - { - [void]$builder.AppendLine(' await using var flow = new TaskFlow();') - } - if ($statements -match '\blogger\b' -and $statements -notmatch '(?:var|ILogger)\s+logger\s*=') - { - [void]$builder.AppendLine(' ILogger logger = NullLogger.Instance;') - } - if ($statements -match '\blifetimeToken\b' -and $statements -notmatch 'CancellationToken\s+lifetimeToken') - { - [void]$builder.AppendLine(' CancellationToken lifetimeToken = default;') - } - if ($statements -match '\bservices\b' -and $statements -notmatch '(?:var|IServiceCollection)\s+services\s*=') - { - [void]$builder.AppendLine(' IServiceCollection services = new ServiceCollection();') - } - if ($statements -match '\bfactory\b' -and $statements -notmatch 'ITaskFlowFactory\s+factory') - { - [void]$builder.AppendLine(' ITaskFlowFactory factory = null!;') - } - - foreach ($line in ($statements -split '\r?\n')) { [void]$builder.AppendLine(" $line") } - - $helpers = @{ - 'SaveAsync' = 'static Task SaveAsync(string value, CancellationToken token) => Task.CompletedTask;' - 'SearchAsync' = 'static Task SearchAsync(CancellationToken token) => Task.CompletedTask;' - 'SendUpdateAsync' = 'static Task SendUpdateAsync(CancellationToken token) => Task.CompletedTask;' - 'PersistOrdersAsync' = 'static Task PersistOrdersAsync(CancellationToken token) => Task.CompletedTask;' - 'PersistAsync' = 'static Task PersistAsync(CancellationToken token) => Task.CompletedTask;' - 'ImportAsync' = 'static Task ImportAsync(CancellationToken token) => Task.CompletedTask;' - 'ExportAsync' = 'static Task ExportAsync(CancellationToken token) => Task.CompletedTask;' - 'RefreshAsync' = 'static Task RefreshAsync(CancellationToken token) => Task.CompletedTask;' - 'ProcessAsync' = 'static Task ProcessAsync(string value, CancellationToken token) => Task.CompletedTask;' - } - foreach ($name in $helpers.Keys) - { - if ($statements -match "\b$name\s*\(" -and $statements -notmatch "(?:static\s+)?(?:async\s+)?Task(?:<[^>]+>)?\s+$name\s*\(") - { - [void]$builder.AppendLine(" $($helpers[$name])") - } - } - - [void]$builder.AppendLine(' }') - [void]$builder.AppendLine(' }') - } - - if ($types.Length -gt 0) - { - foreach ($line in ($types -split '\r?\n')) { [void]$builder.AppendLine(" $line") } - } - [void]$builder.AppendLine('}') - - $fileName = "Snippet$($snippetNumber.ToString('D2')).cs" - Set-Content -LiteralPath (Join-Path $generatedRoot $fileName) -Value $builder.ToString() - $relativePath = [IO.Path]::GetRelativePath($repositoryRoot, $markdownFile.FullName) - $manifest.Add("$fileName`t$relativePath") - } -} - -Set-Content -LiteralPath (Join-Path $generatedRoot 'manifest.txt') -Value $manifest -Write-Output "Generated $snippetNumber C# snippets from $($documentationFiles.Count) Markdown files." - -dotnet build (Join-Path $generatedRoot 'DocumentationExamples.csproj') --configuration Release -if ($LASTEXITCODE -ne 0) { exit $LASTEXITCODE } From 8c2b02abf8e1ea69cf554d734b2b0be8e5b06779 Mon Sep 17 00:00:00 2001 From: Volodymyr Dombrovskyi <5788605+dombrovsky@users.noreply.github.com> Date: Sun, 16 Aug 2026 12:24:05 -0600 Subject: [PATCH 5/8] Refocus README on TaskFlow value --- README.md | 152 ++++++++++++++---------------------------------------- 1 file changed, 40 insertions(+), 112 deletions(-) diff --git a/README.md b/README.md index b7afa3c..90f1bd1 100644 --- a/README.md +++ b/README.md @@ -1,158 +1,86 @@ # TaskFlow for .NET -TaskFlow provides owned FIFO execution lanes for asynchronous .NET work, with composable cancellation, timeout, diagnostics, and thread-affinity policies. +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) -Use TaskFlow to: +## Why TaskFlow? -- serialize asynchronous access to a mutable or non-thread-safe resource; -- preserve event order when synchronous callbacks start asynchronous work; -- bind background work to a component or dependency-injection scope; -- cancel obsolete work when a newer request arrives; -- compose cancellation, timeout, logging, annotations, and error observation; and -- run work on the thread pool, a dedicated thread, or a caller-owned thread. +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. -## Install +TaskFlow packages those concerns into a reusable execution lane. It is useful when you need to: -```shell -dotnet add package TaskFlow -``` +- 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. -Optional integrations: +One flow is one sequential lane. Create separate flows for work that should proceed independently. -```shell -dotnet add package TaskFlow.Microsoft.Extensions.DependencyInjection -dotnet add package TaskFlow.Microsoft.Extensions.Logging -dotnet add package TaskFlow.Extensions.Time -``` +## Features + +| 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. | -`TaskFlow.Extensions.Time` is needed only by consumers that resolve TaskFlow's `netstandard2.0` asset and use `WithThrottle`; .NET 8 and .NET 10 receive that extension from the core package. +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. -## A FIFO execution lane +## Serialize synchronous work for asynchronous callers ```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 => +public interface IDataStore { - Console.WriteLine("second"); - return Task.CompletedTask; -}); - -await Task.WhenAll(first, second); -``` - -`second` starts only after `first` finishes. Each call returns a task for that operation's result, exception, or cancellation. One failed operation does not stop later queued operations. - -## Serialize a resource + void Save(Data data); +} -```csharp -public sealed class SerializedStore : IAsyncDisposable +public sealed class SerializedStore(IDataStore inner) : IAsyncDisposable { - private readonly IDataStore _inner; private readonly TaskFlow _flow = new(); - public SerializedStore(IDataStore inner) - { - _inner = inner; - } - - public Task SaveAsync( - Data data, - CancellationToken cancellationToken = default) - { - return _flow.Enqueue( - token => _inner.SaveAsync(data, token), - cancellationToken); - } + public Task SaveAsync(Data data) => + _flow.Enqueue(() => inner.Save(data)); public ValueTask DisposeAsync() => _flow.DisposeAsync(); } ``` -Callers remain asynchronous while access to the wrapped resource stays ordered and non-concurrent. - -## Latest request wins +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. -```csharp -await using var flow = new TaskFlow(); -ITaskScheduler latest = flow.CreateCancelPrevious(); - -Task search = latest.Enqueue(async token => -{ - await Task.Delay(TimeSpan.FromMilliseconds(250), token); - await SearchAsync(token); -}); +## Install -await search; +```shell +dotnet add package TaskFlow ``` -Every new submission requests cancellation of older unfinished work. The delay creates a latest-request-wins pattern when delegates cooperate with cancellation. - -## Features - -| Capability | API or implementation | Documentation | -|---|---|---| -| FIFO asynchronous execution | `TaskFlow` | [Concepts and lifecycle](https://dombrovsky.github.io/TaskFlow/concepts-and-lifecycle/) | -| Thread affinity | `DedicatedThreadTaskFlow`, `CurrentThreadTaskFlow` | [Execution models](https://dombrovsky.github.io/TaskFlow/execution-models/) | -| Latest request wins | `CreateCancelPrevious` | [Cancellation](https://dombrovsky.github.io/TaskFlow/extensions/cancellation/) | -| Component cancellation | `CreateCancellationScope` | [Cancellation](https://dombrovsky.github.io/TaskFlow/extensions/cancellation/) | -| Queue-and-execution timeout | `WithTimeout` | [Reliability](https://dombrovsky.github.io/TaskFlow/extensions/reliability/) | -| Leading-edge admission throttle | `WithThrottle` | [Reliability](https://dombrovsky.github.io/TaskFlow/extensions/reliability/) | -| Error observation | `OnError` | [Reliability](https://dombrovsky.github.io/TaskFlow/extensions/reliability/) | -| Structured lifecycle logging | `WithLogging` | [Observability](https://dombrovsky.github.io/TaskFlow/extensions/observability/) | -| Scoped and named registration | DI integration package | [Dependency injection](https://dombrovsky.github.io/TaskFlow/dependency-injection/) | - -## When to use something else - -| Primitive | Prefer it when | -|---|---| -| `lock` | The entire critical section is synchronous. | -| `SemaphoreSlim` | Mutual exclusion is enough and callers will manage acquisition, release, ordering, and lifetime. | -| `Channel` | The application is fundamentally a producer/consumer data stream. | -| `BackgroundService` | Work belongs to the application host lifetime rather than a smaller component. | -| TaskFlow | Each submission needs FIFO execution, its own result task, composable policies, and an owned shutdown boundary. | - -## Packages and frameworks - -| Package | Frameworks | -|---|---| -| `TaskFlow` | `netstandard2.0`, `net8.0`, `net10.0` | -| `TaskFlow.Extensions.Time` | `netstandard2.0` | -| `TaskFlow.Microsoft.Extensions.DependencyInjection` | `netstandard2.0`, `net8.0`, `net10.0` | -| `TaskFlow.Microsoft.Extensions.Logging` | `netstandard2.0`, `net8.0`, `net10.0` | - -See the [compatibility matrix](https://dombrovsky.github.io/TaskFlow/compatibility/) for feature-level availability. +Optional dependency-injection, logging, and time integrations are available from the [TaskFlow packages on NuGet](https://www.nuget.org/profiles/dombrovsky). ## Lifecycle essentials -- Prefer `await using`; `DisposeAsync` requests cancellation and waits for the lane to finish. -- Cancellation is cooperative. Synchronous disposal can time out while noncooperative work continues. -- Observe every task returned by `Enqueue`, even when the flow owns the work's lifetime. -- Built-in flows invoke accepted queued delegates with canceled tokens instead of removing them from the lane. +- Observe every task returned by `Enqueue`; ownership does not make failures unobservable. +- 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 [Semantics and pitfalls](https://dombrovsky.github.io/TaskFlow/semantics-and-pitfalls/) before using timeouts or owning long-running background work. +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. ## Documentation -- [Documentation home](https://dombrovsky.github.io/TaskFlow/) - [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/) -Build the repository with a .NET 10 SDK. Tests run against .NET 8 and .NET 10. - TaskFlow is available under the [MIT License](LICENSE). Contributions and problem reports are welcome through [GitHub issues](https://github.com/dombrovsky/TaskFlow/issues). From a55c3e85dbd28b2d91422753ccbbdc863cf2b9a2 Mon Sep 17 00:00:00 2001 From: Volodymyr Dombrovskyi <5788605+dombrovsky@users.noreply.github.com> Date: Sun, 16 Aug 2026 12:47:21 -0600 Subject: [PATCH 6/8] Document DI scheduler pipelines --- docs/dependency-injection.md | 51 +++++++++++++++++++++++++++++++----- 1 file changed, 45 insertions(+), 6 deletions(-) diff --git a/docs/dependency-injection.md b/docs/dependency-injection.md index 9b8aa60..bf64346 100644 --- a/docs/dependency-injection.md +++ b/docs/dependency-injection.md @@ -99,15 +99,54 @@ static Task ImportAsync(CancellationToken token) => Task.CompletedTask; The returned `ITaskFlow` belongs to the caller and must be disposed. -## Advanced registration +## Compose decorators at registration -The advanced `AddTaskFlow` overload accepts delegates for: +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. -- creating the underlying `ITaskFlow`; -- resolving named `TaskFlowOptions`; and -- composing an `ITaskScheduler` decorator chain. +This is especially useful when a named TaskFlow configuration is exposed as a keyed DI service: -Use it to centralize cancellation, timeout, logging, or application-specific wrappers. Preserve the root flow separately from the decorated scheduler so the dependency-injection scope disposes the actual owner. +```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 From 9bd27695ffd26c07868a15d827f0e85d0ddb6fae Mon Sep 17 00:00:00 2001 From: Volodymyr Dombrovskyi <5788605+dombrovsky@users.noreply.github.com> Date: Sun, 16 Aug 2026 12:52:13 -0600 Subject: [PATCH 7/8] List available packages in README --- README.md | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index 90f1bd1..bd849df 100644 --- a/README.md +++ b/README.md @@ -58,13 +58,14 @@ public sealed class SerializedStore(IDataStore inner) : IAsyncDisposable 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. -## Install +## Packages -```shell -dotnet add package TaskFlow -``` - -Optional dependency-injection, logging, and time integrations are available from the [TaskFlow packages on NuGet](https://www.nuget.org/profiles/dombrovsky). +| 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 From dcfd54ed67bb1eead2d49dc633b264afad36999d Mon Sep 17 00:00:00 2001 From: Volodymyr Dombrovskyi <5788605+dombrovsky@users.noreply.github.com> Date: Sun, 16 Aug 2026 15:49:49 -0600 Subject: [PATCH 8/8] Document owned fire-and-forget patterns --- README.md | 2 +- docs/extensions/cancellation.md | 2 +- docs/extensions/index.md | 2 +- docs/getting-started.md | 6 +- docs/recipes.md | 98 +++++++++++++++------------------ docs/semantics-and-pitfalls.md | 6 +- docs/troubleshooting.md | 2 +- 7 files changed, 55 insertions(+), 63 deletions(-) diff --git a/README.md b/README.md index bd849df..1099c84 100644 --- a/README.md +++ b/README.md @@ -69,7 +69,7 @@ Callers receive a task instead of blocking on `Save`. The wrapped synchronous me ## Lifecycle essentials -- Observe every task returned by `Enqueue`; ownership does not make failures unobservable. +- 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`. diff --git a/docs/extensions/cancellation.md b/docs/extensions/cancellation.md index 684792c..24afa23 100644 --- a/docs/extensions/cancellation.md +++ b/docs/extensions/cancellation.md @@ -73,4 +73,4 @@ This is a useful latest-request-wins recipe, but it is not a dedicated trailing- ## Disposal and ownership -Neither cancellation decorator is disposable. Dispose the flow that owns the execution lane. During shutdown, continue observing returned operation tasks so expected cancellation and unexpected failures are distinguished deliberately. +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 index 72348aa..6b6d89b 100644 --- a/docs/extensions/index.md +++ b/docs/extensions/index.md @@ -43,7 +43,7 @@ static Task PersistAsync(CancellationToken token) => Task.CompletedTask; ## Common rules -- Always observe the task returned by `Enqueue`. +- 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. diff --git a/docs/getting-started.md b/docs/getting-started.md index 0598099..9e17c5b 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -38,9 +38,9 @@ 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. -## Await operation tasks +## Choose whether to await an operation -TaskFlow keeps the lane moving after an operation fails, but it does not consume the failure on behalf of the caller. Await returned tasks, return them to a caller, retain them as a component completion signal, or deliberately observe them with application-specific telemetry. +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)); @@ -57,6 +57,8 @@ catch (IOException exception) 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. diff --git a/docs/recipes.md b/docs/recipes.md index 4ea8a03..a8179ed 100644 --- a/docs/recipes.md +++ b/docs/recipes.md @@ -6,7 +6,7 @@ permalink: /recipes/ # Recipes -These examples focus on ownership and observable operation tasks rather than treating queued work as untracked fire-and-forget work. +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 @@ -41,39 +41,37 @@ All callers remain asynchronous while access to `_inner` stays FIFO and non-conc ```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 async void OnReadingReceived(object? sender, ReadingEventArgs args) + private void OnReadingReceived(object? sender, ReadingEventArgs args) { Reading reading = args.Reading; - try - { - await _flow.Enqueue( - token => _sink.HandleAsync(reading, token)); - } - catch (OperationCanceledException) - { - // Expected when the subscriber is disposed. - } - catch (Exception exception) - { - Console.Error.WriteLine(exception); - } + _ = _events.Enqueue( + token => _sink.HandleAsync(reading, token)); } - public ValueTask DisposeAsync() => _flow.DisposeAsync(); + public async ValueTask DisposeAsync() + { + _source.ReadingReceived -= OnReadingReceived; + await _flow.DisposeAsync(); + } } ``` -The event handler copies the event data and calls `Enqueue` before its first suspension, so TaskFlow preserves callback submission order. Event handlers are the conventional exception to avoiding `async void`; this one catches every operation outcome locally. In production, unsubscribe the event before disposing the flow so no callback can submit during shutdown. +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 @@ -162,59 +160,47 @@ Each submission cancels unfinished older submissions. The initial delay means ra ## Own a recoverable background loop ```csharp -public sealed class InboxPump : IAsyncDisposable +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 _flow = new(); - private Task? _completion; + private readonly ILogger _logger; + private readonly TaskFlow _lifetime = new(); - public InboxPump(IInbox inbox, ILogger logger) + public InboxPump(IInbox inbox, ILogger logger) { _inbox = inbox; _logger = logger; } - public Task Completion => _completion ?? Task.CompletedTask; - - public void Start() + public Task StartAsync(CancellationToken cancellationToken) { - _completion ??= _flow.Enqueue(RunAsync); + cancellationToken.ThrowIfCancellationRequested(); + _ = _lifetime.Enqueue(RunAsync); + return Task.CompletedTask; } - public async ValueTask DisposeAsync() - { - await _flow.DisposeAsync(); - - if (_completion is null) - { - return; - } - - try - { - await _completion; - } - catch (OperationCanceledException) - { - // Disposal canceled the loop: normal shutdown. - } - } + public Task StopAsync(CancellationToken cancellationToken) => + _lifetime.DisposeAsync().AsTask().WaitAsync(cancellationToken); private async Task RunAsync(CancellationToken cancellationToken) { - while (true) + while (!cancellationToken.IsCancellationRequested) { - cancellationToken.ThrowIfCancellationRequested(); - - try + await using (var iteration = new TaskFlow()) { - await _inbox.ProcessAvailableAsync(cancellationToken); - } - catch (TransientInboxException exception) - when (!cancellationToken.IsCancellationRequested) - { - _logger.LogWarning(exception, "Inbox iteration failed"); + 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); @@ -223,7 +209,9 @@ public sealed class InboxPump : IAsyncDisposable } ``` -Recoverable failures are caught per iteration so the loop survives them. `Completion` exposes unexpected terminal failures, while disposal cancellation is handled as normal shutdown. An outer `OnError` decorator can report a terminal failure but cannot restart a failed loop. +`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 diff --git a/docs/semantics-and-pitfalls.md b/docs/semantics-and-pitfalls.md index 6fa9f28..4b7adcb 100644 --- a/docs/semantics-and-pitfalls.md +++ b/docs/semantics-and-pitfalls.md @@ -52,9 +52,11 @@ The timeout is cooperative. A delegate that ignores the token can continue runni Prefer asynchronous disposal for component-owned background work and make long-running delegates observe cancellation promptly. -## Disposal is not operation-task observation +## Disposal owns lifetime, not operation outcomes -Disposal waits for lane completion and suppresses operation failures internally. It does not replace awaiting or otherwise observing the task returned for each operation. Store a background loop's completion task so unexpected terminal failures remain visible. +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 diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index 4e534bd..67197d0 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -40,7 +40,7 @@ That is expected for `TaskFlow`, `DedicatedThreadTaskFlow`, and `CurrentThreadTa ## A background loop stopped after one failure -`OnError` observes and rethrows; it does not retry or suppress. Catch recoverable exceptions inside each loop iteration. Retain the loop's returned task so unexpected terminal failures remain observable. See the [background-loop recipe](recipes.md#own-a-recoverable-background-loop). +`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