STM32 и FreeRTOS. 3. Встаем в очередь
Раньше: про потоки и про семафоры«Вас много, а я одна!» — классическая фраза продавщицы, которую затерроризировали покупатели с вопросами «А есть …?». Вот и в микроконтроллерах случаются полностью аналогичные ситуации, когда несколько потоков требуют внимания от какой-либо медленной штуки, которая просто физически не способна обслужить всех разом.
Возьмем наиболее яркий и богатый проблемами пример, на котором «валятся» большинство неопытных программистов. Есть мощный и достаточно быстрый микроконтроллер. К нему подключен с одной стороны адаптер com-порта, через который пользователь подает команды и получает результаты, а с другой — шаговый двигатель, который согласно этим командам поворачивается на какой-то угол. И конечно же, прикольная кнопочка, которая тоже что-то этакое значит для пользователя. Где можно наловить проблем?
Пойдем со стороны пользователя. Com-порт, или USART (universal synchronous/asynchronous receiver/transmitter) — штука очень нежная и капризная. Основной каприз заключается в том, что ее нельзя оставлять без внимания. По одной простой причине — для экономии выводов в 99% случаев выводят только сигналы приема и передачи, оставляя сигналы разрешения приема и передачи (rtr/rts, dtr, cts и прочие) за бортом. И стоит микропроцессору хоть чуть-чуть замешкаться, как символ будет потерян. Судите сами: магические цифры 9600/8N1 говорят нам, что в секунду по линии летит 9600 бод, а один символ кодируется 10 (1 стартовый, 8 данных и 1 стоповый) импульсами. В итоге максимальная скорость передачи составляет 9600/10 = 960 байт в секунду. Или чуть больше одной миллисекунды на байт. А если по ТЗ скорость передачи 115200? А ведь микроконтроллеру надо еще обработать эти данные. Кажется, что все плохо, но в реальности куча устройств работает и не доставляет проблем. Какие же решения применяются?
Cамым распространенным (и правильным) решением является перекладка всей работы по приему или передаче символов на плечи микроконтроллера. Делать аппаратный usart порт научились сейчас все, поэтому типовое решение в большинстве случаев выглядит примерно вот так:
void URARTInterrupt () { a=GetCharFromUSART (); if (a!=0×13) buffer[count++]=a; else startProcessing (); } Где проблема? Во-первых, проблема в вызове startProcessing. Стоит этой функции хоть чуть-чуть задержаться с работой, как очередной символ будет потерян. Берем STM32L1 (который на минимальной частоте успевает за 1 мс обработать всего 84 команды) и при более-менее развесистой логике полученная конструкция будет терять символы.
После дружеского похлопывания по голове у нас в студии программист переписывает код так, что бы вместо вызова startProcessing поднимался семафор, который и запускал бы обработку полученных данных. И тут же получил следующую проблему в полный рост: пока startProcessing что-то там перемалывает из буфера, обработчик нового символа начинает планомерно затирать еще не обработанный буфер. После очередного разорванного свитера и скуренной книжки по буферам программист реализует кольцевой буфер с указателями, чем закапывает проблему еще глубже.
Обычно где-то на этом месте я замечаю удивленные глаза народа и возмущенные выкрики в духе «ну такие алгоритмы же работают в куче проектов и ничего, никаких проблем». Да, работают. Но в каких проектах? Только в тех, где можно все общение с контроллером перевести на полудуплексный режим, как в рациях. Вот пример диалога пользователя (П) и контроллера (К)
П: ATZ (Контроллер, ты живой?)К: ОК (Ага, я тут)П: М340,1 (Сделай чего-то)(тут может быть пауза, иногда очень большая)К: ОК (Сделал типа)
Где проблемы? Во-первых, нельзя послать несколько команд подряд. Значит либо надо будет менять логику или делать сложные команды с несколькими параметрами. Во-вторых, между запросом и ответом не может быть никаких других запросов и ответов. А вдруг пользователь в процессе движения нажмет кнопку? Что делать? Выход только один — использовать родную природу порта, а именно его асинхронность. В результате диалог между пользователем и контроллером выглядит примерно так
П: ATZ (жив?)К: ОК (ага)П: M340,1К: К1=1 (кнопку нажали)П: Е333,25,2(тишина)К: Е=1 (задачу Е выполнил)К: М=1 (задачу М выполнил)
Конечно, логика обработки подобного потока немного усложняется, зато благодаря такой асинхронности мы сразу получаем множество преимуществ перед «классической школой».
Во-первых, резко повышается отзывчивость интерфейса с пользователем. Ну работает там где-то моторчик или считается что-то, но это же не повод «замирать» или «тормозить». А когда заказчик понимает, что величина тормозов практически не зависит от мощности контроллера, у него возникают резонные вопросы…Во-вторых, такое поведение ближе к поведению небольшого начальника в реальной жизни. И скопировать такое поведение очень легко — все видели, все участвовали.И наконец, мы разделяем поток управления на «командный» и «статусный». И в результате реализация всяких индикаторов объемов выполнения задачи или углов поворота в реальном времени не представляет трудностей.
А теперь со всеми этими задумками взглянем на противоположную сторону — на сторону исполнителя. Он достаточно медлителен, что бы попросту не успеть обработать команды по порядку. И время запуска-остановки моторчика очень большое, поэтому хорошо бы сделать примитивную оптимизацию в духе «если поступило две команды на поворот в одну сторону, то поверни за раз».
Что делает обычный программист? Так как он прочитал предыдущие статьи и кучку книжек, то он рисует логику расставляя семафоры по необходимости, а для блокирования одновременного доступа к моторчику использует мутексы. Все хорошо, но код получается громоздкий и тяжело читаемый. Что делать? У нас в studiovsemoe.com мы используем рецепт Шарикова: «В очередь, сукины дети! В очередь!». Опять же, концепция очереди впитывается чуть ли не с молоком матери, поэтому проблем с пониманием никаких.
В данном примере можно просто создать три очереди. Первая это команды, полученные от пользователя. В нее засовывается все (ну или после минимальной проверки), что принято со всех входных портов. Вторая это те данные, которые необходимо выдать пользователю. Состояние кнопок, результаты расчетов и так далее. И наконец, третья очередь служит для заданий моторчику/считалке/кому-то еще. А между этими очередями потоки для преобразования данных от пользователя в задания для считалки. Так как в FreeRTOS есть официальные функции для «заглядывания» внутрь очереди, то легко сделать оптимизацию для случаев, когда следующее действие продолжает/повторяет текущее.
Итак, всего три очереди, а дикая куча проблем решена. Во-первых, нет даже потенциальной проблемы потери или переписи буфера принятых символов. Правда результатом станет чуть больший расход памяти, но это допустимая цена. Во-вторых, нет проблем с выводом. Все процессы просто пишут в одну очередь, а как выводить, в каком формате и прочее — это уже не их забота. Надо оформлять вывод в рамочку — переписываем одну функцию, а не все, которые что-то могут выводить. Опять же, нет проблем с одновременным/перекрывающимся выводом (когда один поток выводит 12345, а другой qwerty, но пользователь получает что-то типа 1qw234er5ty). И наконец благодаря такому подходу задачи очень легко разделить на более мелкие и раскидать их по потокам/ядрам микропроцессора. А все это означает ускорение разработки и снижение стоимости поддержки. Все рады, все довольны.
Что бы не отходить от практики мигания светодиодами, сделаем следующую демонстрацию: создадим очередь, из которой раз в секунду будет забираться элемент (имитация обсчета). По нажатию кнопки в очередь два раза в секунду будет помещаться элемент (управляющие команды). Ну, а светодиодики будут показывать загруженность очереди. Опять же, предупреждаю что код для повышения читаемости написан без обработки ошибок, поэтому аккуратней.
Где-то в начале кода определим очередь
xQueueHandle q; Код зажигания светодиодов поменяем по принципу «в очереди больше Н элементов? зажигаем, если нет, то нет»
if (uxQueueMessagesWaiting (q)>1) HAL_GPIO_WritePin (GPIOE, GPIO_PIN_9, GPIO_PIN_SET); else НAL_GPIO_WritePin (GPIOE, GPIO_PIN_9, GPIO_PIN_RESET); osDelay (100); Перед запуском планировщика проинициализируем очередь так, что бы в ней могли «стоять» 8 элементов размером с байт.
q = xQueueCreate (8, sizeof (unsigned char)); Ну и код кнопки для помещения символов в очередь
if (HAL_GPIO_ReadPin (GPIOA, GPIO_PIN_0)==GPIO_PIN_SET) { unsigned char toSend; xQueueSend (q, (void *) &toSend, portMAX_DELAY); } osDelay (500); Чего не хватает? Воркера, который забирает из очереди задания. Пишем.
static void WorkThread (void const * argument) { for (;;) { unsigned char rec; xQueueReceive (q, &(rec), portMAX_DELAY); osDelay (1000); } } Как видите, все параметры аналогичны тем, что используются в семафорах, поэтому повторяться не буду. И как обычно, результат проще показать.
[embedded content]
Код со всеми потрохами доступен по адресу kaloshin.ru/stm32/freertos/stage3.rar
