Contrapresión y buffering: cuando los ritmos no cuadran

En cuanto un flujo tiene dos extremos con relojes independientes, aparece la pregunta que ninguna abstracción puede esquivar: qué se hace con los elementos que llegan mientras el consumidor sigue ocupado con el anterior. Solo existen tres respuestas posibles —acumularlos, tirarlos o frenar a quien los produce— y cada una compra una propiedad pagando con otra. Acumular preserva todos los datos y sacrifica memoria y frescura; descartar preserva memoria y frescura y sacrifica datos; frenar lo preserva todo pero exige que la fuente acepte ser frenada, cosa que muchas no admiten. La política de búfer de una AsyncStream no es un parámetro de afinado: es la declaración explícita de cuál de esos tres precios estás dispuesto a pagar.

El desajuste y sus dos síntomas

Supongamos una fuente que emite cien elementos por segundo y un consumidor que procesa ochenta. La diferencia no es un detalle de rendimiento: es una acumulación de veinte elementos por segundo que crece sin techo mientras la situación dure. Con la política por defecto —ilimitada— el programa no falla ni avisa; simplemente empieza a consumir memoria de forma lineal en el tiempo.

El segundo síntoma es más traicionero que el primero. Una cola larga no solo ocupa memoria: retrasa. La relación es la que describe la ley de Little: la cantidad media de elementos en la cola es igual a la tasa de llegada multiplicada por el tiempo medio de espera. Despejando, el tiempo que un elemento pasa esperando es proporcional a la longitud de la cola, de modo que un búfer generoso convierte silenciosamente una pérdida de datos en un retraso creciente. La interfaz responde a un gesto que ocurrió hace tres segundos, el gráfico dibuja una posición que el sensor ya olvidó, el registro cuenta un pasado que no interesa a nadie. Es lo que en redes se llamó bufferbloat, y su moraleja se traslada íntegra: un búfer grande no arregla un desajuste sostenido, solo lo disfraza de latencia.

Hay además un tercer efecto que rara vez se mide y casi siempre se sufre: una cola larga retrasa también la observación del error. Si la fuente falla en el elemento mil, el consumidor no se enterará hasta haber procesado los novecientos que esperaban delante, y para entonces el diagnóstico llegará descolocado en el tiempo respecto a lo que de verdad ocurrió.

De ahí sale la primera regla operativa. Un búfer absorbe ráfagas, es decir, desajustes transitorios en los que la tasa media del consumidor supera a la del productor y solo puntualmente se invierte. Un búfer no absorbe desajustes sostenidos: si el productor es estructuralmente más rápido, ninguna capacidad basta y la única salida honesta es descartar o frenar.

Las políticas, con precisión

La enumeración tiene tres casos y sus nombres describen qué se conserva, no qué se tira. Es el matiz que más confusiones causa.

public enum BufferingPolicy {
    case unbounded                  // sin limite
    case bufferingOldest(Int)       // conserva los N mas antiguos
    case bufferingNewest(Int)       // conserva los N mas recientes
}

let (flujo, cont) = AsyncStream.makeStream(of: Muestra.self,
                                           bufferingPolicy: .bufferingNewest(1))

Con bufferingOldest(n), cuando el búfer está lleno el elemento entrante se descarta: la cola queda congelada con los primeros que llegaron. Es la política de quien quiere el principio de la historia —el primer error, la primera lectura anómala— y considera prescindible lo que venga después.

Con bufferingNewest(n), cuando el búfer está lleno se descarta el más antiguo para hacer sitio al recién llegado: la cola es una ventana deslizante sobre lo último ocurrido. Es la política de quien mira un estado que se refresca, donde una muestra vieja no tiene ningún valor.

El caso límite bufferingNewest(1) merece nombre propio: convierte el flujo en un registro de último valor conocido, donde el consumidor siempre ve lo más reciente y jamás se retrasa. Para posiciones, temperaturas, progresos o cualquier magnitud que represente un estado continuo, suele ser la elección correcta y casi nadie la usa.

En todos los casos el productor puede enterarse, porque yield devuelve el veredicto:

switch cont.yield(muestra) {
case .enqueued(let restantes) where restantes < 4:
    reducirCadencia()            // aviso temprano de saturacion
case .dropped:
    contadorDePerdidas += 1      // el flujo va por detras
case .terminated:
    detenerSensor()              // ya no escucha nadie
default:
    break
}

