Add SlowConsumingPressureMonitor

src/Orleans/Providers/IOrleansProvider.cs

@@ -104,6 +104,17 @@ public static int GetIntProperty(this IProviderConfiguration config, string key,
             return config.Properties.TryGetValue(key, out s) ? int.Parse(s) : settingDefault;
         }
 
+        public static bool TryGetDoubleProperty(this IProviderConfiguration config, string key, out double setting)
+        {
+            if (config == null)
+            {
+                throw new ArgumentNullException("config");
+            }
+            string s;
+            setting = 0;
+            return config.Properties.TryGetValue(key, out s) ? double.TryParse(s, out setting) : false;
+        }
+
         public static string GetProperty(this IProviderConfiguration config, string key, string settingDefault)
         {
             if (config == null)
@@ -163,6 +174,17 @@ public static TimeSpan GetTimeSpanProperty(this IProviderConfiguration config, s
             string s;
             return config.Properties.TryGetValue(key, out s) ? TimeSpan.Parse(s) : settingDefault;
         }
+
+        public static bool TryGetTimeSpanProperty(this IProviderConfiguration config, string key, out TimeSpan setting)
+        {
+            if (config == null)
+            {
+                throw new ArgumentNullException("config");
+            }
+            string s;
+            setting = TimeSpan.Zero;
+            return config.Properties.TryGetValue(key, out s) ? TimeSpan.TryParse(s, out setting) : false;
+        }
     }
 
     /// <summary>

src/OrleansServiceBus/OrleansServiceBus.csproj

@@ -47,6 +47,10 @@
     </Compile>
     <Compile Include="ErrorCode.cs" />
     <Compile Include="Properties\AssemblyInfo.cs" />
+    <Compile Include="Providers\Streams\EventHub\CachePressureMonitors\AggregatedCachePressureMonitor.cs" />
+    <Compile Include="Providers\Streams\EventHub\CachePressureMonitors\AveragingCachePressureMonitor.cs" />
+    <Compile Include="Providers\Streams\EventHub\CachePressureMonitors\ICachePressureMonitor.cs" />
+    <Compile Include="Providers\Streams\EventHub\CachePressureMonitors\SlowConsumingPressureMonitor.cs" />
     <Compile Include="Providers\Streams\EventHub\DefaultEventHubReceiverMonitor.cs" />
     <Compile Include="Providers\Streams\EventHub\EventDataExtensions.cs" />
     <Compile Include="Providers\Streams\EventHub\EventHubAdapterFactory.cs" />

src/OrleansServiceBus/Providers/Streams/EventHub/CachePressureMonitors/AggregatedCachePressureMonitor.cs

@@ -0,0 +1,67 @@
+using Orleans.Runtime;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace Orleans.ServiceBus.Providers
+{
+    /// <summary>
+    /// Aggregated cache pressure monitor
+    /// </summary>
+    public class AggregatedCachePressureMonitor : List<ICachePressureMonitor>, ICachePressureMonitor
+    {
+        private bool isUnderPressure;
+        private Logger logger;
+
+        /// <summary>
+        /// Constructor
+        /// </summary>
+        /// <param name="logger"></param>
+        public AggregatedCachePressureMonitor(Logger logger)
+        {
+            this.isUnderPressure = false;
+            this.logger = logger.GetSubLogger(this.GetType().Name);
+        }
+
+        /// <summary>
+        /// Record cache pressure to every monitor in this aggregated cache monitor group
+        /// </summary>
+        /// <param name="cachePressureContribution"></param>
+        public void RecordCachePressureContribution(double cachePressureContribution)
+        {
+            this.ForEach(monitor =>
+            {
+                monitor.RecordCachePressureContribution(cachePressureContribution);
+            });
+        }
+
+        /// <summary>
+        /// Add one monitor to this aggregated cache monitor group
+        /// </summary>
+        /// <param name="monitor"></param>
+        public void AddCachePressureMonitor(ICachePressureMonitor monitor)
+        {
+            this.Add(monitor);
+        }
+
+        /// <summary>
+        /// If any mornitor in this aggregated cache monitor group is under pressure, then return true
+        /// </summary>
+        /// <param name="utcNow"></param>
+        /// <returns></returns>
+        public bool IsUnderPressure(DateTime utcNow)
+        {
+            bool underPressure = this.Any(monitor => monitor.IsUnderPressure(utcNow));
+            if (this.isUnderPressure != underPressure)
+            {
+                this.isUnderPressure = underPressure;
+                logger.Info(this.isUnderPressure
+                    ? $"Ingesting messages too fast. Throttling message reading."
+                    : $"Message ingestion is healthy.");
+            }
+            return underPressure;
+        }
+    }
+}

