2
\$\begingroup\$

I have a class, 'TaskCoordinator' which has 'BaseTask's applied to it.

The BaseTasks have OnStarted, OnFinished and OnCancelled events which are triggered to notify the calling class of their events.

GroupTasks consist of a collection of BaseTasks to be executed one after another or all at the same time.

Tasks which derive from PropertyTransformation apply to a .Net Property and as such I needed a way to ensure that there are never more then one task targeting the same property, to achieve this I override GetHashCode to return the hash of the object + property name and use it to determine if there are any collisions when applying a task.

Below is the class in question.

public class TaskCoordinator : IDisposable
{
    private readonly CancellationTokenSource _cancellationToken;
    private readonly ConcurrentDictionary<BaseTask, ConcurrentQueue<BaseTask>> _tasks;

    /// <summary>
    ///     Orchestrate the order of execution of tasks
    /// </summary>
    public TaskCoordinator()
    {
        _cancellationToken = new CancellationTokenSource();
        _tasks = new ConcurrentDictionary<BaseTask, ConcurrentQueue<BaseTask>>();
    }

    /// <summary>
    ///     Clean up by cancelling all pending & running tasks
    /// </summary>
    public void Dispose()
    {
        _cancellationToken.Cancel();
    }

    /// <summary>
    ///     Add a task to be executed
    /// </summary>
    /// <param name="action"></param>
    public void Apply(BaseTask task)
    {
        if (task == null)
        {
            return;
        }

        TryToStartTask(task);
    }

    private void TryToStartTask(BaseTask task)
    {
        if (_tasks.TryGetValue(task, out var taskQueue))
        {
            if (taskQueue == null)
            {
                taskQueue = new ConcurrentQueue<BaseTask>();
                _tasks.TryAdd(task, taskQueue);
            }

            taskQueue.Enqueue(task);
        }
        else
        {
            StartTask(task).ConfigureAwait(false);
        }
    }

    private async Task StartTask(BaseTask task)
    {
        task.RaiseOnStarted(EventArgs.Empty);
        try
        {
            if (task is ParallelGroupTask parallelGroupTask)
            {
                await RunParallel(parallelGroupTask).ConfigureAwait(false);
            }
            else if (task is SequentialGroupTask sequentialGroupTask)
            {
                await RunSequential(sequentialGroupTask).ConfigureAwait(false);
            }
            else
            {
                await Run(task).ConfigureAwait(false);
            }

            task.RaiseOnFinished(EventArgs.Empty);
            RunNextTask(task);
        }
        catch (TaskCanceledException)
        {
            task.RaiseOnCancelled();
        }
    }

    private async Task Run(BaseTask task)
    {
        await task.StartAsync(_cancellationToken).ConfigureAwait(false);
    }

    private async Task RunParallel(ParallelGroupTask task)
    {
        if (task.Tasks.Count == 0)
        {
            return;
        }

        bool isRunning = true;

        int finishedTaskCount = 0;
        foreach (var taskTask in task.Tasks)
        {
            taskTask.OnFinished += (s, e) => {
                if (++finishedTaskCount >= task.Tasks.Count)
                {
                    isRunning = false;
                }
            };
            Apply(taskTask);
        }

        while (isRunning)
        {
            await Task.Delay(1).ConfigureAwait(false);
        }
    }


    private async Task RunSequential(SequentialGroupTask task)
    {
        if (task.Tasks.Count == 0)
        {
            return;
        }

        ConnectSequentialTasks(task.Tasks);

        var isRunning = true;
        //When the last task finishes set running to false
        task.Tasks[task.Tasks.Count - 1].OnFinished += (s, e) => { isRunning = false; };

        Apply(task.Tasks[0]);

        while (isRunning)
        {
            await Task.Delay(1).ConfigureAwait(false);
        }
    }

    private void ConnectSequentialTasks(List<BaseTask> tasks)
    {
        for (var i = 1; i < tasks.Count; i++)
        {
            var localIndex = i;
            tasks[localIndex - 1].OnFinished += (s, e) => { Apply(tasks[localIndex]); };
        }
    }

    private void RunNextTask(BaseTask completedTask)
    {
        if (_tasks.TryGetValue(completedTask, out var taskQueue) && taskQueue.TryDequeue(out var nextTask))
        {
            StartTask(nextTask).ConfigureAwait(false);
        }
        else
        {
            _tasks.TryRemove(completedTask, out _);
        }
    }
}

Here is the abstract BaseTask which all Tasks derive:

 public abstract class BaseTask
{
    protected BaseTask(TimeSpan duration)
    {
        Duration = duration;
        _cancellationToken = new CancellationTokenSource();
    }

    public TimeSpan Duration { get; set; }
    protected CancellationTokenSource _cancellationToken { get; set; }

    public event EventHandler<EventArgs> OnStarted;
    public event EventHandler<EventArgs> OnFinished;
    public event EventHandler<EventArgs> OnCancelled;

    /// <summary>
    ///     Start the task
    /// </summary>
    /// <returns></returns>
    public Task StartAsync()
    {
        return StartAsync(_cancellationToken);
    }

    /// <summary>
    ///     Start the task with the given CancellationToken
    /// </summary>
    /// <returns></returns>
    public Task StartAsync(CancellationTokenSource cancellationToken)
    {
        _cancellationToken = cancellationToken;
        if (_cancellationToken.IsCancellationRequested)
        {
            RaiseOnCancelled();
            return Task.FromCanceled(_cancellationToken.Token);
        }

        return InternalTask();
    }

    /// <summary>
    ///     Interrupt the task
    /// </summary>
    public void Cancel()
    {
        _cancellationToken.Cancel();
    }

    internal void RaiseOnStarted(EventArgs args)
    {
        OnStarted?.Invoke(this, args);
    }

    internal void RaiseOnFinished(EventArgs args)
    {
        if (_cancellationToken.IsCancellationRequested)
        {
            RaiseOnCancelled();
        }
        else
        {
            OnFinished?.Invoke(this, args);
        }
    }

    internal void RaiseOnCancelled()
    {
        OnCancelled?.Invoke(this, EventArgs.Empty);
    }

    protected abstract Task InternalTask();
}

I am quite happy with my solution in its current state as it seems to works well for my needs. Given the topics used in this class I am worried the solution may be some what 'hacked' together and a more elegant / reliable solution exists for me needs.

Reliablity and deadlock aversion are #1 priority with low CPU overhead coming in second and finally speed being third in my priority list.

For additional context here is the GitHub Repo.

https://github.com/Timmoth/AptacodeTaskPlex

\$\endgroup\$
  • \$\begingroup\$ I really don't understand the problem you're solving with this code. Can you explain the business requirement some more? \$\endgroup\$ – Enigmativity Oct 21 '19 at 9:06

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.