DEV Community: Josef Matějka

Asynchronous Streams in C#

Josef Matějka — Sat, 31 Jul 2021 08:16:55 +0000

C# Asynchronous streams

I think everyone will agree that collections and data enumeration is usually bread and butter of C# programming. As a programmers we have at our disposal great tools for synchronous enumeration, we have IEnumerable and its generic variant, which can be combined with foreach expression and/or with LINQ. But until C# 8.0 the options for asynchronous enumeration were limited. Iterators were strictly synchronous, thus programmers had to come up with their own solutions - the usual aproach was to create a method returning type Task<IEnumerable<T>>. A task that will asynchronously get a part of the collection and then synchronously enumerate it, when the task succesfully finished. In this article we will explore IAsyncEnumerable<T> which provides an fast and elegant solution to asynchronous enumeration.

Motivation

As I said in some cases it is necesarry to load data asynchronously because

they are behind a blocking call (files, data send through the network),
or we have to wait before data are generated (like in case of Channel<T>), and we don't want to block a thread by waiting for the result. Obtaining part of data asynchronously and then enumerating the result works good from the perfomance perspective, but syntactically it can be a bit awkward and over-complicated considering how elegant does the enumeration look in case of synchronous code.

IAsyncEnumerable and IAsycnEnumerator

I assume the reader is already familiar with IEnumerable<T> and IEnumerator<T>, the asynchronous variant closesly mirrors the synchronous version, but there are few differences.

GetAsyncEnumerator (counterpart of GetEnumerator) accepts a CancellationToken,
MoveNextAsync (counterpart of MoveNext) returns ValueTask<T> instead of T,
and disposing is done via DisposeAsync instead of Dispose.
Reset method is missing (but for synchronous part it is usually not implemented).

As in the case of IEnumerable<T> and IEnumerator<T> programmers usually does not have to implement explicitly, but they can use iterators which will generate it. Since C# 8.0 can also generate IAsyncEnumerable<T> and IAsyncEnumerator<T> via asynchronous iterators and C# provides direct support which allows you to use asynchronous stream in similar way as their synchronous counterpart. Which is why they should feel familiar, therefore it should not be problem to embrace them.

Creating asynchronous iterator block

Rules for the creating a asynchronous iterator with yield return or yield break are again analogous to the synchronous version. Your function must be of a return type IAsyncEnumerable<T> or IAsyncEnumerator<T> it also must be async function. That means no parameter can be ref or out and it should contain an await expression. For example, consider this function:

private static async IAsyncEnumerable<int> FibonacciAsync([EnumeratorCancellation] CancellationToken token = default)
{
    var f0 = 0;
    var f1 = 1;
    yield return f0;
    yield return f1;
    while (true)
    {
        var fn = f0 + f1;
        if (fn < 1)
        {
            yield break;
        }
        yield return fn;
        await Task.Delay(TimeSpan.FromSeconds(1), token).ConfigureAwait(false);
        f0 = f1;
        f1 = fn;
    }
}

This function returns a Fibonacci sequence of all numbers that fit into an int, overflow is check by this condition if (fn < 1), we use Task.Delay for simulation of asynchronous execution, in reality the asynchronous part would be for example an database fetch or any other similar blocking operation.

Since we are talking about asynchronous execution, we have to ponder a bit about cancellation of execution. If we look at GetAsyncEnumerator it accepts an optional CancellationToken. But in our case the IAsyncEnumerable<T> is generated by compiler, how can we work with a token passed to GetAsyncEnumerator? If you put attribute EnumeratorCancellation in front of an CancellationToken argument and compiler knows that for the enumerator cancellation serves said argument.

Enumerating throught asynchronous stream

Let's continue by enumeration of first 5 Fibonacci's numbers. Seasoned C# programmer would immediatelly use Take from System.Linq. Sadly for asynchronous enumerable it is necessary to download nuget package System.Linq.Async first. It contains overloads of classic extension methods for IAsyncEnumerable<T>. We expect that in future version of C# you will not have to download this nuget package, the methods will be automatically incorporated in base framework and no additional packages will be requiered.

public static async Task Main()
{
    var source = new CancellationTokenSource();
    await foreach (var number in FibonacciAsync().Take(5).WithCancellation(source.Token).ConfigureAwait(false))
    {
        Console.WriteLine(number);
    }
}

As you can see, C# supports asynchronous version of foreach with await word in front of it, the difference is that the underneath MoveNextAsync calls are awaited and after finishing the enumerator is disposed asynchronously. If you need to pass a CancellationToken you can do that via extension method WithCancellation, which distributes the token to each internal MoveNextAsync call. This token we get in our FibonacciAsync, because we decorated our token argument with EnumeratorCancellation attribute.

Performance concerns

