Introduction aux POSIX MQ

Sur le forum, lors de la rédaction de cet article, une question très intéressante a été posée : « Quelle est la différence avec OpenMPI ? »

Cette question est très intéressante du fait que l'idée des MP/MPI est de partager des calculs par l'intermédiaire de « passage de messages ». Voici la réponse qui a été apportée :

• « Les POSIX MQ restent en “ local ” sur la machine (c'est l'inconvénient des System V MQ et POSIX MQ), mais elles sont “ normalement ” disponibles sur tous les systèmes POSIX ou SUS, et sont totalement gratuites (contrairement à beaucoup de produits travaillant sur les “ MQ ”, mais qui eux travaillent sur plusieurs machines en parallèle comme MPI).En cherchant plus de détails, OpenMPI parait extrêmement plus complet que les “ MQ ” classiques. Il y a UNE différence majeure qui change tout, et cela est lié à l'usage des MQ/Communication par rapport au MP/Parallélisme. »

Le MPI propose à l'envoi :

int MPI_Send (
	message, 	/* actual information sent */
	length, 	/* length of the message */
	datatype, 	/* MPI datatype of the message */
	destination, 	/* rank of the processor getting the message */
	tag, 		/* tag helps sort messages - likely an int and the same as in MPI_Recv */
	MPI_Comm 	/* almost alway MPI_Comm_World */
);

D'après ces slides (Introduction MPI ProgrammingIntroduction MPI Programming), on voit que le « rank » sert à identifier les processus qui vont exécuter une tâche, et que le « tag » sert à identifier une suite de messages liés entre eux… et SURTOUT (slide 25) :

Used to ensure messages are read in the right order
- standard says nothing about the order of message arrival

Ceci est la différence avec les MQ.

Les MQ assurent l'ordre d'envoi/arrivée ainsi que la gestion des priorités, tandis que les MP/MPI assurent la distribution et l'échange de structures au sein d'un système.

Le but du MP/MPI étant de calculer, et garder un « ensemble logique » de données entre elles, il faut surtout que tous les messages représentant la même donnée arrivent à un moment ou à un autre pour conserver l'intégrité de celle-ci, quelle que soit sa taille. À l'inverse, sur les MQ, on n'essaye pas d'envoyer des structures, mais simplement des messages (à but informatif), dont l'ordre d'arrivée est important.

Les points semblables entre MQ et MP/MPI :

• échange de données entre processus/machines : une MQ peut être lue/écrite par plusieurs processus (voire machines sur les produits commerciaux) ; plusieurs thread/processus/machines communiquent via plusieurs routines send/recv/sendrecv sur OpenMPI ;
• identification des communications : le nom d'une MQ sert à l'identifier sur l'ensemble du système, ensemble rank + tag permet de séparer les tâches et destinataires en « groupes ».

Les différences entre MQ et MP/MPI :

• les MQ travaillent sur l'ordre d'arrivée (paramètre priority) : on envoie avec une priorité au choix, on reçoit nécessairement le message le plus ancien avec la priorité la moins basse (donc priorité max sortira TOUJOURS en premier) ;
• le but d'une MQ est d'informer/transférer une petite donnée plutôt que de partager du calcul/des structures en mémoire ;
• les MQ (commerciales) se trouvent sur des systèmes d'informations « professionnels » pour coordonner l'ensemble des progiciels entre eux et remonter au plus vite une alerte (communications entre micro-ordinateurs, miniordinateurs et mainframes), tandis que les MP/MPI servent au calcul massivement parallèle sur des super calculateurs (grilles de calculs).

Les sockets permettent des échanges entre seulement 2 interlocuteurs, les pipes permettent des échanges entre plusieurs processus, mais tous doivent être de même parentée. Il est possible de faire des communications entre plus d'interlocuteurs, mais de nombreux autres pipes/sockets doivent être ouvert(e)s, toute leur gestion implique beaucoup d'appels système.

Les FIFO permettent des échanges entre N interlocuteurs, mais, il faut au moins qu'un interlocuteur soit en écriture et un en lecture avant de pouvoir envoyer des données, les priorités ne sont pas gérées, et la FIFO disparait après que tous les interlocuteurs ont fini leurs communications.

