Коммуникация

Дополнительные главы

chrns

Для организации обмена данными между узлами используется множество разнообразных интерфейсов: CAN, Ethernet, UART, SPI, I2C, 1-Wire и т.д. Какие-то из них параллельные, какие-то последовательные. Одни могут работать в дуплексном режиме, некоторые только в полудуплексном или вообще в симплексном. Какие-то интерфейсы являются синхронными, а другие асинхронными. Так или иначе, главное, для чего они предназначены — это передача и прием данных. Микроконтроллеры, как правило, имеют аппаратные реализации популярных интерфейсов и позволяют легко передавать и принимать последовательность бит, абстрагируясь от временных интервалов и физического уровня в целом. Когда нужно принять или отправить один байт, то проблем с ожиданием отправки и обработкой не возникает: условно, один байт — одна функция (отправки или обработчик прерывания). Но что делать, если принять или отправить нужно целую строчку? Каким образом организовать программу, чтобы обработка массивов символов не вызывала головной боли?

Возьмем интерфейс асинхронного приемопередатчика UART (сокр. от Universal Asynchronous Receiver-Transmitter). Для работы ему нужны всего две линии: RXD (она же RX, англ. receive) и TXD (она же TX, англ. transmission), принимающая и передающая линии соответственно. Подключать их нужно крест-накрест, т.е. RX к TX, TX к RX. Стандартная посылка занимает 10 бит (зависит от настроек). Передача происходит последовательно, бит за битом в равные промежутки времени, которые определяются скоростью для конкретного соединения и указывается в бодах (в данном случае соответствует битам в секунду). Существует общепринятый ряд стандартных скоростей: 300; 600; 1200; 2400; 4800; 9600; 19200; 38400; 57600; 115200; 230400; 460800; 921600 бод. Скорость (S, бод) и длительность бита (T, секунд) связаны соотношением T=S-1. В начале посылки идет стартовый бит, сигнализирующий о начале передачи. Далее идут данные, обычно 8 бит. Для увеличения помехозащищенности иногда используется бит четности/нечетности, по которому можно определить, есть ли в принятых данных ошибки (если их немного). Довольно часто он опускается, тогда такой режим называют No parity, и соответственно другие два — Even parity с проверкой на четность и Odd parity с проверкой на нечетность. Завершает посылку стоп-бит, длительность которого может равняться, в зависимости от настроек, 1, 1.5, 2 длительностям бита. Условимся, что наш интерфейс настроен на скорость 9600, бит четности не используется, а длина стоп-бита равна 1 (9600/8N1).

Теперь, когда принцип работы интерфейса UART описан, самое время описать проблемы, связанные с отправкой данных. Представьте некоторое устройство с автономным питанием. Его задача — собирать данные и отправлять их на удаленный узел при помощи радиомодуля, который общается с МК через интерфейс UART. Так как устройство работает от батарейки, тратить энергию на работу радиомодуля тогда, когда он ничего не отправляет, очень глупо. При этом при каждом его включении, перед отправкой данных, приходится инициализировать протокол и частоту радиоканала, а потом ждать, пока модуль выйдет в рабочий режим. Затраты энергии на эти операции относительно невелики, однако повторяя их много раз (скажем, раз в минуту), мы впустую потратим внушительное количество энергии. Разработчики решили, что их устроит сценарий, когда данные будут сохраняться во флеш-память и отправляться раз в неделю. Допустим, данные, которые нам нужны, форматируются по стандарту NMEA 0183. Пример сообщения представлен ниже.

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

$AMRTGR,300318,092747,51.245627,N,137.457394,W,65,*3A

В данной строчке 53 символа. Каждый символ кодируется 8 битами плюс 2 сервисными битами UART (стартовым и стоповым). Таким образом, в передатчик нужно отправить 530 бит. Следовательно, время, которое необходимо для передачи строки, можно найти по формуле:

t(9600, 53) = bits / S = (53 · 10) / 9600 = 55,2 ms

Вспомним, что квант времени в ОС обычно выставляется равным 1 мс, т.е. в 55 раз меньше, чем время, которое необходимо потратить на передачу. Очевидно, что отключать планировщик на всё это время не самая лучшая идея, которая вам может прийти в голову: вы не сможете реагировать ни на что другое, а будете заняты передачей данных. В прошивке bare-metal проблема точно такая же.

