Статическое распределение объектов FreeRTOS

По умолчанию все объекты в системе FreeRTOS распределяются динамически — очереди, семафоры, таймеры, задачи (потоки), и мьютексы. Программист видит только «кучу» — область где динамически выделяется память по запросу программы или системы, а что там творится внутри – не ясно. Сколько еще осталось? Неизвестно. Не занимает ли что нибудь больше чем нужно? Кто его знает? Лично я предпочитаю решать вопросы организации памяти еще на этапе написания прошивки, не доводя до ошибок во время выполнения, когда память неожиданно закончилась.

Эта статья является логическим продолжением вчерашней про статическое распределение объектов в памяти микроконтроллера, только теперь применительно к объектам FreeRTOS. Сегодня мы научимся размещать объекты FreeRTOS статически, что позволит более четко понимать что происходит в оперативной памяти микроконтроллера, как именно расположены и сколько занимают наши объекты.

Но просто взять и начать размещать объекты FreeRTOS статически много ума не требуется — FreeRTOS начиная с версии 9.0 как раз предоставляет функции создания объектов размещенных статически. Такие функции имеют суффикс Static в названии и на эти функции имеется отличная документация с примерами. Мы же напишем удобные и красивые C++ обертки над функциями FreeRTOS, которые не только будут размещать объекты статически, но и скрывать все потроха, а также предоставлять более удобный интерфейс.

Статья рассчитана на начинающих программистов, но которые уже знакомы с основами FreeRTOS и примитивами синхронизации многопоточный программ. Поехали.

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

На мой взгляд, любая прошивка где нужно одновременно делать две (и более) задачи решается намного проще и элегантнее, если использовать FreeRTOS. Например, считывать показания с медленных датчиков и одновременно обслуживать дисплей. Только чтобы без тормозов, пока считываются датчики. В общем must have! Всячески рекомендую к изучению.

Как я уже сказал и писал в прошлой статье, мне не очень нравится подход с созданием объектов динамически в случае если мы еще на этапе компиляции знаем их количество и размер. Если такие объекты размещать статически, то мы можем получить более четкую и понятную картину распределения памяти в микроконтроллере, а значит и избежать сюрпризов когда память внезапно закончилась.

Вопросы организации памяти FreeRTOS будем рассматривать на примере платы BluePill на микроконтроллере STM32F103C8T6. Чтобы не париться с компилятором и системой сборки работать будем в среде ArduinoIDE, установив поддержку для этой платы. Есть несколько реализаций Arduino под STM32 — в принципе подойдет любая. У меня установлена stm32duino согласно инструкции из Readme.md проекта, бутлоадер как сказано в этой статье. FreeRTOS версии 10.0 установлена через менеджер библиотек ArduinoIDE. Компилятор — gcc 8.2

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

• 2 задачи (потока) работают параллельно
• также работает таймер, который время от времени посылает нотификацию первой задаче используя семафор в режиме signal-wait
• первая задача, получив нотификацию от таймера, посылает сообщение (случайное число) второй задаче через очередь
• вторая, получив сообщение, печатает его в консоль
• пускай первая задача тоже что нибудь печатает в консоль, а чтобы они не подрались консоль будет защищена мьютексом.
• размер очереди можно было бы ограничить одним элементом, но для того, чтобы было интереснее поставим 1000

Стандартная реализация (согласно документации и туториалам) может выглядеть так.

#include <STM32FreeRTOS.h>
 
TimerHandle_t xTimer;
xSemaphoreHandle xSemaphore;
xSemaphoreHandle xMutex;
xQueueHandle xQueue;
 
void vTimerCallback(TimerHandle_t pxTimer)
{
  xSemaphoreGive(xSemaphore);
}
 
void vTask1(void *)
{
  while(1)
  {
	xSemaphoreTake(xSemaphore, portMAX_DELAY);
	int value = random(1000);
	xQueueSend(xQueue, &value, portMAX_DELAY);
 
	xSemaphoreTake(xMutex, portMAX_DELAY);
	Serial.println("Test");
	xSemaphoreGive(xMutex);
  }
}
 
void vTask2(void *)
{
  while(1)
  {
	int value;
	xQueueReceive(xQueue, &value, portMAX_DELAY);
 
	xSemaphoreTake(xMutex, portMAX_DELAY);
	Serial.println(value);
	xSemaphoreGive(xMutex);
  }
}
 
