Шины сообщений (магистрали). Черновой вариант API

Действующие объекты

Потоки - отправители и получатели сообщений. Для потоков-получателей ядро ведет очереди сообщений (в очередях находятся указатели на сообщения)
Кэш страниц памяти, используемой при передаче сообщений. Когда нужно передать сообщение, память для него выбирается из этого кэша. Соответственно, при освобождении сообщения, память возвращается в кэш
Список выделенных сообщений. Каждое сообщение является объектом ядра и имеет список потоков, которые владеют сообщением
Список шин сообщений. Каждая шина сообщений (магистраль) имеет список потоков, зарегистрированных на ней

Функции

Несколько упрощая, всю работу IPC можно описать с помощью такого API:

int RegisterMessageBus(const char *bus_name); 
void UnregisterMessageBus(int bus_id); 
int EnumMessageBus(char *buffer, int buffer_size);

int RegisterMessageFilter(int bus_id, int *msg_types, int count);

void* AllocMessage(int size); void FreeMessage(void *ptr_msg); 

void SendMessage(int bus_id, int type, void *ptr_msg); 
void* GetMessage(int &msg_type);

Далее приводится упрощенное описание реализации вышеприведенных функций. Описание в двух вариантах: со стороны пользовательского процесса (USER) и со стороны ядра (KERNEL).

Реализация

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

Регистрация шины сообщений

int RegisterMessageBus(const char *bus_name);

USER

Функция регистрирует поток на магистрали bus_name. bus_name - наименование магистрали. Функция возвращает идентификатор, который фигурирует во всех остальных обращениях к магистрали.

KERNEL

Ядро проходит по списку своих объектов типа "магистраль" в поисках объекта с наименованием bus_name. Если такой объект не найден, создается новый объект типа "магистраль" с именем bus_name
Указатель на объект ядра "поток" добавляется к списку зарегистрированных потоков

Освобождение ресурсов

Если зарегистрированный поток по каким-то причинам не дерегистрировался перед завершением (core dumped или просто программист забыл), ядро выполняет освобождение ресурсов. В качестве идентификатора ядро может возвращать указатель на свой объект "магистраль" (из соображений скорости доступа), но это может привести к большой нестабильности ядра. Лучше возвращать искусственный id, который ссылается на объект ядра "магистраль"

KERNEL

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

Дерегистрация шины сообщений

void UnregisterMessageBus(int bus_id);

USER

Поток снимается с регистрации на магистрали bus_id. Поток больше не будет получать сообщения с этой шины.

KERNEL

Ядро находит магистраль bus_id и если поток зарегистрирован в этой магистрали удаляет его фильтр сообщений. Если поток был единственным зарегистрированным на магистрали, то удаляется и магистраль. При этом сообщения, полученные с этой магистрали и находящиеся в очереди потока не удаляются.

Получение списка магистралей

int EnumMessageBus(char *buffer, int buffer_size);

USER

Функция формирует блок памяти (в buffer) такого формата:

"bus_name"\0"bus_name"\0..."bus_name"\0\0
buffer_size - размер буфера

Если размер буфера достаточен для хранения всего списка магистралей, функция возвращает 0. В противном случае возвращается необходимый размер буфера.

KERNEL

Ядро проходит по списку объектов типа "магистраль" и заносит в buffer информацию о магистралях.

Следующие две функции управляют распределением памяти сообщений.

Создать сообщение

void* AllocMessage(int size);

USER

Функция возвращает указатель на начало сообщения размером size в адресном пространстве потока. Если не удалось создать сообщение, возвращается 0.

KERNEL

Ядро пытается выделить из кэша страниц память для размещения сообщения размером size. Если это сделать невозможно, возвращается 0
Ядро пытается найти в адресном пространстве потока незанятый регион размером не менее size. Если это сделать невозможно, возвращается 0
Создается объект ядра "сообщение", и в список его потоков-владельцев добавляется текущий поток
Ядро отображает выделенную память в найденный регион с атрибутами RW и возвращает виртуальный адрес региона

Освободить сообщение

void FreeMessage(void *ptr_msg);

USER

Выделенное ранее сообщение удаляется.

KERNEL

В адресном пространстве ядро потока ищет регион, к которому относится адрес ptr_msg. Если этот адрес не принадлежит сообщению или текущий поток не является владельцем сообщения, производится выход из функции (хотя можно и "выбросить" поток)
Регион, который занимало сообщение, удаляется из адресного пространства потока
Из списка потоков-владельцев объекта ядра "сообщение" удаляется текущий поток. Если поток был единственным владельцем сообщения, то удаляется и само сообщение, а занимаемая им память возвращается в кэш

Далее 3 функции управляющие приемом/передачей сообщений.

Зарегистрировать фильтр сообщений

int RegisterMessageFilter(int bus_id, int *msg_types, int count);

USER

Функция устанавливает на шине bus_id фильтр сообщений для текущего потока. Потоку будут доставляться сообщения, типы которых указаны в фильтре.

   msg_types - массив типов сообщений 
   count - количество сообщений в фильтре

В случае удачного завершения возвращается 0. В противном случае (например, если поток не зарегистрирован на магистрали) возвращается -1.

KERNEL

Ядро в списке шин ищет магистраль с идентификатором bus_id. Если такой магистрали не существует или поток в ней не зарегистрирован, возвращается -1
Типы сообщений из msg_types добавляются в фильтр магистрали для текущего потока

Отправить сообщение

void SendMessage(int bus_id, int type, void *ptr_msg);

USER

Функция отправляет сообщение ptr_msg типа type на магистраль bus_id.

KERNEL

Проверяется:

принадлежность адреса ptr_msg корректному региону адресного пространства потока
принадлежность региона сообщению
наличие у региона атрибутов RW
наличие прав владельца у текущего потока по отношению к сообщению

Если хотя бы одно из этих условий не выполняется, отправка сообщения не производится (может имеет смысл завершить поток аварийно).

Атрибуты региона содержащего сообщение заменяются на RO. Это необходимо по двум причинам:

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

В принципе, можно ввести функцию повторного "быстрого" выделения сообщения, когда атрибуты снова меняются на RW, единственным условием является наличие только одного владельца сообщения.

Ядро проходит по потокам, зарегистрированным на магистрали и сравнивает тип передаваемого сообщения (type) с их фильтрами. Если сообщение можно передать потоку-получателю, выполняются такие действия:

в адресном пространстве потока-получателя отыскивается незанятый регион способный вместить сообщение
страницы сообщения отображаются на адресное пространство потока-получателя с атрибутом RO
поток получатель вносится в список владельцев сообщения
указатель на сообщение помещается в очередь сообщений потока получателя
если поток получатель приостановлен (вынесен из списка планировщика), он планируется на выполнение

Получить сообщение

void* GetMessage(int &msg_type);

USER

Возвращает указатель на первое выбранное из очереди сообщение. Тип сообщения сохраняется в msg_type. Если очередь сообщений пуста, поток будет приостановлен.

KERNEL

Ядро пытается выбрать первое сообщение из очереди потока. Если очередь пуста, поток выносится из списка потоков планировщика.