Limited concurrency level task scheduler (with task priority) handling wrapped tasks

Question

I'm having a hard time finding a task scheduler on which I can schedule prioritised tasks but can also handle "wrapped" tasks. It is something like what Task.Run tries to solve, but you cannot specify a task scheduler to Task.Run. I have been using a QueuedTaskScheduler from the Parallel Extensions Extras Samples to solve the task priority requirement (also suggested by this post).

Here is my example:

class Program
{
    private static QueuedTaskScheduler queueScheduler = new QueuedTaskScheduler(targetScheduler: TaskScheduler.Default, maxConcurrencyLevel: 1);
    private static TaskScheduler ts_priority1;
    private static TaskScheduler ts_priority2;
    static void Main(string[] args)
    {
        ts_priority1 = queueScheduler.ActivateNewQueue(1);
        ts_priority2 = queueScheduler.ActivateNewQueue(2);

        QueueValue(1, ts_priority2);
        QueueValue(2, ts_priority2);
        QueueValue(3, ts_priority2);
        QueueValue(4, ts_priority1);
        QueueValue(5, ts_priority1);
        QueueValue(6, ts_priority1);

        Console.ReadLine();           
    }

    private static Task QueueTask(Func<Task> f, TaskScheduler ts)
    {
        return Task.Factory.StartNew(f, CancellationToken.None, TaskCreationOptions.HideScheduler | TaskCreationOptions.DenyChildAttach, ts);
    }

    private static Task QueueValue(int i, TaskScheduler ts)
    {
        return QueueTask(async () =>
        {
            Console.WriteLine("Start {0}", i);
            await Task.Delay(1000);
            Console.WriteLine("End {0}", i);
        }, ts);
    }
}

The typical output of the example above is:

Start 4
Start 5
Start 6
Start 1
Start 2
Start 3
End 4
End 3
End 5
End 2
End 1
End 6

What I want is:

Start 4
End 4
Start 5
End 5
Start 6
End 6
Start 1
End 1
Start 2
End 2
Start 3
End 3

EDIT:

I think I'm looking for a task scheduler, similar to QueuedTaskScheduler, that will solve this problem. But any other suggestions are welcome.

Well, what you want is handle priority of tasks, but not run them in parallel mode ? could you not just limit the number of simultaneous threads in your scheduler ? — Kek, Nov 14 '12 at 12:02
@Kek `new QueuedTaskScheduler(targetScheduler: TaskScheduler.Default, maxConcurrencyLevel: 1);` above does exactly that (limit the number of simultaneous threads to 1) — Francois Nel, Nov 14 '12 at 12:52

score 4 · Answer 1 · answered Nov 14 '12 at 13:44

Unfortunately, this can't be solved with a TaskScheduler, because they always work at the Task level, and an async method almost always contains multiple Tasks.

You should use a SemaphoreSlim in conjunction with a prioritizing scheduler. Alternatively, you could use AsyncLock (which is also included in my AsyncEx library).

class Program
{
  private static QueuedTaskScheduler queueScheduler = new QueuedTaskScheduler(targetScheduler: TaskScheduler.Default, maxConcurrencyLevel: 1);
  private static TaskScheduler ts_priority1;
  private static TaskScheduler ts_priority2;
  private static SemaphoreSlim semaphore = new SemaphoreSlim(1);
  static void Main(string[] args)
  {
    ts_priority1 = queueScheduler.ActivateNewQueue(1);
    ts_priority2 = queueScheduler.ActivateNewQueue(2);

    QueueValue(1, ts_priority2);
    QueueValue(2, ts_priority2);
    QueueValue(3, ts_priority2);
    QueueValue(4, ts_priority1);
    QueueValue(5, ts_priority1);
    QueueValue(6, ts_priority1);

    Console.ReadLine();           
  }

  private static Task QueueTask(Func<Task> f, TaskScheduler ts)
  {
    return Task.Factory.StartNew(f, CancellationToken.None, TaskCreationOptions.HideScheduler | TaskCreationOptions.DenyChildAttach, ts).Unwrap();
  }

