Bibliothèque parallèle TPL et programmation parallèle en C#

En .NET, la Task Parallel Library (TPL) simplifie l’exécution parallèle en tirant parti des processeurs multi-cœurs. Avec TPL, vous pouvez utiliser des outils tels que Task, Parallel, les collections Concurrent et PLINQ pour accélérer les opérations intensives en CPU. Cet article présente les concepts fondamentaux, les cas d’utilisation appropriés, la gestion des annulations et des exceptions, ainsi que des exemples pratiques.

Qu’est-ce que TPL ? Commencer avec Task

Une Task représente une unité de travail. Task.Run envoie les tâches intensives en CPU au pool de threads. Vous pouvez lancer plusieurs tâches simultanément et attendre leur achèvement avec Task.WhenAll.

using System;
using System.Threading.Tasks;

class Program
{
    static async Task Main()
    {
        Task t1 = Task.Run(() => Travail("A", 1000));
        Task t2 = Task.Run(() => Travail("B", 800));
        Task t3 = Task.Run(() => Travail("C", 1200));

        await Task.WhenAll(t1, t2, t3);
        Console.WriteLine("Toutes les tâches sont terminées.");
    }

    static void Travail(string nom, int ms)
    {
        Console.WriteLine($"{nom} a commencé");
        Task.Delay(ms).Wait(); // attente simulée (non intensive en CPU)
        Console.WriteLine($"{nom} terminé");
    }
}

Remarque : pour les opérations d’attente I/O, préférez async/await ; Task.Run est destiné aux travaux intensifs en CPU.

Parallel.For / Parallel.ForEach

La classe Parallel divise automatiquement les boucles et les distribue sur plusieurs threads. Lors de l’accès à des variables partagées, utilisez des techniques thread-safe comme Interlocked.

using System;
using System.Threading;
using System.Threading.Tasks;

class Program
{
    static void Main()
    {
        int total = 0;

        Parallel.For(0, 1_000_000, i =>
        {
            // Utilisation d’Interlocked pour éviter les conflits (race conditions)
            Interlocked.Add(ref total, 1);
        });

        Console.WriteLine($"Total : {total}");
    }
}

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

class Program
{
    static void Main()
    {
        var nombres = new List<int> { 1, 2, 3, 4, 5 };
        Parallel.ForEach(nombres, n =>
        {
            Console.WriteLine($"{n} traité (Thread : {Environment.CurrentManagedThreadId})");
        });
    }
}

ParallelOptions : Degré de parallélisme et annulation

Avec MaxDegreeOfParallelism, vous pouvez limiter le degré de parallélisme ; avec CancellationToken, vous pouvez ajouter la prise en charge de l’annulation.

using System;
using System.Threading;
using System.Threading.Tasks;

class Program
{
    static void Main()
    {
        var cts = new CancellationTokenSource();
        var po = new ParallelOptions
        {
            MaxDegreeOfParallelism = Environment.ProcessorCount - 1, // selon le nombre de cœurs
            CancellationToken = cts.Token
        };

        // Annulation après 2 secondes
        Task.Run(async () => { await Task.Delay(2000); cts.Cancel(); });

        try
        {
            Parallel.For(0, 1000, po, i =>
            {
                po.CancellationToken.ThrowIfCancellationRequested();
                TravailCPU(i); // travail intensif en CPU
            });
        }
        catch (OperationCanceledException)
        {
            Console.WriteLine("Les opérations ont été annulées.");
        }
    }

    static void TravailCPU(int i)
    {
        // Exemple de travail
        double x = 0;
        for (int k = 0; k < 50_000; k++) x += Math.Sqrt(k + i);
    }
}

Accumulation avec des variables locales (Local Init/Finally)

Au lieu de verrouiller une variable partagée, utilisez un accumulateur local pour chaque thread, puis combinez les résultats à la fin.

using System;
using System.Threading.Tasks;

class Program
{
    static void Main()
    {
        long totalGlobal = 0;

        Parallel.For<long>(0, 10_000_000,
            () => 0, // total local par thread
            (i, loop, totalLocal) =>
            {
                totalLocal += i % 10;
                return totalLocal;
            },
            totalLocal => { System.Threading.Interlocked.Add(ref totalGlobal, totalLocal); });

        Console.WriteLine($"Total : {totalGlobal}");
    }
}

