Конечные автоматы в прошивке микроконтроллера

Зачем: задержки против неблокирующего кода

Типичная первая прошивка выглядит как сценарий: включить насос, delay(2000), перейти на полную скорость, ждать кнопку в цикле. Пока идёт задержка, программа слепа: не видит нажатие «стоп», не забирает байты из UART, не мигает светодиодом. Добавить вторую задачу в такой код невозможно, не переписав первую.

Автомат решает это иначе. У каждой задачи есть переменная состояния и функция шага, которая:

1. смотрит на текущее состояние;
2. проверяет события: уровень вывода, флаг от прерывания, принятый байт, истёкший таймер;
3. если событие подходит, выполняет действие и меняет состояние;
4. сразу возвращается.

Главный цикл по очереди вызывает функции шага всех задач. Ни одна не ждёт, поэтому цикл проходит за микросекунды, и каждая задача реагирует на своё событие почти мгновенно. Ожидание превращается в состояние: «жду окончания разгона» — это не delay(2000), а состояние STARTING и таймер, который проверяется на каждом проходе. Попутно у программы появляется явное описание поведения, которое можно нарисовать, проверить и протестировать.

Состояния, события, переходы, действия

Термины удобно брать из спецификации UML, где автоматы описаны в главе 14; по её словам, автоматы UML основаны на объектно-ориентированном варианте statecharts Дэвида Харела и в нескольких местах отличаются от оригинала по семантике (UML 2.5.1, 14.1).

• Состояние (state) — ситуация, в которой выполняется некоторое условие, обычно понятное из названия: «насос выключен», «идёт разгон» (14.2.3.4). Состояние не обязано быть статичным: в состоянии RUN двигатель работает, но автомат ждёт следующего события.
• Событие (event) и триггер — то, что может запустить переход: нажатие, таймаут, принятый байт, ошибка. В прошивке события — это обычно значения перечисления.
• Переход (transition) — дуга из одного состояния в другое. У него есть триггеры, необязательное условие (guard) и действие (effect), которое выполняется при переходе (14.2.3.8); переход разрешён, только если условие истинно (14.2.3.9.2). Текстовая запись перехода в UML — событие [условие] / действие (14.2.4.8), например TIMEOUT [давление > 0,5 бар] / motor_full_speed().
• Действия входа и выхода (entry, exit) — код, который выполняется при каждом входе в состояние и при каждом выходе из него (14.2.3.4.3). Они избавляют от повторения одного и того же действия на всех входящих переходах. Порядок такой: сначала exit старого состояния, затем effect перехода, затем entry нового (14.2.3.4.5, 14.2.3.9.6).
• Внутренний переход (internal) реагирует на событие, не выходя из состояния: entry и exit не выполняются (14.2.3.8.1). Пример — счёт импульсов в состоянии MEASURE.

Событие, для которого в текущем состоянии нет перехода, по UML просто отбрасывается (14.2.3.9.1). В прошивке это тоже решение по умолчанию, но о нём нужно помнить: «нажали СТОП во время аварии — ничего не произошло» может быть и правильным поведением, и ошибкой.

Автоматы Мура и Мили

Два классических вида автоматов различаются тем, от чего зависит выход.

У автомата Мили выход и следующее состояние определяются текущим входом и текущим состоянием — так модель последовательностной схемы определена в статье Дж. Мили 1955 года (разд. 2.1). В прошивке это действия, привязанные к переходам: событие «короткое нажатие» появляется в момент перехода из «кнопка отпускается» в «отпущена», а не пока автомат находится в каком-то состоянии.

У автомата Мура выход определяется только текущим состоянием. В прошивке это выходы, которые соответствуют состоянию целиком: светодиод «авария» горит всё время, пока автомат в FAULT, ШИМ насоса равен 100 %, пока он в RUN. Такие выходы естественно задавать в действии входа в состояние или вычислять по состоянию на каждом шаге. Название дано по работе Э. Ф. Мура «Gedanken-Experiments on Sequential Machines» из сборника Automata Studies (1956); на неё, тогда ещё готовившуюся к печати, ссылается и Мили.

На практике автомат в прошивке почти всегда смешанный, и называть его «Муром» или «Мили» нужно разве что для выбора, где писать код: постоянный выход — в состоянии, разовое действие — в переходе.

Пример 1: кнопка с коротким и длинным нажатием на switch-case

Механическая кнопка при нажатии и отпускании дребезжит — выдаёт серию переключений вместо одного (подробнее — в статье о дребезге контактов). Задача: отфильтровать дребезг и различить короткое нажатие и удержание дольше 800 мс. Длинное нажатие сообщаем сразу по достижении порога, не дожидаясь отпускания, а короткое — только после отпускания: до этого момента неизвестно, не станет ли оно длинным.

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

#define DEBOUNCE_MS    20u
#define LONG_PRESS_MS 800u

