24 Jun 2023, 8:32 PM
24 Jun 2023, 8:32 PM

9. Message Queues

Una coda di messaggi, message queue, è una lista concatenata memorizzata all’interno del kernel ed identificata con un ID (un intero positivo univoco), chiamato queue identifier.

Questo ID viene condiviso tra i processi interessati, e viene generato attraverso una chiave univoca.

Una coda deve essere innanzitutto generata in maniera analoga ad una FIFO, impostando dei permessi. Ad una coda esistente si possono aggiungere o recuperare messaggi tipicamente in modalità “autosincrona”: la lettura attende la presenza di un messaggio, la scrittura attende che via sia spazio disponibile. Questi comportamenti possono però essere configurati.

Queue identifier e queue key

Quanto trattiamo di message queue, abbiamo due identificativi:

Key: intero che identifica un insieme di risorse condivisibili nel kernel, come semafori, memoria condivisa e code. Questa chiave univoca deve essere nota a più processi, e viene usata per ottenere il queue identifier. Il kernel utilizza questa chiave per cercare la coda di messaggi corrispondente e restituire l'ID al processo.

Queue identifier: id univoco della coda, generato dal kernel ed associato ad una specifica key. Questo ID viene usato per interagire con la coda, ovvero per inviare e ricevere messaggi dalla coda.

Creazione coda

int msgget(key_t key, int msgflg)
Restituisce l’identificativo di una coda basandosi sulla chiave “key” e sui flags:

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h> //msgget.c
key_t queueKey = 56; //Unique key
int queueId = msgget(queueKey, 0777 | IPC_CREAT | IPC_EXCL); //0777 indica che tutti gli utenti possono leggere, scrivere ed eseguire operazioni sulla coda di messaggi

Ottenere chiave univoca

key_t ftok(const char *path, int id)

Restituisce una chiave basandosi sul path (una cartella o un file), esistente ed accessibile nel file-system, e sull’id numerico. La chiave dovrebbe essere univoca e sempre la stessa per ogni coppia <path,id> in ogni istante sullo stesso sistema.

Un metodo d’uso, per evitare possibili conflitti, potrebbe essere generare un path (es. un file) temporaneo univoco, usarlo, eventualmente rimuoverlo, ed usare l’id per per rappresentare diverse “categorie” di code, a mo’ di indice.

#include <sys/ipc.h> //ftok.c
key_t queue1Key = ftok("/tmp/unique", 1);
key_t queue2Key = ftok("/tmp/unique", 2);

Se si cerca di ottenere una chiave da una queue che non era stata ancora creata (usando la funzione create()), la funzione ftok() ritorna -1.

Esempio creazione

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <stdio.h>
#include <fcntl.h> //ipcCreation.c
int main()
{
	remove("/tmp/unique"); //Rimuove il file chiamato /tmp/unique se esiste
	key_t queue1Key = ftok("/tmp/unique", 1); //Tenta di ottenere una chiave univoca attraverso il percorso del file e un ID di 1. Dato che il file non esiste, la chiamata a ftok fallisce. queue1Key conterrà -1
	creat("/tmp/unique", 0777); //Crea il file
	queue1Key = ftok("/tmp/unique", 1); //Prova ad ottenere una chiave univoca e questa volta la chiamata ha successo e quindi restituisce una chiave univoca
	int queueId = msgget(queue1Key ,0777 | IPC_CREAT); //Il programma utilizza questa chiave per creare una nuova coda di messaggi
	queueId = msgget(queue1Key , 0777); //Ottiene l'ID di una coda di messaggi esistente. La chiave univoca e' specificata come queue1Key. Se la coda di messaggi non esiste allora restituira' un errore, ma in questo caso va a buon fine
	msgctl(queue1Key,IPC_RMID,NULL); //Tenta di rimuovere una coda di messaggi, ma dato che viene passato queue1Key invece dell'ID della coda di messaggi, la chiamata a msgctl fallra' e restituira' un errore
	msgctl(queueId,IPC_RMID,NULL); //Rimuove correttamente una coda di messaggi
	queueId = msgget(queue1Key , 0777); // tenta di ottenere l'ID di una coda di messaggi esistente, ma poiche' la coda di messaggi e' stata appena rimossa dalla riga di codice precedente, questa chiamata fallira' e restituira' un errore
	queueId = msgget(queue1Key , 0777 | IPC_CREAT); //Crea una nuova coda di messaggi, perche' non esiste una coda con quella chiave
	queueId = msgget(queue1Key , 0777 | IPC_CREAT); //Ottiene l'ID della coda di messaggi esistente. L'operazione ha successo perche' esiste una coda con la stessa chiave
	queueId = msgget(queue1Key , 0777 | IPC_CREAT | IPC_EXCL); //Tenta di creare una nuova coda di messaggi, ma dato che esiste gia' una coda di messaggi con la stessa chiave ed essendo specificato il flag IPC_EXCL, la chiamata fallira' e restituira' un errore
	return 0;
}

