M B A R K's Blog
Threading Channels in C#

Threading Channels in C#

M B A R K's photo
M B A R K
·

10 min read

Table of contents

Threading channels are a synchronization mechanism that allows multiple threads to communicate with each other. They can be used to pass data between threads, or to signal when a particular event has occurred. In this article, we will take a closer look at threading channels in C# and how they can be used to improve the performance of multithreaded applications.

What are Threading Channels?

A threading channel is a communication mechanism that allows multiple threads to communicate with each other. They can be thought of as a pipe that connects two or more threads, through which data or signals can be passed. Threading channels are a type of synchronization mechanism, which means that they are used to coordinate the actions of different threads in a multithreaded application.

There are several types of threading channels available in C#, including:

  • BlockingCollection<T>: This is a thread-safe collection that blocks a thread when it is empty or full, depending on the mode it is in.

  • Channel<T>: This is a lightweight channel that can be used to pass data between threads.

  • Pipe<T>: This is a channel that can be used to pass data between threads in a producer/consumer model.

How to Use Threading Channels in C#

Using threading channels in C# is relatively straightforward. First, you will need to create a new instance of the threading channel you want to use. For example, to create a new BlockingCollection<T>, you can use the following code:

BlockingCollection<int> numbers = new BlockingCollection<int>();

Next, you will need to create one or more threads that will use the threading channel to communicate with each other. This can be done using the Thread class, as shown in the following example:

Thread producerThread = new Thread(() =>
{
    for (int i = 0; i < 100; i++)
    {
        numbers.Add(i);
    }
    numbers.CompleteAdding();
});

Thread consumerThread = new Thread(() =>
{
    while (!numbers.IsCompleted)
    {
        int number = numbers.Take();
        Console.WriteLine(number);
    }
});

In this example, the producerThread adds a series of numbers to the BlockingCollection, while the consumerThread takes them out and prints them to the console. Note that the BlockingCollection has a CompleteAdding() method that can be called to signal that no more items will be added to the collection. This allows the consumer thread to know when to stop processing items.

Once the threads have been created, you can start them using the Start() method:

producerThread.Start();
consumerThread.Start();

Example

Let's take a look at a more detailed example of how to use a threading channel to pass data between threads. In this example, we will use a Channel<T> to pass a series of numbers from one thread to another.

First, we will create the Channel<T> and the two threads:

Channel<int> channel = Channel.CreateUnbounded<int>();
Thread producerThread = new Thread(() =>
{
    for (int i = 0; i < 100; i++)
    {
        channel.Writer.TryWrite(i);
    }
    channel.Writer.TryComplete();
});

This thread will write a series of numbers to the Channel<T>, and then call the TryComplete() method to signal that no more items will be written.

Next, we can create the consumer thread:

Thread consumerThread = new Thread(() =>
{
    while (await channel.Reader.WaitToReadAsync())
    {
        int number = channel.Reader.TryRead();
        Console.WriteLine(number);
    }
});

This thread will use the WaitToReadAsync() method to wait until there is data available in the channel, and then use the TryRead() method to read the data and print it to the console.

Finally, we can start both threads using the Start() method:

producerThread.Start();
consumerThread.Start();

Using the CreateBounded Method

The CreateBounded method is a factory method that can be used to create a new instance of a Channel<T> with a fixed capacity. This can be useful when you want to limit the amount of data that can be stored in the channel at any given time.

Why Use a Bounded Channel?

There are several reasons why you might want to use a bounded channel in your C# application:

  • To limit the amount of memory used by the channel: When using an unbounded channel, the amount of data that can be stored in the channel is limited only by the available memory on the system. This can be a problem if the channel is used to store large amounts of data, as it can quickly consume a significant portion of the available memory. By using a bounded channel, you can limit the amount of data that can be stored in the channel, which can help to prevent memory usage from getting out of control.

  • To improve performance: Bounded channels can also be useful for improving the performance of multithreaded applications. When using an unbounded channel, a producer thread may need to wait for a consumer thread to process data before it can add more data to the channel. This can lead to delays and reduced performance. By using a bounded channel, the producer thread can continue to add data to the channel without waiting for the consumer thread, as long as there is still space available in the channel. This can help to improve the overall performance of the application.

