ProducerConsumerBuffer.cs

Go to the documentation of this file.

using System;
using System.Collections.Concurrent;
using System.Diagnostics;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
 
using AnalyzeRe.Utilities;
 
namespace AnalyzeRe.LargeDataUpload
{
    public abstract class ProducerConsumerBuffer<TProduced> : IProducerConsumerBuffer<TProduced>
    {
        #region Public Properties
        protected BlockingCollection<TProduced> Produced { get; }
 
        protected bool IsConsumerWaiting { get; private set; }
 
        public bool IsRunning => _runningThread != null;
 
        [System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Design",
            "CA1065:DoNotRaiseExceptionsInUnexpectedLocations")]
        public bool CanTake
        {
            get
            {
                if (_producerLoopException != null)
                    throw new BufferProducerException(_producerLoopException);
                // ReSharper disable once InvertIf (order important to avoid race condition)
                if (Produced.IsAddingCompleted)
                    if (!Produced.Any())
                        return false;
                return true;
            }
        }
        #endregion Public Properties
 
        #region Internal Properties
        protected CancellationTokenSource Cancellation { get; }
 
        private Task _runningThread;
 
        private bool _isDisposed = false;
 
        private Exception _producerLoopException = null;
        #endregion Internal Properties
 
        #region Constructor and Destructor
        protected ProducerConsumerBuffer(int max_queue_size)
        {
            Produced = new BlockingCollection<TProduced>(max_queue_size);
            Cancellation = new CancellationTokenSource();
        }
 
        public void Dispose()
        {
            Dispose(true);
            GC.SuppressFinalize(this);
        }
 
        ~ProducerConsumerBuffer()
        {
            Dispose(false);
        }
 
        protected virtual void Dispose(bool disposing)
        {
            if (!disposing) return;
            if (_isDisposed) return;
            Cancel();
            OnDispose();
            Produced.Dispose();
            _isDisposed = true;
        }
 
        protected virtual void OnDispose() { }
        #endregion Constructor and Destructor
 
        #region Public Methods
        public void Start(CancellationToken? cancellationToken = null)
        {
            string logPrefix = !Utilities.EnableDebugLogging ? null :
                $"{GetType().NiceTypeName()} on thread {Thread.CurrentThread.ManagedThreadId} - ";
            if (Utilities.EnableDebugLogging)
                Debug.WriteLine($"{logPrefix} Starting...");
 
            cancellationToken?.Register(Cancel);
            if (_runningThread != null)
                throw new InvalidOperationException("Producer is already running.");
            OnStart();
            _runningThread = new Task(AsyncProducer);
            _runningThread.Start(TaskScheduler.Default);
 
            if (Utilities.EnableDebugLogging)
                Debug.WriteLine($"{logPrefix} Initialized and new thread started.");
        }
 
        public TProduced TryTake(out bool success, CancellationToken? cancellation)
        {
            string logPrefix = !Utilities.EnableDebugLogging ? null :
                $"{GetType().NiceTypeName()} on thread {Thread.CurrentThread.ManagedThreadId} - ";
 
            try
            {
                // Start the producer thread if it's not already running.
                if (!IsRunning)
                    throw new InvalidOperationException("Producer is not running. You must invoke " +
                        "the Start() method to spawn the producer before blocking for items.");
                IsConsumerWaiting = true;
                if (_producerLoopException != null)
                    throw new BufferProducerException(_producerLoopException);
 
                if(Utilities.EnableDebugLogging)
                    Debug.WriteLine($"{logPrefix}TryTake running with {Produced.Count} buffers in queue.");
 
                success = !cancellation.HasValue ? Produced.TryTake(out TProduced item, -1) :
                    Produced.TryTake(out item, -1, cancellation.Value);
                if (_producerLoopException != null)
                    throw new BufferProducerException(_producerLoopException);
 
                if (Utilities.EnableDebugLogging)
                    Debug.WriteLine($"{logPrefix}TryTake Completed. success={success}. " +
                        $"{Produced.Count} buffers remain in queue.");
                return item;
            }
            finally { IsConsumerWaiting = false; }
        }
 
        public virtual void Cancel()
        {
            Cancellation.Cancel();
            _runningThread?.Wait();
            _runningThread = null;
        }
        #endregion Public Methods
 
        #region Private Methods
        private void AsyncProducer()
        {
            string logPrefix = !Utilities.EnableDebugLogging ? null :
                $"{GetType().NiceTypeName()} on thread {Thread.CurrentThread.ManagedThreadId} - ";
            try
            {
                while (!Cancellation.Token.IsCancellationRequested && !IsProducerFinished())
                {
                    if (Utilities.EnableDebugLogging)
                        Debug.WriteLine($"{logPrefix}Try Produce Next has queued up {Produced.Count} buffers.");
                    if (TryProduceNext(out TProduced next))
                    {
                        if (Utilities.EnableDebugLogging)
                            Debug.WriteLine(logPrefix + "Produced (size: " +
                                (next is BufferedBytes ? ((BufferedBytes)(object)next).LengthFilled + " of " +
                                    ((BufferedBytes)(object)next).Bytes.Length : next.GetType().NiceTypeName()) + ")");
                        Produced.Add(next, Cancellation.Token);
                    }
                    else if (Utilities.EnableDebugLogging)
                        Debug.WriteLine($"{logPrefix}Try Produce Next returned nothing new.");
                }
            }
            catch (OperationCanceledException) when (Cancellation.Token.IsCancellationRequested)
            {
                if (Utilities.EnableDebugLogging)
                    Debug.WriteLine($"{logPrefix}Caught cancellation.");
            }
            catch (Exception ex)
            {
                if (Utilities.EnableDebugLogging)
                    Debug.WriteLine($"{logPrefix}Caught unhandled exception:\n{ex}");
                _producerLoopException = ex;
            }
            finally
            {
                Produced.CompleteAdding();
                if (Utilities.EnableDebugLogging)
                    Debug.WriteLine($"{logPrefix}Completed Adding");
            }
        }
        #endregion Private Methods
 
        #region Abstract and Virtual Methods
        protected virtual void OnStart() { }
 
        protected abstract bool IsProducerFinished();
 
        protected abstract bool TryProduceNext(out TProduced next);
        #endregion Abstract and Virtual Methods
    }
}