Implementar una cola bloqueante acotada y segura para subprocesos (Thread-Safe Bounded Blocking Queue)

Problema y cuándo aplica

Implementa BoundedBlockingQueue<E>. Su constructor acepta una capacidad positiva. put(item) agrega elementos en orden FIFO y espera mientras la cola esté llena. take() remueve desde la cabeza y espera mientras la cola esté vacía. Múltiples productores y consumidores pueden invocar ambos métodos de forma concurrente. Ningún elemento puede perderse, retornarse dos veces ni retornarse después de un elemento que haya ingresado antes. Si un subproceso es interrumpido mientras adquiere el bloqueo o espera en una condición, el método lanza InterruptedException.

El ejercicio no permite envolver ArrayBlockingQueue. La API base excluye operaciones no bloqueantes offer/poll, tiempos de espera (timeouts), remoción masiva y semántica de apagado (shutdown), y no promete equidad estricta (fairness) entre los subprocesos en espera. Estas son preguntas de seguimiento. Al igual que BlockingQueue de Java, esta implementación rechaza null, ya que las API de colas a menudo usan null para indicar que no hay ningún elemento disponible.

Este es un problema de codificación de estructuras de datos concurrentes. La indexación del arreglo es la parte fácil. La verdadera prueba es si el candidato puede enunciar un contrato auditable: qué sincronización protege el estado compartido, por qué no se pueden perder las notificaciones, por qué un subproceso reactivado debe volver a verificar su condición y en qué instante exacto una operación surte efecto para otros subprocesos.

Qué evalúa el entrevistador

La primera señal es si el candidato separa la seguridad (safety) de la vitalidad (liveness). La seguridad requiere que el tamaño permanezca en [0, capacity], que cada elemento se remueva como máximo una vez y que el orden de remoción coincida con el orden de inserción. La vitalidad requiere que un productor tenga la oportunidad de continuar cuando una cola llena deja de estarlo y que un consumidor tenga la oportunidad cuando una cola vacía deja de estarlo. La interrupción también debe ser capaz de cancelar una espera.

La segunda señal es si las primitivas de sincronización se corresponden con los predicados de estado. Un solo bloqueo protege el arreglo, la cabeza, la cola y el tamaño, por lo que verificar una condición y cambiar el estado ocurren en la misma sección crítica. notFull representa size < capacity; notEmpty representa size > 0. Los productores solo esperan por la primera condición, los consumidores solo por la segunda, y un cambio de estado que cruce el límite notifica al rol opuesto.

La tercera señal es si el candidato puede explicar el uso de while en lugar de usarlo como un patrón memorizado. Condition permite reactivaciones espurias (spurious wakeups). Incluso después de un signal real, otro subproceso competidor puede readquirir el bloqueo primero y llenar o vaciar la cola nuevamente. Una vez que el subproceso reactivado readquiere el bloqueo, debe volver a evaluar el predicado. Ser notificado no demuestra que la condición todavía se cumpla.

Finalmente, el entrevistador puede indagar sobre el diseño: cómo los índices del anillo preservan FIFO, por qué la actualización del tamaño pertenece al punto de linealización, por qué un solo bloqueo evita el interbloqueo (deadlock) por orden de bloqueos, cuándo es suficiente signal y por qué los timeouts y el apagado requieren nueva semántica de API.

Preguntas para aclarar antes de responder

