Ir al contenido
wannacode

Grupo de hilos y tareas

En resumen: Un grupo de hilos son unos pocos trabajadores delante de una cola de tareas en lugar de un hilo nuevo por tarea; en .NET es el grupo compartido con Task.Run, Parallel.For y Task.WhenAll.

Problema

Un hilo por cada imagen

La tienda empieza unas rebajas. Los vendedores suben imágenes de sus productos, y cada imagen necesita una miniatura para la lista de productos. Hacer una miniatura es trabajo que calcula: recorre cada píxel (mira trabajo que espera y trabajo que calcula).

La primera versión inicia un hilo por cada imagen subida:

var threads = new List<Thread>();
foreach (string image in uploads)
{
    var thread = new Thread(() => SaveThumbnail(image));
    thread.Start();
    threads.Add(thread);
}

Con 20 imágenes todo funciona en tu computadora: «Listas: 20 miniaturas». En la primera hora de las rebajas llegan 10 000 a la vez.

  • Cada hilo necesita su propia pila (normalmente hasta 1 MB) y un lugar en el sistema operativo. Miles de hilos se comen la memoria del servidor.
  • Aunque haya memoria, 10 000 hilos en 8 núcleos solo se turnan. El procesador pierde tiempo cambiando de uno a otro, y cada imagen se vuelve más lenta.

El servidor no se vuelve más lento; se cae. Y un servidor caído tampoco atiende a los compradores.

La idea

Unos pocos trabajadores y una cola

Un grupo de hilos son unos pocos trabajadores, creados una vez, y delante de ellos una cola de tareas. Una tarea es «reduce esta imagen». Un trabajador libre toma la siguiente tarea de la cola, la termina y vuelve por una nueva.

  • El número de trabajadores es constante. 10 000 imágenes no hacen 10 000 hilos, sino 10 000 tareas en la cola. Una tarea es un objeto pequeño; un hilo es caro.
  • Los trabajadores se reutilizan. No se crea ni se destruye un hilo por cada imagen.
  • La cola absorbe el pico. Cuando las subidas llegan más rápido de lo que los trabajadores pueden atender, las tareas esperan en lugar de tumbar el servidor.

Esto es productor y consumidor: el código que acepta las subidas pone tareas en la cola, y los trabajadores son los consumidores.

Ya conociste el grupo en la lección sobre hilos: Task.Run, Task<int> y .Result. Aquí veremos qué pasa por dentro, cómo ordenar los resultados, qué pasa con los errores y cuántos trabajadores necesitas.

Míralo

Tres trabajadores

Hacer una miniatura aquí es una pausa, 2 ms por cada KB, y el tamaño sale del nombre de la imagen. Así la salida es la misma en cada ejecución:

// Crear una miniatura: aquí es una pausa, y el tamaño sale del nombre.
static int Thumbnail(string image)
{
    if (image.StartsWith("dañada"))
    {
        throw new ArgumentException("archivo dañado");
    }
    int kb = 20 + 3 * image.Length;
    Thread.Sleep(2 * kb);
    return kb;
}

Ocho imágenes, un grupo de tres trabajadores:

int[] sizes = new int[images.Length];
var options = new ParallelOptions
{
    MaxDegreeOfParallelism = 3,
};
Parallel.For(0, images.Length, options, i =>
{
    sizes[i] = Thumbnail(images[i]);
});
int total = 0;
for (int i = 0; i < images.Length; i++)
{
    Console.WriteLine($"{images[i]}: {sizes[i]} KB");
    total += sizes[i];
}
Console.WriteLine($"Total: {total} KB");
Paso a paso, en palabras
  1. 8 imágenes esperan. Parallel.For usará como máximo 3 hilos del grupo de .NET.
  2. Parallel.For reparte los índices. Los tres hilos toman las tres primeras imágenes.
  3. Hilo 3 termina primero: Top es la tercera imagen, y su resultado va al tercer lugar. El trabajador no se detiene: toma la siguiente tarea.
  4. La cola está vacía. Hilo 1 terminó y no tiene nada que tomar, así que espera. Los demás terminan sus últimas imágenes.
  5. Cada resultado está en sizes[i], así que el orden es el de las imágenes. Total 379 KB.
