Cancelling BlockingCollection.GetConsumingEnumerable() and processing what's left

Question

I have one process generating work and a second process with a BlockingCollection<> that consumes that work. When I close my program, I need my consumer to stop consuming work, but I still need to quickly log the work that was pending but hadn't been consumed.

Right now, my consumer spawns a thread that has a foreach (<object> in BlockingCollection.GetConsumingEnumerable()) loop. When I stop my program, my producer calls Consumer.BlockingCollection.CompleteAdding(). What I find is that my consumer continues to process everything in the queue.

Googling the issues tells me that I need to use a CancellationToken. So I tried that out:

private void Process () { // This method runs in a separate thread
    try {
        foreach (*work* in BlockingCollection.GetConsumingEnumerable(CancellationToken)) {
            // Consume
        }
    }
    catch (OperationCancelledException) {
        foreach (*work* in BlockingCollection.GetConsumingEnumerable()) {
            // quickly log
        }
    }
}

My producer has:

private CancellationTokenSource StopFlag = new CancellationTokenSource ();
MyConsumer.CancellationToken = StopFlag.Token;
// Make the consumer spawn it's consuming thread...
StopFlag.Cancel ();
MyConsumer.BlockingCollection.CompleteAdding ();

When I try this, I get no indication that the OperationCancelledException ever happened.

This question tries to explain the use of a cancellation token, but it seems like it's not using it correctly. (Argument: if it works, then it's "correct enough".) And this question would appear to be an exact duplicate of my question but with no example. (Same here.)

So to reiterate: How do I properly use a CancellationToken on BlockingCollection.GetConsumingEnumerable() with the caveat that I need to process the remaining items in the queue after it gets cancelled using a different method?

(I think my problem is centered around the proper use of the CancellationToken. None of my testing indicates that the process is actually being cancelled. (StopFlag.IsCancellationRequested always equals false.))

Answer 1

When you pass in the CancellationToken to GetConsumingEnumerable it won't throw an exception of cancellation is requested, it'll just stop spitting out items. Rather than catching the exception just check the token:

foreach (var item in BlockingCollection.
    GetConsumingEnumerable(CancellationToken))
{
    //consume item
}
if (CancellationToken.IsCancellationRequested)
    foreach (var item in BlockingCollection)
    {
        //log item
    }

Also note that, if cancellation is requested, and it's possible that CompletedAdding hasn't been called then you should just iterate the collection, not call GetConsumingEnumerable. If you know that the producer will complete adding when the operation is cancelled then that's not a problem.

Answer 2

My problem was in how I was trying to cancel the operation. Instead of having my producer owning the CancellationTokenSource, I put it all in the consumer.

public class cProducer {
    private cConsumer myConsumer = new cConsumer ();

    public void onStart () {
        myConsumer.OnStart ();
    }

    public void onStop () {
        myConsumer.OnStop ();
    }

    public void OnOrderReceived (cOrder newOrder) {
        myConsumer.orderQueue.Add (cOrder);
    }
}

public class cConsumer {
    private CancellationTokenSource stopFlag;
    public BlockingCollection<cOrder> orderQueue = new BlockingCollection<cOrder> ();
    private Task processingTask;

    public void OnStart () {
        stopFlag = new CancellationTokenSource ();
        processingTask = Task.Factory.StartNew (() => Process ());
    }

    public void OnStop () {
        stopFlag.Cancel ();
        orderQueue.CompleteAdding ();
        processingTask.Wait ();
    }

    private void Process () {
        try {
            foreach (cOrder newOrder in orderQueue.GetConsumingEnumerable (stopFlag.Token)) {
                // process
            }
        }
        catch (OperationCanceledException) {
            foreach (cOrder cancelledOrder in orderQueue.GetConsumingEnumerable ()) {
                // log it
            }
        }
    }
}

Answer 3

I had the exact same problem. The BlockingCollection seemed to be deadlocked when I cancelled the procedure. The OperationCanceledException was not propagated to the calling method. I figured out that my Producer did not take the cancellation token in consideration and was therefore waiting for the queue to be consumed. All I had to do was to provide the cancellation token in the Add() method. To translate this into Jason's solution above, all I did was this:

public void OnOrderReceived (cOrder newOrder, CancellationToken cancellationToken) 
{
    myConsumer.orderQueue.Add (cOrder, cancellationToken);
}

The Process() method does not need any try-catch clause. However you need to throw if the process is cancelled:

private void Process () 
{
    foreach (cOrder newOrder in orderQueue.GetConsumingEnumerable (stopFlag.Token)) 
    {
        // process
        stopFlag.Token.ThrowIfCancellationRequested();
    }
}