• ¿Qué capacidades y elementos son válidos? La capacidad debe ser mayor que cero y los elementos null son rechazados.
• ¿Las esperas por lleno o vacío deben hacer spin? No. Un subproceso en espera debe liberar el bloqueo y esperar en una condición. No debe consumir CPU en un bucle de sondeo activo (spin) ni dormir reteniendo el bloqueo.
• ¿Cómo debe comportarse la interrupción? Tanto put como take propagan InterruptedException. No suprimen la interrupción ni modifican la cola después de que falla la operación interrumpida.
• ¿Se requiere equidad estricta (fairness)? No en el problema base. El ReentrantLock no equitativo por defecto puede permitir que un subproceso posterior adquiera el bloqueo primero. Un bloqueo equitativo altera el rendimiento (throughput) y el comportamiento de planificación.
• ¿La cola necesita apagado (shutdown)? No en el problema base. Si lo hiciera, define si los elementos existentes se pueden drenar, si los subprocesos en espera reciben una excepción o un resultado especial, y quién despierta a todos los que esperan.
• ¿Son la misma garantía el FIFO linealizado de elementos y el FIFO de los subprocesos en espera? No. Los elementos salen en el orden de linealización de los puts exitosos. Eso no significa que los productores o consumidores bloqueados sean admitidos en orden de llegada.
• ¿Se puede utilizar una clase de la biblioteca estándar? El código de producción normalmente debería preferir un ArrayBlockingQueue probado. La implementación manual aquí sirve específicamente para evaluar los invariantes de concurrencia y la semántica de espera de condición.

Estructura de respuesta en 30 segundos

«Utilizaré un arreglo de tamaño fijo como búfer circular, con head, tail y size representando la siguiente posición de lectura, la siguiente posición de escritura y la cantidad actual de elementos. Un único ReentrantLock protege todo el estado compartido, y dos objetos Condition representan que no está vacía y que no está llena. put espera dentro de un while (size == capacity), inserta e incrementa size, y luego notifica a un consumidor. take simétricamente espera a que no esté vacía, limpia la cabeza y decrementa size, y luego notifica a un productor. Cada verificación y transición ocurre bajo el mismo bloqueo. Dado que await libera atómicamente el bloqueo y lo readquiere antes de retornar, no hay una ventana de notificación perdida entre verificar y dormir. Cada operación exitosa es de tiempo O(1), con un espacio de O(capacity)».

Análisis detallado paso a paso

Deriva la representación a partir del cuello de botella ingenuo. Un arreglo simple que desplaza los elementos restantes tras cada remoción hace que take sea de orden O(n). Mantener un índice de lectura que solo crezca desperdicia el prefijo liberado. Un arreglo circular fijo reutiliza las ranuras liberadas: head apunta a la siguiente posición de lectura, tail apunta a la siguiente posición de escritura y size es la cantidad actual de elementos. Un índice vuelve a cero tras alcanzar el final, por lo que ni la inserción ni la remoción desplazan elementos existentes.

La implementación mantiene cuatro invariantes:

1. 0 <= size <= items.length.
2. Comenzando en head, las primeras size ranuras en orden circular contienen la secuencia FIFO no consumida.
3. tail == (head + size) % items.length. Cuando la cola está llena, head == tail, por lo que size distingue el estado lleno del vacío.
4. Cada lectura o escritura de items, head, tail y size ocurre mientras se retiene el mismo bloqueo.

Aquí está la implementación central:

import java.util.Objects;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public final class BoundedBlockingQueue<E> {
  private final Object[] items;
  private final ReentrantLock lock = new ReentrantLock();
  private final Condition notEmpty = lock.newCondition();
  private final Condition notFull = lock.newCondition();

  private int head;
  private int tail;
  private int size;

  public BoundedBlockingQueue(int capacity) {
    if (capacity <= 0) {
      throw new IllegalArgumentException("capacity must be positive");
    }
    items = new Object[capacity];
  }

  public void put(E item) throws InterruptedException {
    Objects.requireNonNull(item, "item");
    lock.lockInterruptibly();
    try {
      while (size == items.length) {
        notFull.await();
      }

      items[tail] = item;
      tail = (tail + 1) % items.length;
      size++;
      notEmpty.signal();
    } finally {
      lock.unlock();
    }
  }

  @SuppressWarnings("unchecked")
  public E take() throws InterruptedException {
    lock.lockInterruptibly();
    try {
      while (size == 0) {
        notEmpty.await();
      }

      E item = (E) items[head];
      items[head] = null;
      head = (head + 1) % items.length;
      size--;
      notFull.signal();
      return item;
    } finally {
      lock.unlock();
    }
  }
}

