1
\$\begingroup\$

To overview async tasks as "Loading configuration", "Saving configuration" and "Discarding configuration" I created a TaskQueue type.

I want to be able to schedule tasks which can be canceled an such which can't. The "Saving configuration" and "Discarding configuration" for example cannot be canceled but the "Loading configuration" can be.

I have the use case that the "Loading configuration" task can be triggered while the previous "Loading configuration" task is still running. In that case the cancable taks have to halted and the new one should start pretty much immediately.

I also give the possiblity to await until the enqueued task is done.

Now I wanted to know if my implementation is good or can be improved? Also if other types exist which solve the same problem (better or worse) as maybe the TaskScheduler? How could I solve this with the other types.

To synchronise my taks I'm using the SemaphoreSlim type. Is this a good choice? Are there alternatives? If so which and what are the up and downsides?

Is the Queue type good for the purpose I'm using it for. Are there alternatives?

What would you have done different to solve this problem?

public class TaskQueue
{
    public bool IsRunning => _longRunningTasks.Count > 0 || _semaphoreSlim.CurrentCount != _maxConcurentActions;

    public event EventHandler<bool> IsRunningChanged;

    private bool _oldIsRunningValue;

    private readonly int _maxConcurentActions;
    private readonly Queue<LongRunningTask> _longRunningTasks;
    private readonly SemaphoreSlim _semaphoreSlim;

    public TaskQueue(int maxConcurentActions)
    {
        _maxConcurentActions = maxConcurentActions;
        _semaphoreSlim = new SemaphoreSlim(maxConcurentActions);
        _longRunningTasks = new Queue<LongRunningTask>();
    }

    public async void Enqueue(Action action)
        => await EnqueueAsync(action);
    public async void Enqueue(Action action, bool canCancel)
        => await EnqueueAsync(action, canCancel);
    public async void Enqueue(Action<CancellationToken> action)
        => await EnqueueAsync(action);
    public async void Enqueue(Action<CancellationToken> action, bool canCancel)
        => await EnqueueAsync(action, canCancel);
    public async void Enqueue(Func<CancellationToken, Task> action)
        => await EnqueueAsync(action);
    public async void Enqueue(Func<CancellationToken, Task> action, bool canCancel)
        => await EnqueueAsync(action, canCancel);
    public async void Enqueue(Func<Task> action)
        => await EnqueueAsync(action);
    public async void Enqueue(Func<Task> action, bool canCancel)
        => await EnqueueAsync(action, canCancel);
    public async Task EnqueueAsync(Action action)
        => await EnqueueAsync(action, true);
    public async Task EnqueueAsync(Action action, bool canCancel)
        => await EnqueueAsync(cancellationToken => action(), canCancel);
    public async Task EnqueueAsync(Action<CancellationToken> action)
        => await EnqueueAsync(action, true);
    public async Task EnqueueAsync(Action<CancellationToken> action, bool canCancel)
        => await EnqueueAsync(async cancellationToken =>
        {
            action(cancellationToken);
            await Task.CompletedTask;
        }, canCancel);
    public async Task EnqueueAsync(Func<Task> action)
        => await EnqueueAsync(action, true);
    public async Task EnqueueAsync(Func<Task> action, bool canCancel)
        => await EnqueueAsync(async cancelationToken => await action(), canCancel);
    public async Task EnqueueAsync(Func<CancellationToken, Task> action)
        => await EnqueueAsync(action, true);
    public async Task EnqueueAsync(Func<CancellationToken, Task> action, bool canCancel)
    {
        var id = Guid.NewGuid();
        _longRunningTasks.Enqueue(new LongRunningTask
        {
            Id = id,
            CanCancel = canCancel,
            CancellationTokenSource = new CancellationTokenSource(),
            Action = action
        });

        if (_oldIsRunningValue != true)
        {
            _oldIsRunningValue = true;
            IsRunningChanged?.Invoke(this, true);
        }

        var nextId = Guid.Empty;
        while (nextId != id)
        {
            await _semaphoreSlim.WaitAsync();
            nextId = _longRunningTasks.Peek().Id;
        }

        var myTask = _longRunningTasks.Dequeue();
        if (!myTask.CancellationTokenSource.IsCancellationRequested)
        {
            await myTask.Action(myTask.CancellationTokenSource.Token);
        }
        _semaphoreSlim.Release();

        if (!IsRunning && _oldIsRunningValue)
        {
            _oldIsRunningValue = false;
            IsRunningChanged?.Invoke(this, false);
        }
    }

    public void CancleAllCancelables()
    {
        foreach (var item in _longRunningTasks)
        {
            if (item.CanCancel)
            {
                item.CancellationTokenSource.Cancel();
            }
        }
    }

    private class LongRunningTask
    {
        public Guid Id { get; set; }
        public bool CanCancel { get; set; }
        public CancellationTokenSource CancellationTokenSource { get; set; }
        public Func<CancellationToken, Task> Action { get; set; }
    }
}

The TaskQueue type can for example be used like the following.

var taskQueue = new TaskQueue(1);

taskQueue.IsRunningChanged += (sender, b) => Console.WriteLine($"{GetCurrentTime()} - IsRunning changed to `{b}`");

taskQueue.Enqueue(async cancellationToken => await WriteAndWait(cancellationToken, 1));
taskQueue.Enqueue(async cancellationToken => await WriteAndWait(cancellationToken, 2));

taskQueue.Enqueue(() =>
{
    Console.WriteLine($"{GetCurrentTime()} - Task numer `{7}` started...");
    taskQueue.CancleAllCancelables();

    taskQueue.Enqueue(async cancellationTokenInner => await WriteAndWait(cancellationTokenInner, 5));
    taskQueue.Enqueue(async cancellationTokenInner => await WriteAndWait(cancellationTokenInner, 6));

    Console.WriteLine($"{GetCurrentTime()} - Task numer `{7}` stoped");
});

taskQueue.Enqueue(async cancellationToken => await WriteAndWait(cancellationToken, 3));
taskQueue.Enqueue(async cancellationToken => await WriteAndWait(cancellationToken, 4));

while (taskQueue.IsRunning)
{
    Console.WriteLine($"{GetCurrentTime()} - Waiting...");
    Thread.Sleep(200);
}

Console.WriteLine($"{GetCurrentTime()} - Done");

private static async Task WriteAndWait(CancellationToken cancellationToken, int taskNumber)
{
    Console.WriteLine($"{GetCurrentTime()} - Task numer `{taskNumber}` started...");
    await Task.Delay(3000, cancellationToken);
    Console.WriteLine($"{GetCurrentTime()} - Task numer `{taskNumber}` stoped");
}

private static string GetCurrentTime()
{
    var dateTime = DateTime.Now;
    return $"{dateTime.Hour}:{dateTime.Minute}:{dateTime.Second}.{dateTime.Millisecond}";
}
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.