50

I have some methods returning Task<T> on which I can await at will. I'd like to have those Tasks executed on a custom TaskScheduler instead of the default one.

var task = GetTaskAsync ();
await task;

I know I can create a new TaskFactory (new CustomScheduler ()) and do a StartNew () from it, but StartNew () takes an action and create the Task, and I already have the Task (returned behind the scenes by a TaskCompletionSource)

How can I specify my own TaskScheduler for await ?

4
  • 2
    Doesn´t await expect a hot (aka already running) task? Also tasks from TaskCompletionSource cannot be "run", as one is responsible for completing the task by calling SetResult. Do you mean how to specify the scheduler on which the continuation from await is run?
    – sanosdole
    Mar 15, 2013 at 9:26
  • 2
    the task is hot and running, while returned by the CTS. But some scheduling happens between the SetResult and the continuation after the await. I need to control that scheduling with my own scheduler, if possible Mar 15, 2013 at 9:58
  • @StephaneDelcroix Can't you run the whole method on your scheduler? I think that would be the best solution.
    – svick
    Mar 15, 2013 at 22:18
  • 2
    From your comments, I understand that you want to use a specific scheduler for the continuation of the initial Task and not the initial Task itself, as it is already hot and running. For configuring continuation you may want to look at custom awaiter implementations. I think you should be able to write something similar to "await GetTaskAsync().ContinueIn(MyScheduler)" Oct 13, 2017 at 8:17

6 Answers 6

55

I think what you really want is to do a Task.Run, but with a custom scheduler. StartNew doesn't work intuitively with asynchronous methods; Stephen Toub has a great blog post about the differences between Task.Run and TaskFactory.StartNew.

So, to create your own custom Run, you can do something like this:

private static readonly TaskFactory myTaskFactory = new TaskFactory(
    CancellationToken.None, TaskCreationOptions.DenyChildAttach,
    TaskContinuationOptions.None, new MyTaskScheduler());
private static Task RunOnMyScheduler(Func<Task> func)
{
  return myTaskFactory.StartNew(func).Unwrap();
}
private static Task<T> RunOnMyScheduler<T>(Func<Task<T>> func)
{
  return myTaskFactory.StartNew(func).Unwrap();
}
private static Task RunOnMyScheduler(Action func)
{
  return myTaskFactory.StartNew(func);
}
private static Task<T> RunOnMyScheduler<T>(Func<T> func)
{
  return myTaskFactory.StartNew(func);
}

Then you can execute synchronous or asynchronous methods on your custom scheduler.

7
  • 2
    I'm sorry to say but the real Task doing work is scheduled on the default ThreadPoolTaskScheduler -- your MyTaskScheduler is just used to schedule the inner task.
    – springy76
    Sep 15, 2016 at 14:55
  • 2
    @springy76: No, that's incorrect. MyTaskScheduler will run func, and in the case of an asynchronous func, any awaits will resume on MyTaskScheduler by default. Sep 15, 2016 at 15:09
  • 2
    I took the example LimitedConcurrencyLevelTaskScheduler and modified it to be LIFO. But all tasks get executed in FIFO order and are not limited, but TaskScheduler.Current returns my custom scheduler at any possible debugging point. The scheduler works as expected when being used with Parallel.ForEach or when executing a real Func<TResult> using myTaskFactory.StartNew() -- but the combination of Func<Task<TResult>> and Unwrap() does not even execute the code before the first await in neither the right order nor with limited concurrency, saying the scheduler is totally ignored.
    – springy76
    Sep 15, 2016 at 15:58
  • @springy76: That is not expected behavior. I recommend you post a minimal, reproducible example as a separate question. Sep 15, 2016 at 16:01
  • 2
    @springy76: Well, I would expect a bit of oddness - specifically, asynchronous methods sent to the thread scheduler "leave" the FIFO queue while running, and when they're ready to resume after an await, they'll have to re-enter the FIFO queue (at the end). So the behavior won't be obvious, but it should certainly be running them with that task scheduler. Sep 15, 2016 at 16:20
12

The TaskCompletionSource<T>.Task is constructed without any action and the scheduler is assigned on the first call to ContinueWith(...) (from Asynchronous Programming with the Reactive Framework and the Task Parallel Library — Part 3).

