27. Очереди

Задачи не должны трогать общие переменные напрямую — это прямая дорога к гонкам, о которых раздел 30. Обмениваются они через очереди: механизм, который одновременно передаёт данные и синхронизирует момент передачи.

Как это называется в документации

Оригинал	Перевод
queue	очередь
item	элемент, одно сообщение
queue length	длина очереди, число мест
item size	размер одного сообщения
send to back / to front	отправить в конец / в начало
peek	заглянуть, не забирая
block time / ticks to wait	время ожидания
queue set	набор очередей
overwrite	перезапись единственного места
messages waiting	число сообщений в очереди

27.1 Главное свойство: очередь копирует

Решение, определяющее всё остальное поведение.

Отправка кладёт в очередь копию значения, а не ссылку на него.

sample_t s;

s.raw = ADC1->DR;
xQueueSend(q, &s, portMAX_DELAY);   /* адрес передан, скопировано значение */

Адрес передаётся только затем, чтобы ядру было откуда копировать. После возврата из функции переменная s больше ни при чём: она может измениться, выйти из области видимости, исчезнуть вместе с кадром стека — в очереди лежит независимая копия.

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

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

Где лежат копии — в окне памяти: переменная отправителя, хранилище очереди и переменная получателя:

Создание

QueueHandle_t q = xQueueCreate(16, sizeof(sample_t));
if (q == NULL) {
    /* не хватило кучи */
}

Два числа: сколько мест и какого размера каждое. Оба фиксированы на всю жизнь очереди.

⚠️ Очередь занимает свой максимум с момента создания. Место под все шестнадцать сообщений выделяется сразу, независимо от того, сколько их лежит. Пустая очередь и полная стоят одинаково.

Что она стоит

Обратите внимание на соотношение: служебные 80 байт дороже полезных 64. На мелких сообщениях заголовок доминирует, и очередь из восьми однобайтовых значений обходится в десять раз дороже, чем весят её данные.

Практический вывод: лучше одна очередь на несколько полей, чем несколько очередей по одному полю. Шестнадцать четырёхбайтовых сообщений в одной очереди стоят 144 байта; те же данные, разложенные по четырём однобайтовым очередям, — 384.

27.2 Отправка и приём

/* задача-производитель */
xQueueSend(q, &s, pdMS_TO_TICKS(10));       /* ждать место до 10 мс */

/* задача-потребитель */
xQueueReceive(q, &s, portMAX_DELAY);        /* ждать сообщение сколько нужно */

Третий аргумент — время ожидания, и именно он превращает очередь из контейнера в средство синхронизации.

Ожидающая задача не тратит процессор. Она переходит в состояние ожидания, и ядро будит её ровно в момент появления сообщения. Это и делает связку «производитель — потребитель» через очередь такой удобной: потребитель написан как простой бесконечный цикл, а спит он ровно столько, сколько нужно.

static void consumer_task(void *arg)
{
    sample_t s;

    for (;;) {
        if (xQueueReceive(q, &s, portMAX_DELAY) == pdPASS) {
            process(&s);
        }
    }
}

⚠️ Результат проверять обязательно. При ожидании с ограничением отказ означает, что сообщения не появилось, и переменная s осталась с прежним содержимым — то есть с мусором при первом проходе. Обработка непроверенных данных даёт ошибку, которую потом ищут в самом неподходящем месте. Проверяют его и при portMAX_DELAY: без INCLUDE_vTaskSuspend 1 это тоже ограничение, только длинное.

Напомним мину из раздела 24: отличить «очередь пуста» от «очередь полна» по коду возврата невозможно — errQUEUE_EMPTY, errQUEUE_FULL и pdFAIL это один и тот же ноль. Смысл определяется тем, какую функцию вы вызвали.

Отправка в начало очереди

xQueueSendToBack(q, &s, 0);      /* обычный порядок; то же, что xQueueSend */
xQueueSendToFront(q, &s, 0);     /* вперёд всех */

Второй вариант ломает порядок намеренно — так проталкивают срочное сообщение мимо очереди. Приём: команда останова в очереди команд двигателя должна обгонять накопившиеся команды движения.

Заглянуть, не забирая

xQueuePeek(q, &s, 0);            /* сообщение остаётся в очереди */

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

Почтовый ящик: очередь на одно место

Для «последнего значения» — температуры, состояния, режима — очередь делают длиной ровно 1 и пишут в неё с перезаписью:

QueueHandle_t mb_temp = xQueueCreate(1, sizeof(int16_t));   /* длина строго 1 */