How to Use the CreateBounded Method

Using the CreateBounded method is simple. First, you will need to specify the capacity of the channel as an integer argument. For example:

Channel<int> channel = Channel.CreateBounded<int>(10);

This creates a new Channel<int> with a capacity of 10 items.

Next, you can use the Writer property of the channel to add data to the channel, and the Reader property to read data from the channel. For example:

channel.Writer.TryWrite(1);
channel.Writer.TryWrite(2);
channel.Writer.TryWrite(3);

int number = channel.Reader.TryRead();

In this example, three numbers are added to the channel, and then the first number is read from the channel and stored in the number variable.

Example

Here is a complete example of how to use a bounded channel to coordinate the actions of two threads:

Channel<int> channel = Channel.CreateBounded<int>(10);

Thread producerThread = new Thread(() =>
{
    for (int i = 0; i < 100; i++)
    {
        channel.Writer.TryWrite(i);
    }
});

Thread consumerThread = new Thread(() =>
{
    while (channel.Reader.TryRead(out int number))
    {
        Console.WriteLine(number);
    }
});

producerThread.Start();
consumerThread.Start();

In this example, the producerThread adds a series of numbers to the channel, and the consumerThread reads them and prints them to the console. The bounded channel ensures that the producer thread will not be able to add more than 10 numbers to the channel at a time, which can help to improve the performance of the application.

Note that the TryRead() method used in the consumer thread takes an out parameter, which is used to store the value of the item read from the channel. This allows the consumer thread to read the value of the item without having to store it in a separate variable first.

The CreateBounded method is a useful tool for creating threading channels with a fixed capacity in C#. By using a bounded channel, you can limit the amount of data that can be stored in the channel, which can help to improve the performance and memory usage of your application. In this article, we have seen how to use the CreateBounded method to create a bounded channel, and how to use the channel to coordinate the actions of multiple threads.

Real-World Examples

Here are two real-world examples of how threading channels can be used in C#:

  1. Data processing pipeline: Threading channels can be used to create a data processing pipeline that is composed of multiple stages. For example, a pipeline might consist of a producer thread that reads data from a file, a series of worker threads that perform various transformations on the data, and a consumer thread that writes the transformed data to a database. Each stage of the pipeline can be implemented as a separate thread, and the threads can communicate using threading channels to pass the data from one stage to the next.

  2. Image processing application: Threading channels can also be used to improve the performance of an image processing application. For example, a producer thread could read images from a file and add them to a channel, while a series of worker threads could perform various image processing tasks on the images. The processed images could then be written to another channel, which could be consumed by a consumer thread that displays the images on the screen. By using threading channels to coordinate the actions of the different threads, it is possible to achieve a significant improvement in performance over a single-threaded implementation.

Data processing pipeline

Here is an example of how threading channels could be used to implement a data processing pipeline in C#:

// Create the input and output channels
Channel<string> inputChannel = Channel.CreateUnbounded<string>();
Channel<string> outputChannel = Channel.CreateUnbounded<string>();

// Create the producer thread
Thread producerThread = new Thread(() =>
{
    // Read data from a file
    using (StreamReader reader = new StreamReader("data.txt"))
    {
        string line;
        while ((line = reader.ReadLine()) != null)
        {
            // Add the data to the input channel
            inputChannel.Writer.TryWrite(line);
        }
    }

    // Signal that no more data will be added to the input channel
    inputChannel.Writer.TryComplete();
});

// Create the worker threads
List<Thread> workerThreads = new List<Thread>();
for (int i = 0; i < 4; i++)
{
    Thread workerThread = new Thread(() =>
    {
        // Process data from the input channel
        while (inputChannel.Reader.TryRead(out string line))
        {
            // Perform some transformation on the data
            string transformedLine = line.ToUpper();

            // Add the transformed data to the output channel
            outputChannel.Writer.TryWrite(transformedLine);
        }
    });
    workerThreads.Add(workerThread);
}

