Ir al contenido
wannacode

Productor y consumidor

En resumen: Productor y consumidor es un patrón en el que unos hilos ponen tareas en una cola limitada y otros las sacan; una cola llena hace esperar al productor (backpressure).

Problema

Black Friday

A las 10:00 empieza la venta y el sitio recibe 50 pedidos por segundo. El almacén empaca 10. Hay tres malas maneras de manejarlo.

El sitio empaca por sí mismo. El cliente hace clic en «Pedir» y espera hasta que alguien empaca la caja. El sitio es tan rápido como el almacén.

El sitio pone los pedidos en una lista simple y el almacén los toma de ahí. Dos empacadores pueden tomar el mismo pedido y otro puede perderse: eso es una condición de carrera. Si añades un cerrojo, un empacador con la lista vacía tiene que comprobar una y otra vez si ha llegado algo.

La lista no tiene límite. El sitio está contento: nunca espera. Pero cada segundo la lista crece 40 pedidos. Después de una hora tiene 144 000, el último ha esperado una hora y se acaba la memoria.

La idea

Una cola con espacio limitado

Entre los dos lados va una cola. El productor pone tareas al final, los consumidores las toman del frente. Este es el patrón productor y consumidor. Los dos lados no se conocen: el sitio no sabe cuántos empacadores hay, y los empacadores no saben de dónde vienen los pedidos.

La cola es bloqueante y tiene espacio limitado:

  • si está vacía, el consumidor espera, sin hacer girar el procesador, hasta que llegue algo;
  • si está llena, el productor espera hasta que un consumidor libere un lugar;
  • el cerrojo está dentro de la cola: cada pedido va a exactamente un consumidor.
En C# esto es BlockingCollection<T> con boundedCapacity: 3. Add espera si está llena. GetConsumingEnumerable() es un bucle que toma mientras haya algo y espera si está vacía. Vive en System.Collections.Concurrent, como ConcurrentBag, donde juntamos los pedidos empacados.

Míralo

El sitio y los empacadores

static void Front(List<Order> orders,
                  BlockingCollection<Order> queue)
{
    foreach (var order in orders)
    {
        queue.Add(order); // espera si está llena
    }
    queue.CompleteAdding(); // fin: los empacadores paran
}


static void Pack(BlockingCollection<Order> queue,
                 ConcurrentBag<Order> packed)
{
    // espera si está vacía
    foreach (var order in queue.GetConsumingEnumerable())
    {
        packed.Add(order);
    }
}
Paso a paso, en palabras
  1. El sitio (productor) pone pedidos en una cola con 3 lugares; dos empacadores (consumidores) los toman del frente. Está vacía, así que esperan en GetConsumingEnumerable.
  2. El sitio pone 1001 y 1002 al final de la cola.
  3. Cada empacador toma el pedido del frente y lo empaca. Empacar es más lento que recibir pedidos.
  4. El sitio pone 1003, 1004 y 1005. Los 3 lugares están ocupados.
  5. 1006 no cabe: Add espera y el sitio deja de recibir pedidos. Esto es backpressure: el lado lento frena al rápido.
  6. El empacador 1 termina con 1001 y toma 1003. Se libera un lugar y el put que esperaba deja entrar 1006.
  7. Y así sigue: los empacadores toman del frente, el sitio añade al final y en la cola nunca hay más de 3.
  8. Se acabaron los pedidos. CompleteAdding le dice a la cola que no llegará nada más.
  9. Vacía y completada: GetConsumingEnumerable termina y cada foreach se detiene. Sin CompleteAdding los empacadores esperarían para siempre.
  10. 8 de 8, cada uno una vez: 3 exprés por 90 €, 5 normales por 180 €. Quién empaca cuál cambia de una ejecución a otra; los totales no.

El pedido:

record Order(int Id, int Euro, bool Express);

El sitio y los empacadores:

static void Front(List<Order> orders,
                  BlockingCollection<Order> queue)
{
    foreach (var order in orders)
    {
        queue.Add(order); // espera si está llena
    }
    queue.CompleteAdding(); // fin: los empacadores paran
}


static void Pack(BlockingCollection<Order> queue,
                 ConcurrentBag<Order> packed)
{
    // espera si está vacía
    foreach (var order in queue.GetConsumingEnumerable())
    {
        packed.Add(order);
    }
}
front y pack corren en tres hilos: un sitio y dos empacadores. El sitio pone los ocho pedidos y termina, y los empacadores toman hasta que reciben una señal para parar.
var orders = new List<Order>
{
    new(1001, 25, false),
    new(1002, 40, true),
    new(1003, 15, false),
    new(1004, 60, false),
    new(1005, 30, true),
    new(1006, 45, false),
    new(1007, 20, true),
    new(1008, 35, false),
};
var queue = new BlockingCollection<Order>(boundedCapacity: 3);
var packed = new ConcurrentBag<Order>();
var threads = new List<Thread>
{
    new Thread(() => Front(orders, queue)),
    new Thread(() => Pack(queue, packed)),
    new Thread(() => Pack(queue, packed)),
};
foreach (var thread in threads) thread.Start();
foreach (var thread in threads) thread.Join();
Report(orders, packed);

El informe comprueba cosas que no dependen de qué empacador tomó qué pedido:

