Code e stack Producer-Consumer

1. Introduzione

Cominciamo dalle basi. Queue — è una struttura dati fondamentale che funziona secondo il principio FIFO (First-In, First-Out), cioè "chi arriva primo esce primo". Immagina la fila al supermercato: chi si è messo per primo viene servito per primo.

Nella programmazione multithread il pattern Producer-Consumer è uno dei più comuni e potenti.

• Producers — sono thread o parti dell'applicazione che creano dati o task e li mettono in una coda condivisa. Loro "producono" lavoro.
• Consumers — sono thread o parti dell'applicazione che prendono dati o task dalla coda e li processano. Loro "consumano" lavoro.

Questo pattern aiuta a gestire il flusso di dati, disaccoppia i componenti del sistema (il producer non deve sapere chi e come processerà i dati), rende l'applicazione più reattiva e aiuta a distribuire il carico tra i thread.

Esempio: ConcurrentQueue — aggiunta e rimozione

Vediamo come aggiungere e rimuovere elementi da ConcurrentQueue<T>.

using System.Collections.Concurrent;

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

// Aggiunta di elementi (producer)
tasks.Enqueue("Scaricare il file");
tasks.Enqueue("Elaborare l'immagine");
Console.WriteLine($"Task nella coda: {tasks.Count}"); // Output: Task nella coda: 2

// Rimozione di elementi (consumer)
if (tasks.TryDequeue(out string task1))
{
    Console.WriteLine($"Task completato: {task1}"); // Output: Task completato: Scaricare il file
}

if (tasks.TryDequeue(out string task2))
{
    Console.WriteLine($"Task completato: {task2}"); // Output: Task completato: Elaborare l'immagine
}

if (!tasks.TryDequeue(out string emptyTask))
{
    Console.WriteLine("La coda è vuota, non ci sono nuovi task."); // Output: La coda è vuota, non ci sono nuovi task.
}

Basi del funzionamento: Enqueue(), TryDequeue()

Enqueue(T item): Si usa per aggiungere un elemento alla fine della coda. Questa operazione è thread-safe. Puoi chiamare Enqueue contemporaneamente da 10 thread diversi e tutti gli elementi verranno aggiunti correttamente.

TryDequeue(out T item): Si usa per provare a rimuovere un elemento dall'inizio della coda. È il metodo chiave per i consumer. Restituisce true se un elemento è stato rimosso con successo (il valore viene assegnato al parametro out item), e false se la coda è vuota. Importante: TryDequeue non blocca il thread se la coda è vuota.

2. Importanza di TryDequeue() e atomicità delle operazioni

Il metodo TryDequeue() non è solo comodo; è critico per il corretto funzionamento thread-safe. È atomico: il controllo se la coda è vuota e la rimozione dell'elemento avvengono come un'unica operazione indivisibile.