On peut donc retenir que les POSIX MQ sont créées et « existent » (ainsi que leur contenu) tant qu'elles ne sont pas détruites (impossible avec les FIFO, pipes et sockets).

Les priorités et l'ordre d'arrivée assuré permettent de conserver l'arrangement des messages en fonction de l'importance.

Avant de travailler sur une MQ, il faut évidemment l'ouvrir ou la créer avec cette la fonction mq_open(). La fonction mq_open() ne supprime AUCUN message, et n'en crée aucun. Cette fonction renvoie un identificateur de file qui est similaire à un file descriptor pour les fichiers : s'il est égal à -1, alors l'ouverture/création a échoué (voir errno pour les détails), sinon l'appel a réussi.

mqd_t mq_open(const char *, int, ...);

Le 1^er argument est le nom de la MQ. Le nom DOIT commencer par un slash « / », et contenir une chaîne alphanumérique. POSIX impose que l'ouverture d'un même nom commençant par un slash « / » ouvre la même MQ, quel que soit le processus.

Si d'autres slashs « / » sont présents dans le nom (ou aucun), POSIX ne donne aucune consigne, c'est à l'OS de décider comment réagir. Certains OS vont suivre ce chemin dans l'arborescence réelle, d'autres dans la partie dédiée aux MQ… à vous de vous renseigner sur la façon dont votre OS gère ces noms (s'il les gère !).

L'apparition dans le FS de la MQ n'est pas obligatoire. Mais si la gestion des MQ est faite avec des file descriptors classiques, alors le maximum de MQ ouvrables sera défini par la macro OPEN_MAX, en comptabilisant les autres file descriptors classiques.

Le 2^e argument est un ensemble de flags similaires à ceux que l'on trouve pour les fichiers :

• O_RDONLY : la file est ouverte en lecture seulement, donc pour lire. On pourra donc utiliser la fonction mq_receive(). Un même processus peut ouvrir plusieurs fois la même MQ avec différents droits ! ;
• O_WRONLY : la file est ouverte en écriture seulement, on pourra donc utiliser mq_send() ;
• O_RDWR : ouvre la file en lecture et en écriture. Il est donc possible d'utiliser mq_send() et mq_receive() ;
• O_CREAT : permet de créer une MQ. Voir l'encadré plus bas pour les détails, car ce flag nécessite que mq_open() prenne 2 paramètres supplémentaires. Si la MQ existe déjà, et que le flag O_EXCL n'est pas activé, ce flag est ignoré ;
• O_EXCL : avec O_CREAT, il permet d'empêcher l'ouverture de la file si celle-ci existe. Le comportement de O_EXCL sans O_CREAT n'est pas décrit (à ne pas tester donc) ;
• O_NONBLOCK : permet de dire si mq_send() et mq_receive() doivent attendre que les ressources nécessaires soient disponibles ou non (mq_send() attendra que la file dispose d'une place supplémentaire pour le message, et mq_receive() attendra qu'un message arrive dans la file). Si les ressources ne sont pas disponibles, les deux fonctions échouent et placent errno à EAGAIN.

struct mq_attr
{
  long mq_flags;	/* Flags de la file */
  long mq_maxmsg;	/* Nombre maximum de messages dans la file */
  long mq_msgsize;	/* Taille maximale de chaque message */
  long mq_curmsgs;	/* Nombre de messages actuellement dans la file */
};

L'écriture dans la file se fait en envoyant un message. Il n'est pas nécessaire de préparer de forme spécifique au message, c'est mq_send() qui se charge de créer la structure.

La fonction renvoie 0 en cas de succès, sinon -1.

int mq_send(mqd_t, const char *, size_t, unsigned);