  private static Task QueueValue(int i, TaskScheduler ts)
  {
    return QueueTask(async () =>
    {
      await semaphore.WaitAsync();
      try
      {
        Console.WriteLine("Start {0}", i);
        await Task.Delay(1000);
        Console.WriteLine("End {0}", i);
      }
      finally
      {
        semaphore.Release();
      }
    }, ts);
  }
}

This looks like an interesting solution. However, I see one problem with this. Although the solution will (at first) result in the correct output (as in this question), but it will break the priority of the executed tasks. The scheduler will execute all tasks (in the correct priority) up until the `await semaphore.WaitAsync()` but tasks with higher priority will not be released from the lock before tasks of lower priority. This is especially true if higher priority tasks are scheduled after lower priority tasks (that are still waiting to be released from the lock). — Francois Nel, Nov 14 '12 at 14:53
In that case, you'll need an actual priority-based lock, which does not exist because AFAIK no one else has needed one. You'll have to build your own. — Stephen Cleary, Nov 14 '12 at 15:13
I have added my own [answer](http://stackoverflow.com/a/13414364/1514235). Please have a look and see what you think. — Francois Nel, Nov 16 '12 at 10:18

score 3 · Accepted Answer · answered Nov 16 '12 at 10:08

The best solution I could find is to make my own version of the QueuedTaskScheduler (original found in the Parallel Extensions Extras Samples source code).

I added a bool awaitWrappedTasks parameter to the constructors of the QueuedTaskScheduler.

public QueuedTaskScheduler(
        TaskScheduler targetScheduler,
        int maxConcurrencyLevel,
        bool awaitWrappedTasks = false)
{
    ...
    _awaitWrappedTasks = awaitWrappedTasks;
    ...
}

public QueuedTaskScheduler(
        int threadCount,
        string threadName = "",
        bool useForegroundThreads = false,
        ThreadPriority threadPriority = ThreadPriority.Normal,
        ApartmentState threadApartmentState = ApartmentState.MTA,
        int threadMaxStackSize = 0,
        Action threadInit = null,
        Action threadFinally = null,
        bool awaitWrappedTasks = false)
{
    ...
    _awaitWrappedTasks = awaitWrappedTasks;

    // code starting threads (removed here in example)
    ...
}

I then modified the ProcessPrioritizedAndBatchedTasks() method to be async

private async void ProcessPrioritizedAndBatchedTasks()

I then modified the code just after the part where the scheduled task is executed:

private async void ProcessPrioritizedAndBatchedTasks()
{
    bool continueProcessing = true;
    while (!_disposeCancellation.IsCancellationRequested && continueProcessing)
    {
        try
        {
            // Note that we're processing tasks on this thread
            _taskProcessingThread.Value = true;

            // Until there are no more tasks to process
            while (!_disposeCancellation.IsCancellationRequested)
            {
                // Try to get the next task.  If there aren't any more, we're done.
                Task targetTask;
                lock (_nonthreadsafeTaskQueue)
                {
                    if (_nonthreadsafeTaskQueue.Count == 0) break;
                    targetTask = _nonthreadsafeTaskQueue.Dequeue();
                }

                // If the task is null, it's a placeholder for a task in the round-robin queues.
                // Find the next one that should be processed.
                QueuedTaskSchedulerQueue queueForTargetTask = null;
                if (targetTask == null)
                {
                    lock (_queueGroups) FindNextTask_NeedsLock(out targetTask, out queueForTargetTask);
                }

                // Now if we finally have a task, run it.  If the task
                // was associated with one of the round-robin schedulers, we need to use it
                // as a thunk to execute its task.
                if (targetTask != null)
                {
                    if (queueForTargetTask != null) queueForTargetTask.ExecuteTask(targetTask);
                    else TryExecuteTask(targetTask);

                    // ***** MODIFIED CODE START ****
                    if (_awaitWrappedTasks)
                    {
                        var targetTaskType = targetTask.GetType();
                        if (targetTaskType.IsConstructedGenericType && typeof(Task).IsAssignableFrom(targetTaskType.GetGenericArguments()[0]))
                        {
                            dynamic targetTaskDynamic = targetTask;
                            // Here we await the completion of the proxy task.
                            // We do not await the proxy task directly, because that would result in that await will throw the exception of the wrapped task (if one existed)
                            // In the continuation we then simply return the value of the exception object so that the exception (stored in the proxy task) does not go totally unobserved (that could cause the process to crash)
                            await TaskExtensions.Unwrap(targetTaskDynamic).ContinueWith((Func<Task, Exception>)(t => t.Exception), TaskContinuationOptions.ExecuteSynchronously);
                        }
                    }
                    // ***** MODIFIED CODE END ****
                }
            }
        }
        finally
        {
            // Now that we think we're done, verify that there really is
            // no more work to do.  If there's not, highlight
            // that we're now less parallel than we were a moment ago.
            lock (_nonthreadsafeTaskQueue)
            {
                if (_nonthreadsafeTaskQueue.Count == 0)
                {
                    _delegatesQueuedOrRunning--;
                    continueProcessing = false;
                    _taskProcessingThread.Value = false;
                }
            }
        }
    }
}

The change of method ThreadBasedDispatchLoop was a bit different, in that we cannot use the async keyword or else we will break the functionality of executing scheduled tasks in the dedicated thread(s). So here is the modified version of ThreadBasedDispatchLoop

private void ThreadBasedDispatchLoop(Action threadInit, Action threadFinally)
{
    _taskProcessingThread.Value = true;
    if (threadInit != null) threadInit();
    try
    {
        // If the scheduler is disposed, the cancellation token will be set and
        // we'll receive an OperationCanceledException.  That OCE should not crash the process.
        try
        {
            // If a thread abort occurs, we'll try to reset it and continue running.
            while (true)
            {
                try
                {
                    // For each task queued to the scheduler, try to execute it.
                    foreach (var task in _blockingTaskQueue.GetConsumingEnumerable(_disposeCancellation.Token))
                    {
                        Task targetTask = task;
                        // If the task is not null, that means it was queued to this scheduler directly.
                        // Run it.
                        if (targetTask != null)
                        {
                            TryExecuteTask(targetTask);
                        }
                        // If the task is null, that means it's just a placeholder for a task
                        // queued to one of the subschedulers.  Find the next task based on
                        // priority and fairness and run it.
                        else
                        {
                            // Find the next task based on our ordering rules...                                    
                            QueuedTaskSchedulerQueue queueForTargetTask;
                            lock (_queueGroups) FindNextTask_NeedsLock(out targetTask, out queueForTargetTask);

                            // ... and if we found one, run it
                            if (targetTask != null) queueForTargetTask.ExecuteTask(targetTask);
                        }

                        if (_awaitWrappedTasks)
                        {
                            var targetTaskType = targetTask.GetType();
                            if (targetTaskType.IsConstructedGenericType && typeof(Task).IsAssignableFrom(targetTaskType.GetGenericArguments()[0]))
                            {
                                dynamic targetTaskDynamic = targetTask;
                                // Here we wait for the completion of the proxy task.
                                // We do not wait for the proxy task directly, because that would result in that Wait() will throw the exception of the wrapped task (if one existed)
                                // In the continuation we then simply return the value of the exception object so that the exception (stored in the proxy task) does not go totally unobserved (that could cause the process to crash)
                                TaskExtensions.Unwrap(targetTaskDynamic).ContinueWith((Func<Task, Exception>)(t => t.Exception), TaskContinuationOptions.ExecuteSynchronously).Wait();
                            }
                        }
                    }
                }
                catch (ThreadAbortException)
                {
                    // If we received a thread abort, and that thread abort was due to shutting down
                    // or unloading, let it pass through.  Otherwise, reset the abort so we can
                    // continue processing work items.
                    if (!Environment.HasShutdownStarted && !AppDomain.CurrentDomain.IsFinalizingForUnload())
                    {
                        Thread.ResetAbort();
                    }
                }
            }
        }
        catch (OperationCanceledException) { }
    }
    finally
    {
        // Run a cleanup routine if there was one
        if (threadFinally != null) threadFinally();
        _taskProcessingThread.Value = false;
    }
}

I have tested this and it gives the desired output. This technique could also be used for any other scheduler. E.g. LimitedConcurrencyLevelTaskScheduler and OrderedTaskScheduler

Waiting on the task in the scheduler destroys the value of async IO. If you don't need async IO anyway you could switch to synchronous task bodies. — usr, Nov 16 '12 at 11:56
+1 then. I learned a lot in this question. Not entirely convinced that this solution is preferable to an `AsyncSemaphore` but I'll think about it. — usr, Nov 16 '12 at 14:56
You are executing an `async-void` method from within a `TaskScheduler` implementation? Scary, I wonder that @StephenCleary hat nothing to say about this. — springy76, Feb 15 '17 at 09:22
@springy76 this actually worked very well, but I have since moved away from any explicit TaskScheduler implementations when I discovered the correct use of [ActionBlock in the TPL DataFlow framework](https://msdn.microsoft.com/en-us/library/hh228609(v=vs.110).aspx) — Francois Nel, Feb 16 '17 at 11:10

usr · Answer 3 · 2012-11-14T15:43:34.810

0

I think it is impossible to achieve this goal. A core problem seems to be that a TaskScheduler can only be used to run code. But there are tasks that do not run code, such as IO tasks or timer tasks. I don't think the TaskScheduler infrastructure can be used to schedule those.

From the perspective of a TaskScheduler it looks like this:

1. Select a registered task for execution
2. Execute its code on the CPU
3. Repeat

Step (2) is synchronous which means that the Task to be executed must start and finish as part of step (2). This means that this Task cannot do async IO because that would be non-blocking. In that sense, TaskScheduler only supports blocking code.

I think you would be served best by implementing yourself a version of AsyncSemaphore that releases waiters in priority order and does throttling. Your async methods can await that semaphore in a non-blocking way. All CPU work can run on the default thread-pool so there is no need to start custom threads inside of a custom TaskScheduler. IO tasks can continue to use non-blocking IO.

edited Nov 14 '12 at 15:43

answered Nov 14 '12 at 11:44

usr

168,620
35
240
369

what you explained here I have already tried and it has basically the same output (as in the original problem). In your suggestion `firstPartTask` is scheduled on the queued task scheduler, but is complete as soon as it hits the first `await` and the scheduler simply executes the next "first part" in the queue even if the previous "inner task" (the reset of the task after the first `await`) has not completed. I can only think that this will be solved by a **scheduler** that handles this scenario I'm looking for and cannot be solved by some magic outside the scheduler. – Francois Nel Nov 14 '12 at 13:11
I come to believe you're right. I added some thoughts and a suggestion. Please let me know what you think. – usr Nov 14 '12 at 13:39
Thanks for your update. Your suggestion using a semaphore lock is exactly what the user suggested in the following [answer](http://stackoverflow.com/a/13379980/1514235) (see my comments). Your suggestion that a scheduler only executes its tasks synchronously is somewhat true, but what if the scheduler awaits the "wrapped" task of each task before it executes any other tasks in the queue. I think this gave me an idea... thanks (will let you know if I come up with something). – Francois Nel Nov 14 '12 at 15:08
You could create a custom scheduler that knows that task execution has completed only after the wrapped task has completed as well as the wrapping task. It would need to do a runtime cast to see if the queued `Task` is actually a `Task` and in this case add a continuation and so on. – usr Nov 14 '12 at 15:46
I have added my own [answer](http://stackoverflow.com/a/13414364/1514235). Please have a look and see what you think. – Francois Nel Nov 16 '12 at 10:18

Limited concurrency level task scheduler (with task priority) handling wrapped tasks

3 Answers3

Linked