armat.threading 3.0.0

Armat Threading Library

Armat.Threading is a powerful alternative to the .NET Task Parallel Library (TPL) that provides fine-grained control over asynchronous execution. The Job and Job<T> classes offer functionality similar to Task and Task<T>, with enhanced capabilities for multi-scheduler scenarios and context propagation.

Why Choose Armat.Threading?

• 🔀 Multiple Thread Pools: Create independent thread pools (JobScheduler instances) within a single application for isolated workload management
• 🔗 Context Propagation: JobRuntimeScope enables seamless passing of contextual data through async call chains
• 🎯 Familiar API: Consistent with System.Threading.Tasks APIs for easy adoption by .NET developers
• ⚡ Async-Await Support: Full compatibility with C# async-await patterns
• 🔧 Extensible: Clear, maintainable codebase designed for customization
• 📊 Observable: Built-in statistics and monitoring capabilities
• 🚀 Performance: Designed to match or exceed TPL performance

Table of Contents

• Installation
• Quick Start
• Core Concepts
  • Job and Job<T>
  • JobScheduler
  • JobRuntimeScope
  • CorrelationIdScope
• Advanced Usage
• Version 3.0.0
• Known Limitations
• Resources
• License

Installation

Install the package via NuGet:

dotnet add package armat.threading

Or via Package Manager:

Install-Package armat.threading

Quick Start

Here's a simple example to get you started:

using Armat.Threading;

// Run a simple asynchronous job
await Job.Run(() =>
{
    Console.WriteLine($"Running on thread {Thread.CurrentThread.ManagedThreadId}");
}).ConfigureAwait(false);

// Run a job with a return value
int result = await Job<int>.Run(() =>
{
    return 42;
}).ConfigureAwait(false);

Console.WriteLine($"Result: {result}");

Core Concepts

Job and Job<T>

Job and Job<T> are the fundamental types for asynchronous operations in Armat.Threading, equivalent to Task and Task<T> in the standard .NET TPL.

Key Features

• Full async-await support with configurable context capture
• Cancellation token support
• Exception handling and aggregation
• Job continuations and parent-child relationships
• Custom scheduling options

Important

⚠️ Always use ConfigureAwait() when awaiting Jobs. The parameterless GetAwaiter() method is obsolete.

Basic Examples

    // run a Job with no args and no return value
    await Job.Run(
        () =>
        {
            Console.WriteLine("Running job asynchronously");
        }).ConfigureAwait(false);

    // run a Job with args but no return value
    await Job.Run(
        (Object? state) =>
        {
            Double number = (Double)state!;
            Console.WriteLine("Running job asynchronously with arg {0}", number);
        }, 0.1248).ConfigureAwait(false);

    // run a Job with no args and returning T
    T result = await Job<T>.Run(
        () =>
        {
            Console.WriteLine("Running job asynchronously");
            return default(T);
        }).ConfigureAwait(false);

    // run a Job with args and returning T
    T result = await Job<T>.Run(
        (Object? state) =>
        {
            Double number = (Double)state!;
            Console.WriteLine("Running job asynchronously with arg {0}", number);
            return new T(number);
        }, 0.1248).ConfigureAwait(false);

There are several overloads of instantiating and running the jobs. It is possible to

• Run a Job with a given cancellation token. This makes possible to request Job cancellation from the other thread. Job execution methods should properly handle the cancellation request and stop the asynchronous execution.

• Run a Job within a given JobScheduler. There may be multiple JobScheduler instances within a single .Net application. JobSchedulers can be used to run asynchronous operation, and all continuations of a Job will derive (run within) the same JobScheduler unless explicitly specified otherwise.

• Pass an argument (state) to the Job. The state is defined as an Object and may be used to pass arguments to the asynchronous operation triggered by the JobScheduler.

• Get the result (return value) of a Job upon completion. Generic implementations of Job<TResult> are running a function with a return value of type TResult. The result of a Job becomes available once it's completed successfully.

• Create a Job with custom JobCreationOptions.

The JobCreationOptions enumeration provides flags that can be combined to control job execution behavior:

During execution of a Job, there's a static property Armat.Threading.Job.Current to be used for identifying the instance of currently running Job.

The Initiator property returns the the instance of parent Job in context of which this one has been triggered. The property Armat.Threading.Job.Root returns the top level Job from which the asynchronous operation has begun.

I'm not going to describe the whole interface of Armat.Threading.Job class taking into account that it generally matches to the System.Threading.Tasks.Task from .Net CLR.

