Task

De Banane Atomic
Aller à la navigationAller à la recherche

Description

Permet d'effectuer des tâches en parallèle ou des tâches sans consommer le thread de l'UI.
L'utilisation de Task permet de gagner en réactivité mais pas en performance.

Task

A Task represents an asynchronous operation.

Cs.svg 
// create and run a task which will wait for 4 seconds then return a result
var result = await Task.Run(async () => 
{
    await Task.Delay(4000);
    return 0;
});

TaskFactory et sous-tâches

Cs.svg 
Task<int[]> parent = Task.Run(() =>
{
    var results = new int[3];
    TaskFactory tf = new TaskFactory(TaskCreationOptions.AttachedToParent,
    TaskContinuationOptions.ExecuteSynchronously);
    tf.StartNew(() => results[0] = 0);
    tf.StartNew(() => results[1] = 1);
    tf.StartNew(() => results[2] = 2);
    return results;
});

var finalTask = parent.ContinueWith(parentTask =>
{
    foreach (int i in parentTask.Result)
        Console.WriteLine(i);
});

finalTask.Wait();

// exécution simultanée des tâches
Task[] tasks;
Task.WaitAll(tasks);

// attende de la fin d'une des tâches
int index = Task.WaitAny(tasks);

Annuler une tâche

Cs.svg 
var cancellationTokenSource = new CancellationTokenSource();
CancellationToken token = cancellationTokenSource.Token;

Task task = Task.Run(() =>
{
    while (!token.IsCancellationRequested)
    {
        Console.Write("*");
        Thread.Sleep(1000);
    }
    token.ThrowIfCancellationRequested();
}, token);

Console.WriteLine("Press enter to stop the task");
Console.ReadLine();
cancellationTokenSource.Cancel();
try
{
    task.Wait();
}
catch (AggregateException e)
{
    Console.WriteLine(e.InnerExceptions[0].Message);
}

async await

Permet d'écrire facilement du code asynchrone, exemple: ne pas bloquer le thread graphique lors de tâches longues à exécuter.
Alternative élégante au BackgroundWorker
Une méthode async tourne séquentiellement mais permet de faire des appels await sur des blocs de code.
await permet d’exécuter des tâches longues sans bloquer le thread UI.

Cs.svg 
button.Clicked += Button_Clicked;

// La méthode async doit retourner Task ou Task<T>, void est possible mais réservé au handler d'événement
async void Button_Clicked(object sender, EventArgs e)
{
    // appel asynchrone d'une méthode asynchrone
    await DoItAsync();

    // ou exécution direct du code avec le mot clé await
    label.Text = "Traitement en cours !!!";
    await Task.Delay(1000);
    label.Text = "Traitement terminé !!!";
}

// La méthode async doit retourner Task ou Task<T>, void est possible mais réservé au handler d'événement
// Le nom de la méthode doit se terminer par Async (convention)
async Task DoItAsync()
{
    label1.Text = "Traitement en cours !!!";

    await Task.Delay(1000);

    // le code suivant ne sera exécute qu'une fois la tache terminée
    // ce qui simplifie le code en évitant la création d'un event TaskCompleted pour avertir de la fin de la tache
    label1.Text = "Traitement terminé !!!";
}

Callback

Cs.svg 
private async Task DoItAsync(string param, Action callback)
{
    // Do some stuff
    await MyMethodAsync();

    // Then, when it's done, call the callback
    callback();

Paramètre out

Les méthodes async ne supportent pas les paramètres out.
Utiliser un tuple comme paramètre de retour à la place.

Cs.svg 
public async Task<(int status, string result)> MyMethod() {}

Application Console

Cs.svg 
// permet d'appeler une méthode async dans la méthode Main d'un projet Console
string result = MyMethodAsync().GetAwaiter().GetResult();

static async Task<string> MyMethodAsync()
{
}

Fire and forget

Call an async method without waiting for the response. Exceptions will be lost.

Cs.svg 
_ = MyMethodAsync().ConfigureAwait(false);  // ConfigureAwait to avoid deadlock
MyMethodAsync().Forget();

async Task MyMethodAsync() {}

static class TaskExtension
{
    public static void Forget(this Task _)
    {
    }
}

Propriété WPF

Nuget:

  • Nito.Mvvm.Async prerelease
  • FontAwesome.WPF
MainWindow.xaml 
<Window xmlns:Controls="clr-namespace:System.Windows.Controls;assembly=PresentationFramework"
        xmlns:fa="http://schemas.fontawesome.io/icons/">
    <Window.Resources>
        <Controls:BooleanToVisibilityConverter x:Key="BooleanToVisibilityConverter" />
    </Window.Resources>

    <TextBox Text="{Binding Query, UpdateSourceTrigger=PropertyChanged}" />
    <TextBlock Text="{Binding ResultTask.Result}" />

    <Label Content="Loading..." 
           Visibility="{Binding ResultTask.IsNotCompleted, 
                                Converter={StaticResource BooleanToVisibilityConverter}, 
                                FallbackValue=Collapsed}"/>
    <fa:ImageAwesome Icon="Refresh" Spin="True" Height="16" Width="16" 
                     Visibility="{Binding ResultTask.IsNotCompleted, 
                     Converter={StaticResource BooleanToVisibilityConverter}, 
                     FallbackValue=Collapsed}" />

    <Label Content="{Binding NotifyValuesTask.ErrorMessage}" 
           Visibility="{Binding ResultTask.IsFaulted, 
                                Converter={StaticResource BooleanToVisibilityConverter}}"/>
