C# - Playing with Records - Part Three

Preface

Goal: A simple case example of concurrency using C#.

Concurency means two functions can run separately, without aware of each other presence. Furthermore, we can pass data, between two different functions that run separately. Apart from previously linq topic with C#, we can build a simple concurreny example based on previous data

6: Concurrency with Channel

I have been wondering, if C# have something easy to code, just like goroutine. After searching for days, I finally find a good article about it:

• An Introduction to System.Threading.Channels

Now it is time to stripped down the given example on article above, to suit the songs record use case. Make a simpler example to be learnt with.

The Skeleton

We should be ready for the real demo. This is only consist of one short file.

• gitlab.com/…/csharp/Step17Channel.cs

using …

class Step17Channel
{
  public async Task Sender(ChannelWriter<string> writer){…}

  public async Task Receiver(ChannelReader<string> reader){…}

  public async Task RunBoth() {…}

  public void Show() {…}
}

And in Program class we call as below:

• gitlab.com/…/csharp/SongsApp/Program.cs

class Program
{
  static void Main(string[] args) {
    Step17Channel step = new Step17Channel();
    step.Show();
  }
}

1. Sender(), produce message to channel,

2. Receiver(), consume message from channel,

3. RunBoth(), runinng both thread in asynchronous fashioned.

4. Show(), called in program entry point Main().

Library

We require these two threading libraries below:

using System;
using System.Linq;
using System.Threading.Channels;
using System.Threading.Tasks;

Sender

Produce Messages

The Sender() function will flatten our songs records, and sent each result as message through channel.

  public async Task Sender(ChannelWriter<string> writer)
  {
    MySongs mySongs = new MySongs();

    Array.ForEach(mySongs.songs, song => {
      if (song.Tags != null) {
        Array.ForEach(song.Tags, async (tag) => {
          await writer.WriteAsync(tag);
        });
      };
    });

    …
  }

Every sync function need await, since all previous await wrapped in lambda, we actually do not have await in this funciton. We need to await something, so we have this workround code as below:

  public async Task Sender(ChannelWriter<string> writer)
  {
    …

    writer.Complete();
    await Task.FromResult(1);
    return;
  }

We would gather all event separately to form a new tags list.

Receiver

Consume Messages

The Receiver() function walk for each messaage, and then it push each received message into tags array.

  public async Task Receiver(ChannelReader<string> reader)
  {
    string[] tags = new string[] {};

    while (await reader.WaitToReadAsync()) {
      if (reader.TryRead(out var tag)) {
        tags = tags.Append(tag).ToArray();
      }
    }

   Console.WriteLine("[{0}]",
      string.Join(", ", tags));
  }

Running Both Separately

This should be an async task

Consider gather both functions, wrapped in asynchrounus function.

• gitlab.com/…/csharp/Step17Channel.cs

  public async Task RunBoth()
  {
    var channel = Channel.CreateUnbounded<string>();

    Task consume = Receiver(channel.Reader); 
    Task produce = Sender(channel.Writer);

    await Task.WhenAll(produce, consume);
    return;
  }

Handle as Synchronous Function

Pretty short right!

We just need this Wait() function. Now we can call Show() as usual in Main() entry point.

  public void Show() {
    Task task = RunBoth();
    task.Wait();
  }

With the result as below array:

$ dotnet run
[60s, jazz, 60s, rock, 70s, rock, 70s, pop]

This is the end of our C# journey in this article. We shall meet again in other article.

Further Optimization with Parallel Execution

Parallel.ForEach

You can however, write down above method as below:

• gitlab.com/…/csharp/Step18Parallel.cs

  public async Task Sender(ChannelWriter<string> writer)
  {
    MySongs mySongs = new MySongs();

    await Task.Run(() => 
      Parallel.ForEach(mySongs.songs, song => {
        if (song.Tags != null) {
          Parallel.ForEach(song.Tags, tag => {
            writer.WriteAsync(tag);
          });
        };
      })
    );

    Task.WaitAll();
    writer.Complete();
    return;
  }

With the result exactly the same as previous array:

$ dotnet run
[60s, jazz, 60s, rock, 70s, rock, 70s, pop]

What is Next 🤔?

We have alternative way to do this. And we will also discuss about concurrency.

Consider continue reading [ Python - Playing with Records - Part One ].