Le 1^er argument est le descripteur de la MQ (on l'a reçu lors du mq_open()).

Le 2^e argument est le message que l'on souhaite envoyer. Il ne doit pas être plus grand que la taille déclarée à l'initialisation de la file !

Le 3^e argument est la taille du message envoyé.

Le 4^e argument est la priorité avec laquelle on souhaite envoyer le message. Par défaut, il est conseillé de mettre une priorité de 0 pour les messages classiques, et plus si l'information à transmettre n'est pas prévue dans le cadre du fonctionnement normal. Par exemple : plusieurs services émettent des demandes/traitements classiques, puis subitement une alerte doit être remontée. Les demandes seront de priorité 0, et l'alerte pourra être de priorité 1 à 31.

La lecture (destructrice) d'un message se fait avec mq_received(). On est sûr que le message qui sera extrait par mq_receive() sera celui de plus haute priorité présent dans la file, et le plus ancien. La valeur de retour est le nombre de caractères lus, ou -1 en cas d'échec.

ssize_t mq_receive(mqd_t, char *, size_t, unsigned *);

Le 1^er argument est le descripteur de MQ (récupéré avec mq_open()).

Le 2^e argument est un buffer de taille suffisante pour réceptionner le message. Étant donné que les MQ ont une taille maximale fixe, il est facile à la compilation d'inclure un buffer fixe (même chose à l'exécution avec malloc). Si le buffer est trop petit, le message ne sera pas lu ni détruit.

Le 3^e argument est la taille du buffer. Si cette taille est inférieure à la taille donnée à la file à sa création (via mq_attr.mq_msgsize), alors l'appel échoue.

Le 4^e argument est un pointeur vers un entier non signé. Précisément, si l'argument n'est pas NULL, mq_receive() va écrire dedans la priorité du message réceptionné. On peut donc créer une variable locale et la passer par référence à la fonction.

La fermeture d'une file ne sert qu'à retirer le descripteur de file associé au processus. La file n'est PAS détruite suite à mq_close() ! Le retour indique seulement si la fermeture a réussi (0) ou non (-1).

int mq_close(mqd_t);

Le seul paramètre est un descripteur de MQ. Il doit évidemment être celui d'une file précédemment ouverte.

Une fois la MQ fermée, il est possible de la supprimer avec mq_unlink(). Si la MQ est encore référencée par d'autres processus ou descripteurs, elle sera supprimée après que la dernière référence a été fermée, et l'appel à mq_unlink() ne sera pas bloquant. La fonction renvoie 0 si tout a bien fonctionné, sinon -1.

int mq_unlink(const char *);

Le seul paramètre est le nom de la file.

gcc -Wextra -Wall -Werror -c main.c

gcc -lrt main.o -o MQTestSimple

main.h

#ifndef MAIN_H_
# define MAIN_H_

# include <unistd.h>
# include <stdlib.h>
# include <stdio.h>
# include <fcntl.h>	/* For O_* constants */
# include <sys/stat.h>	/* For mode constants */
# include <mqueue.h>
# include <string.h>

#endif /* !MAIN_H_ */

main.c

# include "main.h"

# define MQ_NAME "/queuetest"
# define BUF_LEN 512

int main(void)
{
  struct mq_attr	attr;
  ssize_t 		len_recv;
  mqd_t 		My_MQ;
  char 			*text;
  char 			recv[BUF_LEN];

  attr.mq_flags = 0;
  attr.mq_maxmsg = 10;
  attr.mq_msgsize = BUF_LEN;
  attr.mq_curmsgs = 0;
  
  My_MQ = mq_open(MQ_NAME, O_RDWR | O_CREAT, 0644, &attr);
  if ((int) My_MQ == -1)
  {
    perror("Error : mq_open failed !\n");
    return(-1);
  }
  
  printf("%s\n", "Envoi msg de priorite 42");
  text = strdup("Message test prio : 42 !");
  mq_send(My_MQ, text, strlen(text), 42);
  free(text);
  
  printf("%s\n", "Envoi msg de priorite 21");
  text = strdup("Message test prio : 21 !");
  mq_send(My_MQ, text, strlen(text), 21);
  free(text);
  
  memset(recv, 0, BUF_LEN);
  len_recv = mq_receive(My_MQ, recv, BUF_LEN, NULL);
  printf("Premier msg recu (len : %u) : %s\n", (unsigned int) len_recv, recv);
  
  memset(recv, 0, BUF_LEN);
  len_recv = mq_receive(My_MQ, recv, BUF_LEN, NULL);
  printf("Deuxieme msg recu (len : %u) : %s\n", (unsigned int) len_recv, recv);
  
  mq_close(My_MQ);
  mq_unlink(MQ_NAME);
  
  return (0);
}

Cette fonction sert à récupérer les propriétés de la file pour connaitre le nombre maximum de messages que l'on peut laisser en attente de traitement, leur longueur maximale autorisée au sein de la file, etc. La fonction renvoie un entier, 0 en cas de succès, sinon -1.

