Мне давно было интересно попробовать запустить корутины на микроконтроллере, но стандарт С++ совершенно не учитывает мир embedded с его ограничениями и не даёт возможности создать корутину гарантированно не требующую аллокаций в куче.
Недавно я обнаружил что LLM агенты уже могут реализовывать довольно сложные задачи самостоятельно с неплохим качеством кода.
И тут я понял - чего бы не попробовать сделать своё расширение для clang с корутинами, которые гарантированно не используют аллокации и поверх этого реализовать небольшую операционную систему на корутинах для микроконтроллера.
Что такое корутина
Корутина - это функция, выполнение которой можно приостановить, а затем продолжить с того же места. При этом сохраняются её локальные переменные и информация о том, на каком шаге она остановилась.
Например, мигание светодиодом можно записать так:
task<void> blink() { while (true) { led.write(1); co_await delay(100); led.write(0); co_await delay(500); } }
Корутина включает светодиод, ждёт 100 мс, выключает его и ждёт ещё 500 мс. Во время каждого ожидания она отдаёт управление, и процессор может выполнять другие задачи или спать. После возобновления выполнение продолжается с того места, на котором остановилось.
Здесь task и delay предоставляет библиотека. Сам язык даёт механизм приостановки через co_await, а библиотека определяет, чего мы ждём и кто возобновит выполнение. Если результат ожидания уже готов, co_await может вообще не приостанавливать корутину.
Для embedded в этом интересно то, что последовательность действий остаётся записана в одной функции. Без корутин такой код часто приходится разбивать на состояния: включили периферию, запомнили текущий шаг, вышли из функции; при следующем вызове проверили событие и перешли к следующему шагу. Корутины C++ позволяют поручить это преобразование компилятору.
Если посмотреть на сгенерированный Clang код, вместо одного тела blink появятся отдельные функции создания, возобновления и уничтожения корутины. Между вызовами функции возобновления хранится кадр с её состоянием.
Логику реально получившейся функции возобновления можно переписать на C++ так. Служебные поля promise и awaiter здесь опущены:
struct blink_frame { void (*resume)(blink_frame*); void (*destroy)(blink_frame*); // Promise и данные объектов ожидания. unsigned char state = 0; // 0 - первое выполнение; // 1 - продолжение после delay(100); // 2 - продолжение после delay(500). }; void blink_resume(blink_frame* frame) { if (frame->state == 1) { led.write(0); frame->state = 2; register_timer(frame, 500); return; } // Первое выполнение или продолжение после delay(500). led.write(1); frame->state = 1; register_timer(frame, 100); return; }
register_timer здесь обозначает работу объекта ожидания delay: зарегистрировать задержку и сохранить handle, через который корутину позже возобновят. Компилятор не знает, как устроены наши часы и планировщик; он вызывает методы awaiter, а библиотечный код уже выполняет регистрацию.
При первом запуске state равен 0: светодиод включается, в кадре сохраняется состояние 1, регистрируется таймер на 100 мс, функция возвращает управление. Когда таймер срабатывает, вызывается та же blink_resume, но теперь она попадает в ветку выключения светодиода и ставит задержку на 500 мс. Следующее возобновление снова включает светодиод.
Исходного бесконечного цикла внутри одного вызова blink_resume уже нет. Цикл получается из последовательности возобновлений одного и того же кадра. В настоящей реализации также есть вызовы await_ready, await_suspend и await_resume; в этом примере ожидания всегда приостанавливают выполнение, а await_resume ничего не делает, поэтому после оптимизации остаётся именно такая логика.
Функция создания инициализирует кадр и возвращает handle. Функция уничтожения вызывает необходимые деструкторы объектов кадра. Если бы в blink были локальные переменные, значения которых нужны после co_await, они тоже находились бы в этом кадре.
При этом корутина сама по себе не создаёт поток. Несколько корутин могут по очереди выполняться в одном потоке, уступая управление в точках ожидания. У корутин C++20 нет отдельных машинных стеков: данные, которые нужны после приостановки, сохраняются в кадре корутины. Дальше как раз разберёмся, где этот кадр хранить на микроконтроллере.
В чём проблема стандартных корутин C++
Мы уже увидели, что корутина превращается в несколько функций и кадр с данными. Функции лежат во flash, а кадру нужна RAM. Казалось бы, осталось объявить под него обычную переменную и передать её при запуске blink.
Но в стандартном C++ так сделать напрямую нельзя. Компилятор создаёт тип кадра для своих нужд и не предоставляет его программе. Написанный выше blink_frame - иллюстрация: получить такой тип из blink и применить к нему sizeof в стандартном C++ невозможно.
sizeof(task<void>) тоже не поможет. В нашей библиотеке task хранит только handle - указатель на кадр. Размер такого объекта не говорит, сколько памяти занимает сама корутина. Две функции могут возвращать один и тот же task<void>, но иметь совершенно разные кадры: одна ждёт таймер, другая сохраняет между ожиданиями большой буфер.
По умолчанию память под кадр выделяется через operator new. Можно определить собственный operator new в promise_type; если подходящего объявления там нет, используется глобальная функция выделения памяти. Размер необходимого блока компилятор передаёт ей аргументом. Этот механизм описан в разделе стандарта о корутинах:https://eel.is/c++draft/dcl.fct.def.coroutine#9
То есть компилятор знает, сколько памяти нужно, но библиотека получает этот размер только при запуске корутины. А нам он нужен раньше - чтобы объявить статический объект нужного размера и увидеть его расход RAM ещё при сборке прошивки.
Собственный operator new позволяет обойтись без кучи. Например, можно возвращать память из статического пула или передать буфер аргументом корутины и использовать его в функции выделения. Но размер и выравнивание этого буфера всё равно придётся откуда-то взять. Заложить запас, проверить размер при запуске, посмотреть результат сборки - можно. Объявить хранилище точно под конкретную корутину средствами стандартного C++ - нельзя.
Есть ещё устранение аллокации: компилятор может встроить кадр в вызывающую функцию, если способен доказать, что это безопасно. В LLVM за такую оптимизацию отвечает CoroElide, описанный в документации о корутинах (https://llvm.org/docs/Coroutines.html#coroutine-allocation-elision). Это полезно, но не даёт никаких гарантий.
Для прошивки мне хотелось явно задать, где хранится кадр и сколько памяти он занимает. Объявить его рядом с остальными объектами, передать при запуске и получить ошибку компиляции, если такой запуск невозможен. Сам механизм корутин для этого подходит; в стандартном интерфейсе не хватает доступа к размеру кадра и способа создать его в предоставленном хранилище.
Что добавили в Clang
В Clang добавили атрибут [[coro::noalloc]] и три встроенные функции: две позволяют узнать размер и выравнивание кадра при компиляции, третья - создать корутину в заранее выделенном хранилище.
[[coro::noalloc]] coro::task<void> some_task(); __builtin_coro_frame_size(some_task); __builtin_coro_frame_align(some_task); __builtin_coro_invoke(some_task, storage, occupied, /* аргументы */);
Первые две дают размер и выравнивание кадра для выбранной целевой архитектуры. Третья создаёт корутину в переданном хранилище. Адрес флага occupied нужен, чтобы связать освобождение хранилища с фактическим уничтожением кадра.
Поверх этих builtin была реализована библиотека. Её тип frame оборачивает их примерно так - здесь опущены проверки типов и часть интерфейса:
template<auto F> class frame { alignas(__builtin_coro_frame_align(F)) unsigned char storage[__builtin_coro_frame_size(F)]{}; bool occupied = false; public: template<class... Args> auto operator()(Args&&... args) noexcept { if (occupied) __builtin_trap(); occupied = true; return __builtin_coro_invoke( F, static_cast<void*>(storage), &occupied, static_cast<Args&&>(args)...); } };
Таким образом, размер буфера и его выравнивание задаёт компилятор. Флаг занятости и padding тоже входят в sizeof(frame<F>).
Определение функции должно находиться до создания соответствующего frame в той же единице трансляции. По одному объявлению компилятор не узнает, какие переменные переживают точки приостановки и какой кадр получится после преобразования.
Вызывать помеченную функцию обычным способом нельзя:
auto task = some_task(); // Ошибка компиляции.
Также нельзя получить обычный вызываемый указатель на неё. Передать функцию как аргумент шаблона для frame<some_task> можно. Ограничения сделаны чтобы не существовало пути запуска без предоставленного хранилища.
Сам frame не копируемый и не перемещаемый, это нужно чтобы его адрес гарантированно был стабилен. В других местах программы могут быть указатели на объекты на фрейме корутины.
Почему узнать размер кадра оказалось непросто
Главная сложность здесь это момент, к которому нужно знать размер frame.
Чтобы разобрать массив внутри frame<F>, frontend уже должен знать его размер. Но окончательное устройство кадра появляется позже, когда coroutine passes преобразуют функцию: выделяют сохраняемое состояние, строят resume/destroy и раскладывают данные кадра.
В реализации используется отдельный сервис layout. Для уже определённой функции он запускает изолированный CodeGen, выполняет необходимые coroutine passes и получает размер и выравнивание из построенного LLVM-представления кадра. Получившийся преобразованный IR сохраняется, а затем импортируется в итоговый модуль.
Размер берётся из того же кадра, который затем используется в сгенерированном коде.
У текущей версии есть множество ограничений. Поддерживаются свободные нешаблонные функции; методы, coroutine templates и лямбды не поддерживаются. PCH/modules для расширения тоже не поддерживаются. Рекурсивная зависимость размеров кадров и динамический layout (alloca) запрещены.
Теперь поверх этого можно написать планировщик
Следующим шагом стал coro_os - небольшое переносимое кооперативное ядро с портом для Cortex-M.
Слово "операционная система" здесь не означает аналог полноценной RTOS со всеми её возможностями. В текущем ядре один активный планировщик, нет вытеснения, отдельных стеков задач и переключения аппаратного контекста. Готовые корутины исполняются из основного цикла.
Разделение объектов получилось следующим:
Объект | Для чего нужен |
| Память под конкретную корутину |
| Владение корутиной и получение её результата |
| Связь корневой корутины с очередями планировщика |
| Обслуживание готовых задач, таймеров и пробуждений |
Сам по себе task не знает, каким планировщиком его запускают. А job содержит служебный узел, через который ядро учитывает состояние и ожидания задачи.
Внутри планировщика три очереди. Ready - FIFO готовых задач. Timer - упорядоченный список относительных задержек. Pending - очередь событий, которые нужно обработать, прежде чем продолжить корутину.
Последняя особенно нужна для прерываний. ISR не вызывает resume, а отмечает уведомление, добавляет узел в pending, если его там ещё нет, и будит основной цикл. Когда управление доходит до poll, планировщик обрабатывает pending и переносит ожидающую задачу в ready.
static board_endpoint hardware_endpoint; extern "C" void Peripheral_IRQHandler() { // Сначала board-код подтверждает источник прерывания. hardware_endpoint.signal_from_isr(); }
Endpoint получается из job.notifier(). Его нужно сохранить до включения IRQ. Это невладеющая ссылка на задачу, поэтому перед уничтожением job источник прерывания нужно отключить.
Обычное уведомление бинарное: несколько сигналов до их потребления сливаются в один.
Другую корутину можно разбудить тем же endpoint, вызвав signal() из thread mode.
Для времени есть sleep_for(ticks) и wait_for(ticks). В первом случае уведомление не сокращает сон. Во втором ожидание возвращает notified или timeout.
yield() ставит задачу в конец ready. poll() исполняет очередь до опустошения.
Порт для Cortex-M
Архитектурный порт отвечает за critical sections, пробуждение процессора и ожидание работы. Его тип задаётся параметром шаблона scheduler<Port>, поэтому вызовы порта известны при компиляции. Пустой объект Cortex-M порта не добавляет памяти в scheduler благодаря [[no_unique_address]].
Для коротких critical sections Cortex-M порт использует PRIMASK с сохранением исходного значения. Пробуждение процессора выполняется через SEV. В idle используется последовательность:
SEV; WFE; if (!ready()) { WFE; }
Первый WFE потребляет event, установленный предшествующим SEV. После этого проверяется наличие работы. Если прерывание приходит между проверкой и вторым WFE, оно устанавливает event, и процессор не засыпает, потеряв пробуждение.
Ready-проверка должна учитывать и изменение часов после последнего образца времени. В примере SysTick увеличивает счётчик и вызывает SEV; после пробуждения основной цикл снова проверяет время и очереди.
Весь этот механизм находится в архитектурном порте. Ядру не нужно знать, что у Cortex-M есть event register или инструкция WFE. Для host-тестов используется отдельный тестовый порт.
Вложенные корутины: кадр внутри кадра
Мигать светодиодом полезно для первого теста, но хочется разбивать программу на функции. Например, ждать пакет в дочерней корутине и получить результат:
struct received_packet { std::uint32_t bytes; }; [[coro::noalloc]] coro::task<received_packet> receive_packet() { co_await board_scheduler::wait(); // В настоящем драйвере здесь читаются метаданные принятого пакета. co_return {1}; } [[coro::noalloc]] coro::task<void> parent() { coro::frame<receive_packet> child_frame; auto packet = co_await child_frame(); // Используем packet. }
child_frame записан как локальная переменная родителя. Поскольку он должен пережить приостановку, его хранилище попадает внутрь родительского кадра. Оно не остаётся на общем машинном стеке после возврата в планировщик.
Для всей цепочки нужен один job. Родитель ждёт ребёнка, а планировщик возобновляет текущую активную корутину цепочки. Уведомления и таймер относятся к этому общему job.
В promise хранится информация о корне, активном кадре и продолжении. Корневой resume обрабатывает переходы к ребёнку и обратно до внешнего ожидания.
Результат task<T> конструируется в promise при co_return. При завершении ожидания значение перемещается родителю, после чего дочерний кадр уничтожается. Тип может быть move-only и не иметь конструктора по умолчанию; перемещение и разрушение должны быть noexcept.
А если дети вызываются по очереди?
После вложенных корутин появился следующий вопрос. Если родитель последовательно вызывает несколько разных корутин, зачем хранить память под все их кадры одновременно?
Для этого добавили frame_slot:
[[coro::noalloc]] coro::task<void> process_packet(received_packet packet) { co_await board_scheduler::sleep_for(packet.bytes); } [[coro::noalloc]] coro::task<void> packet_loop() { coro::frame_slot<receive_packet, process_packet> children; for (;;) { auto packet = co_await children.invoke<receive_packet>(); co_await children.invoke<process_packet>(packet); } }
В слоте один буфер с максимальными размером и выравниванием кадров перечисленных функций. Одновременно в нём может жить только одна корутина. co_await уничтожает завершившегося ребёнка до возврата родителю, поэтому следующий вызов может сразу использовать тот же буфер.
Это явное переиспользование памяти. Компилятор не должен доказывать, что произвольные вызовы не пересекаются по времени: программист выражает это выбором одного слота. Попытка использовать занятый слот вызывает trap.
В измеренном примере на Cortex-M0 при -Os отдельные frame для receive и process занимают 48 и 40 байт. Общий слот - 48 байт. Весь родительский frame со слотом занимает 100 байт, с двумя отдельными frame - 140. В этих размерах уже учтены служебные данные и выравнивание.
Синхронизация и уничтожение задач
Кооперативность не отменяет необходимость синхронизации. Пока одна корутина ждёт завершения операции с периферией, другая вполне может попытаться обратиться к тому же ресурсу.
В библиотеке есть FIFO mutex и semaphore. Mutex нерекурсивный и возвращает RAII guard:
coro_os::mutex<board_port> resource; // Внутри корутины: auto guard = co_await resource.lock(); co_await board_scheduler::sleep_for(10); // При выходе из scope mutex освобождается.
Guard сохраняет владение через точки приостановки. Его можно освободить раньше через unlock().
У семафора есть ограничение ёмкости, ожидание с таймаутом и выдача разрешения из ISR:
coro_os::semaphore<board_port, 8> packets{0}; // Из IRQ: bool accepted = packets.release_from_isr(); // Из корутины: bool received = co_await packets.acquire_for(100); // Или без таймаута: co_await packets.acquire();
Нулевой таймаут проверяет доступность без ожидания. Release возвращает false, если достигнута ёмкость. ISR только назначает разрешение и ставит пробуждение в pending, вся операция занимает O(1).
Записи ожидания находятся в awaiter внутри кадра корутины. Отдельных аллокаций для очередей нет. Примитивы должны жить дольше использующих их задач, а mutex guard - оставаться в своей цепочке job.
Проверка на микроконтроллере
Для проверки я взял P-NUCLEO-WB55 с STM32WB55RG на Cortex-M4.
Чтобы было с чем сравнить, сделал три прошивки одного приложения: без операционной системы, на FreeRTOS и на coro_os.
Три светодиода независимо мигают с периодами 200, 500 и 1000 мс. Нажатие SW1 переключает скорость между обычной и удвоенной. Кнопка подключена через EXTI, дребезг подавляется, удержание не вызывает повторных переключений.
Startup, таблица векторов, настройки часов и GPIO у всех вариантов общие. Частота CPU - 16 МГц от HSI, SysTick - 1 кГц. Доступ к периферии через регистры и CMSIS, без HAL. Отличается способ выполнения приложения.
Без OS это обычный superloop: три срока следующего переключения LED и явная машина состояний кнопки. Основной цикл проверяет, что пора сделать, и засыпает до события.
В FreeRTOS - три задачи для светодиодов и одна для кнопки. Например, задача светодиода:
static void led_task(void* argument) { const auto led = reinterpret_cast<std::uintptr_t>(argument); for (;;) { vTaskDelay(nucleo::half_period(led, nucleo::fast())); nucleo::led_toggle(led); } }
В coro_os тот же код выглядит так:
[[coro::noalloc]] coro::task<void> blink(unsigned led) { for (;;) { co_await scheduler::sleep_for( nucleo::half_period(led, nucleo::fast())); nucleo::led_toggle(led); } }
По коду самой задачи разница невелика. Разница в том, какую память нужно зарезервировать для её выполнения.
В FreeRTOS все задачи создаются статически, через xTaskCreateStatic. Под четыре задачи приложения выделено по 256 слов стека, под idle - 128 слов.
У корутин явно выделены три frame<blink> и один кадр обработчика кнопки. Все четыре задачи выполняются на общем машинном стеке.
Обработчик EXTI в варианте FreeRTOS отправляет direct-to-task notification через vTaskNotifyGiveFromISR. В coro_os вызывает signal_from_isr у endpoint. Подавление дребезга выполняется уже в задаче, после пробуждения.
Полный код трёх вариантов находится в https://github.com/rus084/coro_os/tree/main/examples/nucleo_wb55
Сколько памяти получилось
Все три варианта собраны одним Clang 23.1.3 для Cortex-M4: -Os, без LTO, с удалением неиспользуемых секций. В размер Flash входят startup, таблица векторов, код приложения, служебные счётчики, .ARM.exidx и выравнивание. Размер взят из итогового BIN.
Вариант | Flash, байт | .data + .bss, байт |
Без OS | 1404 | 48 |
FreeRTOS | 4932 | 5224 |
coro_os | 3352 | 224 |
С RAM нужно разобраться подробнее. В 5224 байта FreeRTOS уже входят 4608 байт статически выделенных стеков: четыре по 1024 байта и idle на 512 байт. На самом деле можно выделить стека и поменьше, но нужно явно вычислять требуемый размер.
А 224 байта coro_os - только статические данные. Планировщик и четыре job находятся в main, поэтому в bss не попадают. Сгенерированный стековый кадр main занимает 184 байта, из которых 20 - scheduler и 160 - четыре job. База получается 224 + 184 = 408 байт, поверх которой нужен стек выполняющихся вызовов и IRQ.
В общем linker script для всех трёх вариантов задан одинаковый резерв общего стека - 2 КиБ. Если считать именно зарезервированную RAM, получаются такие числа:
Вариант | .data + .bss | Общий стек | Всего зарезервировано |
Без OS | 48 | 2048 | 2096 |
FreeRTOS | 5224 | 2048 | 7272 |
coro_os | 224 | 2048 | 2272 |
Получилось ожидаемо: superloop самый маленький. Корутины добавляют расходы на кадры и планирование, зато последовательность ожиданий остаётся записана обычным кодом. FreeRTOS тоже позволяет писать последовательный код, но каждой задаче нужен собственный стек, и она получает вытесняющее планирование. Для задач, которым достаточно кооперативной модели, у корутин есть возможность обойтись без этих стеков.
Сколько стоит переключение задач
Кроме памяти интересно было измерить, сколько тактов CPU занимает передача управления от одной задачи другой. Для этого я сделал две отдельные тестовые прошивки для той же платы: на FreeRTOS и на coro_os.
В каждой работают две задачи. Задача A отправляет уведомление задаче B и уходит в ожидание. B принимает уведомление, отвечает задаче A и тоже уходит в ожидание. Получается обмен A -> B -> A. В FreeRTOS используются xTaskNotifyGive и ulTaskNotifyTake, у корутин - signal и co_await scheduler::wait().
Время было измерено через аппаратный счётчик тактов DWT_CYCCNT. Число тактов за серию делится на 2048: в каждом обмене две передачи управления. Тест повторён три раза после перезапуска платы.
Вариант | Тактов на передачу управления | Время при 16 МГц |
FreeRTOS | 684,5 | 42,78 мкс |
coro_os | 411,0 | 25,69 мкс |
В этом тесте корутины тратят примерно на 40% меньше тактов. В число входят отправка уведомления, переход в ожидание, работа планировщика, само переключение, цикл теста и счётчик ответов. Отдельный вызов resume() или только обработчик PendSV здесь не измеряются.
Обе прошивки собраны с -Os, без LTO. У задач FreeRTOS одинаковый приоритет, поэтому отправка уведомления сама по себе не вытесняет отправителя: переключение происходит, когда он блокируется. SysTick на время замера отключён, PendSV работает; внешние прерывания не включены.
FreeRTOS при переключении сохраняет и восстанавливает машинный контекст задачи, включая указатель её стека. Корутина возвращается в планировщик обычным вызовом, а затем планировщик вызывает функцию возобновления другой корутины. Но и этот путь требует работы: нужно обработать уведомление, поменять состояние задачи и очередь готовых. Поэтому расходы у корутин тоже есть.
Что в этой работе делали агенты
Еще интерес был ещё и в том, смогут ли агенты справиться с такой задачей. Основную работу по реализации расширения и библиотеки я поручал им: разбор исходников, изменения, тесты, сборку и исправление найденных ошибок.
Это не выглядело как один запрос «напиши мне компилятор и ОС на корутинах». Работа шла итерациями. Я задавал требования, смотрел на API и размеры объектов, спрашивал, зачем нужно то или иное поле, и просил менять архитектурные решения, которые мне не нравились.
Например, сначала архитектурный порт был структурой с указателями на функции. Потом стал параметром шаблона.
Отдельное ревью оказалось полезным. Сам факт, что код написал агент и другой агент его проревьювил, еще не значит что код production-ready. Читать код, принимать архитектурные решения и проверять результат всё равно нужно.
Эксперимент выглядит удачным: идея, ради которой пришлось бы надолго отложить остальные дела и самому погружаться в Clang, дошла до рабочего решения.
Исходники всего доступны на гитхабе:
- Расширение Clang (ветка noalloc-coroutines) https://github.com/rus084/llvm-project/tree/noalloc-coroutines
- Библиотека coro_os, тесты и примеры для платы https://github.com/rus084/coro_os
