6 Wait Notify¶

Condiciones en un bloque sincronizado¶

Cuando usamos varios bloques sincronizados sobre el mismo cerrojo intrínseco debemos tener en cuenta que si existe alguna condición por la que el hilo no pueda continuar la ejecución de la sección crítica, el hilo debería suspender su ejecución y liberar cerrojo, en espera de que deje de cumplirse la condición que no le permite continuar. De esta manera, otro hilo podrá ejecutar el mismo u otro bloque sincronizado sobre protegido por dicho cerrojo. Esto es de vital importancia cuando el hecho de que la condición que bloquea al hilo deje de cumplirse depende de que otro hilo acceda a otra sección crítica protegida por el mismo cerrojo, ya que si el primer hilo no liberara dicho cerrojo, el segundo nunca podría entrar en dicha sección crítica y por tanto la condición que bloquea al primer hilo nunca dejaría de cumplirse.

Por otra parte, el hecho de que el hilo se suspenda esperando que la condición que lo bloquea deje de cumplirse implica que habrá que notificarle en el futuro cuando debería volver a estar listo para comprobar de nuevo la condición. Así, cuando otro hilo realice alguna acción que suponga que la condición que bloquea al primer hilo pueda dejar de cumplirse deberá notificarlo al sistema para que el primer hilo trate de volver a adquirir el cerrojo y retome la sección crítica, volviendo a comprobar la condición, de manera que si deja de cumplirse, continúe con la ejecución de la sección crítica.

Esta condición de bloqueo recibe habitualmente el nombre de centinela (guard en inglés) o condición de paso.

El ejemplo típico de lo que acabamos de explicar es el llamado problema del productor-consumidor, que se describe a continuación. Tenemos un buffer de datos en el que uno o más productores almacenan datos y del que uno o más consumidores extraen datos. Como el buffer es una estructura de datos compartida por ellos, deberemos controlar su acceso usando el mecanismo de sincronización descrito en el apartado anterior. Sin embargo, tenemos más limitaciones, como el hecho de que un proveedor no puede almacenar más datos en el buffer si éste está lleno o que un consumidor no puede extraer datos del buffer si está vacío.

Para este tipo de situaciones, Java proporciona una serie de métodos específicos en la clase Object. Una vez obtenido mediante synchronized el cerrojo intrínseco asociado a un determinado objeto, un hilo puede llamar al método wait() sobre dicho objeto. Al llamar a wait(), el hilo se suspende y se libera el cerrojo que protege el bloque sincronizado, lo que permite que otros hilos ejecuten otros bloques de código sincronizados protegidos por el mismo cerrojo.

El método wait(timeoutMillis) se encuentra sobrecargado para que podamos especificar un tiempo máximo que un hilo puede estar suspendido esperando a que deje de cumplirse la condición de bloqueo, transcurrido el cual el hilo será reactivado automáticamente. Si se pasa un valor negativo para el parámetro timeoutMilllis se lanza la excepción IllegalArgumentException. Sin embargo, bajo nuestro punto de vista, el método wait(timoutMillis) tiene el inconveniente de que no nos informa de si se ha dejado de esperar porque nos han notificado o porque se ha consumido el tiempo máximo especificado.

Como la mayoría de métodos bloqueantes, los método wait() y wait(timeoutMillis) lanzan la excepción InterruptedException si el hilo correspondiente es interrumpido mientras está suspendido en espera de que deje de cumplirse la condición de bloqueo, o si ya había sido marcado como interrumpido antes de llamar a wait() o wait(timeoutMillis), reactivando inmediatamente el hilo en ambos casos.

Si se llama a wait() o wait(timeoutMillis) desde hace fuera de un bloque o método synchronized, se lanzará la excepción IllegalMonitorStateException, ya que el empleo de estos métodos está asociado al cerrojo intrínseco del objeto sobre el que se ejecutan.

Para reactivar un hilo que ha sido bloqueado mediante wait() o wait(timeoutMillis), otro hilo debe llamar al método notify() o notifyAll(), también desde dentro de un bloque de código protegido por el mismo cerrojo intrínseco, para notificar que se ha producido algún "evento" que hace que sea posible que alguna condición de bloqueo (guard) deje de cumplirse, y por tanto los hilos que estaban bloqueados en dicha condición puedan continuar.

Como consecuencia de la llamada a notifyAll() todos los hilos que hubieran sido bloqueados mediante wait() o wait(timeMillis) en dicho cerrojo intrínseco por cualquier condición de bloqueo serán reactivados.

Un aspecto importante es que cuando se hace notifyAll() no hay garantía de que el evento concreto que se está notificando sea por el que están interesados los hilos bloqueados en dicho cerrojo. De hecho, existe otro método, llamado notify() que reactiva un sólo hilo, pero no se recomienda su uso cuando haya distintas condiciones de espera, porque el hilo reactivado podría no estar interesado en dicho "evento" y otros hilos que sí estarían interesados no serían reactivados.

Además, el método notify() no proporciona ninguna garantía sobre qué hilo será reactivado si varios hilos están esperando en el mismo objeto monitor, lo que puede llevar a que un determinado hilo sufra inanición, es decir que nunca sea reactivado cuando se llama notify() porque siempre es reactivado algún otro hilo que estuviera esperando. Para solucionar este problema, como estudiaremos más adelante, deberemos usar un objeto ReentrantLock con el modo fair activado y definir un objeto Condition por el que esperar, asociado a dicho cerrojo.

Si existe una única condición de espera en dicho objeto monitor y el evento sólo va a permitir a un único hilo continuar, entonces es más adecuado llamar a notify() en vez de llamar a notifyAll(), ya que estaríamos reactivando a varios hilos innecesariamente, a lo que se conoce como el thundering herd problem (problema de la "manada atronadora").