Objects.requireNonNull se ejecuta antes de bloquear porque solo valida un argumento y no depende del estado compartido. lockInterruptibly() hace que la espera para adquirir el bloqueo sea interrumpible. Una vez que el método entra en el bloque try, finally libera el bloqueo retenido por el subproceso actual ya sea tras un retorno normal, una interrupción durante una espera de condición o una excepción en tiempo de ejecución.

La garantía esencial de await() es que libera atómicamente el bloqueo asociado y entra en espera, para luego readquirir dicho bloqueo antes de retornar. Un productor no debe desbloquear manualmente y registrarse como en espera solo después. Eso crearía una ventana en la que un consumidor podría liberar espacio y enviar una notificación antes de que el productor haya comenzado efectivamente a esperar, perdiendo la notificación para siempre. Una variable de condición combina «liberar y comenzar a esperar» en una sola acción de sincronización y cierra esa ventana.

El bucle while maneja dos casos diferentes. Uno es la reactivación espuria permitida por la especificación. El otro es la competencia ordinaria: dos consumidores pueden ser despertados, uno readquiere el bloqueo primero y remueve el único elemento, y el segundo encuentra una cola vacía cuando finalmente obtiene el bloqueo. Una notificación de condición solo indica que el estado pudo haber cambiado; el predicado de estado es la autoridad para proceder.

El punto de linealización de un put exitoso es la transición bajo bloqueo que agrega el elemento y pasa size de k a k + 1. Para take, es la remoción correspondiente y la transición de k a k - 1. El bloqueo asegura que otro subproceso pueda observar o bien el estado completo anterior a la transición o bien el estado completo posterior a ella, nunca una escritura en el arreglo sin su correspondiente actualización de tamaño. Liberar y después adquirir el mismo bloqueo también establece visibilidad de memoria, cumpliendo el objetivo de happens-before especificado para pasar elementos a través de colas concurrentes estándar.

¿Por qué usar dos condiciones? Con un solo conjunto de espera, un take podría despertar a otro consumidor aunque la cola continúe vacía, mientras que un productor que podría usar la nueva ranura permanece dormido. Separar notEmpty y notFull permite que cada transición notifique únicamente al rol que ahora puede proceder. Una operación base put o take de un solo elemento crea solo un nuevo elemento o ranura, por lo que signal() es suficiente; el subproceso reactivado aún vuelve a verificar en un bucle while. Si una operación modifica múltiples ranuras, o si el apagado requiere que cada subproceso en espera observe un nuevo estado, reconsidera signalAll().

La corrección se deduce por inducción sobre los invariantes. Inicialmente, head = tail = size = 0. Una inserción se ejecuta solo cuando size < capacity; escribe en tail, avanza la cola e incrementa el tamaño exactamente una vez, por lo que se mantiene dentro de la capacidad y se agrega después de todos los elementos no consumidos. Una remoción se ejecuta solo cuando size > 0; lee en head, limpia la ranura, avanza la cabeza y decrementa el tamaño exactamente una vez, por lo que retorna el elemento no consumido más antiguo. El bloqueo serializa estas transiciones, haciendo que cualquier orden de ejecución de muchos productores y consumidores sea equivalente a alguna ejecución secuencial válida.

Cada put o take exitoso realiza un número fijo de operaciones de arreglos, enteros y sincronización, por lo que su trabajo es de orden O(1) excluyendo el tiempo de espera. El arreglo fijo usa un espacio de O(capacity). Bajo contención, la espera por el bloqueo y los cambios de contexto pueden dominar la latencia; la notación asintótica no captura ese costo. Un diseño de un solo bloqueo no tiene ciclos multibloqueo, pero los llamadores aún pueden crear un problema mayor de orden de bloqueos si invocan un método bloqueante mientras retienen algún otro bloqueo.