typedef enum {
    BTN_IDLE,                /* released */
    BTN_PRESS_DEBOUNCE,      /* contact closed, waiting for it to settle */
    BTN_PRESSED,             /* press confirmed, long press not reached yet */
    BTN_LONG_HELD,           /* long press already reported */
    BTN_RELEASE_DEBOUNCE     /* contact opened, waiting for it to settle */
} btn_state_t;

typedef enum { BTN_EV_NONE, BTN_EV_SHORT, BTN_EV_LONG } btn_event_t;

typedef struct {
    btn_state_t state;
    uint32_t t_state;        /* when the current debounce interval started */
    uint32_t t_press;        /* when the press was confirmed */
    bool was_long;
} button_t;

static bool elapsed(uint32_t now, uint32_t start, uint32_t timeout)
{
    return (uint32_t)(now - start) >= timeout;   /* correct across wraparound */
}

/* Called every 1-10 ms from the main loop; pressed is the raw pin level. */
btn_event_t button_update(button_t *b, bool pressed, uint32_t now)
{
    btn_event_t ev = BTN_EV_NONE;

    switch (b->state) {
    case BTN_IDLE:
        if (pressed) {
            b->state = BTN_PRESS_DEBOUNCE;
            b->t_state = now;
        }
        break;
    case BTN_PRESS_DEBOUNCE:
        if (!pressed) {
            b->state = BTN_IDLE;                         /* bounce or noise */
        } else if (elapsed(now, b->t_state, DEBOUNCE_MS)) {
            b->state = BTN_PRESSED;
            b->t_press = now;
        }
        break;
    case BTN_PRESSED:
        if (!pressed) {
            b->state = BTN_RELEASE_DEBOUNCE;
            b->t_state = now;
            b->was_long = false;
        } else if (elapsed(now, b->t_press, LONG_PRESS_MS)) {
            b->state = BTN_LONG_HELD;
            ev = BTN_EV_LONG;                            /* reported while still held */
        }
        break;
    case BTN_LONG_HELD:
        if (!pressed) {
            b->state = BTN_RELEASE_DEBOUNCE;
            b->t_state = now;
            b->was_long = true;
        }
        break;
    case BTN_RELEASE_DEBOUNCE:
        if (pressed) {
            b->state = b->was_long ? BTN_LONG_HELD : BTN_PRESSED;   /* bounce */
        } else if (elapsed(now, b->t_state, DEBOUNCE_MS)) {
            b->state = BTN_IDLE;
            if (!b->was_long)
                ev = BTN_EV_SHORT;                       /* known only after release */
        }
        break;
    }
    return ev;
}

Как это работает по времени. Кнопку нажали в момент 1000 мс, контакт дребезжит до 1004 мс. В 1000 мс автомат уходит в PRESS_DEBOUNCE, при первом же размыкании возвращается в IDLE, при следующем замыкании снова запоминает время. Последнее замыкание в 1004 мс, и в 1024 мс — через 20 мс стабильного уровня — автомат переходит в PRESSED. Если кнопку отпустили в 1300 мс, через 20 мс без дребезга выдаётся SHORT. Если держат дальше, в 1824 мс выдаётся LONG, и последующее отпускание ничего не выдаёт. 20 мс — константа примера: время дребезга своей кнопки лучше посмотреть осциллографом и взять с запасом.

Функцию вызывают из главного цикла раз в 1–10 мс; чем реже вызов, тем грубее отсчёт времени, но логика не меняется. Флаг was_long — небольшая уступка: без него понадобились бы два состояния отпускания. Это обычный компромисс между числом состояний и дополнительными переменными; важно только, чтобы таких переменных было мало и каждая имела ясный смысл.

Обратите внимание на switch без default. Так компилятор помогает: с ключом -Wall GCC включает -Wswitch и предупреждает, если в switch по перечислению пропущено какое-то значение — например, после добавления нового состояния. Но если default есть, -Wswitch молчит; предупреждение и при наличии default даёт отдельный ключ -Wswitch-enum (GCC Manual, Warning Options). Если default нужен для защиты от испорченной переменной состояния, включайте -Wswitch-enum.

Время: тик в миллисекундах и таймауты без блокировки

Автомату нужны часы. Обычно это 32-битный счётчик миллисекунд, который увеличивает прерывание таймера с частотой 1 кГц: SysTick на Cortex-M, таймер в режиме CTC на AVR. Настройка таймеров разобрана в статьях о таймерах AVR и о таймерах STM32, предделитель и период поможет посчитать калькулятор таймера.

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

static volatile uint32_t ms_ticks;         /* incremented by a 1 kHz timer ISR */

void tick_isr(void)                        /* SysTick_Handler or TIMER0_COMPA_vect body */
{
    ms_ticks++;
}

uint32_t millis(void)
{
    return ms_ticks;                       /* on AVR read it inside ATOMIC_BLOCK */
}