// Create the consumer thread
Thread consumerThread = new Thread(() =>
{
    // Write data to the database
    using (SqlConnection connection = new SqlConnection("..."))
    {
        connection.Open();

        while (outputChannel.Reader.TryRead(out string line))
        {
            using (SqlCommand command = new SqlCommand("INSERT INTO table (column) VALUES (@value)", connection))
            {
                command.Parameters.AddWithValue("@value", line);
                command.ExecuteNonQuery();
            }
        }
    }
});

// Start the producer, worker, and consumer threads
producerThread.Start();
foreach (Thread workerThread in workerThreads)
{
    workerThread.Start();
}
consumerThread.Start();

// Wait for all threads to complete
producerThread.Join();
foreach (Thread workerThread in workerThreads)
{
    workerThread.Join();
}
consumerThread.Join();

In this example, the producer thread reads data from a file and adds it to the inputChannel. The worker threads read the data from the inputChannel, perform some transformation on it, and then add it to the outputChannel. Finally, the consumer thread reads the transformed data from the outputChannel and writes it to a database.

By using threading channels to communicate between the different threads, it is possible to create a highly efficient data processing pipeline that is able to process large amounts of data in parallel.

Image processing application

Here is an example of how threading channels could be used to implement an image processing application in C#:

// Create the input and output channels
Channel<Bitmap> inputChannel = Channel.CreateUnbounded<Bitmap>();
Channel<Bitmap> outputChannel = Channel.CreateUnbounded<Bitmap>();

// Create the producer thread
Thread producerThread = new Thread(() =>
{
    // Read images from a file
    using (DirectoryInfo directory = new DirectoryInfo("images"))
    {
        FileInfo[] files = directory.GetFiles("*.jpg");
        foreach (FileInfo file in files)
        {
            // Load the image and add it to the input channel
            using (Bitmap image = new Bitmap(file.FullName))
            {
                inputChannel.Writer.TryWrite(image);
            }
        }
    }

    // Signal that no more images will be added to the input channel
    inputChannel.Writer.TryComplete();
});

// Create the worker threads
List<Thread> workerThreads = new List<Thread>();
for (int i = 0; i < 4; i++)
{
    Thread workerThread = new Thread(() =>
    {
        // Process images from the input channel
        while (inputChannel.Reader.TryRead(out Bitmap image))
        {
            // Perform some image processing on the image
            image = image.Rotate(90);

            // Add the processed image to the output channel
            outputChannel.Writer.TryWrite(image);
        }
    });
    workerThreads.Add(workerThread);
}

// Create the consumer thread
Thread consumerThread = new Thread(() =>
{
    // Display the processed images on the screen
    while (outputChannel.Reader.TryRead(out Bitmap image))
    {
        using (Form form = new Form())
        {
            form.Width = image.Width;
            form.Height = image.Height;
            form.BackgroundImage = image;
            form.ShowDialog();
        }
    }
});

// Start the producer, worker, and consumer threads
producerThread.Start();
foreach (Thread workerThread in workerThreads)
{
    workerThread.Start();
}
consumerThread.Start();

// Wait for all threads to complete
producerThread.Join();
foreach (Thread workerThread in workerThreads)
{
    workerThread.Join();
}
consumerThread.Join();

In this example, the producer thread reads images from a file and adds them to the inputChannel. The worker threads read the images from the inputChannel, perform some image processing on them, and then add them to the outputChannel. Finally, the consumer thread reads the processed images from the outputChannel and displays them on the screen.

By using threading channels to coordinate the actions of the different threads, it is possible to create a highly efficient image processing application that is able to process multiple images in parallel. This can significantly improve the performance of the application compared to a single-threaded implementation.

Conclusion

Threading channels are a powerful synchronization mechanism that can be used to communicate between threads in a multithreaded application. In this article, we have looked at how to use threading channels in C#, including the BlockingCollection<T>, Channel<T>, and Pipe<T> classes. We have also seen an example of how to use a Channel<T> to pass data between threads. With the knowledge of threading channels, you should now be able to design and implement multithreaded applications more effectively in C#.

C#