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

Merge pull request #6326 from elsa-workflows/perf/activity-resumption

Browse source 
Refactor workflow execution handling to improve performance
This commit is contained in:
Sipke Schoorstra 2025-01-24 14:32:55 +01:00 committed by GitHub
parent f67d96976d 184b159c6f
commit 09c31d9e45
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 98 additions and 63 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.github/workflows/packages.yml vendored
View file

@ -9,7 +9,7 @@ on:
release:
types: [ prereleased, published ]
env:
base_version: '3.3.0'
base_version: '3.3.1'
feedz_feed_source: 'https://f.feedz.io/elsa-workflows/elsa-3/nuget/index.json'
nuget_feed_source: 'https://api.nuget.org/v3/index.json'

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

@ -104,19 +104,18 @@ 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;
var items = await context.GetItemSource<object>(Items).ToListAsync(context.CancellationToken);
var count = items.Count;
await foreach (var item in items)
{
await DispatchChildWorkflowAsync(context, item);
dispatchedInstancesCount++;
}
// Dispatch the child workflows.
foreach (var item in items)
await DispatchChildWorkflowAsync(context, item, waitForCompletion);
context.SetProperty(DispatchedInstancesCountKey, dispatchedInstancesCount);
// Store the number of dispatched instances for tracking.
context.SetProperty(DispatchedInstancesCountKey, 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,11 +124,12 @@ public class BulkDispatchWorkflows : Activity
Stimulus = new BulkDispatchWorkflowsStimulus(workflowInstanceId)
{
ParentInstanceId = context.WorkflowExecutionContext.Id,
ScheduledInstanceIdsCount = dispatchedInstancesCount
ScheduledInstanceIdsCount = count
},
IncludeActivityInstanceId = false,
AutoBurn = false,
};
context.CreateBookmark(bookmarkOptions);
}
else
@ -139,9 +139,15 @@ public class BulkDispatchWorkflows : Activity
}
}
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);
@ -151,6 +157,9 @@ public class BulkDispatchWorkflows : Activity
["ParentInstanceId"] = parentInstanceId
};
if (waitForCompletion)
properties["WaitForCompletion"] = true;
var itemDictionary = new Dictionary<string, object>
{
[defaultInputItemKey] = item
@ -168,12 +177,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)

30
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,38 @@ 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 workflowDefinitionService = context.GetRequiredService<IWorkflowDefinitionService>();
var workflowGraph = await workflowDefinitionService.FindWorkflowGraphAsync(workflowDefinitionId, VersionOptions.Published, context.CancellationToken);
if (workflowGraph == null)
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
};
input["ParentInstanceId"] = context.WorkflowExecutionContext.Id;
// If we need to wait for the child workflow to complete, set the property. This will be used by the ResumeDispatchWorkflowActivity handler.
if (waitForCompletion)
properties["WaitForCompletion"] = true;
input["ParentInstanceId"] = parentInstanceId;
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.");
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,
};

49
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,33 +40,33 @@ 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);
if(!waitForCompletion || result.Status == WorkflowStatus.Finished)
var result = await ExecuteWorkflowAsync(context, waitForCompletion);
if (!waitForCompletion || result.Status == WorkflowStatus.Finished)
{
context.SetResult(result);
await context.CompleteActivityAsync();
return;
}
// Since the child workflow is still running, we need to wait for it to complete using a bookmark.
var bookmarkOptions = new CreateBookmarkArgs
{
@ -77,23 +77,36 @@ 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 we need to wait for the child workflow to complete, set the property. This will be used by the ResumeExecuteWorkflowActivity to resume the parent workflow.
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()
};
@ -109,7 +122,7 @@ public class ExecuteWorkflow : Activity<ExecuteWorkflowResult>
return info;
}
private async ValueTask OnChildWorkflowCompletedAsync(ActivityExecutionContext context)
{
var input = context.WorkflowInput;

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);

21
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;
@ -16,33 +15,35 @@ namespace Elsa.Workflows.Runtime.Handlers;
internal class ResumeDispatchWorkflowActivity(IBookmarkQueue bookmarkQueue, IStimulusHasher stimulusHasher, ILogger<ResumeDispatchWorkflowActivity> logger) : INotificationHandler<WorkflowExecuted>
{
private static readonly string ActivityTypeName = ActivityTypeNameHelper.GenerateTypeName<DispatchWorkflow>();
public async Task HandleAsync(WorkflowExecuted notification, CancellationToken cancellationToken)
{
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)
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
}
};
logger.LogDebug("Resuming any blocking DispatchWorkflow activities for workflow {WorkflowInstanceId} using stimulus hash {StimulusHash}", notification.WorkflowState.Id, bookmarkQueueItem.StimulusHash);
await bookmarkQueue.EnqueueAsync(bookmarkQueueItem, cancellationToken);
}
}

13
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,22 @@ 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
}