src/OrleansServiceBus/Providers/Streams/EventHub/CachePressureMonitors/AveragingCachePressureMonitor.cs

@@ -0,0 +1,93 @@
+using Orleans.Runtime;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace Orleans.ServiceBus.Providers
+{
+    /// <summary>
+    /// Cache pressure monitor whose back pressure algorithm is based on averaging pressure value
+    /// over all pressure contribution
+    /// </summary>
+    public class AveragingCachePressureMonitor : ICachePressureMonitor
+    {
+        /// <summary>
+        /// Default flow control threshold
+        /// </summary>
+        public static readonly double DefaultThreshold = 1 / 3;
+        private static readonly TimeSpan checkPeriod = TimeSpan.FromSeconds(2);
+        private readonly Logger logger;
+
+        private double accumulatedCachePressure;
+        private double cachePressureContributionCount;
+        private DateTime nextCheckedTime;
+        private bool isUnderPressure;
+        private double flowControlThreshold;
+
+        /// <summary>
+        /// Constructor
+        /// </summary>
+        /// <param name="logger"></param>
+        public AveragingCachePressureMonitor(Logger logger)
+            :this(DefaultThreshold, logger)
+        { }
+
+        /// <summary>
+        /// Contructor
+        /// </summary>
+        /// <param name="flowControlThreshold"></param>
+        /// <param name="logger"></param>
+        public AveragingCachePressureMonitor(double flowControlThreshold, Logger logger)
+        {
+            this.flowControlThreshold = flowControlThreshold;
+            this.logger = logger.GetSubLogger(this.GetType().Name);
+            nextCheckedTime = DateTime.MinValue;
+            isUnderPressure = false;
+        }
+
+        public void RecordCachePressureContribution(double cachePressureContribution)
+        {
+            // Weight unhealthy contributions thrice as much as healthy ones.
+            // This is a crude compensation for the fact that healthy consumers wil consume more often than unhealthy ones.
+            double weight = cachePressureContribution < flowControlThreshold ? 1.0 : 3.0;
+            accumulatedCachePressure += cachePressureContribution * weight;
+            cachePressureContributionCount += weight;
+        }
+
+        public bool IsUnderPressure(DateTime utcNow)
+        {
+            if (nextCheckedTime < utcNow)
+            {
+                CalculatePressure();
+                nextCheckedTime = utcNow + checkPeriod;
+            }
+            return isUnderPressure;
+        }
+
+        private void CalculatePressure()
+        {
+            // if we don't have any contributions, don't change status
+            if (cachePressureContributionCount < 0.5)
+            {
+                // after 5 checks with no contributions, check anyway
+                cachePressureContributionCount += 0.1;
+                return;
+            }
+
+            double pressure = accumulatedCachePressure / cachePressureContributionCount;
+            bool wasUnderPressure = isUnderPressure;
+            isUnderPressure = pressure > flowControlThreshold;
+            // If we changed state, log
+            if (isUnderPressure != wasUnderPressure)
+            {
+                logger.Verbose(isUnderPressure
+                    ? $"Ingesting messages too fast. Throttling message reading. AccumulatedCachePressure: {accumulatedCachePressure}, Contributions: {cachePressureContributionCount}, AverageCachePressure: {pressure}, Threshold: {flowControlThreshold}"
+                    : $"Message ingestion is healthy. AccumulatedCachePressure: {accumulatedCachePressure}, Contributions: {cachePressureContributionCount}, AverageCachePressure: {pressure}, Threshold: {flowControlThreshold}");
+            }
+            cachePressureContributionCount = 0.0;
+            accumulatedCachePressure = 0.0;
+        }
+    }
+}

src/OrleansServiceBus/Providers/Streams/EventHub/CachePressureMonitors/ICachePressureMonitor.cs