typedef struct {
    uint32_t start;
    uint32_t period;
    bool armed;
} soft_timer_t;

void timer_start(soft_timer_t *t, uint32_t period_ms)
{
    t->start = millis();
    t->period = period_ms;
    t->armed = true;
}

/* true exactly once when the period has elapsed */
bool timer_expired(soft_timer_t *t)
{
    if (t->armed && (uint32_t)(millis() - t->start) >= t->period) {
        t->armed = false;
        return true;
    }
    return false;
}

На Cortex-M чтение выровненного 32-битного слова неделимо, а 8-битный AVR читает uint32_t четырьмя командами, и прерывание таймера может вклиниться между ними. Поэтому на AVR millis() читает счётчик внутри ATOMIC_BLOCK(ATOMIC_RESTORESTATE) — подробно в статье о прерываниях и общих данных.

Почему сравнивают разность, а не моменты времени

32-битный счётчик миллисекунд переполняется через 2^32 мс = 4 294 967 296 мс ≈ 49,7 суток и начинает с нуля. Устройство, которое работает дольше, обязательно встретит этот момент, и сравнение должно его пережить.

Стандарт C гарантирует: вычисление с беззнаковыми операндами не переполняется, результат берётся по модулю числа, на единицу большего максимального значения типа (C11, 6.2.5p9). Для uint32_t — по модулю 2^32. Поэтому разность now - start равна прошедшему времени, даже если между start и now счётчик перешёл через ноль. А сумма start + timeout в момент перехода переполняется и даёт маленькое число.

Пример. Таймер на 500 мс запущен при start = 0xFFFFFF00 (4 294 967 040), через 16 мс now = 0xFFFFFF10.

Через 500 мс now = 0x000000F4 = 244, и разность снова верна: 244 − 4 294 967 040 по модулю 2^32 = 500.

Два ограничения. Первое: проверять таймер нужно хотя бы раз за 2^32 − timeout мс, иначе разность «обернётся» и таймаут потеряется; при опросе в главном цикле это не проблема. Второе касается коротких счётчиков. Если счётчик 16-битный, а int 32-битный, как на ARM, операнды uint16_t перед вычитанием продвигаются до int (C11, 6.3.1.1p2), и разность может быть отрицательной: при start = 65530 и now = 5 получится −65525, и сравнение now - start >= 10 никогда не станет истинным. Явное приведение (uint16_t)(now - start) берёт результат по модулю 65536 (C11, 6.3.1.3) и даёт верные 11. Приведение к типу счётчика стоит писать всегда — тогда код одинаково работает на AVR с 16-битным int и на ARM с 32-битным.

Пример 2: таблица переходов

Когда автомат описывается фразами «в состоянии A по событию E перейти в B и выполнить X», его удобно хранить прямо в виде такой таблицы — массива структур. Пример — контроллер насоса с плавным пуском и защитой от перегрузки.

Остальные пары игнорируются: START в состоянии RUN или FAULT ничего не делает, и выйти из аварии можно только сбросом. Код почти дословно повторяет таблицу:

/* pump_fsm.h */
typedef enum { ST_OFF, ST_STARTING, ST_RUN, ST_FAULT, ST_COUNT } pump_state_t;
typedef enum { EV_START, EV_STOP, EV_TIMEOUT, EV_OVERLOAD, EV_RESET, EV_COUNT } pump_event_t;

void pump_init(void);
void pump_dispatch(pump_event_t ev);
pump_state_t pump_state(void);

/* pump_fsm.c */
#include <stddef.h>
#include "pump_fsm.h"
#include "pump_hw.h"     /* motor_soft_start(), motor_full_speed(), motor_off(), ... */

typedef void (*action_t)(void);

typedef struct {
    pump_state_t state;          /* current state */
    pump_event_t event;          /* trigger */
    pump_state_t next;           /* target state */
    action_t     action;         /* effect, may be NULL */
} transition_t;

static const transition_t table[] = {
    { ST_OFF,      EV_START,    ST_STARTING, motor_soft_start },
    { ST_STARTING, EV_TIMEOUT,  ST_RUN,      motor_full_speed },
    { ST_STARTING, EV_STOP,     ST_OFF,      motor_off        },
    { ST_STARTING, EV_OVERLOAD, ST_FAULT,    fault_trip       },
    { ST_RUN,      EV_STOP,     ST_OFF,      motor_off        },
    { ST_RUN,      EV_OVERLOAD, ST_FAULT,    fault_trip       },
    { ST_FAULT,    EV_RESET,    ST_OFF,      fault_clear      },
};

static pump_state_t current;

void pump_init(void)
{
    current = ST_OFF;
    motor_off();
}

