Реализация протокола обмена: кадры, CRC, таймауты

Как найти границу кадра

Пауза: Modbus RTU

В Modbus RTU кадр отделяется от соседних тишиной на линии не меньше 3,5 символа, а пауза больше 1,5 символа внутри кадра означает, что кадр неполный и его надо отбросить (спецификация Modbus over Serial Line, 2.5.1.1). Символ RTU — 11 бит: старт-бит, 8 бит данных, бит чётности и стоп-бит; без чётности ставят два стоп-бита (2.5.1). Отсюда интервалы:

Для скоростей выше 19 200 бит/с спецификация рекомендует фиксированные значения, чтобы не перегружать приёмник прерываниями таймера (2.5.1.1). Пауза требует таймера, который перезапускается при каждом принятом байте; на STM32 удобнее флаг IDLE или встроенный таймаут приёмника, где он есть. Слабое место паузы — посредники: преобразователь USB-UART или радиомодуль копит байты и выдаёт их пачками, и паузы внутри кадра появляются там, где их не было. Подробнее о самом протоколе — в статье о Modbus RTU, а расчёт интервалов и кадров — в инструменте «Modbus RTU» и в глоссарии: интервал тишины.

Заголовок и длина

Формат «маркер, длина, данные, CRC» — самый распространённый в самодельных протоколах:

Проблема в том, что байт 0xA5 может встретиться в данных или в CRC. Если приёмник включился посреди передачи или потерял байт, он примет такой байт за начало кадра. Защита строится из трёх частей:

1. Ограничение длины. LEN больше 64 — заведомо ложный заголовок, его отбрасывают сразу, не дожидаясь 200 байт «данных».
2. CRC покрывает и длину. Ложный кадр с ложной длиной почти наверняка не пройдёт проверку.
3. Поиск внутри отброшенного. После неудачи следующий маркер ищут, начиная со второго байта отброшенного кандидата. Настоящий кадр мог начаться внутри байт, которые разборщик уже принял как «данные» ложного кадра.

Третий пункт часто пропускают, и тогда после каждой ошибки теряется ещё один, совершенно исправный кадр.

Разделитель и байт-стаффинг: SLIP и COBS

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

SLIP (RFC 1055) заканчивает кадр байтом END = 0xC0. Байт 0xC0 в данных передаётся парой ESC ESC_END (0xDB 0xDC), а сам 0xDB — парой ESC ESC_ESC (0xDB 0xDD). Схема проста, но в худшем случае — данные из одних 0xC0 — кадр удваивается. В RFC оговорено, что у SLIP нет ни контроля ошибок, ни адресации: CRC добавляют в данные сами.

COBS (Consistent Overhead Byte Stuffing, Cheshire и Baker, 1999) кодирует данные так, что в них не остаётся нулевых байтов, а ноль служит разделителем. Каждый блок начинается с байта-кода: «через N − 1 байт будет ноль». Накладные расходы предсказуемы: не больше одного байта на 254 байта данных, примерно 0,4 %, а для кадров короче 254 байт — ровно один байт.

#include <stddef.h>
#include <stdint.h>

/* Encodes len bytes into dst (needs len + len / 254 + 1 bytes).
   Returns the encoded length; the 0x00 delimiter is not included. */
size_t cobs_encode(const uint8_t *src, size_t len, uint8_t *dst)
{
    size_t code_pos = 0, out = 1;
    uint8_t code = 1;

    for (size_t i = 0; i < len; i++) {
        if (src[i] == 0u) {
            dst[code_pos] = code;           /* close the block at the zero */
            code_pos = out++;
            code = 1;
        } else {
            dst[out++] = src[i];
            if (++code == 0xFFu) {          /* 254 non-zero bytes: full block */
                dst[code_pos] = code;
                code_pos = out++;
                code = 1;
            }
        }
    }
    dst[code_pos] = code;
    return out;
}

/* Decodes one frame without the delimiter. Returns the length or -1 if corrupted. */
int cobs_decode(const uint8_t *src, size_t len, uint8_t *dst)
{
    size_t in = 0, out = 0;

    while (in < len) {
        uint8_t code = src[in++];
        if (code == 0u)
            return -1;                      /* a zero cannot appear inside a frame */
        for (uint8_t i = 1; i < code; i++) {
            if (in >= len || src[in] == 0u)
                return -1;                  /* truncated block */
            dst[out++] = src[in++];
        }
        if (code != 0xFFu && in < len)
            dst[out++] = 0u;                /* the zero this block stood for */
    }
    return (int)out;
}