Thankfully you can customize the await behavior slightly by implementing your own class deriving from INotifyCompletion and then using it in a pattern similar to await SomeTask.ConfigureAwait(false) to configure the scheduler that the task should start using in the OnCompleted(Action continuation) method (from await anything;).

Here is the usage:

    TaskCompletionSource<object> source = new TaskCompletionSource<object>();

    public async Task Foo() {
        // Force await to schedule the task on the supplied scheduler
        await SomeAsyncTask().ConfigureScheduler(scheduler);
    }

    public Task SomeAsyncTask() { return source.Task; }

Here is a simple implementation of ConfigureScheduler using a Task extension method with the important part in OnCompleted:

public static class TaskExtension {
    public static CustomTaskAwaitable ConfigureScheduler(this Task task, TaskScheduler scheduler) {
        return new CustomTaskAwaitable(task, scheduler);
    }
}

public struct CustomTaskAwaitable {
    CustomTaskAwaiter awaitable;

    public CustomTaskAwaitable(Task task, TaskScheduler scheduler) {
        awaitable = new CustomTaskAwaiter(task, scheduler);
    }

    public CustomTaskAwaiter GetAwaiter() { return awaitable; }

    public struct CustomTaskAwaiter : INotifyCompletion {
        Task task;
        TaskScheduler scheduler;

        public CustomTaskAwaiter(Task task, TaskScheduler scheduler) {
            this.task = task;
            this.scheduler = scheduler;
        }

        public void OnCompleted(Action continuation) {
            // ContinueWith sets the scheduler to use for the continuation action
            task.ContinueWith(x => continuation(), scheduler);
        }

        public bool IsCompleted { get { return task.IsCompleted; } }
        public void GetResult() { }
    }
}

Here's a working sample that will compile as a console application:

using System;
using System.Collections.Generic;
using System.Runtime.CompilerServices;
using System.Threading.Tasks;

namespace Example {
    class Program {
        static TaskCompletionSource<object> source = new TaskCompletionSource<object>();
        static TaskScheduler scheduler = new CustomTaskScheduler();

        static void Main(string[] args) {
            Console.WriteLine("Main Started");
            var task = Foo();
            Console.WriteLine("Main Continue ");
            // Continue Foo() using CustomTaskScheduler
            source.SetResult(null);
            Console.WriteLine("Main Finished");
        }

        public static async Task Foo() {
            Console.WriteLine("Foo Started");
            // Force await to schedule the task on the supplied scheduler
            await SomeAsyncTask().ConfigureScheduler(scheduler);
            Console.WriteLine("Foo Finished");
        }

        public static Task SomeAsyncTask() { return source.Task; }
    }

    public struct CustomTaskAwaitable {
        CustomTaskAwaiter awaitable;

        public CustomTaskAwaitable(Task task, TaskScheduler scheduler) {
            awaitable = new CustomTaskAwaiter(task, scheduler);
        }

        public CustomTaskAwaiter GetAwaiter() { return awaitable; }

        public struct CustomTaskAwaiter : INotifyCompletion {
            Task task;
            TaskScheduler scheduler;

            public CustomTaskAwaiter(Task task, TaskScheduler scheduler) {
                this.task = task;
                this.scheduler = scheduler;
            }

            public void OnCompleted(Action continuation) {
                // ContinueWith sets the scheduler to use for the continuation action
                task.ContinueWith(x => continuation(), scheduler);
            }

            public bool IsCompleted { get { return task.IsCompleted; } }
            public void GetResult() { }
        }
    }

    public static class TaskExtension {
        public static CustomTaskAwaitable ConfigureScheduler(this Task task, TaskScheduler scheduler) {
            return new CustomTaskAwaitable(task, scheduler);
        }
    }

