diff --git a/src/modules/Elsa.Scheduling/Services/DefaultBookmarkScheduler.cs b/src/modules/Elsa.Scheduling/Services/DefaultBookmarkScheduler.cs index c1d86fdc6d..ce4946b9ff 100644 --- a/src/modules/Elsa.Scheduling/Services/DefaultBookmarkScheduler.cs +++ b/src/modules/Elsa.Scheduling/Services/DefaultBookmarkScheduler.cs @@ -10,6 +10,7 @@ namespace Elsa.Scheduling.Services; /// public class DefaultBookmarkScheduler : IBookmarkScheduler { + private const int MaxConcurrentSchedulingOperations = 16; private readonly IWorkflowScheduler _workflowScheduler; /// @@ -28,58 +29,13 @@ public async Task ScheduleAsync(IEnumerable bookmarks, Cancellat var startAtBookmarks = bookmarkList.Filter(SchedulingStimulusNames.StartAt); var timerBookmarks = bookmarkList.Filter(SchedulingStimulusNames.Timer); var cronBookmarks = bookmarkList.Filter(SchedulingStimulusNames.Cron); + var schedules = delayBookmarks + .Select(bookmark => CreateScheduleAt(bookmark.Id, bookmark.WorkflowInstanceId, bookmark.GetPayload().ResumeAt)) + .Concat(startAtBookmarks.Select(bookmark => CreateScheduleAt(bookmark.Id, bookmark.WorkflowInstanceId, bookmark.GetPayload().ExecuteAt))) + .Concat(timerBookmarks.Select(bookmark => CreateScheduleAt(bookmark.Id, bookmark.WorkflowInstanceId, bookmark.GetPayload().ResumeAt))) + .Concat(cronBookmarks.Select(bookmark => CreateCronSchedule(bookmark.Id, bookmark.WorkflowInstanceId, bookmark.GetPayload().CronExpression))); - // Schedule each Delay bookmark. - foreach (var bookmark in delayBookmarks) - { - var payload = bookmark.GetPayload(); - var resumeAt = payload.ResumeAt; - var request = new ScheduleExistingWorkflowInstanceRequest - { - WorkflowInstanceId = bookmark.WorkflowInstanceId, - BookmarkId = bookmark.Id - }; - await _workflowScheduler.ScheduleAtAsync(bookmark.Id, request, resumeAt, cancellationToken); - } - - // Schedule each StartAt bookmark. - foreach (var bookmark in startAtBookmarks) - { - var payload = bookmark.GetPayload(); - var executeAt = payload.ExecuteAt; - var request = new ScheduleExistingWorkflowInstanceRequest - { - WorkflowInstanceId = bookmark.WorkflowInstanceId, - BookmarkId = bookmark.Id - }; - await _workflowScheduler.ScheduleAtAsync(bookmark.Id, request, executeAt, cancellationToken); - } - - // Schedule each Timer bookmark. - foreach (var bookmark in timerBookmarks) - { - var payload = bookmark.GetPayload(); - var resumeAt = payload.ResumeAt; - var request = new ScheduleExistingWorkflowInstanceRequest - { - WorkflowInstanceId = bookmark.WorkflowInstanceId, - BookmarkId = bookmark.Id - }; - await _workflowScheduler.ScheduleAtAsync(bookmark.Id, request, resumeAt, cancellationToken); - } - - // Schedule each Cron bookmark. - foreach (var bookmark in cronBookmarks) - { - var payload = bookmark.GetPayload(); - var cronExpression = payload.CronExpression; - var request = new ScheduleExistingWorkflowInstanceRequest - { - WorkflowInstanceId = bookmark.WorkflowInstanceId, - BookmarkId = bookmark.Id - }; - await _workflowScheduler.ScheduleCronAsync(bookmark.Id, request, cronExpression, cancellationToken); - } + await ScheduleAsync(schedules, cancellationToken); } /// @@ -90,58 +46,13 @@ public async Task ScheduleAsync(string workflowInstanceId, IEnumerable var startAtBookmarks = bookmarkList.Filter(SchedulingStimulusNames.StartAt); var timerBookmarks = bookmarkList.Filter(SchedulingStimulusNames.Timer); var cronBookmarks = bookmarkList.Filter(SchedulingStimulusNames.Cron); + var schedules = delayBookmarks + .Select(bookmark => CreateScheduleAt(bookmark.Id, workflowInstanceId, bookmark.GetPayload().ResumeAt)) + .Concat(startAtBookmarks.Select(bookmark => CreateScheduleAt(bookmark.Id, workflowInstanceId, bookmark.GetPayload().ExecuteAt))) + .Concat(timerBookmarks.Select(bookmark => CreateScheduleAt(bookmark.Id, workflowInstanceId, bookmark.GetPayload().ResumeAt))) + .Concat(cronBookmarks.Select(bookmark => CreateCronSchedule(bookmark.Id, workflowInstanceId, bookmark.GetPayload().CronExpression))); - // Schedule each Delay bookmark. - foreach (var bookmark in delayBookmarks) - { - var payload = bookmark.GetPayload(); - var resumeAt = payload.ResumeAt; - var request = new ScheduleExistingWorkflowInstanceRequest - { - WorkflowInstanceId = workflowInstanceId, - BookmarkId = bookmark.Id - }; - await _workflowScheduler.ScheduleAtAsync(bookmark.Id, request, resumeAt, cancellationToken); - } - - // Schedule each StartAt bookmark. - foreach (var bookmark in startAtBookmarks) - { - var payload = bookmark.GetPayload(); - var executeAt = payload.ExecuteAt; - var request = new ScheduleExistingWorkflowInstanceRequest - { - WorkflowInstanceId = workflowInstanceId, - BookmarkId = bookmark.Id - }; - await _workflowScheduler.ScheduleAtAsync(bookmark.Id, request, executeAt, cancellationToken); - } - - // Schedule each Timer bookmark. - foreach (var bookmark in timerBookmarks) - { - var payload = bookmark.GetPayload(); - var resumeAt = payload.ResumeAt; - var request = new ScheduleExistingWorkflowInstanceRequest - { - WorkflowInstanceId = workflowInstanceId, - BookmarkId = bookmark.Id - }; - await _workflowScheduler.ScheduleAtAsync(bookmark.Id, request, resumeAt, cancellationToken); - } - - // Schedule each Cron bookmark. - foreach (var bookmark in cronBookmarks) - { - var payload = bookmark.GetPayload(); - var cronExpression = payload.CronExpression; - var request = new ScheduleExistingWorkflowInstanceRequest - { - WorkflowInstanceId = workflowInstanceId, - BookmarkId = bookmark.Id - }; - await _workflowScheduler.ScheduleCronAsync(bookmark.Id, request, cronExpression, cancellationToken); - } + await ScheduleAsync(schedules, cancellationToken); } /// @@ -171,4 +82,41 @@ public async Task UnscheduleAsync(IEnumerable bookmarks, Cancell foreach (var bookmark in bookmarksToUnSchedule) await _workflowScheduler.UnscheduleAsync(bookmark.Id, cancellationToken); } -} \ No newline at end of file + + private Task ScheduleAsync(IEnumerable schedules, CancellationToken cancellationToken) + { + var options = new ParallelOptions + { + MaxDegreeOfParallelism = MaxConcurrentSchedulingOperations, + CancellationToken = cancellationToken + }; + + return Parallel.ForEachAsync(schedules, options, ScheduleBookmarkAsync); + } + + private async ValueTask ScheduleBookmarkAsync(BookmarkSchedule schedule, CancellationToken cancellationToken) + { + var request = new ScheduleExistingWorkflowInstanceRequest + { + WorkflowInstanceId = schedule.WorkflowInstanceId, + BookmarkId = schedule.BookmarkId + }; + + if (schedule.CronExpression != null) + await _workflowScheduler.ScheduleCronAsync(schedule.BookmarkId, request, schedule.CronExpression, cancellationToken); + else + await _workflowScheduler.ScheduleAtAsync(schedule.BookmarkId, request, schedule.ExecuteAt, cancellationToken); + } + + private static BookmarkSchedule CreateScheduleAt(string bookmarkId, string workflowInstanceId, DateTimeOffset executeAt) + { + return new BookmarkSchedule(bookmarkId, workflowInstanceId, executeAt, null); + } + + private static BookmarkSchedule CreateCronSchedule(string bookmarkId, string workflowInstanceId, string cronExpression) + { + return new BookmarkSchedule(bookmarkId, workflowInstanceId, default, cronExpression); + } + + private readonly record struct BookmarkSchedule(string BookmarkId, string WorkflowInstanceId, DateTimeOffset ExecuteAt, string? CronExpression); +} diff --git a/test/unit/Elsa.Scheduling.UnitTests/Services/DefaultBookmarkSchedulerTests.cs b/test/unit/Elsa.Scheduling.UnitTests/Services/DefaultBookmarkSchedulerTests.cs new file mode 100644 index 0000000000..bd2842cc5d --- /dev/null +++ b/test/unit/Elsa.Scheduling.UnitTests/Services/DefaultBookmarkSchedulerTests.cs @@ -0,0 +1,71 @@ +using Elsa.Scheduling.Bookmarks; +using Elsa.Scheduling.Services; +using Elsa.Workflows.Runtime.Entities; +using NSubstitute; + +namespace Elsa.Scheduling.UnitTests.Services; + +public class DefaultBookmarkSchedulerTests +{ + private readonly IWorkflowScheduler _workflowScheduler; + private readonly DefaultBookmarkScheduler _scheduler; + + public DefaultBookmarkSchedulerTests() + { + _workflowScheduler = Substitute.For(); + _scheduler = new DefaultBookmarkScheduler(_workflowScheduler); + } + + [Fact] + public async Task ScheduleAsync_WithStoredBookmarks_StartsSchedulingConcurrently() + { + var releaseScheduler = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var scheduledCount = 0; + var bookmarks = CreateStoredDelayBookmarks(2); + _workflowScheduler + .ScheduleAtAsync(Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(_ => + { + Interlocked.Increment(ref scheduledCount); + return new ValueTask(releaseScheduler.Task); + }); + + var scheduleTask = _scheduler.ScheduleAsync(bookmarks, CancellationToken.None); + + var allSchedulingStarted = await WaitUntilAsync(() => Volatile.Read(ref scheduledCount) == bookmarks.Count, TimeSpan.FromSeconds(5)); + var completedBeforeRelease = scheduleTask.IsCompleted; + releaseScheduler.SetResult(); + await scheduleTask; + + Assert.True(allSchedulingStarted); + Assert.False(completedBeforeRelease); + } + + private static IReadOnlyCollection CreateStoredDelayBookmarks(int count) + { + var now = DateTimeOffset.UtcNow; + return Enumerable.Range(1, count).Select(index => new StoredBookmark + { + Id = $"bookmark-{index}", + Name = SchedulingStimulusNames.Delay, + Hash = $"hash-{index}", + WorkflowInstanceId = $"workflow-{index}", + Payload = new DelayPayload(now.AddMinutes(index)), + CreatedAt = now + }).ToList(); + } + + private static async Task WaitUntilAsync(Func condition, TimeSpan timeout) + { + var stopAt = DateTimeOffset.UtcNow.Add(timeout); + while (DateTimeOffset.UtcNow < stopAt) + { + if (condition()) + return true; + + await Task.Delay(10); + } + + return condition(); + } +}