void pump_dispatch(pump_event_t ev)
{
    for (size_t i = 0; i < sizeof table / sizeof table[0]; i++) {
        if (table[i].state == current && table[i].event == ev) {
            if (table[i].action != NULL)
                table[i].action();
            current = table[i].next;
            return;
        }
    }
    /* no row: the event is ignored in this state */
}

pump_state_t pump_state(void)
{
    return current;
}

Таймаут разгона получается из программного таймера: действие motor_soft_start() запускает его на 2000 мс, а главный цикл проверяет и отправляет автомату событие:

static soft_timer_t start_timer;

void motor_soft_start(void)
{
    /* pwm_ramp_begin(); */
    timer_start(&start_timer, 2000u);      /* EV_TIMEOUT in 2 s */
}

void motor_off(void) { start_timer.armed = false; /* pwm_set(0); */ }

void app_loop_once(void)
{
    if (timer_expired(&start_timer))
        pump_dispatch(EV_TIMEOUT);         /* main context: no need to queue */

    int ev;
    while ((ev = evq_get()) >= 0)          /* events posted by ISRs, see below */
        pump_dispatch((pump_event_t)ev);
}

Что даёт таблица:

• Обзор. Всё поведение — семь строк в одном месте. Добавить переход — добавить строку, а не искать нужную ветку if в пятидесяти строках switch.
• Константа. Таблица static const и на Cortex-M целиком лежит во Flash, не занимая ОЗУ. На классических AVR константы по умолчанию копируются в ОЗУ: секция .rodata у них входит в .data (см. статью о скриптах компоновщика), поэтому большую таблицу там размещают во Flash через PROGMEM и читают функциями pgm_read_* — подробнее в статье о языке C для микроконтроллеров.
• Проверяемость. Таблицу можно перебрать тестом — пример ниже.

Линейный поиск проходит таблицу строка за строкой, и время выбора перехода растёт с её длиной. Если состояний и событий десятки, вместо списка строк используют двумерный массив next[ST_COUNT][EV_COUNT] и такой же массив действий: выбор перехода становится одной индексацией, но пустые клетки занимают память, а саму таблицу труднее читать. Условия (guard) добавляют пятым полем — указателем на функцию bool (*guard)(void): строка срабатывает, только если условие истинно, а если нет — поиск продолжается со следующей строки той же пары. Так можно записать TIMEOUT [давление в норме] → RUN и TIMEOUT [давления нет] → FAULT.

В pump_dispatch() действие выполняется до смены current. Значит, внутри действия автомат ещё в старом состоянии, и если действие само вызовет pump_dispatch(), обработка пойдёт по неверной строке. Это нарушение run-to-completion; правило простое — действия не вызывают диспетчер своего автомата, а кладут событие в очередь.

Пример 3: функции-состояния для разборщика пакетов UART

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

Формат пакета в примере:

#include <stdint.h>

#define PKT_HEADER 0xA5u
#define PKT_MAX      32u
#define PKT_GAP_MS    5u          /* silence that drops a half-received packet */

typedef struct parser parser_t;
typedef void (*parser_state_t)(parser_t *p, uint8_t byte);

struct parser {
    parser_state_t state;         /* the current state is a function */
    uint8_t len, pos, sum;
    uint8_t payload[PKT_MAX];
    uint32_t t_last;              /* reception time of the previous byte */
    uint16_t errors;
    void (*on_packet)(const uint8_t *data, uint8_t len);
};

static void st_header(parser_t *p, uint8_t byte);
static void st_length(parser_t *p, uint8_t byte);
static void st_payload(parser_t *p, uint8_t byte);
static void st_checksum(parser_t *p, uint8_t byte);

static void st_header(parser_t *p, uint8_t byte)
{
    if (byte == PKT_HEADER)
        p->state = st_length;     /* anything else is skipped: resync */
}

static void st_length(parser_t *p, uint8_t byte)
{
    if (byte == 0u || byte > PKT_MAX) {
        p->errors++;
        p->state = st_header;
        return;
    }
    p->len = byte;
    p->pos = 0u;
    p->sum = byte;
    p->state = st_payload;
}

static void st_payload(parser_t *p, uint8_t byte)
{
    p->payload[p->pos++] = byte;
    p->sum = (uint8_t)(p->sum + byte);
    if (p->pos == p->len)
        p->state = st_checksum;
}

static void st_checksum(parser_t *p, uint8_t byte)
{
    if ((uint8_t)(p->sum + byte) == 0u)
        p->on_packet(p->payload, p->len);
    else
        p->errors++;
    p->state = st_header;
}

void parser_init(parser_t *p, void (*on_packet)(const uint8_t *, uint8_t))
{
    p->state = st_header;
    p->errors = 0u;
    p->t_last = 0u;
    p->on_packet = on_packet;
}

/* Called from the main loop for every byte taken from the UART ring buffer;
 * t_rx is the time the ISR received the byte, not the time it is processed. */