Like in syncrhonous case, using an async iterator means, that a state machine is generated. It is represented by a hidden class implementing these interfaces: IAsyncEnumerable<T>, IAsyncEnumerator<T>, IAsyncStateMachine, IValueTaskSource<bool>, and IValueTaskSource. The reason why one class implements both IAsyncEnumerable<T> and IAsyncEnumerator<T> is to avoid unnecessary allocations. If you have a class implementing IAsyncEnumerable<T> and another one implementing IAsyncEnumerator<T> you need to create a new instance of enumerator each enumeration. That can be avoided if one class implement both interfaces. If no one is using the object, it can be reused as an enumerator (the same principle also goes for synchronous iterators).

The main trick to keep the allocation low is to use ValueTask<T> over a Task and implementing IValueTaskSource<bool> and IValueTaskSource. As the name suggest, the ValueTask<T> has value semantics (whereas classic Task<T> is a reference type). This is important in cases when the MoveNextAsync call runs synchronously, because in that case the ValueTask<T> can be put on the stack and we can avoid heap allocation and garbage collection of this task - which hleps performance-wise. So it is expected to use async enumeration with some kind of buffering, when in asynchronous call some data are loaded and then they're returned synchronously. If you are interested in the ValueTask<T> more, I recommend reading this blog post: Understanding the Whys, Whats, and Whens of ValueTask, which explains this type and its usage in depth.

Last interface IAsyncStateMachine is only intended for compiler, the runtime and internal APIs uses this interface to perform awaits.

The number of unecessary allocations is low, the microsoft itself states this: "The compiler implementation has been designed to keep allocations incredibly low; in fact, no matter how many times an async iterator yields, the most common case is that it incurs at most two allocations of overhead."

Thread safety warning

This may be surprising to some, but in general MoveNextAsync is not thread-safe. On a given enumerator each call to MoveNextAsync must happen after the last MoveNextAsync has finished, therefore you cannot invoke enumeration concurrently. It is not a problem for a MoveNextAsync to be called from a different thread, but you must not call this function from multiple threads at the same time. The main problem is, that anything can happen. It can seem to be running correctly in some cases, in some cases an exception could be emitted. So if you want to save yourself some debugging time, stay away from enumerating IAsyncEnumerable<T> concurrently.

Conclusion

Newly added asynchronous iterators provide a great tool for dealing with reading data collections asynchronously, this can be useful when loading data from database, files, internet or it can be usefull for in-memory collections too, like in case of Channel data structure. Plus side is, there's very small performance overhead and they closely mirror their synchronous counterpart, which makes them easier to use, because most programmers have already dealt with enumerables.

C# Channels

Josef Matějka — Sat, 31 Jul 2021 08:14:33 +0000

C# Channels

The .NET Core 3.0 has brought a new namespace System.Threading.Channels which contains a synchronization data structure crucial for concurrent programming. In this article I will briefly explain what concurrency is, how is it connected to channels, and why is concurrency in many cases an easy way to parallel computation. It is a programing paradigm that does not necessary need mutexes, semaphores and other synchronization primitives, while it does not inherently suffer from deadlocks and race conditions. Plus, it is simple to test and reason about, because the building blocks are simple single thread oriented units.

What is concurrency?

Concurrency is a style of programming popularized mainly by Go language and its creator Rob Pike. Concurrent program means that It parts (classes/functions/methods) of the program can run with no order, independently. In our case we create classes that do some work and connect them together by a thread-safe pipeline for your data called channel. Therefore we can safely send the data between the classes and we do not have to worry about making the data structures thread-safe. The safety is guaranteed by the channels.

Concurrency does not necessary mean parallel computation, you can run a concurrent program only in one thread. But since the parts of your program are independent and can be run in any order, it is easy to transform any concurrent program to a parallel one.

Let us have some analogy for channels. Imagine a little unorthodox bakery, where three workers work. Each of them is specialized in one task.

WorkerOne can prepare dough,
WorkerTwo can bake bread,
WorkerThree can work at the checkout.

They decided to have two baskets, BasketOne is for the dough and BasketTwo is for the bread and they agreed to this algorithm of work.

When WorkerOne is done, he puts dough into the BasketOne,
WorkerTwo will take dough from the basket whenever he can and starts working,
When WorkerTwo is done, he puts the bread into the BasketTwo,
WorkerThree will take the bread from the basket and puts it on the counter.

The baskets here represent channels, the product is safely stored in a basket until it is needed. All workers are synchronized by the baskets, if a basket is full, the worker has to wait until it is empty again. When the basket is empty and the worker has nothing to do, he must wait until the basket is filled again. But otherwise, they can work independently. And if they find out, that WorkerTwo is a bottleneck for their production, they can always hire a worker with the same specialization and just like that they made the process of baking faster - no other change is necessary, they just need one more worker.

Code Example

Imagine you want to program a concurrent prime sieve (inspired by Go channel tutorial), to demonstrate how you can utilize channels in your C# program. First we need to identify our independent parts of the program. Let us create two classes

