Files et piles Producer-Consumer

1. Introduction

Commençons par les bases. La file est une structure de données fondamentale qui fonctionne selon le principe FIFO (First-In, First-Out), c'est-à-dire "le premier entré — le premier sorti". Imaginez une file au supermarché : celui qui s'est mis en premier est servi en premier.

En programmation multithread, le pattern Producteur-Consommateur (Producer-Consumer) est l'un des plus répandus et puissants.

• Producteurs (Producers) — ce sont des threads ou des parties de l'application qui créent des données ou des tâches et les placent dans une file commune. Ils "produisent" du travail.
• Consommateurs (Consumers) — ce sont des threads ou des parties de l'application qui prennent des données ou des tâches de la file et les traitent. Ils "consomment" le travail.

Ce pattern aide à gérer le flux de données, découple les composants du système (le producteur n'a pas besoin de savoir qui et comment traitera les données), rend le système plus réactif et aide à répartir la charge entre les threads de façon uniforme.

Exemple : ConcurrentQueue — ajout et extraction

Voyons comment ajouter et extraire des éléments de ConcurrentQueue<T>.

using System.Collections.Concurrent;

ConcurrentQueue<string> tasks = new ConcurrentQueue<string>();

// Ajout d'éléments (producteur)
tasks.Enqueue("Download file"); 
tasks.Enqueue("Process image");
Console.WriteLine($"Tasks in queue: {tasks.Count}"); // Sortie : Tasks in queue: 2

// Extraction d'éléments (consommateur)
if (tasks.TryDequeue(out string task1))
{
    Console.WriteLine($"Completed task: {task1}"); // Sortie : Completed task: Download file
}

if (tasks.TryDequeue(out string task2))
{
    Console.WriteLine($"Completed task: {task2}"); // Sortie : Completed task: Process image
}

if (!tasks.TryDequeue(out string emptyTask))
{
    Console.WriteLine("Queue is empty, no new tasks."); // Sortie : Queue is empty, no new tasks.
}

Principes de base : Enqueue(), TryDequeue()

Enqueue(T item) : Sert à ajouter un élément à la fin de la file. Cette opération est thread-safe. Vous pouvez appeler Enqueue simultanément depuis 10 threads différents, et tous les éléments seront correctement ajoutés.

TryDequeue(out T item) : Sert à tenter d'extraire un élément depuis le début de la file. C'est la méthode clé pour les consommateurs. Elle retourne true si un élément a été extrait avec succès (la valeur est placée dans le paramètre de sortie item), et false si la file est vide. Important : TryDequeue ne bloque pas le thread si la file est vide.

2. Importance de TryDequeue() et atomicité des opérations

La méthode TryDequeue() n'est pas seulement pratique ; elle est cruciale pour un fonctionnement thread-safe correct. Elle est atomique : la vérification de la vacuité de la file et l'extraction elle-même se font comme une seule opération indivisible.

Si nous avions des méthodes séparées IsEmpty (vérifier si la file est vide) et Dequeue (extraire un élément), entre leurs appels un autre thread pourrait vider la file. Finalement votre Dequeue lancerait une exception ou retournerait des données incorrectes. TryDequeue élimine complètement ce risque.

Exemple : Producer-Consumer avec plusieurs threads

Ici on lance deux threads producteurs et un thread consommateur.

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

ConcurrentQueue<int> dataQueue = new ConcurrentQueue<int>();
bool producersDone = false; // Flag pour signaler au consommateur

void Producer(int start, int count)
{
    for (int i = 0; i < count; i++)
    {
        dataQueue.Enqueue(start + i);
        Console.WriteLine($"[P] Added: {start + i}");
        Thread.Sleep(10); 
    }
}

void Consumer()
{
    while (!producersDone || dataQueue.Count > 0) // On continue tant qu'il y a des données ou que les producteurs bossent
    {
        if (dataQueue.TryDequeue(out int item))
        {
            Console.WriteLine($"[K] Processed: {item}");
        }
        else
        {
            Thread.Sleep(50); // On attend si la file est vide
        }
    }
    Console.WriteLine("[K] Finished work.");
}

// Lancement de l'exemple dans Main:
// Task.Run(() => Producer(1, 5));
// Task.Run(() => Producer(100, 5)); // Deuxième producteur
// Task.Run(() => Consumer());
// Thread.Sleep(600); // On laisse le temps aux threads de bosser
// producersDone = true; // On signale que les producteurs ont fini
// Thread.Sleep(200); // On laisse le temps au consommateur de prendre le reste

Notez que dans cet exemple simple le flag producersDone et Thread.Sleep sont utilisés pour simuler la fin. Dans des applications réelles, pour une synchronisation plus fiable de la terminaison on utilise souvent CancellationTokenSource ou BlockingCollection<T>.

ConcurrentQueue<T> convient parfaitement aux scénarios où :

• L'ordre de traitement des éléments est important (FIFO).
• Plusieurs threads ajoutent des éléments, et/ou plusieurs threads les retirent.
• On a besoin de haute performance sans gestion manuelle des locks.

3. La pile pour producer-consumer (LIFO)

La pile est une autre structure de données fondamentale qui fonctionne selon le principe LIFO (Last-In, First-Out), c'est-à-dire "le dernier entré — le premier sorti". Imaginez une pile d'assiettes : vous prenez toujours celle du dessus, et une nouvelle assiette est toujours posée sur le dessus de la pile.

ConcurrentStack<T> est aussi thread-safe que ConcurrentQueue<T>, et peut aussi être utilisé dans le pattern "producteur-consommateur", mais avec l'ordre de traitement inversé.

Exemple : ConcurrentStack — ajout et extraction

using System.Collections.Concurrent;

ConcurrentStack<string> commandStack = new ConcurrentStack<string>();

// Ajout de commandes (producteur)
commandStack.Push("Select text");
commandStack.Push("Change font");
commandStack.Push("Save document");
Console.WriteLine($"Commands in stack: {commandStack.Count}"); // Sortie : Commands in stack: 3

// Extraction de commandes (consommateur)
if (commandStack.TryPop(out string cmd1))
{
    Console.WriteLine($"Cancelled command: {cmd1}"); // Sortie : Cancelled command: Save document
}

if (commandStack.TryPop(out string cmd2))
{
    Console.WriteLine($"Cancelled command: {cmd2}"); // Sortie : Cancelled command: Change font
}

if (!commandStack.TryPop(out string emptyCmd))
{
    Console.WriteLine("Command stack is empty."); // Sortie : Command stack is empty.
}

4. Principes de base : Push(), TryPop()

Push(T item) : Sert à ajouter un élément au sommet de la pile. L'opération est thread-safe.

TryPop(out T item) : Sert à tenter d'extraire un élément du sommet de la pile. Elle retourne true si un élément a été extrait avec succès, et false si la pile est vide. Comme TryDequeue, c'est une opération atomique qui prévient les data races.

Exemple : utilisation de ConcurrentStack pour un pool d'objets

La pile convient bien pour implémenter des pools d'objets : on prend — on utilise — on rend.

using System.Collections.Concurrent;

class Connection { /* Simple placeholder */ public Guid Id { get; } = Guid.NewGuid(); }

ConcurrentStack<Connection> connectionPool = new ConcurrentStack<Connection>();

// On remplit le pool avec des connexions initiales
for (int i = 0; i < 3; i++)
{
    connectionPool.Push(new Connection());
}
Console.WriteLine($"Connections in pool: {connectionPool.Count}"); // Sortie : Connections in pool: 3

void UseConnection()
{
    if (connectionPool.TryPop(out Connection conn))
    {
        Console.WriteLine($"[Pool] Used connection: {conn.Id}");
        // Simulation de travail avec la connexion
        Thread.Sleep(50); 
        connectionPool.Push(conn); // On rend au pool
        Console.WriteLine($"[Pul] Vozvrashcheno soedinenie: {conn.Id}. V pule: {connectionPool.Count}");
    }
    else
    {
        Console.WriteLine("[Pool] Pool of connections is empty. Creating new..");
        // En général on crée une nouvelle connexion si le pool est vide
        connectionPool.Push(new Connection()); 
    }
}

// Lancement de l'exemple dans Main:
Task.Run(() => UseConnection());
Task.Run(() => UseConnection());
Task.Run(() => UseConnection());
Thread.Sleep(500);

Dans cet exemple plusieurs threads peuvent prendre et rendre des connexions au pool en toute sécurité.

5. Cas d'utilisation et comparaison avec ConcurrentQueue

ConcurrentStack<T> est utilisé quand :

• L'ordre LIFO est critique (par exemple l'historique d'actions pour une fonction "Undo").
• On a besoin d'un accès rapide aux éléments les plus récemment ajoutés (souvent "chauds" dans le cache CPU).
• On implémente des algorithmes basés sur des piles (parcours en profondeur, parsing d'expressions).

Comparaison et choix de la collection appropriée

ConcurrentQueue

ConcurrentStack

Le choix entre ConcurrentQueue et ConcurrentStack dépend entièrement de l'ordre de traitement requis dans votre scénario producer-consumer. Les deux collections fournissent des performances élevées et la thread-safety "out of the box", vous libérant de la gestion manuelle de la synchronisation et aidant à construire des systèmes multithread scalables.

2

Mission

C# SELF, niveau 58, leçon 1

Bloqué

Implémentation du modèle Producteur-Consommateur avec une file d'attente

Implémentation du modèle Producteur-Consommateur avec une file d'attente

Ouvrir