Process a ConcurrentStack when not empty?

Question

I've got a ConcurrentStack that I'm dumping items into. What's a good way to process those items one at a time when the stack isn't empty? I'd like to do this in a way that isn't eating up CPU cycles when the stack isn't being processed.

What I've currently got is basically this and it doesn't seem like an ideal solution.

private void AddToStack(MyObj obj)
{
    stack.Push(obj);
    HandleStack();
}

private void HandleStack()
{
    if (handling)
        return;

    Task.Run( () =>
    {
        lock (lockObj)
        {
            handling = true;
            if (stack.Any())
            {
                //handle whatever is on top of the stack
            }
            handling = false;
        }
    }
}

So the bool is there so multiple threads don't get backed up waiting on the lock. But I don't want multiple things handling the stack at once hence the lock. So if two separate threads do end up calling HandleStack simultaneously and get past the bool, the lock is there so both aren't iterating through the stack at once. But once the second gets through the lock the stack'll be empty and doesn't do anything. So this does end up giving me the behavior I want.

So really I'm just writing a pseudo concurrent wrapper around the ConcurrentStack and it seems like there's got to be a different way to achieve this. Thoughts?

Answer 1

You could consider using the Microsoft TPL Dataflow to do this kind of thing.

Here's a simple example showing how to create a queue. Try it out and play around with the settings for MaxDegreeOfParallelism and BoundedCapacity to see what happens.

For your example, I think you'll want to set MaxDegreeOfParallelism to 1 if you don't want more than one thread handling a data item simultaneously.

(Note: You need to use .Net 4.5x and install TPL Dataflow for the project using Nuget.)

Also have a read of Stephen Cleary's blog about TPL.

using System;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;

namespace SimpleTPL
{
    class MyObj
    {
        public MyObj(string data)
        {
            Data = data;
        }

        public readonly string Data;
    }

    class Program
    {
        static void Main()
        {
            var queue = new ActionBlock<MyObj>(data => process(data), actionBlockOptions());
            var task = queueData(queue);

            Console.WriteLine("Waiting for task to complete.");
            task.Wait();
            Console.WriteLine("Completed.");
        }

        private static void process(MyObj data)
        {
            Console.WriteLine("Processing data " + data.Data);
            Thread.Sleep(200); // Simulate load.
        }

        private static async Task queueData(ActionBlock<MyObj> executor)
        {
            for (int i = 0; i < 20; ++i)
            {
                Console.WriteLine("Queuing data " + i);
                MyObj data = new MyObj(i.ToString());

                await executor.SendAsync(data);
            }

            Console.WriteLine("Indicating that no more data will be queued.");

            executor.Complete(); // Indicate that no more items will be queued.

            Console.WriteLine("Waiting for queue to empty.");

            await executor.Completion; // Wait for executor queue to empty.
        }

        private static ExecutionDataflowBlockOptions actionBlockOptions()
        {
            return new ExecutionDataflowBlockOptions
            {
                MaxDegreeOfParallelism = 4,
                BoundedCapacity        = 8
            };
        }
    }
}

Answer 2

ConcurrentStack<T> is one of the collections that implements IProducerConsumerCollection<T>, and as such can be wrapped by BlockingCollection<T>. BlockingCollection<T> has several convenience members for common operations like "consume while the stack is not empty". E.g., you could call TryTake in a loop. Or, you could just use GetConsumingEnumerable:

private BlockingCollection<MyObj> stack;
private Task consumer;

Constructor()
{
  stack = new BlockingCollection<MyObj>(new ConcurrentStack<MyObj>());
  consumer = Task.Run(() =>
  {
    foreach (var myObj in stack.GetConsumingEnumerable())
    {
      ...
    }
  });
}

private void AddToStack(MyObj obj)
{
  stack.Add(obj);
}

Answer 3

Looks like you want a typical producer consumer.

I would recommend using an autoresetevent

Have your consumer wait when the stack is empty. Call Set when the producer method is called.

Read this thread

Fast and Best Producer/consumer queue technique BlockingCollection vs concurrent Queue