xQueueOverwrite(mb_temp, &t);      /* писатель: не ждёт и не отказывает, старое затёрто */
xQueuePeek(mb_temp, &t, 0);        /* читатели: последнее значение, не забирая */

xQueueOverwrite допустима только для очереди длиной 1: на более длинной сработает configASSERT. Из прерывания — xQueueOverwriteFromISR. Это и есть прямой способ «отбрасывать старое вместо нового» из 27.6.

27.3 Из прерывания — только FromISR

У каждой функции есть двойник для обработчиков прерываний:

void ADC1_2_IRQHandler(void)
{
    BaseType_t woken = pdFALSE;
    sample_t s = { .raw = ADC1->DR };

    xQueueSendFromISR(q, &s, &woken);
    portYIELD_FROM_ISR(woken);
}

Два отличия от обычной функции.

Нет времени ожидания. Прерывание не может ждать — ждать в нём некому. Не получилось положить в очередь — значит не получилось.

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

Механизм и его тонкости разобраны в разделе 28 целиком — там же о том, какие прерывания вообще имеют право обращаться к ядру.

27.4 Приём из нескольких источников

Задача часто должна реагировать на события из разных мест: команды от связи, отсчёты от датчика, срабатывание таймера. Ждать на нескольких очередях по очереди нельзя — заснув на первой, задача не увидит второй.

Три решения, в порядке предпочтения.

Одна очередь с размеченным сообщением — самое простое и обычно лучшее:

typedef struct {
    enum { EV_SAMPLE, EV_COMMAND, EV_TIMEOUT } kind;
    union {
        uint16_t sample;
        uint8_t  command;
    } data;
} event_t;

Все источники пишут в одну очередь, задача разбирает по полю kind. Порядок событий сохраняется, память тратится один раз, код читается линейно.

Набор очередей (configUSE_QUEUE_SETS) позволяет ждать сразу на нескольких. Механизм работает, но обходится дороже и по памяти, и по времени, а код с ним заметно сложнее. Берут его, когда очереди уже существуют и переделывать их нельзя.

Группа событий (раздел 31) — если передавать нужно не данные, а сам факт события.

27.5 Данные переменной длины и крупные сообщения

Очередь хранит элементы одного фиксированного размера. Кадр переменной длины в неё не положишь, а килобайтный копировать дорого.

Приём стандартный: в очередь кладут указатель.

static uint8_t frame_pool[4][256];
QueueHandle_t q_frames = xQueueCreate(4, sizeof(uint8_t *));

uint8_t *p = frame_pool[i];
xQueueSend(q_frames, &p, 0);        /* передаём четыре байта вместо 256 */

Экономия очевидна: очередь из четырёх указателей — шестнадцать байт хранилища вместо килобайта, и копируется четыре байта вместо двухсот пятидесяти шести.

Но появляется вопрос, которого не было раньше, — кто владеет буфером.

⚠️ Указатель на локальную переменную — самая частая ошибка с очередями. Отправитель кладёт в очередь адрес переменной из своего стека и выходит из функции. Получатель забирает адрес, который теперь указывает на чужой мусор. Работает, пока стек не переиспользовали, — то есть неустойчиво и с виду случайно.

Правило простое: буфер обязан пережить приём. Годятся статический массив, пул буферов, память из кучи; не годится ничто локальное.

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

27.6 ⚠️ Проблемы очередей

Очередь переполняется

Означает ровно одно: производитель быстрее потребителя. Никакой размер очереди этого не лечит — он лишь откладывает момент.

if (xQueueSend(q, &s, 0) != pdPASS) {
    dropped++;              /* считать потери, а не игнорировать */
}

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

Наблюдать за заполнением дёшево:

UBaseType_t waiting = uxQueueMessagesWaiting(q);

Число, устойчиво растущее со временем, — прямой признак того, что потребитель не справляется. Это стоит выводить при отладке всегда.

Ожидание с portMAX_DELAY в обоих направлениях

Если и отправитель, и получатель ждут без ограничения, а третья сторона почему-то молчит, обе задачи остановятся навсегда, и внешне это будет выглядеть как зависание прибора. Ограниченное ожидание с обработкой отказа надёжнее.

Очередь не решает вопрос приоритетов

Очередь не наследует приоритет — этим занимается мьютекс (раздел 30). Если высокоприоритетная задача ждёт сообщения, которое должна прислать низкоприоритетная, а та не получает процессора, ждать она будет долго. Очередь передаёт данные, но не поднимает никого в очереди на процессор.

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