Пример: данные 11 22 00 33 кодируются в 03 11 22 02 33, а на линию уходят с разделителем: 03 11 22 02 33 00. Байт 03 значит «два байта данных, затем ноль», 02 — «один байт данных», а ноль в конце блока при последнем блоке не добавляется. Приёмник копит байты до нуля, декодирует и проверяет CRC; любая ошибка теряет не больше одного кадра.

Целостность: какая CRC

Простая сумма байтов не замечает перестановки байтов и компенсирующих друг друга ошибок. CRC (циклический избыточный код) — остаток от деления кадра на полином — к таким искажениям чувствителен, и для протоколов на UART с кадрами до сотен байт обычно хватает 16 бит.

Главное правило — указывать все параметры, а не только полином. «CRC-16 с полиномом 0x1021» — это минимум три разных алгоритма с разными результатами:

Каталог отдельно отмечает, что CRC-16/IBM-3740 часто ошибочно называют CRC-CCITT, хотя исторически это имя относится к отражённому варианту (CRC-16/KERMIT). Контрольное значение — CRC строки из девяти ASCII-символов «123456789» — первый тест любой реализации. Перебрать параметры и сверить результат можно в калькуляторе CRC.

Реализация CRC-16/MODBUS без таблицы: полином 0x8005 в отражённом виде записывается как 0xA001, так он и указан в спецификации Modbus (6.2.2):

#include <stddef.h>
#include <stdint.h>

uint16_t crc16_modbus(const uint8_t *data, size_t len)
{
    uint16_t crc = 0xFFFFu;
    while (len--) {
        crc ^= *data++;
        for (int i = 0; i < 8; i++)
            crc = (crc & 1u) ? (uint16_t)((crc >> 1) ^ 0xA001u) : (uint16_t)(crc >> 1);
    }
    return crc;
}

Побитовый вариант — 8 итераций на байт; табличный быстрее, но занимает 512 байт флеш-памяти под таблицу из 256 слов. Для кадров в десятки байт на 115 200 бод побитового обычно хватает с большим запасом. В Modbus CRC передаётся младшим байтом вперёд (2.5.1.2). У этого алгоритма нулевой остаток: CRC кадра вместе с дописанной CRC равна 0, что даёт вторую удобную проверку. Подробнее о том, как устроен CRC, — в глоссарии: CRC.

Разборщик потока

Разборщик для формата «SOF, LEN, DATA, CRC» из таблицы выше. Он получает байты по одному вместе с текущим временем, копит кандидата в кадр и после каждого байта проверяет, не собран ли кадр целиком. При ошибке CRC или недопустимой длине он сдвигает буфер к следующему маркеру внутри уже принятых байт.

/* frame.h */
#ifndef FRAME_H
#define FRAME_H

#include <stddef.h>
#include <stdint.h>

#define FRAME_SOF          0xA5u
#define FRAME_PAYLOAD_MAX  64u
#define FRAME_MAX          (FRAME_PAYLOAD_MAX + 4u)   /* SOF + LEN + data + CRC16 */
#define FRAME_GAP_MS       20u                        /* silence that drops a half frame */

typedef struct {
    uint8_t  buf[FRAME_MAX];    /* candidate frame, buf[0] == FRAME_SOF */
    size_t   n;
    uint32_t t_last_ms;
    uint32_t crc_errors, bad_len, timeouts;
} frame_parser_t;

uint16_t crc16_modbus(const uint8_t *data, size_t len);
size_t frame_build(uint8_t *dst, const uint8_t *payload, uint8_t len);
int frame_feed(frame_parser_t *p, uint8_t byte, uint32_t now_ms, uint8_t *payload);
int frame_next(frame_parser_t *p, uint8_t *payload);

#endif

/* frame.c */
#include <string.h>
#include "frame.h"

size_t frame_build(uint8_t *dst, const uint8_t *payload, uint8_t len)
{
    dst[0] = FRAME_SOF;
    dst[1] = len;
    memcpy(&dst[2], payload, len);
    uint16_t crc = crc16_modbus(&dst[1], (size_t)len + 1u);   /* LEN and data */
    dst[2u + len] = (uint8_t)(crc & 0xFFu);                  /* low byte first */
    dst[3u + len] = (uint8_t)(crc >> 8);
    return (size_t)len + 4u;
}

/* Drops buf[0..from) and then everything up to the next SOF,
   so that the buffer is either empty or starts with FRAME_SOF. */