Below is a simple example of running an asynchronous operation using jobs:

    public Int32 Sum(Int32[] array)
    {
        // create the Job
        Job<Int32> asyncJob = new Job<Int32>(SumFn, array);

        // Run asynchronously
        asyncJob.Run();

        // wait for it to finish and return the result
        return asyncJob.Result;
    }

    private Int32 SumFn(Object? args)
    {
        Int32[] array = (Int32[])args;

        return array.Sum();
    }

JobScheduler

The JobScheduler class manages thread pools and job execution queues. It derives from JobSchedulerBase and implements the IJobScheduler interface.

Architecture Overview

┌─────────────────────────────────────┐
│      IJobScheduler Interface        │
├─────────────────────────────────────┤
│  + Default (static)                 │
│  + Current (static)                 │
│  + Enqueue(job)                     │
│  + Cancel(job)                      │
│  + EnterScope()                     │
└─────────────────────────────────────┘
              ▲
              │ implements
              │
┌─────────────┴───────────────────────┐
│    JobSchedulerBase (abstract)      │
├─────────────────────────────────────┤
│  + EnterScope()                     │
│  # UpdateJobStatus()                │
│  # ExecuteJobProcedure()            │
└─────────────────────────────────────┘
              ▲
              │ extends
              │
┌─────────────┴───────────────────────┐
│         JobScheduler                │
├─────────────────────────────────────┤
│  + Statistics                       │
│  + MethodBuilderStatistics          │
│  - Thread Pool Management           │
│  - Job Queue Management             │
└─────────────────────────────────────┘

IJobScheduler Interface

• static IJobScheduler Default { get; } static property returns the default instance of IJobScheduler.
• static IJobScheduler Current { get; } static property returns the instance of IJobScheduler which is currently running a Job on the caller thread, or IJobScheduler.Default otherwise. See JobSchedulerScope for more information about how to change the Current job scheduler.
• void Enqueue(Job job) enqueues a Job in the scheduler. To successfully enqueue a Job in a JobScheduler, the job must have Job.Status = JobStatus.Created (never run before).
• void Enqueue(Job job, DateTimeOffset schedulingTime) enqueues a Job to start the execution at the given time. The Job is kept in a delayed queue and does not occupy a scheduler thread until the scheduling time comes; jobs scheduled for a present or past time are enqueued for immediate execution. See Delayed Execution for details.
• Boolean Cancel(Job job) cancels Job execution in the JobScheduler before it begins. The method returns false if the Job is already running or finished.
• Int32 PendingJobsCount { get; } property returns the number of jobs currently waiting in the queue. Use this to monitor the current load on the JobScheduler.
• JobSchedulerScope EnterScope() updates the IJobScheduler.Current property to point to this instance for the executing thread. The scope is reset when the returned JobSchedulerScope is disposed.

Armat.Threading.JobSchedulerBase abstract class

The class Armat.Threading.JobSchedulerBase provides means for changing the Default and Current Job Schedulers:

• public static IJobScheduler Default { get; protected set; } gets or sets the default IJobScheduler. If not set, returns JobScheduler.WithDefaultConfiguration.
  • Note: To protect setting the default JobScheduler by arbitrary code, the setter is made protected, requiring a public setter in derived classes.
  • Note: JobScheduler.Default can be set only once during process lifetime.
• public JobSchedulerScope EnterScope() updates the IJobScheduler.Current property to point to this job scheduler in context of the calling thread.
  • Note: The class JobSchedulerScope implements IDisposable interface and automatically restores the previous IJobScheduler.Current value upon disposal.

The below example illustrates how to set the Current JobScheduler within a given scope.

    private Int32 ImplicitJobExecutionInCustomScheduler()
    {
        // this is the scheduler to be used in scope of the given method
        using JobScheduler ajs = new("Async Job Scheduler (ajs)");

        // After this line all Jobs will be executed by ajs scheduler (unless overridden by another one)
        // The scope will be disposed when exiting the method, and the previous IJobScheduler.Current will be restored
        using var scope = ajs.EnterScope();

        // create the Job
        Job<Int32> asyncJob = new Job<Int32>(SumFn, new Int32[] { 1, 2, 3 });

        // Run asynchronously
        asyncJob.Run();

        // wait for it to finish and return the result
        return asyncJob.Result;
	}

    private Int32 SumFn(Object? args)
    {
        Int32[] array = (Int32[])args;

        return array.Sum();
    }

Armat.Threading.JobScheduler class

Armat.Threading.JobScheduler is the default implementation of the Armat.Threading.IJobScheduler interface. It can be constructed with a JobSchedulerConfiguration argument to configure:

• Scheduler name (used for naming threads)
• Minimum and maximum number of threads for asynchronous operations
• Maximum number of long-running threads
• Maximum size of job queues within the scheduler