@@ -0,0 +1,28 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace Orleans.ServiceBus.Providers
+{
+    /// <summary>
+    /// Cache pressure monitor records pressure contribution to the cache, and determine if the cache is under pressure based on its 
+    /// back pressure algorithm
+    /// </summary>
+    public interface ICachePressureMonitor
+    {
+        /// <summary>
+        /// Record cache pressure contribution to the monitor
+        /// </summary>
+        /// <param name="cachePressureContribution"></param>
+        void RecordCachePressureContribution(double cachePressureContribution);
+
+        /// <summary>
+        /// Determine if the monitor is under pressure
+        /// </summary>
+        /// <param name="utcNow"></param>
+        /// <returns></returns>
+        bool IsUnderPressure(DateTime utcNow);
+    }
+}

src/OrleansServiceBus/Providers/Streams/EventHub/CachePressureMonitors/SlowConsumingPressureMonitor.cs

@@ -0,0 +1,114 @@
+using Orleans.Runtime;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace Orleans.ServiceBus.Providers
+{
+    /// <summary>
+    /// Pressure monitor which is in favor of the slow consumer in the cache
+    /// </summary>
+    public class SlowConsumingPressureMonitor : ICachePressureMonitor
+    {
+        /// <summary>
+        /// DefaultPressureWindowSize
+        /// </summary>
+        public static TimeSpan DefaultPressureWindowSize = TimeSpan.FromMinutes(1);
+        private const double DefaultFlowControlThreshold = 0.5;
+
+        /// <summary>
+        /// PressureWindowSize
+        /// </summary>
+        public TimeSpan PressureWindowSize { get; set; }
+        /// <summary>
+        /// FlowControlThreshold
+        /// </summary>
+        public double FlowControlThreshold { get; set; }
+
+        private readonly Logger logger;
+        private double biggestPressureInCurrentWindow;
+        private DateTime nextCheckedTime;
+        private bool wasUnderPressure;
+
+        /// <summary>
+        /// Constructor
+        /// </summary>
+        /// <param name="logger"></param>
+        public SlowConsumingPressureMonitor(Logger logger)
+            : this(DefaultFlowControlThreshold, DefaultPressureWindowSize, logger)
+        { }
+
+        /// <summary>
+        /// Constructor
+        /// </summary>
+        /// <param name="pressureWindowSize"></param>
+        /// <param name="logger"></param>
+        public SlowConsumingPressureMonitor(TimeSpan pressureWindowSize, Logger logger)
+            : this(DefaultFlowControlThreshold, pressureWindowSize, logger)
+        {
+        }
+
+        /// <summary>
+        /// Constructor
+        /// </summary>
+        /// <param name="flowControlThreshold"></param>
+        /// <param name="logger"></param>
+        public SlowConsumingPressureMonitor(double flowControlThreshold, Logger logger)
+            : this(flowControlThreshold, DefaultPressureWindowSize, logger)
+        {
+        }
+
+        /// <summary>
+        /// Constructor
+        /// </summary>
+        /// <param name="flowControlThreshold"></param>
+        /// <param name="pressureWindowSzie"></param>
+        /// <param name="logger"></param>
+        public SlowConsumingPressureMonitor(double flowControlThreshold, TimeSpan pressureWindowSzie, Logger logger)
+        {
+            this.FlowControlThreshold = flowControlThreshold;
+            this.logger = logger.GetSubLogger(this.GetType().Name);
+            this.nextCheckedTime = DateTime.MinValue;
+            this.biggestPressureInCurrentWindow = 0;
+            this.wasUnderPressure = false;
+            this.PressureWindowSize = pressureWindowSzie;
+        }
+
+        public void RecordCachePressureContribution(double cachePressureContribution)
+        {
+            if (cachePressureContribution > this.biggestPressureInCurrentWindow)
+                biggestPressureInCurrentWindow = cachePressureContribution;
+        }
+
+        public bool IsUnderPressure(DateTime utcNow)
+        {
+            //if any pressure contribution in current period is bigger than flowControlThreshold
+            //we see the cache is under pressure
+            bool underPressure = this.biggestPressureInCurrentWindow > this.FlowControlThreshold;
+
+            if (underPressure && !this.wasUnderPressure)
+            {
+                //if under pressure, extend the nextCheckedTime, make sure wasUnderPressure is true for a whole window  
+                this.wasUnderPressure = underPressure;
+                this.nextCheckedTime = utcNow + this.PressureWindowSize;
+                logger.Verbose($"Ingesting messages too fast. Throttling message reading. BiggestPressureInCurrentPeriod: {biggestPressureInCurrentWindow}, Threshold: {FlowControlThreshold}");
+                this.biggestPressureInCurrentWindow = 0;
+            }
+
+            if (this.nextCheckedTime < utcNow)
+            {
+                //at the end of each check period, reset biggestPressureInCurrentPeriod
+                this.nextCheckedTime = utcNow + this.PressureWindowSize;
+                this.biggestPressureInCurrentWindow = 0;
+                //if at the end of the window, pressure clears out, log
+                if(this.wasUnderPressure && !underPressure)
+                    logger.Verbose($"Message ingestion is healthy. BiggestPressureInCurrentPeriod: {biggestPressureInCurrentWindow}, Threshold: {FlowControlThreshold}");
+                this.wasUnderPressure = underPressure;
+            }
+
+            return this.wasUnderPressure;
+        }
+    }
+}

