0
\$\begingroup\$

I've developed a toy example to investigate how async works in C#. Though I've learned a lot this week, I'm still unsure of the finer details, as this is my first C# program.

Compilation was done with .Net Core 2.2.

Usage

  • The program runs in an infinite loop. Press any key to stop the program.

Goal

  • Process stream data such that:
    • The stream is never blocked.
    • Data from the stream is bucketed into separate processing tasks based on an Id.
    • This bucketing allows processing tasks to be done in parallel.
    • This bucketing also ensures that data are processed in order.
    • Processed data from all buckets is then fed to a mutual IO operation.

Implementation

  • Data streams from set of Machines.
  • This data is accessible from a single producer -- here a while(true) loop.
  • The data consists of ordered records containing fields {machineId, sensorId, sensorValue}.
    • A single machine may have multiple sensors.
  • To maintain order and introduce parallelism, records are processed in a separate BlockingCollection for each Id.
  • After raw data processing, a new aggregate value is created and then passed to a common IO BlockingCollection.

Issues

  1. Is it reasonable to start multiple BlockingCollection.​getConsumingEnumerable() loops with Task.Run in preparation for data ingestion?
  2. Is passing data between BlockingCollections the way I do it reasonable?
  3. Are there any performance issues with passing data between multiple BlockingCollections?
  4. Am I using Task.Run and await properly?
  5. Am I using CancellationTokens properly?
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

public struct SensorReading
{
    public int machineId;
    public int sensorId;
    public int value;

    public SensorReading(int machineId, int sensorId, int value)
    {
        this.machineId = machineId;
        this.sensorId = sensorId;
        this.value = value;
    }
}

public struct MachineAggregateReading
{
    public int machineId    { get; set; }
    public int aggregateSum { get; set; }

   public MachineAggregateReading(int machineId, int aggregateSum)
    {
        this.machineId = machineId;
        this.aggregateSum = aggregateSum;
    }
}

public class IOQueue {
    private BlockingCollection<MachineAggregateReading> collection;
    private CancellationToken token;

    public IOQueue(CancellationToken token) {
        collection = new BlockingCollection<MachineAggregateReading>(
            new ConcurrentQueue<MachineAggregateReading>());
        this.token = token;
    }

    public void addToQueue(MachineAggregateReading m)
    {
        if (!collection.IsAddingCompleted) {
            collection.Add(m);
        }

    }

    // Return `Task` for an async method that performs an operation but returns no value.
    public async Task processQueue()
    {
        foreach (var s in collection.GetConsumingEnumerable())
        {
            Console.WriteLine("IO: machineId: {0}. aggregateSum {1}. queue count: {2}",
                    s.machineId, s.aggregateSum, collection.Count);
            await Task.Delay(20 * 5, token);

        }
    }

    public void completeAdding() {
        collection.CompleteAdding();
    }
}

public class MachineSensorReadingCollector
{
    private int machineId;
    private IOQueue ioQueue;
    private CancellationToken token;

    private BlockingCollection<SensorReading> collection;
    private Dictionary<int, int> aggregateSensorReadings;

    public MachineSensorReadingCollector(
        int machineId, IOQueue ioQueue, CancellationToken token)
    {
        this.machineId = machineId;
        this.ioQueue = ioQueue;
        this.token = token;

        this.collection = new BlockingCollection<SensorReading>(
            new ConcurrentQueue<SensorReading>());
        aggregateSensorReadings = new Dictionary<int, int>();
    }

    public void addToQueue(SensorReading s)
    {
        if (!collection.IsAddingCompleted) {
            collection.Add(s);
        }

    }

    // Return `Task` for an async method that performs an operation
    // but returns no value.
    public async Task processQueue()
    {
        int aggregateSum;

        foreach (var s in collection.GetConsumingEnumerable())
        {
            // Add to aggregate sensor readings
            if (!aggregateSensorReadings.ContainsKey(s.sensorId)) {
                aggregateSensorReadings[s.sensorId] = 0;
            }

            aggregateSensorReadings[s.sensorId] += s.value;

            Console.WriteLine("Consuming: machineId: {0}. sensorId: {1}. value: {2}. aggregate: {3}. queue count: {4}",
                    machineId, s.sensorId, s.value,
                    aggregateSensorReadings[s.sensorId], collection.Count);

            await Task.Delay(10 * 5, token);

            aggregateSum = 0;
            foreach ((_, int value) in aggregateSensorReadings) {
                aggregateSum += value;
            }

            ioQueue.addToQueue(
                new MachineAggregateReading(machineId, aggregateSum));
        }
    }