void parser_feed(parser_t *p, uint8_t byte, uint32_t t_rx)
{
    if (p->state != st_header && (uint32_t)(t_rx - p->t_last) > PKT_GAP_MS) {
        p->errors++;              /* gap on the line inside a packet: start over */
        p->state = st_header;
    }
    p->t_last = t_rx;
    p->state(p, byte);
}

Пример с числами. Пакет с данными 0x10, 0x20, 0x30: LEN = 3, сумма 3 + 0x10 + 0x20 + 0x30 = 0x63, и CHK = 0x100 − 0x63 = 0x9D. Поток A5 03 10 20 30 9D проходит состояния header → length → payload ×3 → checksum, проверка (0x63 + 0x9D) mod 256 = 0 сходится, вызывается on_packet(). Если в линии был мусор, st_header пропускает всё до байта 0xA5. Если 0xA5 случайно встретился внутри мусора, разборщик начнёт собирать ложный пакет. Обычно он отбросит его по длине или контрольной сумме, но не всегда: 8-битная сумма случайных байт совпадёт с нулём примерно в одном случае из 256, и тогда мусор будет принят как пакет. Для ответственных протоколов поэтому берут CRC-8 или CRC-16 (посчитать их поможет калькулятор CRC). Кроме того, пока набирается ложный пакет, разборщик съедает байты следующего настоящего, и тот тоже может потеряться, если его не спасёт таймаут.

Таймаут между байтами защищает от зависания на половине пакета: если передатчик перезагрузился посреди посылки, без таймаута разборщик принял бы первые байты следующего пакета за продолжение старого. Важно, какое время сравнивать. Байты попадают в разборщик не из обработчика прерывания, а из кольцевого буфера приёма: обработчик UART кладёт байт в буфер, главный цикл забирает и вызывает parser_feed(). Если брать время в момент обработки, любая задержка главного цикла — запись во Flash, обновление дисплея — растянет промежуток между уже принятыми байтами, и годный пакет будет отброшен; и наоборот, настоящая пауза в линии сожмётся, если байты забраны из буфера пачкой. Поэтому время ставит обработчик прерывания, и оно хранится в буфере рядом с байтом:

/* UART receive ring buffer: the ISR stores each byte with its reception time */
#include <stdbool.h>
#include <stdint.h>

#define RX_SIZE 64u                        /* power of two, at most 128 */

typedef struct {
    uint32_t t;                            /* ms tick when the byte arrived */
    uint8_t byte;
} rx_item_t;

uint32_t millis(void);

static volatile uint32_t rx_time[RX_SIZE];
static volatile uint8_t rx_data[RX_SIZE];
static volatile uint8_t rx_head;           /* written only by the ISR */
static volatile uint8_t rx_tail;           /* written only by the main loop */

/* Body of the receive ISR, called with the byte read from the data register.
 * On AVR interrupts are disabled inside an ISR, so millis() is read whole. */
void uart_rx_isr(uint8_t byte)
{
    uint8_t head = rx_head;
    if ((uint8_t)(head - rx_tail) == RX_SIZE)
        return;                            /* overflow: the byte is dropped */
    rx_data[head & (RX_SIZE - 1u)] = byte;
    rx_time[head & (RX_SIZE - 1u)] = millis();   /* reception time, not processing time */
    rx_head = (uint8_t)(head + 1u);
}

/* Main loop: take one byte with its timestamp; false when the buffer is empty. */
bool uart_rx_get(rx_item_t *out)
{
    uint8_t tail = rx_tail;
    if (rx_head == tail)
        return false;
    out->byte = rx_data[tail & (RX_SIZE - 1u)];
    out->t = rx_time[tail & (RX_SIZE - 1u)];
    rx_tail = (uint8_t)(tail + 1u);
    return true;
}

Главный цикл передаёт разборщику именно это время:

rx_item_t it;
while (uart_rx_get(&it))
    parser_feed(&parser, it.byte, it.t);   /* compare reception times */

Внутри обработчика AVR прерывания запрещены, поэтому 32-битный тик читается целиком; на Cortex-M неделимо само чтение выровненного слова. Точность метки — один тик, то есть ±1 мс, и порог должен быть заметно больше.

Сам порог зависит от передатчика. На скорости 115 200 бит/с в формате 8N1 байт идёт 10 / 115 200 ≈ 86,8 мкс, так что 5 мс — время почти 58 байт. Если передатчик — другой микроконтроллер, который отправляет пакет из буфера подряд, паузы такой длины внутри пакета не ожидаются, и 5 мс в примере — выбранная для этого случая константа. Если передатчик — программа на компьютере, гарантий нет: она не работает в реальном времени, и пакет может уйти по частям с паузами; порог тогда выбирают по измерениям с запасом. Обратное направление тоже не мгновенно: USB-UART адаптер FTDI отправляет принятые байты в компьютер, когда наберётся 62 байта данных или истечёт таймер задержки, по умолчанию 16 мс (FTDI AN232B-04, разд. 3.1). Разборщик на стороне компьютера видит такие пачки, и паузы между ними не совпадают с паузами в линии.

