La Rueda de Tiempo (Timing Wheel) es un algoritmo fundamental para la gestión eficiente de tareas diferidas y programadas. En sistemas que requieren un alto volumen de programación de tareas, este modelo permite optimizar el uso de los hilos mediante la ejecución por lotes. A diferencia de las implementaciones nativas de Java como Timer o la combinación de DelayQueue con ScheduledThreadPoolExecutor, la rueda de tiempo ofrece un rendimiento superior en escenarios de alta concurrencia.
Sin embargo, es importante notar que la precisión de este algoritmo está limitada por la resolución mínima de sus "ticks". Si la rueda avanza cada segundo, no podrá programar tareas con una precisión inferior a ese intervalo.
Fundamentos y Arquitectura
Este patrón de diseño es ampliamente adoptado en sistemas de producción como Apache Kafka, Netty, Akka y Quartz. La programación de tareas se puede clasificar en dos enfoques: tiempo relativo (ejecutar después de X milisegundos) y tiempo absoluto (ejecutar en un timestamp específico). Ambos son intercambiables.
En su forma más básica, la rueda de tiempo es una cola circular implementada mediante un array. Cada posición del array (o ranura) contiene una lista de tareas (TaskBucket), que internamente es una lista doblemente enlazada de nodos (TaskNode).
Componentes del Modelo
- tickIntervalMs: El intervalo de tiempo base que representa cada ranura.
- bucketCount: El número total de ranuras en la rueda.
- wheelSpanMs: La capacidad total de tiempo de la rueda, calculada como
tickIntervalMs * bucketCount. - pointerTimeMs: El puntero que indica el tiempo actual. Siempre es un múltiplo entero de
tickIntervalMsy divide la rueda en tareas epxiradas y pendientes.
Mecánica de Funcionamiento
Imaginemos una rueda con tickIntervalMs = 1 y bucketCount = 20. Su capacidad total es de 20ms. Si insertamos una tarea para dentro de 2ms, se aloja en la ranura 2. Cuando el puntero avanza a la ranura 2, las tareas se ejecutan. Si el puntero da la vuelta completa y llega de nuevo a la ranura 2, esta se reutiliza para nuevas tareas.
El Problema de los Retardos Prolongados y la Solución Jerárquica
¿Qué ocurre si necesitamos programar una tarea para dentro de 350ms en una rueda con una capacidad de solo 20ms? Aumentar el bucketCount a 350 o más sería un desperdicio masivo de memoria y degradaría el rendimiento.
La solución es implementar una Rueda de Tiempo Jerárquica, similar a un sistema de numeración posicional. Cuando el tiempo de expiración de una tarea excede la capacidad de la rueda actual, esta se "promociona" a una rueda de nivel superior. Cada nivel superior tiene un tickIntervalMs igual al wheelSpanMs del nivel inferior.
Promoción y Degradación de Tareas
- Si una tarea de 350ms no cabe en la primera rueda (20ms), se envía a la segunda rueda (capacidad 400ms).
- Una tarea de 450ms requeriría una tercera rueda (capacidad 8000ms).
- Cuando el puntero de la rueda de nivel superior alcanza la ranura de la tarea, esta aún no ha expirado por completo. En este punto, la tarea se "degrada" y se reinserta en la rueda de nivel inferior, repitiendo el proceso hasta que finalmente llega a la ranura exacta de la rueda base para su ejecución.
Optimización del Avance del Puntero
Avanzar el puntero de la rueda en cada milisegundo mediante un bucle infinito generaría un consumo innecesario de CPU. Para evitar los "ticks vacíos", se utiliza una DelayQueue de Java como mecanismo de alerta.
Cada TaskBucket que recibe una tarea se inserta en la DelayQueue. El hilo principal se bloquea esperando a que el bucket con la expiración más temprana esté listo. Esto permite un avance del puntero "bajo demanda" y preciso, eliminando las iteraciones innecesarias.
Implementación en Java
A continuación, se presenta una implementación reestructurada de este algoritmo, modificando las estructuras de datos y la concurrencia para demostrra el concepto.
Nodos y Tareas Ejecutables
public class RunnableTask implements Runnable {
private final String identifier;
private final long delayMs;
private TaskNode assignedNode;
public RunnableTask(String identifier, long delayMs) {
this.identifier = identifier;
this.delayMs = delayMs;
}
public long getDelayMs() { return delayMs; }
public String getIdentifier() { return identifier; }
public synchronized void assignNode(TaskNode node) {
if (this.assignedNode != null && this.assignedNode != node) {
this.assignedNode.detach();
}
this.assignedNode = node;
}
@Override
public void run() {
System.out.println("Ejecutando tarea: " + identifier);
}
}
public class TaskNode {
private final RunnableTask task;
private final long expirationTimeMs;
TaskNode previous;
TaskNode next;
volatile TaskBucket parentBucket;
public TaskNode(RunnableTask task, long expirationTimeMs) {
this.task = task;
this.expirationTimeMs = expirationTimeMs;
}
public long getExpirationTimeMs() { return expirationTimeMs; }
public RunnableTask getTask() { return task; }
public void detach() {
TaskBucket currentBucket = this.parentBucket;
while (currentBucket != null) {
currentBucket.unlinkNode(this);
currentBucket = this.parentBucket;
}
}
}
Contenedores de Tareas (Buckets)
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;
public class TaskBucket implements Delayed {
private final TaskNode sentinel = new TaskNode(null, -1);
private final AtomicLong bucketExpiration = new AtomicLong(-1L);
public TaskBucket() {
sentinel.next = sentinel;
sentinel.previous = sentinel;
}
public long getExpiration() {
return bucketExpiration.get();
}
public boolean setExpiration(long expirationMs) {
return bucketExpiration.getAndSet(expirationMs) != expirationMs;
}
public boolean insertNode(TaskNode node) {
boolean success = false;
while (!success) {
node.detach();
synchronized (this) {
if (node.parentBucket == null) {
node.parentBucket = this;
TaskNode tail = sentinel.previous;
node.previous = tail;
node.next = sentinel;
tail.next = node;
sentinel.previous = node;
success = true;
}
}
}
return true;
}
public void unlinkNode(TaskNode node) {
synchronized (this) {
if (node.parentBucket == this) {
node.previous.next = node.next;
node.next.previous = node.previous;
node.next = null;
node.previous = null;
node.parentBucket = null;
}
}
}
public void flushAndClear(Consumer<TaskNode> action) {
synchronized (this) {
TaskNode current = sentinel.next;
while (current != sentinel) {
unlinkNode(current);
action.accept(current);
current = sentinel.next;
}
bucketExpiration.set(-1L);
}
}
@Override
public long getDelay(TimeUnit unit) {
long remaining = bucketExpiration.get() - System.currentTimeMillis();
return unit.convert(Math.max(0, remaining), TimeUnit.MILLISECONDS);
}
@Override
public int compareTo(Delayed other) {
if (other instanceof TaskBucket) {
return Long.compare(this.bucketExpiration.get(), ((TaskBucket) other).bucketExpiration.get());
}
return 0;
}
}
Rueda de Tiempo Jerárquica
import java.util.concurrent.DelayQueue;
public class TimingWheel {
private final long tickIntervalMs;
private final int bucketCount;
private final long wheelSpanMs;
private long pointerTimeMs;
private final TaskBucket[] slots;
private volatile TimingWheel overflowWheel;
private final DelayQueue<TaskBucket> sharedDelayQueue;
public TimingWheel(long tickIntervalMs, int bucketCount, long startTimeMs, DelayQueue<TaskBucket> delayQueue) {
this.tickIntervalMs = tickIntervalMs;
this.bucketCount = bucketCount;
this.wheelSpanMs = tickIntervalMs * bucketCount;
this.pointerTimeMs = startTimeMs - (startTimeMs % tickIntervalMs);
this.slots = new TaskBucket[bucketCount];
this.sharedDelayQueue = delayQueue;
for (int i = 0; i < bucketCount; i++) {
slots[i] = new TaskBucket();
}
}
public boolean scheduleTask(TaskNode node) {
long expiration = node.getExpirationTimeMs();
if (expiration < pointerTimeMs + tickIntervalMs) {
return false;
} else if (expiration < pointerTimeMs + wheelSpanMs) {
long virtualTick = expiration / tickIntervalMs;
int slotIndex = (int) (virtualTick % bucketCount);
TaskBucket targetBucket = slots[slotIndex];
targetBucket.insertNode(node);
if (targetBucket.setExpiration(virtualTick * tickIntervalMs)) {
sharedDelayQueue.offer(targetBucket);
}
return true;
} else {
return getOverflowWheel().scheduleTask(node);
}
}
private TimingWheel getOverflowWheel() {
if (overflowWheel == null) {
synchronized (this) {
if (overflowWheel == null) {
overflowWheel = new TimingWheel(wheelSpanMs, bucketCount, pointerTimeMs, sharedDelayQueue);
}
}
}
return overflowWheel;
}
public void advancePointer(long timestamp) {
if (timestamp > pointerTimeMs + tickIntervalMs) {
pointerTimeMs = timestamp - (timestamp % tickIntervalMs);
if (overflowWheel != null) {
overflowWheel.advancePointer(timestamp);
}
}
}
}
Motor de Programación (Scheduler)
import java.util.concurrent.*;
public class SchedulerEngine {
private final TimingWheel baseWheel;
private final DelayQueue<TaskBucket> delayQueue = new DelayQueue<>();
private final ExecutorService taskExecutor;
private final ScheduledExecutorService clockPusher;
public SchedulerEngine() {
this.baseWheel = new TimingWheel(1, 20, System.currentTimeMillis(), delayQueue);
this.taskExecutor = Executors.newFixedThreadPool(50);
this.clockPusher = Executors.newSingleThreadScheduledExecutor();
this.clockPusher.scheduleAtFixedRate(() -> advanceClock(20), 0, 20, TimeUnit.MILLISECONDS);
}
public void submitTask(RunnableTask task) {
long expiration = System.currentTimeMillis() + task.getDelayMs();
TaskNode node = new TaskNode(task, expiration);
task.assignNode(node);
processNode(node);
}
private void processNode(TaskNode node) {
if (!baseWheel.scheduleTask(node)) {
taskExecutor.submit(node.getTask());
}
}
private void advanceClock(long timeoutMs) {
try {
TaskBucket expiredBucket = delayQueue.poll(timeoutMs, TimeUnit.MILLISECONDS);
if (expiredBucket != null) {
baseWheel.advancePointer(expiredBucket.getExpiration());
expiredBucket.flushAndClear(this::processNode);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
public void shutdown() {
clockPusher.shutdown();
taskExecutor.shutdown();
}
}