SemianiakaVY/elsa-core
1
0
Fork 0
Code Issues Pull requests Actions 1 Packages Projects Releases Wiki Activity

Implement batch processing in BulkDispatchWorkflows

Browse source 
The changes introduce a batch processing mechanism in the BulkDispatchWorkflows activity. Previously, each workflow was dispatched individually. Now, the process groups the workflows in batches, processes each batch concurrently, and handles errors. This change improves performance, particularly for large data sources.
This commit is contained in:
Sipke Schoorstra 2023-12-16 11:45:31 +01:00
parent 1204ac0319
commit 65cb2640e3
1 changed files with 40 additions and 10 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

50
src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs
View file

@ -23,7 +23,7 @@ namespace Elsa.Workflows.Runtime.Activities;
/// <summary>
/// Creates new workflow instances of the specified workflow for each item in the data source and dispatches them for execution.
/// </summary>
[Activity("Elsa", "Composition", "Create new workflow instances for each item in the data source and dispatch them for execution.")]
[Activity("Elsa", "Composition", "Create new workflow instances for each item in the data source and dispatch them for execution.", Kind = ActivityKind.Task)]
[FlowNode("Finished", "Canceled", "Done")]
[UsedImplicitly]
public class BulkDispatchWorkflows : Activity
@ -98,19 +98,24 @@ public class BulkDispatchWorkflows : Activity
var waitForCompletion = WaitForCompletion.GetOrDefault(context);
var items = GetItemsAsync(context).WithCancellation(context.CancellationToken);
var dispatchedInstancesCount = 0L;
var batchSize = 1000;
var batch = new List<object>();
try
await foreach (var item in items)
{
await foreach (var item in items)
{
await DispatchChildWorkflowAsync(context, item);
dispatchedInstancesCount++;
}
batch.Add(item);
if (batch.Count < batchSize)
continue;
await ProcessBatch(context, batch);
batch.Clear();
}
catch (TaskCanceledException)
// Process the last batch if it has any items.
if (batch.Count > 0)
{
await context.CompleteActivityWithOutcomesAsync("Canceled");
return;
await ProcessBatch(context, batch);
}
context.SetProperty(DispatchedInstancesCountKey, dispatchedInstancesCount);
@ -136,6 +141,31 @@ public class BulkDispatchWorkflows : Activity
// Otherwise, we can complete immediately.
await context.CompleteActivityAsync();
}
// Cancelling children
}
private async Task ProcessBatch(ActivityExecutionContext context, List<object> items)
{
var tasks = items.Select(async item =>
{
try
{
await DispatchChildWorkflowAsync(context, item);
}
catch (TaskCanceledException)
{
await context.CompleteActivityWithOutcomesAsync("Canceled");
}
catch (Exception ex)
{
context.JournalData.Add("Error", ex.Message);
}
});
await Task.WhenAll(tasks);
}
private async ValueTask<string> DispatchChildWorkflowAsync(ActivityExecutionContext context, object item)