Le queue sono persistenti

#include <sys/ipc.h>
#include <stdio.h>
#include <sys/msg.h> //persistent.c
int main()
{
	key_t queue1Key = ftok("/tmp/unique", 1);
	int queueId = msgget(queue1Key , 0777 | IPC_CREAT | IPC_EXCL);
	perror("Error:");
}

Se eseguiamo questo programma dopo aver eseguito il precedente "ipeCreation.e" verrà generato un errore dato che la coda esiste già ed abbiamo usato il flag IPC_EXCL.

Comunicazione - 1

Ogni messaggio inserito nella coda ha:

Al contrario delle FIFO, i messaggi in una coda possono essere recuperati anche sulla base del tipo e non solo del loro ordine “assoluto” di arrivo. Così come i files, le code sono delle strutture persistenti che continuano ad esistere, assieme ai messaggi in esse salvati, anche alla terminazione del processo che le ha create. L’eliminazione della coda deve essere esplicita.

struct msg_buffer
{
	long mtype;
	char mtext[100];
} message;

Comunicazione - 2

Il payload del messaggio non deve essere necessariamente un campo testuale: può essere una qualsiasi struttura dati (può anche essere vuoto).

typedef struct book
{
	char title[10];
	char description[100];
	unsigned short chapters;
} Book;

struct msg_buffer
{
	long mtype;
	Book mtext;
} message_rcv;

struct msg_empty
{
	long mtype;
} message_empty;

Inviare messaggi

int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg);

Aggiunge una copia del messaggio puntato da msgp, con un payload di dimensione msgsz, alla coda identificata da msqid. Il messaggio viene inserito immediatamente se c’è abbastanza spazio disponibile, altrimenti la chiamata si blocca fino a che abbastanza spazio diventa disponibile. Se msgflg è IPC_NOWAIT allora la chiamata fallisce in assenza di spazio.

NB: msgsz è la grandezza del payload del messaggio, non del messaggio intero (che contiene anche il tipo)! Per esempio, sizeof(msgp.mtext)

Ricevere messaggi - 1

ssize_t msgrcv(int msqid,void *msgp,size_t msgsz,long msgtyp,int msgflg)

Rimuove un messaggio dalla coda msqid e lo salva nel buffer msgp. msgsz specifica la lunghezza massima del payload del messaggio (per esempio mtext della struttura msgp). Se il payload ha una lunghezza maggiore e msgflg è MSG_NOERROR allora il payload viene troncato (viene persa la parte in eccesso), se MSG_NOERROR non è specificato allora il payload non viene eliminato e la chiamata fallisce.

Se non sono presenti messaggi, la chiamata si blocca in loro attesa. Il flag IPC_NOWAIT fa fallire la syscall se non sono presenti messaggi.

Ricevere messaggi - 2

ssize_t msgrcv(int msqid,void *msgp,size_t msgsz,long msgtyp,int msgflg)

A seconda di msgtyp viene recuperato il messaggio:

Esempio comunicazione - 1

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <string.h>
#include <stdio.h> //ipc.c
struct msg_buffer
{
	long mtype;
	char mtext[100];
} msgp, msgp2; //Vengono dichiarati due buffer di messaggio

