UNIX: взаимодействие процессов
вернуться

Стивенс Уильям Ричард
Шрифт:

13 struct mymq_info *mqinfo;

14 mqinfo = mqd;

15 if (mqinfo->mqi_magic != MQI_MAGIC) {

16 errno = EBADF;

17 return(-1);

18 }

19 mqhdr = mqinfo->mqi_hdr; /* указатель struct */

20 mptr = (int8_t *) mqhdr; /* указатель на байт */

21 attr = &mqhdr->mqh_attr;

22 if ((n = pthread_mutex_lock(&mqhdr->mqh_lock)) != 0) {

23 errno = n;

24 return(-1);

25 }

26 if (maxlen < attr->mq_msgsize) {

27 errno = EMSGSIZE;

28 goto err;

29 }

30 if (attr->mq_curmsgs = 0) { /* очередь пуста */

31 if (mqinfo->mqi_flags & O_NONBLOCK) {

32 errno = EAGAIN;

33 goto err;

34 }

35 /* ожидаем помещения сообщения в очередь */

36 mqhdr->mqh_nwait++;

37 while (attr->mq_curmsgs == 0)

38 pthread_cond_wait(&mqhdr->mqh_wait, &mqhdr->mqh_lock);

39 mqhdr->mqh_nwait--;

40 }

Листинг 5.28. Функция mq_receive: вторая половина

//my_pxmsg_mmap/mq_receive.c

41
if ((index = mqhdr->mqh_head) == 0)

42 err_dump("mymq_receive: curmsgs = %ld; head = 0", attr->mq_curmsgs);

43 msghdr = (struct mymsg_hdr *) &mptr[index];

44 mqhdr->mqh_head = msghdr->msg_next; /* новое начало списка */

45 len = msghdr->msg_len;

46 memcpy(ptr, msghdr + 1, len); /* копирование самого сообщения */

47 if (priop != NULL)

48 *priop = msghdr->msg_prio;

49 /* только что считанное сообщение становится первым в списке пустых */

50 msghdr->msg_next = mqhdr->mqr_free;

51 mqhdr->mqh_free = index;

52 /* запуск любого процесса, заблокированного в вызове mq_send */

53 if (attr->mq_curmsgs == attr->mq_maxmsg)

54 pthread_cond_signal(&mqhdr->mqh_wait);

55 attr->mq_curmsgs--;

56 pthread_mutex_unlock(&mqhdr->mqh_lock);

57 return(len);

58 err:

59 pthread_mutex_unlock(&mqhdr->mqh_lock);

60 return(-1);

61 }

Возвращение сообщения вызвавшему процессу

43-51 msghdr указывает на msg_hdr первого сообщения в очереди, которое мы и возвратим. Освободившееся сообщение становится первым в списке свободных. 

Разблокирование процесса, заблокированного в вызове mq_send

52-54 Если очередь была полной в момент считывания сообщения, мы вызываем pthread_cond_signal для отправки сообщения любому из процессов, заблокированных в вызове mq_send.

5.9. Резюме

Очереди сообщений Posix просты в использовании: новая очередь создается (или существующая открывается) функцией mq_open; закрываются очереди вызовом mq_close, а удаляются mq_unlink. Поместить сообщение в очередь можно функцией mq_send, а считать его оттуда можно с помощью mq_receive. Атрибуты очереди можно считать и установить с помощью функций mq_getattr и mq_setattr, а функция mq_notify позволяет зарегистрировать процесс на уведомление о помещении нового сообщения в пустую очередь. Каждое сообщение в очереди обладает приоритетом (небольшое целое число), и функция mq_receive всегда возвращает старейшее сообщение с наивысшим приоритетом.

Изучая mq_notify, мы познакомились с сигналами реального времени стандарта Posix, которые обладают номерами от SIGMIN до SIGMAX. При установке обработчика для этих сигналов с флагом SA_SIGINFO они будут помещаться в очередь, доставляться в порядке очереди и сопровождаться двумя дополнительными аргументами (при вызове обработчика).

Наконец, мы реализовали большую часть возможностей очереди сообщений Posix в приблизительно 500 строках кода на языке С, используя отображаемые в память файлы, взаимные исключения и условные переменные Posix. Эта реализация иллюстрирует обработку ситуации гонок при создании новой очереди; еще раз нам придется столкнуться с такой ситуацией в главе 10 при реализации семафоров Posix.