Особенности разработки встраиваемого программного обеспечения. Часть 2
В предыдущей части статьи я рассказал об особенностях реализации многозадачности для ВПО. Однако, какой бы ни был выбран подход к многозадачности, архитектура самих задач напрямую влияет на эффективность работы всей встраиваемой системы.
Для достижения оптимальной работы системы, задачи необходимо проектировать по принципу эффективного использования процессорного времени. Т.е. задача должна использовать процессорное время исключительно для полезной нагрузки. Данный принцип выполняет 2 основные функции: Улучшение отзывчивости системы и эффективного потребления электроэнергии.
Улучшение отзывчивости системы за счёт эффективного использования процессорного времени.
Этот вопрос тесно связан с такими понятиями как блокирующий/неблокирующий метод, синхронная/асинхронная операция и активное/пассивное ожидание.
Блокирующий метод называется таковым из-за невозможности продолжить работу вызывающей его единицы выполнения кода (поток или задача) до завершения своей работы (ошибка или успех). В случае с неблокирующим методом напротив, вызывающая его единица выполнения кода может продолжить свою работу.
При этом важно, как именно будет получен результат: сразу – синхронная операция, или позже (например, в прерывании) – асинхронная операция. С другой стороны, ожидание результата может потребовать активного участия единицы выполнения кода – активное ожидание, или не потребовать – пассивное ожидание.
Примером неблокирующего асинхронного метода из высокоуровневого программирования может быть механизм promise, когда результат выполнения метода отслеживается асинхронно относительно других задач или потоков, а вызывающая единица выполнения кода продолжает свою работу.
Из-за специфики работы встраиваемых систем, вопрос блокировок встает достаточно часто. Работа таких систем тесно связана с приемом-передачей и обработкой данных, которые могут передаваться по различным каналам и подвергаться обработке различной степени сложности. При всем этом, нужно принимать во внимание посредственные вычислительные возможности целевого, в контексте статьи, однопоточного МК.
Для более наглядного объяснения рассмотрим условную задачу чтения и обработки данных внешнего аналого-цифрового преобразователя (далее АЦП).
Самая простоя рабочая архитектура такой задачи может выглядеть так:
// Прототипы
bool adc_task( void );
uint32_t systick_get( void );
bool timeout_check( const uint32_t c_timeout_check_start_tick,
const uint32_t c_timeout_check_period_tick );
uint32_t ms_to_tick( const uint32_t c_ms );
gpio_state_t gpio_pin_read( const size_t c_gpio_pin_id );
spi_err_t spi_poll_read( uint8_t* const cp_arr,
const size_t c_size_byte );
adc_err_t adc_ch_data_processing( uint8_t* const cp_arr,
const size_t c_size_byte );
void spi_cs_assert( void );
void spi_cs_deassert( void );
bool adc_task( void )
{
bool is_err = false;
uint8_t adc_ch_data_arr[ADC_CH_DATA_ARR_SIZE_BYTE] = { 0, };
{ // Метод ожидания готовности преобразования
bool is_drdy = false;
const uint32_t c_timeout_check_start_tick = systick_get( );
do
{
if( timeout_check( c_timeout_check_start_tick,
ms_to_tick( ADC_DRDY_TIMEOUT_MS ) ) )
{ is_err = true; goto Exit; }
if( gpio_pin_read( ADC_PIN_DRDY ) == GPIO_STATE_RESET )
// Активный уровень низкий
{ is_drdy = true; }
}
while( !is_drdy );
}
{ // Метод чтения значений всех аналоговых каналов
spi_cs_assert( );
if( spi_poll_read( adc_ch_data_arr,
sizeof( adc_ch_data_arr ) ) != SPI_OK )
{ is_err = true; goto Exit; }
}
{ // Метод Обработки значений всех аналоговых каналов
if( adc_ch_data_processing( adc_ch_data_arr,
sizeof( adc_ch_data_arr ) ) != ADC_OK )
{ is_err = true; goto Exit; }
}
Exit:
spi_cs_deassert( );
return is_err;
}
spi_err_t spi_poll_read( uint8_t* const cp_arr,
const size_t c_size_byte )
{
spi_err_t err = SPI_OK;
size_t xfer_curr_size_byte = c_size_byte;
const uint32_t c_timeout_check_start_tick = systick_get( );
uint8_t* p_arr = cp_arr;
while( xfer_curr_size_byte > 0U )
{
if( timeout_check( c_timeout_check_start_tick,
ms_to_tick( SPI_BYTE_RECEIVE_TIMEOUT_MS ) ) )
{ err = SPI_ERR_TIMEOUT; goto Exit; }
const uint32_t c_spi_ctrl_data = reg32_read( SPI_CTRL_REG_ADDR );
if( ( c_spi_ctrl_data & SPI_CTRL_REG_ERR ) != 0U )
{ err = SPI_ERR_HW; goto Exit; }
if( ( c_spi_ctrl_data & SPI_CTRL_REG_RX_NOT_EMPTY ) != 0U )
{
*p_arr++ = reg32_read( SPI_RX_DATA_REG_ADDR );
xfer_curr_size_byte--;
}
}
Exit:
return err;
}Данная задача раскладывается на 3 метода:
Ожидание готовности преобразования (дискретный сигнал от м/с);
Чтение результата преобразования (значения всех каналов м/с);
Обработка полученных значений.
С последним, в большинстве случаев, сделать ничего не получиться, это вычислительный метод (CPU-bound), который будет блокировать задачу. Исключением может быть использование другой единицы выполнения кода, на которую можно вынести вычисления и продолжить работу задачи, если это необходимо. Или, в рамках своей же задачи, если это возможно, декомпозировать вычисления, скажем, по каналам. При таком дроблении тяжелых вычислений можно улучшить отзывчивость системы за счет выделения какой-то части процессорного времени другим задачам, но это неизбежно увеличит время самого вычисления.
Метод ожидания готовности преобразования в данном примере блокирует работу задачи до готовности преобразования. Задача ожидает результат сразу – синхронно, а в цикле проверяет состояния ножки МК в ожидании необходимого значения – активно ждет.
Метод чтения результата преобразования в данном примере также блокирует работу задачи в очередной синхронной (ждет результат сразу) операции с активным ожиданием – ожидание флага готовности принятого байта данных по SPI.
При этом, задача работает внутри единственного потока, а никаких менеджеров/автоматов нет – задача блокирует весь поток своими активными ожиданиями. Это принуждает всю систему простаивать, остальные задачи не обновляются – отзывчивость системы ухудшается.
Первое, что необходимо сделать – интегрировать конечный автомат машинных состояний (далее FSM). Интегрирование FSM поможет решить некоторые основные задачи эффективного кода – предсказуемость, тестируемость, масштабируемость и читаемость кода. В процессе проектирования и тестирования FSM станет понятно, где и как можно эффективно использовать процессорное время.
FSM справедлив для архитектуры кода, как его фундамент, но как обстоят дела с аппаратным обеспечением?
Для эффективного использования процессорного времени, разработчики МК интегрируют специализированные устройства для решения вопроса работы с аппаратным обеспечением, в частности, прием-передача данных – контроллер прямого доступа к памяти (далее DMA). Контроллер DMA представляет собой, своего рода, сопроцессор, который управляет потоком данных без вмешательства процессорного ядра (потока). Контроллер DMA имеет определенное число каналов для приема-передачи данных, а самих контроллеров в МК может быть несколько. Это позволяет реализовать эффективные способы (неблокирующий, параллельный, асинхронный) приема-передачи данных, практически, в пределах всего адресного пространства МК. В том числе адреса регистров периферийных блоков, например, последовательных интерфейсов передачи данных.
Для приема-передачи данных по каналам DMA разработчику необходимо лишь сконфигурировать конкретный канал DMA: указать размер транзакции, адрес памяти источника (src) и назначения (dst), а также некоторые режимы работы самого канала DMA.
Принимая во внимание вышесказанное, можно спроектировать задачу по-новому:
// Прототипы
bool adc_task( void );
uint32_t systick_get( void );
bool timeout_check( const uint32_t c_timeout_check_start_tick,
const uint32_t c_timeout_check_period_tick );
uint32_t ms_to_tick( const uint32_t c_ms );
gpio_state_t gpio_pin_read( const size_t c_gpio_pin_id );
spi_err_t spi_dma_read( uint8_t* const cp_arr,
const size_t c_size_byte );
bool spi_dma_read_cmplt_is( void );
adc_err_t adc_ch_data_processing( uint8_t* const cp_arr,
const size_t c_size_byte );
void spi_cs_assert( void );
void spi_cs_deassert( void );
typedef enum
{
ADC_FSM_S_CONVERSION_WAIT_READ = 0U,
ADC_FSM_S_CONVERSION_READ_WAIT,
ADC_FSM_S_CONVERSION_PROCESSING,
ADC_FSM_S_ERROR
} adc_fsm_t;
struct
{
adc_fsm_t stateMachine;
uint32_t adcDrdyTimeoutCheckStartTick;
uint32_t spiReadTimeoutCheckStartTick;
uint8_t adcChDataArr[ADC_CH_DATA_ARR_SIZE_BYTE];
} adc_tcb;
bool adc_task( void )
{
bool is_err = false;
// ...
switch( adc_tcb.stateMachine )
{
// Метод ожидания готовности преобразования
// Метод чтения значений всех аналоговых каналов
case ADC_FSM_S_CONVERSION_WAIT_READ :
{
if( timeout_check( adc_tcb.adcDrdyTimeoutCheckStartTick,
ms_to_tick( ADC_DRDY_TIMEOUT_MS ) ) )
{
adc_tcb.stateMachine = ADC_FSM_S_ERROR;
is_err = true;
goto Exit;
}
if( gpio_pin_read( ADC_PIN_DRDY ) == GPIO_STATE_RESET )
// Активный уровень низкий
{
spi_cs_assert( );
if( spi_dma_read( adc_tcb.adcChDataArr,
sizeof( adc_tcb.adcChDataArr ) ) != SPI_OK )
{
adc_tcb.stateMachine = ADC_FSM_S_ERROR;
is_err = true;
goto Exit;
}
adc_tcb.spiReadTimeoutCheckStartTick = systick_get( );
adc_tcb.stateMachine = ADC_FSM_S_CONVERSION_READ_WAIT;
}
else
{
/* Stay here */
}
break;
}
// Метод ожидания чтения значений всех аналоговых каналов
case ADC_FSM_S_CONVERSION_READ_WAIT :
{
if( timeout_check( adc_tcb.spiReadTimeoutCheckStartTick,
ms_to_tick( SPI_READ_TIMEOUT_MS ) ) )
{
adc_tcb.stateMachine = ADC_FSM_S_ERROR;
is_err = true;
goto Exit;
}
if( spi_dma_read_cmplt_is( ) )
{
spi_cs_deassert( );
adc_tcb.adcDrdyTimeoutCheckStartTick = systick_get( );
adc_tcb.stateMachine = ADC_FSM_S_CONVERSION_PROCESSING;
}
else
{
/* Stay here */
}
}
// Метод Обработки значений всех аналоговых каналов
case ADC_FSM_S_CONVERSION_PROCESSING :
{
if( adc_ch_data_processing( adc_tcb.adcChDataArr, // <-- Stay here
sizeof( adc_tcb.adcChDataArr ) ) != ADC_OK )
{
adc_tcb.stateMachine = ADC_FSM_S_ERROR;
is_err = true;
goto Exit;
}
else
{
// Restart FSM
adc_tcb.stateMachine = ADC_FSM_S_CONVERSION_WAIT_READ;
}
}
case ADC_FSM_S_ERROR :
default :
{
// ...
}
}
Exit:
return is_err;
}
spi_err_t spi_dma_read( uint8_t* const cp_arr,
const size_t c_size_byte )
{
spi_err_t err = SPI_OK;
// ...
// Set DMA src - SPI RX REG
reg32_write( DMA_CH_SPI_RX_SRC_REG_ADDR,
SPI_RX_REG_ADDR );
// Set DMA dst - memory
reg32_write( DMA_CH_SPI_RX_DST_REG_ADDR,
(uint32_t)cp_arr );
// Set XFER size
reg32_write( DMA_CH_SPI_RX_XFER_REG_ADDR,
c_size_byte );
// Enable DMA channel - start data receive
reg32_or( DMA_CH_SPI_RX_CTRL_REG_ADDR,
DMA_CTRL_REG_CH_ENABLE );
Exit:
return err;
}Теперь задача имеет архитектуру FSM, в рамках одного вызова задачи выполняется работа одного машинного состояния.
Метод ожидания преобразования теперь не загоняет задачу в цикл проверки состояния ножки МК, а проверяет ее лишь 1 раз за вызов задачи, отдавая процессорное время другим задачам во время фазы пассивного ожидания. Несмотря на то что, задача пассивно ждет в заблокированном состоянии (работа дальше не продолжается), сам метод ожидания остается синхронным – результат ожидается сразу. Этот вариант лучше предыдущего в любом случае, но именно синхронность данной операции остается проблемой, т.к. отслеживание готовности преобразования происходит только во время вызова задачи – ухудшение отзывчивости.
Метод чтения теперь лишь запускает прием, работа задачи продолжается (до состояния ожидания приема), т.е. метод неблокирующий. Единственный поток свободен от побайтового приема данных по SPI и, помимо этой задачи, может уделить время другим задачам – улучшение отзывчивость.
Вопрос возникает в работе метода ожидания приема данных по DMA, в реализации spi_dma_read_cmplt_is( ).
Классический способ будет иметь вид:
bool spi_dma_read_cmplt_is( )
{
bool is_read_cmplt = false;
if( ( reg32_read( DMA_CH_SPI_RX_CTRL_REG_ADDR ) & DMA_CTRL_REG_XFER_CMPLT ) != 0U )
{ is_read_cmplt = true; }
return is_read_cmplt;
}Данный метод является блокирующим (задача ждет завершения приема и не выполняет работу дальше), а ожидание аналогично методам ожидания и чтения преобразования – пассивное ожидание синхронной операции.
Новая архитектура задачи эффективно использует процессорное время – работает только в фазу активных действий (однократными проверками можно пренебречь), но отзывчивость такой задачи будет зависеть от реализации многозадачности и от других задач в том числе.
Приведенный пример характерен для классической кооперативной многозадачности без планировщика – суперцикл. Поэтому, для улучшения отзывчивости системы, в частности задачи работы с АЦП, необходимо добавить нотки вытеснения.
Для методов ожидания готовности преобразования и ожидания завершения приема данных по SPI можно использовать аппаратные прерывания, предоставляемые разработчиком МК.
Готовность преобразования представляет собой отрицательный фронт дискретного сигнала, что можно использовать через внешние прерывания (EXTI IRQ), а завершение приема данных – прерывание канала DMA.
Для корректной интеграции асинхронных (результат ожидается не сразу) операций по отслеживанию готовности преобразования и завершению приема данных, вполне может быть достаточно вызывать задачу работы с АЦП из обработчиков соответствующих прерываний:
void exti_isr( void )
{
adc_task( );
exti_handler( );
}
void dma_ch_isr( void )
{
adc_task( );
dma_ch_handler( );
}В таком случае, возможно, не придется адаптировать код задачи под асинхронный вызов.
Особенности DMA в микроконтроллерах.
Стоит также упомянуть про некоторые особенности DMA в микроконтроллерах. Периферия МК должна быть аппаратно готова к работе с DMA, т.к. канал DMA и сама периферия должны уметь управлять потоком данных без вмешательства процессорного ядра. На практике, это достаточно сложная структуру различных сигналов, линий синхронизации и т.д. Поэтому, несмотря на то что, периферия МК может лежать в адресном пространстве DMA, DMA доступ может быть не реализован.
Из вышесказанного можно вынести то, что DMA умеет управлять периферией, что может использоваться как трюк – реализовывать прием-передачу через DMA даже когда он и не нужен, опуская работу с, в некоторых случаях, сложной последовательностью побайтового (polling) приема-передачи.
Также важно помнить, что DMA работает именно с памятью, что критично для транзакций mem-pheriperal. Большинство периферийных блоков интерфейсов передачи данных, в частности последовательных, имеют сложную структуру регистров данных. Например, помимо регистра данных, куда имеет доступ DMA (а на практике это может быть еще и FIFO-буфер), есть еще сдвиговый регистр, который выставляет состояния в физическую шину. В таком случае, когда DMA сообщит о завершении транзакции, фактически транзакция еще не будет завершена – сдвиговый регистр еще не закончил выставлять данные в физическую шину. Тогда корректный способ отслеживания завершения транзакции может выглядеть так:
bool spi_dma_transmit_cmplt_is( )
{
bool is_transmit_cmplt = false;
if( ( reg32_read( DMA_CH_SPI_TX_CTRL_REG_ADDR ) & DMA_CTRL_REG_XFER_CMPLT ) != 0U
&& ( reg32_read( SPI_CTRL_REG_ADDR ) & SPI_CTRL_REG_TRANSMIT_CMPLT ) != 0U )
{ is_transmit_cmplt = true; }
return is_transmit_cmplt;
}Однако, такой способ уже не сработает в случае асинхронного получения результата, т.к. прерывание DMA канала сработает 1 раз, до фактического завершения передачи (поднятие флага SPI_CTRL_REG_TRANSMIT_CMPLT). В таком случае необходимо искать МК с последовательными интерфейсами передачи данных, которые умеют отслеживать пакетные транзакции или отслеживать завершение по времени – аппаратным таймером с прерыванием.
Помимо классических вариантов использования DMA (работа с периферией), можно использовать еще один трюк. DMA умеет работать в режиме mem-mem, что позволяет без вмешательства процессорного ядра, например, скопировать или заполнить паттерном определенную область памяти МК. Это позволяет реализовывать асинхронные, а в некоторых случаях неблокирующие операции и методы.
Стоит понимать, что силы DMA небезграничные, контроллер DMA имеет конечное число каналов, которые могут нагружать шину данных при одновременной работе, что может привести к задержкам.
Улучшение эффективного потребления электроэнергии за счёт эффективного использования процессорного времени.
Интегрирование FSM позволило декомпозировать задачу на составляющие – машинные состояния, что дает полное понимание о ее работе, а в данном контексте, о том, когда задача не выполняет активных действий (заблокирована). Заблокированная задача, при использовании планировщиков, может вовсе не тратить процессорного времени. Тогда встает вопрос, что делать процессорному ядру?
Для систем с батарейным питанием это очень важный вопрос, т.к. одной из задач таких систем является длительная работа на одном заряде батареи. В этом случае система должна переводиться в состояние пониженного энергопотребления (спать) в моменты, когда это возможно. А когда это возможно помогут узнать машинные состояния задач:
#define ADC_FSM_S_SLEEP_POSSIBLE ( ADC_FSM_S_CONVERSION_WAIT_READ | ADC_FSM_S_CONVERSION_READ_WAIT)
#define TASK1_FSM_S_SLEEP_POSSIBLE ( TASK1_FSM_S_X_WAIT )
#define TASK2_FSM_S_SLEEP_POSSIBLE ( TASK2_FSM_S_X_WAIT )
#define ADC_FSM_S_SLEEP_POSSIBLE_CHECK ( ( adc_tcb.stateMachine & ADC_FSM_S_SLEEP_POSSIBLE ) != 0U )
#define TASK1_FSM_S_SLEEP_POSSIBLE_CHECK ( ( task1_tcb.stateMachine & TASK1_FSM_S_SLEEP_POSSIBLE ) != 0U )
#define TASK2_FSM_S_SLEEP_POSSIBLE_CHECK ( ( task2_tcb.stateMachine & TASK2_FSM_S_SLEEP_POSSIBLE ) != 0U )
void sleep_enter_task( void )
{
if( ADC_FSM_S_SLEEP_POSSIBLE_CHECK
&& TASK1_FSM_S_SLEEP_POSSIBLE_CHECK
&& TASK2_FSM_S_SLEEP_POSSIBLE_CHECK )
{
sleep( );
}
}А проснуться система сможет по прерываниям, например, аппаратный таймер или внешний дискретный сигнал (EXTI).
По традиции оставшиеся свои мысли расскажу в следующей части. Спасибо за внимание, услышимся!
KioskNews shows a cleaned-up reading view extracted from the publisher’s page — the original always lives on their site, not ours.