int main(void)
{
	msgp.mtype = 20; //Viene impostato il tipo di messaggio mtype del buffer msgp su 20
	strcpy(msgp.mtext,"This is a message"); //Testo del messaggio
	key_t queue1Key = ftok("/tmp/unique", 1); //Viene generata una chiave per la coda dei messaggi con il percorso /tmp/unique e l'ID progetto 1
	int queueId = msgget(queue1Key , 0777 | IPC_CREAT | IPC_EXCL); //Viene creata una coda di messaggi con la chiave generata, i permessi 0777 ed i flag IPC_CREAT | IPC_EXCL
	int esito = msgsnd(queueId , &msgp, sizeof(msgp.mtext),0); //Viene inviato il messaggio contenuto nel buffer msgp alla coda dei messaggi
	esito = msgrcv(queueId , &msgp2, sizeof(msgp2.mtext),20,0); //Viene ricevuto un messaggio dalla coda dei messaggi con il tipo 20 e viene salvato nel buffer msgp2
	printf("Received %s\n",msgp2.mtext); //Viene stampato il testo del messaggio ricevuto
}

Esempio comunicazione - 2

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <stdio.h>
#include <string.h> //ipcBook.c
typedef struct book
{
	char title[10];
	char description[200];
	short chapters;
} Book;

struct msg_buffer
{
	long mtype;
	Book mtext;
} msgp_snd, msgp_rcv; //Two different message buffers

int main(void)
{
	msgp_snd.mtype = 20; //Viene impostato il tipo di messaggio mtype del buffer msgp_snd su 20
	strcpy(msgp_snd.mtext.title,"Title"); //Viene impostato il titolo
	strcpy(msgp_snd.mtext.description,"This is a description"); //Viene impostata la descrizione
	msgp_snd.mtext.chapters = 17; //Viene impostato il numero dei capitoli del libro
	key_t queue1Key = ftok("/tmp/unique", 1); //Viene generata una chiave per la coda dei messaggi  con il percorso /tmp/unique e l'ID progetto 1
	int queueId = msgget(queue1Key , 0777 | IPC_CREAT); //Viene creata una coda di messaggi con la chiave generata, i permessi 0777 e l'opzione IPC_CREAT
	int esito = msgsnd(queueId , &msgp_snd, sizeof(msgp_snd.mtext),0); //Viene inviato il messaggio contenuto nel buffer msgp_snd alla coda dei messaggi
	esito = msgrcv(queueId , &msgp_rcv, sizeof(msgp_rcv.mtext),20,0); //Viene ricevuto un messaggio dalla coda dei messaggi con il tipo 20 e viene salvato nel buffer msgp_rcv
	printf("Received: %s %s %d\n",msgp_rcv.mtext.title, msgp_rcv.mtext.description, msgp_rcv.mtext.chapters); //Vengono stampati il titolo, la descrizione e il numero di capitoli del libro ricevuto
}

Esempio comunicazione - 3

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <cstdio>
#include <stdlib.h>
#include <string.h> //ipcType.c
struct msg_buffer
{
	long mtype;
	char mtext[100];
} msgp_snd, msgp_rcv; //Two different message buffers

int main(int argc, char ** argv)
{
	int to_fetch = atoi(argv[1]); //Converte il primo argomento della riga di comando (che rappresenta il tipo di messaggio da ricevere) in un intero
	key_t queue1Key = ftok("/tmp/unique", 1); //Viene generata una chiave per la coda dei messaggi  con il percorso /tmp/unique e l'ID progetto 1
	int queueId = msgget(queue1Key , 0777 | IPC_CREAT); //Viene creata una coda di messaggi con la chiave generata, i permessi 0777 e l'opzione IPC_CREAT
	msgp_snd.mtype = 20;
	strcpy(msgp_snd.mtext,"A message of type 20");
	int esito = msgsnd(queueId , &msgp_snd, sizeof(msgp_snd.mtext),0); //Invia il messaggio msgp_snd alla coda di messaggi identificata dall'ID queueId. Il terzo argomento indica la lunghezza del messaggio da inviare e il quarto e' un flag di opzioni
	msgp_snd.mtype = 10; //Re-use the same message
	strcpy(msgp_snd.mtext,"Another message of type 10");
	esito = msgsnd(queueId , &msgp_snd, sizeof(msgp_snd.mtext),0);
	esito = msgrcv(queueId , &msgp_rcv, sizeof(msgp_rcv.mtext), to_fetch, 0); //Riceve un messaggio dalla coda di messaggi identificata dall'ID queueId e lo memorizza nel buffer msgp_rcv. Il terzo argomento indica la dimensione massima del messaggio da ricevere, il quarto p il tipo di messaggio da ricevere e l'ultimo e' un flag di opzioni
	printf("Received: %s\n", msgp_rcv.mtext);
}