void setup()
{
  Serial.begin(9600);
 
  vSemaphoreCreateBinary(xSemaphore);
  xQueue = xQueueCreate(1000, sizeof(int));
  xMutex = xSemaphoreCreateMutex();
 
  xTimer = xTimerCreate("Timer", 1000, pdTRUE, NULL, vTimerCallback);
  xTimerStart(xTimer, 0);
 
  xTaskCreate(vTask1, "Task 1", configMINIMAL_STACK_SIZE, NULL, tskIDLE_PRIORITY, NULL);
  xTaskCreate(vTask2, "Task 2", configMINIMAL_STACK_SIZE, NULL, tskIDLE_PRIORITY, NULL);
  vTaskStartScheduler();
}
 
void loop() {}

Давайте посмотрим что творится в памяти микроконтроллера, если скомпилировать такой код. По умолчанию все объекты FreeRTOS размещаются в динамической памяти. FreeRTOS предоставляет аж целых 5 реализаций менеджеров памяти, которые отличаются сложностью реализации, но в целом задача у них одна и та же – нарезать кусочки памяти для нужд FreeRTOS и пользователя. Кусочки нарезаются либо из общей кучи микроконтроллера (с помощью malloc) или используют свою отдельную кучу. Какая именно куча используется для нас не важно – все равно внутрь кучи заглянуть мы не сможем.

Например, для кучи имени FreeRTOS это будет выглядеть так (вывод утилиты objdump)

...
200009dc l 	O .bss      	00002000 ucHeap
...

Т.е. видим один большой кусок, внутри которого нарезаются все объекты FreeRTOS – семафоры, мьютексы, таймеры, очереди, и даже сами задачи. Последние 2 пункта очень важны. В зависимости от количества элементов очередь может быть достаточно большой, а задачи гарантировано будут занимать много места из-за стека, который также выделяется вместе с задачей.

Да, это минус многозадачности – у каждой задачи будет свой стек. Причем стек должен быть достаточно большой, чтобы в нем поместились не только вызовы и локальные переменные самой задачи, но и стек прерывания, если такое возникнет. Ну а поскольку прерывание может случиться в любой момент, то каждая задача должна иметь резерв по стеку на случай прерывания. Более того, микроконтроллеры CortexM могут иметь вложенные прерывания, потому стек должен быть достаточно большой, чтобы вместить все прерывания, если они произойдут одновременно.

Размер стека задачи задается при создании задачи параметром функции xTaskCreate. Размер стека не может быть меньше параметра configMINIMAL_STACK_SIZE (задается в конфигурационном файле FreeRTOSConfig.h) – это тот самый резерв для прерываний. Размер кучи задается параметром configTOTAL_HEAP_SIZE и в данном случае равен 8кб.


Стоп, а что за переменные xTimer, xQueue, xSemaphore, и xMutex? Разве они не описывают нужные нам объекты? Нет, это только хендлы – указатели на некую (непрозрачную) структуру, которая и описывает сами объекты синхронизации

200009cc g 	O .bss     	00000004 xTimer
200009d0 g     O .bss    	00000004 xSemaphore
200009cc g 	O .bss     	00000004 xQueue
200009d4 g 	O .bss    	00000004 xMutex

Как я уже упоминал, предлагаю чинить весь этот беспорядок тем же самым способом, что и в предыдущей статье — распределим все наши объекты статически на этапе компиляции. Функции статического распределения становятся доступны если в файле конфигурации FreeRTOS параметр configSUPPORT_STATIC_ALLOCATION установлен в 1.

Начнем с очередей. Вот как предлагает аллоцировать очереди документация на FreeRTOS

struct AMessage
 {
            	char ucMessageID;
            	char ucData[ 20 ];
 };
 
 #define QUEUE_LENGTH 10
 #define ITEM_SIZE sizeof( uint32_t )
 
 // xQueueBuffer will hold the queue structure.
 StaticQueue_t xQueueBuffer;
 
 // ucQueueStorage will hold the items posted to the queue.  Must be at least
 // [(queue length) * ( queue item size)] bytes long.
 uint8_t ucQueueStorage[ QUEUE_LENGTH * ITEM_SIZE ];
 
 void vATask( void *pvParameters )
 {
 QueueHandle_t xQueue1;
 
            	// Create a queue capable of containing 10 uint32_t values.
            	xQueue1 = xQueueCreate( QUEUE_LENGTH, // The number of items the queue can hold.
                                                                                                            	ITEM_SIZE       	  // The size of each item in the queue
                                                                                                            	&( ucQueueStorage[ 0 ] ), // The buffer that will hold the items in the queue.
                                                                                                            	&xQueueBuffer ); // The buffer that will hold the queue structure.
 
            	// The queue is guaranteed to be created successfully as no dynamic memory
            	// allocation is used.  Therefore xQueue1 is now a handle to a valid queue.
 
            	// ... Rest of task code.
 }

