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

Add cancellation handling for executing scheduled tasks (#6819)

Browse source 
* 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>
This commit is contained in:
Sipke Schoorstra 2025-07-25 07:46:35 +02:00 committed by GitHub
parent 9b1a76d047
commit 3fd8a252cd
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 134 additions and 25 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

54
src/modules/Elsa.Scheduling/ScheduledTasks/ScheduledCronTask.cs
View file

@ -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;
/// <summary>
/// Initializes a new instance of <see cref="ScheduledCronTask"/>.
@ -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<ICommandSender>();
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();
}
}

61
src/modules/Elsa.Scheduling/ScheduledTasks/ScheduledRecurringTask.cs
View file

@ -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<ScheduledRecurringTask> _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;
/// <summary>
/// Initializes a new instance of <see cref="ScheduledRecurringTask"/>.
/// </summary>
/// <param name="task">The task to execute.</param>
/// <param name="startAt">The instant at which to start executing the task.</param>
/// <param name="interval">The interval at which to execute the task.</param>
/// <param name="systemClock">The system clock.</param>
/// <param name="scopeFactory">Scope factory to create the scope and get dependancies.</param>
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<ScheduledRecurringTask> 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<ICommandSender>();
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();
}
}
}

44
src/modules/Elsa.Scheduling/ScheduledTasks/ScheduledSpecificInstantTask.cs
View file

@ -18,7 +18,10 @@ public class ScheduledSpecificInstantTask : IScheduledTask, IDisposable
private readonly ILogger<ScheduledSpecificInstantTask> _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;
/// <summary>
/// Initializes a new instance of <see cref="ScheduledSpecificInstantTask"/>.
@ -30,23 +33,37 @@ public class ScheduledSpecificInstantTask : IScheduledTask, IDisposable
_scopeFactory = scopeFactory;
_logger = logger;
_startAt = startAt;
_cancellationTokenSource = new CancellationTokenSource();
_cancellationTokenSource = new();
Schedule();
}
/// <inheritdoc />
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<ICommandSender>();
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();
}
}