1

I was implementing a Producer-Consumer using BufferBlock. The code is working well.

static async Task Produce(ITargetBlock<int> queue)
{
    try
    {
        // Post messages to the block asynchronously. 
        for (int i = 0; i < 100; i++)
        {
            Console.WriteLine("Sending: {0}", i);
            await queue.SendAsync(i);
        }
    }
    finally 
    {
        queue.Complete();
    }
}

static async Task Consume(ISourceBlock<int> queue)
{
    // Read messages from the block asynchronously. 
    while (await queue.OutputAvailableAsync())
    {
        int value = await queue.ReceiveAsync();
        Console.WriteLine("Receiving: {0}", value);
    }
}

static void Main(string[] args)
{
    // Create a BufferBlock<int> object. 
    var queue = new BufferBlock<int>();

    try
    {
        var produce = Produce(queue);
        var consume = Consume(queue);

        Task.WaitAll(produce, consume, queue.Completion);
    }
    catch (Exception exception)
    {
        Console.WriteLine("An exception was thrown: {0}", exception.Message);
        Console.WriteLine("Terminating...");
    }
}

Now I have a throttling issue, which is I want to the maximum concurrent number for consumer is 4. I want to use SemaphoreSlim bot not sure how to apply it.

Notice: it is a concurrent scheduler question, not parallelism problem.

10
  • What do you mean by "concurrent number for consumer"?
    – i3arnon
    Oct 6, 2014 at 12:28
  • Say at any time we can't consume more than 4 items.
    – user1108948
    Oct 6, 2014 at 12:30
  • 1
    @Love: Why don't you just use ActionBlock with a limited concurrency of 4? Oct 6, 2014 at 12:31
  • @StephenCleary, I had a hard time to understand ActionBlock. I thought it is for parallelism instead of concurrentcy. Parallelism is for multiple cpu cores.
    – user1108948
    Oct 6, 2014 at 12:32
  • @Love Is there any reason using the available cores is undesirable?
    – Gusdor
    Oct 6, 2014 at 12:50

1 Answer 1

3

If all you want is to be able to consume a certain amount at a time you can simply call TryRecieve multiple times until it's empty, or the amount was reached. Here's an extension method that handles that:

public static bool TryReceive<T>(this BufferBlock<T> bufferBlock, int count, out IList<T> items)
{
    items = new List<T>();   
    for (var i = 0; i < count; i++)
    {
        T item;
        if (bufferBlock.TryReceive(out item))
        {
            items.Add(item);
        }
        else
        {
            break;
        }
    }
    return items.Any();
}

And so the consumer becomes:

static async Task Consume(BufferBlock<int> queue)
{
    // Read messages from the block asynchronously. 
    while (await queue.OutputAvailableAsync())
    {
        IList<int> values;
        queue.TryReceive(4, out values);
        Console.WriteLine("Receiving: {0}", string.Join(", ", values));
    }
}
6
  • There are a couple of errors. "No overload for method 'TryReceive' takes 2 arguments. And IList does not contain a definition for 'Any' ....
    – user1108948
    Oct 6, 2014 at 12:53
  • @Love The extension method in my answer takes 2 arguments (on top of the bufferBlock it extends). Any is part of the System.Linq namespace.
    – i3arnon
    Oct 6, 2014 at 12:55
  • So I need to put it in a separate class instead of a method? It is the meaning of the extension?
    – user1108948
    Oct 6, 2014 at 12:58
  • @Love yes. Extension methods need to be inside a static class.
    – i3arnon
    Oct 6, 2014 at 12:59
  • I put it in a separate class. Now the error is Instance argument: cannot convert from 'System.Threading.Tasks.Dataflow.ISourceBlock<int>' to 'Sysyem.Threading.Tasks.Dataflow.BufferBlock<int>'
    – user1108948
    Oct 6, 2014 at 13:07

Your Answer

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