В этом примере очередь описывается тремя переменными:

• Массив ucQueueStorage — это место в котором будут размещаться элементы очереди. Размер очереди задается пользователем для каждой очереди индивидуально.
• Структура xQueueBuffer – тут живет описание и состояние очереди, текущий размер, списки ожидающих задач, а также другие атрибуты и поля, нужные FreeRTOS для работы с очередью. Название для переменной, на мой взгляд, не совсем удачное, в самой FreeRTOS эта штука называется QueueDefinition (описание очереди).
• Переменная xQueue1 – это идентификатор очереди (handle). Все функции управления очередью, а также некоторые другие (например, внутренние функции работы с таймерами, семафорами и мьютексами) принимают вот такой хендл. На деле это просто указатель на QueueDefinition, но мы этого (как бы) не знаем, а потому хендл придется тягать за собой отдельно.

Сделать как в примере, конечно же, не составит проблем. Но лично мне не нравится иметь целых 3 переменные на одну сущность. Класс который сможет это инкапсулировать тут аж просится. Одна лишь проблема – размер каждой очереди может отличаться. В одном месте нужна очередь побольше, в другом достаточно пары элементов. Поскольку мы хотим разместить очередь статически мы должны как-то указать этот размер на этапе компиляции. Для этого можно воспользоваться шаблоном.

template<class T, size_t size>
class Queue
{
  QueueHandle_t xHandle;
  StaticQueue_t x QueueDefinition;
  T         	xStorage[size];
 
public:
  Queue()
  {
	xHandle = xQueueCreateStatic(size,
                 	sizeof(T),
                     reinterpret_cast<uint8_t*>(xStorage),
                 	&xQueueDefinition);
  }
 
  bool receive(T * val, TickType_t xTicksToWait = portMAX_DELAY)
  {
	return xQueueReceive(xHandle, val, xTicksToWait);
  }
 
  bool send(const T & val, TickType_t xTicksToWait = portMAX_DELAY)
  {
	return xQueueSend(xHandle, &val, xTicksToWait);
  }
};

Заодно в этот класс поселились также функции отправки и приема сообщений, причем сразу удобного нам типа.

Очередь будет объявляться как глобальная переменная, как-то так

Queue<int, 1000> xQueue;

Отправка сообщения

	xQueue.send(value);

Прием сообщения

	int value;
	xQueue.receive(&value);

Разберемся теперь с семафорами. И хотя технически (внутри FreeRTOS) семафоры и мутексы реализованы через очереди, семантически это 3 разных примитива. А потому будем реализовывать их отдельными классами.

Реализация класса семафора будет достаточно тривиальна – она просто хранит несколько переменных и объявляет несколько функций.

class Sema
{
  SemaphoreHandle_t xSema;
  StaticSemaphore_t xSemaControlBlock;
 
public:
  Sema()
  {
	xSema = xSemaphoreCreateBinaryStatic(&xSemaControlBlock);
  }
 
  BaseType_t give()
  {
	return xSemaphoreGive(xSema);
  }
 
  BaseType_t take(TickType_t xTicksToWait = portMAX_DELAY)
  {
	return xSemaphoreTake(xSema, xTicksToWait);
  }
};

Объявление семафора

Sema xSema;

Захват семафора

  xSema.take();

Отпускание семафора

  xSema.give();

Теперь мьютекс

class Mutex
{
  SemaphoreHandle_t xMutex;
  StaticSemaphore_t xMutexControlBlock;
 
public:
  Mutex()
  {
	xMutex = xSemaphoreCreateMutexStatic(&xSemaControlBlock);
  }
 
  BaseType_t lock(TickType_t xTicksToWait = portMAX_DELAY)
  {
	return xSemaphoreTake(xMutex, xTicksToWait);
  }
 
  BaseType_t unlock()
  {
	return xSemaphoreGive(xMutex);
  } 
};

Как вы можете заметить класс мьютекса практически идентичен классу семафора. Но как я уже сказал семантически это разные сущности. Более того, интерфейсы этих классов не полные, и расширяться они будут совсем в разные стороны. Так, у семафора могут добавится методы giveFromISR() и takeFromISR() для работы с семафором в прерывании, тогда как у мьютекса разве что метод tryLock() добавится — семантически у него нет других операций.


Мьютексом можно пользоваться в стиле С, т.е. напрямую вызывать lock()/unlock(). Но раз уж мы пишем на C++, то можно воспользоваться прелестями RAII и написать более удобную обертку, которая сама будет захватывать и отпускать мьютекс.