Las pruebas no pueden limitarse a un ejemplo de un solo subproceso. Cubre alternancia lleno/vacío con capacidad uno, múltiples vueltas completas al anillo, almacenamiento y remoción por separado de elementos iguales repetidos, aceptación de una cadena vacía mientras se rechaza null, valores únicos producidos por varios productores y drenados por varios consumidores, ausencia de valores faltantes o duplicados en el conjunto final, preservación del orden dentro del flujo de cada productor individual e interrupción tanto de un put bloqueado como de un take bloqueado. Una capacidad enorme asigna un arreglo igualmente grande en el constructor y puede fallar por falta de memoria; la implementación base no es diferida (lazy) y no debe disfrazar una falla de asignación como una cola vacía. Las pruebas concurrentes deben coordinar los inicios con una barrera o un latch y usar un límite de tiempo (deadline) en el ejecutor de pruebas para detectar bloqueos permanentes. Un sleep corto no demuestra que un subproceso haya alcanzado su estado de espera.

Ejemplo de respuesta de alta calidad

«Primero delimitaría el contrato base a una capacidad positiva, elementos no nulos, put/take bloqueantes, orden FIFO, múltiples productores y consumidores, y esperas interrumpibles. Los timeouts, el apagado y la equidad estricta requieren valores de retorno y semántica de estados adicionales, por lo que los mantendría fuera de la implementación central.

La representación es un arreglo circular fijo. head es la siguiente posición de lectura, tail la siguiente posición de escritura y size elimina la ambigüedad de lleno versus vacío cuando head == tail. Un solo ReentrantLock protege los cuatro campos compartidos. El predicado para notEmpty es size > 0, y el predicado para notFull es size < capacity.

put adquiere el bloqueo de forma interrumpible, espera por espacio disponible dentro de un bucle while, escribe en la cola e incrementa size, y luego notifica a un consumidor. take simétricamente espera a que no esté vacía, remueve la cabeza, limpia la referencia y decrementa size, y luego notifica a un productor. El bucle es obligatorio porque las esperas de condición pueden despertarse de forma espuria y porque otro subproceso puede hacer que el predicado vuelva a ser falso mientras el subproceso reactivado compite por readquirir el bloqueo.

await libera atómicamente el bloqueo y comienza la espera, lo que evita perder una notificación entre la comprobación y el momento de dormir. El punto de linealización de una operación exitosa es la transición de estado bajo bloqueo que modifica la pertenencia a la cola y el valor de size. El mismo bloqueo garantiza que otros subprocesos vean un estado previo o posterior completo. La inducción sobre los invariantes de capacidad, FIFO en anillo y mismo bloqueo demuestra que la implementación no desborda, no duplica remociones ni reordena elementos.

Excluyendo el tiempo de bloqueo, ambos métodos son de orden O(1), y el espacio es O(capacity). Iniciaría múltiples productores y consumidores con un latch, verificaría el conjunto de identificadores únicos y el orden por cada productor, y probaría por separado las rutas de lleno, vacío e interrupción».

Errores comunes

• Verificar lleno o vacío con un if → la condición aún puede ser falsa tras una reactivación espuria o competencia por el bloqueo → vuelve a comprobar el predicado de estado en un bucle while.
• Desbloquear manualmente tras la verificación y luego esperar → puede ocurrir un cambio de estado antes del registro de espera, perdiendo la notificación → utiliza Condition.await() asociado al mismo bloqueo.
• Sincronizar el arreglo, los índices y size por separado → otro subproceso puede observar un estado intermedio contradictorio → protege toda la comprobación y transición con un único bloqueo.
• Usar un solo conjunto de espera con un signal arbitrario → la señal puede despertar a un subproceso del mismo rol que no puede avanzar → mantén condiciones separadas para notEmpty y notFull.
• Hacer spin o dormir reteniendo el bloqueo → el subproceso que podría cambiar la condición no puede adquirirlo → una espera de condición debe liberar el bloqueo.
• Suprimir InterruptedException → los llamadores no pueden cancelar el trabajo y el subproceso puede quedar bloqueado indefinidamente → declara y propaga la interrupción, y libera el bloqueo en finally.
• Usar solo head == tail para detectar ambos estados → los estados de anillo lleno y vacío son indistinguibles → mantén un size protegido por bloqueo.
• Dejar la referencia removida en el arreglo → el arreglo retiene un objeto consumido más tiempo del necesario → asigna la ranura a null después de leerla.
• Equiparar el orden FIFO de elementos con la equidad de subprocesos → el bloqueo predeterminado no completa las llamadas en espera en orden de llegada → describe el ordenamiento de elementos y la política de planificación por separado.
• Afirmar soporte de apagado en la implementación base → los subprocesos en espera no tienen un estado cerrado observable y podrían no despertar nunca → define el contrato de apagado antes de agregar estado y notificación por difusión (broadcast).