string[] images = { "Bolso.jpg", "Zapato.jpg", "Top.jpg", "Falda.jpg", "Pantalón.jpg", "Camisa.jpg", "Botín.jpg", "Boa.jpg" };
int[] sizes = new int[images.Length];
var options = new ParallelOptions
{
    MaxDegreeOfParallelism = 3,
};
Parallel.For(0, images.Length, options, i =>
{
    sizes[i] = Thumbnail(images[i]);
});
int total = 0;
for (int i = 0; i < images.Length; i++)
{
    Console.WriteLine($"{images[i]}: {sizes[i]} KB");
    total += sizes[i];
}
Console.WriteLine($"Total: {total} KB");
Salida
Bolso.jpg: 47 KB
Zapato.jpg: 50 KB
Top.jpg: 41 KB
Falda.jpg: 47 KB
Pantalón.jpg: 56 KB
Camisa.jpg: 50 KB
Botín.jpg: 47 KB
Boa.jpg: 41 KB
Total: 379 KB
  • .NET tiene un grupo de hilos (ThreadPool) para todo el programa y decide por sí mismo cuántos hilos mantener. No lo creas ni lo cierras.
  • Parallel.For ejecuta el cuerpo para cada índice en el grupo y espera a todos. MaxDegreeOfParallelism = 3 dice «como máximo tres a la vez»: ese es el grupo de tres.
  • Cada tarea escribe en su propio lugar sizes[i], así que no hace falta ningún lock y los resultados quedan en el orden de las imágenes.
Más precisamente, para curiosos

.NET decide por sí mismo cuántos hilos mantiene el grupo: empieza con tantos como núcleos y añade más si las tareas se quedan mucho tiempo sin calcular. ThreadPool.GetMaxThreads y SetMinThreads lo ajustan, pero rara vez los necesitas.

Parallel.ForEach(images, options, image => …) es lo mismo para una colección, y Parallel.ForEachAsync es para tareas con await. El hilo que llama a Parallel.For también trabaja y espera al final, así que en una aplicación web no lo llames en el hilo de la petición.

Cómo es en los otros lenguajes: en Java el grupo es un objeto que creas y cierras (ExecutorService), en Python es ThreadPoolExecutor, y Node no tiene un grupo de Workers ya hecho.

Pruébalo

Encuentra el error

Un compañero reescribió las miniaturas con un grupo, y las rebajas no fueron más rápidas. Seis imágenes tardan lo mismo que sin grupo:

var watch = Stopwatch.StartNew();
var sizes = new List<int>();
foreach (string image in images)
{
    Task<int> size = Task.Run(() => Thumbnail(image));
    sizes.Add(await size);
}
Console.WriteLine($"Tardó más de 0,5 s: {watch.ElapsedMilliseconds > 500}");
Salida
Tardó más de 0,5 s: True

Elige la línea con el error.

La idea

Resultados en el orden de entrada

Las tareas terminan en un orden que no conoces: depende de las imágenes, de la carga, de qué trabajador estaba libre. Así que nunca imprimas «en el orden en que terminaron» si quieres la misma salida cada vez.

Dos formas dan el mismo resultado en cada ejecución:

  • En el orden de entrada. Guardas las promesas en una lista en el orden de las tareas y las esperas en ese orden. Así funciona el código de arriba. En C# Task.WhenAll da un arreglo en el orden de las tareas, no en el orden en que terminaron:
  • Una suma o una cuenta. Si solo necesitas el total, el orden no importa: 47 + 50 es lo mismo que 50 + 47.
Task<int>[] tasks = images.Select(image => Task.Run(() => Thumbnail(image))).ToArray();
int[] sizes = await Task.WhenAll(tasks);
for (int i = 0; i < images.Length; i++)
{
    Console.WriteLine($"{images[i]}: {sizes[i]} KB");
}
Salida
Bolso.jpg: 47 KB
Zapato.jpg: 50 KB
Top.jpg: 41 KB
Falda.jpg: 47 KB

A veces quieres los resultados en el orden en que terminaron: para una barra de «3 de 8 listas» o para mostrar las primeras imágenes listas. Task.WhenAny los da así. Entonces recoges solo cosas que no dependen del orden:

var running = images.Select(image => Task.Run(() => Thumbnail(image))).ToList();
int count = 0;
int total = 0;
while (running.Count > 0)
{
    Task<int> done = await Task.WhenAny(running);
    running.Remove(done);
    total += await done;
    count++;
}
Console.WriteLine($"Listas: {count} de {images.Length}, en total: {total} KB");
Salida
Listas: 8 de 8, en total: 379 KB
Task.WhenAny da la primera tarea terminada de la lista. La quitas y esperas la siguiente. Con muchas tareas esto es O(n²), por eso .NET 9 añadió Task.WhenEach, que las da una por una.

La idea

Un error dentro de una tarea

Una de las imágenes está dañada, y thumbnail lanza una excepción. No sale en el hilo que inició la tarea: ese hilo siguió su camino hace rato. El grupo la captura y la guarda en la promesa del resultado. Sale cuando esperas ese resultado:

string[] images = { "Bolso.jpg", "Zapato.jpg", "dañada.jpg", "Top.jpg" };
var tasks = images.Select(image => Task.Run(() => Thumbnail(image))).ToList();
for (int i = 0; i < images.Length; i++)
{
    try
    {
        int kb = await tasks[i];
        Console.WriteLine($"{images[i]}: {kb} KB");
    }
    catch (ArgumentException e)
    {
        Console.WriteLine($"{images[i]}: error: {e.Message}");
    }
}
Salida
Bolso.jpg: 47 KB
Zapato.jpg: 50 KB
dañada.jpg: error: archivo dañado
Top.jpg: 41 KB
await lanza la excepción misma (ArgumentException). .Result la lanzaría envuelta en una AggregateException. Las otras imágenes no sufren.

