AdaptiveBufferedBytesProducer.cs

Go to the documentation of this file.

using System;
 
namespace AnalyzeRe.LargeDataUpload
{
    public abstract class AdaptiveBufferedBytesProducer : ProducerConsumerBuffer<BufferedBytes>
    {
        private const int MaxQueueSize = 3;
 
        private const int BufferScalingFactor = 4;
 
        #region Public Properties
        public int MaximumBufferLength { get; set; } = 33554432; // 2^25 (32 MiB)
 
        public int MinimumBufferLength { get; set; } = 16384; // 2^14 (16 KiB)
 
        public long TotalBytesRead { get; private set; }
        #endregion Public Properties
 
        #region Private Properties
        // Stores the size of the last BufferedBytes byte array produced
        private int _currentBufferSize;
        #endregion Private Properties
 
        #region Constructor
        protected AdaptiveBufferedBytesProducer(int? min_buffer_length = null,
            int? max_buffer_length = null)
            : base(MaxQueueSize)
        {
            MinimumBufferLength = min_buffer_length ?? MinimumBufferLength;
            MaximumBufferLength = max_buffer_length ?? MaximumBufferLength;
        }
        #endregion Constructor
 
        #region Private Methods
        protected override void OnStart()
        {
            if (MaximumBufferLength < MinimumBufferLength)
                MinimumBufferLength = MaximumBufferLength;
            _currentBufferSize = MinimumBufferLength;
        }
 
        protected abstract bool TryProduceNext(out BufferedBytes next, int buffer_size);
 
        protected sealed override bool TryProduceNext(out BufferedBytes buffer)
        {
            if (!TryProduceNext(out buffer, _currentBufferSize))
                return false;
 
            // If the consumer is currently waiting, it took us too long to fill this buffer,
            // so we should reduce the size of the next buffer so that we can deliver it faster.
            if (IsConsumerWaiting)
            {
                while (_currentBufferSize > MinimumBufferLength && _currentBufferSize >= buffer.LengthFilled)
                    _currentBufferSize = Math.Max(_currentBufferSize / BufferScalingFactor, MinimumBufferLength);
            }
            // Otherwise, if the last buffer we produced is still queued up, the consumer is not
            // blocked on this producer. We can try to increase the size of the next buffer
            // we will produce to decrease the overhead of generating multiple smaller buffers.
            else if (Produced.Count >= 1 && _currentBufferSize < MaximumBufferLength)
            {
                _currentBufferSize = Math.Min(_currentBufferSize * BufferScalingFactor, MaximumBufferLength);
            }
 
            TotalBytesRead += buffer.LengthFilled;
            return true;
        }
        #endregion Private Methods
    }
}