Кольцевой буфер для UART на прерываниях

Сколько времени у обработчика и сколько памяти у буфера

Кадр 8N1 — старт-бит, 8 бит данных и стоп-бит — занимает 10 битовых интервалов. Отсюда скорость в байтах и время одного символа:

Посчитать кадр с чётностью или двумя стоп-битами помогает инструмент «Кадр UART», а погрешность скорости при данной тактовой частоте — «Скорость UART».

Ограничений два, и они разного масштаба.

Обработчик должен забрать байт, пока аппаратный буфер не переполнился. У STM32F103 регистр данных USART один: пока принимается следующий символ, предыдущий ждёт в DR, и если к концу приёма нового символа флаг RXNE всё ещё установлен, взводится ORE (RM0008, 27.6.1). Запас — одно время символа: 86,8 мкс на 115 200 бод, 10 мкс на 1 Мбит/с, то есть 720 тактов ядра на 72 МГц. У ATmega328P приёмный буфер — FIFO на два символа, и переполнение DOR0 наступает, когда буфер полон, третий символ принят в сдвиговый регистр и обнаружен старт-бит четвёртого (DS40002061B, 20.11.2). Запас — около двух символов. Это время делят все обработчики с тем же или более высоким приоритетом, а на AVR — все обработчики и все участки с запрещёнными прерываниями.

Программный буфер должен вместить всё, что придёт, пока основной цикл не читает. Если цикл может отвлечься на 15 мс, на 115 200 бод за это время придёт 0,015 · 11 520 ≈ 173 байта, и буфер берут на 256. Для коротких команд с ответом — меньше; для потока данных GPS или логов — больше. Если задержки цикла никто не измерял, это стоит сделать до выбора размера: например, счётчиком циклов DWT, как показано в статье о логировании и трассировке.

Устройство буфера

Буфер — массив из N ячеек и два индекса. Производитель пишет в ячейку head и увеличивает head, потребитель читает ячейку tail и увеличивает tail. Индексы — свободно растущие беззнаковые счётчики: они не заворачиваются в 0 на границе массива, а просто переполняются, как положено беззнаковой арифметике C (C11, 6.2.5p9). Номер ячейки получается маской индекс & (N − 1), поэтому N — степень двойки, а число байт в буфере равно head − tail по модулю 2^k, где k — разрядность индекса.

Отсюда ограничение на размер: N не больше половины диапазона индекса плюс один, то есть 128 при 8-битных индексах. При N = 256 и 8-битных индексах полный буфер дал бы разность 256 mod 256 = 0 — как у пустого.

Почему индекс на AVR 8-битный: чтение и запись uint8_t — одна команда, и обработчик не может вклиниться в середину. 16-битный индекс читается двумя командами, и при неудачном моменте основная программа получит половину старого и половину нового значения. На Cortex-M чтение и запись выровненного 16- и 32-битного значения неделимы (ARMv7-M ARM, A3.5.3), и индексы делают uint16_t или uint32_t, а буфер — сколько нужно.

/* rb.h: single-producer single-consumer byte ring buffer */
#ifndef RB_H
#define RB_H

#include <stdbool.h>
#include <stdint.h>

#ifndef RB_IDX_T
#define RB_IDX_T uint8_t                 /* uint16_t on 32-bit cores for bigger buffers */
#endif

#ifndef RB_BARRIER                        /* compiler barrier: no reordering across it */
#define RB_BARRIER() __asm__ __volatile__ ("" ::: "memory")
#endif

typedef RB_IDX_T rb_idx_t;

typedef struct {
    uint8_t *data;
    rb_idx_t mask;                        /* size - 1, size is a power of two */
    volatile rb_idx_t head;               /* written only by the producer */
    volatile rb_idx_t tail;               /* written only by the consumer */
} rb_t;

#define RB_IS_POW2(n) ((n) != 0u && (((n) & ((n) - 1u)) == 0u))