Копирование стоит времени

При частом обмене крупными сообщениями двойное копирование становится заметным. Это ровно тот случай, когда переходят на передачу указателей.

Очередь между прерыванием и задачей — не всегда лучший выбор

Для передачи одного факта «событие произошло» очередь избыточна: 80 байт заголовка ради флага. Уведомление задачи (раздел 31) делает то же самое дешевле и быстрее.

27.7 🔧 Кнопка → очередь → светодиод

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

typedef struct {
    uint8_t    button;                      /* какая кнопка */
    TickType_t at;                          /* когда нажата, в тиках */
} btn_event_t;                              /* очередь копирует структуру целиком */

static void button_task(void *arg)
{
    uint32_t stable = 0, last = 0, same = 0;
    (void)arg;
    for (;;) {
        uint32_t now = button_read();
        same = (now == last) ? same + 1 : 0;        /* сколько чтений подряд одинаковые */
        last = now;
        if (same == 2 && now != stable) {           /* третье одинаковое — состояние устоялось */
            stable = now;
            if (stable) {
                btn_event_t ev = { 0, xTaskGetTickCount() };
                if (xQueueSend(q, &ev, 0) == pdPASS)    /* задача опроса не ждёт места */
                    presses++;
                else
                    dropped++;
            }
        }
        vTaskDelay(pdMS_TO_TICKS(10));
    }
}

static const uint16_t half_period_ms[] = { 500, 250, 100 };

static void led_task(void *arg)
{
    unsigned mode = 0;
    btn_event_t ev;
    (void)arg;
    for (;;) {
        /* ожидание очереди и есть таймер мигания */
        if (xQueueReceive(q, &ev, pdMS_TO_TICKS(half_period_ms[mode])) == pdPASS)
            mode = (mode + 1) % 3;                  /* нажатие — следующая частота */
        else
            led_toggle();                           /* тайм-аут — переключить */
    }
}

Дребезг — три одинаковых чтения подряд. Состояние считается устоявшимся, когда вход трижды подряд прочитан одинаково, то есть через 20–30 мс; скачки контакта короче этого не проходят (другие способы — в 8.5).

Тайм-аут ожидания — это таймер. xQueueReceive с тайм-аутом, равным полупериоду мигания, возвращает pdPASS, если пришло нажатие, и отказ, если время вышло. Отказ и есть момент переключить светодиод, а нажатие меняет полупериод по кругу: 500 → 250 → 100 мс. Отдельный таймер не нужен.

Опрос не ждёт очередь. Задача кнопки кладёт событие с нулевым ожиданием: если очередь полна, лучше посчитать потерю (27.6), чем застрять и перестать опрашивать кнопку.

Опыт поставлен на настоящем ядре с портом POSIX. «Нажиматель» трижды жмёт кнопку, каждый раз с дребезгом — по пять скачков через тик при нажатии и при отпускании, — а между нажатиями считаются переключения светодиода за две секунды.

Нажатий поймано 3 из 3, потерь в очереди — 0. В первом окне три переключения, а не четыре: четвёртое приходится ровно на границу окна.

27.8 Итоги

• Очередь копирует значение, а не ссылается на него: переменная отправителя после вызова свободна.
• Длина и размер элемента фиксированы навсегда; место под все элементы занимается сразу.
• Заголовок очереди — 80 байт; на мелких сообщениях он дороже данных, поэтому одна очередь со структурой лучше нескольких по полю.
• Время ожидания превращает очередь в средство синхронизации: ждущая задача процессор не потребляет.
• Результат приёма проверять обязательно — при отказе приёмная переменная не заполнена.
• Из прерываний вызываются только двойники с окончанием FromISR; времени ожидания у них нет, зато есть признак пробуждения.
• Ждать на нескольких очередях нельзя; вместо этого — одна очередь с размеченным сообщением.
• Крупные и переменные данные передают указателем, и тогда буфер обязан пережить приём, а владение — быть оговорено.
• Переполнение означает, что производитель быстрее потребителя; размер очереди это откладывает, но не лечит.
• uxQueueMessagesWaiting — дешёвый и надёжный признак того, что потребитель не справляется.
• Тайм-аут ожидания очереди может служить таймером: отказ — время вышло, данные — событие. Задача опроса кладёт в очередь без ожидания и считает потери (27.7).

Дальше: раздел 28 «Прерывания в RTOS» — какие функции ядра можно вызывать из обработчика, почему их всего часть и что за порог приоритета отделяет одни прерывания от других.