    public void completeAdding() {
        collection.CompleteAdding();
    }
}

public class Client
{
    private Random random;
    private CancellationTokenSource tokenSource { get; set; }
    private CancellationToken token { get; set; }

    private int[] machineIdArray;
    private List<Task> machineTasksList;

    private IOQueue ioQueue;
    private Dictionary<int, MachineSensorReadingCollector> machineCollectorDict;

    public Client(int[] machineIdArray) {
        random = new Random();
        tokenSource = new CancellationTokenSource();
        this.token = tokenSource.Token;

        ioQueue = new IOQueue(this.token);
        this.machineIdArray = machineIdArray;
        machineTasksList = new List<Task>();
        machineCollectorDict =
            new Dictionary<int, MachineSensorReadingCollector>();

        foreach (var machineId in machineIdArray)
        {
            machineCollectorDict[machineId] =
                new MachineSensorReadingCollector(
                    machineId: machineId, ioQueue: ioQueue, token: token);
        }
    }

    public async Task Start()
    {
        // Start IO Queue
        //   `Task.Run` must be used here rather than `await`.
        //
        //   Possible:
        //     var ioQueueProcessor = Task.Run(() => ioQueue.processQueue());
        //
        //   Not possible:
        //     await ioQueue.processQueue();
        //
        //   If `await ioQueue.processQueue();` is used the executing thread
        //   will get stuck inside the GetConsumingEnumerable method.
        var ioQueueProcessor = Task.Run(() => ioQueue.processQueue());

        // Start Machine Collectors
        foreach (var (_, machineCollector) in machineCollectorDict)
        {
            // Again, `Task.Run` must be used here rather than `await` for
            // the same reason given above.
            var machineCollectorProcessor =
                Task.Run(() => machineCollector.processQueue());
            machineTasksList.Add(machineCollectorProcessor);
        }

        Console.WriteLine("Press any key to send cancellation token.");

        while (true)
        {

            if (Console.KeyAvailable) {
                tokenSource.Cancel();
                Console.WriteLine("Key pressed. Cancellation token activated.");
                break;
            }

            var s = new SensorReading(
                machineId: machineIdArray[random.Next(machineIdArray.Length)],
                sensorId: random.Next(1, 4),
                value: random.Next(-2, 2));

            Console.WriteLine("Producing: machineId {0}. sensorId {1}. value {2}",
                s.machineId, s.sensorId, s.value);

            machineCollectorDict[s.machineId].addToQueue(s);
            await Task.Delay(30, token);
        }
    }

    public static void Main(string[] args)
    {
        int[] machineIdArray = new int[] { 1, 3, 5 };
        var c = new Client(machineIdArray);

        try
        {
            c.Start().Wait();
        }
        catch (AggregateException agg)
        {
            foreach (var ex in agg.InnerExceptions)
            {
                if (ex is TaskCanceledException)
                {
                    Console.WriteLine("Task successfully canceled.");
                }
                else
                {
                    Console.WriteLine(ex);
                }
            }
        }

        Console.WriteLine("All actions completed.");
    }
}
```
\$\endgroup\$
  • \$\begingroup\$ Your Main() method can be async Task instead of void. This way you can await the Start() task instead of Wait(). \$\endgroup\$ – Brad M Feb 7 at 23:57
  • \$\begingroup\$ Thanks @brad-m. Any other feedback would be much appreciated. \$\endgroup\$ – Mike Ekim Feb 8 at 2:12
  • 1
    \$\begingroup\$ We cannot currently review your code because it contains bugs that make your solution not working properly. We'll be happy to take a look a it when you've fixed them. \$\endgroup\$ – t3chb0t Feb 8 at 7:46
  • 1
    \$\begingroup\$ @t3chb0t I've edited the code. Thanks for reaching out. \$\endgroup\$ – Mike Ekim Feb 8 at 14:01
  • \$\begingroup\$ Writing code like this line while (true) if (!token.IsCancellationRequested) is a very bad style and error-prone - also in many other places where you use a foreach followed by an if without any {} - this is difficult to refactor and can very quickly go sideways. \$\endgroup\$ – t3chb0t Feb 9 at 11:47

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.