Contrapresión gratis

En la documentación de cualquier librería reactiva clásica hay una sección larga dedicada a decidir qué hacer cuando el productor va más rápido que el consumidor, con una enumeración de estrategias que el programador debe elegir: acumular, descartar lo viejo, descartar lo nuevo, fallar o no hacer nada y confiar. Esa sección existe porque en aquellos diseños la emisión es una llamada ordinaria que no puede detenerse, y un valor que ya salió del productor tiene que ir a alguna parte. En Kotlin esa sección no existe, y no por omisión sino porque la pregunta no se puede formular: el productor no puede ir más rápido que el consumidor, ya que la llamada con la que emite no retorna hasta que el consumidor ha acabado. Lo interesante de este nivel no es aprender a resolver el problema, sino entender qué se está pagando por no tenerlo y cómo comprar deliberadamente el problema de vuelta cuando interesa.

El problema que no llegas a tener

Retomemos la firma del método consumidor y leámosla ahora buscando una sola palabra, que es la que hace todo el trabajo.

public suspend fun emit(value: T)

Como está marcada como suspendida, la llamada del productor no retorna hasta que el cuerpo del recolector ha terminado con ese valor. Si el recolector tarda un segundo, el productor lleva un segundo detenido en esa línea, sin haber consumido hilo alguno y sin haber acumulado nada en ninguna parte. La demanda no hay que pedirla porque siempre vale exactamente uno, y el crédito no hay que gestionarlo porque el propio mecanismo de suspensión lo lleva.

val t = measureTimeMillis {
    flow { repeat(3) { delay(100); emit(it) } }   // 100 ms por valor
        .collect { delay(200) }                    // 200 ms por valor
}
// t ronda los 900 ms: es la suma, no el maximo

Ese número es el precio y conviene mirarlo sin dramatismo. Novecientos milisegundos es la suma de todo lo que hace el productor más todo lo que hace el consumidor, porque en ningún instante los dos están trabajando a la vez. A cambio se obtienen dos garantías fuertes: no se pierde ningún valor y no crece ninguna estructura de memoria por mucho que se desequilibren los ritmos. Un programa así puede correr durante semanas sin que nadie vigile el tamaño de nada.

Comparado con el nivel anterior, esto es exactamente el régimen de cita de los canales, con la diferencia de que aquí no hubo que elegirlo: es lo que sale de no haber puesto ningún canal en medio. La contrapresión no es una funcionalidad de los flujos, es lo que queda cuando no hay búfer y el emisor puede detenerse.

Conviene además notar que la suspensión del productor no cuesta un hilo. Mientras el bloque espera dentro de emit, su continuación está guardada y el hilo que la ejecutaba se ha ido a hacer otra cosa. Esa es la diferencia entre esta contrapresión y la que se obtiene en un modelo de hilos bloqueantes, donde detener al productor significa dejar un hilo entero parado y por tanto solo se puede hacer unas pocas miles de veces antes de que la máquina se rinda. Aquí se puede hacer un millón de veces, y esa escala es la razón de que la solución sencilla sea también la solución viable.

buffer parte la cadena en dos corrutinas

El operador que compra el solapamiento no hace nada exótico: lanza el tramo de arriba en una corrutina propia y conecta ambos tramos con un canal de los que ya conoces. A partir de esa línea de la cadena hay dos corrutinas, no una.

val t = measureTimeMillis {
    flow { repeat(3) { delay(100); emit(it) } }
        .buffer()                 // capacidad por defecto, politica de suspension
        .collect { delay(200) }
}
// t ronda los 700 ms: productor y consumidor se solapan

El ahorro es modesto en tres elementos y crece con la longitud del flujo hasta acercarse al máximo de las dos sumas en lugar de a su total. Lo que se ha gastado a cambio es precisamente lo que la lección segunda describía como el valor real del diseño: ahora hay una frontera entre las dos mitades, con un canal por el que viajan los valores, y con ella se van la pila única, la latencia mínima y la garantía de que un valor emitido ya fue consumido.

flowchart LR
A[Bloque productor] --> B[emit]
B --> C[Canal con capacidad]
C --> D[Corrutina recolectora]
D --> E[Cuerpo del usuario]
C -.-> F[Si se llena el productor suspende]

Los parámetros del operador son los mismos que los de la factoría de canales, y no por casualidad: son literalmente los del canal que construye. La capacidad decide cuántos valores caben por delante y la política de desbordamiento decide qué ocurre cuando ese hueco se agota.