int mq_getattr(mqd_t, struct mq_attr *);

Le 1^er paramètre est le descripteur de la file dont on souhaite récupérer les propriétés.

Le 2^e paramètre est un pointeur sur une structure mq_attr, qui sera rempli avec les propriétés. Il parait approprié d'utiliser une référence sur une variable locale si un ou quelques tests uniques (non répétitifs) sont effectués.

Voici un rappel de la structure mq_attr :

struct mq_attr
{
  long mq_flags;	/* Flags de la file */
  long mq_maxmsg;	/* Nombre maximum de messages dans la file */
  long mq_msgsize;	/* Taille maximale de chaque message */
  long mq_curmsgs;	/* Nombre de messages actuellement dans la file */
};

La fonction permet de modifier certains paramètres de la file lors du fonctionnement du programme.

Un paramètre permet notamment de sauvegarder l'ancien état des propriétés. Les valeurs contenues dans mq_maxmsg, mq_msgsize et mq_curmsgs sont ignorées, on en déduit donc que seuls les flags d'accès sont modifiés (mq_flags). L'appel à cette fonction renvoie l'entier 0 en cas de succès, et -1 en cas d'échec.

int mq_setattr(mqd_t, const struct mq_attr * restrict, struct mq_attr * restrict);

Le 1^er argument est le descripteur de la file dont on souhaite modifier les propriétés.

Le 2^e est un pointeur sur structure mq_attr, contenant les nouvelles valeurs. Une variable locale passée en référence est suffisante dans la plupart des cas.

Le 3^e argument est aussi un pointeur sur structure mq_attr, et il peut être mis à NULL. Celui-ci sert à sauvegarder l'état précédent de la file ! Plutôt que de faire deux appels successifs à mq_getattr puis mq_setattr, il est possible de sauvegarder l'état précédent en même temps que l'on en écrit un nouveau. Il est vivement conseillé d'utiliser malloc dans ce cas, et de sauvegarder cet état quelque part à long terme.

La fonction agit de façon très similaire à mq_send(), mais diffère uniquement lorsque la file est pleine et que O_NONBLOCK est inactif : dans ce cas, la fonction patientera « au maximum » le temps déclaré dans le dernier argument (instant absolu en secondes au format time_t UNIX). La clock utilisée sera prioritairement celle utilisée par le module temps réel, mais si celle-ci n'existe pas, la clock système sera utilisée. Si la file n'est pas pleine, le message est obligatoirement posté, que le temps maximum soit révolu ou non. Si la fonction réussit, elle renvoie l'entier 0, sinon -1.

int mq_timedsend(mqd_t, const char *, size_t, unsigned, const struct timespec *);

Le 1^er argument, comme pour mq_send(), est le descripteur de la file que l'on souhaite utiliser.

Le 2^e argument est un pointeur vers le message à envoyer.

Le 3^e argument est la taille du message à envoyer (ne doit pas dépasser la taille maximum définie à la création de la file).

Le 4^e argument est la priorité du message (ne doit pas dépasser le maximum).

Le 5^e argument, unique à mq_timedsend(), est un pointeur sur structure timespec. Ce temps est un instant fixe à définir avec la structure. Si la file n'est pas pleine, l'argument est ignoré et le message posté. Voici la structure timespec :

struct timespec
{
  time_t tv_sec;	/* Temps en secondes depuis le 1er janvier 1970 */
  long tv_nsec;		/* Temps en nanosecondes */
};

Tout comme mq_receive(), la fonction effectue les mêmes opérations de la même façon, excepté lorsque la file est vide et que O_NONBLOCK est inactif. Dans ce cas précis, la fonction patientera jusqu'au moment indiqué par le dernier argument, et si celui-ci est dépassé, alors la fonction échoue. Si l'instant absolu déclaré est dépassé, mais que la file n'est pas vide, la fonction extrait malgré tout un message et ignore le dernier paramètre. La fonction renvoie la taille du message qu'elle a extrait, ou -1 en cas d'échec.

ssize_t mq_timedreceive(mqd_t, char * restrict, size_t, unsigned * restrict, const struct timespec * restrict);

Le 1^er argument est le descripteur de file.

Le 2^e argument est un buffer où sera copié le message. Il doit être suffisamment long et peut être retrouvé grâce à mq_getattr().

