26.5. Очереди сообщений.

26.5.1. Основные структуры ядра для работы с очередями.

Очередь сообщений — это связный список, находящийся в адресном пространстве ядра. Каждая очередь имеет свой уникальный идентификатор IPC.

Структура ядра msgbuf (описана в файле /usr/src/linux/include/linux/msg.h) является буфером сообщений:

Struct msgbuf {

 long mtype; /* тип сообщения */

 char mtext[1]; /* текст сообщения */

};

Тут все ясно: тип сообщения и само сообщение. Используя тип сообщения, вы можете помещать в одну очередь разные сообщения, а не создавать еще одну очередь. Например, у нас есть два приложения — клиент и сервер. Вы можете использовать для них одну и ту же очередь: сообщения клиента будут с номером 1 (mtype = 1), а сообщения сервера — с номером 0 (mtype = 0).

Ясное дело, что сообщения из одного символа нас не устраивают, поэтому вы можете переопределить структуру msgbuf в своей программе:

Struct my_buf {

 long mtype;

 char mtext[128];

}

Вы также можете добавлять новые поля в эту структуру (но только в своей программе! Код ядра модифицировать не нужно):

Struct my_buf {

 long mtype;

 char mtext[128];

 char info[50];

 int status;

}

He бойтесь создавать свои структуры: ядру все равно, с какими данными работать, вам нужно учитывать только максимальный размер сообщения, который определен в файле /usr/src/linux/include/linux/msg.h:

#define MSGMAX 4056.

4056 байтов — это максимальный размер не ваших данных, а всей структуры. включая тип сообщения. Размер типа long равен 4 байтам.

Сами сообщения хранятся ядром в структуре msg, которая также определена в файле msg.h:

Struct msg {

 struct msg *msg_next; /* указатель на след. сообщение.

                          в очереди */

 long msg_type;  /* тип сообщения */

 char *msg_spot; /* адрес самого сообщения (текста) */

 short msg_ts;   /* размер сообщения (текста) */

};

Сообщения хранятся в виде односвязного списка. Первый член структуры msg_next — это указатель на следующее сообщение в очереди. Второй член msg_type — это тип сообщения, такой же, как в структуре msg_buf.

Следующий член структуры — это указатель на начало текста сообщения, а последний член msg_ts — размер текста сообщения.

Каждый тип объекта IPC представляется в ядре определенной структурой. Для очередей сообщений это структура msqid_ds (описана в файле /usr/src/linux/include/linux/msg.h).

Struct msqid_ds {

 struct ipc_perm msg_perm; /* информация о правах.

                              доступа */

 struct msg *msg_first; /* указатель на первое сообщение.

                           в очереди */

 struct msg *msg_last; /* указатель на последнее сообщение.

                          в очереди*/

 time_t msg_stime; /* время последнего вызова msgsnd */

 time_t msg_rtime; /* время последнего вызова msgrcv */

 time_t msg_ctime; /* время последнего изменения */

 struct wait_queue *wwait;

 struct wait_queue *rwait;

 ushort msg_cbytes;

 ushort msg_qnum;

 ushort msg_qbytes; /* максимальное число байтов на очередь */

 ushort msg_lspid;  /* pid последнего испустившего msgsnd */

 ushort msg_lrpid;  /* последний полученный pid */

};

♦ msg_perm.

Это структура типа ipc_perm (ipc_perm определена в файле linux/ipc.h). Данная структура содержит информацию о владельце и правах доступа:

Struct ipc_perm {

 key_t key;

 ushort uid;  /* uid и gid владельца */

 ushort gid;

 ushort cuid; /* uid и gid создателя */

 ushort cgid;

 ushort mode; /* режим доступа */

 ushort seq;  /* системное поле. Вас оно не касается. */

};

♦ msg_ftrst.

Указатель на первое сообщение в очереди.

♦ msg_last.

Указатель на последнее сообщение в очереди.

♦ msg_stime.

Время отправки последнего сообщения из очереди,