flujo.buffer()                                     // capacidad razonable, suspende
flujo.buffer(1)                                    // un solo hueco por delante
flujo.buffer(Channel.UNLIMITED)                    // sin freno, sin garantia
flujo.buffer(64, BufferOverflow.DROP_OLDEST)       // prioriza frescura

La posición del operador dentro de la cadena decide qué se solapa con qué, y por eso no es una decoración que se pueda colocar al final por costumbre. Todo lo que queda por encima pasa a la corrutina nueva; todo lo que queda por debajo se queda con el recolector.

lento.map { caro(it) }.buffer().collect { rapido(it) }   // caro se solapa
lento.buffer().map { caro(it) }.collect { rapido(it) }   // caro no se solapa

En la primera línea el mapeo costoso corre en la corrutina de arriba y avanza mientras el consumidor trabaja; en la segunda corre en la del recolector y vuelve a sumarse a su tiempo. Las dos cadenas producen exactamente los mismos valores en el mismo orden y tardan cosas muy distintas, lo cual convierte la posición del operador en una decisión de diseño y no en un detalle de estilo.

Hay un detalle de implementación con consecuencias observables. Las aplicaciones adyacentes de los operadores que introducen un canal se fusionan en un único canal correctamente configurado, de modo que escribir dos búferes seguidos, o un cambio de contexto seguido de un búfer, no produce dos fronteras sino una. Esa fusión es la que hace que componer funciones que devuelven flujos no vaya acumulando canales invisibles cada vez que alguien añade un tramo.

Descartar valores o cancelar trabajos

Cuando el consumidor no puede seguir el ritmo y acumular no es aceptable, quedan dos respuestas distintas que se confunden constantemente porque su efecto superficial se parece.

La primera es descartar valores. conflate es exactamente un búfer que se queda con el último y tira el anterior, es decir, el régimen conflado de los canales con otro nombre. El productor no se detiene nunca y el consumidor procesa cada valor entero, pero muchos valores no llegan a verse.

flujo.conflate().collect { procesar(it) }
// Cada procesar termina; los valores intermedios simplemente no llegaron

La segunda es no descartar nada en la entrada y cancelar el trabajo en curso cuando llega algo más nuevo. collectLatest recibe todos los valores, pero si mientras procesa uno aparece el siguiente, cancela el cuerpo a medias y lo reinicia con el valor nuevo.

flujo.collectLatest { valor ->
    procesar(valor)   // puede quedar cancelado a medias
}

La diferencia importa muchísimo y se resume así: con conflate todo lo que empieza termina, pero no todo empieza; con collectLatest todo empieza, pero no todo termina. Si el cuerpo del consumidor tiene efectos parciales, escribe en disco o abre recursos, la primera opción es segura y la segunda exige que ese cuerpo sea cancelable y limpie tras de sí. Si el cuerpo es un cálculo puro y caro cuyo resultado obsoleto no sirve de nada, la segunda es claramente mejor porque no malgasta tiempo terminando algo que ya nadie quiere.

La familia del último operador incluye variantes intermedias que hacen lo mismo dentro de la cadena en lugar de al final: mapLatest cancela la transformación anterior, transformLatest cancela el bloque anterior y flatMapLatest cancela la recolección del flujo anterior. Todas comparten la misma condición de uso y el mismo peligro.

Conviene mencionar dos operadores más que atacan el mismo desequilibrio desde el eje del tiempo y no del ritmo. debounce deja pasar un valor solo cuando ha transcurrido un intervalo sin que llegue otro, y es lo correcto para entradas que se estabilizan. sample deja pasar el último valor visto cada cierto periodo, y es lo correcto para observar algo que cambia continuamente. Ninguno de los dos es un búfer: son filtros con reloj, y por tanto pierden valores por diseño.

Hay un matiz sobre la variante que cancela que conviene fijar porque contradice la intuición. Esos operadores no introducen ninguna concurrencia real: el productor y el consumidor siguen compartiendo corrutina, y lo que ocurre es que el cuerpo del consumidor se ejecuta dentro de una corrutina hija que se cancela y se relanza. Por eso el productor sí puede quedarse esperando si el consumidor tarda en responder a la cancelación, y por eso un cuerpo que no comprueba la cancelación convierte a collectLatest en un collect corriente con más objetos por medio.