class MutexLocker
{
  Mutex & mtx;
 
public:
  MutexLocker(Mutex & mutex)
	: mtx(mutex)
  {
	mtx.lock();
  }
 
  ~MutexLocker()
  {
	mtx.unlock();
  }
};

При выходе из области видимости мьютекс будет автоматически освобожден.

Это особенно удобно, если выходов из функции несколько и не нужно постоянно помнить о необходимости освобождения ресурсов.

	MutexLocker lock(xMutex);
	Serial.println(value);
  } // mutex will be unlocked here

Теперь очередь таймеров.

class Timer
{
  TimerHandle_t xTimer;
  StaticTimer_t xTimerControlBlock;
 
public:
  Timer(const char * const pcTimerName,
    	const TickType_t xTimerPeriodInTicks,
    	const UBaseType_t uxAutoReload,
    	void * const pvTimerID,
    	TimerCallbackFunction_t pxCallbackFunction)
	{
  	xTimer = xTimerCreateStatic(pcTimerName, xTimerPeriodInTicks, uxAutoReload, pvTimerID, pxCallbackFunction, &xTimerControlBlock);
	}
 
	void start(TickType_t xTicksToWait = 0)
	{
  	xTimerStart(xTimer, xTicksToWait);
	}
};

В целом тут все аналогично предыдущим классам, не буду подробно останавливаться. Возможно, API оставляет желать лучшего, ну или как минимум требует расширения. Но моя цель показать принцип, а не доводить до состояния production ready.

Ну и, наконец, задачи. У каждой задачи есть стек и его нужно заранее разместить в памяти. Воспользуемся тем же приемом, что и с очередями – напишем шаблонный класс

template<const uint32_t ulStackDepth>
class Task
{
protected:
  StaticTask_t xTaskControlBlock;
  StackType_t xStack[ ulStackDepth ];
  TaskHandle_t xTask;
 
public:
  Task(TaskFunction_t pxTaskCode,
   	const char * const pcName,
   	void * const pvParameters,
   	UBaseType_t uxPriority)
  {
	xTask = xTaskCreateStatic(pxTaskCode, pcName, ulStackDepth, pvParameters, uxPriority, xStack, &xTaskControlBlock);
  }
};

Поскольку объекты задач теперь объявляются как глобальные переменные, то и инициализироваться они будут как глобальные переменные – до вызова main(). А значит и параметры, которые передаются в задачи также должны быть известны на этом этапе. Этот нюанс нужно учитывать, если в Вашем случае передается нечто, что нужно вычислять перед созданием задачи (у меня же там просто NULL). Если Вам это все равно не подходит — рассмотрите вариант с локальными статическими переменными из прошлой статьи.

Компилируем и получаем ошибку:

tasks.c:(.text.vTaskStartScheduler+0x10): undefined reference to `vApplicationGetIdleTaskMemory'
timers.c:(.text.xTimerCreateTimerTask+0x1a): undefined reference to `vApplicationGetTimerTaskMemory'

Дело вот в чем. В каждой ОС есть специальная задача – Idle Task (задача по умолчанию, задача ничего не делания). Операционная система исполняет эту задачу если все другие задачи выполняться не могут (например спят, или чего-то ждут). В целом, это самая обычная задача, только с самым низким приоритетом. Но вот создается она внутри ядра FreeRTOS и влиять на ее создание мы не можем. Но раз уж мы начали размещать задачи статически, то нужно как-то сказать ОС где нужно разместить управляющий блок и стек этой задачи. Вот для этого FreeRTOS и просит нас определить специальную функцию vApplicationGetIdleTaskMemory().

Аналогичная ситуация и с задачей таймеров. Таймеры в системе FreeRTOS живут не сами по себе – в ОС крутится специальная задача, которая и обслуживает эти таймеры. И эта задача также требует управляющий блок и стек. И точно также ОС просит нас указать где они находятся с помощью функции vApplicationGetTimerTaskMemory().

Сами функции тривиальны и просто возвращают соответствующие указатели на статически размещенные объекты.

extern "C" void vApplicationGetIdleTaskMemory( StaticTask_t **ppxIdleTaskTCBBuffer, StackType_t **ppxIdleTaskStackBuffer, uint32_t *pulIdleTaskStackSize)
{
  static StaticTask_t Idle_TCB;
  static StackType_t  Idle_Stack[configMINIMAL_STACK_SIZE];
 
  *ppxIdleTaskTCBBuffer = &Idle_TCB;
  *ppxIdleTaskStackBuffer = Idle_Stack;
  *pulIdleTaskStackSize = configMINIMAL_STACK_SIZE;
}
 