void send_data(uint8_t *data, uint32_t n) {
#ifdef FREERTOS
    vTaskSuspendAll();
    {
#endif
        for (uint32_t i = 0; i < n; i++) {
            UART_SEND(data[i]);
            while(!IS_UART_READY());
        }
#ifdef FREERTOS
    }
    xTaskResumeAll();
#endif
}

Теперь представьте, что будет при большом объеме данных. Мы собирали данные раз в минуту 7 дней подряд. Каждая запись — 53 символа. Тогда количество байт к передаче будет равно 534240. Рассчитаем время:

t(9600, 534240) = (10 · 534240) / 9600 = 556,5 s

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

Решений может быть несколько. Например, в функции send_data() можно отправить только первый символ из буфера, а блок UART настроить на прерывание при завершении отправки байта. В таком случае можно внутри прерывания итерировать переменную index до конца буфера и отправлять байт из него по соответствующему индексу.

void USART1_IRQHandler(void) {
    if (index < BUFFER_SIZE)
        send_byte(buffer[++index]);
    // ...
}

Отзывчивость системы возрастет, но прерывания будут срабатывать достаточно часто, примерно каждые 0.1 мс для скорости 9600, т.е. по 10 раз в квант времени. С увеличением скорости UART станет намного хуже: для скорости 115200 период прерываний составит 8,68 мкс (примерно 115 раз за квант времени!), что сделает работу задач неэффективной.

Для того чтобы не тратить время на простое копирование данных и ожидание их отправки (или приема), в микроконтроллерах предусматривают специальный блок прямого доступа к памяти (англ. Direct Memory Access, DMA). Мы не будем рассматривать работу этого блока, реализация может отличаться от производителя к производителю, но отметим, что он довольно гибок в настройке. Он может копировать данные либо из одного массива в другой, либо из массива в регистр, либо из регистра в массив (инкрементируя указатель), при этом не тратя процессорное время. Если привязать DMA к событию UART (сбросу/выставлению флага о завершении передачи), он скопирует значение из ячейки указанного массива в регистр UART, что вызовет новую передачу данных. Таким образом, процессор практически не участвует в процессе передачи и может заняться более важными вещами.

void send_data(uint8_t *data, uint32_t n) {
    DMA_START(data, n);
}

Осталось разобраться с приемом данных — и здесь опять не всё так просто. Очевидно, что использование прерываний при плотном потоке данных вызовет такие же проблемы, как и при их отправке: прерывания будут происходить очень часто, что пагубно скажется на производительности системы. Снова можно прибегнуть к модулю DMA, который разгрузит процессор от тупого копирования данных из одной области памяти в другую. Однако здесь всплывает еще одна проблема — получатель в общем случае не знает, какое количество байт было отправлено. Допустим, сообщение приходит раз в 5-10 минут, и его длина варьируется от 10 до 20 символов. Если предписать DMA скопировать 10 чисел, в то время как было отправлено 15, то 5 символов просто будут утрачены. Такое поведение может привести к нежелательным последствиям.

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

Решить проблему можно несколькими способами, основанными на одной идее — необходимо детектировать отсутствие приема данных в течение какого-то периода времени. Удаленный узел обычно отправляет команды целиком, т.е. байт за байтом, без задержек. В таком случае после приема 10 символов можно ожидать 11-й вслед за ними. Если его нет — значит, передача завершена. Сделать это можно либо через таймер, который будет сбрасываться каждый раз, когда байт будет принят, либо через специальный детектор пустого символа UART.

Таймер следует настроить в режим захвата (англ. capture mode). Описание метода можно найти в документе AN3019.

USART_ITConfig(USART2, USART_IT_IDLE, ENABLE);

В обработчике этого прерывания модуль DMA может быть принудительно остановлен, что приведет к вызову прерывания о конце приема данных. Далее остается лишь обработать полученные данные.

Вернемся, однако, к методу с прерываниями, ведь не все микроконтроллеры оснащены модулем DMA. Мы не говорили об этом напрямую, но подразумевалось, что работа осуществляется с буфером (англ. buffer), представляющим собой обычный массив.

#define BUFFER_SIZE       16
volatile uint32_t index = 0;
volatile uint8_t rx_buffer[BUFFER_SIZE];

В тех случаях, когда нам есть на что ориентироваться в самом принимаемом сообщении (на символ перевода каретки \r, начала новой строки \n, конца сообщения \0 или чему-то еще), можно не заморачиваться с организацией буфера. Довольно часто можно встретить модули, которые работают посредством AT-команд:

AT-команды (они же — набор команд Hayes) были разработаны еще в 1977 году, когда компания Hayes выпустила свой модем Smartmodem 300 baud. Решение оказалось таким удачным, что AT-командами и по сей день пользуются различные компании. Структура команды очень проста: она начинается с букв AT (англ. attention), после чего пишется имя команды, и завершается строка символом конца строки (\r).

AT+CWSAP="WICHAITER","12345678",5,0\n\r

Приведенная выше строка задает Wi-Fi модулю ESP8266 название точки доступа и ее пароль. Пример реализации буфера для такой ситуации приведен ниже.

#define END     '\n'
void USART1_IRQHandler(void) {
    uint8_t ch = USART1_Read();
    if (ch != END) {
        rx_buffer[index++];
    } else {
        index = 0;
        ready = 1; // or vTaskNotifyGiveFromISR()
    }
    // ...
}

В обработчике прерывания пришедшие данные складываются в массив, и переменная index итерируется до тех пор, пока не встретится символ окончания передачи. (Здесь, однако, нет защиты от переполнения буфера.) Всё просто. Но не во всех случаях такой подход можно использовать. Допустим, есть некоторое гипотетическое устройство, которое должно принять 456 байт информации от компьютера и сохранить их себе на флешку. Буфер, который был создан разработчиком для принятия входного потока, позволяет хранить до 256 символов, что меньше требуемого объема. Ориентироваться в самом потоке данных у нас нет никакой возможности — нам даже не известен объем данных, который будет прислан. Что же делать?

Элегантным решением является циклический буфер (англ. cyclic buffer), который также называют круговым (англ. circular buffer), или кольцевым (англ. ring buffer).

Такая структура подразумевает массив фиксированного размера и является самым простым способом организации очереди с логикой FIFO (First In — First Out). Приведенное изображение — условность, на самом деле это линейный кусок памяти, в котором программным путем при переходе от последнего элемента новая запись записывается в первую ячейку.

Реализация может быть разной; мы рассмотрим самую простую и интуитивно понятную, оформив ее в виде модуля. Создадим заголовочный файл и объявим все необходимые прототипы функций, а также тип данных циклического буфера:

Предложенная реализация не является потокобезопасной.

#ifndef __CIRCULAR_BUFFER_H__
#define __CIRCULAR_BUFFER_H__

#define BUFFER_SIZE  16

typedef struct {
    uint8_t head;
    uint8_t tail;
    uint8_t data[BUFFER_SIZE];
} CB_t;

void cb_reset(CB_t *buf);
uint32_t cb_is_empty(CB_t *buf);
uint32_t cb_is_full(CB_t *buf);
void cb_put(CB_t *buf, uint8_t ch);
uint8_t cb_get(CB_t *buf);

#endif /* __CIRCULAR_BUFFER_H__ */

Предполагается, что разработчик создаст переменную типа CB_t и будет пользоваться API-функциями, предоставляемыми заголовочным файлом circular_buffer.h. Первая функция, которую он должен вызвать — cb_reset(), задача которой — сбросить поля head и tail в 0. К сожалению, Си не предлагает более элегантного метода.

void cb_reset(CB_t *buf) {
    if(buf) {
        buf->head = 0;
        buf->tail = 0;
    }
}

Далее реализуем функции проверки состояния буфера. Для упрощения реализации условимся, что head никогда не будет равен tail: мы потеряем одну ячейку, но избавим себя от необходимости создавать дополнительные проверки и усложнять реализацию.

uint32_t cb_is_full(CB_t *buf) {
    return buf ? (((buf->head + 1) % BUFFER_SIZE) == cbuf->tail) : NULL;
}

uint32_t cb_is_empty(CB_t *buf) {
    return buf ? (buf->head == buf->tail) : NULL;
}

Допишем оставшиеся две функции для добавления и считывания данных из буфера.

void cb_put(CB_t *buf, uint8_t ch) {
    if(buf) {
    buf->data[buf->head] = ch;
    buf->head = (buf->head + 1) % BUFFER_SIZE;
        if(buf->head == buf->tail)
            buf->tail = (buf->tail + 1) % BUFFER_SIZE;
    }
}

uint8_t cb_get(CB_t *buf) {
    uint8_t ch;
    if(buf && !cb_is_empty(buf)) {
        ch = buf->data[buf->tail];
        buf->tail = (buf->tail + 1) % BUFFER_SIZE;
    }
    return ch;
}

Изменено: