WPF async ObservableTaskQueue class

Question

I'm currently working on a WPF application, making calls to various web APIs. Thus, I'm looking after a special observable tasks queue:

• Progress value must be available
• Tasks (API calls)/seconds must be limited
• Failed tasks must be restarted (API calls may unexpectedly raise exceptions)

---

Since the API calls are not very CPU intensive (mostly waiting for web responses), Based on Servy's answer, I kept it simple by using async/await rather than using threads, but I may be wrong.

I came up with the following class, that seems to work well with WPF. But could you tell me what you think about this simple implementation? Any advice about improvements or errors I might have done is welcome!

public class ObservableTaskQueue : INotifyPropertyChanged
{
    //========================================================================================================
    // Events
    //========================================================================================================

    /// <summary>
    /// Occurs when the first task is enqueued.
    /// </summary>
    public event Action StartWorking;

    /// <summary>
    /// Occurs when the last enqueued task is done.
    /// </summary>
    public event Action StopWorking;


    //========================================================================================================
    // Fields
    //========================================================================================================

    private SemaphoreSlim semaphore;

    private Stopwatch timer = new Stopwatch();

    private int maxTasksCount = 0;

    private int queuedTaskCount = 0;



    //========================================================================================================
    // Properties
    //========================================================================================================


    /// <summary>
    /// Get or set the minimum duration between two task starts (in milliseconds).
    /// </summary>
    public int MinimumTaskDuration { get; set; }

    /// <summary>
    /// Get or set whether a task needs to be restarted after raising an exception.
    /// </summary>
    public bool RestartTaskOnError { get; set; }

    /// <summary>
    /// Get the task queue's progress (0-100).
    /// </summary>
    public int Progress {
        get
        {
            return (maxTasksCount == 0) ? 0 : (maxTasksCount - queuedTaskCount) * 100 / maxTasksCount;
        }
    }



    //========================================================================================================
    // Constructors
    //========================================================================================================

    public TaskQueue()
    {
        semaphore = new SemaphoreSlim(1);
    }



    //========================================================================================================
    // Methods
    //========================================================================================================

    public async void Enqueue(Func<Task> taskCallback)
    {
        // Raise event when the queue gets busy.
        if (Interlocked.Increment(ref queuedTaskCount) == 1)
        {
            Interlocked.Exchange(ref maxTasksCount, 0);

            if (StartWorking != null)
            {
                StartWorking();
            }
        }

        // Update queue's progress.
        Interlocked.Increment(ref maxTasksCount);
        NotifyPropertyChanged("Progress");

        // Only one task can be running at one time.
        await semaphore.WaitAsync();

        // If a maximum task rate is set, wait a bit if the previous task finished too fast.
        if (MinimumTaskDuration > 0 && timer.IsRunning)
        {
            timer.Stop();

            if (timer.ElapsedMilliseconds < MinimumTaskDuration)
            {
                Console.WriteLine("  Waiting {0}ms", MinimumTaskDuration - (int)timer.ElapsedMilliseconds);
                await Task.Delay(MinimumTaskDuration - (int)timer.ElapsedMilliseconds + 1);
            }
        }

        // Start measuring task's duration and run the task.
        try
        {
            timer.Restart();
            await taskCallback();
        }

        // Restart failed tasks on error.
        catch
        {
            if (RestartTaskOnError)
            {
                Interlocked.Decrement(ref maxTasksCount); // don't notify progress yet

                Console.WriteLine("Error, requeue the same task");
                Enqueue(taskCallback);
            }
        }

        finally
        {
            Console.WriteLine("  Ended in {0}ms", timer.ElapsedMilliseconds);

            // Raise event when the queue gets idle.
            if (Interlocked.Decrement(ref queuedTaskCount) == 0)
            {
                timer.Stop();

                if (StopWorking != null)
                {
                    StopWorking();
                }
            }

            NotifyPropertyChanged("Progress");

            semaphore.Release();
        }
    }



    //========================================================================================================
    // INotifyPropertyChanged implementation
    //========================================================================================================

    public event PropertyChangedEventHandler PropertyChanged;

    private void NotifyPropertyChanged(String propertyName)
    {
        if (PropertyChanged != null)
        {
            PropertyChanged(this, new PropertyChangedEventArgs(propertyName));
        }
    }


}

Usage :

public MainWindow()
{
    InitializeComponent();

    TaskQueue queue = new TaskQueue();
    queue.MinimumTaskDuration = 300;
    queue.RestartTaskOnError = true;
    queue.StartWorking += delegate() { Console.WriteLine("START WORKING"); };
    queue.StopWorking += delegate() { Console.WriteLine("STOP WORKING"); };

    Console.WriteLine("Enqueue (Task-1)");
    Queue.Enqueue(Task1);

    Console.WriteLine("Enqueue (Task-2)");
    Queue.Enqueue(Task2);

    Console.WriteLine("Enqueue (Task-3)");
    Queue.Enqueue(Task3);
}


bool restart = true;

private async Task Task1()
{
    Console.WriteLine("(Task-1) started");
    await Task.Delay(50);
    Console.WriteLine("  (Task-1) completed");
}

private async Task Task2()
{
    Console.WriteLine("(Task-2) started");
    await Task.Delay(100);

    if (restart) // fails on the first try!
    {
        restart = false;
        throw new Exception();
    }

    Console.WriteLine("  (Task-2) completed");
}

private async Task Task3()
{
    Console.WriteLine("(Task-3) started");
    await Task.Delay(85);
    Console.WriteLine("  (Task-3) completed");
}

Result:

Enqueue (Task-1)
[START WORKING]
(Task-1) started
Enqueue (Task-2)
Enqueue (Task-3)
  (Task-1) completed
  Ended in 60ms
  Waiting 239ms
(Task-2) started
A first chance exception of type 'System.Exception' occured in OkSEO SearchConsole.exe
A first chance exception of type 'System.Exception' occured in mscorlib.dll
Error occured, requeue the same task
  Ended in 133ms
  Waiting 167ms
(Task-3) started
  (Task-3) completed
  Ended in 86ms
  Waiting 213ms
(Task-2) started
  (Task-2) completed
  Ended in 114ms
[STOP WORKING]

---

Known bugs/things to do:

• The class is lacking a generic Enqueue<T> method.
• Task delay is not respected between two separate queue processings (if a new task is added immediately after the triggering of StopWorking event).
• maxTasksCount and queuedTaskCount might be public observable properties.

Answer 1

So, there's a lot going on here. First things first, this class is really doing too much. It's running a queue of tasks one after the other, it's handling retrying errors, it's adding a delay between tasks, it's got business logic of what "progress" means that seems rather separate from the idea of a queue of tasks, etc. The code needs to be broken up a bit.

So to break it up we'll create a queue that executes arbitrary tasks with a single degree of parallelism, and that can retry the tasks and add a delay between executions. This class won't need to know about reporting progress, and it won't have a notion of starting or stopping.

public class TaskQueue
{
    private SemaphoreSlim semaphore = new SemaphoreSlim(1);
    public TimeSpan TimeBetweenTaskExecutions { get; set; }
    public int NumberOfRetries { get; set; }

    public async Task<T> Enqueue<T>(Func<Task<T>> taskGenerator)
    {
        await semaphore.WaitAsync();

        int numberOfTriesRemaining = NumberOfRetries + 1;
        Task delay = null;
        try
        {
            Func<Task<T>> wrappedGenerator = () =>
            {
                delay = Task.Delay(TimeBetweenTaskExecutions);
                return taskGenerator();
            };
            return await TaskUtilities.RetryOnFailure(wrappedGenerator, NumberOfRetries, TimeBetweenTaskExecutions)
                .ConfigureAwait(false);
        }
        finally
        {
            ReleaseAfterDelay(delay);
        }
    }

    public Task Enqueue(Func<Task> taskGenerator)
    {
        return Enqueue(() => TaskUtilities.WithResult(taskGenerator(), true));
    }

    private async void ReleaseAfterDelay(Task delay)
    {
        try
        {
            await delay
                    .ConfigureAwait(false);
        }
        finally
        {
            semaphore.Release();
        }
    }
}

public static class TaskUtilities
{
    public static async Task<T> WithResult<T>(Task task, T value)
    {
        await task.ConfigureAwait(false);
        return value;
    }

    public static async Task<T> RetryOnFailure<T>(Func<Task<T>> taskGenerator,
        int numberOfRetries = 1, TimeSpan? timeBetweenExecutions = null)
    {
        int numberOfTriesRemaining = numberOfRetries + 1;
        while (true)
        {
            var delayTask = Task.Delay(timeBetweenExecutions ?? TimeSpan.Zero);
            try
            {
                return await taskGenerator();
            }
            catch
            {
                numberOfTriesRemaining--;
                if (numberOfTriesRemaining == 0)
                    throw;
                await delayTask;
            }
        }
    }
}

So there are a number of things going on here that are different from how you approached these aspects of your problem. First off, the delay. TimeSpan is your friend here. It means you don't need to think about units, and it has lots of useful operations you can perform on it. You can pull out a given unit of measurement (say, milliseconds) when displaying the timespan to the user, although that won't be relevant here. When the caller provides one they can also use whatever measurement they want, using TimeSpan.FromSeconds, TimeSpan.FromMilliseconds, etc. Additionally, we don't use a Stopwatch to keep track of the delay, we use Timespan.Delay, and simply don't release the semaphore until that has finished. This is not only much simpler to read, and a lot less code, but removes the bug you noted in your own code of newly enqueued items not observing the delay properly.

Of note here, we want the Task returned from Enqueue to finish as soon as the generated task has finished and not after they delay between tasks. Because of this, we need to make sure that we don't release the semaphore until after the delay while letting the async method complete without waiting on that. We do this here with the use of an async void method, since in this case this behavior really should be fire and forget. We want to ensure that the semaphore is released after the delay, but we don't want the Enqueue method to wait on it.

Next we have retries. While recursing to handle retries is simple in terms of code, it has a number of problems. It only really works if you retry forever. Normally this isn't desirable; typically you want to retry a fixed number of times before just giving up, and recursing as you do doesn't easily allow for that to be added. Additionally, it results in an item being run, then potentially other items being run before it is retried. Perhaps that was by design, but it's rather atypical behavior. Generally you'd want to retry the failing task without moving to others until it succeeds or you're done with it. Finally, because you're dropping the Task returned by Enqueue on the floor, it means that you don't have any good way of exposing the completion of that operation to the caller, which isn't great; in your code in particular the async method finishes, implying the completion of the operation, before the operation has actually completed.

Given all that, I refactored that into a loop that it breaks out of by either finishing successfully or running out of retry attempts and throwing. I've also separated it out into its own method, to separate the mechanism of retrying a Task from the rest of your queue. Additionally, since it seems like the kind of thing that'd be fairly widely useful, I have it in a separate class exposed publicly rather than as a private method.

Exposing the result of the operation, as you noted that your solution doesn't do, is done by simply returning the result of awaiting the task in question and ensuring all work to be done when the operation completes (successfully or unsuccessfully) is in a finally. Your own code was actually not far off from doing this, if it weren't for your solution of using recursion to handle retrying on failure (you couldn't await the recursive call without deadlocking, so it'd need a bit more refactoring to avoid that).

As for the non-result Task overload, I cheated a bit. Performance wise, it'd be best to copy-paste the method and just return nothing rather than the result of the Task. In order to avoid the code repetition I'm simply wrapping the task into one that has a dummy result so I can call the other overload. This adds the cost of having the wrapper (small, but not nothing) with the benefit of not needing to replicate the logic of the code.

Now we're finally ready for your observable queue. There's not a ton left to do here (by design) given that we've already done all of the hard work in your separate queue. It just needs to hold onto an instance of it, expose the members of it as needed, and then just fire off the events as each task is added/removed from the queue. I renamed some of the variables you used, to ones that I felt better explained what they represented, but the logic here isn't particularly different from your own, other than that it's separated from all of the task scheduling messiness, and with a few fairly minor changes here and there that I would admittedly consider mostly personal preference.

I think the only real bug in this event code is that your event firing code isn't thread safe; removing the last remaining handler after the if check would break your code.

public class ObservableTaskQueue : INotifyPropertyChanged
{
    public event Action StartWorking;
    public event Action StopWorking;
    public event PropertyChangedEventHandler PropertyChanged;

    private TaskQueue queue = new TaskQueue();
    private int tasksQueuedSinceQueueWasLastIdle = 0;
    private int pendingTasks = 0;
    public int Progress
    {
        get
        {
            if (tasksQueuedSinceQueueWasLastIdle == 0)
                return 0;

            int numberOfCompletedTasks = tasksQueuedSinceQueueWasLastIdle - pendingTasks;
            return numberOfCompletedTasks * 100 / tasksQueuedSinceQueueWasLastIdle;
        }
    }
    public TimeSpan TimeBetweenTaskExecutions
    {
        get { return queue.TimeBetweenTaskExecutions; }
        set { queue.TimeBetweenTaskExecutions = value; }
    }
    public int NumberOfRetries
    {
        get { return queue.NumberOfRetries; }
        set { queue.NumberOfRetries = value; }
    }

    public async Task<T> Enqueue<T>(Func<Task<T>> taskGenerator)
    {
        Interlocked.Increment(ref tasksQueuedSinceQueueWasLastIdle);
        if (Interlocked.Increment(ref pendingTasks) == 1)
            StartWorking?.Invoke();
        NotifyPropertyChanged("Progress");
        try
        {
            return await queue.Enqueue(taskGenerator).ConfigureAwait(false);
        }
        finally
        {
            NotifyPropertyChanged("Progress");
            if (Interlocked.Decrement(ref pendingTasks) == 0)
            {
                Interlocked.Exchange(ref tasksQueuedSinceQueueWasLastIdle, 0);
                StopWorking?.Invoke();
            }
        }
    }

    public Task Enqueue(Func<Task> taskGenerator)
    {
        return Enqueue(() => TaskUtilities.WithResult(taskGenerator(), true));
    }

    private void NotifyPropertyChanged(String propertyName)
    {
        PropertyChanged?.Invoke(this, new PropertyChangedEventArgs(propertyName));
    }
}

So, one final point to raise. ConfigureAwait. So, normally, as this is API code, and nothing here really needs to run on whatever the current synchronization context is, we'd just ConfigureAwait(false) everything, but we can't quite do that here. It's entirely possible that the Func<Task> that the consumer provides to us is something they want to run in the current context. Because of this, we can't ConfigureAwait(false) any code that has a code path that will later go on to invoke one of those delegates. That does still leave us with a fair few places that we can safely ConfigureAwait(false) in order to not have to wait to get back into a context we don't need.