performance tweaks for event bursts

danielgerlag · danielgerlag · commit 0b50ab7956ef · 2020-08-08T19:57:02.000-07:00
diff --git a/src/WorkflowCore/Interface/IGreyList.cs b/src/WorkflowCore/Interface/IGreyList.cs
@@ -0,0 +1,9 @@
+﻿namespace WorkflowCore.Interface
+{
+    public interface IGreyList
+    {
+        void Add(string id);
+        void Remove(string id);
+        bool Contains(string id);
+    }
+}
diff --git a/src/WorkflowCore/ServiceCollectionExtensions.cs b/src/WorkflowCore/ServiceCollectionExtensions.cs
@@ -48,6 +48,7 @@ public static IServiceCollection AddWorkflow(this IServiceCollection services, A
             services.AddTransient<IWorkflowErrorHandler, TerminateHandler>();
             services.AddTransient<IWorkflowErrorHandler, SuspendHandler>();
 
+            services.AddSingleton<IGreyList, GreyList>();
             services.AddSingleton<IWorkflowController, WorkflowController>();
             services.AddSingleton<IActivityController, ActivityController>();
             services.AddSingleton<IWorkflowHost, WorkflowHost>();
diff --git a/src/WorkflowCore/Services/BackgroundTasks/EventConsumer.cs b/src/WorkflowCore/Services/BackgroundTasks/EventConsumer.cs
@@ -16,13 +16,15 @@ internal class EventConsumer : QueueConsumer, IBackgroundTask
         private readonly IEventRepository _eventRepository;
         private readonly IDistributedLockProvider _lockProvider;
         private readonly IDateTimeProvider _datetimeProvider;
+        private readonly IGreyList _greylist;
         protected override int MaxConcurrentItems => 2;
         protected override QueueType Queue => QueueType.Event;
 
-        public EventConsumer(IWorkflowRepository workflowRepository, ISubscriptionRepository subscriptionRepository, IEventRepository eventRepository, IQueueProvider queueProvider, ILoggerFactory loggerFactory, IServiceProvider serviceProvider, IWorkflowRegistry registry, IDistributedLockProvider lockProvider, WorkflowOptions options, IDateTimeProvider datetimeProvider)
+        public EventConsumer(IWorkflowRepository workflowRepository, ISubscriptionRepository subscriptionRepository, IEventRepository eventRepository, IQueueProvider queueProvider, ILoggerFactory loggerFactory, IServiceProvider serviceProvider, IWorkflowRegistry registry, IDistributedLockProvider lockProvider, WorkflowOptions options, IDateTimeProvider datetimeProvider, IGreyList greylist)
             : base(queueProvider, loggerFactory, options)
         {
             _workflowRepository = workflowRepository;
+            _greylist = greylist;
             _subscriptionRepository = subscriptionRepository;
             _eventRepository = eventRepository;
             _lockProvider = lockProvider;
@@ -41,6 +43,11 @@ protected override async Task ProcessItem(string itemId, CancellationToken cance
             {
                 cancellationToken.ThrowIfCancellationRequested();
                 var evt = await _eventRepository.GetEvent(itemId);
+                if (evt.IsProcessed)
+                {
+                    _greylist.Add($"evt:{evt.Id}");
+                    return;
+                }
                 if (evt.EventTime <= _datetimeProvider.UtcNow)
                 {
                     IEnumerable<EventSubscription> subs = null;
@@ -60,7 +67,7 @@ protected override async Task ProcessItem(string itemId, CancellationToken cance
                         subs = await _subscriptionRepository.GetSubscriptions(evt.EventName, evt.EventKey, evt.EventTime);
                     }
 
-                    var toQueue = new List<string>();
+                    var toQueue = new HashSet<string>();
                     var complete = true;
 
                     foreach (var sub in subs.ToList())
@@ -79,7 +86,7 @@ protected override async Task ProcessItem(string itemId, CancellationToken cance
             }
         }
         
-        private async Task<bool> SeedSubscription(Event evt, EventSubscription sub, List<string> toQueue, CancellationToken cancellationToken)
+        private async Task<bool> SeedSubscription(Event evt, EventSubscription sub, HashSet<string> toQueue, CancellationToken cancellationToken)
         {            
             foreach (var eventId in await _eventRepository.GetEvents(sub.EventName, sub.EventKey, sub.SubscribeAsOf))
             {
diff --git a/src/WorkflowCore/Services/BackgroundTasks/RunnablePoller.cs b/src/WorkflowCore/Services/BackgroundTasks/RunnablePoller.cs
@@ -13,12 +13,14 @@ internal class RunnablePoller : IBackgroundTask
         private readonly IDistributedLockProvider _lockProvider;
         private readonly IQueueProvider _queueProvider;
         private readonly ILogger _logger;
+        private readonly IGreyList _greylist;
         private readonly WorkflowOptions _options;
         private Timer _pollTimer;
 
-        public RunnablePoller(IPersistenceProvider persistenceStore, IQueueProvider queueProvider, ILoggerFactory loggerFactory, IServiceProvider serviceProvider, IWorkflowRegistry registry, IDistributedLockProvider lockProvider, WorkflowOptions options)
+        public RunnablePoller(IPersistenceProvider persistenceStore, IQueueProvider queueProvider, ILoggerFactory loggerFactory, IServiceProvider serviceProvider, IWorkflowRegistry registry, IDistributedLockProvider lockProvider, IGreyList greylist, WorkflowOptions options)
         {
             _persistenceStore = persistenceStore;
+            _greylist = greylist;
             _queueProvider = queueProvider;            
             _logger = loggerFactory.CreateLogger<RunnablePoller>();
             _lockProvider = lockProvider;
@@ -55,7 +57,13 @@ private async void PollRunnables(object target)
                         var runnables = await _persistenceStore.GetRunnableInstances(DateTime.Now);
                         foreach (var item in runnables)
                         {
+                            if (_greylist.Contains($"wf:{item}"))
+                            {
+                                _logger.LogDebug($"Got greylisted workflow {item}");
+                                continue;
+                            }
                             _logger.LogDebug("Got runnable instance {0}", item);
+                            _greylist.Add($"wf:{item}");
                             await _queueProvider.QueueWork(item, QueueType.Workflow);
                         }
                     }
@@ -80,8 +88,15 @@ private async void PollRunnables(object target)
                         var events = await _persistenceStore.GetRunnableEvents(DateTime.Now);
                         foreach (var item in events.ToList())
                         {
+                            if (_greylist.Contains($"evt:{item}"))
+                            {
+                                _logger.LogDebug($"Got greylisted event {item}");
+                                _greylist.Add($"evt:{item}");
+                                continue;
+                            }
                             _logger.LogDebug($"Got unprocessed event {item}");
-                            await _queueProvider.QueueWork(item, QueueType.Event);                            
+                            _greylist.Add($"evt:{item}");
+                            await _queueProvider.QueueWork(item, QueueType.Event);
                         }
                     }
                     finally
diff --git a/src/WorkflowCore/Services/BackgroundTasks/WorkflowConsumer.cs b/src/WorkflowCore/Services/BackgroundTasks/WorkflowConsumer.cs
@@ -14,14 +14,16 @@ internal class WorkflowConsumer : QueueConsumer, IBackgroundTask
         private readonly IDateTimeProvider _datetimeProvider;
         private readonly IPersistenceProvider _persistenceStore;
         private readonly IWorkflowExecutor _executor;
+        private readonly IGreyList _greylist;
 
         protected override int MaxConcurrentItems => Options.MaxConcurrentWorkflows;
         protected override QueueType Queue => QueueType.Workflow;
 
-        public WorkflowConsumer(IPersistenceProvider persistenceProvider, IQueueProvider queueProvider, ILoggerFactory loggerFactory, IServiceProvider serviceProvider, IWorkflowRegistry registry, IDistributedLockProvider lockProvider, IWorkflowExecutor executor, IDateTimeProvider datetimeProvider, WorkflowOptions options)
+        public WorkflowConsumer(IPersistenceProvider persistenceProvider, IQueueProvider queueProvider, ILoggerFactory loggerFactory, IServiceProvider serviceProvider, IWorkflowRegistry registry, IDistributedLockProvider lockProvider, IWorkflowExecutor executor, IDateTimeProvider datetimeProvider, IGreyList greylist, WorkflowOptions options)
             : base(queueProvider, loggerFactory, options)
         {
             _persistenceStore = persistenceProvider;
+            _greylist = greylist;
             _executor = executor;
             _lockProvider = lockProvider;
             _datetimeProvider = datetimeProvider;
@@ -43,7 +45,7 @@ protected override async Task ProcessItem(string itemId, CancellationToken cance
                 cancellationToken.ThrowIfCancellationRequested();
                 workflow = await _persistenceStore.GetWorkflowInstance(itemId);
                 if (workflow.Status == WorkflowStatus.Runnable)
-                {                        
+                {
                     try
                     {
                         result = await _executor.Execute(workflow);
@@ -52,6 +54,7 @@ protected override async Task ProcessItem(string itemId, CancellationToken cance
                     {
                         await _persistenceStore.PersistWorkflow(workflow);
                         await QueueProvider.QueueWork(itemId, QueueType.Index);
+                        _greylist.Remove($"wf:{itemId}");
                     }
                 }
             }
diff --git a/src/WorkflowCore/Services/GreyList.cs b/src/WorkflowCore/Services/GreyList.cs
@@ -0,0 +1,63 @@
+﻿using Microsoft.Extensions.Logging;
+using System;
+using System.Collections.Concurrent;
+using System.Threading;
+using WorkflowCore.Interface;
+
+namespace WorkflowCore.Services
+{
+    public class GreyList : IGreyList, IDisposable
+    {
+        private readonly Timer _cycleTimer;
+        private readonly ConcurrentDictionary<string, DateTime> _list;
+        private readonly ILogger _logger;
+        private const int CYCLE_TIME = 600;
+
+        public GreyList(ILoggerFactory loggerFactory)
+        {
+            _logger = loggerFactory.CreateLogger<GreyList>();
+            _list = new ConcurrentDictionary<string, DateTime>();
+            _cycleTimer = new Timer(new TimerCallback(Cycle), null, TimeSpan.FromSeconds(CYCLE_TIME), TimeSpan.FromSeconds(CYCLE_TIME));
+        }
+
+        public void Add(string id)
+        {
+            _list.AddOrUpdate(id, DateTime.Now, (key, val) => DateTime.Now);
+        }
+
+        public bool Contains(string id)
+        {
+            if (!_list.TryGetValue(id, out var start))
+                return false;
+
+            var result = start > (DateTime.Now.AddMinutes(-2));
+
+            if (!result)
+                _list.TryRemove(id, out var _);
+
+            return result;
+        }
+
+        private void Cycle(object target)
+        {
+            try
+            {
+                _list.Clear();
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError(ex, ex.Message);
+            }
+        }
+
+        public void Dispose()
+        {
+            _cycleTimer.Dispose();
+        }
+
+        public void Remove(string id)
+        {
+            _list.TryRemove(id, out var _);
+        }
+    }
+}
diff --git a/src/WorkflowCore/WorkflowCore.csproj b/src/WorkflowCore/WorkflowCore.csproj
@@ -15,12 +15,12 @@
     <GenerateAssemblyCompanyAttribute>false</GenerateAssemblyCompanyAttribute>
     <GenerateAssemblyProductAttribute>false</GenerateAssemblyProductAttribute>
     <Description>Workflow Core is a light weight workflow engine targeting .NET Standard.</Description>
-    <Version>3.2.0</Version>
-    <AssemblyVersion>3.2.0.0</AssemblyVersion>
-    <FileVersion>3.2.0.0</FileVersion>
+    <Version>3.2.1</Version>
+    <AssemblyVersion>3.2.1.0</AssemblyVersion>
+    <FileVersion>3.2.1.0</FileVersion>
     <PackageReleaseNotes></PackageReleaseNotes>
     <PackageIconUrl>https://github.com/danielgerlag/workflow-core/raw/master/src/logo.png</PackageIconUrl>
-    <PackageVersion>3.2.0</PackageVersion>
+    <PackageVersion>3.2.1</PackageVersion>
   </PropertyGroup>
 
   <ItemGroup>

Original file line number	Diff line number	Diff line change
`@@ -16,13 +16,15 @@ internal class EventConsumer : QueueConsumer, IBackgroundTask`
`16`	`16`	`private readonly IEventRepository _eventRepository;`
`17`	`17`	`private readonly IDistributedLockProvider _lockProvider;`
`18`	`18`	`private readonly IDateTimeProvider _datetimeProvider;`
	`19`	`+ private readonly IGreyList _greylist;`
`19`	`20`	`protected override int MaxConcurrentItems => 2;`
`20`	`21`	`protected override QueueType Queue => QueueType.Event;`
`21`	`22`
`22`		`- public EventConsumer(IWorkflowRepository workflowRepository, ISubscriptionRepository subscriptionRepository, IEventRepository eventRepository, IQueueProvider queueProvider, ILoggerFactory loggerFactory, IServiceProvider serviceProvider, IWorkflowRegistry registry, IDistributedLockProvider lockProvider, WorkflowOptions options, IDateTimeProvider datetimeProvider)`
	`23`	`+ public EventConsumer(IWorkflowRepository workflowRepository, ISubscriptionRepository subscriptionRepository, IEventRepository eventRepository, IQueueProvider queueProvider, ILoggerFactory loggerFactory, IServiceProvider serviceProvider, IWorkflowRegistry registry, IDistributedLockProvider lockProvider, WorkflowOptions options, IDateTimeProvider datetimeProvider, IGreyList greylist)`
`23`	`24`	`: base(queueProvider, loggerFactory, options)`
`24`	`25`	`{`
`25`	`26`	`_workflowRepository = workflowRepository;`
	`27`	`+ _greylist = greylist;`
`26`	`28`	`_subscriptionRepository = subscriptionRepository;`
`27`	`29`	`_eventRepository = eventRepository;`
`28`	`30`	`_lockProvider = lockProvider;`
`@@ -41,6 +43,11 @@ protected override async Task ProcessItem(string itemId, CancellationToken cance`
`41`	`43`	`{`
`42`	`44`	`cancellationToken.ThrowIfCancellationRequested();`
`43`	`45`	`var evt = await _eventRepository.GetEvent(itemId);`
	`46`	`+ if (evt.IsProcessed)`
	`47`	`+ {`
	`48`	`+ _greylist.Add($"evt:{evt.Id}");`
	`49`	`+ return;`
	`50`	`+ }`
`44`	`51`	`if (evt.EventTime <= _datetimeProvider.UtcNow)`
`45`	`52`	`{`
`46`	`53`	`IEnumerable<EventSubscription> subs = null;`
`@@ -60,7 +67,7 @@ protected override async Task ProcessItem(string itemId, CancellationToken cance`
`60`	`67`	`subs = await _subscriptionRepository.GetSubscriptions(evt.EventName, evt.EventKey, evt.EventTime);`
`61`	`68`	`}`
`62`	`69`
`63`		`- var toQueue = new List<string>();`
	`70`	`+ var toQueue = new HashSet<string>();`
`64`	`71`	`var complete = true;`
`65`	`72`
`66`	`73`	`foreach (var sub in subs.ToList())`
`@@ -79,7 +86,7 @@ protected override async Task ProcessItem(string itemId, CancellationToken cance`
`79`	`86`	`}`
`80`	`87`	`}`
`81`	`88`
`82`		`- private async Task<bool> SeedSubscription(Event evt, EventSubscription sub, List<string> toQueue, CancellationToken cancellationToken)`
	`89`	`+ private async Task<bool> SeedSubscription(Event evt, EventSubscription sub, HashSet<string> toQueue, CancellationToken cancellationToken)`
`83`	`90`	`{`
`84`	`91`	`foreach (var eventId in await _eventRepository.GetEvents(sub.EventName, sub.EventKey, sub.SubscribeAsOf))`
`85`	`92`	`{`
Original file line number	Diff line number	Diff line change
`@@ -14,14 +14,16 @@ internal class WorkflowConsumer : QueueConsumer, IBackgroundTask`
`14`	`14`	`private readonly IDateTimeProvider _datetimeProvider;`
`15`	`15`	`private readonly IPersistenceProvider _persistenceStore;`
`16`	`16`	`private readonly IWorkflowExecutor _executor;`
	`17`	`+ private readonly IGreyList _greylist;`
`17`	`18`
`18`	`19`	`protected override int MaxConcurrentItems => Options.MaxConcurrentWorkflows;`
`19`	`20`	`protected override QueueType Queue => QueueType.Workflow;`
`20`	`21`
`21`		`- public WorkflowConsumer(IPersistenceProvider persistenceProvider, IQueueProvider queueProvider, ILoggerFactory loggerFactory, IServiceProvider serviceProvider, IWorkflowRegistry registry, IDistributedLockProvider lockProvider, IWorkflowExecutor executor, IDateTimeProvider datetimeProvider, WorkflowOptions options)`
	`22`	`+ public WorkflowConsumer(IPersistenceProvider persistenceProvider, IQueueProvider queueProvider, ILoggerFactory loggerFactory, IServiceProvider serviceProvider, IWorkflowRegistry registry, IDistributedLockProvider lockProvider, IWorkflowExecutor executor, IDateTimeProvider datetimeProvider, IGreyList greylist, WorkflowOptions options)`
`22`	`23`	`: base(queueProvider, loggerFactory, options)`
`23`	`24`	`{`
`24`	`25`	`_persistenceStore = persistenceProvider;`
	`26`	`+ _greylist = greylist;`
`25`	`27`	`_executor = executor;`
`26`	`28`	`_lockProvider = lockProvider;`
`27`	`29`	`_datetimeProvider = datetimeProvider;`
`@@ -43,7 +45,7 @@ protected override async Task ProcessItem(string itemId, CancellationToken cance`
`43`	`45`	`cancellationToken.ThrowIfCancellationRequested();`
`44`	`46`	`workflow = await _persistenceStore.GetWorkflowInstance(itemId);`
`45`	`47`	`if (workflow.Status == WorkflowStatus.Runnable)`
`46`		`- {`
	`48`	`+ {`
`47`	`49`	`try`
`48`	`50`	`{`
`49`	`51`	`result = await _executor.Execute(workflow);`
`@@ -52,6 +54,7 @@ protected override async Task ProcessItem(string itemId, CancellationToken cance`
`52`	`54`	`{`
`53`	`55`	`await _persistenceStore.PersistWorkflow(workflow);`
`54`	`56`	`await QueueProvider.QueueWork(itemId, QueueType.Index);`
	`57`	`+ _greylist.Remove($"wf:{itemId}");`
`55`	`58`	`}`
`56`	`59`	`}`
`57`	`60`	`}`