Iterate an array of BlockingCollections

Question

I have an array of BlockingCollections that i have initiated like this:

BlockingCollection<FxDataMapper>[] _fxDataByPair ;
arrayOfBlockingCollection = new BlockingCollection<MyClassInstance>[4];
            for (int i = 0; i < 4; i++)
                _arrayOfBlockingCollection[i] = new BlockingCollection<MyClassInstance>();

Now it want to iterate the array and extract each of the array elements and consume the BlockingCollections elements My code is this and although it works i think its not the best one

private void testArrayConsuming()
{
    Task processor = Task.Factory.StartNew(() =>
    {
        foreach (var x in arrayOfBlockingCollection )
        {
            Task processor2 = Task.Factory.StartNew(() =>
            {
                foreach (var y in x.GetConsumingEnumerable())
                        Console.WriteLine(x.Element.Value.ToString());

            });
        }
    });
}

Answer 1

Your code has a couple of problems:

• It is a void (synchronous) whose only task is to start a Task, lying to the caller since it's in fact an asynchronous Task.
• You are using Task.Factory.StartNew instead of the preferred Task.Run

So, a better approach to this would be something like this:

private async Task testArrayConsuming()
{
    foreach (var x in arrayOfBlockingCollection)
    {
        await Task.Run(() =>
        {
            foreach (var y in x.GetConsumingEnumerable())
            {
                Console.WriteLine(x.Element.Value.ToString());
            }
        });
    }
}

This would block execution, consuming one x at a time. If you want this to be parallel, use this instead:

private async Task testArrayConsuming()
{
    var tasks = new List<Task>();
    foreach (var x in arrayOfBlockingCollection)
    {
        tasks.Add(Task.Run(() =>
        {
            foreach (var y in x.GetConsumingEnumerable())
            {
                Console.WriteLine(x.Element.Value.ToString());
            }
        }));
    }

    await Task.WhenAll(tasks); // wait for all of them
}

Answer 2

You could just stick with the TPL DataFlow Blocks to do this. You can create an extension method to get an element from a blocking collection using the TryTakeFromAny

public static class TPLExtensionMethods
{
    public static IEnumerable<TSource> GetConsumingIndex<TSource>(this BlockingCollection<TSource>[] sources)
    {
        int index;
        do
        {
            TSource result;
            index = BlockingCollection<TSource>.TryTakeFromAny(sources, out result);
            yield return result;
        } while (index != -1);
    }
}

This isn't paralleled out it just reads and pushes a value out when asked for it. This is a building block for us to get to being parallel.

To use it we will need to take your array and transform it using the TransformManyBlock

var transform =
    new TransformManyBlock<BlockingCollection<MyClassInstance>[], MyClassInstance>(
        instances => instances.GetConsumingIndex());

We can now use the ActionBlock to consume the TransformManyBlock output.

var action = new ActionBlock<MyClassInstance>(myclass =>
{
    Console.WriteLine(Thread.CurrentThread.ManagedThreadId +
                      Environment.NewLine +
                      "Class Value: " +
                      myclass.Element.Value.ToString());

}, new ExecutionDataflowBlockOptions()
{
    MaxDegreeOfParallelism = 4 // how parallel do you want it
});

// link up the transform to our action we want to run 
transform.LinkTo(action, new DataflowLinkOptions()
{
    PropagateCompletion = true
});

All that missing now is posting the array to the TransformManyBlock

transform.Post(arrayOfBlockingCollection);

You can tweak the amount of tasks you want by changing the MaxDegreeOfParallelism on the action block. When you are done with your array of BlockingCollections then you need to call transform.Complete()

Answer 3

With the help of your knowledge i have tweaked the code and i come to this which seems to be the best in terms in CPU Usage

private async Task testArrayConsuming()
        {
            await Task.Run(() =>
            {
                foreach (var x in arrayOfBlockingCollection)
                {
                     Task.Run(() =>
                    {
                        foreach (var y in x.GetConsumingEnumerable())
                        {
                            Console.WriteLine(y.Element.Value.ToString());
                        }
                    });
                }
            });
        }

Hope you give me your opinion...