Biblioteca TPL y programación paralela en C#

Aprende Task Parallel Library y programación paralela en C# con Task, Parallel y ejemplos prácticos.

En .NET, la Task Parallel Library (TPL) simplifica la ejecución paralela aprovechando los procesadores multinúcleo. Con TPL, se pueden utilizar herramientas como Task, Parallel, las colecciones Concurrent y PLINQ para acelerar las operaciones intensivas en CPU. Este artículo cubre los conceptos básicos, los casos de uso adecuados, la administración de cancelaciones y excepciones, así como ejemplos prácticos.

¿Qué es TPL? Comenzando con Task

Una Task representa una unidad de trabajo. Task.Run envía tareas intensivas en CPU al grupo de subprocesos (thread pool). Puede iniciar varias tareas simultáneamente y esperar a que todas terminen con Task.WhenAll.


using System;
using System.Threading.Tasks;

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

        await Task.WhenAll(t1, t2, t3);
        Console.WriteLine("Todas las tareas han terminado.");
    }

    static void Trabajo(string nombre, int ms)
    {
        Console.WriteLine($"{nombre} ha comenzado");
        Task.Delay(ms).Wait(); // espera simulada (no intensiva en CPU)
        Console.WriteLine($"{nombre} ha finalizado");
    }
}

Nota: para operaciones de espera I/O, prefiera async/await; Task.Run está pensado para trabajos intensivos en CPU.

Parallel.For / Parallel.ForEach

La clase Parallel divide automáticamente los bucles y los distribuye entre subprocesos. Al acceder a variables compartidas, utilice técnicas seguras para subprocesos como 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 =>
        {
            // Uso de Interlocked para evitar condiciones de carrera
            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 numeros = new List<int> { 1, 2, 3, 4, 5 };
        Parallel.ForEach(numeros, n =>
        {
            Console.WriteLine($"{n} procesado (Hilo: {Environment.CurrentManagedThreadId})");
        });
    }
}

ParallelOptions: Nivel de paralelismo y cancelación

Con MaxDegreeOfParallelism puede limitar el grado de paralelismo, y con CancellationToken puede añadir soporte para cancelación.


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, // según número de núcleos
            CancellationToken = cts.Token
        };

        // Cancelar después de 2 segundos
        Task.Run(async () => { await Task.Delay(2000); cts.Cancel(); });

        try
        {
            Parallel.For(0, 1000, po, i =>
            {
                po.CancellationToken.ThrowIfCancellationRequested();
                TrabajoCPU(i); // trabajo intensivo en CPU
            });
        }
        catch (OperationCanceledException)
        {
            Console.WriteLine("Las operaciones fueron canceladas.");
        }
    }

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

Acumulación con variables locales (Local Init/Finally)

En lugar de bloquear una variable compartida, use un acumulador local para cada hilo y combine los resultados al final.


using System;
using System.Threading.Tasks;

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

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

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

Colecciones Concurrentes

En escenarios de productor/consumidor, ConcurrentBag<T>, ConcurrentQueue<T> y ConcurrentDictionary<TKey,TValue> proporcionan acceso sin bloqueo o con bajo nivel de bloqueo.


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 contador = 0;
        while (bag.TryTake(out _)) contador++;

        Console.WriteLine($"Elementos extraídos: {contador}");
    }
}

PLINQ (Parallel LINQ)

Use AsParallel() para ejecutar consultas en paralelo sobre grandes colecciones. Si se necesita mantener el orden, agregue AsOrdered(); de lo contrario, la ejecución sin orden mejora el rendimiento.


using System;
using System.Linq;

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

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

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

Manejo de excepciones

En llamadas Parallel pueden lanzarse múltiples excepciones; se manejan con AggregateException. En flujos basados en Task, las excepciones aparecen durante el 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($"Error: {e.GetType().Name} - {e.Message}");
        }
    }
}

async/await vs Parallel

async/await: para operaciones I/O (archivo, red, base de datos) sin bloquear la interfaz de usuario.
Parallel/TPL: para tareas intensivas en CPU que aprovechan los múltiples núcleos.
Evite usar Task.Run innecesariamente con tareas I/O; el verdadero beneficio se obtiene con trabajos intensivos en CPU.

Consejos de rendimiento y buenas prácticas

Minimice el estado compartido; combine resultados con acumuladores locales siempre que sea posible.
Evite bloqueos innecesarios; use Interlocked o colecciones Concurrent cuando sea necesario.
Use MaxDegreeOfParallelism para evitar un cambio de contexto excesivo.
En aplicaciones de interfaz gráfica, evite llamadas bloqueantes (.Wait() / .Result).

Ejemplo: Procesamiento de imágenes en paralelo

El siguiente ejemplo simula el procesamiento paralelo de archivos de imagen en una carpeta. En un escenario real, se podrían realizar operaciones intensivas en CPU, como filtrado o generación de miniaturas.


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

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

        var po = new ParallelOptions { MaxDegreeOfParallelism = Environment.ProcessorCount };
        Parallel.ForEach(archivos, po, archivo =>
        {
            // Simulación de procesamiento intensivo en CPU
            var nombre = Path.GetFileName(archivo);
            Console.WriteLine($"Procesando: {nombre}");
            Thread.SpinWait(3000000); // Simulación
            Console.WriteLine($"Completado: {nombre}");
        });

        Console.WriteLine("Todos los archivos han sido procesados.");
    }
}

TL;DR (Resumen)

TPL simplifica la programación paralela: Task, Parallel, PLINQ.
Parallel.For/ForEach distribuye los bucles entre varios núcleos de CPU.
MaxDegreeOfParallelism y CancellationToken proporcionan control y cancelación.
Seguridad de subprocesos esencial: use Interlocked, acumuladores locales y colecciones Concurrent.
async/await ↔ Parallel: elija según la naturaleza del trabajo (I/O o CPU).