Patrones embebidos comunes: búfer circular

Esta es la primera parte de una nueva serie donde vuelvo a explorar los patrones embebidos más comunes, empezando por el búfer circular (ring buffer). Es uno de esos patrones que se usan en la comunicación serial para asegurarnos de no perder ni una trama de datos, y además es un bloque de construcción para otros patrones que veremos más adelante.

El código fuente completo y probado vive en el repositorio complementario: ileanmjr88/tetzontli. Cada patrón de esta serie es su propio módulo con una suite completa de pruebas en GoogleTest, así que puedes clonarlo y ejecutar las pruebas tú mismo.

Concepto básico

El búfer circular usa dos índices, head y tail. Para agregar datos hacemos put de un byte en head e incrementamos head en uno. Para quitar datos hacemos get de un byte en tail e incrementamos tail en uno. Ambos índices solo avanzan hacia adelante: tail persigue a head por todo el búfer.

Este patrón se puede implementar con una lista enlazada o con un arreglo (array) de longitud fija. La lista enlazada suena tentadora, pero en un sistema embebido no quieres asignación dinámica de memoria. En una computadora moderna con gigabytes de memoria, llamar a malloc no es problema. En un microcontrolador tienes kilobytes, y tres problemas más grandes: el heap se fragmenta en un dispositivo que funciona durante meses, el tiempo que tarda una asignación no es determinista, y muchas veces necesitas agregar datos desde dentro de una interrupción, donde no te puedes dar el lujo de ninguna de las dos cosas. Un arreglo fijo evita todo eso, porque el espacio queda garantizado en tiempo de compilación.

Ese espacio garantizado es justo lo que le da tiempo al consumidor: los bytes que llegan más rápido de lo que los puedes procesar se quedan seguros en el búfer hasta que llegues a ellos, en lugar de perderse.

Dar la vuelta dentro del arreglo

Un arreglo no da la vuelta. Cuando llegas al final e intentas pasarte, no regresa al inicio. Te sales del espacio asignado, lo que puede provocar un fallo de segmentación, devolverte un puntero NULL o entregarte basura.

La intuición

La forma intuitiva de regresar al inicio es el operador módulo %, que devuelve el residuo de una división.

Array of length/capacity: 10

Formula: index % capacity -> array index

- index 0: 0 % 10 = 0
- index 1: 1 % 10 = 1
- index 4: 4 % 10 = 4
- index 10: 10 % 10 = 0
- index 11: 11 % 10 = 1

El módulo nos deja tratar el arreglo fijo como un anillo, lo que lo hace ideal para un búfer circular. Funciona bien en una computadora moderna con varios núcleos, frecuencias altas y una unidad de división por hardware. Pero el módulo tiene una desventaja: requiere división, y muchos microcontroladores no tienen división por hardware. El compilador la emula por software, que es lento, y pagamos ese costo con cada byte que pasa por el búfer.

Optimizando con una máscara de bits

Hay una forma más rápida de hacer exactamente lo mismo que el operador módulo, con una restricción: la longitud del arreglo debe ser una potencia de dos. Esa restricción es la llave que nos permite recrear el módulo con una sola operación a nivel de bits. La fórmula es index = head & mask, donde mask = capacity - 1. Repasemos las matemáticas.

Array of length/capacity: 8 (power of two)

mask = 8 - 1 = 7 = 0b0111

- index 0: 0b0000 & 0b0111 = 0
- index 1: 0b0001 & 0b0111 = 1
- index 2: 0b0010 & 0b0111 = 2
- index 8: 0b1000 & 0b0111 = 0
- index 10: 0b1010 & 0b0111 = 2
- index 14: 0b1110 & 0b0111 = 6

Fíjate que son las mismas respuestas que daría el módulo: 8 % 8 = 0, 10 % 8 = 2, 14 % 8 = 6. No es coincidencia. La máscara 0b0111 conserva solo los tres bits más bajos y limpia todo lo que está arriba, y los tres bits más bajos de cualquier número son exactamente ese número módulo 8. Quedarte con los bits bajos es sacar el residuo. Como la capacidad es potencia de dos, capacity - 1 es una secuencia limpia de unos, y eso es lo que hace que el AND coincida perfectamente con el módulo. Intenta esto con un número que no sea potencia de dos y se cae todo: 11 & 9 = 9, pero 11 % 10 = 1.

Implementación

La estructura del búfer circular

Creamos un struct para mantener juntos todos los datos. Esto nos facilita la vida, porque con un solo puntero a la estructura podemos acceder a buffer, capacity, mask, head y tail. También hace que los datos sean fáciles de pasar por puntero a cualquier función que necesite agregar o leer de él.