./ipcType 10 #Per testarlo

Modificare la coda

int msgctl(int msqid, int cmd, struct msqid_ds *buf);

Modifica la coda identificata da msqid secondo i comandi cmd, riempiendo buf con informazioni sulla coda (ad esempio timestamp dell’ultima scrittura, dell’ultima lettura, numero messaggi nella coda, etc…). Valori di cmd possono essere:

msqid_ds structure

struct msqid_ds 
{
	struct ipc_perm msg_perm; //Struttura di tipo ipc_perm che contiene informazioni sui permessi e la proprietà della coda di messaggi
	time_t msg_stime; //Rappresenta l'ora dell'ultimo invio di un messaggio alla coda utilizzando la funzione msgsnd. msgsnd(2) indica la sezione del manuale in cui è documentata la funzione
	time_t msg_rtime; //Rappresenta l’ora dell’ultimo ricevimento di un messaggio dalla coda utilizzando la funzione msgrcv. msgrcv(2)
	time_t msg_ctime; //Rappresenta l’ora della creazione della coda o dell’ultima modifica della coda utilizzando la funzione msgctl
	unsigned long msg_cbytes; //Rappresenta il numero di byte attualmente presenti nella coda di messaggi
	msgqnum_t msg_qnum; //Rappresenta il numero di messaggi attualmente presenti nella coda di messaggi
	msglen_t msg_qbytes; //Rappresenta il numero massimo di byte che possono essere presenti nella coda di messaggi
	pid_t msg_lspid; //Rappresenta il PID (Process ID) del processo che ha inviato l’ultimo messaggio alla coda utilizzando la funzione msgsnd. msgsnd(2)
	pid_t msg_lrpid; //Rappresenta il PID (Process ID) del processo che ha ricevuto l’ultimo messaggio dalla coda utilizzando la funzione msgrcv. msgrcv(2)
};

ipc_perm structure

struct ipc_perm //Memorizza informazioni sui permessi e la proprieta' di un oggetto IPC (Interprocess Communication), come una coda di messaggi, un segmento di memoria condivisa o un semaforo
{
	key_t __key; //Rappresenta la chiave fornita alla funzione msgget (o altre funzioni simili per creare oggetti IPC) per creare l'oggetto IPC. msgget(2)
	uid_t uid; //Rappresenta l'UID (User ID) effettivo del proprietario dell'oggetto IPC
	gid_t gid; //Rappresenta il GID (Group ID) effettivo del proprietario dell'oggetto IPC
	uid_t cuid; //Rappresenta l'UID (User ID) effettivo del creatore dell'oggetto IPC
	gid_t cgid; //Rappresenta rappresenta il GID (Group ID) effettivo del creatore dell'oggetto IPC
	unsigned short mode; //Rappresenta i permessi dell'oggetto IPC. Questo campo puo' essere impostato utilizzando le costanti simboliche per i permessi dei file, come S_IRUSR, S_IWUSR, ecc
	unsigned short __seq; //Rappresenta il numero di sequenza dell'oggetto IPC. Questo campo e' utilizzato internamente dal sistema operativo e non dovrebbe essere modificato direttamente dall'utente
};

Esempio modifica coda

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <fcntl.h>
#include <cstdio>

