Merge pull request #50 from danielgerlag/read-ahead

ReleaseNotes/1.3.0.md

@@ -0,0 +1,20 @@
+# Workflow Core 1.3.0
+
+* Added support for async steps
+
+Simply inherit from `StepBodyAsync` instead of `StepBody`
+
+```c#
+public class DoSomething : StepBodyAsync
+{
+    public override async Task<ExecutionResult> RunAsync(IStepExecutionContext context)
+    {
+        await Task.Delay(2000);
+        return ExecutionResult.Next();
+    }
+}
+```
+
+* Migrated from managing own thread pool to TPL datablocks for queue consumers
+
+* After executing a workflow, will determine if it is scheduled to run before the next poll, if so, will delay queue it

src/WorkflowCore/Interface/IBackgroundWorker.cs → src/WorkflowCore/Interface/IBackgroundTask.cs

@@ -1,7 +1,7 @@
 
 namespace WorkflowCore.Interface
 {
-    public interface IBackgroundWorker
+    public interface IBackgroundTask
     {
         void Start();
         void Stop();

src/WorkflowCore/Interface/IDateTimeProvider.cs

@@ -0,0 +1,9 @@
+using System;
+
+namespace WorkflowCore.Interface
+{
+    public interface IDateTimeProvider
+    {
+        DateTime Now { get; }
+    }
+}

src/WorkflowCore/Interface/IDistributedLockProvider.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Threading;
 using System.Threading.Tasks;
 
 namespace WorkflowCore.Interface
@@ -11,7 +12,7 @@ namespace WorkflowCore.Interface
     /// </remarks>
     public interface IDistributedLockProvider
     {
-        Task<bool> AcquireLock(string Id);
+        Task<bool> AcquireLock(string Id, CancellationToken cancellationToken);
 
         Task ReleaseLock(string Id);
 

src/WorkflowCore/Interface/IPersistenceProvider.cs

@@ -16,7 +16,7 @@ public interface IPersistenceProvider
 
         Task PersistWorkflow(WorkflowInstance workflow);
 
-        Task<IEnumerable<string>> GetRunnableInstances();
+        Task<IEnumerable<string>> GetRunnableInstances(DateTime asAt);
 
         Task<IEnumerable<WorkflowInstance>> GetWorkflowInstances(WorkflowStatus? status, string type, DateTime? createdFrom, DateTime? createdTo, int skip, int take);
 
@@ -32,7 +32,7 @@ public interface IPersistenceProvider
 
         Task<Event> GetEvent(string id);
 
-        Task<IEnumerable<string>> GetRunnableEvents();
+        Task<IEnumerable<string>> GetRunnableEvents(DateTime asAt);
 
         Task<IEnumerable<string>> GetEvents(string eventName, string eventKey, DateTime asOf);
 

src/WorkflowCore/Interface/IQueueProvider.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Threading;
 using System.Threading.Tasks;
 using WorkflowCore.Models;
 
@@ -25,7 +26,9 @@ public interface IQueueProvider : IDisposable
         /// If the queue is empty, NULL is returned
         /// </summary>
         /// <returns></returns>
-        Task<string> DequeueWork(QueueType queue);                
+        Task<string> DequeueWork(QueueType queue, CancellationToken cancellationToken);
+
+        bool IsDequeueBlocking { get; }
 
         Task Start();
         Task Stop();

src/WorkflowCore/Interface/IStepBody.cs

@@ -8,6 +8,6 @@ namespace WorkflowCore.Interface
 {
     public interface IStepBody
     {        
-        ExecutionResult Run(IStepExecutionContext context);
+        Task<ExecutionResult> RunAsync(IStepExecutionContext context);
     }
 }

src/WorkflowCore/Interface/IWorkflowExecutor.cs

@@ -6,6 +6,6 @@ namespace WorkflowCore.Interface
 {
     public interface IWorkflowExecutor
     {
-        WorkflowExecutorResult Execute(WorkflowInstance workflow, WorkflowOptions options);
+        Task<WorkflowExecutorResult> Execute(WorkflowInstance workflow, WorkflowOptions options);
     }
 }

src/WorkflowCore/Models/StepBody.cs

@@ -8,9 +8,14 @@ namespace WorkflowCore.Models
 {
     public abstract class StepBody : IStepBody
     {
-        
+
         public abstract ExecutionResult Run(IStepExecutionContext context);
 
+        public Task<ExecutionResult> RunAsync(IStepExecutionContext context)
+        {
+            return Task.FromResult(Run(context));
+        }
+
         protected ExecutionResult OutcomeResult(object value)
         {
             return new ExecutionResult()

src/WorkflowCore/Models/StepBodyAsync.cs

@@ -0,0 +1,14 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+using WorkflowCore.Interface;
+
+namespace WorkflowCore.Models
+{
+    public abstract class StepBodyAsync : IStepBody
+    {
+        public abstract Task<ExecutionResult> RunAsync(IStepExecutionContext context);
+
+    }
+}

src/WorkflowCore/Models/WorkflowInstance.cs

@@ -16,6 +16,8 @@ public class WorkflowInstance
 
         public string Description { get; set; }
 
+        public string Reference { get; set; }
+
         public List<ExecutionPointer> ExecutionPointers { get; set; } = new List<ExecutionPointer>();
 
         public long? NextExecution { get; set; }

src/WorkflowCore/Models/WorkflowOptions.cs

@@ -12,17 +12,15 @@ public class WorkflowOptions
         internal Func<IServiceProvider, IPersistenceProvider> PersistanceFactory;
         internal Func<IServiceProvider, IQueueProvider> QueueFactory;
         internal Func<IServiceProvider, IDistributedLockProvider> LockFactory;
-        internal int ThreadCount;
         internal TimeSpan PollInterval;
         internal TimeSpan IdleTime;
         internal TimeSpan ErrorRetryInterval;        
 
         public WorkflowOptions()
         {
             //set defaults
-            ThreadCount = Environment.ProcessorCount;
             PollInterval = TimeSpan.FromSeconds(10);
-            IdleTime = TimeSpan.FromMilliseconds(500);
+            IdleTime = TimeSpan.FromMilliseconds(100);
             ErrorRetryInterval = TimeSpan.FromSeconds(60);            
 
             QueueFactory = new Func<IServiceProvider, IQueueProvider>(sp => new SingleNodeQueueProvider());
@@ -45,11 +43,6 @@ public void UseQueueProvider(Func<IServiceProvider, IQueueProvider> factory)
             QueueFactory = factory;
         }
 
-        public void UseThreads(int count)
-        {
-            ThreadCount = count;
-        }
-
         public void UsePollInterval(TimeSpan interval)
         {
             PollInterval = interval;

src/WorkflowCore/Models/WorkflowStep.cs

@@ -65,7 +65,6 @@ public virtual IStepBody ConstructBody(IServiceProvider serviceProvider)
             }
             return body;
         }
-
     }
 
     public enum ExecutionPipelineDirective { Next = 0, Defer = 1, EndWorkflow = 2 }

src/WorkflowCore/ServiceCollectionExtensions.cs

@@ -7,7 +7,9 @@
 using WorkflowCore.Services;
 using WorkflowCore.Models;
 using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.ObjectPool;
 using WorkflowCore.Primitives;
+using WorkflowCore.Services.BackgroundTasks;
 
 namespace Microsoft.Extensions.DependencyInjection
 {
@@ -23,13 +25,19 @@ public static void AddWorkflow(this IServiceCollection services, Action<Workflow
             services.AddSingleton<IDistributedLockProvider>(options.LockFactory);
             services.AddSingleton<IWorkflowRegistry, WorkflowRegistry>();
             services.AddSingleton<WorkflowOptions>(options);
+
+            services.AddTransient<IBackgroundTask, WorkflowConsumer>();
+            services.AddTransient<IBackgroundTask, EventConsumer>();
+            services.AddTransient<IBackgroundTask, RunnablePoller>();
+
             services.AddSingleton<IWorkflowHost, WorkflowHost>();
             services.AddTransient<IWorkflowExecutor, WorkflowExecutor>();
             services.AddTransient<IWorkflowBuilder, WorkflowBuilder>();
-            services.AddTransient<IWorkflowThread, WorkflowThread>();
-            services.AddTransient<IEventThread, EventThread>();
-            services.AddTransient<IRunnablePoller, RunnablePoller>();
-
+            services.AddTransient<IDateTimeProvider, DateTimeProvider>();
+
+            services.AddTransient<IPooledObjectPolicy<IPersistenceProvider>, InjectedObjectPoolPolicy<IPersistenceProvider>>();
+            services.AddTransient<IPooledObjectPolicy<IWorkflowExecutor>, InjectedObjectPoolPolicy<IWorkflowExecutor>>();
+
             services.AddTransient<Foreach>();
         }
     }

src/WorkflowCore/Services/BackgroundTasks/EventConsumer.cs

@@ -0,0 +1,95 @@
+using System;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Logging;
+using WorkflowCore.Interface;
+using WorkflowCore.Models;
+
+namespace WorkflowCore.Services.BackgroundTasks
+{
+    internal class EventConsumer : QueueConsumer, IBackgroundTask
+    {
+        private readonly IPersistenceProvider _persistenceStore;
+        private readonly IDistributedLockProvider _lockProvider;
+        private readonly IDateTimeProvider _datetimeProvider;
+
+        protected override QueueType Queue => QueueType.Event;
+
+        public EventConsumer(IPersistenceProvider persistenceStore, IQueueProvider queueProvider, ILoggerFactory loggerFactory, IServiceProvider serviceProvider, IWorkflowRegistry registry, IDistributedLockProvider lockProvider, WorkflowOptions options, IDateTimeProvider datetimeProvider)
+            : base(queueProvider, loggerFactory, options)
+        {
+            _persistenceStore = persistenceStore;
+            _lockProvider = lockProvider;
+            _datetimeProvider = datetimeProvider;
+        }
+
+        protected override async Task ProcessItem(string itemId, CancellationToken cancellationToken)
+        {
+            if (await _lockProvider.AcquireLock($"evt:{itemId}", cancellationToken))
+            {
+                try
+                {
+                    cancellationToken.ThrowIfCancellationRequested();
+                    var evt = await _persistenceStore.GetEvent(itemId);
+                    if (evt.EventTime <= _datetimeProvider.Now.ToUniversalTime())
+                    {
+                        var subs = await _persistenceStore.GetSubcriptions(evt.EventName, evt.EventKey, evt.EventTime);
+                        var success = true;
+
+                        foreach (var sub in subs.ToList())
+                            success = success && await SeedSubscription(evt, sub, cancellationToken);
+
+                        if (success)
+                            await _persistenceStore.MarkEventProcessed(itemId);
+                    }
+                }
+                finally
+                {
+                    await _lockProvider.ReleaseLock($"evt:{itemId}");
+                }
+            }
+            else
+            {
+                Logger.LogInformation($"Event locked {itemId}");
+            }
+        }
+
+        private async Task<bool> SeedSubscription(Event evt, EventSubscription sub, CancellationToken cancellationToken)
+        {
+            if (await _lockProvider.AcquireLock(sub.WorkflowId, cancellationToken))
+            {
+                try
+                {
+                    var workflow = await _persistenceStore.GetWorkflowInstance(sub.WorkflowId);
+                    var pointers = workflow.ExecutionPointers.Where(p => p.EventName == sub.EventName && p.EventKey == sub.EventKey && !p.EventPublished);
+                    foreach (var p in pointers)
+                    {
+                        p.EventData = evt.EventData;
+                        p.EventPublished = true;
+                        p.Active = true;
+                    }
+                    workflow.NextExecution = 0;
+                    await _persistenceStore.PersistWorkflow(workflow);
+                    await _persistenceStore.TerminateSubscription(sub.Id);
+                    return true;
+                }
+                catch (Exception ex)
+                {
+                    Logger.LogError(ex.Message);
+                    return false;
+                }
+                finally
+                {
+                    await _lockProvider.ReleaseLock(sub.WorkflowId);
+                    await QueueProvider.QueueWork(sub.WorkflowId, QueueType.Workflow);
+                }
+            }
+            else
+            {
+                Logger.LogInformation("Workflow locked {0}", sub.WorkflowId);
+                return false;
+            }
+        }
+    }
+}