Проверка таймаута в parser_feed() срабатывает с приходом следующего байта; если нужно сообщить об ошибке сразу, в главном цикле сравнивают время последнего принятого байта с текущим. Кольцевой буфер подробно разобран в статье о прерываниях, приём по прерыванию — в статье об интерфейсе UART.

У функций-состояний есть цена: таблицу всех переходов приходится собирать по телу функций. Сравнение указателей p->state != st_header законно, а вот сохранить состояние во Flash или передать по сети в виде указателя нельзя — для этого нужен номер.

Очередь событий из прерываний

Событие часто рождается в обработчике прерывания: перегрузка от компаратора, нажатие по внешнему прерыванию, конец преобразования АЦП. Вызывать pump_dispatch() прямо из обработчика опасно. Если главный цикл в этот момент сам находится внутри pump_dispatch() — например, обрабатывает таймаут, — обработчик выполнит второй переход посреди первого: прочитает старое current, вызовет действие, запишет новое состояние, а после возврата первый вызов перезапишет его своим. Автомат окажется в состоянии, которое не соответствует ни одной строке таблицы.

UML описывает правильное поведение как run-to-completion: следующее событие отдаётся автомату только после того, как обработка предыдущего полностью закончилась и автомат пришёл в устойчивое состояние (14.2.3.9.1). Спецификация уточняет, что это не запрет прерываний: поток, выполняющий шаг автомата, может быть вытеснен, лишь бы другой шаг того же автомата не начался раньше. В прошивке без ОС это достигается очередью: обработчики только кладут события, а главный цикл достаёт и обрабатывает по одному.

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

#define EVQ_SIZE 16u                       /* power of two, at most 128 */

static volatile uint8_t evq_buf[EVQ_SIZE];
static volatile uint8_t evq_head;          /* written only by evq_post() */
static volatile uint8_t evq_tail;          /* written only by evq_get() */
static volatile uint8_t evq_lost;          /* overflow counter, for debugging */

/* Producer. Lock-free only with a single writer. If several ISRs that can
 * preempt each other post, or the main loop posts too (FSM actions do),
 * call it inside a critical section on any MCU. */
bool evq_post(uint8_t ev)
{
    uint8_t head = evq_head;
    if ((uint8_t)(head - evq_tail) == EVQ_SIZE) {
        evq_lost++;
        return false;
    }
    evq_buf[head & (EVQ_SIZE - 1u)] = ev;  /* store the event before publishing it */
    evq_head = (uint8_t)(head + 1u);
    return true;
}

/* Consumer: called only from the main loop. Returns -1 when empty. */
int evq_get(void)
{
    uint8_t tail = evq_tail;
    if (evq_head == tail)
        return -1;
    uint8_t ev = evq_buf[tail & (EVQ_SIZE - 1u)];
    evq_tail = (uint8_t)(tail + 1u);
    return ev;
}

Это кольцевой буфер с индексами-счётчиками из статьи о прерываниях и общих данных, только массив здесь тоже volatile. Документация GCC разрешает компилятору переставлять обращения к volatile-объектам только в пределах участка между соседними точками следования, но не через точку следования (GCC Manual, When is a Volatile Object Accessed?), поэтому запись события в ячейку и запись индекса в соседних операторах выполнятся в порядке программы, и барьеры для компилятора не нужны. Это верно для одного ядра: процессор видит собственные обращения в порядке программы. Если писатель и читатель работают на разных ядрах двухъядерного микроконтроллера, между записью данных и записью флага или индекса в общей памяти Arm требует команду DMB или DSB (Application Note 321, разд. 4.18), то есть __DMB(); подробнее — в статье о прерываниях.

Буфер корректен без запрета прерываний, только пока писатель один. Писателей становится несколько в двух случаях, и тогда evq_post() выполняют в критической секции на любом микроконтроллере (на AVR — ATOMIC_BLOCK(ATOMIC_RESTORESTATE), на Cortex-M — с сохранением и восстановлением PRIMASK):

• события кладут несколько обработчиков, которые могут прервать друг друга, как при разных приоритетах в NVIC; на AVR обработчики по умолчанию не вложены, и этот случай там не возникает;
• в очередь пишет ещё и главный цикл — например, действия автомата, которые по правилу из примера 2 кладут события в очередь, а не вызывают диспетчер. Обработчик может прервать evq_post() главного цикла между чтением и записью evq_head, и одно из событий потеряется.

Другой вариант — завести для внутренних событий главного цикла отдельную очередь, в которую обработчики не пишут.

Несколько правил для очереди:

• Размер считают по худшему случаю: сколько событий может прийти, пока главный цикл занят самым долгим своим делом. Счётчик evq_lost в отладочной сборке покажет, хватает ли.
• События — не флаги. Два нажатия подряд будут двумя событиями, а не одним, как у флага. Если важно только последнее значение (текущая температура), очередь не нужна: достаточно переменной.
• Сон. Когда очередь пуста, главный цикл может усыпить процессор до следующего прерывания. Проверка «очередь пуста» и засыпание должны быть атомарны, иначе событие, пришедшее между ними, пролежит до следующего пробуждения. Как это сделать на AVR и Cortex-M, разобрано в той же статье о прерываниях.

Иерархические автоматы

В плоском автомате общие реакции приходится повторять. У насоса событие OVERLOAD ведёт в FAULT и из STARTING, и из RUN; добавьте состояния «прогрев», «реверс», «промывка» — и строка с OVERLOAD появится у каждого. Statecharts решают это вложенностью.

• Составное состояние содержит вложенные подсостояния (UML 2.5.1, 14.2.3.4.1). Можно ввести состояние ON с подсостояниями STARTING и RUN и один переход OVERLOAD → FAULT из ON — он действует, в каком бы подсостоянии ни был автомат.
• Приоритет. Если событие обрабатывают и подсостояние, и объемлющее состояние, срабатывает переход подсостояния (14.2.3.9.4). Так подсостояние может переопределить общую реакцию.
• Порядок entry и exit. При выходе из составного состояния сначала выполняется exit самого внутреннего активного состояния, затем — объемлющих, наружу (14.2.3.4.6); при входе — наоборот, снаружи внутрь.
• История. Переход в псевдосостояние history возвращает составное состояние в то подсостояние, в котором оно было при последнем выходе (shallow — верхний уровень, deep — вся вложенная конфигурация, 14.2.3.4.4). Пример: «пауза» прерывает цикл стиральной машины, а «продолжить» возвращает его в ту же фазу.
• Ортогональные регионы — несколько независимых автоматов внутри одного состояния, активных одновременно (14.2.3.4.1): например, режим двигателя и режим индикации.

На C иерархию обычно делают на функциях-состояниях: обработчик подсостояния возвращает признак «не обработано», и диспетчер передаёт событие обработчику родителя. Для прошивки с несколькими автоматами часто хватает двух уровней, сделанных вручную: общие события (авария, выключение питания) проверяются до switch или до поиска по таблице.

Тестирование таблицы переходов на компьютере

Логика автомата не зависит от железа, если действия вынесены в отдельные функции. Тогда модуль pump_fsm.c компилируется для компьютера без изменений, а вместо настоящего pump_hw.c с ШИМ и выводами к тесту подключают поддельные действия, которые пишут в журнал по букве:

/* test_pump.c */
#include <stdio.h>
#include <string.h>
#include "pump_fsm.h"
#include "pump_hw.h"

static char log_buf[64];

static void log_add(const char *s)
{
    strncat(log_buf, s, sizeof log_buf - strlen(log_buf) - 1u);
}

void motor_soft_start(void) { log_add("S"); }
void motor_full_speed(void) { log_add("F"); }
void motor_off(void)        { log_add("O"); }
void fault_trip(void)       { log_add("T"); }
void fault_clear(void)      { log_add("C"); }

/* The specification, written independently of the table in pump_fsm.c. */
static const pump_state_t expected_next[ST_COUNT][EV_COUNT] = {
    /*               START        STOP      TIMEOUT      OVERLOAD  RESET */
    [ST_OFF]      = { ST_STARTING, ST_OFF,   ST_OFF,      ST_OFF,   ST_OFF      },
    [ST_STARTING] = { ST_STARTING, ST_OFF,   ST_RUN,      ST_FAULT, ST_STARTING },
    [ST_RUN]      = { ST_RUN,      ST_OFF,   ST_RUN,      ST_FAULT, ST_RUN      },
    [ST_FAULT]    = { ST_FAULT,    ST_FAULT, ST_FAULT,    ST_FAULT, ST_OFF      },
};

static const char *const expected_log[ST_COUNT][EV_COUNT] = {
    /*               START STOP TIMEOUT OVERLOAD RESET */
    [ST_OFF]      = { "S", "",  "",     "",      ""  },
    [ST_STARTING] = { "",  "O", "F",    "T",     ""  },
    [ST_RUN]      = { "",  "O", "",     "T",     ""  },
    [ST_FAULT]    = { "",  "",  "",     "",      "C" },
};

static int failures;

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

/* Bring the FSM into state s through valid events, then clear the log. */
static void enter(pump_state_t s)
{
    pump_init();
    if (s == ST_STARTING || s == ST_RUN || s == ST_FAULT) pump_dispatch(EV_START);
    if (s == ST_RUN)   pump_dispatch(EV_TIMEOUT);
    if (s == ST_FAULT) pump_dispatch(EV_OVERLOAD);
    log_buf[0] = '\0';
}