/* Static buffer of a given size; the size is checked at compile time. */
#define RB_DEFINE(name, size)                                                   \
    _Static_assert(RB_IS_POW2(size) && (size) <= (rb_idx_t)-1 / 2u + 1u,         \
                   #name ": size must be a power of two that fits the index");  \
    static uint8_t name##_data[size];                                           \
    static rb_t name = { name##_data, (rb_idx_t)((size) - 1u), 0u, 0u }

static inline rb_idx_t rb_count(const rb_t *rb)
{
    return (rb_idx_t)(rb->head - rb->tail);
}

/* Producer side. Returns false when the buffer is full. */
static inline bool rb_put(rb_t *rb, uint8_t byte)
{
    rb_idx_t head = rb->head;
    if ((rb_idx_t)(head - rb->tail) > rb->mask)
        return false;
    RB_BARRIER();                         /* see the freed slot before overwriting it */
    rb->data[head & rb->mask] = byte;
    RB_BARRIER();                         /* store the byte before publishing it */
    rb->head = (rb_idx_t)(head + 1u);
    return true;
}

/* Consumer side. Returns the byte or -1 when the buffer is empty. */
static inline int rb_get(rb_t *rb)
{
    rb_idx_t tail = rb->tail;
    if (rb->head == tail)
        return -1;
    RB_BARRIER();                         /* see the new head before reading the slot */
    uint8_t byte = rb->data[tail & rb->mask];
    RB_BARRIER();                         /* read the byte before freeing the slot */
    rb->tail = (rb_idx_t)(tail + 1u);
    return byte;
}

#endif /* RB_H */

Детали, которые легко упустить:

• Приведения (rb_idx_t)(...). Операнды uint8_t продвигаются до int (C11, 6.3.1.1), и разность head − tail может оказаться отрицательной. Приведение к беззнаковому типу берёт её по модулю 256 (6.3.1.3p2). Без него проверка «полон» сломается на переходе счётчика через 255.
• volatile только у индексов. Данные объявлены без volatile, а обращения к не-volatile объектам компилятор вправе переставлять относительно volatile (так прямо сказано в руководстве GCC) — поэтому порядок «данные, затем индекс» задают барьеры. Можно было объявить volatile и массив: обращения к volatile-объектам выполняются строго по правилам абстрактной машины и между собой не переставляются (C11, 5.1.2.3p6), и на одном ядре такой буфер тоже корректен. Барьер выбран ради эффективности: он ограничивает компилятор в двух точках, а volatile на данных запрещает оптимизировать каждое обращение к буферу, включая копирование блоками.
• Барьер компилятора, а не процессора. На одном ядре Cortex-M3 и M4 обращения к памяти завершаются в порядке программы, и команда DMB не нужна (AN321). Она нужна, если другая сторона — DMA или второе ядро.
• Макросы с #ifndef позволяют подставить другую ширину индекса для STM32 и другой барьер при сборке теста на компьютере.

Вместо барьеров и volatile можно взять атомарные операции C11: индексы объявляют атомарными — _Atomic rb_idx_t head, tail; вместо volatile, потому что функции <stdatomic.h> работают только с объектами атомарного типа, — и пишут atomic_store_explicit(&rb->head, head + 1, memory_order_release) у производителя и atomic_load_explicit(&rb->head, memory_order_acquire) у потребителя (C11, 7.17.7). Компилятор сам поставит то, что нужно для данного процессора. На однопроцессорном микроконтроллере это бывает избыточно, а на ядрах без подходящих команд GCC превращает такие операции в вызовы библиотечных функций, поэтому для AVR чаще оставляют вариант с барьером.

ATmega328P: приём и передача

Векторы USART в avr-libc: USART_RX_vect — принят символ, USART_UDRE_vect — регистр данных передатчика пуст, USART_TX_vect — передача кадра завершена. Последний нужен для полудуплексных линий вроде RS-485; для простого вывода хватает первых двух.

/* uart_avr.c: ATmega328P, USART0, F_CPU and BAUD given on the command line */
#include <avr/io.h>
#include <avr/interrupt.h>
#include <util/atomic.h>
#include <util/setbaud.h>
#include "rb.h"

RB_DEFINE(rx_rb, 64);
RB_DEFINE(tx_rb, 64);

static volatile uint16_t rx_overruns;     /* bytes lost in hardware (DOR0) */
static volatile uint16_t rx_frame_errors; /* bad stop bit (FE0) */
static volatile uint16_t rx_dropped;      /* software buffer full */

void uart_init(void)
{
    UBRR0H = UBRRH_VALUE;
    UBRR0L = UBRRL_VALUE;
    UCSR0A = USE_2X ? _BV(U2X0) : 0u;            /* error flags written as zero */
    UCSR0C = _BV(UCSZ01) | _BV(UCSZ00);          /* 8N1 */
    UCSR0B = _BV(RXCIE0) | _BV(RXEN0) | _BV(TXEN0);
}

ISR(USART_RX_vect)
{
    uint8_t status = UCSR0A;                     /* flags belong to the byte in UDR0 */
    uint8_t byte = UDR0;                         /* always read: clears RXC0 */

    if (status & _BV(DOR0))
        rx_overruns++;
    if (status & _BV(FE0)) {
        rx_frame_errors++;
        return;                                  /* the byte is garbage */
    }
    if (!rb_put(&rx_rb, byte))
        rx_dropped++;
}

ISR(USART_UDRE_vect)
{
    int c = rb_get(&tx_rb);
    if (c < 0)
        UCSR0B &= (uint8_t)~_BV(UDRIE0);         /* nothing left: stop UDRE interrupts */
    else
        UDR0 = (uint8_t)c;
}

int uart_getc(void)
{
    return rb_get(&rx_rb);                        /* -1 when nothing has arrived */
}

bool uart_putc(uint8_t c)
{
    if (!rb_put(&tx_rb, c))
        return false;                             /* full: caller decides to wait or drop */
    UCSR0B |= _BV(UDRIE0);                        /* start or keep the transmission going */
    return true;
}

uint16_t uart_overruns(void)
{
    uint16_t n;
    ATOMIC_BLOCK(ATOMIC_RESTORESTATE) {           /* 16-bit read is two instructions */
        n = rx_overruns;
    }
    return n;
}

Что здесь важно.

Порядок чтения в обработчике приёма. Флаги ошибок FE0, DOR0 и UPE0 хранятся в FIFO вместе с символом, к которому относятся, и чтение UDR0 сдвигает FIFO. Поэтому сначала UCSR0A, потом UDR0 (DS40002061B, 20.7.4). И UDR0 читают всегда, даже если байт испорчен: прерывание приёма повторяется, пока установлен RXC0, а сбрасывается он только опустошением FIFO (20.7.3).

DOR0 означает, что байт уже потерян где-то перед текущим. Сам текущий байт корректен, поэтому он идёт в буфер, а счётчик переполнений сообщает основной программе, что поток повреждён — для протокола с кадрами это повод сбросить разбор текущего кадра.

Передача запускается включением прерывания. Флаг UDRE0 установлен всегда, когда передатчик готов принять байт, — в том числе сразу после сброса (20.11.2). Поэтому достаточно положить байт в буфер и разрешить UDRIE0: обработчик вызовется немедленно, если передатчик свободен, или когда освободится. Опустошив буфер, обработчик запрещает UDRIE0, иначе он вызывался бы непрерывно.

Гонка на UCSR0B безопасна, но только здесь. UCSR0B |= _BV(UDRIE0) — это чтение, OR и запись регистра в три команды: UCSR0B лежит в пространстве данных по адресу 0xC1, а команды SBI и CBI работают только с регистрами ввода-вывода 0x00–0x1F (DS40002061B, разд. 36, примечание 2). Если между чтением и записью сработает USART_UDRE_vect и выключит UDRIE0, основная программа запишет его обратно включённым. Последствие безвредно: обработчик вызовется ещё раз, увидит пустой буфер и снова выключит прерывание. Но если какой-то другой обработчик меняет иные биты UCSR0B — например, переключает RXEN0 для полудуплекса, — его изменение может быть затёрто, и запись в UCSR0B из основной программы нужно взять в ATOMIC_BLOCK.

Счётчики ошибок 16-битные, и на 8-битном AVR их чтение в основной программе — две команды; между ними обработчик может изменить значение. Отсюда ATOMIC_BLOCK в uart_overruns().

Скорость задаётся макросами F_CPU и BAUD на этапе компиляции, а util/setbaud.h вычисляет делитель и решает, нужен ли режим U2X. Как то же сделать вручную и проверить погрешность — в уроке «UART и printf» курса «AVR с нуля».

STM32F103: один обработчик на все события

У USART в STM32 один вектор на все события, и обработчик сам выясняет, что произошло, по регистру состояния USART_SR. Для 256-байтных буферов индексы делают 16-битными: проект собирают с -DRB_IDX_T=uint16_t.

/* uart_stm32.c: STM32F103, USART1, PA9/PA10 already configured */
#include "stm32f1xx.h"
#include "rb.h"

RB_DEFINE(rx_rb, 256);
RB_DEFINE(tx_rb, 256);

static volatile uint32_t rx_overruns, rx_frame_errors, rx_dropped;

void uart_init(void)
{
    RCC->APB2ENR |= RCC_APB2ENR_USART1EN;
    USART1->BRR = 0x271u;                       /* 72 MHz / 115200 = 625 = 0x271 (USARTDIV 39.0625) */
    USART1->CR1 = USART_CR1_UE | USART_CR1_TE | USART_CR1_RE | USART_CR1_RXNEIE;
    NVIC_SetPriority(USART1_IRQn, 5);
    NVIC_EnableIRQ(USART1_IRQn);
}

void USART1_IRQHandler(void)
{
    uint32_t sr = USART1->SR;

    /* RXNE, or ORE alone: reading DR after SR clears ORE, NE and FE */
    if (sr & (USART_SR_RXNE | USART_SR_ORE)) {
        uint8_t byte = (uint8_t)USART1->DR;
        if (sr & USART_SR_ORE)
            rx_overruns++;
        if (sr & USART_SR_FE)
            rx_frame_errors++;
        else if (!rb_put(&rx_rb, byte))
            rx_dropped++;
    }

    if ((USART1->CR1 & USART_CR1_TXEIE) && (sr & USART_SR_TXE)) {
        int c = rb_get(&tx_rb);
        if (c < 0)
            USART1->CR1 &= ~USART_CR1_TXEIE;     /* buffer empty: stop TXE interrupts */
        else
            USART1->DR = (uint8_t)c;             /* the write clears TXE */
    }
}

int uart_getc(void)
{
    return rb_get(&rx_rb);
}

bool uart_putc(uint8_t c)
{
    if (!rb_put(&tx_rb, c))
        return false;
    USART1->CR1 |= USART_CR1_TXEIE;
    return true;
}

ORE — главная ловушка. Флаг переполнения взводится, только пока установлен RXNE, при RXNEIE = 1 тоже вызывает прерывание и сбрасывается последовательностью «чтение SR, затем чтение DR» (RM0008, 27.6.1). Обычно обработчик так и делает, и чтение DR сбрасывает оба флага. Но если следующий символ закончится между чтением SR и чтением DR, ORE взведётся уже после чтения SR: чтение DR сбросит RXNE, а ORE останется. При следующем входе обработчик увидит ORE = 1 и RXNE = 0. Если он проверяет один RXNE, то ничего не прочитает и выйдет, ORE так и не сбросится, и прерывание будет вызываться снова и снова, отнимая всё процессорное время. Поэтому условие чтения — RXNE или ORE. Содержимое DR при переполнении не теряется: пропадает символ, принятый в сдвиговый регистр, а в DR остаётся предыдущий корректный байт.

Проверка TXEIE перед TXE. Флаг TXE установлен всё время, пока передатчик свободен, а обработчик вызывается и по приёму. Без проверки разрешения обработчик при каждом принятом байте лез бы в пустой буфер передачи.

Делитель BRR. USARTDIV = f_PCLK / (16 · скорость) = 72 000 000 / (16 · 115 200) = 39,0625: мантисса 39 (0x27) в битах 15:4, дробная часть 0,0625 · 16 = 1 в битах 3:0, итого 0x271. Здесь деление точное, на других частотах появится погрешность — её покажет калькулятор скорости UART. Настройка выводов и тактирования шин — в инструменте «Инициализация STM32» и в статье о USART в STM32.

Завершение передачи для RS-485. TXE означает, что байт ушёл в сдвиговый регистр, а не на линию. Чтобы отключить передатчик RS-485 вовремя, после последнего байта включают TCIE и снимают сигнал DE в обработчике по флагу TC — он взводится, когда кадр передан полностью (RM0008, 27.6.1). Если обработчик управляет и TXEIE, и TCIE, гонка на CR1 из предыдущего раздела перестаёт быть безвредной, и uart_putc меняет CR1 при запрещённых прерываниях. Подробнее о полудуплексе — в статье об интерфейсе RS-485.

Пример с числами: перенос счётчиков

N = 64, маска 63, индексы 8-битные. Пусть основной цикл долго не читал:

Счётчик head перешёл через 255, но разность осталась правильной. Ячейки: последний записанный байт лежит в (8 − 1) & 63 = 7, следующий читаемый — в 210 & 63 = 18.

Сборка строк из буфера

Основной цикл обычно ждёт не байты, а команды, заканчивающиеся переводом строки. Неблокирующая сборка строки забирает из буфера всё, что есть, и сообщает о готовой строке:

#include <stdbool.h>
#include <stddef.h>

#define LINE_MAX 64u

typedef struct {
    char buf[LINE_MAX];
    size_t len;
    bool overflow;                       /* the current line was too long */
} line_t;

/* Returns true when a complete line is in l->buf (zero-terminated). */
bool line_poll(line_t *l, int (*getc_fn)(void))
{
    int c;
    while ((c = getc_fn()) >= 0) {
        if (c == '\r')
            continue;                    /* accept both CRLF and LF */
        if (c == '\n') {
            bool ok = !l->overflow;
            l->buf[l->len] = '\0';
            l->len = 0;
            l->overflow = false;
            if (ok)
                return true;
            continue;                    /* drop the over-long line */
        }
        if (l->len < LINE_MAX - 1u)
            l->buf[l->len++] = (char)c;
        else
            l->overflow = true;
    }
    return false;
}

Функция возвращается сразу, когда буфер пуст, и не держит основной цикл. Слишком длинная строка не обрезается молча, а отбрасывается целиком: обрезанная команда опаснее потерянной. Для двоичных протоколов с кадрами вместо строк нужен разборщик с длиной и контрольной суммой — он разобран в статье о реализации протоколов, а устройство такого разборщика как автомата — в статье о конечных автоматах.

Приём через DMA

На высоких скоростях прерывание на каждый байт становится дорогим: при 1 Мбит/с это 100 000 входов в обработчик в секунду. STM32 позволяет переложить приём на DMA: канал в циклическом режиме (бит CIRC) пишет байты из USART_DR в массив и после последнего начинает сначала, автоматически перезагружая счётчик (RM0008, 13.3.3). Регистр DMA_CNDTRx уменьшается после каждой передачи (13.4.4), поэтому индекс записи равен N − CNDTR — это и есть head кольцевого буфера, который обновляет аппаратура, а tail по-прежнему ведёт программа.

Остаётся узнать, когда читать. Удобный повод — флаг IDLE: он взводится, когда после приёма на линии наступила тишина длиной в кадр (RM0008, 27.6.1; см. свободная линия), то есть пакет закончился. Обработчик IDLE просто будит основной цикл, а тот забирает всё от tail до N − CNDTR. Переполнение в такой схеме программа не замечает: DMA перезапишет непрочитанное молча, поэтому размер буфера считают с запасом. Каналы и запросы DMA у STM32 описаны в статье о DMA в STM32.

Проверка на компьютере

Буфер не зависит от железа, и его логику удобно проверить на компьютере: барьер для обычного компилятора подменяется пустым макросом, производитель и потребитель вызываются по очереди в псевдослучайном порядке, а тест сравнивает порядок байт и поведение на границах.

/* test_rb.c: host test, build: cc -std=c11 -Wall '-DRB_BARRIER()=' test_rb.c */
#include <stdio.h>
#include "rb.h"

RB_DEFINE(t_rb, 16);

static unsigned failures;
#define CHECK(cond) do { if (!(cond)) { failures++; \
    printf("FAIL line %d: %s\n", __LINE__, #cond); } } while (0)

static uint32_t rnd_state = 12345u;
static uint32_t rnd(void)                     /* xorshift32, deterministic */
{
    rnd_state ^= rnd_state << 13;
    rnd_state ^= rnd_state >> 17;
    rnd_state ^= rnd_state << 5;
    return rnd_state;
}

int main(void)
{
    /* empty and full */
    CHECK(rb_get(&t_rb) == -1);
    for (int i = 0; i < 16; i++)
        CHECK(rb_put(&t_rb, (uint8_t)i));
    CHECK(!rb_put(&t_rb, 99));                /* 17th byte is rejected */
    CHECK(rb_count(&t_rb) == 16);
    for (int i = 0; i < 16; i++)
        CHECK(rb_get(&t_rb) == i);
    CHECK(rb_get(&t_rb) == -1);

    /* random interleaving across many counter wraps */
    uint8_t next_in = 0, next_out = 0;
    for (int step = 0; step < 100000; step++) {
        unsigned n = rnd() % 8u;
        if (rnd() & 1u) {
            while (n--) {
                if (rb_put(&t_rb, next_in))
                    next_in++;
                else
                    CHECK(rb_count(&t_rb) == 16);
            }
        } else {
            while (n--) {
                int c = rb_get(&t_rb);
                if (c < 0) {
                    CHECK(rb_count(&t_rb) == 0);
                    break;
                }
                CHECK(c == next_out);
                next_out++;
            }
        }
        CHECK(rb_count(&t_rb) == (uint8_t)(next_in - next_out));
    }

    printf("%s, %u failures\n", failures ? "FAILED" : "OK", failures);
    return failures != 0;
}

Ключ '-DRB_BARRIER()=' (в кавычках из-за скобок) определяет барьер как пустой макрос: в однопоточном тесте он не нужен, а синтаксис __asm__ есть не у всех компиляторов. За 100 000 шагов счётчики 8-битных индексов проходят через 255 тысячи раз, так что тест проверяет именно ту арифметику, которая ломается при забытом приведении типа. Этот тест не ловит гонки — для них нужен второй поток или реальное железо, — зато ловит ошибки границ, которые составляют большую часть ошибок в буферах. Как организовать такие тесты в проекте и подключить Unity, рассказано в статье о модульном тестировании прошивки.

Типичные ошибки

• Размер не степень двойки при маскировании индексов — часть ячеек не используется, а часть перезаписывается. _Static_assert в RB_DEFINE ловит это при компиляции.
• Размер 256 при 8-битных индексах — полный буфер неотличим от пустого.
• 16-битный индекс на AVR без защиты — редкие «скачки» заполнения и потерянные или повторённые байты.
• Нет чтения регистра данных при ошибке — бесконечное прерывание на AVR (RXC0 не сбрасывается) и на STM32 (ORE не сбрасывается).
• Данные без volatile и без барьера — компилятор вправе переставить запись байта и публикацию индекса; нужен барьер (или volatile и на массиве — ценой оптимизации).
• Ожидание места в буфере передачи при запрещённых прерываниях — uart_putc в цикле ожидания из обработчика или из ATOMIC_BLOCK никогда не дождётся: опустошить буфер может только обработчик, который сейчас не может выполниться.
• Два писателя в один буфер — например, printf и из основного цикла, и из обработчика таймера. Схема «один писатель, один читатель» перестаёт работать, и rb_put нужно выполнять в критической секции.
• Байт с ошибкой кадра в данных — FE означает испорченный стоп-бит, обычно из-за несовпадения скорости или помехи; такой байт отбрасывают и считают.