В повседневной практике разработчика встраиваемых систем приходится сталкиваться с необходимостью запуска двух и более разноплановых ОС на n-ядерных системах на кристалле. Это, как правило, Linux и специализированная RTOS. На плечи Linux ложится работа с тяжеловесными стеками протоколов, а RTOS же занимается задачами реального времени.
Одна из основных задач, которая встает при такой организации системы — обеспечение механизма взаимодействия, то есть межъядерный обмен данными. Если вам интересно узнать один из вариантов решения на базе открытой библиотеки OpenMCAPI, пролистать пару десятков строк программного кода и увидеть реальные цифры пропускной способности при использовании этой библиотеки, добро пожаловать под кат.
Задача межъядерного обмена данными успешно решается за счет использования разделяемой памяти и межъядерных прерываний с написанием своей прослойки взаимодействия и портированием ее на различные ОС. Для приведения такого API к стандартизированному виду Multicore Association (MCA) разработала и выпустила в свет первую версию спецификации MCAPI (Multicore Communications API), вскоре была выпущена и вторая версия.
Рассматриваемая библиотека OpenMCAPI основана на спецификации MCAPI 2.0, разработана компанией Mentor Graphics Corporation и имеет открытый исходный код под свободной лицензией BSD/GPL. Исходные коды можно получить, воспользовавшись сайтом проекта, там же находится краткая информация по запуску и портированию.
Библиотека OpenMCAPI изначально предоставляет возможность работы под управлением ОС Linux с использованием виртуального транспорта либо разделяемой памяти (но только на платформах mpc85xx и mv78xx0).
Предлагаемая структура взаимодействия Linux и RTOS через OpenMCAPI c разделением на абстрактные уровни имеет следующий вид (см. рис 1):
Рисунок. 1. Структура взаимодействия Linux и RTOS через OpenMCAPI
Рассмотрим реализацию приведенной структуры на примерах исходного кода порта для Linux:
- MCAPI Generic — реализация внешнего MCAPI API.
- OS Layer — часть уровня MCAPI Generic, который содержит код, зависящий от операционной системы. Эта часть представлена в файле libmcapi/mcapi/linux/mcapi_os.c и содержит реализацию:
- Transport Generic — это уровень абстракции, который предоставляет механизм для работы с разделяемой памятью на уровне пользовательского пространства (userspace). Он представлен файлами libmcapi/shm/shm.c и libmcapi/shm/linux/shm_os.c и содержит реализацию:
- OS Specific Driver представлен модулем ядра Linux, обеспечивающим прямой доступ к оборудованию из пользовательского пространства. Модуль находится в папке libmcapi/shm/linux/kmod и содержит реализацию:
Для полного понимания механизма взаимодействия через транспорт, использующий разделяемую память, который реализован в библиотеке OpenMCAPI, необходимо рассмотреть механизм межъядерного сигнализирования и структуру данных в разделяемой памяти.
Дальнейшее рассмотрение будет проводиться на основе платформы mpc85xx (чип P1020 компании Freescale). Программное обеспечение: ядро Linux версии 2.6.35 с патчами, которое поставляется с комплектом средств разработки (SDK) Freescale QorIQ_SDK_V1_03 (доступен для скачивания после регистрации на их сайте), в качестве операционной системы реального времени (RTOS) использована RTEMS, исходные коды которой можно получить в git-репозиторие по ссылке git://git.rtems.org/rtems.git.
Для реализации межъядерного сигнализирования Freescale предоставляет как минимум два механизма:
- Interprocessor Interrupts (IPIs) — межъядерные прерывания, до 4 штук с поддержкой мультикастовых прерываний.
- Message Interrupts (MSGRs) — межъядерные 32-битные сообщения с генерированием прерывания при записи сообщения в регистр, до 8 штук.
В библиотеке оpenmcapi при реализации OS Specific Driver для этой платформы используется механизм MSGRs.
Рассмотрим структуру данных, содержащуюся в разделяемой памяти (см. рис. 2):
Рисунок 2. Структура данных в разделяемой памяти
Область разделяемой памяти по использованию пространству можно разбить на два блока:
- Первый блок — область SHM_MGMT_BLOCK, представлена структурой:
/* SM driver mamagement block */
struct _shm_drv_mgmt_struct_
{
shm_lock shm_init_lock;
mcapi_uint32_t shm_init_field;
struct _shm_route_ shm_routes[CONFIG_SHM_NR_NODES];
struct _shm_buff_desc_q_ shm_queues[CONFIG_SHM_NR_NODES];
struct _shm_buff_mgmt_blk_ shm_buff_mgmt_blk;
};
Структура содержит следующие элементы:
- Глобальная блокировка разделяемой памяти — shm_init_lock, используемая для разграничения доступа n-ядер к разделяемой (shared) области.
- Переменная shm_init_field содержит ключ окончания инициализации мастера, принимает значение SHM_INIT_COMPLETE_KEY по окончанию инициализации.
- Shm_routes — таблица маршрутизации со связями межъядерных сообщений, содержит CONFIG_SHM_NR_NODES связей по числу участвующих в обмене ядер (узлов). В нашем случае 2 узла.
- Shm_queues — очереди сообщений с привязкой к конкретному узлу, содержит CONFIG_SHM_NR_NODES. В нашем случае 2 очереди.
- Shm_buff_mgmt_blk — структура управления буферами (SHM_BUFFER) в области данных.
- Вторая область — область данных, содержит SHM_BUFF_COUNT (по умолчанию 128) структур SHM_BUFFER. Эта область служит непосредственно для хранения передаваемых данных. Структура SHM_BUFFER состоит из массива размером MCAPI_MAX_DATA_LEN и дополнительной структуры управления элементом.
Перед тем как рассмотреть процесс портирования необходимо привести блок-схему работы низкоуровневого механизма коммуникации через разделяемую память (механизм реализован в OpenMCAPI, см. рис. 3):
Рисунок. 3. SDL-диаграммы низкоуровневого механизма коммуникации через разделяемую память (реализован в OpenMCAPI).
Некоторые пояснения к диаграммам:
- “HW-Notification” — диаграмма описывает процесс отсылки уведомления удаленному или текущему ядру:
- Вызов функции принимает id ядра, для которого предназначено сообщение (функция openmcapi_shm_notify).
- Если уведомление предназначено для удаленного ядра “target id”, то генерируется удаленное сообщение через механизм MSGRs (рассмотрен выше) со значением в поле данных равным единице (см. блок 3), иначе происходит явный вызов обработчика прерывания interrupt_handle (диаграмма “HW-Receive”), см. блок 4.
- “HW-Receive” — диаграмма описывает процесс приема уведомления от удаленного или текущего ядра:
- Interrupt_handle является обработчиком прерывания, настроенным на срабатывание при приеме сообщения по MSGRs, используется также для явного вызова.
- В блоках 2—6 организована проверка статуса всех сообщений MSGRs, если поле данных MSGR-сообщения не равно 0, происходит разблокировка потока, который производит обработку данных в области разделяемой памяти.
- В блоках 7—8 происходит проверка места вызова обработчика прерывания (“interrupt_handler”). Если вызов был в прерывании, сбрасывается флаг присутствия сообщения MSGR.
Перед тем как перейти к описанию портирования для RTEMS, вкратце рассмотрим эту ОС.
RTEMS (Real-Time Executive for Multiprocessor Systems) — это RTOS c открытым исходным кодом, полнофункциональная операционная система реального времени с поддержкой множества открытых стандартных интерфейсов прикладного программирования (API), стандартов POSIX и BSD-сокетов. Она предназначена для использования в космических, медицинских, сетевых и многих других встраиваемых устройствах. RTEMS содержит широкий набор процессорных архитектур, таких как ARM, PowerPC, Intel, Blackfin, MIPS, Microblaze и др. Содержит большой стек реализованных сетевых протоколов, в частности tcp/ip, http, ftp, telnet. Предоставляет стандартизированный доступ к RTC, NAND, UART и другому оборудованию.
Перейдем к процессу портирования OpenMCAPI. Исходя из документа, расположенного по ссылке [1] требуется:
- Реализовать OS Layer, файлы:
- libmcapi/mcapi/rtems/mcapi_os.c;
- libmcapi /include/rtems/mgc_mcapi_impl_os.h.
- Реализовать поддержку совместимого транспорта разделяемой памяти, файл:
- libmcapi/shm/rtems/shm_os.c.
- Добавить рецепты в waf-сборщик, который используется для OpenMCAPI.
Так как целевая платформа P1020 (powerpc, 500v2) и портирование проводилось на RTOS, где допускается отсутствие разделения пространства kernel/user space, отпадает необходимость в написании:
- libmcapi/include/arch/powerpc/atomic.h;
- libmcapi/shm/rtems/kmod/.
Также отпадает необходимость в реализации OS Layer, так как RTEMS поддерживает POSIX-совместимые вызовы, файлы mcapi_os.c и mgc_mcapi_impl_os.h просто были скопированы из реализации для Linux.
Реализация транспорта разделяемой памяти полностью выполнена в файле shm_os.c и включает адаптацию вызовов из уровня абстракции Transport Generic (файл libmcapi/shm/shm.c) и реализацию механизма обмена через MSGRs.
Функции, требующие реализации:
1) mcapi_status_t openmcapi_shm_notify (mcapi_uint32_t unit_id, mcapi_uint32_t node_id) — функция отправляет нотификацию удаленному ядру(ам), реализация представлена диаграммой (см. рис 3). Исходный код приведен ниже:
/* send notify remote core */
mcapi_status_t openmcapi_shm_notify(mcapi_uint32_t unit_id,
mcapi_uint32_t node_id)
{
mcapi_status_t mcapi_status = MCAPI_SUCCESS;
int rc;
rc = shm_rtems_notify(unit_id);
if (rc) {
mcapi_status = MGC_MCAPI_ERR_NOT_CONNECTED;
}
return mcapi_status;
}
static inline int shm_rtems_notify(const mcomm_core_t target_core)
{
struct mcomm_qoriq_data *const data = &mcomm_qoriq_data;
/* If the target is the local core, call the interrupt handler directly. */
if (target_core == mcomm_qoriq_cpuid()) {
_mcomm_interrupt_handler(NO_IRQ, data);
} else {
mcomm_qoriq_notify(target_core);
}
return 0;
}
/* Wake up the process(es) corresponding to the mailbox(es) which just received
* packets. */
static int _mcomm_interrupt_handler(rtems_vector_number irq, struct mcomm_qoriq_data *data)
{
register int i;
void *mbox = data->mbox_mapped;
for (i = 0; i < data->nr_mboxes; i++) {
int active;
switch (data->mbox_size) {
case 1:
active = readb(mbox);
break;
case 4:
active = readl(mbox);
break;
default:
active = 0;
}
if (active) {
LOG_DEBUG("%s: waking mbox %d\n", __func__, i);
(void) rtems_event_send( data->rid, MMCAPI_RX_PENDING_EVENT );
}
mbox += data->mbox_stride;
}
if (irq != NO_IRQ) {
mcomm_qoriq_ack();
}
return 0;
}
2) mcapi_uint32_t openmcapi_shm_schedunitid(void) — функция возвращает номер текущего ядра (то есть ядра, исполняющего этот код), реализуется тривиально чтением регистра процессора. Исходный код приведен ниже:
/* Get current cpu id */
mcapi_uint32_t openmcapi_shm_schedunitid(void)
{
return (mcapi_uint32_t) ppc_processor_id();
}
3) mcapi_status_t openmcapi_shm_os_init(void) — функция создает и запускает низкоуровневый поток приема данных, реализуется посредством вызова функций rtems_task_create и rtems_task_start. Исходный код приведен ниже:
/* Now that SM_Mgmt_Blk has been initialized, we can start the RX thread. */
mcapi_status_t openmcapi_shm_os_init(void)
{
struct mcomm_qoriq_data *const data = &mcomm_qoriq_data;
rtems_id id;
rtems_status_code sc;
if( RTEMS_SELF != data->rid ) {
return MCAPI_ERR_GENERAL;
}
sc = rtems_task_create(
rtems_build_name( 'S', 'M', 'C', 'A' ),
MMCAPI_RX_TASK_PRIORITY,
RTEMS_MINIMUM_STACK_SIZE,
RTEMS_DEFAULT_MODES,
RTEMS_DEFAULT_ATTRIBUTES,
&id);
if( RTEMS_SUCCESSFUL != sc ) {
return MCAPI_ERR_GENERAL;
}
/* global save task id */
data->rid = id;
sc = rtems_task_start( id, mcapi_receive_thread, 0 );
if( RTEMS_SUCCESSFUL != sc ) {
perror( "rtems_task_start\n" );
return MCAPI_ERR_GENERAL;
};
return MCAPI_SUCCESS;
}
static rtems_task mcapi_receive_thread(rtems_task_argument argument)
{
int rc;
do {
rc = shm_rtems_wait_notify(MCAPI_Node_ID);
if (rc < 0) {
perror("shm_rtems_wait_notify");
break;
}
MCAPI_Lock_RX_Queue();
/* Process the incoming data. */
shm_poll();
MCAPI_Unlock_RX_Queue(0);
} while (1);
printk("%s exiting!\n", __func__);
}
static inline int shm_rtems_wait_notify(const mcapi_uint32_t unitId)
{
rtems_event_set event_out;
int ret = 0;
while(1) {
LOG_DEBUG("mcomm_mbox_pending start\n");
(void) rtems_event_receive(
MMCAPI_RX_PENDING_EVENT,
RTEMS_DEFAULT_OPTIONS,
RTEMS_NO_TIMEOUT,
&event_out
);
LOG_DEBUG("rtems_event_receive\n");
ret = mcomm_mbox_pending(&mcomm_qoriq_data,
(mcomm_mbox_t)unitId);
LOG_DEBUG("mcomm_mbox_pending end ret=%d\n", ret);
if(ret != 0) {
return ret;
};
}
return 0;
}
4) mcapi_status_t openmcapi_shm_os_finalize(void) — функция останавливает низкоуровневый поток приема данных, реализуется посредством вызова функции rtems_task_delete. Исходный код приведен ниже:
/* Finalize the SM driver OS specific layer. */
mcapi_status_t openmcapi_shm_os_finalize(void)
{
struct mcomm_qoriq_data *const data = &mcomm_qoriq_data;
rtems_id id = data->rid;
rtems_status_code sc;
sc = rtems_task_delete(id);
if( RTEMS_SUCCESSFUL != sc ) {
return MCAPI_ERR_GENERAL;
}
return MCAPI_SUCCESS;
}
5) void *openmcapi_shm_map(void) — функция подготовки и настройка интерфейса MSGRs, подготовка разделяемой памяти. Исходный код приведен ниже:
/* full open mcom device and get memory map addres*/
void *openmcapi_shm_map(void)
{
void *shm;
int rc;
size_t shm_bytes;
// low level init //
mcomm_qiroq_probe();
shm_bytes = shm_rtems_read_size();
if (shm_bytes <= 0) {
perror("read shared memory size\n");
return NULL;
}
/* initialized device. */
rc = shm_rtems_init_device();
if (rc < 0) {
perror("couldn't initialize device\n");
goto out;
}
shm = shm_rtems_read_addr();
if (shm == NULL) {
perror("mmap shared memory");
goto out;
}
return shm;
out:
return NULL;
}
static size_t shm_rtems_read_size(void)
{
struct mcomm_qoriq_data *const data = &mcomm_qoriq_data;
return (size_t) (data->mem.end - data->mem.start);
}
static inline int shm_rtems_init_device(void)
{
struct _shm_drv_mgmt_struct_ *mgmt = NULL; /* xmmm */
return mcomm_dev_initialize(&mcomm_qoriq_data,
(uint32_t)&mgmt->shm_queues[0].count,
CONFIG_SHM_NR_NODES,
sizeof(mgmt->shm_queues[0].count),
((void *)&mgmt->shm_queues[1].count - (void *)&mgmt->shm_queues[0].count));
}
static void *shm_rtems_read_addr(void)
{
struct mcomm_qoriq_data *const data = &mcomm_qoriq_data;
return (void*)data->mem.start;
}
6. void openmcapi_shm_unmap(void *shm) — функция закрывает интерфейс MSGRs, отменяет использование разделяемой памяти. Исходный код приведен ниже:
/* full close mcom device and revert memory */
void openmcapi_shm_unmap(void *shm)
{
/* deinitialized device. */
shm_rtems_deinit_device();
// low level deinit //
mcomm_qoriq_remove();
}
static inline int shm_rtems_deinit_device(void)
{
return mcomm_dev_finalize(&mcomm_qoriq_data);
}
Отдельно следует рассмотреть реализацию функции низкоуровневого потока приема mcapi_receive_thread (исходный код см. выше). При запуске потока вызовом функции rtems_event_receive он переводится в режим ожидания события (реализуется доступным в RTEMS механизмом событий). Далее при приходе события запуска, отсылаемого в обработчике interrupt_handler (см. рис. 3, диаграмма “HW-Receive”), происходит обработка изменений в области разделяемой памяти (вызов внутренней функции openmcapi — shm_poll()), c ее предварительной блокировкой, после чего поток возвращается в состояние ожидания.
Ниже приводятся результаты, полученные при взаимодействии Linux и RTEMS через OpenMCAPI. Тестовый стенд представляет собой отладочная плата от Freescale P1020RDB-PB с установленным процессором P1020 (2 ядра). Частоты: частота ядра — 800 МГц, DDR2 — 400 МГц, CCB — 400 МГц. На ядрах 0/1 были запущены соответственно Linux/RTEMS. Обмен был двухсторонним, замерялось время, затраченное на 10000 двухсторонних посылок. Результаты тестов сведены в таблицу:
| № |
Описание теста |
Времени на одну посылку, мкс |
| 1 |
Симметричные пакеты размером 512 байт |
37,5 |
| 2 |
Симметричные пакеты размером 52430 байт |
121 |
| 3 |
Симметричные пакеты размером 100 кБайт |
346 |
| 4 |
Ассиметричные пакеты размерами 1к/100к-linux/rtems |
185 |
Из всего выше изложенного можно сделать вывод, что библиотека OpenMCAPI предоставляет собой достойный вариант реализации спецификации MCAPI, имеющей четкую структуру исходного кода, облегчающую портирование; наглядные примеры портирования (платформы powerpc и arm); свободную лицензию и производительность, достаточную для большинства применений.
[!?] Вопросы и комментарии приветствуются. На них будет отвечать автор статьи Руслан Филипович, программист дизайн-центра электроники Promwad.
