Removed hard coded Max Concurrent Items in Event Consumer. That signi…

docs/performance.md

@@ -0,0 +1,87 @@
+# Performance Test
+
+Workflow-core version 3.7.0 was put under test to evaluate its performance. The setup used was single node with the default MemoryPersistenceProvider persistence provider.
+
+## Methodology
+
+- Test Environment - Test were run on following two environments one after the other to see how workflow-core performance with a lower vs higher hardware configuration.
+  - Lower configuration
+    - Cores: 8 vCPU ([Standard_D8s_v3](https://learn.microsoft.com/azure/virtual-machines/dv3-dsv3-series))
+    - RAM: 32 GB
+    - OS: Linux Ubuntu 20.04
+    - dotNet 6
+  - Higher configuration
+    - Cores: 32 vCPU ([Standard_D32as_v4](https://learn.microsoft.com/azure/virtual-machines/dav4-dasv4-series))
+    - RAM: 128 GB
+    - OS: Linux Ubuntu 20.04
+    - dotNet 6
+- Test Workflow: Workflow consist of 3 basic steps. These 3 simple steps were chosen to test the performance of the workflow engine with minimal yet sufficient complexity and to avoid any external dependencies.
+  - Step1 : Generate a [random number](https://learn.microsoft.com/dotnet/api/system.random?view=net-6.0) between 1 to 10 and print it on standard output.
+  - Step2 :  [Conditional step](https://github.com/danielgerlag/workflow-core/blob/master/docs/control-structures.md)
+    - Step 2.1: If value generate in step1 is > 5 then print it on standard output.
+    - Step 2.2: If value generate in step1 is <= 5 then print it on standard output.
+  - Step3: Prints a good bye message on standard output.
+- Test tools:
+  - [NBomber](https://nbomber.com/docs/getting-started/overview/) was used as performance testing framework with C# console app as base.
+
+- Test scenarios:
+  - Each type of test run executed for 20 minutes.
+  - NBomber Load Simulation of type [KeepConstant](https://nbomber.com/docs/using-nbomber/basic-api/load-simulation#keep-constant) copies was used. This type of simulation keep a constant amount of Scenario copies(instances) for a specific period.
+  - Concurrent copies [1,2,3,4,5,6,7,8,10,12,14,16,32,64,128,256,512,1024] were tested.
+    - For example if we take Concurrent copies=4 and Duration=20 minutes this means that NBomber will ensure that we have 4 instance of Test Workflow running in parallel for 20 minutes.
+
+## Results
+
+- Workflow per seconds - Below tables shows how many workflows we are able to execute per second on two different environment with increasing number of concurrent copies.
+
+| **Concurrent Copies** | **8 vCPU** | **32 vCPU** |
+| :-------------------: | :--------: | :---------: |
+|         **1**         |   300.6    |    504.7    |
+|         **2**         |   310.3    |    513.1    |
+|         **3**         |   309.6    |    519.3    |
+|         **4**         |   314.7    |    521.3    |
+|         **5**         |   312.4    |    519.0    |
+|         **6**         |   314.7    |    517.7    |
+|         **7**         |   318.9    |    516.7    |
+|         **8**         |   318.4    |    517.5    |
+|        **10**         |   322.6    |    517.1    |
+|        **12**         |   319.7    |    517.6    |
+|        **14**         |   322.4    |    518.1    |
+|        **16**         |   327.0    |    515.5    |
+|        **32**         |   327.7    |    515.8    |
+|        **64**         |   330.7    |    523.7    |
+|        **128**        |   332.8    |    526.9    |
+|        **256**        |   332.8    |    529.1    |
+|        **512**        |   332.8    |    529.1    |
+|       **1024**        |   341.3    |    529.1    |
+
+![Workflows Per Second](./images/performance-test-workflows-per-second.png)
+
+- Latency - Shows Mean, P99 and P50 latency in milliseconds on two different environment with increasing number of concurrent copies.
+
+| **Concurrent Copies** | **Mean 8 vCPU** | **Mean 32 vCPU** | **P.99 8 vCPU** | **P.99 32 vCPU** | **P.50 8 vCPU** | **P.50 32 vCPU** |
+| :-------------------: | :-------------: | :--------------: | :-------------: | :--------------: | :-------------: | :--------------: |
+|         **1**         |      3.32       |       1.98       |      12.67      |       2.49       |      3.13       |       1.85       |
+|         **2**         |      6.43       |       3.89       |      19.96      |       5.67       |      6.17       |       3.65       |
+|         **3**         |      9.67       |       5.77       |      24.96      |       8.2        |      9.14       |       5.46       |
+|         **4**         |      12.7       |       7.76       |      27.44      |      13.57       |      12.02      |       7.22       |
+|         **5**         |      15.99      |       9.63       |      34.59      |      41.89       |      15.14      |       9.08       |
+|         **6**         |      19.05      |      11.58       |      38.69      |      45.92       |      18.02      |      10.93       |
+|         **7**         |      21.94      |      13.54       |      42.18      |       48.9       |      20.72      |      12.66       |
+|         **8**         |      25.11      |      15.45       |      44.35      |      51.04       |      23.92      |      14.54       |
+|        **10**         |      30.98      |      19.33       |      52.29      |      56.64       |      29.31      |      18.21       |
+|        **12**         |      37.52      |      23.18       |      59.2       |      63.33       |      35.42      |      21.82       |
+|        **14**         |      43.44      |      27.01       |      67.33      |      67.58       |      41.28      |      25.55       |
+|        **16**         |      48.93      |      31.03       |      72.06      |      72.77       |      46.11      |      28.93       |
+|        **32**         |      97.65      |      62.03       |     130.05      |      104.96      |      94.91      |      58.02       |
+|        **64**         |     193.53      |      122.24      |     235.14      |      168.45      |     191.49      |      115.26      |
+|        **128**        |     384.63      |      243.74      |     449.79      |      294.65      |     379.65      |      236.67      |
+|        **256**        |     769.13      |      486.82      |     834.07      |      561.66      |     766.46      |      498.22      |
+|        **512**        |     1538.29     |      968.02      |     1725.44     |     1052.67      |     1542.14     |      962.05      |
+|       **1024**        |     2999.36     |     1935.32      |     3219.46     |     2072.57      |     3086.34     |     1935.36      |
+
+![Latency](./images/performance-test-workflows-latency.png)
+
+## References
+
+- [NBomber](https://nbomber.com/docs/getting-started/overview/)

src/Directory.Build.props

@@ -4,10 +4,10 @@
     <PackageLicenseUrl>https://github.com/danielgerlag/workflow-core/blob/master/LICENSE.md</PackageLicenseUrl>
     <RepositoryType>git</RepositoryType>
     <RepositoryUrl>https://github.com/danielgerlag/workflow-core.git</RepositoryUrl>
-    <Version>3.8.2</Version>
-    <AssemblyVersion>3.8.2.0</AssemblyVersion>
-    <FileVersion>3.8.2.0</FileVersion>
+    <Version>3.9.0</Version>
+    <AssemblyVersion>3.9.0.0</AssemblyVersion>
+    <FileVersion>3.9.0.0</FileVersion>
     <PackageIconUrl>https://github.com/danielgerlag/workflow-core/raw/master/src/logo.png</PackageIconUrl>
-    <PackageVersion>3.8.2</PackageVersion>
+    <PackageVersion>3.9.0</PackageVersion>
   </PropertyGroup>
 </Project>

src/WorkflowCore.DSL/WorkflowCore.DSL.csproj

@@ -1,4 +1,4 @@
-<Project Sdk="Microsoft.NET.Sdk">
+<Project Sdk="Microsoft.NET.Sdk">
 
   <PropertyGroup>
     <TargetFramework>netstandard2.0</TargetFramework>
@@ -11,7 +11,7 @@
   <ItemGroup>
     <PackageReference Include="Newtonsoft.Json" Version="13.0.1" />
     <PackageReference Include="SharpYaml" Version="1.6.5" />
-    <PackageReference Include="System.Linq.Dynamic.Core" Version="1.2.13" />
+    <PackageReference Include="System.Linq.Dynamic.Core" Version="1.3.0" />
   </ItemGroup>
 
   <ItemGroup>

src/WorkflowCore/Services/BackgroundTasks/EventConsumer.cs

@@ -17,7 +17,7 @@ internal class EventConsumer : QueueConsumer, IBackgroundTask
         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, IGreyList greylist)

src/providers/WorkflowCore.Persistence.EntityFramework/WorkflowCore.Persistence.EntityFramework.csproj

@@ -22,7 +22,7 @@
   </ItemGroup>
 
   <ItemGroup Condition=" '$(TargetFramework)' == 'net6.0' ">
-    <PackageReference Include="Microsoft.EntityFrameworkCore.Relational" Version="6.0.3" />
+    <PackageReference Include="Microsoft.EntityFrameworkCore.Relational" Version="7.*" />
   </ItemGroup>
 
   <ItemGroup Condition=" '$(TargetFramework)' == 'netstandard2.1' ">

src/providers/WorkflowCore.Persistence.MySQL/WorkflowCore.Persistence.MySQL.csproj

@@ -35,11 +35,11 @@
   </ItemGroup>
 
   <ItemGroup Condition=" '$(TargetFramework)' == 'net6.0' ">
-    <PackageReference Include="Microsoft.EntityFrameworkCore.Tools" Version="6.0.3">
+    <PackageReference Include="Microsoft.EntityFrameworkCore.Tools" Version="7.*">
       <PrivateAssets>all</PrivateAssets>
       <IncludeAssets>runtime; build; native; contentfiles; analyzers</IncludeAssets>
     </PackageReference>
-    <PackageReference Include="Pomelo.EntityFrameworkCore.MySql" Version="6.0.1" />
+    <PackageReference Include="Pomelo.EntityFrameworkCore.MySql" Version="7.*" />
   </ItemGroup>
 
   <ItemGroup>

src/providers/WorkflowCore.Persistence.PostgreSQL/WorkflowCore.Persistence.PostgreSQL.csproj

@@ -23,12 +23,12 @@
   </ItemGroup>
 
   <ItemGroup Condition="'$(TargetFramework)' == 'net6.0'">
-	<PackageReference Include="Npgsql" Version="6.0.6" />
-    <PackageReference Include="Npgsql.EntityFrameworkCore.PostgreSQL" Version="6.0.6" />
-    <PackageReference Include="Microsoft.EntityFrameworkCore.Tools" Version="6.0.3">
+	<PackageReference Include="Npgsql" Version="7.*" />
+    <PackageReference Include="Npgsql.EntityFrameworkCore.PostgreSQL" Version="7.*" />
+    <PackageReference Include="Microsoft.EntityFrameworkCore.Tools" Version="7.*">
       <PrivateAssets>All</PrivateAssets>
     </PackageReference>
-    <PackageReference Include="Microsoft.EntityFrameworkCore.Design" Version="6.0.3">
+    <PackageReference Include="Microsoft.EntityFrameworkCore.Design" Version="7.*">
       <PrivateAssets>all</PrivateAssets>
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
     </PackageReference>

src/providers/WorkflowCore.Persistence.SqlServer/WorkflowCore.Persistence.SqlServer.csproj

@@ -24,11 +24,11 @@
   </ItemGroup>
 
   <ItemGroup Condition=" '$(TargetFramework)' == 'net6.0' ">
-    <PackageReference Include="Microsoft.EntityFrameworkCore.SqlServer" Version="6.0.3" />
-    <PackageReference Include="Microsoft.EntityFrameworkCore.Tools" Version="6.0.3">
+    <PackageReference Include="Microsoft.EntityFrameworkCore.SqlServer" Version="7.*" />
+    <PackageReference Include="Microsoft.EntityFrameworkCore.Tools" Version="7.*">
       <PrivateAssets>All</PrivateAssets>
     </PackageReference>
-    <PackageReference Include="Microsoft.EntityFrameworkCore.Design" Version="6.0.3">
+    <PackageReference Include="Microsoft.EntityFrameworkCore.Design" Version="7.*">
       <PrivateAssets>all</PrivateAssets>
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
     </PackageReference>

src/providers/WorkflowCore.Persistence.Sqlite/WorkflowCore.Persistence.Sqlite.csproj

@@ -24,7 +24,7 @@
   </ItemGroup>
 
   <ItemGroup Condition=" '$(TargetFramework)' == 'net6.0' ">
-    <PackageReference Include="Microsoft.EntityFrameworkCore.Sqlite" Version="6.0.3" />
+    <PackageReference Include="Microsoft.EntityFrameworkCore.Sqlite" Version="7.*" />
   </ItemGroup>
 
   <ItemGroup Condition=" '$(TargetFramework)' == 'netstandard2.1' ">

src/providers/WorkflowCore.Providers.AWS/README.md

@@ -34,6 +34,18 @@ services.AddWorkflow(cfg =>
 If any AWS resources do not exists, they will be automatcially created. By default, all DynamoDB tables and indexes will be provisioned with a throughput of 1, you can modify these values from the AWS console.
 You may also specify a prefix for the dynamo table names.
 
+If you have a preconfigured dynamoClient, you can pass this in instead of the credentials and config
+```C#
+var client = new AmazonDynamoDBClient();
+var sqsClient = new AmazonSQSClient();
+services.AddWorkflow(cfg =>
+{
+    cfg.UseAwsDynamoPersistenceWithProvisionedClient(client, "table-prefix");
+    cfg.UseAwsDynamoLockingWithProvisionedClient(client, "workflow-core-locks");
+    cfg.UseAwsSimpleQueueServiceWithProvisionedClient(sqsClient, "queues-prefix");
+});
+```
+
 
 ## Usage (Kinesis)
 

src/providers/WorkflowCore.Providers.AWS/ServiceCollectionExtensions.cs

@@ -1,6 +1,7 @@
 using System;
 using Amazon;
 using Amazon.DynamoDBv2;
+using Amazon.Kinesis;
 using Amazon.Runtime;
 using Amazon.SQS;
 using Microsoft.Extensions.Logging;
@@ -15,28 +16,55 @@ public static class ServiceCollectionExtensions
     {
         public static WorkflowOptions UseAwsSimpleQueueService(this WorkflowOptions options, AWSCredentials credentials, AmazonSQSConfig config, string queuesPrefix = "workflowcore")
         {
-            options.UseQueueProvider(sp => new SQSQueueProvider(credentials, config, sp.GetService<ILoggerFactory>(), queuesPrefix));
+            var sqsClient = new AmazonSQSClient(credentials, config);
+            return options.UseAwsSimpleQueueServiceWithProvisionedClient(sqsClient, queuesPrefix);
+        }
+
+        public static WorkflowOptions UseAwsSimpleQueueServiceWithProvisionedClient(this WorkflowOptions options, AmazonSQSClient sqsClient, string queuesPrefix = "workflowcore")
+        {
+            options.UseQueueProvider(sp => new SQSQueueProvider(sqsClient, sp.GetService<ILoggerFactory>(), queuesPrefix));
             return options;
         }
 
         public static WorkflowOptions UseAwsDynamoLocking(this WorkflowOptions options, AWSCredentials credentials, AmazonDynamoDBConfig config, string tableName)
         {
-            options.UseDistributedLockManager(sp => new DynamoLockProvider(credentials, config, tableName, sp.GetService<ILoggerFactory>(), sp.GetService<IDateTimeProvider>()));
+            var dbClient = new AmazonDynamoDBClient(credentials, config);
+            return options.UseAwsDynamoLockingWithProvisionedClient(dbClient, tableName);
+        }
+
+        public static WorkflowOptions UseAwsDynamoLockingWithProvisionedClient (this WorkflowOptions options, AmazonDynamoDBClient dynamoClient, string tableName)
+        {
+            options.UseDistributedLockManager(sp => new DynamoLockProvider(dynamoClient, tableName, sp.GetService<ILoggerFactory>(), sp.GetService<IDateTimeProvider>()));
             return options;
         }
 
         public static WorkflowOptions UseAwsDynamoPersistence(this WorkflowOptions options, AWSCredentials credentials, AmazonDynamoDBConfig config, string tablePrefix)
         {
-            options.Services.AddTransient<IDynamoDbProvisioner>(sp => new DynamoDbProvisioner(credentials, config, tablePrefix, sp.GetService<ILoggerFactory>()));
-            options.UsePersistence(sp => new DynamoPersistenceProvider(credentials, config, sp.GetService<IDynamoDbProvisioner>(), tablePrefix, sp.GetService<ILoggerFactory>()));
+            var dbClient = new AmazonDynamoDBClient(credentials, config);
+            return options.UseAwsDynamoPersistenceWithProvisionedClient(dbClient, tablePrefix);
+        }
+
+        public static WorkflowOptions UseAwsDynamoPersistenceWithProvisionedClient(this WorkflowOptions options, AmazonDynamoDBClient dynamoClient, string tablePrefix)
+        {
+            options.Services.AddTransient<IDynamoDbProvisioner>(sp => new DynamoDbProvisioner(dynamoClient, tablePrefix, sp.GetService<ILoggerFactory>()));
+            options.UsePersistence(sp => new DynamoPersistenceProvider(dynamoClient, sp.GetService<IDynamoDbProvisioner>(), tablePrefix, sp.GetService<ILoggerFactory>()));
             return options;
         }
 
         public static WorkflowOptions UseAwsKinesis(this WorkflowOptions options, AWSCredentials credentials, RegionEndpoint region, string appName, string streamName)
         {
-            options.Services.AddTransient<IKinesisTracker>(sp => new KinesisTracker(credentials, region, "workflowcore_kinesis", sp.GetService<ILoggerFactory>()));
-            options.Services.AddTransient<IKinesisStreamConsumer>(sp => new KinesisStreamConsumer(credentials, region, sp.GetService<IKinesisTracker>(), sp.GetService<IDistributedLockProvider>(), sp.GetService<ILoggerFactory>(), sp.GetService<IDateTimeProvider>()));
-            options.UseEventHub(sp => new KinesisProvider(credentials, region, appName, streamName, sp.GetService<IKinesisStreamConsumer>(), sp.GetService<ILoggerFactory>()));
+            var kinesisClient = new AmazonKinesisClient(credentials, region);
+            var dynamoClient = new AmazonDynamoDBClient(credentials, region);
+
+            return options.UseAwsKinesisWithProvisionedClients(kinesisClient, dynamoClient,appName, streamName);
+
+        }
+
+        public static WorkflowOptions UseAwsKinesisWithProvisionedClients(this WorkflowOptions options, AmazonKinesisClient kinesisClient, AmazonDynamoDBClient dynamoDbClient, string appName, string streamName)
+        {
+            options.Services.AddTransient<IKinesisTracker>(sp => new KinesisTracker(dynamoDbClient, "workflowcore_kinesis", sp.GetService<ILoggerFactory>()));
+            options.Services.AddTransient<IKinesisStreamConsumer>(sp => new KinesisStreamConsumer(kinesisClient, sp.GetService<IKinesisTracker>(), sp.GetService<IDistributedLockProvider>(), sp.GetService<ILoggerFactory>(), sp.GetService<IDateTimeProvider>()));
+            options.UseEventHub(sp => new KinesisProvider(kinesisClient, appName, streamName, sp.GetService<IKinesisStreamConsumer>(), sp.GetService<ILoggerFactory>()));
             return options;
         }
     }

src/providers/WorkflowCore.Providers.AWS/Services/DynamoDbProvisioner.cs

@@ -15,10 +15,10 @@ public class DynamoDbProvisioner : IDynamoDbProvisioner
         private readonly IAmazonDynamoDB _client;
         private readonly string _tablePrefix;
 
-        public DynamoDbProvisioner(AWSCredentials credentials, AmazonDynamoDBConfig config, string tablePrefix, ILoggerFactory logFactory)
+        public DynamoDbProvisioner(AmazonDynamoDBClient dynamoDBClient, string tablePrefix, ILoggerFactory logFactory)
         {
             _logger = logFactory.CreateLogger<DynamoDbProvisioner>();
-            _client = new AmazonDynamoDBClient(credentials, config);
+            _client = dynamoDBClient;
             _tablePrefix = tablePrefix;
         }
 

src/providers/WorkflowCore.Providers.AWS/Services/DynamoLockProvider.cs

@@ -25,10 +25,10 @@ public class DynamoLockProvider : IDistributedLockProvider
         private readonly AutoResetEvent _mutex = new AutoResetEvent(true);
         private readonly IDateTimeProvider _dateTimeProvider;
 
-        public DynamoLockProvider(AWSCredentials credentials, AmazonDynamoDBConfig config, string tableName, ILoggerFactory logFactory, IDateTimeProvider dateTimeProvider)
+        public DynamoLockProvider(AmazonDynamoDBClient dynamoDBClient, string tableName, ILoggerFactory logFactory, IDateTimeProvider dateTimeProvider)
         {
             _logger = logFactory.CreateLogger<DynamoLockProvider>();
-            _client = new AmazonDynamoDBClient(credentials, config);
+            _client = dynamoDBClient;
             _localLocks = new List<string>();
             _tableName = tableName;
             _nodeId = Guid.NewGuid().ToString();

src/providers/WorkflowCore.Providers.AWS/Services/DynamoPersistenceProvider.cs

@@ -26,10 +26,10 @@ public class DynamoPersistenceProvider : IPersistenceProvider
 
         public bool SupportsScheduledCommands => false;
 
-        public DynamoPersistenceProvider(AWSCredentials credentials, AmazonDynamoDBConfig config, IDynamoDbProvisioner provisioner, string tablePrefix, ILoggerFactory logFactory)
+        public DynamoPersistenceProvider(AmazonDynamoDBClient dynamoDBClient, IDynamoDbProvisioner provisioner, string tablePrefix, ILoggerFactory logFactory)
         {
             _logger = logFactory.CreateLogger<DynamoPersistenceProvider>();
-            _client = new AmazonDynamoDBClient(credentials, config);
+            _client = dynamoDBClient;
             _tablePrefix = tablePrefix;
             _provisioner = provisioner;
         }