Preguntas de seguimiento y respuestas

Pregunta de seguimiento 1: ¿Cómo agregarías offer y poll con límite de tiempo?

Convierte la duración restante a nanosegundos y llama a awaitNanos(remaining) dentro del mismo bucle while del predicado. Tras cada retorno, continúa con el tiempo restante que reporte. Si el valor no es positivo y el predicado sigue siendo falso, retorna falla. Reutilizar el timeout completo tras cada reactivación espuria podría extender la espera real indefinidamente. La API también debe distinguir un timeout de un elemento nulo, otra razón para rechazar null.

Pregunta de seguimiento 2: ¿Cómo implementarías shutdown()?

Define primero la máquina de estados. Por ejemplo, rechaza nuevos puts tras el cierre pero permite que los elementos existentes se drenen; cuando esté cerrada y vacía, take lanza una excepción dedicada o retorna un resultado explícito. shutdown debe modificar la bandera de cerrado bajo el mismo bloqueo y llamar a signalAll() en ambas condiciones para que todos los que esperan puedan readquirir el bloqueo y observar el cierre. Cada predicado en bucle de espera necesita una rama para el estado cerrado; agregar solo un campo booleano es insuficiente.

Pregunta de seguimiento 3: ¿Por qué usar signal() aquí y cuándo usarías signalAll()?

Una inserción base crea un elemento consumible y una remoción crea una ranura libre. Despertar a un solo subproceso del rol opuesto es suficiente para progresar y reduce la competencia inútil. La inserción masiva, remoción masiva, cambios dinámicos de capacidad o el apagado pueden hacer que varios subprocesos en espera sean elegibles a la vez, por lo que normalmente necesitan signalAll() o una cantidad de notificaciones que coincida con el cambio de estado. Independientemente de cuántos se despierten, la reverificación en el bucle while sigue siendo obligatoria.

Pregunta de seguimiento 4: ¿Cómo proporcionarías equidad (fairness)?

new ReentrantLock(true) hace que la adquisición del bloqueo favorezca al subproceso que lleva más tiempo esperando, pero aun así no garantiza un orden absoluto de finalización en tiempo real para put/take; la interrupción y la planificación también influyen. Una política equitativa generalmente reduce el barging y el riesgo de inanición (starvation), pero puede reducir el throughput. Paga y verifica ese costo únicamente cuando el contrato del llamador realmente requiera el ordenamiento de los subprocesos en espera.

Pregunta de seguimiento 5: ¿Podría ser esta una cola lock-free?

Una cola MPMC acotada y lock-free usualmente necesita números de secuencia atómicos, CAS y una prueba de ordenamiento de memoria mucho más compleja. Su comportamiento «bloqueante» todavía necesita un mecanismo de parking y reactivación; el sondeo continuo (spin) con CAS por sí solo no constituye una cola bloqueante. El diseño introduce problemas de ABA, false sharing, garantías de progreso y consideraciones del modelo de memoria de la plataforma. A menos que las mediciones demuestren que el bloqueo único es el cuello de botella y el equipo pueda mantener la prueba formal y las pruebas de estrés, una clase de la biblioteca estándar o una implementación clara basada en bloqueos es más segura.

Pregunta de seguimiento 6: ¿Por qué no usar dos semáforos directamente?

Un semáforo contador puede representar las ranuras libres y otro los elementos disponibles, pero las actualizaciones de head/tail en el anillo aún necesitan exclusión mutua. Adquirir múltiples primitivas de sincronización también requiere un manejo cuidadoso de excepciones, interrupciones y reversión de permisos (permits). Una solución con semáforos puede ser correcta, pero no es automáticamente más breve que un solo bloqueo más dos condiciones. Cualquier diseño debe demostrar que el recuento de permisos y el estado real del arreglo nunca divergen.