extern "C" void vApplicationGetTimerTaskMemory (StaticTask_t **ppxTimerTaskTCBBuffer, StackType_t **ppxTimerTaskStackBuffer, uint32_t *pulTimerTaskStackSize)
{
  static StaticTask_t Timer_TCB;
  static StackType_t  Timer_Stack[configTIMER_TASK_STACK_DEPTH];
 
  *ppxTimerTaskTCBBuffer   = &Timer_TCB;
  *ppxTimerTaskStackBuffer = Timer_Stack;
  *pulTimerTaskStackSize   = configTIMER_TASK_STACK_DEPTH;
}

Давайте посмотрим что у нас получилось.



Код основной программы целиком.

Timer xTimer("Timer", 1000, pdTRUE, NULL, vTimerCallback);
Sema xSema;
Mutex xMutex;
Queue<int, 1000> xQueue;
 
Task<configMINIMAL_STACK_SIZE> task1(vTask1, "Task 1", NULL, tskIDLE_PRIORITY);
Task<configMINIMAL_STACK_SIZE> task2(vTask2, "Task 2", NULL, tskIDLE_PRIORITY);
 
 
void vTimerCallback(TimerHandle_t pxTimer)
{
  xSema.give();
 
  MutexLocker lock(xMutex);
  Serial.println("Test");
}
 
void vTask1(void *)
{
  while(1)
  {
	xSema.take();
	int value = random(1000);
	xQueue.send(value);
  }
}
 
void vTask2(void *)
{
  while(1)
  {
	int value;
	xQueue.receive(&value);
 
	MutexLocker lock(xMutex);
	Serial.println(value);
  }
}
 
void setup()
{
  Serial.begin(9600);
 
  xTimer.start();
 
  vTaskStartScheduler();
}
 
void loop() {}

Можно дизассемблировать полученный бинарник и посмотреть что и как там расположилось (вывод objdump’а немного подкрашен для лучшей читаемости):

0x200000b0    	.bss    	512     	vApplicationGetIdleTaskMemory::Idle_Stack
0x200002b0    	.bss    	92       	vApplicationGetIdleTaskMemory::Idle_TCB
0x2000030c     	.bss    	1024   	vApplicationGetTimerTaskMemory::Timer_Stack
0x2000070c     	.bss    	92       	vApplicationGetTimerTaskMemory::Timer_TCB
0x200009c8     	.bss    	608     	task1
0x20000c28     	.bss    	608     	task2
0x20000e88    	.bss    	84       	xMutex
0x20000edc    	.bss    	4084   	xQueue
0x20001ed0    	.bss    	84       	xSema
0x20001f24     	.bss    	48       	xTimer

Цель достигнута — теперь все как на ладони. Каждый объект виден и понятен его размер (ну разве что составные объекты типа Task считают все свои запчасти одним куском). Статистика компилятора также предельно точна и на этот раз весьма полезна.

Sketch uses 20,800 bytes (15%) of program storage space. Maximum is 131,072 bytes.
Global variables use 9,332 bytes (45%) of dynamic memory, leaving 11,148 bytes for local variables. Maximum is 20,480 bytes.

Заключение

И хотя система FreeRTOS позволяет на лету создавать и удалять задачи, очереди, семафоры и мьютексы, во многих случаях это не нужно. Как правило достаточно один раз создать все объекты на старте и они будут работать до следующей перезагрузки. А это хороший повод распределить такие объекты статически на этапе компиляции. Как результат мы получим четкое понимание о занимаемой нашими объектами памяти, где что лежит и сколько еще свободной памяти осталось.

Очевидно, что предложенный способ подходит только для размещения объектов время жизни которых сравнимо с временем жизни всего приложения. В противном случае стОит использовать динамическую память.

Помимо статического размещения объектов FreeRTOS мы еще написали удобные обертки над примитивами FreeRTOS, что позволило несколько упростить клиентский код, а также энкапсулировать

Интерфейс при необходимости можно упростить (например, не проверять код возврата, или не использовать таймауты). Также стоит отметить, что реализация неполная – я не заморачивался реализацией всех возможных способов отправки и приема сообщений через очередь (например из прерывания, отправку в начало или конец очереди), не реализована работа с примитивами синхронизации из прерываний, счетные (не бинарные) семафоры, и много чего другого.

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

Пример из статьи находится тут.

Всем спасибо кто дочитал эту статью до конца. Я буду рад конструктивной критике. Мне также будет интересно обсудить нюансы в коментариях.