int main(void)
{
    for (int s = 0; s < ST_COUNT; s++) {
        for (int e = 0; e < EV_COUNT; e++) {
            enter((pump_state_t)s);
            CHECK(pump_state() == (pump_state_t)s);
            pump_dispatch((pump_event_t)e);
            if (pump_state() != expected_next[s][e] || strcmp(log_buf, expected_log[s][e]) != 0) {
                printf("FAIL state %d event %d: next %d (want %d), actions \"%s\" (want \"%s\")\n",
                       s, e, (int)pump_state(), (int)expected_next[s][e],
                       log_buf, expected_log[s][e]);
                failures++;
            }
        }
    }

    /* a whole scenario: start, run, stop */
    enter(ST_OFF);
    pump_dispatch(EV_START);
    pump_dispatch(EV_TIMEOUT);
    pump_dispatch(EV_STOP);
    CHECK(pump_state() == ST_OFF && strcmp(log_buf, "SFO") == 0);

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

Тест собирают тем же компилятором, что и любую программу для компьютера, с предупреждениями, например gcc -std=c11 -Wall -Wextra pump_fsm.c test_pump.c -o test_pump, и запускают. Главное в нём — две матрицы ожидаемого поведения, записанные отдельно от таблицы в pump_fsm.c: для каждой из 4 × 5 = 20 пар «состояние + событие» — следующее состояние и последовательность действий. Цикл приводит автомат в каждое состояние, подаёт каждое событие и сравнивает оба результата с матрицами. Проверяется и то, что игнорируемые события не вызывают действий: для них ожидается пустой журнал.

Такой перебор ловит ошибку, которую трудно заметить глазами, — строку, скопированную с чужим состоянием. Если добавить в таблицу ошибочную строку { ST_RUN, EV_START, ST_STARTING, motor_soft_start }, тест сообщит: FAIL state 2 event 0: next 1 (want 2), actions "S" (want "") — в состоянии RUN событие START должно игнорироваться. Проверка только «игнорируемые события ничего не делают» эту ошибку бы пропустила: состояние меняется, и проверка не применяется. Сценарий в конце теста проверяет порядок действий на цепочке событий: журнал «SFO» означает «разгон, полная скорость, выключение» именно в этом порядке.

Так же проверяют кнопку и разборщик: функции получают время параметром, поэтому тест подаёт любые моменты, в том числе около 0xFFFFFF00, где счётчик вот-вот перейдёт через ноль, и за миллисекунды «проживает» минуты нажатий. Тесты запускают при каждой сборке, до прошивки платы; что попадает в образ для программатора, рассказано в статье о файлах прошивки.

Три способа рядом

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

• Ожидание внутри состояния. while (!button_pressed()) {} или delay() в ветке case превращают автомат обратно в блокирующий код. Каждое ожидание — это отдельное состояние и проверка на следующем проходе.
• Сравнение моментов времени now >= start + timeout вместо разности: раз в 49,7 суток таймауты срабатывают мгновенно. Для 16-битных счётчиков — ещё и разность без приведения к uint16_t.
• Неатомарное чтение тика на AVR: uint32_t читается четырьмя командами, и время изредка «прыгает» вперёд или назад на 256 мс и больше.
• Вызов автомата из прерывания и из главного цикла. Переход прерывается посередине, состояние портится. Обработчики кладут события в очередь.
• Рекурсивный вызов диспетчера из действия. Действие выполняется, пока автомат ещё в старом состоянии; новое событие нужно класть в очередь.
• Забытая инициализация. Статическая переменная без инициализатора равна нулю (C11, 6.7.9p10; на микроконтроллере её обнуляет код запуска), и если первое значение перечисления — не начальное состояние, автомат стартует не там. У функций-состояний хуже: указатель окажется нулевым, и первый же вызов p->state(p, byte) уведёт программу по нулевому адресу. Начальное состояние задают явно в функции инициализации — как parser_init() — и, если у начального состояния есть действие входа, выполняют его там же, как pump_init() вызывает motor_off().
• Неучтённые пары «состояние + событие». Событие тихо игнорируется там, где должно было что-то сделать, или наоборот. Помогают таблица на бумаге со всеми клетками и тест, перебирающий все пары.
• Скрытое состояние в переменных. Десяток флагов вида is_running, was_paused, need_retry рядом с переменной состояния — признак того, что часть состояний не выделена явно. Комбинации флагов, которых не должно быть, рано или поздно возникнут.
• default, который глушит предупреждение. С default ключ -Wswitch не сообщит о забытом новом состоянии; включайте -Wswitch-enum или обходитесь без default.
• Действия выхода и входа размазаны по переходам. Если «выключить ШИМ» написано в пяти переходах из RUN, шестой забудут. Общее для всех выходов из состояния — в одну функцию exit, которую вызывает диспетчер.