2
\$\begingroup\$

I have a process that does a bunch of document processing. One of the tasks being performed is OCR, which is the slowest part of the process, so I decided to run that in another thread.

I decided the producer/consumer would work best because I will be adding document ids as each document is being processed (and there will be thousands of documents processed).

I also decided to use the Parallel.ForEach instead of kicking off many consumer threads because I figured the Parallel.ForEach will utilize the resources much better than I could do it on my own.

Another requirement is that I need to be able to cancel the process.

Question: Are my decision above sound decision (regarding using the BlockingCollection and Parallel.ForEach)?

Review: Is my code resilient? Am I doing things correctly regarding handling the cancellation?

Note: I have some questions within the code.

public class DocumentOcrQueue
{
    private readonly BlockingCollection<int> _documentIds = new BlockingCollection<int>();
    private Task _consumer;
    private CancellationTokenSource _cts;

    public void Add(int documentId)
    {
        _documentIds.Add(documentId);
    }

    /// <summary>
    /// Start the OCR Service.  Once the service has started it will start processing document Ids in the queue
    /// </summary>
    public void Start()
    {
        try
        // The token source for issuing the cancelation request.
        _cts = new CancellationTokenSource();

        var cts = _cts;

        // Question:  I put the cancellation token as the parameter of the 
        //            Task.Run but I am not sure if this is needed.  
        //            Any advice on this?
        _consumer = Task.Run(() => OcrConsumer(cts.Token), cts.Token);
    }

    private void OcrConsumer(CancellationToken cancellationToken)
    {
        var options = new ParallelOptions
        {
            MaxDegreeOfParallelism = Environment.ProcessorCount,
            CancellationToken = cancellationToken
        };

        try
        {
            // GetConsumingPartitioner is an extension method.  See class below.
            Parallel.ForEach(_documentIds.GetConsumingPartitioner(), options, id =>
            {
                options.CancellationToken.ThrowIfCancellationRequested();
                PerformOcr(id);
            });
        }
        catch (OperationCanceledException)
        {
        }
        catch (AggregateException ae)
        {
            bool notCancelledExceptionOccurred = false;
            foreach (var ex in ae.InnerExceptions.Where(e => !(e is OperationCanceledException)))
            {
                notCancelledExceptionOccurred = true;
                // Log exception
            }

            if (notCancelledExceptionOccurred)
                throw;
        }
        catch (Exception ex)
        {
            // Log exception
            throw;
        }
        finally
        {
            _cts.Dispose();
        }
    }

    private void PerformOcr(int documentId)
    {
        // Do OCR work here.
    }

    public void Stop(bool stopImmediately = false)
    {
        _documentIds.CompleteAdding();

        if (stopImmediately)
            _cts.Cancel();

        // Question
        // Does it matter if I use Task.WaitAll or _consumer.Wait()?
        // Is there an advantage of one over the other?
        Task.WaitAll(_consumer);
        // _consumer.Wait();
    }
}

// BlockingCollection.GetConsumingEnumerable and Parallel.ForEach can work against each other so found this sample code which solves this.
// http://blogs.msdn.com/b/pfxteam/archive/2010/04/06/9990420.aspx
static class BlockingCollectionExtention
{
    public static Partitioner<T> GetConsumingPartitioner<T>(this BlockingCollection<T> collection)
    {
        return new BlockingCollectionPartitioner<T>(collection);
    }

    private class BlockingCollectionPartitioner<T> : Partitioner<T>
    {
        private readonly BlockingCollection<T> _collection;

        internal BlockingCollectionPartitioner(
            BlockingCollection<T> collection)
        {
            if (collection == null)
                throw new ArgumentNullException(nameof(collection));

            _collection = collection;
        }

        public override bool SupportsDynamicPartitions => true;

        public override IList<IEnumerator<T>> GetPartitions(int partitionCount)
        {
            if (partitionCount < 1)
                throw new ArgumentOutOfRangeException(nameof(partitionCount));

            var dynamicPartitioner = GetDynamicPartitions();
            return Enumerable.Range(0, partitionCount).Select(_ => dynamicPartitioner.GetEnumerator()).ToArray();
        }