Este problema ocurre cuando un gran número de procesos o hilos en espera de un evento ("la manada") son despertados repentinamente cuando se produce el evento, pero en realidad un sólo hilo es capaz de sacar provecho de ello y avanzar. Todos los hilos lo intentarán, compitiendo por los recursos, consumiendo ciclos de procesamiento, pero sólo uno de ellos lo conseguirá. Ese consumo de recursos cesará cuando los hilos descubran que no pueden avanzar y vuelvan a esperar, es decir, "cuando la manada vuelve a calmarse".

Al usar notifyAll() debemos tener en cuenta que todos los hilos reactivados compiten por volver a adquirir el cerrojo, lo que puede implicar que para cuando un hilo reactivado consiga adquirir el cerrojo para continuar con la ejecución de su sección crítica la condición de bloqueo que lo bloqueaba vuelva a cumplirse, a pesar de que en algún momento dejó de cumplirse y otro hilo lo notificó. Por este motivo se debe evaluar la condición de bloqueo en un bucle while y no en un simple if.

synchronized(this) {
    while(condicion_bloqueo) {
        wait();
    }
    // ...
}

Proyecto WaitNotify¶

En este proyecto crearemos una aplicación que ejecuta dos hilos, un hilo productor dedicado a producir donuts y almacenarlos en una bandeja de la pastelería y otro hilo consumidor dedicado a extraer elementos de la bandeja de la pastelería y consumirlos. El problema radica en que ambos pueden acceder simultáneamente a la misma estructura de datos para agregar o extraer un donut, lo que implica que deberemos proveer mecanismos de sincronización sobre dichos accesos. Por otra parte, un productor no puede agregar un donut a la estructura de datos si la bandeja ya esta llena y un consumidor no puede extraer un elemento de la bandeja si ésta está vacía. Estas condiciones de bloqueo se ejecutan dentro del bloque sincronizado, por lo que deberemos usar el método wait() para suspender los hilos mientras no puedan continuar y usar el método notifyAll() para reactivarlos cuando deba volver a evaluarse la condición centinela.

public class Main {

    public static void main(String[] args) {
        Bakery bakery = new Bakery();
        Thread doughnutProducerThread = new Thread(new DoughnutProducer(bakery), "Doughnut producer");
        Thread doughnutConsumerThread = new Thread(new DoughnutConsumer(bakery), "Doughnut consumer");
        doughnutProducerThread.start();
        doughnutConsumerThread.start();
    }

}

import java.util.ArrayList;

public class Bakery {

    private static final int TRAY_CAPACITY = 10;

    private final ArrayList<Integer> tray = new ArrayList<>();

    public void addToTray(Integer doughnut) throws InterruptedException {
        synchronized (this) {
            while (tray.size() >= TRAY_CAPACITY) {
                System.out.println("Producer waiting for the tray to have room");
                wait();
            }
            tray.add(doughnut);
            System.out.printf("Producer puts doughnut #%d on the tray\n", doughnut);
            notifyAll();
        }
    }

    public Integer extractFromTray() throws InterruptedException {
        Integer doughnut;
        synchronized (this) {
            while (tray.isEmpty()) {
                System.out.println("Consumer waiting for the tray to have a doughnut");
                wait();
            }
            doughnut = tray.remove(0);
            System.out.printf("Consumer extracts doughnut #%d from tray\n", doughnut);
            notifyAll();
            return doughnut;
        }
    }

}

import java.util.Objects;
import java.util.concurrent.TimeUnit;

public class DoughnutConsumer implements Runnable {

    private final Bakery bakery;

    public DoughnutConsumer(Bakery bakery) {
        Objects.requireNonNull(bakery);
        this.bakery = bakery;
    }

    @Override
    public void run() {
        Integer doughnut;
        while (!Thread.currentThread().isInterrupted()) {
            try {
                doughnut = bakery.extractFromTray();
            } catch (InterruptedException e) {
                System.out.println("Consumer has been interrupted while extracting from tray");
                return;
            }
            try {
                eat(doughnut);
            } catch (InterruptedException e) {
                System.out.println("Consumer has been interrupted while eating");
                return;
            }
        }
        System.out.println("Consumer has been interrupted");
    }

    private void eat(int doughnut) throws InterruptedException {
        System.out.printf("Consumer is eating doughnut #%d\n", doughnut);
        TimeUnit.SECONDS.sleep(1);
    }

}

import java.util.Objects;
import java.util.concurrent.TimeUnit;

public class DoughnutProducer implements Runnable {

    private final Bakery bakery;
    private int doughnutNumber;

    public DoughnutProducer(Bakery bakery) {
        Objects.requireNonNull(bakery);
        this.bakery = bakery;
    }

    @Override
    public void run() {
        Integer doughnut;
        while (!Thread.currentThread().isInterrupted()) {
            try {
                doughnut = makeDoughnut();
            } catch (InterruptedException e) {
                System.out.println("Producer has been interrupted while making a doughnut");
                return;
            }
            try {
                bakery.addToTray(doughnut);
            } catch (InterruptedException e) {
                System.out.println("Producer has been interrupted while adding a doughnut to the tray");
                return;
            }
        }
        System.out.println("Producer has been interrupted");
    }

    private int makeDoughnut() throws InterruptedException {
        Integer doughnut = ++doughnutNumber;
        System.out.printf("Producer is making doughnut #%d\n", doughnut);
        TimeUnit.SECONDS.sleep(3);
        return doughnut;
    }

}