int main(void)
{
	struct msqid_ds mod; //Dichiara una variabile di tipo msqid_ds per memorizzare le informazioni sulla coda di messaggi
	int esito = open("/tmp/unique", O_CREAT, 0777); //Apre (o crea, se non esiste) un file chiamato /tmp/unique con i permessi 0777 (tutti gli utenti hanno i permessi di lettura, scrittura ed esecuzione sul file)
	key_t queue1Key = ftok("/tmp/unique", 1); //Genera una chiave per la coda di messaggi utilizzando la funzione ftok con il percorso /tmp/unique e l'ID progetto 1
	int queueId = msgget(queue1Key , IPC_CREAT | S_IRWXU ); //Crea una coda di messaggi con la chiave generata. L'opzione IPC_CREAT specifica che se la coda di messaggi identificata dalla chiave fornita non esiste deve essere creata, mentre S_IRWXU specifica i permessi iniziali della coda di messaggi se viene creata una nuova coda, in modo che il proprietario della coda abbia i permessi di lettura, scrittura ed esecuzione sulla coda
	msgctl(queueId,IPC_RMID,NULL); //Svuota la coda di messaggi utilizzando la funzione msgctl con il comando IPC_RMID
	queueId = msgget(queue1Key , IPC_CREAT | S_IRWXU ); //Ricrea la coda di messaggi con la stessa chiave e l'opzione IPC_CREAT utilizzando la funzione msgget 
	esito = msgctl(queueId,IPC_STAT,&mod); //Ottiene informazioni sulla coda di messaggi utilizzando la funzione msgctl con il comando IPC_STAT e memorizza queste informazioni nella struttura mod
	printf("Current permission on queue: %d\n",mod.msg_perm.mode); //Stampa i permessi correnti della coda utilizzando il campo mode della struttura msg_perm
	mod.msg_perm.mode = 0000; //Modifica i permessi della coda impostando il campo mode della struttura msg_perm a 0000, che rimuove tutti i permessi dalla coda
	esito = msgctl(queueId,IPC_SET,&mod); //Modifica i permessi della coda utilizzando la funzione msgctl con il comando IPC_SET
	printf("Current permission on queue: %d\n\n",mod.msg_perm.mode); //Stampa nuovamente i permessi correnti della coda
}

Esempio 4

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <string.h>
#include <stdio.h>
#include <unistd.h>
#include <wait.h> //ipc4.c

struct msg_buffer
{
	long mtype;
	char mtext[100];
} msgpSND,msgpRCV;