Se avessimo metodi separati IsEmpty (controllare se la coda è vuota) e Dequeue (rimuovere l'elemento), tra le chiamate un altro thread potrebbe svuotare la coda. Di conseguenza il tuo Dequeue potrebbe lanciare eccezioni o restituire dati non validi. TryDequeue elimina completamente questa situazione.

Esempio: Producer-Consumer con più thread

Qui lanciamo due thread producer e un thread consumer.

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

ConcurrentQueue<int> dataQueue = new ConcurrentQueue<int>();
bool producersDone = false; // Flag per segnalare al consumer

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

void Consumer()
{
    while (!producersDone || dataQueue.Count > 0) // Continuiamo finché ci sono dati o i producer lavorano
    {
        if (dataQueue.TryDequeue(out int item))
        {
            Console.WriteLine($"[C] Processato: {item}");
        }
        else
        {
            Thread.Sleep(50); // Aspettiamo se la coda è vuota
        }
    }
    Console.WriteLine("[C] Ho finito.");
}

// Esecuzione dell'esempio in Main:
// Task.Run(() => Producer(1, 5));
// Task.Run(() => Producer(100, 5)); // Secondo producer
// Task.Run(() => Consumer());
// Thread.Sleep(600); // Diamo tempo ai thread di lavorare
// producersDone = true; // Segnaliamo che i producer hanno finito
// Thread.Sleep(200); // Diamo tempo al consumer di prendere il resto

Nota che in questo esempio semplice il flag producersDone e Thread.Sleep vengono usati per simulare la terminazione. Nelle applicazioni reali per una sincronizzazione più affidabile si usano spesso CancellationTokenSource o BlockingCollection<T>.

ConcurrentQueue<T> è ideale per scenari dove:

• L'ordine di elaborazione degli elementi è importante (FIFO).
• Molti thread aggiungono elementi e/o molti thread li consumano.
• Serve alta performance senza gestire manualmente i lock.

3. Stack per producer-consumer (LIFO)

Stack — è un'altra struttura dati fondamentale che funziona secondo il principio LIFO (Last-In, First-Out), cioè "chi arriva per ultimo esce per primo". Immagina una pila di piatti: prendi sempre quello in cima, e quando ne aggiungi uno nuovo lo metti in cima alla pila.

ConcurrentStack<T> è thread-safe tanto quanto ConcurrentQueue<T>, e può essere usato nel pattern producer-consumer ma con ordine di elaborazione invertito.

Esempio: ConcurrentStack — aggiunta e rimozione

using System.Collections.Concurrent;

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

// Aggiunta di comandi (producer)
commandStack.Push("Selezionare testo");
commandStack.Push("Cambiare font");
commandStack.Push("Salvare documento");
Console.WriteLine($"Comandi nello stack: {commandStack.Count}"); // Output: Comandi nello stack: 3

// Rimozione di comandi (consumer)
if (commandStack.TryPop(out string cmd1))
{
    Console.WriteLine($"Comando annullato: {cmd1}"); // Output: Comando annullato: Salvare documento
}

if (commandStack.TryPop(out string cmd2))
{
    Console.WriteLine($"Comando annullato: {cmd2}"); // Output: Comando annullato: Cambiare font
}

if (!commandStack.TryPop(out string emptyCmd))
{
    Console.WriteLine("Lo stack dei comandi è vuoto."); // Output: Lo stack dei comandi è vuoto.
}

4. Basi del funzionamento: Push(), TryPop()

Push(T item): Si usa per aggiungere un elemento in cima allo stack. L'operazione è thread-safe.

TryPop(out T item): Si usa per provare a rimuovere l'elemento in cima allo stack. Restituisce true se l'elemento è stato rimosso con successo, e false se lo stack è vuoto. Come TryDequeue, è un'operazione atomica che previene race condition.

Esempio: usare ConcurrentStack per un object pool

Lo stack è ottimo per implementare object pool: prendi — usi — restituisci.

using System.Collections.Concurrent;

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

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

// Riempire il pool con connessioni iniziali
for (int i = 0; i < 3; i++)
{
    connectionPool.Push(new Connection());
}
Console.WriteLine($"Connessioni nel pool: {connectionPool.Count}"); // Output: Connessioni nel pool: 3

void UseConnection()
{
    if (connectionPool.TryPop(out Connection conn))
    {
        Console.WriteLine($"[Pool] Usata connessione: {conn.Id}");
        // Simulazione di lavoro con la connessione
        Thread.Sleep(50); 
        connectionPool.Push(conn); // Restituiamo al pool
        Console.WriteLine($"[Pool] Restituita connessione: {conn.Id}. Nel pool: {connectionPool.Count}");
    }
    else
    {
        Console.WriteLine("[Pool] Il pool è vuoto. Creo una nuova connessione.");
        // Di solito qui si crea una nuova connessione se il pool è vuoto
        connectionPool.Push(new Connection()); 
    }
}

// Esecuzione dell'esempio in Main:
Task.Run(() => UseConnection());
Task.Run(() => UseConnection());
Task.Run(() => UseConnection());
Thread.Sleep(500);

In questo esempio più thread possono prendere e restituire connessioni al pool in modo sicuro.

5. Esempi d'uso e confronto con ConcurrentQueue

ConcurrentStack<T> si usa quando:

• L'ordine LIFO è critico (per esempio la cronologia delle azioni per la funzione "Undo").
• Si vuole accesso rapido agli elementi più recentemente aggiunti (spesso rimangono "caldi" nella cache della CPU).
• Si implementano algoritmi basati su stack (visita DFS di grafi, parsing sintattico delle espressioni).

Confronto e scelta della collezione adatta

ConcurrentQueue

ConcurrentStack

La scelta tra ConcurrentQueue e ConcurrentStack dipende interamente dall'ordine di elaborazione richiesto nel tuo scenario producer-consumer. Entrambe le collezioni offrono alte prestazioni e thread-safety "out of the box", liberandoti dalla gestione manuale della sincronizzazione e aiutandoti a costruire sistemi multithread scalabili.