src/OrleansServiceBus/Providers/Streams/EventHub/EventHubAdapterFactory.cs

@@ -143,7 +143,10 @@ public virtual void Init(IProviderConfiguration providerCfg, string providerName
             {
                 var bufferPool = new FixedSizeObjectPool<FixedSizeBuffer>(adapterSettings.CacheSizeMb, () => new FixedSizeBuffer(1 << 20));
                 var timePurge = new TimePurgePredicate(adapterSettings.DataMinTimeInCache, adapterSettings.DataMaxAgeInCache);
-                CacheFactory = (partition,checkpointer,cacheLogger) => new EventHubQueueCache(checkpointer, bufferPool, timePurge, cacheLogger, this.SerializationManager);
+                CacheFactory = (partition, checkpointer, cacheLogger) =>
+                {
+                    return CreateCacheFactory(partition, checkpointer, cacheLogger, bufferPool, timePurge);
+                };
             }
 
             if (StreamFailureHandlerFactory == null)
@@ -257,6 +260,29 @@ private EventHubAdapterReceiver GetOrCreateReceiver(QueueId queueId)
             return receivers.GetOrAdd(queueId, q => MakeReceiver(queueId));
         }
 
+        private IEventHubQueueCache CreateCacheFactory(string partition, IStreamQueueCheckpointer<string> checkpointer, Logger cacheLogger,
+            FixedSizeObjectPool<FixedSizeBuffer> bufferPool, TimePurgePredicate timePurge)
+        {
+            var cache = new EventHubQueueCache(checkpointer, bufferPool, timePurge, cacheLogger, this.SerializationManager);
+            if (adapterSettings.AveragingCachePressureMonitorFlowControlThreshold.HasValue)
+            {
+                var avgMonitor = new AveragingCachePressureMonitor(adapterSettings.AveragingCachePressureMonitorFlowControlThreshold.Value, cacheLogger);
+cache.AddCachePressureMonitor(avgMonitor);
+            }
+            if (adapterSettings.SlowConsumingMonitorPressureWindowSize.HasValue
+            || adapterSettings.SlowConsumingMonitorFlowControlThreshold.HasValue)
+            {
+
+                var slowConsumeMonitor = new SlowConsumingPressureMonitor(cacheLogger);
+                if (adapterSettings.SlowConsumingMonitorFlowControlThreshold.HasValue)
+                    slowConsumeMonitor.FlowControlThreshold = adapterSettings.SlowConsumingMonitorFlowControlThreshold.Value;
+                if (adapterSettings.SlowConsumingMonitorPressureWindowSize.HasValue)
+                    slowConsumeMonitor.PressureWindowSize = adapterSettings.SlowConsumingMonitorPressureWindowSize.Value;
+cache.AddCachePressureMonitor(slowConsumeMonitor);
+            }
+            return cache;
+        }
+
         private EventHubAdapterReceiver MakeReceiver(QueueId queueId)
         {
             var config = new EventHubPartitionSettings