        public override IEnumerable<T> GetDynamicPartitions()
        {
            return _collection.GetConsumingEnumerable();
        }
    }
}
\$\endgroup\$
2
\$\begingroup\$
  1. Nothing prevents Start method from being called multiple times.
  2. Start method code is broken
  3. Task.Run(…, token) allows task cancellation before starting.
  4. Task.Wait() is OK.
  5. DocumentOcrQueue violates SRP. It does job management, error handling, and document OCR.
  6. Also, running LongTask with PreferFairness option could be a way simpler – they worked out well for me in such kind of situation under a heavy load. Scheduler looks smart enough and basically has the same blocking queue inside.

I use the following class for such kind of things:

class TaskGroup : IDisposable
{
    public Task Run(Action<CancellationToken> action)
    {
        var task = new Task(() => action(Cts.Token), 
            Cts.Token, 
            TaskCreationOptions.LongRunning | TaskCreationOptions.PreferFairness);

        byte dummy;
        task.ContinueWith(t => Tasks.TryRemove(task, out dummy));
        task.ContinueWith(t => Error?.Invoke(this, t.Exception), 
            TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.NotOnCanceled);

        Tasks.TryAdd(task, 0);
        task.Start();
        return task; 
    }

    public void Dispose()
    {
        Cts.Cancel();
        Task.WhenAll(Tasks.Keys) // do not throw
            .ContinueWith(t => { }).Wait();
    }

    public event EventHandler<Exception> Error;
    CancellationTokenSource Cts { get; } = new CancellationTokenSource();
    ConcurrentDictionary<Task, byte> Tasks { get; } = new ConcurrentDictionary<Task, byte>();
}

It runs a bunch of tasks and allows to cancel them at once:

    static void Main(string[] args)
    {
        var tg = new TaskGroup();
        tg.Error += (sender, e) => Console.WriteLine(e);
        tg.Run(ct => Ocr(ct, 1));
        tg.Run(ct => Ocr(ct, 2));
        tg.Run(ct => Ocr(ct, 3));

        Console.ReadLine();
        tg.Dispose();
    }

    static void Ocr(CancellationToken token, int id)
    {
        while (!token.IsCancellationRequested)
            Console.WriteLine($"{id}...");
    }
\$\endgroup\$
  • \$\begingroup\$ I will give this a try. A couple questions. It seems your implementation kicks off a different task for every document that needs to be OCR'ed. Am I understanding this correctly? If so, why do a while loop inside the OCR code? It seems like (in your example) the while loop just keeps running the same code until someone presses Enter. Am I missing something? \$\endgroup\$ – Gene S Jun 14 '16 at 16:25
  • \$\begingroup\$ Oh, please put your own code in OCR. The class above (TaskGroup) is OK to be used as it is. OCR is just a dummy thing to simulate real long running job. Do not forget to check the same way the cancelation token somewhere inside your OCR logic. \$\endgroup\$ – Dmitry Nogin Jun 14 '16 at 16:32
  • \$\begingroup\$ I get that part, I am just trying to understand if the expectation is that I should run one task for every document I will be doing OCR against or if I should run a small set of tasks and have each task run a subset of the total documents that need to be OCR'ed. I will have thousands of documents being OCR'ed and it seems like kicking off thousands of threads is a bad idea, but I don't understand the underlying processes of the Task Scheduler so maybe kicking off thousands of threads is okay? \$\endgroup\$ – Gene S Jun 14 '16 at 17:16
  • \$\begingroup\$ I was requesting hundreds of long running tasks at once to generate report pages - it was actually executing around 15 of them on my server - it actually creates a pool of jobs to process when resources becomes available, PreferFairness option converts it to something like a queue. Feel free to invoke TaskGroup.Run whenever you have a document ID. \$\endgroup\$ – Dmitry Nogin Jun 14 '16 at 17:25
  • \$\begingroup\$ Then there is probably something with the OCR process that makes the TaskScheduler think the OCR process takes little resources because I requested 100 documents to be OCR'ed and the Task Scheduler ran all 100 of them at the same time. Within 5 seconds I was getting exceptions occurring within my OCR process. Is there a way you can set the maximum number of threads allowed to be executed at once? \$\endgroup\$ – Gene S Jun 14 '16 at 17:31

Your Answer

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

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