TPL dataflow action block

Author: danielgerlag

src/WorkflowCore/Services/QueueTaskDispatcher.cs

@@ -1,9 +1,9 @@
 using Microsoft.Extensions.Logging;
 using System;
 using System.Collections.Generic;
-using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
+using System.Threading.Tasks.Dataflow;
 using WorkflowCore.Interface;
 using WorkflowCore.Models;
 
@@ -17,8 +17,7 @@ public abstract class QueueTaskDispatcher : IBackgroundTask
         protected readonly IQueueProvider QueueProvider;
         protected readonly ILogger Logger;
         protected readonly WorkflowOptions Options;
-        protected Task DispatchTask;
-        private SemaphoreSlim _semaphore;
+        protected Task DispatchTask;        
         private CancellationTokenSource _cancellationTokenSource;
 
         protected QueueTaskDispatcher(IQueueProvider queueProvider, ILoggerFactory loggerFactory, WorkflowOptions options)
@@ -36,7 +35,7 @@ public virtual void Start()
                 throw new InvalidOperationException();
 
             _cancellationTokenSource = new CancellationTokenSource();
-            _semaphore = new SemaphoreSlim(MaxConcurrentItems);
+                        
             DispatchTask = new Task(Execute);
             DispatchTask.Start();
         }
@@ -45,41 +44,38 @@ public virtual void Stop()
         {
             _cancellationTokenSource.Cancel();
             DispatchTask.Wait();
-
-            for (var i = 0; i < MaxConcurrentItems; i++)
-                _semaphore.Wait();
-
             DispatchTask = null;
         }
 
         private async void Execute()
         {
             var cancelToken = _cancellationTokenSource.Token;
+            var opts = new ExecutionDataflowBlockOptions()
+            {
+                MaxDegreeOfParallelism = MaxConcurrentItems,
+                BoundedCapacity = MaxConcurrentItems + 1
+            };
+
+            var actionBlock = new ActionBlock<string>(ExecuteItem, opts);
+
             while (!cancelToken.IsCancellationRequested)
             {
                 try
                 {
-                    await _semaphore.WaitAsync(cancelToken);
-                    string item;
-                    try
-                    {
-                        item = await QueueProvider.DequeueWork(Queue, cancelToken);
-                    }
-                    catch
+                    if (SpinWait.SpinUntil(() => actionBlock.InputCount == 0, Options.IdleTime))
                     {
-                        _semaphore.Release();
-                        throw;
-                    }
+                        var item = await QueueProvider.DequeueWork(Queue, cancelToken);
 
-                    if (item == null)
-                    {
-                        _semaphore.Release();
-                        if (!QueueProvider.IsDequeueBlocking)
-                            await Task.Delay(Options.IdleTime, cancelToken);
-                        continue;
-                    }
+                        if (item == null)
+                        {
+                            if (!QueueProvider.IsDequeueBlocking)
+                                await Task.Delay(Options.IdleTime, cancelToken);
+                            continue;
+                        }
 
-                    new Task(() => ExecuteItem(item, cancelToken)).Start();
+                        if (!actionBlock.Post(item))
+                            await QueueProvider.QueueWork(item, Queue);
+                    }
                 }
                 catch (OperationCanceledException)
                 {
@@ -89,13 +85,15 @@ private async void Execute()
                     Logger.LogError(ex.Message);
                 }
             }
+            actionBlock.Complete();
+            await actionBlock.Completion;
         }
 
-        private async void ExecuteItem(string itemId, CancellationToken cancellationToken)
+        private async Task ExecuteItem(string itemId)
         {
             try
             {
-                await ProcessItem(itemId, cancellationToken);
+                await ProcessItem(itemId, _cancellationTokenSource.Token);
             }
             catch (OperationCanceledException)
             {
@@ -105,10 +103,6 @@ private async void ExecuteItem(string itemId, CancellationToken cancellationToke
             {
                 Logger.LogError($"Error executing item {itemId} - {ex.Message}");
             }
-            finally
-            {
-                _semaphore.Release();
-            }
         }
     }
 }

src/WorkflowCore/WorkflowCore.csproj

@@ -27,6 +27,7 @@
   <ItemGroup>
     <PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="1.1.0" />
     <PackageReference Include="Microsoft.Extensions.Logging" Version="1.1.1" />
+    <PackageReference Include="System.Threading.Tasks.Dataflow" Version="4.7.0" />
     <PackageReference Include="System.Threading.Tasks.Parallel" Version="4.3.0" />
   </ItemGroup>
 

src/samples/WorkflowCore.Sample03/Steps/AddNumbers.cs

@@ -19,8 +19,8 @@ public class AddNumbers : StepBodyAsync
         public override async Task<ExecutionResult> RunAsync(IStepExecutionContext context)
         {
             if (Input1 == 2)
-                //await Task.Delay(2000);
-                System.Threading.Thread.Sleep(2000);
+                await Task.Delay(2000);
+                //System.Threading.Thread.Sleep(2000);
 
             Output = (Input1 + Input2);
             return ExecutionResult.Next();