Generator - generates natural numbers from 2,
Filter - checks if a number is divisible by a prime number.

Generator class could look like this:

public class Generator
{
    public readonly ChannelReader<int> Reader;
    //from here other classes will get a next natural number

    public Generator()
    {
        var channel = Channel.CreateBounded<int>(1);
        //we create a channel with capacity 1
        Reader = channel.Reader;
        _ = Generate(channel.Writer); //runs task Generate.
    }

    private static async Task Generate(ChannelWriter<int> writer)
    {
        var i = 2;
        while (true)
        {
            await writer.WriteAsync(i).ConfigureAwait(false);
            //puts a natural number into the channel, waits if the channel is full
            i++;
        }
    }
}

When a instance of Generator gets initialized, it will run a task on background, which

puts a natural number i into the channel,
increases i by one,
waits until the channel is empty again.

Filter on the other hand could look like this:

public class Filter
{
    public readonly ChannelReader<int> Reader;
    //from here will other filters get a next potential prime number

    public Filter(ChannelReader<int> input, int primeNumber)
    {
        var channel = Channel.CreateBounded<int>(1);
        //we create a channel with capacity 1
        Reader = channel.Reader;
        _ = FilterLoop(input, channel.Writer, primeNumber); //runs task FilterLoop
    }

    private static async Task FilterLoop(ChannelReader<int> reader, ChannelWriter<int> writer, int primeNumber)
    {
        await foreach (var i in reader.ReadAllAsync().ConfigureAwait(false))
        //reads a value if available, otherwise waits
        {
            if (i % primeNumber != 0) //checks if i is divisible by primeNumber
            {
                await writer.WriteAsync(i).ConfigureAwait(false);
                //puts non-divisible number into the writer
            }
        }
    }
}

When an instance of filter is created, the FilterLoop task will do this:

reads a number from the ChannelReader,
tries to divide it by a prime number,
1. if it is divisible it waits for the next number,
2. if it is not divisible, it will write it into the output channel,
waits for the next number.

The main method then can look like this:

class Program
{
    static async Task Main(string[] args)
    {
        var generator = new Generator();
        var primeReader = generator.Reader;
        for (int i = 0; i < 10; i++) //the loop will print the first ten prime numbers
        {
            var prime = await primeReader.ReadAsync().ConfigureAwait(false);
            Console.WriteLine($"Prime number: {prime}");
            var filter = new Filter(primeReader, prime);
            primeReader = filter.Reader;
        }
    }
}

In the main loop we connect the ChannelReader from Generator instance with the first Filter instance. Then we take the reader from the last filter and connect it to the new instance of Filter. So we end up with a chain of channels, where the Generator generates natural numbers and each filter tests if a number can be prime. Here is a simple diagram to demonstrate the workflow:

First i that passes all filters is the next prime number. Therefore we can print it and use it as a argument for a new Filter instance. Each instance can run independently. If there is something in the input, it will process it, otherwise it will wait in background not blocking any thread.

Unbounded variant

In the example above we have worked with bounded channels, but C# also offers a unbounded variant. The API is the same, but the ChannelWriter will not wait in WriteAsync function - because the channel is never full. This can lead to an exception when the memory is full. I would recommend to use this channel when you know that output of generator (the part that uses ChannelWriter) is somehow limited and we are sure that

we have enough memory,
we don't need to limit generator's throughput.

Consider what would happen if we used unbounded Channel in Generator class. Since the generator class can generate a new number wihout any waiting, the default C# TaskScheduler has no reason to run the Generate function asynchronously, therefore the Channel would be filled by integers until the memory is full and a memory exception is thrown. As you can see using unbounded channels can be potentionally dangerous if we are not careful enough. For that reason I recommend use bounded channels by default and switch to the unbounded ones only when necessary. Since the in the unbounded variant there is no waiting on write, there is no need for TaskScheduler to intervene, which could result in faster computation.

Why C# even needs channels?

You could argue that we do not need channels at all. In the example we could just use TPL in combination with async/await and build an asynchronous prime sieve that would be doing the exact same thing. Of course you avoid channels and use other tools C# provides to build parallel programs. But from my experience it seems that in some cases it is easier to reason, test and scale concurrent programs.

In the current project I am working on, we have noticed that using channels simplifies parallelism, because we only need to create and test few simple classes and then connect them via channels. With almost no effort we end with correct parallel programs. We do not have to be afraid of deadlock, race conditions, we do not have to think how to make our clasess thread safe. Other big advantage is if any class in the concurrent program seems to be a bottleneck. For example if we notice our ChannelReader is most of the time full, we can always spawn another instance of it and connect it to the ChannelReader and with almost no effort we have made our application faster.

If the channel data structure interests you, I recommend to read through Microsoft documentation and look at Rob Pike's presentations Concurrency is not parallelism and Go Concurrency patterns. Eventhought the presentations are aimed for Golang the same principles apply to C#.