Colonnello Mortimer: È un posto ideale per un'imboscata. Lo prenderemo tra due fuochi. Il Monco: Ah già... tu dall'esterno, io dall'interno ... come al solito...
E siamo arrivati alla seconda parte della Trilogia del dollaro… oops: la Trilogia del IPC, da non confondere con i mitici tre film del grandissimo Sergio Leone (film che sono un must-view per qualsiasi cinefilo che si rispetti). Il titolo di questo articolo somiglia molto a un altro che trattava un argomento abbastanza diverso. Non è che sono a corto di idee, ma, dividendo in tre parti la trama della POSIX IPC, non posso fare a meno di citare una delle più belle trilogie della storia del Cinema (quello con C maiuscola con il nostro amato C Language). Dovete scusarmi. E, se non volete scusarmi, stavolta vi mando a casa Il Monco per convincervi...
E come anticipato dalla frase introduttiva, questa volta tratteremo altre due POSIX IPC, la Message Queue e la UNIX domain socket (IPC socket) (nella prima parte avevamo parlato di FIFO (Named Pipe), ricordate?). Cominceremo dalla Message Queue che, come già detto, è:
Message Queue: comunicazione tra due o più processi con capacità full duplex. I processi comunicano tra loro pubblicando un messaggio e recuperandolo dalla coda. Una volta recuperato, il messaggio non è più disponibile nella coda.
Il meccanismo di uso è molto (ma molto) simile a quello della FIFO e, praticamente, basta sostituire le chiamate open(), close(), read(), write() e remove() con le corrispondenti chiamate della famiglia mq: mq_open(), mq_close(), mq_send(), mq_receive() e mq_unlink(). Ok, ci siamo: possiamo vedere in una sola botta l’header data.h, il padre processes.c e i due figli writer.c e reader.c… vai col codice!
#ifndef DATA_H
#define DATA_H // path della message queue
#define MQUEUE_PATH "/mymqueue" // numero di messaggi da scambiare per il benchmark
#define N_MESSAGES 2000000 // struttura Data per i messaggi
typedef struct { unsigned long index; // indice dei dati char text[1024]; // testo dei dati
} Data; #endif /* DATA_H */
// processes.c - main processo padre
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <unistd.h>
#include <mqueue.h>
#include <sys/wait.h>
#include "data.h" // funzione main()
int main(int argc, char* argv[])
{ // set attributi della message queue struct mq_attr ma; ma.mq_flags = 0; // flags: 0 o O_NONBLOCK ma.mq_maxmsg = 10; // numero massimo di messaggi permessi nella coda ma.mq_msgsize = sizeof(Data); // massimo size di un messaggio (in byte) ma.mq_curmsgs = 0; // numero di messaggi attualmente nella coda // creo la message queue if (mq_open(MQUEUE_PATH, O_RDWR | O_CREAT, 0700, &ma) == -1) { // errore di creazione printf("%s: non posso creare la message queue (%s)\n", argv[0], strerror(errno)); exit(EXIT_FAILURE); } // crea i processi figli pid_t pid1, pid2; (pid1 = fork()) && (pid2 = fork()); // test pid processi if (pid1 == 0) { // sono il figlio 1 printf("sono il figlio 1 (%d): eseguo il nuovo processo\n", getpid()); char *pathname = "reader"; char *newargv[] = { pathname, NULL }; execv(pathname, newargv); exit(EXIT_FAILURE); // exec non ritorna mai } else if (pid2 == 0) { // sono il figlio 2 printf("sono il figlio 2 (%d): eseguo il nuovo processo\n", getpid()); char *pathname = "writer"; char *newargv[] = { pathname, NULL }; execv(pathname, newargv); exit(EXIT_FAILURE); // exec non ritorna mai } else if (pid1 > 0 && pid2 > 0) { // sono il padre printf("sono il padre (%d): attendo la terminazione dei figli\n", getpid()); int status; pid_t wpid; while ((wpid = wait(&status)) > 0) printf("sono il padre (%d): figlio %d terminato (%d)\n", getpid(), (int)wpid, status); // rimuovo la message queue ed esco printf("%s: processi terminati\n", argv[0]); mq_unlink(MQUEUE_PATH); exit(EXIT_SUCCESS); } else { // errore nella fork(): rimuovo la message queue ed esco printf("%s: fork error (%s)\n", argv[0], strerror(errno)); mq_unlink(MQUEUE_PATH); exit(EXIT_FAILURE); }
}
// writer.c - main processo figlio
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <unistd.h>
#include <mqueue.h>
#include "data.h" // funzione main()
int main(int argc, char *argv[])
{ // apro la message queue in modo scrittura printf("processo %d partito\n", getpid()); mqd_t mq; if ((mq = mq_open(MQUEUE_PATH, O_WRONLY)) == -1) { // errore di apertura printf("%s: non posso aprire la message queue (%s)\n", argv[0], strerror(errno)); exit(EXIT_FAILURE); } // loop di scrittura messaggi per il reader Data my_data; my_data.index = 0; do { // test index per forzare l'uscita if (my_data.index == N_MESSAGES) { // il processo chiude la message queue ed esce per indice raggiunto printf("processo %d terminato (text=%s index=%ld)\n", getpid(), my_data.text, my_data.index); mq_close(mq); exit(EXIT_SUCCESS); } // compongo il messaggio e lo invio my_data.index++; snprintf(my_data.text, sizeof(my_data.text), "un-messaggio-di-test:%ld", my_data.index); } while (mq_send(mq, (char *)(&my_data), sizeof(Data), 1) != -1); // il processo chiude la message queue ed esce per altro motivo (errore) printf("processo %d terminato con errore (%s)\n", getpid(), strerror(errno)); mq_close(mq); exit(EXIT_FAILURE);
}
// reader.c - main processo figlio
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <unistd.h>
#include <mqueue.h>
#include <time.h>
#include <sys/time.h>
#include "data.h" // funzione main()
int main(int argc, char *argv[])
{ // apro la message queue in modo lettura printf("processo %d partito\n", getpid()); mqd_t mq; if ((mq = mq_open(MQUEUE_PATH, O_RDONLY)) == -1) { // errore di apertura printf("%s: non posso aprire la message queue (%s)\n", argv[0], strerror(errno)); exit(EXIT_FAILURE); } // set clock e time per calcolare il tempo di CPU e il tempo di sistema clock_t t_start = clock(); struct timeval tv_start; gettimeofday(&tv_start, NULL); // loop di lettura messaggi dal writer Data my_data; while (mq_receive(mq, (char *)(&my_data), sizeof(Data), NULL) != -1) { // test index per forzare l'uscita if (my_data.index == N_MESSAGES) { // get clock e time per calcolare il tempo di CPU e il tempo di sistema clock_t t_end = clock(); double t_passed = ((double)(t_end - t_start)) / CLOCKS_PER_SEC; struct timeval tv_end, tv_elapsed; gettimeofday(&tv_end, NULL); timersub(&tv_end, &tv_start, &tv_elapsed); // il processo chiude la message queue ed esce per indice raggiunto printf("reader: ultimo messaggio ricevuto: %s\n", my_data.text); printf("processo %d terminato " "(index=%ld CPU time elapsed: %.3f - total time elapsed:%ld.%ld)\n", getpid(), my_data.index, t_passed, tv_elapsed.tv_sec, tv_elapsed.tv_usec / 1000); mq_close(mq); exit(EXIT_SUCCESS); } } // il processo chiude la message queue ed esce per altro motivo (errore) printf("processo %d terminato con errore (%s)\n", getpid(), strerror(errno)); mq_close(mq); exit(EXIT_FAILURE);
}
Tutto chiaro? Sicuramente, perché, come promesso, è quasi sovrapponibile al codice visto nell’ultimo articolo, quindi è veramente molto semplice da interpretare, supponendo che anche il primo esempio lo fosse (e lo spero vivamente!). La differenza maggiore (e più interessante) sta nell’inizializzazione della struttura mq_attr (che ho messo nel processo padre), che permette, rispetto alla Named Pipe, una maggiore personalizzazione del funzionamento. Quindi, semplicità d’uso e e flessibilità e… i risultati? Vediamoli:
sono il padre (19567): attendo la terminazione dei figli sono il figlio 1 (19568): eseguo il nuovo processo sono il figlio 2 (19569): eseguo il nuovo processo processo 19568 partito processo 19569 partito processo 19569 terminato (text=un-messaggio-di-test:2000000 index=2000000) reader: ultimo messaggio ricevuto: un-messaggio-di-test:2000000 processo 19568 terminato (index=2000000 CPU time elapsed: 1.832 - total time elapsed:1.860) sono il padre (19567): figlio 19568 terminato (0) sono il padre (19567): figlio 19569 terminato (0) ./processes: processi terminati
Non male, no? È solo leggermente più lenta della Named Pipe (che era velocissima). In questo caso abbiamo un messaggio ogni 0,92 us (e con la Pipe era uno ogni 0,87 us). Insomma, la Message Queue passa a pieni voti l’esame per modo d’uso e prestazioni e si pone allo stesso livello della Fifo.
Ok, e adesso a che POSIX IPC tocca? Ma alla UNIX Domain Socket! E anche in questo caso può tornare utile un piccolo promemoria:
UNIX domain socket (IPC socket): comunicazione tra processi che usa i socket del Kernel (che non usano TCP/IP). Nel nostro caso (IPC tra processi locali) è preferibile, per ovvi motivi, usare questi socket invece dei Network socket.
Il meccanismo di uso è, ancora una volta, simile a quello della Fifo, anche se il modello utilizzato è il classico dei socket Server/Client. Comunque, per mantenere la coerenza della trattazione, ho mantenuto la nomenclatura reader/writer riuscendo a ottenere, tutto sommato, un codice abbastanza simile a quello degli esempi già visti. E anche in questo caso vediamo in una botta sola l’header data.h, il padre processes.c e i due figli writer.c e reader.c… vai col codice!
#ifndef DATA_H
#define DATA_H // path del file per ipc socket
#define IPCS_PATH "myipcs" // numero di messaggi da scambiare per il benchmark
#define N_MESSAGES 2000000 // struttura Data per i messaggi
typedef struct { unsigned long index; // indice dei dati char text[1024]; // testo dei dati
} Data; #endif /* DATA_H */
// processes.c - main processo padre
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <unistd.h>
#include <sys/wait.h>
#include "data.h" // funzione main()
int main(int argc, char* argv[])
{ // crea i processi figli pid_t pid1, pid2; (pid1 = fork()) && (pid2 = fork()); // test pid processi if (pid1 == 0) { // sono il figlio 1 printf("sono il figlio 1 (%d): eseguo il nuovo processo\n", getpid()); char *pathname = "reader"; char *newargv[] = { pathname, NULL }; execv(pathname, newargv); exit(EXIT_FAILURE); // exec non ritorna mai } else if (pid2 == 0) { // sono il figlio 2 printf("sono il figlio 2 (%d): eseguo il nuovo processo\n", getpid()); char *pathname = "writer"; char *newargv[] = { pathname, NULL }; execv(pathname, newargv); exit(EXIT_FAILURE); // exec non ritorna mai } else if (pid1 > 0 && pid2 > 0) { // sono il padre printf("sono il padre (%d): attendo la terminazione dei figli\n", getpid()); int status; pid_t wpid; while ((wpid = wait(&status)) > 0) printf("sono il padre (%d): figlio %d terminato (%d)\n", getpid(), (int)wpid, status); // esco printf("%s: processi terminati\n", argv[0]); exit(EXIT_SUCCESS); } else { // errore nella fork(): esco printf("%s: fork error (%s)\n", argv[0], strerror(errno)); exit(EXIT_FAILURE); }
}
// writer.c - main processo figlio
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <unistd.h>
#include <sys/socket.h>
#include <sys/un.h>
#include "data.h" // funzione main()
int main(int argc, char *argv[])
{ // creo il socket in modo IPC e Datagram printf("processo %d partito\n", getpid()); int sock; if ((sock = socket(AF_UNIX, SOCK_DGRAM, 0)) == -1) { // errore di creazione printf("%s: non posso creare il socket (%s)\n", argv[0], strerror(errno)); exit(EXIT_FAILURE); } // prepara la struttura sockaddr_un per il reader (è un server) remoto struct sockaddr_un reader; memset(&reader, 0, sizeof(struct sockaddr_un)); reader.sun_family = AF_UNIX; strcpy(reader.sun_path, IPCS_PATH); // loop di scrittura messaggi per il reader Data my_data; my_data.index = 0; do { // test index per forzare l'uscita if (my_data.index == N_MESSAGES) { // il processo chiude il socket ed esce per indice raggiunto printf("processo %d terminato (text=%s index=%ld)\n", getpid(), my_data.text, my_data.index); close(sock); exit(EXIT_SUCCESS); } // compongo il messaggio e lo invio my_data.index++; snprintf(my_data.text, sizeof(my_data.text), "un-messaggio-di-test:%ld", my_data.index); } while (sendto(sock, &my_data, sizeof(my_data), 0, (struct sockaddr *)&reader, sizeof(reader)) != -1); // il processo chiude il socket ed esce per altro motivo (errore) printf("processo %d terminato con errore (%s)\n", getpid(), strerror(errno)); close(sock); exit(EXIT_FAILURE);
}
// reader.c - main processo figlio
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <unistd.h>
#include <time.h>
#include <sys/time.h>
#include <sys/socket.h>
#include <sys/un.h>
#include "data.h" // funzione main()
int main(int argc, char *argv[])
{ // creo il socket in modo IPC e Datagram printf("processo %d partito\n", getpid()); int sock; if ((sock = socket(AF_UNIX, SOCK_DGRAM, 0)) == -1) { // errore di creazione printf("%s: non posso creare il socket (%s)\n", argv[0], strerror(errno)); exit(EXIT_FAILURE); } // prepara la struttura sockaddr_un per questo reader (è un server) struct sockaddr_un reader; memset(&reader, 0, sizeof(struct sockaddr_un)); reader.sun_family = AF_UNIX; strcpy(reader.sun_path, IPCS_PATH); // associa l'indirizzo del reader al socket (questo crea il file IPCS_PATH) if (bind(sock, (struct sockaddr *)&reader, sizeof(reader)) == -1) { // errore di bind printf("%s: errore di bind (%s)", argv[0], strerror(errno)); exit(EXIT_FAILURE); } // set clock e time per calcolare il tempo di CPU e il tempo di sistema clock_t t_start = clock(); struct timeval tv_start; gettimeofday(&tv_start, NULL); // loop di lettura messaggi dal writer Data my_data; while (recv(sock, &my_data, sizeof(my_data), 0) != -1) { // test index per forzare l'uscita if (my_data.index == N_MESSAGES) { // get clock e time per calcolare il tempo di CPU e il tempo di sistema clock_t t_end = clock(); double t_passed = ((double)(t_end - t_start)) / CLOCKS_PER_SEC; struct timeval tv_end, tv_elapsed; gettimeofday(&tv_end, NULL); timersub(&tv_end, &tv_start, &tv_elapsed); // il processo chiude e cancella il socket ed esce per indice raggiunto printf("reader: ultimo messaggio ricevuto: %s\n", my_data.text); printf("processo %d terminato " "(index=%ld CPU time elapsed: %.3f - total time elapsed:%ld.%ld)\n", getpid(), my_data.index, t_passed, tv_elapsed.tv_sec, tv_elapsed.tv_usec / 1000); close(sock); unlink(IPCS_PATH); exit(EXIT_SUCCESS); } } // il processo chiude e cancella il socket ed esce per altro motivo (errore) printf("processo %d terminato con errore (%s)\n", getpid(), strerror(errno)); close(sock); unlink(IPCS_PATH); exit(EXIT_FAILURE);
}
Visto? È un esempio abbastanza classico di Server + Client come quello già mostrato in alcuni miei vecchi articoli (qui e qui), solo che in questo caso ho optato per l’uso del modo datagram-oriented che, con l’obiettivo di una comunicazione veloce in locale, è sicuramente il più adatto e fornisce le prestazioni migliori (la “reliability” in UNIX Domain è molto alta, e non c’è motivo di scegliere la versione stream-oriented). Rispetto ai vari esempi di di questo tipo che si trovano in rete faccio notare alcuni “trucchetti” interessanti che spesso si omettono:
- Anche se la famiglia di funzioni mq_ è orientata a manipolare messaggi di testo, è possibile “ingannare” le funzioni per manipolare quello che si vuole: nel nostro caso vogliamo scrivere/leggere una struct Data e per realizzarlo è sufficiente fare (nelle send/recv) un cast a char* del buffer: non bisogna mai dimenticare che una struttura, anche complessa, è pur sempre un insieme (anzi, un array) di byte, quindi questa operazione è del tutto legale (ma non cercate mai di spedire strutture che contengono pointer… questa si che non è una buona idea).
- Nel reader bisogna rimuovere, in chiusura, il file (corrispondente al socket) che crea automaticamente la bind(): senza questo passo le chiamate successive al reader fallirebbero (con un errore EINVAL “The socket is already bound to an address”).
- Nel reader si può usare la recv() invece della recvFrom() (che è comune nei Server datagram-oriented) perché il nostro reader si limita a leggere e non scrive risposte, quindi non gli interessa l’indirizzo del writer (ottenibile usando la rcvfrom()). Tra l’altro si potrebbe addirittura usare la read() (con qualche accorgimento, però). Infatti, come dice il manuale della recv(2) di Linux:
The only difference between recv() and read(2) is the presence of flags. With a zero flags argument, recv() is generally equivalent to read(2) (but see NOTES). Also, the following call recv(sockfd, buf, len, flags); is equivalent to recvfrom(sockfd, buf, len, flags, NULL, NULL);
E adesso è venuto il momento dei risultati… squillino le trombe!
sono il padre (20035): attendo la terminazione dei figli sono il figlio 1 (20036): eseguo il nuovo processo sono il figlio 2 (20037): eseguo il nuovo processo processo 20036 partito processo 20037 partito processo 20037 terminato (text=un-messaggio-di-test:2000000 index=2000000) reader: ultimo messaggio ricevuto: un-messaggio-di-test:2000000 processo 20036 terminato (index=2000000 CPU time elapsed: 3.607 - total time elapsed:4.552) sono il padre (20035): figlio 20037 terminato (0) sono il padre (20035): figlio 20036 terminato (0) ./processes: processi terminati
Ahi ahi ahi… pare che la IPC socket, sia pur veloce in assoluto (scambia un messaggio ogni 1,8 us) è un po’ più lenta delle Fifo e Message queue. E vabbè, ce ne faremo una ragione: promossa ma con riserva, anche considerando che è un po’ più complicata da usare.
E anche per oggi abbiamo terminato. Nel prossimo articolo parleremo dell’ultimo metodo in analisi, la Shared Memory, e, se avanzano tempo e voglia, faremo anche un interessante confronto con una versione multithread molto simile. Stay tuned!
Ciao, e al prossimo post!
https://italiancoders.it/per-qualche-ipc-in-piu-considerazioni-sulle-prestazioni-della-posix-ipc-pt-2/