7.3.4 Очереди сообщений POSIX

Очередь сообщений POSIX 1003.1b обеспечивает детерминированные и эффективные средства IPC. Она предлагает следующие преимущества для приложений реального времени:

▪Буферы сообщений в очереди сообщений являются предварительно выделенными, обеспечивая доступность ресурсов, когда они необходимы.

▪Сообщениям могут быть назначены приоритеты. Высокоприоритетные сообщения всегда принимаются первыми, независимо от количества сообщений в очереди.

▪Она предлагает асинхронное уведомление при поступлении сообщения, если приёмник не хочет ожидать получения сообщения.

▪Функции отправки и получения сообщений по умолчанию являются блокирующими вызовами. Приложения могут указать время ожидания при отправке или получении сообщений, чтобы избежать недетерминированной блокировки.

Интерфейсы перечислены в Таблице 7.5. Использование некоторых основных функций очереди сообщений иллюстрирует Распечатка 7.8. В этом примере создаются два процесса: один отправляет сообщение в очередь сообщений, а другой получает сообщение из очереди.

Таблица 7.5 Функции для работы с очередью сообщений POSIX.1b

Компиляция и запуск приведённых выше двух программ даёт следующий результат:

# gcc –o mqueue-1 mqueue-1.c –lrt

# gcc –o mqueue-2 mqueue-2.c –lrt

# ./mqueue-1

# ./mqueue-2

O_NONBLOCK not set

Message: Hello Posix World, prio = 1

Временем блокировки приложения для отправки или получения сообщений можно управлять с помощью функций mq_timedsend и mq_timedreceive. Если очередь сообщений заполнена и флаг O_NONBLOCK не установлен, функция mq_timedsend завершается за указанное время (это может произойти, если очередь заполнена и функция отправки блокируется, пока не получит свободный буфер). Точно так же mq_timedreceive завершается за указанное время, если в очереди нет сообщений. Использование функций mq_timedsend и mq_timedreceive иллюстрирует следующий фрагмент кода. Обе ожидают не более 10-ти секунд для отправки или получения сообщений.

/* Отправка сообщения */

struct timespec ts;

/* С этого момента указывает время ожидания как 10 с. */

ts.tv_sec = time(NULL) + 10;

ts.tv_nsec = 0;

if (mq_timedsend(ds, text, SIZE,PRIOTITY, &ts) == -1){

  if (errno == ETIMEDOUT){

    printf("Timeout when waiting for message.");

    return 0;

  }

  return -1;

}

/* Приём сообщения */

if (mq_timedreceive(ds, new_text, SIZE, &prio, &ts) == -1){

  if (errno == ETIMEDOUT){

    printf("Timeout when waiting for message.");

    return 0;

  }

  return -1;

}

Асинхронные уведомления

Асинхронный механизм для процессов для получения уведомления, что в очереди сообщений есть сообщения, вместо синхронной блокировки в mq_receive или mq_timedreceive, обеспечивает функция mq_notify. Этот интерфейс очень удобен для приложений реального времени. Процесс может вызвать функцию mq_notify, чтобы зарегистрироваться для асинхронных уведомлений, а затем может выполнять другую работу. Когда приходит сообщение, в очередь процессу направляется уведомление. После уведомления, для получения сообщения процесс может вызвать mq_receive. Прототип mq_notify:

int mq_notify(mqd_t mqdes,

              const struct sigevent *notification);

Приложение может зарегистрироваться на два типа уведомлений:

▪SIGEV_SIGNAL: отправить процессу сигнал, указанный в notification->sigev_signo, когда в очередь приходит сообщение. Использование иллюстрирует Распечатка 7.9.

▪SIGEV_THREAD: вызвать при поступлении сообщения в очередь в отдельном потоке notification->sigev_notify_function. Использование иллюстрирует Распечатка 7.10.

Реализация в Linux

Подобно реализации в Linux совместно используемой памяти POSIX, Linux реализует очереди сообщений POSIX как файловую систему mqueue. Файловая система mqueue обеспечивает необходимую поддержку ядра для библиотеки пользовательского пространства, которая реализует интерфейсы очереди сообщений POSIX. По умолчанию ядро монтирует файловую систему внутренне и её не видно в пространстве пользователя. Тем не менее, вы можете смонтировать файловую систему mqueue.

# mkdir /dev/mqueue

# mount -t mqueue none /dev/mqueue

Эта команда монтирует файловую систему mqueue в /dev/mqueue. Очередь сообщений представлена в виде файла в /dev/mqueue. Но вы не можете отправить или получить сообщение из очереди "записывая" или "читая" из "файла" очереди сообщений. Чтение файла даёт размер очереди и уведомительную информацию, которая не доступна через стандартные процедуры. Удалите из Распечатки 7.8 mq_unlink, а затем скомпилируйте и запустите.

# gcc mqueue-1.c -lrt

# ./a.out

# cat /dev/mqueue/my_queue

QSIZE:17         NOTIFY:0    SIGNO:0    NOTIFY_PID:0

В выводе, показанном выше:

▪QSIZE: размер очереди сообщений

▪NOTIFY: либо 0, либо SIGEV_SIGNAL или SIGEV_THREAD

▪SIGNAL: номер сигнала, используемого для уведомления

▪NOTIFY_PID: PID процесса, ожидающего уведомление

Для регулировки количества ресурсов, используемых файловой системой, файловая система mqueue также предоставляет управление параметрами (sysctls) в каталоге /proc/sys/fs/mqueue. Этими параметрами являются:

▪queues_max: чтение/запись этого файла позволяет получить/установить максимальное количество очередей сообщений, разрешённых в системе. Например, echo 128 > queues_max позволяет создание максимум 128 очередей сообщений в системе.

▪msg_max: чтение/запись этого файла позволяет получить/установить максимальное количество сообщений в очереди.Максимальное количество сообщений, указанное при вызове mq_open, должно быть меньше или равно msg_max.

▪msgsize_max: чтение/запись файла позволяет получить/установить максимальное значение размера сообщения. Оно является значением по умолчанию, если во время вызова mq_open максимальный размер сообщения не задан.

Что следует помнить

▪Когда процесс получает из очереди сообщение, это сообщение удаляется из очереди.

▪Если флаг O_NONBLOCK не указан, mq_send блокируется, пока в очереди не появится свободное место для добавления сообщения в очередь. Если, чтобы отправить сообщение, ждёт более чем один поток или процесс, и в очереди появляется свободное место, то для отправки сообщения разблокируется поток/процесс самого высокого приоритета, который ждал дольше всех. То же самое касается mq_receive.

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

▪Для отмены существующей регистрации вызовите mq_notify с параметром "уведомление", равным NULL.

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

▪Если какой-нибудь поток процесса заблокирован в mq_receive и этот процесс также зарегистрировал уведомление, прибывающее сообщение поступает в mq_receive и уведомление не посылается.