static void drop_to_sof(frame_parser_t *p, size_t from)
{
    size_t k = from;
    while (k < p->n && p->buf[k] != FRAME_SOF)
        k++;
    memmove(p->buf, &p->buf[k], p->n - k);
    p->n -= k;
}

/* Returns the payload length when a frame is complete, -1 otherwise. */
static int check(frame_parser_t *p, uint8_t *payload)
{
    while (p->n >= 2u) {
        uint8_t len = p->buf[1];
        if (len > FRAME_PAYLOAD_MAX) {
            p->bad_len++;
            drop_to_sof(p, 1);
            continue;
        }
        size_t total = (size_t)len + 4u;
        if (p->n < total)
            return -1;                                       /* wait for more bytes */
        uint16_t rx = (uint16_t)(p->buf[2u + len] | ((uint16_t)p->buf[3u + len] << 8));
        if (crc16_modbus(&p->buf[1], (size_t)len + 1u) == rx) {
            memcpy(payload, &p->buf[2], len);
            drop_to_sof(p, total);                           /* keep the rest from the next SOF */
            return len;
        }
        p->crc_errors++;
        drop_to_sof(p, 1);                                   /* the real frame may start inside */
    }
    return -1;
}

int frame_feed(frame_parser_t *p, uint8_t byte, uint32_t now_ms, uint8_t *payload)
{
    if (p->n > 0u && (uint32_t)(now_ms - p->t_last_ms) > FRAME_GAP_MS) {
        p->n = 0;                                            /* stale half frame */
        p->timeouts++;
    }
    p->t_last_ms = now_ms;

    if (p->n == 0u && byte != FRAME_SOF)
        return -1;                                           /* hunting for SOF */
    p->buf[p->n++] = byte;
    return check(p, payload);
}

/* Further frames already sitting in the buffer: call until it returns -1. */
int frame_next(frame_parser_t *p, uint8_t *payload)
{
    return check(p, payload);
}

Разборщик вызывают так, чтобы после каждого байта забрать все готовые кадры:

int len = frame_feed(&parser, byte, now_ms, payload);
while (len >= 0) {
    handle_frame(payload, (uint8_t)len);
    len = frame_next(&parser, payload);
}

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

• Инвариант: буфер пуст или начинается с SOF. Все сокращения буфера идут через drop_to_sof() — и после ошибки, и после принятого кадра. Без этого остаток после кадра мог бы начинаться с мусора, и его первый байт был бы принят за маркер, а второй — за длину.
• Буфер не переполняется. Длина ограничена FRAME_PAYLOAD_MAX, поэтому кандидат никогда не бывает длиннее FRAME_MAX: когда собрано len + 4 байт, кадр либо принимается, либо буфер сокращается.
• Цикл в check() и frame_next(). После сдвига к следующему маркеру в буфере может уже лежать целый кадр, а иногда и не один: ложный заголовок с большой длиной «проглатывает» несколько настоящих кадров, и они обнаруживаются разом, когда его CRC не сошлась. frame_feed() возвращает первый, а остальные забирает frame_next() сразу, в том же цикле. Если их оставить в буфере до следующего байта, а следующий байт придёт после паузы, таймаут выбросит готовые кадры вместе с полукадрами.
• Таймаут между байтами. Полукадр, после которого линия молчит дольше FRAME_GAP_MS, отбрасывается: иначе первый байт следующего кадра был бы принят как продолжение старого. Разность now_ms − t_last_ms вычисляется в беззнаковом типе и остаётся правильной при переполнении счётчика (C11, 6.2.5p9).
• Время — параметр. Разборщик не вызывает HAL_GetTick(), поэтому таймаут тестируется без ожидания — подробнее в статье о модульном тестировании прошивки.
• Счётчики ошибок. crc_errors, bad_len и timeouts стоит выводить в журнал: по ним видно, плохая ли линия, не совпадает ли скорость, не теряются ли байты в буфере.

Тот же разборщик можно написать как автомат с явными состояниями — «ждём SOF», «ждём LEN», «копим данные», «ждём CRC» — на switch или на функциях-состояниях; такой вариант с похожим форматом кадра разобран в статье о конечных автоматах. Вариант с буфером-кандидатом короче и сам решает задачу поиска маркера внутри отброшенных байт.

Числа в кадре

Многобайтные поля собирают сдвигами в явно заданном порядке байт:

static inline uint16_t get_u16le(const uint8_t *p)
{
    return (uint16_t)(p[0] | ((uint16_t)p[1] << 8));
}

static inline void put_u16le(uint8_t *p, uint16_t v)
{
    p[0] = (uint8_t)(v & 0xFFu);
    p[1] = (uint8_t)(v >> 8);
}

Приведение указателя *(uint16_t *)&buf[3] выглядит короче, но зависит от порядка байт процессора, от выравнивания и нарушает правила совместного доступа типов в C (6.5p7). У Cortex-M0 невыровненных обращений нет вовсе: попытка вызывает HardFault (DUI 0497A, 3.3.4). Сдвиги работают одинаково на AVR, ARM и компьютере. Порядок байт каждого поля задаёт протокол, и его надо записать в описании: в Modbus, например, CRC передаётся младшим байтом вперёд.

Запрос, ответ, таймаут, повтор

В протоколе «запрос — ответ» ведущий отправляет запрос и ждёт. Спецификация Modbus описывает это так: после запроса ведущий запускает таймаут ответа, значение которого зависит от приложения; при ошибке в кадре ответа возможен повтор, а число повторов — настройка ведущего (Modbus over Serial Line, 2.4.1). Типичный таймаут, по той же спецификации, — от 1 с до нескольких секунд на 9600 бит/с.

Нижнюю границу таймаута считают по времени передачи. Пример: Modbus RTU на 9600 бит/с, запрос 8 байт, ответ 69 байт (32 регистра), символ 11 бит:

• запрос: 8 · 11 / 9600 = 9,2 мс;
• ответ: 69 · 11 / 9600 = 79,1 мс;
• пауза перед ответом не меньше 3,5 символа — 4,0 мс;
• время обработки у ведомого — по его документации, допустим, 50 мс.

Итого около 142 мс; с запасом на задержки переходников и планирование задач таймаут берут 300–500 мс. Слишком короткий таймаут даёт ложные ошибки и повторы, которые ещё больше загружают линию; слишком длинный — замедляет опрос, если ведомый отключён.

Повторы порождают вопрос, на который протокол должен ответить заранее: что будет, если запрос дошёл, а потерялся ответ? Ведущий повторит команду, и ведомый выполнит её второй раз. Для «прочитать регистр» или «установить уставку 25 °C» это безвредно — такие команды идемпотентны. Для «увеличить на 1» или «выдать 10 мл» — нет. Решения: делать команды идемпотентными (абсолютные значения вместо приращений) или добавить в кадр номер запроса, по которому ведомый узнаёт повтор и возвращает сохранённый ответ, не выполняя команду снова.

Расширяемость

Протокол живёт дольше первой версии прошивки. Несколько правил, которые потом сэкономят много времени:

• Поле типа сообщения, а не разбор по длине: неизвестные типы ведомый игнорирует или отвечает кодом «не поддерживается».
• Версия протокола в ответе на команду «кто ты» — по ней ведущий решает, какие команды доступны.
• Резервные биты передавать нулями, а при приёме игнорировать: тогда их можно будет задействовать, не ломая старые устройства.
• Описание формата в одном документе: порядок байт каждого поля, все параметры CRC, единицы измерения, диапазоны.

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

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

/* test_frame.c: host test, build: cc -std=c11 -Wall frame.c crc16.c test_frame.c */
#include <stdio.h>
#include <string.h>
#include "frame.h"

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

static int feed_all(frame_parser_t *p, const uint8_t *s, size_t n, uint32_t t, uint8_t *out)
{
    int got = -1;
    for (size_t i = 0; i < n; i++) {
        int r = frame_feed(p, s[i], t, out);
        if (r >= 0) got = r;
    }
    return got;
}