The scheduler provides convenient constructors for common scenarios and access to a default configuration via JobScheduler.WithDefaultConfiguration.

On top of implementing the IJobScheduler interface, the JobScheduler class provides properties to retrieve statistics of Jobs queued within the scheduler. See the Statistics and MethodBuilderStatistics properties for detailed performance metrics.

JobRuntimeScope

The JobRuntimeScope class enables context propagation through asynchronous call chains. It maintains key-value pairs that automatically flow through nested async operations, regardless of call depth or threading complexity.

Use Cases

• Correlation Tracking: Maintain correlation IDs across distributed async operations
• Execution Tracing: Track and log asynchronous execution flows
• Context Delivery: Pass user identity, tenant info, or other context to nested methods
• Request Scoping: Maintain request-specific data in web applications

How It Works

1. Create a scope with JobRuntimeScope.Enter<T>()
2. The scope captures current context and makes data available to all nested calls
3. Data remains accessible through GetValue<T>() at any depth
4. Scope automatically cleans up when disposed (via using statement)

Some examples of using JobRuntimeScope are available as unit tests here. Below is another example of the same:

    private async Job DoSomething()
    {
        // run some user scoped operation
        await RunUserScopedOperation().ConfigureAwait(false);

        // user data is null here because there's no JobRuntimeScope defined at this level
        UserData? userData = JobRuntimeScope.GetValue<UserData>();
    }
    private async Job RunUserScopedOperation()
    {
        // create the scope with some UserData information
        // 'using' keyword guarantees to have the scope Disposed when exiting the method
        using var scope = JobRuntimeScope.Enter<UserData>(() => new UserData("abc", "123"));

        // user data refers to the one declared above
        UserData? userData1 = JobRuntimeScope.GetValue<UserData>();

        // run any asynchronous operation
        // UserData will be accessible in all inner synchronous or asynchronous methods
        await AsyncOperation().ConfigureAwait(false);

        // user data remains the same as above
        UserData? userData2 = JobRuntimeScope.GetValue<UserData>();
    }
    private async Job AsyncOperation()
    {
        // user data remains the same as created in the caller method
        UserData? userData3 = JobRuntimeScope.GetValue<UserData>();

        // running some asynchronous operations
        await Job.Yield();

        // user data remains the same as created in the caller method
        UserData? userData3 = JobRuntimeScope.GetValue<UserData>();
    }

Following are members of the Armat.Threading.JobRuntimeScope class:

• public static JobRuntimeScope Enter(String key, Func<Object> factory) creates a JobRuntimeScope with the given key and uses the factory method to initialize the Value property.
  • Note: If the key already exists in the current scope, JobRuntimeScope.Null is returned to prevent duplicate scopes, and the factory method is never invoked. This includes keys inherited from the Job that initiated the current one — Enter and GetValue resolve keys through the same chain, so an inherited key cannot be shadowed by a nested scope. Always check IsNull on the result.
• public static JobRuntimeScope Enter<T>(String key, Func<T> factory) where T : class is a generic overload of the Enter method.
• public static JobRuntimeScope Enter<T>(Func<T> factory) where T : class is a generic overload that uses the full name of type T as the key.
• public static Object? GetValue(String key) returns the value for the given key in the current scope. Returns null if a scope for the key is not found.
• public static T? GetValue<T>(String key) where T : class returns the value for the given key in the current scope, cast to type T. Returns null if a scope for the key is not found, or if its value is not of type T.
• public static T? GetValue<T>() where T : class is a generic overload that uses the full name of type T as the key.
• public void Leave() leaves the current scope by removing it from the thread context.
• public void Dispose() leaves the current scope as described in the Leave() method. Use with the using keyword to ensure proper disposal when exiting the method scope.
• public Boolean IsNull { get; } returns true for NULL (invalid) scope instances.
• public String Key { get; } property returns the Key of the JobRuntimeScope.
• public Object Value { get; } property returns the Value of the JobRuntimeScope.

CorrelationIdScope class

The class CorrelationIdScope is a value type for JobRuntimeScope that generates auto-incrementing IDs for correlation across asynchronous operations (for logging, tracing, or other needs). It provides convenient factory methods for instantiating JobRuntimeScope with a new CorrelationIdScope value:

• CorrelationIdScope.Create() creates a new scope with an auto-generated correlation ID
• CorrelationIdScope.Create(String name) creates a new scope with a named correlation ID
• CorrelationIdScope.Current() retrieves the current correlation scope (if any)
• CorrelationIdScope.CurrentId() retrieves just the numeric ID (or InvalidID if no scope exists)

The generic version CorrelationIdScope<T> allows for multiple independent correlation ID sequences in the same application.

Example usage:

    private async Job RunCorrelationIDTest(Int32 testNum)
    {
        // create correlation ID and the appropriate scope
        using var scope = CorrelationIdScope.Create();

        // any asynchronous code execution
        await Job.Yield();

        // correlation ID is available here
        Console.WriteLine("RunCorrelationIDTest: Correlation ID for test {0} is {1}",
            testNum,
            CorrelationIdScope.Current()!.CorrelationID);

        // nested method calls
        await NestedAsyncMethodCall(testNum, CorrelationIdScope.Current()!.CorrelationID, 1).ConfigureAwait(false);
    }

    private async Job NestedAsyncMethodCall(Int32 testNum, Int64 expectedCorrID, Int32 depth)
    {
        // any asynchronous code execution
        await Job.Yield();

        // correlation ID remains the same as in the caller method above
        Console.WriteLine("NestedAsyncMethodCall<{0}>: Correlation ID for test {1} is {2}",
            depth,
            testNum,
            CorrelationIdScope.Current()!.CorrelationID);

        // go even deeper
        if (depth < 3)
            await NestedAsyncMethodCall(testNum, expectedCorrID, depth + 1).ConfigureAwait(false);
    }

Appropriate test is available here.

Advanced Usage

Creating Custom Job Schedulers

using Armat.Threading;

// Create a scheduler with custom configuration
var config = new JobSchedulerConfiguration
{
    Name = "HighPriorityScheduler",
    MinThreads = 2,
    MaxThreads = 8,
    MaxLongRunningThreads = 2,
    MaxJobsQueueSize = 1000,
    MaxLongRunningJobsQueueSize = 100
};

using var scheduler = new JobScheduler(config);

// Run jobs in the custom scheduler
await Job.Run(() =>
{
    Console.WriteLine("Running in custom scheduler");
}, CancellationToken.None, JobCreationOptions.None, scheduler)
.ConfigureAwait(false);

Delayed Execution

Jobs can be scheduled to start at a later time. A delayed job is kept in the scheduler's delayed queue — served by a single dispatcher thread that runs only while delayed jobs are pending — and does not occupy a scheduler thread while waiting:

// run a job in 5 seconds
Job delayed = new(() => Console.WriteLine("5 seconds later"));
delayed.Run(DateTimeOffset.UtcNow.AddSeconds(5));

// Job.Delay is built on the same mechanism and does not block a thread either
await Job.Delay(5000).ConfigureAwait(false);

Delayed jobs remain in the Created status until the scheduling time comes, can be canceled with job.Cancel() before they start, and are canceled automatically when their scheduler is disposed. The number of jobs waiting for their start time is available via JobSchedulerBase.DelayedJobsCount.

Monitoring Job Statistics

var scheduler = new JobScheduler("MonitoredScheduler");

// Get real-time statistics
JobSchedulerStatistics stats = scheduler.Statistics;
Console.WriteLine($"Queued: {stats.QueuedJobs}");
Console.WriteLine($"Running: {stats.RunningJobs}");
Console.WriteLine($"Succeeded: {stats.SucceededJobs}");
Console.WriteLine($"Canceled: {stats.CanceledJobs}");
Console.WriteLine($"Faulted: {stats.FaultedJobs}");

// Aggregates: IncompleteJobs = Queued + Running, CompletedJobs = Succeeded + Canceled + Faulted
Console.WriteLine($"Incomplete: {stats.IncompleteJobs}, Completed: {stats.CompletedJobs}");

MethodBuilderStatistics exposes the same counters for the internal jobs created by async Job methods. Counters are updated without locking, so a snapshot is consistent per counter but not across counters — a job in transition may momentarily appear neither queued nor running.

Error Handling

try
{
    await Job.Run(() =>
    {
        throw new InvalidOperationException("Something went wrong");
    }).ConfigureAwait(false);
}
catch (InvalidOperationException ex)
{
    Console.WriteLine($"Job failed: {ex.Message}");
}

Cancellation

Job follows the TPL cancellation model: the token is a request, and the job's final JobStatus is the source of truth.

using CancellationTokenSource cts = new();

Job job = Job.Run(() =>
{
    while (!cts.Token.IsCancellationRequested)
    {
        // ... do a unit of work, then re-check the token
    }

    // acknowledging the request is what marks the Job as Canceled
    cts.Token.ThrowIfCancellationRequested();
}, cts.Token);

cts.Cancel();

try
{
    await job.ConfigureAwait(false);
}
catch (OperationCanceledException)
{
    Console.WriteLine($"Status = {job.Status}");   // Canceled
}

• A running job must cooperate. Nothing aborts a delegate that is already executing — it has to observe the token and throw OperationCanceledException, as above. async Job methods that let an OperationCanceledException escape also complete as Canceled, with nothing recorded in Job.Exception.
• job.Cancel() only cancels a job that has not started yet. It returns true while the job is Created or still waiting in a scheduler queue (the job is dequeued and marked Canceled), and false once execution has begun. Canceling releases everything attached to the job: threads blocked in Wait() wake up, and continuations are scheduled.
• Cancellation is read from the status, not from the token. Once a job has completed successfully, canceling its token afterwards has no effect — Wait(), Result and await still succeed.
• Waiting with a token: Wait(timeout, cancellationToken) throws OperationCanceledException if the token fires before the job completes, while WaitNoThrow(timeout, cancellationToken) returns false. In both cases the job itself keeps running — the token cancels the wait, not the job.
• Already-canceled tokens: use Job.FromCanceled(token) to obtain a pre-canceled job. Passing an already-canceled token to Job.Run throws ArgumentException, because such a job is constructed in the Canceled state and the scheduler refuses to enqueue it. Job.Delay, however, handles this case and returns an already-canceled job.

Combining Multiple Jobs

// Wait for all jobs to complete
var job1 = Job.Run(() => Thread.Sleep(100));
var job2 = Job.Run(() => Thread.Sleep(200));
var job3 = Job.Run(() => Thread.Sleep(150));

Job.WaitAll(job1, job2, job3);
Console.WriteLine("All jobs completed");

// Wait for any job to complete
int completedIndex = Job.WaitAny(job1, job2, job3);
Console.WriteLine($"Job {completedIndex} completed first");

// Use WhenAll for async scenarios
await Job.WhenAll(job1, job2, job3).ConfigureAwait(false);

// Use WhenAny for async scenarios
var completedJob = await Job.WhenAny(job1, job2, job3).ConfigureAwait(false);

Summary

The Armat.Threading library provides a powerful and flexible asynchronous code execution scheduler that serves as an alternative to the .NET Task Parallel Library. Key highlights:

• Full async-await support with Job and Job<T> types
• Multiple independent thread pools within a single application
• Context propagation via JobRuntimeScope for tracing and contextual data
• Production-ready with comprehensive unit tests and performance benchmarks
• .NET 10.0 target framework

Contributions are welcome! Please report bugs, suggest improvements, or submit pull requests.

Version 3.0.0

3.0.0 is a correctness release: it fixes two defects that could block a caller indefinitely (canceling a queued job stranded its waiters; awaiting a job whose token fired never resumed), aligns cancellation with the TPL, and retargets the package to .NET 10.0. It also introduces delayed execution — Job.Delay and jobs started via Job.Run(DateTimeOffset) no longer occupy a scheduler thread while waiting for their start time.

Upgrading from 2.0.x needs attention to these behavior changes:

• .NET 8 and .NET 9 consumers must stay on the 2.0.x line — the package single-targets .NET 10.0.
• Cancellation is reported from JobStatus rather than from the CancellationToken, so a job that completed successfully no longer throws OperationCanceledException once its token is canceled.
• async Job methods that throw OperationCanceledException now complete as Canceled instead of Faulted, with nothing recorded in Job.Exception.
• WaitNoThrow(Int32, CancellationToken) returns false when the token interrupts the wait (it previously reported true).
• JobRuntimeScope.Enter returns JobRuntimeScope.Null for a key inherited from the initiating job, and GetValue<T>(String key) returns null on a type mismatch instead of throwing.

The complete list — every breaking change, fix and improvement — is in ReleaseNotes.md.

Known Limitations

• Initiator chains keep ancestor jobs alive: every Job holds a strong Initiator reference to the job that started it (see Job.Initiator / Job.Root), and the references are not severed on completion. Unbounded continuation chains (e.g. a periodic job endlessly re-scheduling itself from within its own continuation) therefore accumulate completed Job instances that cannot be garbage collected. Break long-living workflows into independent root jobs instead of building unbounded chains.
• WaitAny is limited to 64 jobs: the Job.WaitAny overloads rely on WaitHandle.WaitAny, which supports at most 64 wait handles (63 when a cancellation token is passed). Waiting on more incomplete jobs throws NotSupportedException.

Resources

• NuGet Package: armat.threading
• Source Code: GitHub Repository
• Release Notes: ReleaseNotes.md
• Unit Tests: Threading Tests
• Article: Powerful Alternative to .NET TPL on CodeProject
• Project Website: armat.am/products/threading

License

This project is licensed under the MIT License.