int main()
{
	struct msqid_ds mod; //Variabile utilizzata per modificare le proprieta' della coda dei messaggi
	msgpSND.mtype = 1;
	strcpy(msgpSND.mtext,"This is a message from sender");
	key_t queue1Key = ftok("/tmp/unique", 1); //Genera una chiave univoca basata sul percorso del file "/tmp/unique" e sul numero di progetto 1
	int queueId = msgget(queue1Key , 0777 | IPC_CREAT); //Crea una coda di messaggi con la chiave specificata e i permessi 0777. L'ID della coda creata viene memorizzato nella variabile queueId
	msgctl(queueId,IPC_RMID,NULL); //Rimuove la coda dei messaggi se esiste
	queueId = msgget(queue1Key , 0777 | IPC_CREAT); //Crea una coda di messaggi con la chiave specificata e i permessi 0777. L'ID della coda creata viene memorizzato nuovamente nella variabile queueId
	msgsnd(queueId, &msgpSND, sizeof(msgpSND.mtext),0); //Invia un messaggio alla coda di messaggi. Il primo argomento e' l'ID della coda di messaggi, il secondo argomento e' un puntatore al buffer del messaggio, il terzo argomento e' la dimensione del testo del messaggio e il quarto argomento e' una flag che specifica opzioni aggiuntive 
	msgsnd(queueId, &msgpSND, sizeof(msgpSND.mtext),0); //Send msg
	msgsnd(queueId, &msgpSND, sizeof(msgpSND.mtext),0); //Send msg
	msgctl(queueId,IPC_STAT,&mod); //Recupera le informazioni sulla coda di messaggi con l'opzione IPC_STAT e le memorizza nella variabile mod
	printf("Msg in queue: %ld\nCurrent max bytes in queue: %ld\n\n", mod.msg_qnum, mod.msg_qbytes); //Stampa il numero di messaggi nella coda e il numero massimo di byte nella coda
	mod.msg_qbytes = 200; //Imposta il campo msg_qbytes della variabile mod a 200 per modificare il numero massimo di byte nella coda
	msgctl(queueId,IPC_SET,&mod); //Applica le modifiche alla coda di messaggi
	printf("Msg in queue: %ld --> same number\nCurrent max bytes in queue: %ld\n\n",mod.msg_qnum, mod.msg_qbytes);
	if( fork() != 0 ) //Viene creato un processo figlio ed il padre esegue l'if
	{
		printf("[SND] Sending 4th message with a full queue...\n"); //Indica che un quarto messaggio viene inviato alla coda di messaggi
		msgsnd(queueId, &msgpSND, sizeof(msgpSND.mtext),0); //Invia un quarto messaggio alla coda di messaggi
		printf("[SND] msg sent\n"); //Indica che il quarto messaggio e' stato inviato alla coda di messaggi
		printf("[SND] Sending 5th message with IPC_NOWAIT\n"); //Indica che un quinto messaggio viene inviato alla coda di messaggi con l'opzione IPC_NOWAIT
		if(msgsnd(queueId, &msgpSND, sizeof(msgpSND.mtext),IPC_NOWAIT )== -1) //Invia un quinto messaggio alla coda di messaggi con l'opzione IPC_NOWAIT, ovvero che se la coda di messaggi e' piena e non puo' accettare ulteriori messaggi, questa chiamata a msgsnd() restituisce -1 e non blocca
		{
			perror("Queue is full --> Error"); //Se la coda di messaggi e' piena viene stampato un messaggio di errore
		}
	} else //Blocco eseguito dal processo figlio
	{
		sleep(3); //Il processo figlio attende 3 secondi
		msgrcv(queueId, &msgpRCV, sizeof(msgpRCV.mtext),1,0); //Riceve un messaggio dalla coda di messaggi. Il primo argomento e' l'ID della coda di messaggi, il secondo argomento e' un puntatore al buffer del messaggio, il terzo argomento e' la dimensione del testo del messaggio, il quarto argomento e' il tipo di messaggio da ricevere (in questo caso 1) e il quinto argomento e' una flag che specifica opzioni aggiuntive
		printf("[Reader] Received msg 1 with msg '%s'\n",msgpRCV.mtext); //Stampa un messaggio che indica che il primo messaggio e' stato ricevuto dalla coda di messaggi e mostra il testo del messaggio ricevuto
		sleep(3);
		msgrcv(queueId, &msgpRCV, sizeof(msgpRCV.mtext),1,0);
		printf("[Reader] Received msg 2 with msg '%s'\n",msgpRCV.mtext);
		sleep(3);
		msgrcv(queueId, &msgpRCV, sizeof(msgpRCV.mtext),1,0);
		printf("[Reader] Received msg 3 with msg '%s'\n",msgpRCV.mtext);
		sleep(3);
		msgrcv(queueId, &msgpRCV, sizeof(msgpRCV.mtext),1,0);
		printf("[Reader] Received msg 4 with msg '%s'\n",msgpRCV.mtext);
		sleep(3);
		if(msgrcv(queueId, &msgpRCV, sizeof(msgpRCV.mtext),1,IPC_NOWAIT)== -1) //Tenta di ricevere un quinto messaggio dalla coda di messaggi, se la coda e' vuota e non ci sono ulteriori messaggi da ricevere restituisce -1 e non blocca (altrimenti verrebbe bloccata l'esecuzione del processo finche' non ci fosse un messaggio da ricevere)
		{
			perror("Queue is empty --> Error");
		}else
		{
			printf("[Reader] Received msg 5 with msg '%s'\n", msgpRCV.mtext); //Se la chiamata a msgrcv non restituisce -1 (cioe' se c'e' un quinto messaggio da ricevere), questa riga stampa un messaggio sullo standard output che indica che il quinto messaggio e' stato ricevuto dalla coda di messaggi e mostra il testo del messaggio ricevuto
		}
	}
	while(wait(NULL)>0); //Attende la terminazione di tutti i processi figli
	
	return 0;
}