Come eseguire un'attività su un TaskScheduler personalizzato usando Attendere?

Question

Ho alcuni metodi che restituiscono Task<T> su cui posso await a piacere. Mi piacerebbe avere questi compiti eseguiti su un TaskScheduler personalizzato al posto di quello predefinito.

var task = GetTaskAsync(); 
await task; 

so che posso creare un nuovo TaskFactory (new CustomScheduler()) e fare un StartNew() da esso, ma StartNew() prende un'azione e creare il Task, e ho già il Task (restituito dietro le quinte da un TaskCompletionSource)

Come posso specificare il mio TaskScheduler per await?

Answer 1

Dopo i commenti sembra che si desideri controllare lo scheduler su cui viene eseguito il codice dopo l'attesa.

La compilazione crea una continuazione dall'attesa eseguita su SynchronizationContext corrente per impostazione predefinita. Quindi la soluzione migliore è impostare il SynchronizationContext prima di chiamare Attendere.

Ci sono alcuni modi per attendere un contesto specifico. Vedi Configure Await da Jon Skeet, in particolare la parte su SwitchTo, per ulteriori informazioni su come implementare qualcosa di simile.

MODIFICA: Il metodo SwitchTo di TaskEx è stato rimosso, poiché era troppo facile da utilizzare in modo errato. Vedere lo MSDN Forum per motivi.

Answer 2

Penso che quello che vuoi veramente sia fare un Task.Run, ma con uno schedulatore personalizzato. StartNew non funziona in modo intuitivo con metodi asincroni; Stephen Toub ha un ottimo post sul blog su the differences between Task.Run and TaskFactory.StartNew.

Così, per creare il proprio personalizzato Run, si può fare qualcosa di simile:

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); 
} 

Quindi è possibile eseguire sincroni o metodi asincroni sul scheduler personalizzato.

Answer 3

Il TaskCompletionSource<T>.Task è costruito senza alcuna azione e lo scheduler viene assegnato alla prima chiamata a ContinueWith(...) (da Asynchronous Programming with the Reactive Framework and the Task Parallel Library — Part 3).

Fortunatamente è possibile personalizzare il comportamento attendono leggermente implementando la propria classe derivanti da INotifyCompletion e poi utilizzarlo in un modello simile a await SomeTask.ConfigureAwait(false) per configurare il programma di pianificazione che l'attività deve iniziare ad usare nel metodo OnCompleted(Action continuation) (da await anything;).

Ecco l'utilizzo:

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; } 

Ecco una semplice implementazione di ConfigureScheduler utilizzando un metodo di estensione operativa con la parte importante 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() { } 
    } 
} 

Ecco un esempio di lavoro che compilare come un'applicazione console:

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); 
     } 
    } 
} 

Answer 4

Potete chiamare per questo metodo:

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);