Ese switch es lo más parecido a una contrapresión que AsyncStream puede ofrecer: no frena a nadie, pero informa a tiempo para que el productor decida frenarse solo.

flowchart TB
A[Llega un elemento y el bufer esta lleno] --> B[Que politica se eligio]
B -->|unbounded| C[Crece la memoria y crece la latencia]
B -->|bufferingOldest N| D[Se descarta el elemento entrante]
B -->|bufferingNewest N| E[Se descarta el mas antiguo de la cola]
D --> F[Se conserva el principio de la serie]
E --> G[Se conserva el estado mas reciente]
C --> H[Riesgo de bufferbloat y de agotar memoria]
style C fill:#f38ba8,color:#11111b
style G fill:#a6e3a1,color:#11111b
style F fill:#89b4fa,color:#11111b

Dónde vive la contrapresión de verdad

Frenar al productor exige que el productor pueda ser frenado, y eso solo ocurre si la fuente es de tracción o acepta una señal de vuelta. Hay tres caminos.

El primero es no usar un adaptador de empuje. Si la fuente ya es de tracción —una lectura de fichero, una consulta paginada, una llamada de red por página— escribe una AsyncSequence propia o usa el constructor por desdoblamiento: el trabajo se hace cuando el consumidor pide, y no antes. Ahí la contrapresión es perfecta y gratuita porque no hay cola que llenar.

let paginas = AsyncStream(unfolding: {
    await api.siguientePagina()      // no se pide la siguiente hasta consumir esta
})

El segundo es AsyncChannel, de swift-async-algorithms. Su método de envío es asíncrono y suspende hasta que un consumidor recoge el elemento: capacidad cero, encuentro directo entre las dos partes, ninguna pérdida y ningún crecimiento de memoria. El productor va literalmente al ritmo del consumidor.

let canal = AsyncChannel<Trabajo>()
await canal.send(trabajo)     // suspende hasta que alguien lo recoge

El precio es simétrico al beneficio: solo sirve si el productor vive dentro de una tarea y puede permitirse esperar. Un callback de sistema no puede suspender, y por eso AsyncChannel no sustituye a AsyncStream sino que cubre el caso que esta no puede cubrir.

El tercero son los umbrales: dejar que el búfer crezca hasta una marca alta, pedir a la fuente que pause, y reanudarla al bajar de una marca baja. Es el mecanismo que usan las bibliotecas de red serias, y en Swift aparece formalizado en las utilidades de SwiftNIO. Se puede replicar a mano encima de una AsyncStream si la fuente ofrece métodos de pausa y reanudación.

El criterio de elección

La política no se deduce del rendimiento sino de qué significa el dato, y hay tres familias con respuestas distintas.

Si cada elemento es una muestra de un estado continuo —posición, batería, progreso, precio— lo viejo no vale nada en cuanto llega lo nuevo: bufferingNewest(1), o una ventana pequeña si necesitas suavizar.

Si cada elemento es un evento único con significado propio —una pulsación, una transacción, un mensaje— perderlo es un fallo funcional, no una degradación. Aquí no hay política de descarte aceptable: necesitas contrapresión real, un canal de encuentro o una cola persistente fuera del proceso. Elegir unbounded en este caso es aplazar la decisión, no resolverla.

Si el orden importa más que la completitud —registros y diagnósticos— suele ganar bufferingOldest, porque la primera anomalía explica el fallo y las mil siguientes son ruido derivado.

Conviene además separar dos decisiones que suelen tomarse juntas por inercia: cuánto se guarda y qué se tira al llenarse. La capacidad se dimensiona con datos —tamaño de la ráfaga esperada, tiempo de recuperación del consumidor, memoria disponible— y una capacidad de dos cifras suele bastar para absorber picos reales; la regla de descarte, en cambio, no se mide, se razona a partir del significado del elemento. Confundirlas lleva al vicio habitual de subir la capacidad cuando lo que falla es la regla.

Y hay una última pregunta, previa a todas: ¿de verdad el consumidor tiene que ser tan lento? Antes de dimensionar un búfer merece la pena comprobar si el cuerpo del bucle está haciendo trabajo que podría delegarse, agruparse o mover a otro sitio. Muchos problemas de contrapresión se disuelven cuando el consumidor deja de hacer dentro del bucle algo que no le correspondía.