♦ msg_rtime.

Время последнего изъятия сообщения из очереди.

♦ msg_ctime.

Время последнего изменения очереди.

♦ wwait и rwait.

Указатели в очередь ожидания ядра, которые используются, когда очередь переполнена и процесс вынужден ждать из-за этого.

♦ msg_cbytes.

Суммарный объем всех сообщений в очереди.

♦ msg_qnum.

Количество сообщений в очереди.

♦ msg_qbytes.

Максимальный размер очереди.

♦ msg_lspid.

PID процесса, который послал последнее сообщение в очереди.

♦ msg_lrpid.

PID процесса, который получил сообщение из очереди.

26.5.2. Создание очереди сообщений.

Для создания очереди сообщений используется системный вызов msgget(). Этот же вызов используется для подключения к уже существующей очереди:

Int msgget(key_t key, int msgflg);

Первый аргумент — это ключ, который мы получаем с помощью системного вызова ftok(). Второй аргумент — это режим доступа к очереди:

♦ IPC_CREAT — создать очередь, если она не была создана ранее.

♦ IPC_EXCL — если использовать вместе с IPC_CREAT, то в случае, если очередь существует, мы получим ошибку.

Если использовать только IPC_CREAT (без IPC_EXCL), то вызов msgget() всегда возвращает идентификатор очереди, даже если очередь уже существует (происходит подключение к очереди). Если использовать IPC_EXCL вместе с IPC_CREAT, также будет создана новая очередь, но если очередь уже существует, подключения не произойдет, а функция msgget() возвратит -1 (ошибка).

Вместе с режимом IPC_CREAT можно указывать права доступа к очереди с помощью операции OR:

IPC_CREAT | 0660.

Если произошла ошибка и msgget() вернул -1, то переменная errno устанавливается следующим образом:

♦ EACCESS — у вас нет прав доступа к объекту IPC;

♦ EEXIST — очередь уже существует, создание невозможно, но возможно подключение к очереди;

♦ EIDRM — очередь помечена для удаления;

♦ ENOENT — очередь не существует (в случае подключения);

♦ ENOMEM — не хватает памяти для создания очереди;

♦ ENOSPC — не хватает адресного пространства (то есть превышено максимальное количество очередей).

Следующий код создает очередь сообщений:

Key_t key; /* ключ IPC */

Int id; /* ID очереди сообщений */

/* создаем ключ */

Key = ftok(".", 'd');

/* создаем очередь */

If ((id = msgget(key, IPC_CREAT | 0660 )) == -1) {

 printf("Ошибка при создании очереди\n");

}

26.5.3. Постановка сообщения в очередь.

Для постановки сообщения в очередь используется вызов msgsnd():

Int msgsnd(int msqid, struct msgbuf *msgp, int msgsz,

 int msgflg);

Первый аргумент — это идентификатор очереди, в которую нужно добавить сообщение. Данный идентификатор мы предварительно получаем с помощью системного вызова msgget(). Второй параметр — это указатель на буфер сообщения. Третий аргумент — это длина сообщения без учета типа сообщения (4 байта). Последний аргумент обычно устанавливают равным 0 или IPC_NOWAIT, если вы не хотите, чтобы процесс был блокирован при постановке сообщения в очередь, в случае переполнения очереди. По умолчанию (когда флаг равен 0), если очередь переполнена, ваш процесс будет блокирован до тех пор, пока сообщение не будет поставлено в очередь.

Как обычно, в случае успеха вызов возвращает 0, а если произошла ошибка, то -1. С помощью errno можно анализировать ошибку:

♦ EAGAIN — очередь переполнена, а вы используете флаг IPC_NOWAIT, то есть сообщение будет удалено и вам нужно заново поставить его в очередь (отсюда и название ошибки — AGAIN (опять));

♦ EACCESS — у вас недостаточно прав для записи сообщения в очередь;

♦ EFAULT— неверный адрес буфера msgp (невозможно получить доступ к этому адресу);

♦ EIDRM — очередь сообщений удалена;

♦ EINVAL — ошибка в аргументах, например, неправильное значение идентификатора очереди, отрицательный тип сообщения, неправильный размер сообщения и т.д.

♦ ENOMEM — не хватает памяти.

Следующий фрагмент кода демонстрирует постановку сообщения в очередь;

Int res, length; /* результат операции и длина сообщения */

Struct my_buf *buf; /* само сообщение */

/* определяем длину сообщения — 4 байта */

Length = sizeof(struct my_buf) - sizeof(long);

If ((res = msgsnd( id, &buf, length, 0)) == -1) {

 printf("Ошибка при постановке сообщения в очередь\n");

}

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

Листинг 26.5. Пример работы с очередью.

#include <stdio.h>

#include <stdlib.h>

#include <linux/ipc.h>

#include <linux/msg.h>

Main() {

 int id; /* Идентификатор очереди */

 key_t key; /* Ключ */

 int res, length; /* Результат операции и длина сообщения */

 struct my_buf {

  long mtype; /* тип сообщения */

  /* Далее следуют произвольные поля -

   они зависят от сообщения */

  int op_type; /* тип операции */

  int l_ор; /* первый операнд */

  int r_op; /* второй операнд */

 } msg;

 /* Генерируем IPC-ключ */

 key = ftok(".", 'd');

 /* Создаем очередь или присоединяемся к уже существующей */

 if ((id = msgget(key, IPC_CREAT | 0660)) == -1) {

  printf("Ошибка при создании очереди\n");

  exit(1);

 }

 /* Заполняем сообщение */

 msg.type = 1; /* тип сообщения,

                  должен быть положительным! */

 msg.op_.type = 0; /* тип операции */

 msg.l_op = 6;

 msg.r_op = 5;

 /* определяем длину сообщения - 4 байта */

 length = sizeof(struct my_buf) - sizeof(long);

 if ((res = msgsnd(id, &buf, length, 0)) == -1) {

  printf("Ошибка при постановке сообщения в очередь\n");

  exit(1);

 }

}

После запуска этой программы запустите программу ipcs и посмотрите на статус только что отправленного сообщения. Теперь напишем программу, которая получит это сообщение.

26.5.4. Получение сообщений очереди.

Для получения сообщения используется системный вызов msgrcv():

Int msgrcv(int msqid, struct msgbuf *msgp, int msgsz,

 long mtype, int msgflg);

Первый аргумент определяет очередь, из которой нужно получить сообщение. Второй аргумент — это адрес буфера, в который будет записано сообщение. Третий аргумент — это ограничитель длины сообщения. Четвертый аргумент — это тип сообщения. Ядро будет искать в очереди наиболее старое сообщение данного типа и вернет его копию. Если mtype=0, то ядро вернет самое старое сообщение независимо от типа.

После успешного получения сообщения оно удаляется из очереди.

В случае успеха вызов msqrcv() возвращает число байтов, скопированных в буфер, или -1 в случае ошибки. Переменная errno устанавливается следующим образом:

♦ E2BIG — длина сообщения больше, чем ограничитель msgsz;

♦ EACCESS — у вас недостаточно прав;

♦ EFAULT — недоступен адрес буфера;

♦ EIDRM — очередь уничтожена ядром;

♦ EINTR — операция прервана поступившим сигналом;

♦ EINVAL — ошибка в аргументах, например, отрицательный размер сообщения или неверный номер очереди;

♦ ENOMSG — нет сообщения, удовлетворяющего условию. Посылается, если установлен флаг IPC_NOWAIT, в противном случае процесс будет ждать нужного сообщения.

Последний аргумент предоставляет дополнительные возможности по работе с сообщениями. Если установлен бит MSG_NOERROR в msgflg, то если размер сообщения больше, чем msgsz, оно будет обрезано и вы получите только msgsz байтов. Если флаг MSG_NOERROR не устанавливать, вы получите ошибку E2BIG.

