Переривання та main(): як безпечно ділити дані
Обробник переривання та головний цикл - це дві програми, що працюють в одній пам'яті й нічого не знають одна про одну. Компілятор C не знає, що переривання існує, а CPU не знає, що дві змінні пов'язані між собою. Знає лише програміст, а помилки, які з цього випливають, найгірші з можливих: вони трапляються раз на тисячу запусків, зникають, коли підключаєш налагоджувач, і ніколи не з'являються на code review, бо код виглядає правильно.
Ця стаття розбирає три проблеми спільних даних, одну за одною, на коді, який не працює, а потім показує шаблони, які працюють. Експерименти виконувалися на Cortex-M4 (у QEMU, на моделі плати STM32F4) або на PC, а виводи - справжні.
Проблема 1: компілятор не бачить переривання
Найпростіша комунікація між перериванням і головним циклом - прапорець. Переривання його встановлює, головний цикл на нього чекає:
#ifdef USE_VOLATILE
static volatile bool dataReady;
#else
static bool dataReady;
#endif
void SysTick_Handler(void)
{
dataReady = true;
}
int main(void)
{
SysTick_Start(1000u);
while (!dataReady)
{
/* waiting for the interrupt */
}
Print("the interrupt was seen\n");
Quit();
return 0;
}
Це вся програма, і з оптимізацією -O2 вона ніколи не завершується. Я запустив її в QEMU з SysTick як джерелом переривання (переривання спрацьовує й встановлює прапорець) і порівняв версію з volatile та без нього:
--- without volatile
(no output, the program was killed by the timeout after 5 s)
--- with volatile
the interrupt was seen
Чому? Подивіться, що компілятор згенерував для циклу в першій версії:
8000062: ldrb r3, [r1, #0] @ read the flag once
8000064: cbnz r3, 8000068 @ non-zero: leave the loop
8000066: b.n 8000066 @ zero: jump to itself, forever
Компілятор міркує лише про код main(), а в цьому коді в dataReady ніхто не пише. Отже, значення не може змінитися, його достатньо прочитати один раз, і цикл стає нескінченним. З його точки зору - саме те, що ви просили. З volatile компілятор мусить читати змінну з пам'яті на кожному проході:
8000062: ldrb r3, [r2, #0] @ read the flag in every pass
8000064: cmp r3, #0
8000066: beq.n 8000062
Урок: кожна змінна, яку діляться переривання й основний код, має бути volatile. Те саме стосується кожного регістра периферії, і саме тому визначення регістрів у заголовках CMSIS є volatile. Пам'ятайте, що без цього в debug-збірці (-O0) усе працює, а в release-збірці ламається, тому помилку знаходять пізно.
Проблема 2: volatile не робить операцію атомарною
volatile каже «читай і пиши пам'ять щоразу». Воно не каже «усе одразу». Невинний рядок counter++ - це три інструкції:
Increment_Plain:
ldr r2, [pc, #8] @ address of the counter
ldr r3, [r2, #0] @ 1. load the value
adds r3, #1 @ 2. add one
str r3, [r2, #0] @ 3. store it back
Якщо переривання надходить між завантаженням і записом, і переривання також інкрементує лічильник, то один із двох інкрементів губиться: основний код записує старе значення плюс один поверх значення, яке щойно записало переривання. Часова послідовність:
| Крок | Основний код | Переривання | Лічильник |
|---|---|---|---|
| 1 | завантажує 5 | 5 | |
| 2 | завантажує 5, додає 1, записує 6 | 6 | |
| 3 | додає 1 (у регістрі має 5) | 6 | |
| 4 | записує 6 | 6, мало бути 7 |
Я хотів показати це і на Cortex-M4, але не вийшло: у 200 000 інкрементів із перериванням кожні кілька інструкцій QEMU нічого не втратив. Імовірно, тому що QEMU перевіряє переривання між блоками транслюваного коду, тож жодне ніколи не потрапляє між ldr і str. Це корисне нагадування, що симулятор не доводить відсутності race condition: справжній MCU може перервати після кожної інструкції. Той самий race у місці, де його легко відтворити, - це два потоки на PC:
#include <pthread.h>
#include <stdint.h>
#include <stdio.h>
#define INCREMENTS ( 2000000u )
static volatile uint32_t plainCounter;
static _Atomic uint32_t atomicCounter;
static void *Worker(void *argument)
{
(void)argument;
for (uint32_t index = 0u; index < INCREMENTS; index++)
{
plainCounter++; /* load, add, store: not atomic */
atomic_fetch_add(&atomicCounter, 1u); /* one indivisible operation */
}
return NULL;
}
int main(void)
{
pthread_t first, second;
pthread_create(&first, NULL, Worker, NULL);
pthread_create(&second, NULL, Worker, NULL);
pthread_join(first, NULL);
pthread_join(second, NULL);
printf("expected %u\n", 2u * INCREMENTS);
printf("plain counter %u (lost %u)\n", plainCounter, 2u * INCREMENTS - plainCounter);
printf("atomic counter %u (lost %u)\n", atomicCounter, 2u * INCREMENTS - atomicCounter);
return 0;
}
expected 4000000
plain counter 2652421 (lost 1347579)
atomic counter 4000000 (lost 0)
Третина всіх інкрементів губиться, а результати відрізняються від запуску до запуску. Потік на PC і переривання на MCU в цьому сенсі - одне й те саме: другий потік керування, який може вдарити між двома інструкціями. Те саме стосується всього, що більше за слово CPU: 64-бітова часова мітка або структура з двох полів може бути прочитана наполовину старою, наполовину новою (torn read).
Три способи розв'язання
1. Один записувач на кожну змінну (найкращий)
Найнадійніше рішення - спроєктувати дані так, щоб кожна змінна записувалася лише однією стороною. Переривання записує прапорець, основний код лише читає його й не скидає (або навпаки). Без двох записувачів нема чого губити. Якщо обом сторонам потрібно змінювати те саме, вони не повинні це ділити: у кожної своя змінна (лічильник переривань - isrCalls, лічильник основного коду - інший), а суму рахує читач.
2. Критична секція
Коли двох записувачів уникнути не можна, переривання вимикаються на час операції. Правильний спосіб - зберегти стан і відновити його, а не вмикати переривання наприкінці наосліп, бо функцію могли викликати з уже вимкненими перериваннями:
Increment_CriticalSection:
mrs r1, PRIMASK @ save the state of the mask
cpsid i @ disable the interrupts
ldr r3, [r2, #0]
adds r3, #1
str r3, [r2, #0]
msr PRIMASK, r1 @ restore the state, as it was
Ціна - затримка (latency): під час секції жодне переривання не може виконатися. Тому секція має бути короткою, як кілька інструкцій, і без виклику функції, циклу чи очікування всередині.
3. Атомарна операція
Для одного слова (лічильник, прапорець із лічильником) процесор має інструкції LDREX та STREX: запис вдається лише якщо ніхто не торкався адреси після завантаження, а якщо торкався - цикл повторюється. Компілятор генерує їх за вас зі стандартної операції:
Increment_Atomic:
ldrex r1, [r3] @ load and mark the address
adds r1, #1
strex r2, r1, [r3] @ store only if nobody has touched it, r2 = 0 on success
cmp r2, #0
bne.n Increment_Atomic @ somebody did: try again
у C це один рядок, __atomic_fetch_add(&counter, 1u, __ATOMIC_RELAXED) (або atomic_fetch_add() з <stdatomic.h>). Він не вимикає переривання, тож не додає затримки. Він працює для однієї змінної; для двох змінних, що належать одна одній, не допомагає.
Найуживаніший шаблон: кільцевий буфер
Найпоширеніше, що переривання передає основному коду, - потік даних: байти, прийняті UART, відліки АЦП. Переривання не може їх обробляти (воно має бути коротким), тому кладе їх у чергу, а основний код забирає їх у спокійний момент. Це задача виробник-споживач з одним виробником (переривання) та одним споживачем (основний код), і саме для цього випадку кільцевий буфер не потребує блокування:
#ifndef RING_H
#define RING_H
#include <stdatomic.h>
#include <stdbool.h>
#include <stdint.h>
#define RING_SIZE ( 16u ) /* has to be a power of two */
#define RING_MASK ( RING_SIZE - 1u )
/*
* Single producer, single consumer ring buffer without any lock.
* head: written only by the producer, tail: written only by the consumer.
* Both are free-running counters, the difference is the number of items.
*/
typedef struct
{
uint32_t items[RING_SIZE];
atomic_uint_fast32_t head;
atomic_uint_fast32_t tail;
} ring_t;
static inline bool Ring_Push(ring_t *ring, uint32_t item)
{
const uint_fast32_t head = atomic_load_explicit(&ring->head, memory_order_relaxed);
const uint_fast32_t tail = atomic_load_explicit(&ring->tail, memory_order_acquire);
if (RING_SIZE == (head - tail))
{
return false; /* full, the producer decides what to do */
}
ring->items[head & RING_MASK] = item;
atomic_store_explicit(&ring->head, head + 1u, memory_order_release); /* publish the item */
return true;
}
static inline bool Ring_Pop(ring_t *ring, uint32_t *item)
{
const uint_fast32_t tail = atomic_load_explicit(&ring->tail, memory_order_relaxed);
const uint_fast32_t head = atomic_load_explicit(&ring->head, memory_order_acquire);
if (head == tail)
{
return false; /* empty */
}
*item = ring->items[tail & RING_MASK];
atomic_store_explicit(&ring->tail, tail + 1u, memory_order_release); /* free the slot */
return true;
}
#endif
Чому це правильно?
headзаписує лише виробник, аtail- лише споживач. Немає змінної з двома записувачами, тож жодне оновлення не може загубитися: це перше рішення вище, вбудоване в структуру.- Лічильники вільно біжать (free-running) і ніколи не скидаються. Кількість елементів - це
head - tail, і завдяки беззнаковій арифметиці вона правильна навіть після переповнення лічильників. Якщо розмір - степінь двійки, позиція отримується простою маскою замість ділення. - Порядок має значення: виробник записує елемент і після цього публікує його зсувом
head(memory_order_release), а споживач спершу читаєhead(memory_order_acquire) і лише потім елемент. На одному ядрі Cortex-M важливий лише порядок, у якому компілятор записує, і атомарні операції його зберігають. На MCU з двома ядрами (деякі STM32H7 мають два) або з кешем той самий код усе ще правильний, і це причина писати його в такій формі, а не зvolatile, який «якось працює». - Виробник не чекає. Повний буфер означає «відкинь елемент і порахуй це», бо переривання не може ні на кого чекати. Лічильник відкинутих у тесті каже вам, що буфер замалий.
Я перевірив це двома способами. На PC з двома потоками, виробником і споживачем, пропускаємо 20 мільйонів елементів через буфер на 16 слотів:
items 20000000, order errors 0
і на Cortex-M4 у QEMU: переривання SysTick - виробник, а main() - споживач, 5000 елементів:
items received 5000
order errors 0
full buffer hits 0
Перший тест жорсткіший (два потоки справді виконуються паралельно на двох ядрах), другий - справжня ситуація MCU. Це також ідеальний кандидат для модульного тесту на PC, як у статті про модульне тестування з Unity та CMock: модуль не залежить від hardware.
Правила для обробників переривань
- Короткі. Прочитати регістр, зберегти дані, встановити прапорець, скинути pending-прапорець, повернутися. Обробка виконується в основному коді.
- Без очікування, без виділення пам'яті, без
printf. Нічого, що блокує, триває довго або викликає функцію, небезпечну для виклику з переривання (вона нереентерабельна, якщо всередині використовує статичний буфер). - Усе спільне -
volatileабо атомарне, і в кожної змінної визначений записувач. - Не діліть більше, ніж потрібно. Прапорець або черга між сторонами, а не структура з десяти полів.
- Думайте про пріоритет. Переривання вищого пріоритету може перервати інше переривання. Змінна, спільна для двох переривань, має ту саму проблему, що й спільна з основним кодом.
У Embedbits BSP модулі периферії (наприклад, USART) мають варіанти обробки даних polling, interrupt і DMA, і варіант з перериваннями - саме це: обробник на рівні MCAL, який бере дані з периферії та передає їх застосунку через буфер, тоді як Task модуля працює в головному циклі й виконує обробку.
Підсумок
| Проблема | Симптом | Лікування |
|---|---|---|
| Компілятор не бачить переривання | Працює з -O0, нескінченний цикл з -O2 | volatile на кожній спільній змінній |
counter++ не атомарний | Лічильник, який іноді замалий | Один записувач на змінну, критична секція або atomic_fetch_add |
| Більше за слово | Наполовину старе, наполовину нове значення | Критична секція, або порядковий номер, або черга |
| Потік даних | Втрачені або змішані байти | Кільцевий буфер з одним виробником і одним споживачем |
| Тест проходить у симуляторі | Помилка в полі | Пам'ятайте, що симулятор перериває в інших місцях, ніж MCU |
Три види коду в цій статті мали одну спільну рису: усі вони виглядали правильно.