    public class CustomTaskScheduler : TaskScheduler {
        protected override IEnumerable<Task> GetScheduledTasks() { yield break; }
        protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued) { return false; }
        protected override void QueueTask(Task task) {
            TryExecuteTask(task);
        }
    }
}
5
  • 4
    While it is possible to do this, it's generally not a good idea in production code. The original CTP had a SwitchTo operator that behaved much the same way, but was removed because it encouraged unstructured program flow, which in turn caused complications and pitfalls. For example, if you use a modified await in your try, then your finally block would have to be able to execute either on the original scheduler or the modified one (and the finally code cannot switch schedulers since await is not allowed). Jun 21, 2013 at 20:49
  • @StephenCleary now that catch/finally block can await (c# 6), is it still a bad idea?
    – Kryptos
    Jun 12, 2015 at 11:58
  • @Kryptos: Yes, it's still a bad idea. Enabling await in catch/finally has no effect on the SwitchTo behavior. Jun 12, 2015 at 12:47
  • @StephenCleary do you mean that using any custom awaiter is a bad idea? Or just this implementation of one?
    – Erasmus
    Apr 10, 2020 at 17:20
  • @Erasmus: Just this implementation. Apr 10, 2020 at 21:25
9

There is no way to embed rich async features into a custom TaskScheduler. This class was not designed with async/await in mind. The standard way to use a custom TaskScheduler is as an argument to the Task.Factory.StartNew method. This method does not understand async delegates. It is possible to provide an async delegate, but it is treated as any other delegate that returns some result. To get the actual awaited result of the async delegate one must call Unwrap() to the task returned.

This is not the problem though. The problem is that the TaskScheduler infrastructure does not treat the async delegate as a single unit of work. Each task is split into multiple mini-tasks (using every await as a separator), and each mini-task is processed individually. This severely restricts the asynchronous functionality that can be implemented on top of this class. As an example here is a custom TaskScheduler that is intended to queue the supplied tasks one at a time (to limit the concurrency in other words):

public class MyTaskScheduler : TaskScheduler
{
    private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1);

    protected async override void QueueTask(Task task)
    {
        await _semaphore.WaitAsync();
        try
        {
            await Task.Run(() => base.TryExecuteTask(task));
            await task;
        }
        finally
        {
            _semaphore.Release();
        }
    }

    protected override bool TryExecuteTaskInline(Task task,
        bool taskWasPreviouslyQueued) => false;

    protected override IEnumerable<Task> GetScheduledTasks() { yield break; }
}

The SemaphoreSlim should ensure that only one Task would run at a time. Unfortunately it doesn't work. The semaphore is released prematurely, because the Task passed in the call QueueTask(task) is not the task that represents the whole work of the async delegate, but only the part until the first await. The other parts are passed to the TryExecuteTaskInline method. There is no way to correlate these task-parts, because no identifier or other mechanism is provided. Here is what happens in practice:

var taskScheduler = new MyTaskScheduler();
var tasks = Enumerable.Range(1, 5).Select(n => Task.Factory.StartNew(async () =>
{
    Console.WriteLine($"{DateTime.Now:HH:mm:ss.fff} Item {n} Started");
    await Task.Delay(1000);
    Console.WriteLine($"{DateTime.Now:HH:mm:ss.fff} Item {n} Finished");
}, default, TaskCreationOptions.None, taskScheduler))
.Select(t => t.Unwrap())
.ToArray();
Task.WaitAll(tasks);

Output:

05:29:58.346 Item 1 Started
05:29:58.358 Item 2 Started
05:29:58.358 Item 3 Started
05:29:58.358 Item 4 Started
05:29:58.358 Item 5 Started
05:29:59.358 Item 1 Finished
05:29:59.374 Item 5 Finished
05:29:59.374 Item 4 Finished
05:29:59.374 Item 2 Finished
05:29:59.374 Item 3 Finished

Disaster, all tasks are queued at once.

Conclusion: Customizing the TaskScheduler class is not the way to go when advanced async features are required.

Update: Here is another observation, regarding custom TaskSchedulers in the presence of an ambient SynchronizationContext. The await mechanism by default captures the current SynchronizationContext, or the current TaskScheduler, and invokes the continuation on either the captured context or the scheduler. If both are present, the current SynchronizationContext is preferred, and the current TaskScheduler is ignored. Below is a demonstration of this behavior, in a WinForms application¹:

private async void Button1_Click(object sender, EventArgs e)
{
    await Task.Factory.StartNew(async () =>
    {
        MessageBox.Show($"{Thread.CurrentThread.ManagedThreadId}, {TaskScheduler.Current}");
        await Task.Delay(1000);
        MessageBox.Show($"{Thread.CurrentThread.ManagedThreadId}, {TaskScheduler.Current}");
    }, default, TaskCreationOptions.None,
        TaskScheduler.FromCurrentSynchronizationContext()).Unwrap();
}

Clicking the button causes two messages to popup sequentially, with this information:

1, System.Threading.Tasks.SynchronizationContextTaskScheduler

1, System.Threading.Tasks.ThreadPoolTaskScheduler

