Cómo probar código concurrente

Código en:JavaC#PythonJavaScript

En resumen: El código concurrente se prueba con una prueba de estrés de muchas ejecuciones, una barrera de inicio, entrelazados forzados y la comprobación de un invariante, porque una ejecución en verde no demuestra nada.

Problema

En mi máquina pasa

El contador de ventas de la lección Condición de carrera no tiene cerrojo. Un compañero escribe una prueba: cuatro hilos venden 1000 veces cada uno, y al final los vendidos deben ser 4000:

class Counter:
    def __init__(self):
        self.sold = 0

    def sell(self):
        self.sold += 1


def sell_once():
    counter = Counter()

    def buyer():
        for _ in range(1000):
            counter.sell()

    threads = [threading.Thread(target=buyer) for _ in range(4)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    return counter.sold


print("Vendidos (esperado 4000):")
print(sell_once())

Salida

Vendidos (esperado 4000):
4000

Para ellos la prueba pasa. En la CI también pasa. El código entra en las rebajas, y en la primera hora la tienda ha vendido más artículos de los que muestra el contador.

En Python esto es especialmente fácil: por el GIL, los hilos se alternan cada 5 ms más o menos, y self.sold += 1 son unas pocas instrucciones. La probabilidad de que el cambio caiga justo entre la lectura y la escritura es pequeña. Pequeña, pero no cero: en un servidor con muchas peticiones por hora, ocurre.

Una prueba en verde de código concurrente no demuestra casi nada: en esta ejecución los hilos simplemente no se encontraron justo entre la lectura y la escritura. ¿Cómo escribes una prueba que atrape un error así?

La idea

Una prueba de estrés: muchas ejecuciones y una barrera

Tres cosas hacen visible el error:

• Muchas ejecuciones. La prueba no es una ejecución sino 20 (o 1000 en un trabajo nocturno de CI). Cuentas cuántas salen en rojo.
• Muchos hilos a la vez. Más hilos que núcleos, para que se alternen a mitad del trabajo.
• Una barrera al inicio. Cada hilo espera hasta que lleguen todos, y arrancan juntos. Así se solapan todo el tiempo, no solo al final.

class Counter:
    def __init__(self):
        self.sold = 0

    def sell(self):
        left = self.sold
        time.sleep(0)  # aquí entra otro hilo
        self.sold = left + 1


def lost_sales():
    counter = Counter()

    def buyer():
        for _ in range(1000):
            counter.sell()

    threads = []
    for _ in range(4):
        t = threading.Thread(target=buyer)
        t.start()
        threads.append(t)
    for t in threads:
        t.join()
    return counter.sold != 4000

Paso a paso, en palabras

1. La prueba inicia 4 hilos, cada uno vende 1000 veces sin cerrojo. Sin barrera, cada hilo arranca en cuanto se crea.
2. 20 ejecuciones: 20 de 20 en rojo. Aquí time.sleep(0) abre un hueco ancho, así que falla incluso sin barrera. En código real el hueco es estrecho.
3. Una barrera: cada hilo espera en la salida hasta que lleguen los cuatro. Luego corren juntos.
4. Con la barrera: 20 de 20 en rojo, igual que sin ella: sleep(0) mantiene ancho el hueco. Con un hueco estrecho, la barrera decide.

La prueba con una barrera:

class Counter:
    def __init__(self):
        self.sold = 0

    def sell(self):
        left = self.sold
        time.sleep(0)  # aquí entra otro hilo
        self.sold = left + 1


def lost_sales():
    counter = Counter()
    start = threading.Barrier(4, timeout=1)

    def buyer():
        start.wait()
        for _ in range(1000):
            counter.sell()

    threads = []
    for _ in range(4):
        t = threading.Thread(target=buyer)
        t.start()
        threads.append(t)
    for t in threads:
        t.join()
    return counter.sold != 4000


failed = 0
for run in range(20):
    if lost_sales():
        failed += 1
print("Ejecuciones con ventas perdidas (de 20):")
print(failed)

Salida

Ejecuciones con ventas perdidas (de 20):
20

La salida es «como máximo 20»: cuántas ejecuciones salen en rojo depende de la máquina. En mi máquina son 20 de 20, con y sin barrera.

threading.Barrier(4, timeout=1): wait() espera hasta que 4 hilos lo hayan llamado, y entonces los suelta a todos. Si uno no llega en 1 segundo, lanza BrokenBarrierError en lugar de colgarse.

Aquí sell libera el GIL con time.sleep(0) entre la lectura y la escritura (como en la lección sobre condiciones de carrera). El hueco es tan ancho que la prueba falla incluso sin barrera. En código real el hueco son unas pocas instrucciones, y allí la barrera y las muchas ejecuciones deciden si lo ves siquiera.

La misma prueba contra el contador con cerrojo:

class Counter:
    def __init__(self):
        self.sold = 0
        self.lock = threading.Lock()

    def sell(self):
        with self.lock:
            left = self.sold
            time.sleep(0)  # aquí entra otro hilo
            self.sold = left + 1


def lost_sales():
    counter = Counter()
    start = threading.Barrier(4, timeout=1)

    def buyer():
        start.wait()
        for _ in range(1000):
            counter.sell()

    threads = []
    for _ in range(4):
        t = threading.Thread(target=buyer)
        t.start()
        threads.append(t)
    for t in threads:
        t.join()
    return counter.sold != 4000

Salida

Ejecuciones con ventas perdidas: 0 de 20

Más precisamente, para curiosos

Una prueba de estrés encuentra errores, pero no demuestra que no los haya. «0 de 20» significa «esta vez no lo vimos». Por eso en la CI estas pruebas se ejecutan miles de veces, en máquinas con muchos núcleos, y la revisión de código pregunta aparte: qué se comparte y qué cerrojo lo protege.

Cómo es en los otros lenguajes: en Java la barrera es CyclicBarrier, en C# Barrier, en Node la escribes con Atomics.wait y Atomics.notify. En lugar de una barrera también es común un threading.Event que activa el hilo principal.

Pruébalo

Encuentra el error

Otro compañero escribió una prueba de estrés sin barrera. Pasa 20 de 20 veces, en cada ejecución, en cada máquina, aunque el contador no tiene cerrojo:

class Counter:
    def __init__(self):
        self.sold = 0

    def sell(self):
        left = self.sold
        time.sleep(0)  # aquí entra otro hilo
        self.sold = left + 1


def lost_sales():
    counter = Counter()

    def buyer():
        for _ in range(1000):
            counter.sell()

    for _ in range(4):
        t = threading.Thread(target=buyer)
        t.start()
        t.join()
    return counter.sold != 4000

Salida

Ejecuciones con ventas perdidas: 0 de 20

Elige la línea con el error.

1class Counter:2 def __init__(self):3 self.sold = 045 def sell(self):6 left = self.sold7 time.sleep(0) # aquí entra otro hilo8 self.sold = left + 191011def lost_sales():12 counter = Counter()1314 def buyer():15 for _ in range(1000):16 counter.sell()1718 for _ in range(4):19 t = threading.Thread(target=buyer)20 t.start()21 t.join()22 return counter.sold != 4000

Comprobar

Pruébalo

Completa

¿Qué línea hace que los hilos arranquen juntos?

class Counter:
    def __init__(self):
        self.sold = 0

    def sell(self):
        left = self.sold
        time.sleep(0)  # aquí entra otro hilo
        self.sold = left + 1


def lost_sales():
    counter = Counter()
    start = threading.Barrier(4, timeout=1)

    def buyer():
        
        for _ in range(1000):
            counter.sell()

    threads = []
    for _ in range(4):
        t = threading.Thread(target=buyer)
        t.start()
        threads.append(t)
    for t in threads:
        t.join()
    return counter.sold != 4000

Comprobar

La idea

Forzar el entrelazado

Una prueba de estrés encuentra el error a veces. Para una prueba que falle siempre, la prueba elige el orden de los hilos. Para eso el código tiene una costura: un lugar donde la prueba puede detener el hilo. En el programa real la costura no hace nada.

El escenario es el error de la lección sobre condiciones de carrera: quedan 10 artículos, dos compradores compran uno cada uno.

1. El primer comprador lee 10 y se detiene en la costura.
2. El segundo comprador pasa de punta a punta: lee 10, escribe 9.
3. El primero continúa y escribe su 10 − 1 = 9.

class Stock:
    def __init__(self, count):
        self.count = count
        self.pause = lambda: None

    def buy(self):
        left = self.count
        self.pause()  # una costura para la prueba: aquí la prueba puede detener el hilo
        self.count = left - 1


def forced_test(stock):
    read = threading.Event()
    go = threading.Event()

    def pause():
        if threading.current_thread().name == "first":
            read.set()
            go.wait(timeout=1)

    stock.pause = pause
    first = threading.Thread(target=stock.buy, name="first")
    second = threading.Thread(target=stock.buy, name="second")
    first.start()
    read.wait(timeout=1)  # first ha leído 10 y espera
    second.start()
    second.join(timeout=0.2)  # second compra hasta el final o espera el cerrojo
    go.set()  # first escribe
    first.join(timeout=1)
    second.join(timeout=1)
    verdict = "PASA" if stock.count == 8 else "FALLA"
    return f"Existencias: {stock.count} (esperado 8): {verdict}"

Salida

Existencias: 9 (esperado 8): FALLA

La costura es el campo pause. La prueba le pone una función que detiene solo al hilo llamado first: activa read y espera go. Un threading.Event es un latch: set() lo abre, wait(timeout=1) espera a que se abra, como máximo 1 segundo.

Las existencias son 9 en lugar de 8, en cada ejecución. Ahora la corrección: with self.lock: alrededor de la lectura y la escritura:

class Stock:
    def __init__(self, count):
        self.count = count
        self.pause = lambda: None
        self.lock = threading.Lock()

    def buy(self):
        with self.lock:
            left = self.count
            self.pause()  # una costura para la prueba: aquí la prueba puede detener el hilo
            self.count = left - 1

Salida

Existencias: 8 (esperado 8): PASA

La prueba es la misma. El primer comprador mantiene el cerrojo mientras está detenido en la costura. El segundo espera el cerrojo, así que second.join(timeout=0.2) se agota antes de que haya comprado. Solo cuando el primero escribe 9 y suelta el cerrojo, el segundo lee 9 y escribe 8. Una prueba que fallaba siempre antes de la corrección y pasa siempre después es justo lo que quieres en la CI.

Pruébalo

Tres en la costura

El mismo Stock sin cerrojo. Ahora la costura espera hasta que los tres compradores hayan leído las existencias:

¿Qué imprimirá el programa?

stock = Stock(10)
all_read = threading.Barrier(3, timeout=1)
stock.pause = all_read.wait
buyers = [threading.Thread(target=stock.buy) for _ in range(3)]
for t in buyers:
    t.start()
for t in buyers:
    t.join()
print(f"Existencias: {stock.count}")

Respuesta

Existencias: 9

Existencias: 7

Existencias: 8

Comprobar

La idea

Comprueba el invariante, no el orden

Con 8 compradores, quién compró qué depende de la ejecución. Una prueba que espera «el primer comprador compró 13» falla incluso con código correcto. Por eso la prueba comprueba un invariante: algo que debe ser verdad en cualquier orden.

Para la tienda es: las existencias más el número de vendidos es lo que había al principio, las existencias no bajan de cero, y las compras de los compradores sumadas dan el número de vendidos. 8 compradores hacen 50 intentos cada uno por 100 artículos:

shop = Shop(100)
bought = [0] * 8
start = threading.Barrier(8, timeout=1)


def buyer(me):
    start.wait()
    for _ in range(50):
        if shop.try_buy():
            bought[me] += 1


buyers = [threading.Thread(target=buyer, args=(i,)) for i in range(8)]
for t in buyers:
    t.start()
for t in buyers:
    t.join()
print(f"Existencias: {shop.stock}, vendidos: {shop.sold}")
print(check("existencias + vendidos = 100", shop.stock + shop.sold == 100))
print(check("existencias >= 0", shop.stock >= 0))
print(check("comprados = vendidos", sum(bought) == shop.sold))

Salida

Existencias: 0, vendidos: 100
existencias + vendidos = 100: PASA
existencias >= 0: PASA
comprados = vendidos: PASA

La tienda:

class Shop:
    def __init__(self, stock):
        self.stock = stock
        self.sold = 0
        self.lock = threading.Lock()

    def try_buy(self):
        with self.lock:
            if self.stock == 0:
                return False
            self.stock -= 1
            self.sold += 1
            return True

Cada hilo escribe en su propio lugar bought[me], así que la lista no necesita cerrojo. El total es correcto en cualquier orden; quién compró cuánto no está fijado.

400 intentos por 100 artículos significan que las existencias terminan en 0 en cualquier orden. Eso también es un invariante, por eso la salida es la misma en cada ejecución.

La idea

Un timeout: el deadlock hace fallar la prueba

Dos almacenes se traspasan mercancía entre sí. Cada traslado bloquea el almacén del que saca, luego el que recibe, en orden contrario. Eso es un deadlock: cada hilo sostiene un cerrojo y espera al otro para siempre.

Una prueba sin timeout no falla: se cuelga. La CI espera una hora hasta que la mata, y nadie sabe qué prueba era. Así que una prueba con hilos siempre espera con un timeout:

sofia = threading.Lock()
plovdiv = threading.Lock()


def transfer(source, target):
    with source:
        time.sleep(0.05)
        with target:
            pass  # trasladar la mercancía


def finishes_within(seconds, *parts):
    threads = [threading.Thread(target=part, daemon=True) for part in parts]
    for t in threads:
        t.start()
    end = time.monotonic() + seconds
    for t in threads:
        t.join(timeout=max(0, end - time.monotonic()))
    return not any(t.is_alive() for t in threads)


ok = finishes_within(1, lambda: transfer(sofia, plovdiv), lambda: transfer(plovdiv, sofia))
print("Prueba «traslado entre almacenes»: " + ("PASA" if ok else "no terminó en 1 s: FALLA"))

Salida

Prueba «traslado entre almacenes»: no terminó en 1 s: FALLA

join(timeout=...) espera como máximo ese tiempo. Luego is_alive() dice si el hilo sigue ejecutándose. daemon=True deja que Python termine aunque los dos hilos estén bloqueados para siempre.

Más precisamente, para curiosos

Los frameworks de pruebas traen el timeout ya hecho: pytest-timeout (@pytest.mark.timeout(1)) o faulthandler.dump_traceback_later, que imprime la pila de cada hilo cuando la prueba se atasca. Cómo se escribe una prueba en el lenguaje, mira la lección sobre pruebas.

También hay herramientas que buscan estos errores por ti. Aquí solo las nombramos, sin usarlas:

• jcstress (Java): ejecuta pequeños trozos de código millones de veces en cada orden que permite la JVM y cuenta los resultados.
• ThreadSanitizer (C, C++, Go, Rust) y el detector de carreras de Go (go test -race): vigilan cada lectura y escritura durante la prueba y dicen «estos dos hilos tocan lo mismo sin cerrojo», aunque esta vez el resultado fuera correcto.
• Microsoft.Coyote (.NET): ejecuta código asíncrono y multihilo bajo su propio planificador, prueba muchos órdenes y puede reproducir el orden que falló.
• faulthandler (Python): cuando una prueba se atasca, imprime dónde está cada hilo.

Ejercicio

Una prueba de estrés para la caja

Resuélvelo en tu propio editor.

La caja de las rebajas tiene sell(euros). Escribe una prueba de estrés:

• 20 ejecuciones. En cada una, 4 hilos con una barrera al inicio; cada uno vende 500 veces por 2 €. La suma esperada es 4000 €.
• Una ejecución en la que los hilos no terminan en 1 segundo cuenta como timeout, y la prueba continúa.
• Al final imprime cuántas ejecuciones tuvieron una suma incorrecta y cuántas un timeout, y si la caja pasa.

Salida

Ejecuciones: 20
Con suma incorrecta: 0
Con timeout: 0
La caja: PASA

Más tareas

• Rompe la caja. Quita el cerrojo y mira cuántas ejecuciones salen en rojo. Ejecútala varias veces: el número cambia.
• Un entrelazado forzado. Añade una costura en sell y escribe una prueba que falle siempre sin el cerrojo y pase con él.
• Un deadlock en la caja. Añade un segundo cerrojo (la caja y el almacén) en orden contrario y comprueba que tu prueba informa de un timeout en lugar de colgarse.

Test

Ponte a prueba

1. La prueba del contador sin cerrojo pasa. ¿Qué significa?

El contador es seguro en esta máquinaCasi nada: en esta ejecución los hilos no se solaparonEl contador es seguro

2. ¿Para qué sirve la barrera al inicio de la prueba de estrés?

Para que todos los hilos arranquen juntos y se solapenPara que la prueba sea más rápidaPara que los hilos corran uno tras otro

3. La prueba de estrés pasa 20 de 20 veces. ¿Es seguro el código?

Sí, 20 ejecuciones bastanNo está demostrado: la prueba muestra errores, no demuestra que no los hayaSí, si la barrera estaba puesta

4. ¿Cómo haces que una prueba falle siempre con la misma condición de carrera?

Más ejecucionesThread.sleep con pausas distintasLatches en una costura del código: el primer hilo se detiene entre la lectura y la escritura hasta que el segundo pasa

5. ¿Por qué una prueba con hilos debe tener un timeout?

Si no, un deadlock deja la prueba colgada para siempre y detiene toda la CIPorque los hilos no terminan solosPara que sea más rápida

Comprobar

Empezar la lección de nuevo

Lecciones relacionadas

• Se basa en: Operaciones atómicas y objetos inmutables · Deadlock · Grupo de hilos y tareas · async no es un hilo
• Ejercicios sobre el tema: Concurrencia y condiciones de carrera