Le 3^e argument est la taille du buffer. Si la taille est inférieure à mq_msgsize, définie à la création de la file et disponible avec mq_getattr(), alors l'appel échoue.

Le 4^e argument est un pointeur sur unsigned qui sera rempli avec la priorité du message récupéré. Il vaut mieux déclarer localement une variable et la passer en référence.

Le 5^e argument est un pointeur sur structure timespec. Cette structure définit l'instant maximum que la fonction attendra si la file est vide et que O_NONBLOCK est inactif. Si la file n'est pas vide, et que le maximum est déjà dépassé, la fonction ignore ce paramètre et extrait un message. Voir l'explication avec mq_timedsend() pour les détails.

Cette fonction permet d'être informé quand une file vide ne l'est plus (quand une file vide commence à être alimentée). Attention, si un thread est en attente sur un mq_receive(), alors le thread sera débloqué, et aucune notification ne sera remontée via le mq_notify(). Le mécanisme de notification est lié aux signaux, il faudra donc penser à inclure <signal.h>. Un seul et unique processus peut s'enregistrer pour scruter une file. Une fois la notification remontée le processus est automatiquement désabonné. Si un processus s'enregistre, il peut se désinscrire et permettre qu'un autre s'enregistre.

La fonction renvoie l'entier 0 si elle a réussi, sinon -1. Les causes d'échecs peuvent être que la file désignée n'existe pas (errno à EBADF), ou qu'un autre processus est déjà inscrit (errno à EBUSY).

int mq_notify(mqd_t, const struct sigevent *);

Le 1^er argument est le descripteur de la file que l'on souhaite scruter.

Le 2^e argument est un pointeur sur structure sigevent. Il est possible de mettre ce paramètre à NULL afin de se désinscrire de la notification. Pour s'inscrire voici le contenu d'un sigevent :

struct sigevent
{
  int sigev_notify;					/* Type de notification */
  int sigev_signo;					/* Numero du signal */
  union sigval sigev_value;				/* Valeur du signal */
  void (*) (union sigval) sigev_notify_function;	/* Fonction de notification */
  (pthread_attr_t *) sigev_notify_attributes;		/* Attributs de la notification */
};

La structure sigevent doit contenir au moins le membre sigev_notify avec une des 4 valeurs suivantes :

SIGEV_NONE, SIGEV_SIGNAL, SIGEV_THREAD, [SIGEV_THREAD_ID - spécifique à Linux].

• SIGEV_NONE : rien ne se passe quand l'évènement se produit. La seule utilité dans le cadre des MQ pourrait être d'empêcher tous les autres processus d'effectuer un notify.
• SIGEV_SIGNAL : envoie un signal lorsque l'évènement se produit. Le numéro du signal se trouve dans le 2 membre de la structure sigevent : sigev_signo. Le signal envoyé portera quelques informations transmises depuis la structure sigevent (que vous remplissez lors de l'appel à mq_notify()). Les informations se trouvent dans la structure siginfo_t en lecture lorsque le signal est émis :
  • siginfo_t->si_code : valeur que l'API appelante remplit ;
  • siginfo_t->si_signo : numéro du signal (contenu dans sigevent->sigev_value) ;
  • siginfo_t->si_value : valeur que l'utilisateur (vous) remplit des informations qu'il souhaite depuis le champ sigevent->sigev_value (de type union sigval)

union sigval
{
  int sival_int;
  void *sival_ptr;
};

On comprend donc qu'il faut également ajouter un sigaction du côté de l'application pour récupérer ce signal et effectuer une action « utile ». L'union passée dans la structure permet de passer un entier 32 bits ou un pointeur 32/64 bits, ce qui peut être très utile.

• SIGEV_THREAD : crée un thread (si sigev_notify_attributes n'est pas NULL et contient une structure pthread_attr_t correcte permettant la création d'un thread) ou simule un thread (appel forcé de la fonction, comme un handler de signal) avec comme fonction d'appel celle définie dans sigev_notify_function et en paramètre le contenu de sigev_value (voir l'union dans le paragraphe précédent avec l'entier 32 bits ou le pointeur 32/64 bits).
• SIGEV_THREAD_ID : spécifique à Linux. Je ne me suis pas renseigné dessus, à tester.