This experiment shows that only the first part of the asynchronous delegate, the part before the first await, was scheduled on the non-default scheduler. This behavior limits even further the practical usefulness of custom TaskSchedulers in an async/await-enabled environment.

¹ Windows Forms applications have a WindowsFormsSynchronizationContext installed automatically, when the Application.Run method is called.

4

Can you fit for this method call:

  await Task.Factory.StartNew(
        () => { /* to do what you need */ }, 
        CancellationToken.None, /* you can change as you need */
        TaskCreationOptions.None, /* you can change as you need */
        customScheduler);
4

After the comments it looks like you want to control the scheduler on which the code after the await is run.

The compile creates a continuation from the await that runs on the current SynchronizationContext by default. So your best shot is to set up the SynchronizationContext before calling await.

There are some ways to await a specific context. See Configure Await from Jon Skeet, especially the part about SwitchTo, for more information on how to implement something like this.

EDIT: The SwitchTo method from TaskEx has been removed, as it was too easy to misuse. See the MSDN Forum for reasons.

-1

Faced with same issue, tried to use LimitedConcurrencyLevelTaskScheduler, but it does not support async tasks. So...

Just wrote my own small simple Scheduler, that allow to run async Tasks based on global ThreadPool (and Task.Run method) with ability to limit current max degree of parallelism. It is enough for my exact purposes, maybe will also help you, guys.

Main demo code (console app, dotnet core 3.1) :

    static async Task Main(string[] args)
    {

        //5 tasks to run per time
        int concurrentLimit = 5;
        var scheduler = new ThreadPoolConcurrentScheduler(concurrentLimit);

        //catch all errors in separate event handler
        scheduler.OnError += Scheduler_OnError;

        // just monitor "live" state and output to console
        RunTaskStateMonitor(scheduler);

        // simulate adding new tasks "on the fly"
        SimulateAddingTasksInParallel(scheduler);

        Console.WriteLine("start adding 50 tasks");

        //add 50 tasks
        for (var i = 1; i <= 50; i++)
        {
            scheduler.StartNew(myAsyncTask);
        }

        Console.WriteLine("50 tasks added to scheduler");

        Thread.Sleep(1000000);


    }

Supporting code (place it in the same place) :

    private static void Scheduler_OnError(Exception ex)
    {
        Console.WriteLine(ex.ToString());
    }

    private static int currentTaskFinished = 0;

    //your sample of async task
    static async Task myAsyncTask()
    {
        Console.WriteLine("task started ");

        using (HttpClient httpClient = new HttpClient())
        {
            //just make http request to ... wikipedia!
            //sorry, Jimmy Wales! assume,guys, you will not DDOS wiki :)
            var uri = new Uri("https://wikipedia.org/");
            var response = await httpClient.GetAsync(uri);
            string result = await response.Content.ReadAsStringAsync();
            if (string.IsNullOrEmpty(result))
                Console.WriteLine("error, await is not working");
            else
                Console.WriteLine($"task result : site length is {result.Length}");
        }
        //or simulate it using by sync sleep
        //Thread.Sleep(1000);
        //and for tesing exception : 
        //throw new Exception("my custom error");
        Console.WriteLine("task finished ");

        //just incrementing total ran tasks to output in console
        Interlocked.Increment(ref currentTaskFinished);
    }

    static void SimulateAddingTasksInParallel(ThreadPoolConcurrentScheduler taskScheduler)
    {
        int runCount = 0;
        Task.Factory.StartNew(() =>
        {
            while (true)
            {
                runCount++;

                if (runCount > 5)
                    break;

                //every 10 sec 5 times
                Thread.Sleep(10000);

                //adding new 5 tasks from outer task
                Console.WriteLine("start adding new 5 tasks!");
                for (var i = 1; i <= 5; i++)
                {
                    taskScheduler.StartNew(myAsyncTask);
                }

                Console.WriteLine("new 5 tasks added!");
            }
        }, TaskCreationOptions.LongRunning);
    }

    static void RunTaskStateMonitor(ThreadPoolConcurrentScheduler taskScheduler)
    {
        int prev = -1;
        int prevQueueSize = -1;
        int prevFinished = -1;
        Task.Factory.StartNew(() =>
            {
                while (true)
                {
                    // getting current thread count in working state
                    var currCount = taskScheduler.GetCurrentWorkingThreadCount();
                    // getting inner queue state
                    var queueSize = taskScheduler.GetQueueTaskCount();

                    //just output overall state if something changed
                    if (prev != currCount || queueSize != prevQueueSize || prevFinished != currentTaskFinished)
                    {
                        Console.WriteLine($"Monitor : running tasks:{currCount}, queueLength:{queueSize}. total Finished tasks : " + currentTaskFinished);
                        prev = currCount;
                        prevQueueSize = queueSize;
                        prevFinished = currentTaskFinished;
                    }

                    // check it every 10 ms
                    Thread.Sleep(10);
                }
            }
            , TaskCreationOptions.LongRunning);
    }