Pruébalo

Adónde fue el error

Dos de las imágenes están dañadas. ¿Qué imprime el programa?

string[] images = { "Bolso.jpg", "dañada-1.jpg", "dañada-2.jpg" };
Task<int[]> all = Task.WhenAll(images.Select(image => Task.Run(() => Thumbnail(image))));
try
{
    await all;
}
catch (ArgumentException e)
{
    Console.WriteLine($"Capturado: {e.Message}");
}
Console.WriteLine($"Errores en la tarea: {all.Exception!.InnerExceptions.Count}");
Respuesta

La idea

Cuántos trabajadores

Depende de lo que hagan:

  • Trabajo que calcula (miniaturas, PDF, compresión): tantos como núcleos. Un noveno trabajador en 8 núcleos no tiene dónde ejecutarse; solo se turna con los demás.
  • Trabajo que espera (correo, pago, base de datos): más. Mientras un trabajador espera a la red, el núcleo queda libre para otro. Si una tarea espera W ms y calcula C ms, caben unos 1 + W / C trabajadores en un núcleo.
static int PoolSize(int cores, int waitMs, int computeMs)
{
    return cores * (1 + waitMs / computeMs);
}


Console.WriteLine($"Miniaturas (solo cálculo): {PoolSize(8, 0, 50)}");
Console.WriteLine($"Correos (90 ms de espera, 10 ms de cálculo): {PoolSize(8, 90, 10)}");
Salida
Miniaturas (solo cálculo): 8
Correos (90 ms de espera, 10 ms de cálculo): 80

El número de núcleos viene de Environment.ProcessorCount. La fórmula es un punto de partida, no una ley: mide. Si los correos pasan por un servidor con un límite de 20 conexiones, 80 trabajadores solo esperarán en una cola delante de él.

Pruébalo

Completa

¿Cuántos trabajadores para una tarea que espera waitMs y calcula computeMs?

static int PoolSize(int cores, int waitMs, int computeMs)
{
    return cores * (1 + );
}
Más precisamente, para curiosos

En C# no mantienes hilos para el trabajo que espera: await libera el hilo mientras la petición espera a la red (la lección async y await). Por eso un servidor web atiende miles de peticiones con un grupo pequeño. Los recursos detrás (conexiones a la base de datos) se limitan, por ejemplo con un SemaphoreSlim.

Ejercicio

Las facturas de las rebajas

Resuelve la tarea en tu editor.

Después de las rebajas cada pedido necesita una factura. Invoice(amount) crea el PDF (aquí una pausa de 50 ms) y devuelve el importe con 20 % de IVA. Un pedido con un importe de 0 o menos no es válido, y la función lanza una excepción con el texto «importe no válido».

Los pedidos son del 1001 al 1008 con los importes 50 €, 125 €, 80 €, 0 €, 240 €, 35 €, 90 € y 15 €. Haz las facturas en un grupo de 4 trabajadores. Imprímelas en el orden de los pedidos, y la no válida con el texto del error. Al final: la suma de las facturas buenas y cuántas fallaron.

Salida
Pedido 1001: 60 €
Pedido 1002: 150 €
Pedido 1003: 96 €
Pedido 1004: error: importe no válido
Pedido 1005: 288 €
Pedido 1006: 42 €
Pedido 1007: 108 €
Pedido 1008: 18 €
Total: 762 €, con error: 1

Más tareas

  • Una cola limitada. Haz que el grupo mantenga como máximo 10 tareas esperando en la cola. ¿Qué pasa con la undécima? (Channel.CreateBounded o SemaphoreSlim)
  • Progreso. Imprime «3 de 8 listas» después de cada factura terminada, y al final otra vez la lista ordenada.
  • Un timeout. Una factura que tarda más de 1 segundo cuenta como fallida, sin detener a las demás.

Test

Ponte a prueba

1. ¿Por qué un grupo de pocos hilos y no un hilo por cada imagen subida?
2. Un grupo de 3 trabajadores reduce 8 imágenes. La tercera termina primera. ¿Dónde está su tamaño cuando las lees en el orden de la lista con get, result, Promise.all o sizes[i]?
3. Una tarea del grupo lanza una excepción. ¿Cuándo la ves?
4. El servidor tiene 8 núcleos. Cada correo espera 90 ms a la red y calcula 10 ms. ¿Cuántos trabajadores mantienen ocupados los núcleos, más o menos?
5. ¿Cuántos trabajadores para hacer miniaturas, si solo calcula y el servidor tiene 8 núcleos?

Creado por Ivo Gergov como parte de la campaña de Evo Tech por una educación accesible en programación ·