Collections Concurrentes

Dans les scénarios de production/consommation, ConcurrentBag<T>, ConcurrentQueue<T> et ConcurrentDictionary<TKey,TValue> offrent un accès sans verrouillage ou à faible verrouillage.

using System;
using System.Collections.Concurrent;
using System.Threading.Tasks;

class Program
{
    static void Main()
    {
        var bag = new ConcurrentBag<int>();

        Parallel.For(0, 1000, i => bag.Add(i));

        int compteur = 0;
        while (bag.TryTake(out _)) compteur++;

        Console.WriteLine($"Éléments extraits : {compteur}");
    }
}

PLINQ (Parallel LINQ)

Utilisez AsParallel() pour exécuter des requêtes en parallèle sur de grandes collections. Si l’ordre est important, ajoutez AsOrdered() ; sinon, l’exécution non ordonnée offre de meilleures performances.

using System;
using System.Linq;

class Program
{
    static void Main()
    {
        var tableau = Enumerable.Range(1, 1_000_000).ToArray();

        var resultat = tableau
            .AsParallel()
            .Where(x => x % 3 == 0)
            .Select(x => x * x)
            .Take(10)
            .ToArray();

        Console.WriteLine(string.Join(", ", resultat));
    }
}

Gestion des exceptions

Dans les appels Parallel, plusieurs exceptions peuvent être levées ; elles sont gérées avec AggregateException. Dans les flux basés sur Task, les exceptions apparaissent pendant l’attente avec await.

using System;
using System.Threading.Tasks;

class Program
{
    static void Main()
    {
        try
        {
            Parallel.Invoke(
                () => throw new InvalidOperationException("A"),
                () => throw new ApplicationException("B")
            );
        }
        catch (AggregateException ex)
        {
            foreach (var e in ex.InnerExceptions)
                Console.WriteLine($"Erreur : {e.GetType().Name} - {e.Message}");
        }
    }
}

async/await vs Parallel

• async/await : Pour les opérations I/O (fichiers, réseau, base de données) sans bloquer l’interface utilisateur.
• Parallel/TPL : Pour les travaux intensifs en CPU, répartis sur plusieurs cœurs.
• Évitez d’envoyer inutilement des tâches I/O au pool de threads avec Task.Run ; le gain réel se produit avec les tâches CPU-intensives.

Conseils de performance et bonnes pratiques

• Minimisez l’état partagé ; combinez les résultats à l’aide d’accumulateurs locaux lorsque c’est possible.
• Évitez les verrous inutiles ; utilisez Interlocked ou les collections Concurrent si nécessaire.
• Utilisez MaxDegreeOfParallelism pour limiter le changement de contexte excessif.
• Dans les applications UI, évitez les appels bloquants (.Wait() / .Result).

Exemple : Traitement d’images en parallèle

L’exemple suivant simule le traitement parallèle de fichiers image dans un dossier. Dans un scénario réel, des opérations CPU-intensives comme le filtrage ou la génération de vignettes pourraient être effectuées ici.

using System;
using System.IO;
using System.Threading;
using System.Threading.Tasks;

class Program
{
    static void Main()
    {
        string dossier = "C:\\\\Images";
        var fichiers = Directory.Exists(dossier)
            ? Directory.GetFiles(dossier, "*.jpg")
            : Array.Empty<string>();

        var po = new ParallelOptions { MaxDegreeOfParallelism = Environment.ProcessorCount };
        Parallel.ForEach(fichiers, po, fichier =>
        {
            // Simulation d’un traitement d’image intensif en CPU
            var nom = Path.GetFileName(fichier);
            Console.WriteLine($"Traitement : {nom}");
            Thread.SpinWait(3000000); // Simulation
            Console.WriteLine($"Terminé : {nom}");
        });

        Console.WriteLine("Tous les fichiers ont été traités.");
    }
}

TL;DR (Résumé)

Articles connexes