int main(void)
{
    uint8_t tx[FRAME_MAX], out[FRAME_PAYLOAD_MAX];
    const uint8_t data[] = { 0x10, FRAME_SOF, 0x02, 0x30 };   /* SOF inside the data */
    size_t n = frame_build(tx, data, sizeof data);

    /* 1. clean frame */
    frame_parser_t p = { 0 };
    CHECK(feed_all(&p, tx, n, 0, out) == (int)sizeof data);
    CHECK(memcmp(out, data, sizeof data) == 0);

    /* 2. a stray SOF right before the frame: its LEN byte is the real SOF */
    const uint8_t junk[] = { 0x00, FRAME_SOF };
    memset(&p, 0, sizeof p);
    feed_all(&p, junk, sizeof junk, 0, out);
    CHECK(feed_all(&p, tx, n, 0, out) == (int)sizeof data);
    CHECK(p.bad_len == 1);

    /* 3. corrupted frame immediately followed by a good one */
    uint8_t bad[FRAME_MAX];
    memcpy(bad, tx, n);
    bad[2] ^= 0x01u;                                          /* damage the data */
    memset(&p, 0, sizeof p);
    CHECK(feed_all(&p, bad, n, 0, out) == -1);
    CHECK(feed_all(&p, tx, n, 0, out) == (int)sizeof data);
    CHECK(p.crc_errors >= 1);

    /* 4. a false header swallows two real frames; garbage between them */
    uint8_t a[FRAME_MAX], b[FRAME_MAX], stream[64];
    const uint8_t pa[] = { 0x11 }, pb[] = { 0x22, 0x33 };
    size_t na = frame_build(a, pa, sizeof pa), nb = frame_build(b, pb, sizeof pb);
    size_t k = 0;
    stream[k++] = FRAME_SOF;                       /* false header ... */
    stream[k++] = 10;                              /* 2 + 10 + 2 bytes: exactly A, 0x00, B */
    memcpy(&stream[k], a, na); k += na;
    stream[k++] = 0x00;                            /* garbage after frame A */
    memcpy(&stream[k], b, nb); k += nb;
    memset(&p, 0, sizeof p);
    int got_a = 0, got_b = 0;
    for (size_t i = 0; i < k; i++) {
        int r = frame_feed(&p, stream[i], 0, out);
        while (r >= 0) {
            if (r == 1 && out[0] == 0x11) got_a++;
            if (r == 2 && out[0] == 0x22 && out[1] == 0x33) got_b++;
            r = frame_next(&p, out);
        }
    }
    CHECK(got_a == 1 && got_b == 1);
    CHECK(p.n == 0);                               /* nothing left behind */
    CHECK(p.crc_errors == 1 && p.bad_len == 0);   /* only the false header failed */

    /* 5. half frame, silence, then a full frame */
    memset(&p, 0, sizeof p);
    feed_all(&p, tx, 3, 0, out);
    CHECK(feed_all(&p, tx, n, 100, out) == (int)sizeof data);
    CHECK(p.timeouts == 1);

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

Второй сценарий проверяет поиск маркера внутри отброшенного: лишний байт 0xA5 перед кадром становится ложным заголовком, его «длиной» оказывается настоящий маркер 0xA5 = 165, и кандидат отбрасывается по длине. Если resync() переписать «по-простому» — очищать буфер целиком, — вместе с ложным заголовком пропадёт и маркер настоящего кадра, и тест провалится. В третьем сценарии внутри испорченного кадра есть байт 0xA5 из данных: разборщик начинает с него нового кандидата, тот тоже не проходит CRC, и только потом находится начало исправного кадра. Ещё сильнее такие тесты становятся со случайными данными: тысячи кадров со случайными ошибками, собранные с -fsanitize=address,undefined, находят выход за границы буфера, который на плате проявился бы раз в неделю.

Четвёртый сценарий — самый неприятный для разборщика. Ложный заголовок с длиной 10 «проглатывает» ровно кадр A, байт мусора и кадр B. Когда приходит последний байт B, CRC ложного кадра не сходится, разборщик сдвигается к маркеру A и находит в буфере сразу два готовых кадра, причём между ними лежит ноль. Первый возвращает frame_feed(), второй — frame_next(), а drop_to_sof() убирает мусор, чтобы остаток снова начинался с маркера. Без frame_next() кадр B ждал бы следующего байта и пропал бы по таймауту, если линия замолчит; без drop_to_sof() после принятого кадра ноль стал бы «маркером» нового кандидата: разборщик бы потом выправился поиском маркера, но насчитал бы лишнюю ошибку длины, и счётчики ошибок перестали бы отражать состояние линии — это и проверяет последняя строка сценария.

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

• CRC названа, но не описана — «CRC-16» без начального значения и отражения; две стороны считают по-разному, и ни один кадр не проходит.
• Нет ограничения длины — ложный заголовок с длиной 250 заставляет ждать 250 байт и теряет все кадры за это время.
• После ошибки буфер очищается целиком — теряется исправный кадр, начавшийся внутри отброшенных байт.
• Нет таймаута между байтами — половина кадра после обрыва склеивается со следующим кадром.
• Пауза как граница кадра через USB-UART или радиомодуль — посредник группирует байты, и паузы оказываются не там.
• Многобайтные поля через приведение указателя — разный порядок байт на разных процессорах и HardFault на Cortex-M0.
• Повтор неидемпотентной команды — потерянный ответ превращает одну команду в две.
• Разбор в обработчике прерывания — долгий обработчик и потеря байт; обработчик только кладёт байты в буфер.