Scheduler :

public class ThreadPoolConcurrentScheduler
{
    private readonly int _limitParallelThreadsCount;
    private int _threadInProgressCount = 0;

    public delegate void onErrorDelegate(Exception ex);
    public event onErrorDelegate OnError;

    private ConcurrentQueue<Func<Task>> _taskQueue;
    private readonly object _queueLocker = new object();


    public ThreadPoolConcurrentScheduler(int limitParallelThreadsCount)
    {
        //set maximum parallel tasks to run
        _limitParallelThreadsCount = limitParallelThreadsCount;
        // thread-safe queue to store tasks
        _taskQueue = new ConcurrentQueue<Func<Task>>();
    }

    //main method to start async task
    public void StartNew(Func<Task> task)
    {
        lock (_queueLocker)
        {
            // checking limit
            if (_threadInProgressCount >= _limitParallelThreadsCount)
            {
                //waiting new "free" threads in queue
                _scheduleTask(task);
            }
            else
            {
                _startNewTask(task);
            }
        }
    }

    private void _startNewTask(Func<Task> task)
    {
        Interlocked.Increment(ref _threadInProgressCount);
        Task.Run(async () =>
        {
            try
            {
                await task();
            }
            catch (Exception e)
            {
                //Console.WriteLine(e);
                OnError?.Invoke(e);
            }
        }).ContinueWith(_onTaskEnded);
    }

    //will be called on task end
    private void _onTaskEnded(Task task)
    {
        lock (_queueLocker)
        {
            Interlocked.Decrement(ref _threadInProgressCount);
            //queue has more priority, so if thread is free - let's check queue first
            if (!_taskQueue.IsEmpty)
            {
                if (_taskQueue.TryDequeue(out var result))
                {
                    _startNewTask(result);
                }
            }
        }
    }

    private void _scheduleTask(Func<Task> task)
    {
        _taskQueue.Enqueue(task);
    }

    //returning in progress task count 
    public int GetCurrentWorkingThreadCount()
    {
        return _threadInProgressCount;
    }

    //return number of tasks waiting to run
    public int GetQueueTaskCount()
    {
        lock (_queueLocker) return _taskQueue.Count;
    }
}

Few notes :

  1. First - check comments to it, maybe it is the worst code ever!
  2. Did not test in prod
  3. Did not implement cancellation tokens and any other functionality, that should be there, but i'm too lazy. Sorry
5
  • FYI there is a LimitedConcurrencyLevelTaskScheduler built-in the .NET platform. It is the: var scheduler = new ConcurrentExclusiveSchedulerPair(TaskScheduler.Default, 5).ConcurrentScheduler; Dec 28, 2020 at 7:12
  • @TheodorZoulias, as i know, LimitedConcurrencyLevelTaskScheduler was designed before async/await so it does not support work with it correctly?
    – Nigrimmist
    Dec 28, 2020 at 9:03
  • 1
    Yes, all TaskSchedulers are useless when dealing with asynchronous methods. They schedule the code that runs between the awaited operations, and they are completely unaware about the awaited operations themselves. Does your ThreadPoolConcurrentScheduler class solve this problem somehow? Dec 28, 2020 at 9:40
  • 1
    Oh, I see. Your ThreadPoolConcurrentScheduler is not a TaskScheduler. It's a class that schedules asynchronous delegates Func<Task>, not tasks. I suggest to change the parameter Func<Task> task to Func<Task> taskFactory or Func<Task> actionAsync, because its current name is confusing. Dec 28, 2020 at 9:49
  • Btw the problem of how to limit the concurrency of asynchronous operations has already plenty of solutions, for example here or here. This question here asks specifically about combining TaskSchedulers and await, so your answer is probably not a good fit. Dec 28, 2020 at 10:00

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.