В повседневной практике разработчика встраиваемых систем приходится сталкиваться с необходимостью запуска двух и более разноплановых операционных систем на 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 %dn", __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_startn" );
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 startn");
(void) rtems_event_receive(
MMCAPI_RX_PENDING_EVENT,
RTEMS_DEFAULT_OPTIONS,
RTEMS_NO_TIMEOUT,
&event_out
);
LOG_DEBUG("rtems_event_receiven");
ret = mcomm_mbox_pending(&mcomm_qoriq_data,
(mcomm_mbox_t)unitId);
LOG_DEBUG("mcomm_mbox_pending end ret=%dn", 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 sizen");
return NULL;
}
/* initialized device. */
rc = shm_rtems_init_device();
if (rc < 0) {
perror("couldn't initialize devicen");
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.
Автор: Promwad
Очень интересная статья!