// modules/ring_buffer/ring_buffer.h
typedef struct {
  uint8_t *buffer;
  size_t capacity;
  size_t mask;
  volatile size_t head;
  volatile size_t tail;
} ring_buffer_t;

Como puedes ver, el struct usa varios tipos distintos, y cada uno es deliberado.

• uint8_t: el búfer guarda bytes crudos. Sea cual sea la fuente, datos seriales de UART, I2C o cualquier otra cosa, llegan como octetos, así que un arreglo de bytes es el almacenamiento natural para la cola.
• size_t: capacity, mask, head y tail son todos size_t. Es un tipo sin signo del tamaño de la plataforma (normalmente un alias de unsigned int o unsigned long), lo que mantiene la aritmética de índices portable entre compiladores y evita problemas de desbordamiento con signo. Que sea sin signo importa más adelante, cuando dependemos de que head - tail dé la vuelta limpiamente.
• volatile: head y tail están marcados como volatile para que el compilador siempre los lea de memoria en lugar de guardarse una copia vieja en un registro. El productor y el consumidor pueden ejecutarse en contextos distintos (por ejemplo, una ISR llenando el búfer mientras el bucle principal lo vacía), así que un valor guardado en un registro podría perderse la actualización del otro lado. Ojo: volatile solo garantiza que la lectura ocurra, no hace que el acceso sea atómico ni seguro entre hilos. Veremos por qué este diseño es seguro de todos modos en la sección sobre interrupciones.

Las funciones de inicialización y de potencia de dos

Empecemos con init y su ayudante is_power_of_two. La función init prepara la estructura de datos, y la verificación de potencia de dos es el primer filtro: si quien la llama pide una capacidad que no es potencia de dos, la rechazamos, porque todo el esquema de enmascarado depende de eso. Arranquemos con el ayudante.

// modules/ring_buffer/ring_buffer.c
static bool is_power_of_two(size_t x) {
  return x != 0u && (x & (x - 1u)) == 0u;
}

Esta implementación hace lo siguiente:

• x != 0u: verifica que x no sea 0.
• (x & (x - 1u)) == 0u: esta es la parte que verifica que haya un solo bit encendido. Una potencia de dos tiene exactamente un bit encendido. Restarle uno apaga ese bit y enciende todos los que están debajo, así que x y x - 1 no comparten ningún bit y el AND da cero. Toma x = 8: 8 - 1 = 7, y 0b1000 & 0b0111 = 0. Ahora toma un número que no sea potencia de dos, x = 12: 12 - 1 = 11, y 0b1100 & 0b1011 = 0b1000, que no es cero. Ese bit que sobra delata que x tenía más de un bit encendido.

Si ambas condiciones se cumplen, entonces x es potencia de dos, así que la función devuelve true si y solo si x es potencia de dos. Ahora pasemos a la función ring_buffer_init.

// modules/ring_buffer/ring_buffer.c
bool ring_buffer_init(ring_buffer_t *rb, uint8_t *storage, size_t capacity) {
  if (rb == NULL || storage == NULL || !is_power_of_two(capacity)) {
    return false;
  }

  rb->buffer = storage;
  rb->capacity = capacity;
  rb->mask = capacity - 1u;
  rb->head = 0u;
  rb->tail = 0u;
  return true;
}

init valida primero sus entradas: un puntero nulo al struct, un storage nulo, o una capacidad que no sea potencia de dos hacen que se salga y devuelva false. Una vez que pasan las verificaciones, arma el struct, precalcula mask = capacity - 1 para que put y get nunca tengan que recalcularlo, y pone ambos índices en cero.

Fíjate que init no asigna nada. Quien la llama nos entrega el arreglo storage, así que el búfer puede vivir en memoria estática o en la pila. Este es el principio de no usar malloc que mencionamos antes, hecho concreto: no hay ni un malloc a la vista, y el espacio quedó garantizado en el momento en que el llamador declaró el arreglo.

Las funciones put y get

Estas funciones ponen y sacan un solo byte a la vez. Las limitamos a un byte porque a veces un solo byte es todo lo que necesitamos, y porque las funciones write y read de la siguiente sección están construidas encima de ellas. Empecemos con ring_buffer_put:

// modules/ring_buffer/ring_buffer.c
bool ring_buffer_put(ring_buffer_t *rb, uint8_t byte) {
  if (rb == NULL || ring_buffer_is_full(rb)) {
    return false;
  }
  rb->buffer[rb->head & rb->mask] = byte;
  rb->head++;
  return true;
}

