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

Refactor workflow execution handling to improve performance

Browse source
This commit is contained in:
Sipke Schoorstra 2025-01-24 13:54:37 +01:00
parent f67d96976d
commit fbd16f2d74
6 changed files with 109 additions and 56 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -104,19 +104,11 @@ public class BulkDispatchWorkflows : Activity
protected override async ValueTask ExecuteAsync(ActivityExecutionContext context)
{
var waitForCompletion = WaitForCompletion.GetOrDefault(context);
var items = context.GetItemSource<object>(Items);
var dispatchedInstancesCount = 0;
await foreach (var item in items)
{
await DispatchChildWorkflowAsync(context, item);
dispatchedInstancesCount++;
}
context.SetProperty(DispatchedInstancesCountKey, dispatchedInstancesCount);
var items = await context.GetItemSource<object>(Items).ToListAsync(context.CancellationToken);
var count = items.Count;
// If we need to wait for the child workflows to complete (if any), create a bookmark.
if (waitForCompletion && dispatchedInstancesCount > 0)
if (waitForCompletion && count > 0)
{
var workflowInstanceId = context.WorkflowExecutionContext.Id;
var bookmarkOptions = new CreateBookmarkArgs
@ -125,23 +117,45 @@ public class BulkDispatchWorkflows : Activity
Stimulus = new BulkDispatchWorkflowsStimulus(workflowInstanceId)
{
ParentInstanceId = context.WorkflowExecutionContext.Id,
ScheduledInstanceIdsCount = dispatchedInstancesCount
ScheduledInstanceIdsCount = count
},
IncludeActivityInstanceId = false,
AutoBurn = false,
};
// Create bookmarks first.
context.CreateBookmark(bookmarkOptions);
// Dispatch workflows afterwards.
await DispatchWorkflowsAsync();
}
else
{
// Otherwise, we can complete immediately.
await DispatchWorkflowsAsync();
await context.CompleteActivityWithOutcomesAsync("Done");
}
return;
async Task DispatchWorkflowsAsync()
{
foreach (var item in items)
await DispatchChildWorkflowAsync(context, item, waitForCompletion);
context.SetProperty(DispatchedInstancesCountKey, count);
}
}
private async ValueTask<string> DispatchChildWorkflowAsync(ActivityExecutionContext context, object item)
private async ValueTask<string> DispatchChildWorkflowAsync(ActivityExecutionContext context, object item, bool waitForCompletion)
{
var workflowDefinitionId = WorkflowDefinitionId.Get(context);
var workflowDefinitionService = context.GetRequiredService<IWorkflowDefinitionService>();
var workflowGraph = await workflowDefinitionService.FindWorkflowGraphAsync(workflowDefinitionId, VersionOptions.Published);
if (workflowGraph == null)
throw new($"No published version of workflow definition with ID {workflowDefinitionId} found.");
var parentInstanceId = context.WorkflowExecutionContext.Id;
var input = Input.GetOrDefault(context) ?? new Dictionary<string, object>();
var channelName = ChannelName.GetOrDefault(context);
@ -150,6 +164,9 @@ public class BulkDispatchWorkflows : Activity
{
["ParentInstanceId"] = parentInstanceId
};
if(waitForCompletion)
properties["WaitForCompletion"] = true;
var itemDictionary = new Dictionary<string, object>
{
@ -168,12 +185,6 @@ public class BulkDispatchWorkflows : Activity
var workflowDispatcher = context.GetRequiredService<IWorkflowDispatcher>();
var identityGenerator = context.GetRequiredService<IIdentityGenerator>();
var evaluator = context.GetRequiredService<IExpressionEvaluator>();
var workflowDefinitionService = context.GetRequiredService<IWorkflowDefinitionService>();
var workflowGraph = await workflowDefinitionService.FindWorkflowGraphAsync(workflowDefinitionId, VersionOptions.Published);
if (workflowGraph == null)
throw new Exception($"No published version of workflow definition with ID {workflowDefinitionId} found.");
var correlationId = CorrelationIdFunction != null ? await evaluator.EvaluateAsync<string>(CorrelationIdFunction!, context.ExpressionExecutionContext, evaluatorOptions) : null;
var instanceId = identityGenerator.GenerateId();
var request = new DispatchWorkflowDefinitionRequest(workflowGraph.Workflow.Identity.Id)

35
src/modules/Elsa.Workflows.Runtime/Activities/DispatchWorkflow.cs
View file

@ -72,7 +72,7 @@ public class DispatchWorkflow : Activity<object>
var waitForCompletion = WaitForCompletion.GetOrDefault(context);
// Dispatch the child workflow.
var instanceId = await DispatchChildWorkflowAsync(context);
var instanceId = await DispatchChildWorkflowAsync(context, waitForCompletion);
// If we need to wait for the child workflow to complete, create a bookmark.
if (waitForCompletion)
@ -92,28 +92,39 @@ public class DispatchWorkflow : Activity<object>
}
}
private async ValueTask<string> DispatchChildWorkflowAsync(ActivityExecutionContext context)
private async ValueTask<string> DispatchChildWorkflowAsync(ActivityExecutionContext context, bool waitForCompletion)
{
var workflowDefinitionId = WorkflowDefinitionId.Get(context);
var input = Input.GetOrDefault(context) ?? new Dictionary<string, object>();
var channelName = ChannelName.GetOrDefault(context);
input["ParentInstanceId"] = context.WorkflowExecutionContext.Id;
var correlationId = CorrelationId.GetOrDefault(context);
var workflowDispatcher = context.GetRequiredService<IWorkflowDispatcher>();
var identityGenerator = context.GetRequiredService<IIdentityGenerator>();
var workflowDefinitionService = context.GetRequiredService<IWorkflowDefinitionService>();
var workflowGraph = await workflowDefinitionService.FindWorkflowGraphAsync(workflowDefinitionId, VersionOptions.Published, context.CancellationToken);
if (workflowGraph == null)
throw new Exception($"No published version of workflow definition with ID {workflowDefinitionId} found.");
throw new($"No published version of workflow definition with ID {workflowDefinitionId} found.");
var input = Input.GetOrDefault(context) ?? new Dictionary<string, object>();
var channelName = ChannelName.GetOrDefault(context);
var parentInstanceId = context.WorkflowExecutionContext.Id;
var properties = new Dictionary<string, object>
{
["ParentInstanceId"] = parentInstanceId
};
if(waitForCompletion)
properties["WaitForCompletion"] = true;
input["ParentInstanceId"] = parentInstanceId;
var correlationId = CorrelationId.GetOrDefault(context);
var workflowDispatcher = context.GetRequiredService<IWorkflowDispatcher>();
var identityGenerator = context.GetRequiredService<IIdentityGenerator>();
var instanceId = identityGenerator.GenerateId();
var request = new DispatchWorkflowDefinitionRequest(workflowGraph.Workflow.Identity.Id)
{
ParentWorkflowInstanceId = context.WorkflowExecutionContext.Id,
ParentWorkflowInstanceId = parentInstanceId,
Input = input,
Properties = properties,
CorrelationId = correlationId,
InstanceId = instanceId,
};

40
src/modules/Elsa.Workflows.Runtime/Activities/ExecuteWorkflow.cs
View file

@ -19,7 +19,7 @@ namespace Elsa.Workflows.Runtime.Activities;
public class ExecuteWorkflow : Activity<ExecuteWorkflowResult>
{
/// <inheritdoc />
public ExecuteWorkflow([CallerFilePath] string? source = default, [CallerLineNumber] int? line = default) : base(source, line)
public ExecuteWorkflow([CallerFilePath] string? source = null, [CallerLineNumber] int? line = null) : base(source, line)
{
}
@ -31,7 +31,7 @@ public class ExecuteWorkflow : Activity<ExecuteWorkflowResult>
Description = "The definition ID of the workflow to execute.",
UIHint = InputUIHints.WorkflowDefinitionPicker
)]
public Input<string> WorkflowDefinitionId { get; set; } = default!;
public Input<string> WorkflowDefinitionId { get; set; } = null!;
/// <summary>
/// The correlation ID to associate the workflow with.
@ -40,25 +40,25 @@ public class ExecuteWorkflow : Activity<ExecuteWorkflowResult>
DisplayName = "Correlation ID",
Description = "The correlation ID to associate the workflow with."
)]
public Input<string?> CorrelationId { get; set; } = default!;
public Input<string?> CorrelationId { get; set; } = null!;
/// <summary>
/// The input to send to the workflow.
/// </summary>
[Input(Description = "The input to send to the workflow.")]
public Input<IDictionary<string, object>?> Input { get; set; } = default!;
public Input<IDictionary<string, object>?> Input { get; set; } = null!;
/// <summary>
/// True to wait for the child workflow to complete before completing this activity. If not set, the child workflow will be executed until it either completes or goes idle before this activity completes.
/// </summary>
[Input(Description = "Wait for the child workflow to complete before completing this activity.")]
public Input<bool> WaitForCompletion { get; set; } = default!;
public Input<bool> WaitForCompletion { get; set; } = null!;
/// <inheritdoc />
protected override async ValueTask ExecuteAsync(ActivityExecutionContext context)
{
var result = await ExecuteWorkflowAsync(context);
var waitForCompletion = WaitForCompletion.Get(context);
var result = await ExecuteWorkflowAsync(context, waitForCompletion);
if(!waitForCompletion || result.Status == WorkflowStatus.Finished)
{
@ -77,23 +77,35 @@ public class ExecuteWorkflow : Activity<ExecuteWorkflowResult>
context.CreateBookmark(bookmarkOptions);
}
private async ValueTask<ExecuteWorkflowResult> ExecuteWorkflowAsync(ActivityExecutionContext context)
private async ValueTask<ExecuteWorkflowResult> ExecuteWorkflowAsync(ActivityExecutionContext context, bool waitForCompletion)
{
var workflowDefinitionId = WorkflowDefinitionId.Get(context);
var input = Input.GetOrDefault(context) ?? new Dictionary<string, object>();
var correlationId = CorrelationId.GetOrDefault(context);
var workflowInvoker = context.GetRequiredService<IWorkflowInvoker>();
var identityGenerator = context.GetRequiredService<IIdentityGenerator>();
var workflowDefinitionService = context.GetRequiredService<IWorkflowDefinitionService>();
var workflowGraph = await workflowDefinitionService.FindWorkflowGraphAsync(workflowDefinitionId, VersionOptions.Published, context.CancellationToken);
if (workflowGraph == null)
throw new Exception($"No published version of workflow definition with ID {workflowDefinitionId} found.");
throw new($"No published version of workflow definition with ID {workflowDefinitionId} found.");
var parentInstanceId = context.WorkflowExecutionContext.Id;
var input = Input.GetOrDefault(context) ?? new Dictionary<string, object>();
var correlationId = CorrelationId.GetOrDefault(context);
var workflowInvoker = context.GetRequiredService<IWorkflowInvoker>();
var identityGenerator = context.GetRequiredService<IIdentityGenerator>();
var properties = new Dictionary<string, object>
{
["ParentInstanceId"] = parentInstanceId
};
if(waitForCompletion)
properties["WaitForCompletion"] = true;
input["ParentInstanceId"] = parentInstanceId;
var options = new RunWorkflowOptions
{
ParentWorkflowInstanceId = context.WorkflowExecutionContext.Id,
ParentWorkflowInstanceId = parentInstanceId,
Input = input,
Properties = properties,
CorrelationId = correlationId,
WorkflowInstanceId = identityGenerator.GenerateId()
};

9
src/modules/Elsa.Workflows.Runtime/Handlers/ResumeBulkDispatchWorkflowActivity.cs
View file

@ -21,11 +21,12 @@ internal class ResumeBulkDispatchWorkflowActivity(IBookmarkQueue bookmarkQueue,
if (workflowState.Status != WorkflowStatus.Finished)
return;
var parentInstanceId = workflowState.Properties.TryGetValue("ParentInstanceId", out var parentInstanceIdValue) ? parentInstanceIdValue.ToString() : default;
if (string.IsNullOrWhiteSpace(parentInstanceId))
var waitForCompletion = workflowState.Properties.TryGetValue("WaitForCompletion", out var waitForCompletionValue) && (bool)waitForCompletionValue;
if (!waitForCompletion)
return;
var parentInstanceId = (string)workflowState.Properties["ParentInstanceId"];
var activityTypeName = ActivityTypeNameHelper.GenerateTypeName<BulkDispatchWorkflows>();
var stimulus = new BulkDispatchWorkflowsStimulus(parentInstanceId);
var stimulusHash = stimulusHasher.Hash(activityTypeName, stimulus);

18
src/modules/Elsa.Workflows.Runtime/Handlers/ResumeDispatchWorkflowActivity.cs
View file

@ -2,7 +2,6 @@ using Elsa.Mediator.Contracts;
using Elsa.Workflows.Helpers;
using Elsa.Workflows.Notifications;
using Elsa.Workflows.Runtime.Activities;
using Elsa.Workflows.Runtime.Options;
using Elsa.Workflows.Runtime.Stimuli;
using JetBrains.Annotations;
using Microsoft.Extensions.Logging;
@ -22,21 +21,32 @@ internal class ResumeDispatchWorkflowActivity(IBookmarkQueue bookmarkQueue, ISti
var workflowState = notification.WorkflowState;
logger.LogDebug("Handling workflow executed notification for workflow {WorkflowInstanceId}", notification.WorkflowState.Id);
if (workflowState.Status != WorkflowStatus.Finished)
{
logger.LogDebug("Workflow {WorkflowInstanceId} is not in a finished state. Skipping resumption of any blocking DispatchWorkflow activities", notification.WorkflowState.Id);
return;
}
var props = workflowState.Properties;
var waitForCompletion = props.TryGetValue("WaitForCompletion", out var waitForCompletionValue) && (bool)waitForCompletionValue;
if (!waitForCompletion)
{
logger.LogDebug("Workflow {WorkflowInstanceId} does not have a WaitForCompletion property set to true. Skipping resumption of any blocking DispatchWorkflow activities", notification.WorkflowState.Id);
return;
}
var parentInstanceId = (string) props["ParentInstanceId"];
var stimulus = new DispatchWorkflowStimulus(notification.WorkflowState.Id);
var input = workflowState.Output;
var bookmarkQueueItem = new NewBookmarkQueueItem
{
WorkflowInstanceId = parentInstanceId,
ActivityTypeName = ActivityTypeName,
StimulusHash = stimulusHasher.Hash(ActivityTypeName, stimulus),
Options = new ResumeBookmarkOptions
Options = new()
{
Input = input
}

14
src/modules/Elsa.Workflows.Runtime/Handlers/ResumeExecuteWorkflowActivity.cs
View file

@ -2,7 +2,6 @@ using Elsa.Mediator.Contracts;
using Elsa.Workflows.Helpers;
using Elsa.Workflows.Notifications;
using Elsa.Workflows.Runtime.Activities;
using Elsa.Workflows.Runtime.Options;
using Elsa.Workflows.Runtime.Stimuli;
using JetBrains.Annotations;
@ -11,7 +10,7 @@ namespace Elsa.Workflows.Runtime.Handlers;
/// <summary>
/// Resumes any blocking <see cref="ExecuteWorkflow"/> activities when its child workflow completes.
/// </summary>
[PublicAPI]
[UsedImplicitly]
internal class ResumeExecuteWorkflowActivity(IBookmarkQueue bookmarkQueue, IStimulusHasher stimulusHasher) : INotificationHandler<WorkflowExecuted>
{
private static readonly string ActivityTypeName = ActivityTypeNameHelper.GenerateTypeName<ExecuteWorkflow>();
@ -23,14 +22,23 @@ internal class ResumeExecuteWorkflowActivity(IBookmarkQueue bookmarkQueue, IStim
if (workflowState.Status != WorkflowStatus.Finished)
return;
var props = workflowState.Properties;
var waitForCompletion = props.TryGetValue("WaitForCompletion", out var waitForCompletionValue) && (bool)waitForCompletionValue;
if (!waitForCompletion)
return;
var parentInstanceId = (string)props["ParentInstanceId"];
var stimulus = new ExecuteWorkflowStimulus(notification.WorkflowState.Id);
var input = workflowState.Output;
var bookmarkQueueItem = new NewBookmarkQueueItem
{
WorkflowInstanceId = parentInstanceId,
ActivityTypeName = ActivityTypeName,
StimulusHash = stimulusHasher.Hash(ActivityTypeName, stimulus),
Options = new ResumeBookmarkOptions
Options = new()
{
Input = input
}