Следующий код получает сообщение из очереди:

Int id; /* ID очереди */

Int res, length; /* результат операции и длина */

Struct my_buf buf; /* буфер */

Int type = 1; /* тип сообщения */

Length = sizeof(struct my_buf) — sizeof(long);

If ((res =

 msgrcv(id, &buf, length, type, 0)) == -1) {

 printf("Ошибка!");

 /* можно проанализировать ошибку */

 if (errno==E2BIG) printf("Сообщение слишком большое\n");

 if (errno==EACCESS) printf("Нет доступа\n");

 /* и т.д. */

 exit(1);

}

26.5.5. Проверка наличия сообщения в очереди.

Наверное, вы не хотите, чтобы ваша программа ждала, пока в очереди появится нужное сообщение. Используя особенности системного вызова msgrcv(), можно написать код проверки наличия сообщения определенного типа в очереди. Напишем функцию msg_exists(), которая будет возвращать TRUE, если сообщение есть в очереди, или FALSE, если сообщения в очереди нет.

Int msg_exists(int id, long type) {

 int res;

 if ((result = msgrcv(id, NULL, 0, type, IPC_NOWAIT)) == -1) {

  if (errno == E2BIG).

   return(TRUE);

 }

 return(FALSE);

}

В вызове msgrcv() отсутствует адрес буфера и длина сообщения. Этим мы специально провоцируем ошибку, а вызов IPC_NOWAIT отказывает от блокировки процесса. Мы проверяем errno; если он равен E2BIG, значит, сообщение есть в очереди. Ошибка E2BIG порождается потому, что мы установили размер сообщения равным 0.

26.5.6. Тотальный контроль.

До сих пор мы рассматривали только системные вызовы для работы с сообщениями очереди, сейчас рассмотрим системный вызов msgctl(), предназначенный для контроля самой очереди.

Int msgctl(int msgqid, int cmd, struct msqid_ds *buf);

Первый аргумент — это ID очереди, второй — команда, которую нужно выполнить:

♦ IPC_STAT — записывает в буфер buf структуру msqid_ds для очереди сообщений с идентификатором msgqid.

♦ IPC_SET — устанавливает значение ipc_perm структуры msqid. Значение берется из буфера buf.

♦ IPC_RMID — удаляет очередь.

Системный вызов возвращает 0 в случае успеха и -1, если произошла ошибка. Переменная errno устанавливается следующим образом:

♦ EACCESS — недостаточно прав.

♦ EFAULT — невозможно получить доступ к адресу буфера buf или неверный адрес.

♦ EIDRM — очередь была уничтожена прямо во время запроса.

♦ EINVAL — ошибка в аргументах, например, неправильный ID очереди или отрицательный размер сообщения.

♦ EPERM — у вас нет прав на запись в очередь.

Структура msqid_ds уже рассматривалась ранее, поэтому не вижу смысла приводить ее описание еще раз.

Если подытожить, то все, что мы можем сделать с очередью — это удалить ее и изменить права доступа или информацию о владельце (его UID и GID). Как удалить, думаю, понятно. Напишем функцию change_mode(), которая будет изменять права доступа к нашей очереди. Ей нужно передать два параметра — идентификатор очереди и новый режим доступа в виде строки, например, "0660".

Int change_mode(int id, char *mode) {

 struct msqid_ds buf;

 /* Получаем копию структуры в буфер buf */

 if (msgctl(id, IPC_STAT, &buf) == -1) {

  return(-1);

 }

 /* Изменяем права доступа */

 sscanf(mode, "%ho", &buf.msg_perm.mode);

 /* Модернизируем внутреннюю структуру */

 if (msgctl(id, IPC_SET, &buf) == -1} {

  return(-1);

 }

 return(0);

}

Наша функция возвращает 0 в случае успеха или -1, если произошла ошибка.

На этом обзор средств для работы с очередями сообщений можно считать законченным, теперь с чистой совестью перейдем к следующему средству IPC — семафорам.