When developing applications, we often work with resources that have limited capacity, like external APIs, hardware devices, and connection pools. If we don't manage access to these resources, it can reduce performance, cause errors, and even cause the application to crash. Luckily, .NET provides several ways to handle concurrency, such as the ones we will explore today: SemaphoreSlim, TPL Dataflow, and Polly's Bulkhead policy.
Let's start by running the following command to set up our project:
dotnet new console -n sandbox
Add a Worker.cs file with the following content:
public class Worker
{
public static async Task DoWork(int id)
{
Console.WriteLine($"Task {id} is working.");
var random = new Random();
var delay = random.Next(1, 5);
await Task.Delay(delay * 1000);
Console.WriteLine($"Task {id} has completed.");
}
}
The DoWork method will simulate accessing the limited resource we want to use.
SemaphoreSlim
SemaphoreSlim is a lightweight synchronization primitive that limits the number of threads or tasks that can access a resource or a pool of resources concurrently. It maintains a count of available slots.
When a task wants to access the resource, it calls WaitAsync() (or Wait() for synchronous code).
If a slot is available, the count is decremented, and the task proceeds.
If no slots are available, the task blocks (asynchronously if using WaitAsync()) until another task calls Release().
When a task is done with the resource, it calls Release(), which increments the count and allows a waiting task to proceed.
It's ideal for scenarios where we want to control the degree of parallelism for a specific block of code within our application. Add the SemaphoreSlimExample.cs file with the following content:
public class SemaphoreSlimExample
{
private static SemaphoreSlim _semaphore = new SemaphoreSlim(3);
public static async Task Run()
{
var tasks = new List<Task>();
for (int i = 0; i < 10; i++)
{
var taskId = i;
tasks.Add(Task.Run(async () =>{
await _semaphore.WaitAsync();
try
{
await Worker.DoWork(taskId);
}
finally
{
_semaphore.Release();
}
}));
}
await Task.WhenAll(tasks);
}
}
Even though we launch ten tasks, SemaphoreSlim ensures that the Worker.DoWork method (the part between WaitAsync and Release) is only executed by a maximum of three tasks at any given time. Others will wait their turn.
TPL Dataflow
The Task Parallel Library (TPL) Dataflow provides a higher-level abstraction for building concurrent applications based on message passing and agent-based models. An ActionBlock<TInput> is a dataflow block that executes a provided delegate for each piece of data it receives. One of its key features for resource control is the ExecutionDataflowBlockOptions.MaxDegreeOfParallelism property. This property dictates how many messages the ActionBlock will process concurrently.
It's perfect for situations where we have a stream of items to process and want to manage how many items are processed at the same time. Add the TplDataflowExample.cs file with the following content:
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
public class TplDataflowExample
{
public static async Task Run()
{
var block = new ActionBlock<int>(Worker.DoWork,
new ExecutionDataflowBlockOptions
{
MaxDegreeOfParallelism = 3
});
for (int i = 0; i < 10; i++)
{
block.Post(i);
}
block.Complete();
await block.Completion;
}
}
The ActionBlock takes care of queuing the ten items. It then picks up items from its input queue and executes the Worker.DoWork method for them, ensuring that no more than three calls are active concurrently.
Polly Bulkhead Isolation Policy
Polly is a .NET resilience and transient-fault-handling library that allows developers to express policies such as Retry, Circuit Breaker, Timeout, and, relevant to our discussion, Bulkhead Isolation.
A Bulkhead policy limits the number of concurrent executions and the number of concurrent operations that can be queued waiting for an execution slot. This is a pattern designed to isolate elements of a system so that if one fails or becomes overloaded, it doesn't cascade and bring down the entire system. The library allows us to set up two properties:
maxParallelization: The maximum number of actions that can be executed concurrently through the bulkhead.
maxQueuingActions: The maximum number of actions that can be queued, waiting for an execution slot. If an action arrives when the execution slots and the queue are full, a BulkheadRejectedException is thrown.
Polly's Bulkhead is especially helpful when working with external services or any operation where we need to strictly control concurrency. It can also reject requests if the application is too busy. Run dotnet add sandbox package Polly, to add the package and add the PollyBulkheadExample.cs file with the following content:
using Polly;
using Polly.Bulkhead;
public class PollyBulkheadExample
{
public static async Task Run()
{
var bulkheadPolicy = Policy.BulkheadAsync(maxParallelization: 3, maxQueuingActions: 10);
var tasks = new List<Task>();
for (int i = 0; i < 10; i++)
{
var taskId = i;
tasks.Add(bulkheadPolicy.ExecuteAsync(() => Worker.DoWork(taskId)));
}
await Task.WhenAll(tasks);
}
}
The BulkheadPolicy will allow up to three calls to run concurrently. If more requests come in, up to ten will be queued. Any further requests, if they arrive while others are still processing and the queue is full, will be immediately rejected with a BulkheadRejectedException. Open the Program.cs file and update the content as follows:
Console.WriteLine("SemaphoreSlim");
await SemaphoreSlimExample.Run();
Console.WriteLine("TPL Dataflow");
await TplDataflowExample.Run();
Console.WriteLine("Polly Bulkhead");
await PollyBulkheadExample.Run();
Run dotnet run --project ./sandbox to see the three techniques in action. You can find all the code here. Thanks, and happy coding.