MainVM.cs 
private string _query;
public string Query
{
    get { return _query; }
    set
    {
        Set(() => Query, ref _query, value, true);
        ResultTask = NotifyTask.Create(GetResultAsync(_query, MyCallback));
    }
}

private void MyCallback() { /* ... */ }

private NotifyTask<string> _resultTask;
public NotifyTask<string> ResultTask
{
    get
    {
        return _resultTask;
    }
    set
    {
        Set(() => ResultTask, ref _resultTask, value, true);
    }
}

public async Task<string> GetResultAsync(string query, Action callback)
{
    var url = $"http://localhost:57157/api/v1/test/result/{query}";
    var responseMessage = await _client.GetAsync(url);
    if (responseMessage.IsSuccessStatusCode)
    {
        return await responseMessage.Content.ReadAsStringAsync();
    }
    else
    {
        return await Task.FromResult($"{responseMessage.StatusCode}: {responseMessage.ReasonPhrase}");
    }

    callback();
}

Exceptions

Cs.svg 
catch (AggregateException e)
{
    Console.WriteLine("There where {0} exceptions", e.InnerExceptions.Count);
}

Parallel

Utile si le code n'est pas séquentiel.

Cs.svg 
// for loop
Parallel.For(1, 20, i =>
{
    Console.WriteLine(i);
    Thread.Sleep(1000);
});

// for each loop
Parallel.ForEach(Enumerable.Range(1, 20), i =>
{
    Console.WriteLine(i);
    Thread.Sleep(1000);
});

// invoke actions
Parallel.Invoke(
    () => {
        Console.WriteLine(1);
        Thread.Sleep(1000);
    },
    () => {
        Console.WriteLine(2);
        Thread.Sleep(1000);
    }
);

Parallel options

Cs.svg 
// after 4s throw an OperationCanceledException
// no further operations will start but don't stop currently executing operations
var cancellationTokenSource = new CancellationTokenSource(4000);

var parallelOptions = new ParallelOptions
{
    MaxDegreeOfParallelism = 12, // by default use as much computer power as possible
    TaskScheduler = null,
    CancellationToken = cancellationTokenSource.Token
}

Parallel.ForEach(
    numbers,
    parallelOptions,
    (int i, ParallelLoopState loopState) =>
{
    if (loopState.ShouldExitCurrentIteration) // check if another iteration has requested to break
    {
        loopState.Break(); // break loop
    }

    if (!cancellationTokenSource.Token.IsCancellationRequested) { /* next operation step */ } // useful for long operation to break
});

Handling exceptions

All the exceptions are catched and when all the tasks have been executed then an AggregateException is thrown if any.

Cs.svg 
try
{
    Parallel.Invoke(
        () =>
        {
            var waitTime = DateTime.UtcNow.AddSeconds(4);
            while (DateTime.UtcNow < waitTime) { }
        },
        () =>
        {
            throw new Exception("MyException");
        }
    );
}
catch (AggregateException ex)
{
    ex.InnerExceptions; // ReadOnlyCollection<Exception>
}

Shared variable

Lock

Cs.svg 
private static readonly object sumLock = new();

var sum = 0m; // shared variable, updated by threads
Parallel.For(0, 100, i =>
{
    lock(sumLock) // only 1 thread at a time can access
    {
        sum += 0.5m; // code inside the lock should take as little time as possible
    }
});
To avoid deadlocks:
  • use 1 lock object for each shared resource
  • avoid nested locks
  • use a new object