La función empieza validando que rb no sea NULL, y luego llama al ayudante ring_buffer_is_full. Si el búfer está lleno salimos temprano y devolvemos false, avisándole al llamador que no se guardó ningún byte. Veremos cómo funciona is_full en la siguiente sección. Si hay espacio, hacemos la escritura, y aquí es donde rinde frutos la sección anterior: rb->head & rb->mask calcula el índice del arreglo con el AND a nivel de bits, envolviendo el head de avance libre de regreso a una ranura válida. Escribimos el byte ahí, y luego incrementamos head en uno para que el siguiente put caiga en la ranura que sigue. Fíjate en el orden: primero escribimos la ranura, después avanzamos head. Ese orden importa para la seguridad con interrupciones, y llegaremos a eso más adelante. Ahora veamos ring_buffer_get:

// modules/ring_buffer/ring_buffer.c
bool ring_buffer_get(ring_buffer_t *rb, uint8_t *out) {
  if (rb == NULL || out == NULL || ring_buffer_is_empty(rb)) {
    return false;
  }
  *out = rb->buffer[rb->tail & rb->mask];
  rb->tail++;
  return true;
}

Esta función también empieza validando que rb no sea NULL, y luego verifica que out tampoco lo sea. Ese es el puntero donde se guarda el byte que leemos, que es la forma en que get le devuelve datos al llamador dejando libre su valor de retorno bool para reportar éxito o fracaso. Por último llama al ayudante ring_buffer_is_empty. Si el búfer está vacío salimos temprano y devolvemos false, avisándole al llamador que no se leyó ningún byte. Veremos cómo funciona is_empty en la siguiente sección. Si no está vacío, hacemos la lectura, y otra vez la sección anterior rinde frutos: rb->tail & rb->mask calcula el índice del arreglo con el AND a nivel de bits, envolviendo el tail de avance libre en una ranura válida. Leemos el byte e incrementamos tail en uno. Fíjate que get nunca borra nada. Lee el byte y mueve tail hacia adelante, y esa ranura cuenta como libre en el momento en que tail pasa por encima de ella. Los índices por sí solos deciden qué son datos y qué es espacio libre.

Las funciones is_empty, is_full y count

Estos tres ayudantes, is_empty, is_full y count, son en los que se apoyan put y get para decidir si pueden seguir adelante. Aquí es donde los índices de avance libre por fin se ganan su lugar. Empecemos con ring_buffer_is_empty:

// modules/ring_buffer/ring_buffer.c
bool ring_buffer_is_empty(const ring_buffer_t *rb) {
  return rb->head == rb->tail;
}

El búfer está vacío cuando head y tail tienen el mismo valor. Tiene sentido: tail persigue a head, y cuando lo alcanza por completo ya no queda nada por leer. Aquí está la parte sutil. En muchos diseños de búfer circular los índices se envuelven de regreso al rango válido, y entonces head == tail es ambiguo, porque significa tanto completamente vacío como completamente lleno, y no puedes distinguir cuál. Como nuestros índices avanzan libres y nunca se envuelven, head == tail solo puede significar vacío. En un momento veremos cómo se maneja el caso lleno sin esa ambigüedad.

Sigue ring_buffer_count, que es el corazón de todo esto:

// modules/ring_buffer/ring_buffer.c
size_t ring_buffer_count(const ring_buffer_t *rb) {
  return rb->head - rb->tail;
}

Ambos índices solo aumentan. Cada put incrementa head, cada get incrementa tail, y ninguno se envuelve. Así que head siempre va adelante de tail por exactamente la cantidad de bytes que se han escrito pero todavía no se han leído, que es el nivel de llenado. La diferencia head - tail es la cuenta.

Lo bonito es lo que pasa cuando head finalmente se desborda. size_t tiene un máximo, y después de suficientes bytes head lo rebasa y da la vuelta hacia cero. Podrías esperar que head - tail se rompa en ese punto, pero no pasa. La resta sin signo da la vuelta igual que la suma sin signo, así que la diferencia sigue siendo correcta incluso a través del desbordamiento. Justo por esto head y tail son sin signo, y por esto el búfer puede funcionar indefinidamente sin ningún manejo especial para el desbordamiento.

Por último, ring_buffer_is_full:

// modules/ring_buffer/ring_buffer.c
bool ring_buffer_is_full(const ring_buffer_t *rb) {
  return ring_buffer_count(rb) == rb->capacity;
}

Una vez que tenemos count, lo de lleno es trivial: el búfer está lleno cuando contiene capacity bytes. Sin desperdiciar una ranura, sin un contador aparte que mantener sincronizado, sin ambigüedad con el caso vacío. Los índices de avance libre nos dan vacío (head == tail), lleno (count == capacity) y todo lo que hay en medio, todo a partir de dos números que solo cuentan hacia arriba.

Las funciones write y read

Estas funciones escriben y leen más de un byte a la vez. Empecemos con ring_buffer_write:

// modules/ring_buffer/ring_buffer.c
size_t ring_buffer_write(ring_buffer_t *rb, const uint8_t *data, size_t len) {
  if (rb == NULL || data == NULL) {
    return 0u;
  }
  size_t written = 0u;
  while (written < len && ring_buffer_put(rb, data[written])) {
    written++;
  }
  return written;
}

Esta función devuelve la cantidad de bytes que realmente se escribieron, que puede ser menor a la que se pidió. Devuelve 0 en la salida temprana, cuando el búfer circular rb o el puntero de origen data son NULL. Si no, inicializamos written para llevar la cuenta de cuántos bytes hemos copiado de data al búfer. El ciclo se ejecuta con dos condiciones unidas por &&: written < len nos impide leer más allá del final de la entrada, y ring_buffer_put devuelve false en cuanto el búfer se llena. Mientras ambas se cumplan, seguimos copiando bytes e incrementando written. El ciclo se detiene en cuanto una de las dos falla, ya sea porque se nos acabó la entrada o porque se nos acabó el espacio. La cuenta que devolvemos le dice al llamador exactamente cuánto entró: todo, una parte, o cero si el búfer ya estaba lleno.

Sigue ring_buffer_read:

// modules/ring_buffer/ring_buffer.c
size_t ring_buffer_read(ring_buffer_t *rb, uint8_t *out, size_t len) {
  if (rb == NULL || out == NULL) {
    return 0u;
  }
  size_t count = 0u;
  while (count < len && ring_buffer_get(rb, &out[count])) {
    count++;
  }
  return count;
}

Esta función también devuelve la cantidad de bytes que realmente se leyeron, que puede ser menor a la que se pidió. Devuelve 0 en la salida temprana, cuando el búfer circular rb o el puntero de destino out son NULL. Si no, inicializamos count para llevar la cuenta de cuántos bytes hemos leído del búfer hacia out. El ciclo se ejecuta con dos condiciones unidas por &&: count < len nos impide escribir más allá del final de la salida, y ring_buffer_get devuelve false en cuanto el búfer se vacía. Mientras ambas se cumplan, seguimos leyendo bytes e incrementando count. El ciclo se detiene en cuanto una de las dos falla, ya sea porque el búfer del llamador se llenó o porque el búfer circular se quedó seco. La cuenta que devolvemos le dice al llamador exactamente cuántos bytes recibió.

Seguridad con la rutina de servicio de interrupción (ISR)

Para quienes van empezando en embebidos, puede que no sea obvio por qué se considera seguro este diseño, ni cuáles son sus aplicaciones típicas. El uso principal de un búfer circular, como mencionamos antes, es la comunicación serial. Toma el ejemplo de dos microcontroladores hablando entre sí, cada uno ejecutando su propio firmware hecho a la medida de su tarea. No sabemos cuándo va a necesitar enviar datos ninguno de los dos, así que para asegurarnos de no perder un solo byte, montamos la ruta de recepción de datos (RX) usando la rutina de servicio de interrupción (Interrupt Service Routine, ISR) del microcontrolador. En cuanto llega el inicio de una transmisión, la ISR toma el control y captura cada byte.

Este es el lado productor del búfer. La ISR llama a put por cada byte que recibe. En otro lado, el bucle principal es el consumidor, llamando a get cuando tiene chance de procesar lo que llegó. Los dos se ejecutan en contextos distintos, y aquí está el detalle: la ISR puede dispararse en cualquier momento, entre cualesquiera dos instrucciones que esté ejecutando el bucle principal. Así que la pregunta justa es: si la interrupción cae a la mitad de un get, ¿pueden pisarse entre sí y corromper el búfer?

La respuesta es no, y la razón es que cada índice tiene exactamente un escritor. put es la única función que escribe head, y solo se ejecuta en la ISR. get es la única función que escribe tail, y solo se ejecuta en el bucle principal. Ninguno de los dos lados escribe el índice del otro. El productor lee tail para saber si el búfer está lleno, y el consumidor lee head para saber si está vacío, pero leer es seguro. Como ningún índice tiene dos escritores compitiendo por cambiarlo, no hay nada que corromper, y no necesitamos deshabilitar interrupciones ni recurrir a un candado. A esto se refiere la gente cuando habla de un solo productor y un solo consumidor, o SPSC por sus siglas en inglés.

Esa regla de un solo escritor también es el límite. En el momento en que tengas dos cosas llamando a put, o dos cosas llamando a get, la garantía se acaba y necesitas sincronización de verdad. Este búfer es seguro para exactamente un productor y un consumidor, ni uno más.

Hay un detalle más que lo hace funcionar, y es el orden que señalamos allá en put:

rb->buffer[rb->head & rb->mask] = byte; // 1. write the data
rb->head++; // 2. then publish it

Primero escribimos el byte en la ranura, y después incrementamos head. Ese orden no es accidente. head es la señal que le dice al consumidor que hay un byte nuevo disponible. Si incrementáramos head primero y el tiempo de la interrupción se acomodara mal, el consumidor podría ver el nuevo head, ir a leer esa ranura y sacar un byte que en realidad todavía no habíamos escrito. Al escribir la ranura antes de avanzar head, garantizamos que en el instante en que el consumidor puede ver el byte, el byte de verdad ya está ahí. La actualización del índice publica los datos, así que va al final. get hace el espejo: primero leemos el byte y después avanzamos tail, así nunca marcamos una ranura como libre antes de habernos llevado su contenido.

Un par de advertencias honestas para quien las quiera. Primero, volatile nos da visibilidad, no atomicidad. Este diseño asume que leer o escribir head y tail sucede en una sola operación de máquina ininterrumpible, lo cual es cierto cuando son del tamaño de palabra en el objetivo. En un MCU de 8 bits, donde size_t abarca varios bytes, la actualización de un índice podría partirse en varias instrucciones y quedar rota por una interrupción, y tendrías que tomarlo en cuenta. Segundo, en un microcontrolador de un solo núcleo con una ISR, volatile más el orden de escribir y luego publicar es suficiente. En un sistema multinúcleo con ordenamiento de memoria débil recurrirías a operaciones atómicas con semántica explícita de adquisición y liberación, para garantizar que la escritura de la ranura se vea antes que la del índice. Para el caso clásico de ISR y bucle principal, que es para el que está hecho este búfer, estamos en terreno firme.

Cierre

Eso es un búfer circular completo y funcional en un par de cientos de líneas de C. No hace asignación dinámica, da la vuelta con un solo AND a nivel de bits en lugar de una división, distingue lleno de vacío sin desperdiciar una ranura ni llevar un contador aparte, y es seguro llenarlo desde una interrupción mientras lo vaciamos desde el bucle principal. Para una estructura tan pequeña, trae metida una cantidad sorprendente de sutilezas.

Vale la pena ser honesto sobre lo que este diseño no hace, porque cada una de estas cosas es un intercambio deliberado, no un descuido:

• La capacidad debe ser potencia de dos. Este es el precio de cambiar el módulo por una máscara. Si necesitas guardar exactamente 100 bytes, redondeas a 128 y desperdicias un poco de espacio. En un sistema embebido casi siempre es un intercambio que conviene.
• Solo un productor y un consumidor. Un escritor por índice es lo que lo hace libre de candados. Dos ISR escribiendo, o dos contextos leyendo, y necesitas sincronización de verdad.
• No hay modo de sobrescribir lo más viejo. Cuando el búfer está lleno, put rechaza el byte nuevo en lugar de descartar el más viejo. Algunas aplicaciones quieren lo contrario, conservar las muestras más recientes y tirar las viejas. Esa es una variante razonable, solo que no es esta.
• Solo bytes. Este búfer guarda uint8_t. Guardar elementos más anchos, o structs arbitrarios, implica generalizar el tamaño del elemento, lo que cambia la aritmética de índices y el almacenamiento.

Ninguna de estas limitaciones es difícil de levantar. Son las preguntas que salen naturalmente, y varias de ellas se convierten en sus propios patrones más adelante en esta serie.

Y hablando de eso, el búfer circular es un cimiento, no un destino. Aparece por debajo de varios de los patrones que faltan: una cola de eventos es un búfer circular de mensajes en lugar de bytes, un filtro de promedio móvil es un búfer circular de muestras, y el despachador de comandos que construiremos lee su entrada directamente de uno. Lo que sigue en la serie es el pool de memoria. El búfer circular nos dio almacenamiento fijo para bytes; el pool de memoria nos da almacenamiento fijo para objetos, y es lo que permite que una cola de eventos guarde structs en lugar de datos crudos. Nos vemos ahí.