static void Report(List<Order> orders, ConcurrentBag<Order> packed)
{
    int distinct = packed.Select(o => o.Id).Distinct().Count();
    Console.WriteLine($"Empacados: {packed.Count} de {orders.Count}");
    Console.WriteLine($"Cada pedido una vez: {(distinct == orders.Count ? "sí" : "no")}");
    foreach (var express in new[] { true, false })
    {
        var group = packed.Where(o => o.Express == express).ToList();
        string kind = express ? "Exprés" : "Normales";
        Console.WriteLine($"{kind}: {group.Count} pedidos, {group.Sum(o => o.Euro)} €");
    }
}
Salida
Empacados: 8 de 8
Cada pedido una vez: sí
Exprés: 3 pedidos, 90 €
Normales: 5 pedidos, 180 €

La salida es la misma en cada ejecución. Cuántos pedidos tomó cada empacador cambia, por eso el programa no lo imprime.

La idea

Cómo paran los empacadores

El empacador espera en GetConsumingEnumerable() hasta que llega un pedido. Si el sitio simplemente termina, los empacadores esperan para siempre y el programa nunca acaba.

Por eso, al final el sitio llama a CompleteAdding(): «no llegará nada más». Después, Add lanza una excepción, y GetConsumingEnumerable() termina en cuanto la cola queda vacía. Todo foreach sobre ella se detiene solo, sin importar cuántos empacadores haya. Los pedidos que ya están dentro se empacan primero.

La otra manera, que encontrarás en código ajeno, es la píldora venenosa (poison pill): un valor especial que no es un pedido, uno por consumidor. En C# no la necesitas.

La idea

Backpressure

Una cola llena hace esperar al sitio. No es un error, es el objetivo: el lado lento empuja hacia atrás al rápido. Esto se llama backpressure. El sitio no puede adelantarse al almacén más que el tamaño de la cola, la memoria no crece y el último pedido espera como máximo lo que tarda en empacarse los 3 que tiene delante.

Esperar no es la única respuesta a una cola llena. También hay un método que no espera, sino que dice si funcionó:

TryAdd(order) devuelve false si no hay lugar. TryAdd(order, 1000) espera como máximo un segundo.

Con él, el sitio puede mostrar «Mucha demanda, vuelve a intentarlo en un minuto» en vez de colgarse. Lo que importa es que la decisión es tuya: el pedido no desaparece en silencio.

Pruébalo

Predice la salida

Una cola con 2 lugares y sin empacadores. ¿Qué imprime?

var queue = new BlockingCollection<string>(boundedCapacity: 2);
Console.WriteLine(queue.TryAdd("1001"));
Console.WriteLine(queue.TryAdd("1002"));
Console.WriteLine(queue.TryAdd("1003"));
Console.WriteLine(queue.Take());
Console.WriteLine(queue.TryAdd("1003"));
Console.WriteLine(queue.Count);
Respuesta

Pruébalo

Completa

¿A qué llama el sitio al final para que los empacadores paren?
queue.(); // fin: los empacadores paran

Pruébalo

Encuentra el error

Un colega no quiere que el sitio espere y cambia una línea. Ahora el programa nunca se cuelga, pero el informe a veces dice «Empacados: 5 de 8». ¿En qué línea está el error?

Elige la línea con el error.

Míralo

Hilos, Workers y mensajes

En código C# nuevo también encontrarás Channel<T> (System.Threading.Channels). Channel.CreateBounded<Order>(3) es la misma cola con 3 lugares, pero para código async: await writer.WriteAsync(order) espera sin bloquear un hilo, writer.Complete() es CompleteAdding, y await foreach (var o in reader.ReadAllAsync()) es el consumidor. BlockingCollection bloquea hilos; Channel los suelta mientras espera.

Ejercicio

Dos entradas, tres empacadores

Resuelve la tarea en tu editor.

Los pedidos llegan de dos entradas: el sitio (2001–2006) y la aplicación móvil (3001–3006). Cada entrada es un productor aparte, y hay tres empacadores. La cola tiene 4 lugares. Las píldoras (CompleteAdding) llegan solo cuando las dos entradas han terminado. Imprime:

Salida
Empacados: 12 de 12
Cada pedido una vez: sí
Total: 325 €

Más tareas

  • Rechazo. Haz la cola de 1 lugar y que el sitio use TryAdd(order, 10). Cuenta los pedidos rechazados e imprime «aceptados + rechazados = 12».
  • Por tipo. Los pedidos exprés deben empacarse primero. ¿Ayuda dos colas y un empacador que mira primero la exprés? ¿Qué pasa con las píldoras?
  • Una cadena. Añade una segunda cola después de los empacadores: ellos ponen en ella las cajas empacadas, y un mensajero las toma. ¿Cuántas píldoras hacen falta ahora y quién las pone?

Test

Ponte a prueba

1. La cola entre el sitio y los empacadores está llena. ¿Qué hace put?
2. ¿Por qué tiene la cola un límite? Sin él, el sitio nunca espera.
3. Dos sitios ponen pedidos, tres empacadores los toman. ¿Cuántas píldoras hacen falta y cuándo se ponen?
4. ¿Cómo sabes que el programa con la cola es correcto, si el orden de empaque es distinto en cada ejecución?

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