Interlocked

Create thread-safe atomic operations.

Faster than lock, but Interlocked only works with integers.
Cs.svg 
int sum = 0; // shared variable, updated by threads
Parallel.For(0, 100, i =>
{
    Interlocked.Add(ref sum, 1);
});

AsyncLocal

Allow to have a different variable for each async task.

Cs.svg 
private static AsyncLocal<decimal?> asyncLocal = new();
Parallel.For(0, 100, async (i) =>
{
    asyncLocal.Value = 10; // the asyncLocal is not shared among async tasks
});

Concurrent collections

BlockingCollection<T> ajout et suppression thread-safe. Add, Take. FIFO par défaut.
ConcurrentBag<T> sans ordre, doublons autorisés. Add, TryTake, TryPeek.
ConcurrentDictionary<TKey,T> TryAdd, TryUpdate, AddOrUpdate, GetOrAdd.
ConcurrentQueue<T> FIFO. Enqueue, TryDequeue.
ConcurrentStack<T> LIFO. Push, TryPop.
Cs.svg 
BlockingCollection<string> col = new BlockingCollection<string>();
col.Add("text");
string s = col.Take();

foreach (string v in col.GetConsumingEnumerable())
    Console.WriteLine(v);

Tasks

Cs.svg 
// AsParallel PLINQ
var tasks = Enumerable.Range(1, 30).AsParallel().Select(x => MyTaskAsync(x));
await Task.WhenAll(tasks);

// Task.Run
var tasks = Enumerable.Range(1, 30).Select(x => Task.Run(() => MyTaskAsync(x)));
await Task.WhenAll(tasks);

private async Task<int> MyTaskAsync(int i)
{
    var waitTime = DateTime.UtcNow.AddSeconds(4);
    while (DateTime.UtcNow < waitTime) { }
    await Task.Delay(0);
    return i;
}

Dataflow's ActionBlock

Cs.svg 
using System.Threading.Tasks.Dataflow;

var ids = Enumerable.Range(1, 30).ToList();

var block = new ActionBlock<int>(
    x =>
    {
        Thread.Sleep(1000);
        Console.WriteLine($"{x}/{ids.Count}");
    },
    new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 2 });

ids.ForEach(x => block.Post(x));

block.Complete();       // Signal completion
await block.Completion; // Asynchronously wait for completion.

ParallelQuery

Cs.svg 
using MoreLinq;

var ids = Enumerable.Range(1, 30).ToList();

var dop = 2;
var tasks = ids.Batch((ids.Count / dop) + 1)
    .AsParallel()
    .WithDegreeOfParallelism(dop)
    .WithExecutionMode(ParallelExecutionMode.ForceParallelism)
    .Select(
        (batchedIds, workerId) =>
        {
            var batchedTasks = batchedIds.Select(x => Task.Run(() =>
            {
                Thread.Sleep(1000);
                Console.WriteLine(x);
            }));
            return batchedTasks;
        })
    .SelectMany(x => x);

foreach (var task in tasks)
{
    await task;
}

PLINQ

AsParallel analyses the query to see if it is suitable for parallelization. This analysis adds overhead.
If it is unsafe or faster to run sequentially then it won't be run in parallel.

Cs.svg 
var numbers = Enumerable.Range(0, 100_000_000);

var parallelResult = numbers.AsParallel()
                            .WithDegreeOfParallelism(2)
                            .WithCancellation(token)
                            .WithExecutionMode(ParallelExecutionMode.ForceParallelism)
                            .WithMergeOptions(ParallelMergeOptions.Default)
                            .AsOrdered() // add overhead
                            .Where(i => i % 2 == 0);

// parcourt d'itération en mode parallèle, l'ordre est perdu.
// le parcourt commence même si parallelResult n'est pas au complet
parallelResult.ForAll(e => Console.WriteLine(e));

ForEachAsync

Cs.svg 
// extension de méthode
// dop = degree of parallelism = max number of threads
public static Task ParallelForEachAsync<T>(this IEnumerable<T> source, int dop, Func<T, Task> body)
{
    return Task.WhenAll(
        from partition in Partitioner.Create(source).GetPartitions(dop)
        select Task.Run(async delegate
        {
            using (partition)
                while (partition.MoveNext())
                    await body(partition.Current);
        }));
}

await myItems.ForEachAsync(100, async (myItem) =>
{
    myItem.MyProperty = await MyMethodAsync();
});
Récupérée de « http://wiki.bananeatomic.fr/index.php?title=Task&oldid=27629 »