1
\$\begingroup\$

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());

            });
        }
    });
}
\$\endgroup\$

3 Answers 3

2
\$\begingroup\$

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
}
\$\endgroup\$
3
  • \$\begingroup\$ Don't get me wrong on this ... i tested the code and it works fine but i think its a little more CPU intensive....i will profile it but from judging from the Task manager lets say i see more often devenv climbing to 2% than before....as i said i haven't profile it yet .... \$\endgroup\$
    – John
    Oct 13, 2017 at 10:53
  • \$\begingroup\$ @John In which of the methods would that be? \$\endgroup\$
    – user92754
    Oct 13, 2017 at 11:24
  • \$\begingroup\$ Well i used the 2nd one for testing ...and it seems its just performing different in terms of CPU usage..i am pretty sure it performs a lot better in overall... \$\endgroup\$
    – John
    Oct 13, 2017 at 12:55
1
\$\begingroup\$

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()

\$\endgroup\$
-1
\$\begingroup\$

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...

\$\endgroup\$
2
  • \$\begingroup\$ As I explained in my answer, this approach is wrong regardless of the CPU usage. testArrayConsuming runs a thread just to start another thread \$\endgroup\$
    – user92754
    Oct 13, 2017 at 12:57
  • \$\begingroup\$ ok i will try to stick to your implementation....maybe its interfering with the other Task.Factory.New as i manipulate other BC that manipulate other BC and finally they produce the final...thanks for your help \$\endgroup\$
    – John
    Oct 13, 2017 at 13:18

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.