From 3fd8a252cdf7a0f009c0140cc551c32acdacfcb2 Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Fri, 25 Jul 2025 07:46:35 +0200 Subject: [PATCH] Add cancellation handling for executing scheduled tasks (#6819) * Add cancellation handling for executing scheduled tasks Enhanced `ScheduledRecurringTask`, `ScheduledCronTask`, and `ScheduledSpecificInstantTask` to properly handle cancellation scenarios when tasks are executing. Introduced `_executing` and `_cancellationRequested` flags to ensure clean cancellation processes. * Refactor `ScheduledRecurringTask` to improve readability and fix formatting issues. * Add `SemaphoreSlim` for concurrent task execution control in scheduled tasks Introduce `SemaphoreSlim` to manage and safeguard concurrent executions in `ScheduledRecurringTask`, `ScheduledCronTask`, and `ScheduledSpecificInstantTask`. Enhances thread safety and prevents overlapping executions. Added exception handling and proper semaphore release to ensure robustness. * Add ILogger to scheduled tasks and improve error logging Integrated `ILogger` into `ScheduledRecurringTask` to enhance logging capabilities and replaced the generic comment-based error handling with proper logging for better traceability. Cleaned up and formatted `ScheduledCronTask` for improved code readability. * Dispose of semaphore fields in scheduled task classes to ensure proper resource cleanup. * Update src/modules/Elsa.Scheduling/ScheduledTasks/ScheduledRecurringTask.cs Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Set `_executing` to `false` in task `finally` blocks to ensure state reset after execution. * Update src/modules/Elsa.Scheduling/ScheduledTasks/ScheduledSpecificInstantTask.cs Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Remove redundant `_executing` assignment after `SendAsync` in scheduled task classes. * Update src/modules/Elsa.Scheduling/ScheduledTasks/ScheduledRecurringTask.cs Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Update src/modules/Elsa.Scheduling/ScheduledTasks/ScheduledRecurringTask.cs Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Update src/modules/Elsa.Scheduling/ScheduledTasks/ScheduledCronTask.cs Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Update src/modules/Elsa.Scheduling/ScheduledTasks/ScheduledSpecificInstantTask.cs Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Update semaphore logic to prevent blocked tasks when cancellation is requested Refactored `_executionSemaphore.WaitAsync` usage in `ScheduledRecurringTask`, `ScheduledCronTask`, and `ScheduledSpecificInstantTask` to use non-blocking semaphore acquisition with cancellation token support. This ensures graceful handling of pending tasks during cancellation scenarios. * Refactor delay condition checks to use `TimeSpan.Zero` for improved readability and precision. --------- Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- .../ScheduledTasks/ScheduledCronTask.cs | 54 ++++++++++++++-- .../ScheduledTasks/ScheduledRecurringTask.cs | 61 +++++++++++++++---- .../ScheduledSpecificInstantTask.cs | 44 ++++++++++--- 3 files changed, 134 insertions(+), 25 deletions(-) diff --git a/src/modules/Elsa.Scheduling/ScheduledTasks/ScheduledCronTask.cs b/src/modules/Elsa.Scheduling/ScheduledTasks/ScheduledCronTask.cs index 72b87444c..2b61acc78 100644 --- a/src/modules/Elsa.Scheduling/ScheduledTasks/ScheduledCronTask.cs +++ b/src/modules/Elsa.Scheduling/ScheduledTasks/ScheduledCronTask.cs @@ -20,6 +20,9 @@ public class ScheduledCronTask : IScheduledTask, IDisposable private readonly ICronParser _cronParser; private readonly IServiceScopeFactory _scopeFactory; private readonly CancellationTokenSource _cancellationTokenSource; + private readonly SemaphoreSlim _executionSemaphore = new(1, 1); + private bool _executing; + private bool _cancellationRequested; /// /// Initializes a new instance of . @@ -32,7 +35,7 @@ public class ScheduledCronTask : IScheduledTask, IDisposable _scopeFactory = scopeFactory; _systemClock = systemClock; _logger = logger; - _cancellationTokenSource = new CancellationTokenSource(); + _cancellationTokenSource = new(); Schedule(); } @@ -41,6 +44,13 @@ public class ScheduledCronTask : IScheduledTask, IDisposable public void Cancel() { _timer?.Dispose(); + + if (_executing) + { + _cancellationRequested = true; + return; + } + _cancellationTokenSource.Cancel(); } @@ -54,7 +64,7 @@ public class ScheduledCronTask : IScheduledTask, IDisposable var nextOccurence = _cronParser.GetNextOccurrence(_cronExpression); var delay = nextOccurence - now; - if (!adjusted && delay.Milliseconds <= 0) + if (!adjusted && delay <= TimeSpan.Zero) { adjusted = true; continue; @@ -67,7 +77,7 @@ public class ScheduledCronTask : IScheduledTask, IDisposable private void TrySetupTimer(TimeSpan delay) { - if (delay.Milliseconds <= 0) + if (delay <= TimeSpan.Zero) return; try @@ -82,7 +92,10 @@ public class ScheduledCronTask : IScheduledTask, IDisposable private void SetupTimer(TimeSpan delay) { - _timer = new Timer(delay.TotalMilliseconds) { Enabled = true }; + _timer = new(delay.TotalMilliseconds) + { + Enabled = true + }; _timer.Elapsed += async (_, _) => { @@ -93,8 +106,36 @@ public class ScheduledCronTask : IScheduledTask, IDisposable var commandSender = scope.ServiceProvider.GetRequiredService(); var cancellationToken = _cancellationTokenSource.Token; - if (!cancellationToken.IsCancellationRequested) await commandSender.SendAsync(new RunScheduledTask(_task), cancellationToken); - if (!cancellationToken.IsCancellationRequested) Schedule(); + + if (!cancellationToken.IsCancellationRequested) + { + try + { + var acquired = await _executionSemaphore.WaitAsync(0, cancellationToken); + if (!acquired) return; + + _executing = true; + await commandSender.SendAsync(new RunScheduledTask(_task), cancellationToken); + + if (_cancellationRequested) + { + _cancellationRequested = false; + _cancellationTokenSource.Cancel(); + } + } + catch (Exception e) + { + _logger.LogError(e, "Error executing scheduled task"); + } + finally + { + _executing = false; + _executionSemaphore.Release(); + } + } + + if (!cancellationToken.IsCancellationRequested) + Schedule(); }; } @@ -102,5 +143,6 @@ public class ScheduledCronTask : IScheduledTask, IDisposable { _timer?.Dispose(); _cancellationTokenSource.Dispose(); + _executionSemaphore.Dispose(); } } \ No newline at end of file diff --git a/src/modules/Elsa.Scheduling/ScheduledTasks/ScheduledRecurringTask.cs b/src/modules/Elsa.Scheduling/ScheduledTasks/ScheduledRecurringTask.cs index 21d7fd380..89a75575d 100644 --- a/src/modules/Elsa.Scheduling/ScheduledTasks/ScheduledRecurringTask.cs +++ b/src/modules/Elsa.Scheduling/ScheduledTasks/ScheduledRecurringTask.cs @@ -2,6 +2,7 @@ using Elsa.Common; using Elsa.Mediator.Contracts; using Elsa.Scheduling.Commands; using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; using Timer = System.Timers.Timer; namespace Elsa.Scheduling.ScheduledTasks; @@ -14,24 +15,24 @@ public class ScheduledRecurringTask : IScheduledTask, IDisposable private readonly ITask _task; private readonly ISystemClock _systemClock; private readonly IServiceScopeFactory _scopeFactory; + private readonly ILogger _logger; private readonly TimeSpan _interval; private readonly CancellationTokenSource _cancellationTokenSource; + private readonly SemaphoreSlim _executionSemaphore = new(1, 1); private DateTimeOffset _startAt; private Timer? _timer; + private bool _executing; + private bool _cancellationRequested; /// /// Initializes a new instance of . /// - /// The task to execute. - /// The instant at which to start executing the task. - /// The interval at which to execute the task. - /// The system clock. - /// Scope factory to create the scope and get dependancies. - public ScheduledRecurringTask(ITask task, DateTimeOffset startAt, TimeSpan interval, ISystemClock systemClock, IServiceScopeFactory scopeFactory) + public ScheduledRecurringTask(ITask task, DateTimeOffset startAt, TimeSpan interval, ISystemClock systemClock, IServiceScopeFactory scopeFactory, ILogger logger) { _task = task; _systemClock = systemClock; _scopeFactory = scopeFactory; + _logger = logger; _startAt = startAt; _interval = interval; _cancellationTokenSource = new(); @@ -43,6 +44,13 @@ public class ScheduledRecurringTask : IScheduledTask, IDisposable public void Cancel() { _timer?.Dispose(); + + if (_executing) + { + _cancellationRequested = true; + return; + } + _cancellationTokenSource.Cancel(); } @@ -56,7 +64,7 @@ public class ScheduledRecurringTask : IScheduledTask, IDisposable var now = _systemClock.UtcNow; var delay = startAt - now; - if (!adjusted && delay.Milliseconds <= 0) + if (!adjusted && delay <= TimeSpan.Zero) { adjusted = true; continue; @@ -71,7 +79,10 @@ public class ScheduledRecurringTask : IScheduledTask, IDisposable { if (delay < TimeSpan.Zero) delay = TimeSpan.FromSeconds(1); - _timer = new(delay.TotalMilliseconds) { Enabled = true }; + _timer = new(delay.TotalMilliseconds) + { + Enabled = true + }; _timer.Elapsed += async (_, _) => { @@ -81,10 +92,35 @@ public class ScheduledRecurringTask : IScheduledTask, IDisposable using var scope = _scopeFactory.CreateScope(); var commandSender = scope.ServiceProvider.GetRequiredService(); - var cancellationToken = _cancellationTokenSource.Token; - if (!cancellationToken.IsCancellationRequested) await commandSender.SendAsync(new RunScheduledTask(_task), cancellationToken); - if (!cancellationToken.IsCancellationRequested) Schedule(); + if (!cancellationToken.IsCancellationRequested) + { + try + { + var acquired = await _executionSemaphore.WaitAsync(0, cancellationToken); + if (!acquired) return; + _executing = true; + await commandSender.SendAsync(new RunScheduledTask(_task), cancellationToken); + + if (_cancellationRequested) + { + _cancellationRequested = false; + _cancellationTokenSource.Cancel(); + } + } + catch (Exception e) + { + _logger.LogError(e, "Error executing scheduled task"); + } + finally + { + _executing = false; + _executionSemaphore.Release(); + } + } + + if (!cancellationToken.IsCancellationRequested) + Schedule(); }; } @@ -92,5 +128,6 @@ public class ScheduledRecurringTask : IScheduledTask, IDisposable { _cancellationTokenSource.Dispose(); _timer?.Dispose(); + _executionSemaphore.Dispose(); } -} +} \ No newline at end of file diff --git a/src/modules/Elsa.Scheduling/ScheduledTasks/ScheduledSpecificInstantTask.cs b/src/modules/Elsa.Scheduling/ScheduledTasks/ScheduledSpecificInstantTask.cs index 3d2e9c61e..6979dd410 100644 --- a/src/modules/Elsa.Scheduling/ScheduledTasks/ScheduledSpecificInstantTask.cs +++ b/src/modules/Elsa.Scheduling/ScheduledTasks/ScheduledSpecificInstantTask.cs @@ -18,7 +18,10 @@ public class ScheduledSpecificInstantTask : IScheduledTask, IDisposable private readonly ILogger _logger; private readonly DateTimeOffset _startAt; private readonly CancellationTokenSource _cancellationTokenSource; + private readonly SemaphoreSlim _executionSemaphore = new(1, 1); private Timer? _timer; + private bool _executing; + private bool _cancellationRequested; /// /// Initializes a new instance of . @@ -30,23 +33,37 @@ public class ScheduledSpecificInstantTask : IScheduledTask, IDisposable _scopeFactory = scopeFactory; _logger = logger; _startAt = startAt; - _cancellationTokenSource = new CancellationTokenSource(); + _cancellationTokenSource = new(); Schedule(); } /// - public void Cancel() => _timer?.Dispose(); + public void Cancel() + { + _timer?.Dispose(); + + if (_executing) + { + _cancellationRequested = true; + return; + } + + _cancellationTokenSource.Cancel(); + } private void Schedule() { var now = _systemClock.UtcNow; var delay = _startAt - now; - if (delay.Milliseconds <= 0) + if (delay <= TimeSpan.Zero) delay = TimeSpan.FromMilliseconds(1); - _timer = new Timer(delay.TotalMilliseconds) { Enabled = true }; + _timer = new(delay.TotalMilliseconds) + { + Enabled = true + }; _timer.Elapsed += async (_, _) => { @@ -55,19 +72,31 @@ public class ScheduledSpecificInstantTask : IScheduledTask, IDisposable using var scope = _scopeFactory.CreateScope(); var commandSender = scope.ServiceProvider.GetRequiredService(); - var cancellationToken = _cancellationTokenSource.Token; if (!cancellationToken.IsCancellationRequested) { try { + var acquired = await _executionSemaphore.WaitAsync(0, cancellationToken); + if (!acquired) return; + _executing = true; await commandSender.SendAsync(new RunScheduledTask(_task), cancellationToken); + + if (_cancellationRequested) + { + _cancellationRequested = false; + _cancellationTokenSource.Cancel(); + } } catch (Exception e) { - _logger.LogError(e, "Error scheduled task"); + _logger.LogError(e, "Error executing scheduled task"); + } + finally + { + _executing = false; + _executionSemaphore.Release(); } - } }; } @@ -76,5 +105,6 @@ public class ScheduledSpecificInstantTask : IScheduledTask, IDisposable { _cancellationTokenSource.Dispose(); _timer?.Dispose(); + _executionSemaphore.Dispose(); } } \ No newline at end of file