Pipes, named pipes and message queues

Signals can tell a process that something happened, but they carry no data. This session is about moving data between processes. After fork() the parent and the child have separate memory – a variable changed in one is not changed in the other – so they need the kernel to pass bytes between them.

Three tools do that: an anonymous pipe between related processes, a named pipe (FIFO) that any process can open by its file name, and a message queue that keeps whole, typed messages. They look similar, but they differ in exactly the details that make programs hang: who holds which end, when a read returns, and what happens to the boundaries between messages.

The model in one breath: a pipe is a one-way byte stream in the kernel; every process closes the ends it does not use, and the reader must know how many bytes to expect. A message queue keeps message boundaries and lets the receiver pick messages by type.

pipe(): two ends of a kernel buffer

pipe(fd) creates a buffer inside the kernel and gives you two file descriptors for it: fd[0] to read and fd[1] to write (think of 0 as stdin, 1 as stdout). Bytes written into fd[1] come out of fd[0] in the same order (Fig. 1). You use them with the ordinary read(), write() and close() calls from the first session.

int fd[2];
if (pipe(fd) == -1) {
    perror("pipe");
    exit(EXIT_FAILURE);
}
/* fd[0]: the read end    fd[1]: the write end */

The rules that decide whether your program works or hangs:

A pipe has no name, so the only way to share it is inheritance: create it before fork(), and the child gets copies of both descriptors.

Close what you do not use

Right after fork() both processes hold both ends: four descriptors point at one pipe (Fig. 2). For a pipe from parent to child, the parent closes the read end and the child closes the write end. What remains is one writer and one reader.

This is not tidiness. read() returns end of file only when every write end is closed – in every process. A child that loops until EOF but forgot to close its own inherited write end is waiting for itself: the writer that keeps the pipe alive is the reader (Fig. 3).

pid_t pid = fork();
if (pid == 0) {                                /* child: reader */
    close(fd[1]);                              /* not writing – close it, or EOF never comes */
    char buf[64];
    ssize_t n;
    while ((n = read(fd[0], buf, sizeof(buf))) > 0)
        write(STDOUT_FILENO, buf, n);
    close(fd[0]);
    exit(EXIT_SUCCESS);
}
close(fd[0]);                                  /* parent: writer, not reading */
write(fd[1], "hello\n", 6);
close(fd[1]);                                  /* this is the child's EOF */
waitpid(pid, NULL, 0);

Parent and child: one pipe per direction

A pipe is one-way. If the parent asks and the child answers, you need two pipes (Fig. 4). Using one pipe for both directions does not work: a process can read back its own message before the other side gets it.

In the session example the parent sends a random number, the child reads it and sends back a sentence. The parent closes the child's ends right after fork(), so if the child dies without answering, the parent's read() returns 0 instead of waiting forever, and the buffer is terminated by the number of bytes actually read:

pid_t child_pid = child(io_pipes1[0], io_pipes[1]);   /* child reads 1[0], writes [1] */

close(io_pipes1[0]);            /* the parent uses neither of the child's ends */
close(io_pipes[1]);

write(io_pipes1[1], &r, sizeof(r));                  /* send an int */

ssize_t n = read(io_pipes[0], pufferchar, sizeof(pufferchar) - 1);
if (n == -1) { perror("read"); exit(EXIT_FAILURE); }
pufferchar[n] = '\0';           /* n == 0 (EOF): empty string, not garbage */

int arrived;
ssize_t got = read(pipe_id_rec, &arrived, sizeof(int));
if (got != sizeof(int)) {                     /* -1: error, 0: EOF, less: partial */
    fprintf(stderr, "no complete number arrived\n");
    exit(EXIT_FAILURE);
}
char str[] = "Child kuldi a szulonek: Megkaptam a szamot!";
write(pipe_id_send, str, sizeof(str));        /* sizeof an ARRAY: all chars + '\0' */

The child checks that it got a whole int: read() returning 0 or 2 bytes would leave arrived half-filled with garbage. Finally the parent waits for the child with waitpid() – a pipe does not replace waiting.

Sending numbers, strings and structs

A pipe moves raw bytes. You can write the memory of any variable – an int, an array, a struct – as long as the reader reads the same number of bytes into a variable of the same type: write(fd, &r, sizeof(r)) on one side, read(fd, &x, sizeof(x)) on the other. Strings need care, because the size is easy to get wrong:

char  text[] = "hello";       /* an array of 6 bytes */
char *msg    = "hello";       /* a pointer to them */

write(fd, text, sizeof(text));      /* 6 bytes: h e l l o \0 – correct */
write(fd, msg, sizeof(msg));        /* 8 bytes on 64-bit: the size of the POINTER – wrong */
write(fd, msg, strlen(msg) + 1);    /* 6 bytes – correct for a pointer */

The second problem is boundaries. A pipe is a stream: it does not remember where one write() ended. Two writes can arrive in one read(), and one large write can arrive in several (Fig. 5). Writes of up to PIPE_BUF bytes (4096 on Linux) are atomic – they are not mixed with other writers' data – but a read may still return less than you asked for.

So the reader must know how many bytes belong together. Either both sides use fixed sizes (an int, a char[100], a struct), or the sender sends the length first:

/* sender */
int len = strlen(name) + 1;
write(fd, &len, sizeof(len));       /* first: how many bytes follow */
write(fd, name, len);               /* then: exactly that many */

/* receiver */
int len;
read_full(fd, &len, sizeof(len));
read_full(fd, buf, len);            /* buf must be at least len bytes */

And the reader loops until it has the whole thing:

/* read exactly size bytes, or fail: a pipe may return less than asked for */
int read_full(int fd, void *buf, size_t size)
{
    char *p = buf;
    while (size > 0) {
        ssize_t n = read(fd, p, size);
        if (n == -1 && errno == EINTR) continue;   /* interrupted by a signal: retry */
        if (n <= 0) return -1;                     /* error, or EOF too early */
        p += n;
        size -= n;
    }
    return 0;
}

Structs and arrays of structs

A struct is sent as its bytes. That works only if the struct contains values, not pointers: a char *name is an address in the sender's memory and means nothing in the receiver. Use char name[50]. An array of structs is one block of memory, so a single write() sends all of them:

struct worker {
    int  id;
    char name[50];          /* arrays, not char * – a pointer means nothing in another process */
    char address[50];
};

struct worker team[3] = { {1, "Bela", "Budapest"}, {2, "Janos", "Miskolc"}, {3, "Mark", "Szeged"} };
int count = 3;
write(fd, &count, sizeof(count));
write(fd, team, count * sizeof(struct worker));      /* the whole array in one write */

Child to child

Two children cannot create a pipe for each other – a pipe created inside child 1 is invisible to child 2. The parent creates all pipes before it forks, and every process closes the ends it does not use (Fig. 6). The parent too: if it keeps the write end of the child-to-child pipe open, child 2 never sees end of file.

int p1[2], p12[2];            /* parent -> child1, child1 -> child2 */
pipe(p1);
pipe(p12);                      /* both BEFORE the forks, so every child inherits them */

if (fork() == 0) {              /* child1 */
    close(p1[1]); close(p12[0]);        /* keeps: p1 read, p12 write */
    /* read from p1[0], write to p12[1] */
}
if (fork() == 0) {              /* child2 */
    close(p1[0]); close(p1[1]); close(p12[1]);   /* keeps: p12 read */
    /* read from p12[0] until EOF */
}
close(p1[0]); close(p12[0]); close(p12[1]);      /* parent keeps: p1 write */

Named pipes (FIFO)

An anonymous pipe exists only as inherited descriptors. A named pipe, or FIFO, also has a name in the file system, so any process that knows the name can open it – related or not (Fig. 7). mkfifo(path, mode) creates the name; from then on it is used with the ordinary open(), read(), write() and close().

char pipename[32];
snprintf(pipename, sizeof(pipename), "/tmp/%d", getpid());   /* unique per run */

if (mkfifo(pipename, S_IRUSR | S_IWUSR) == -1) {             /* 0600: only the owner */
    perror("mkfifo");           /* EEXIST: a file with this name is already there */
    exit(EXIT_FAILURE);
}

The mode S_IRUSR | S_IWUSR (0600) lets only the owner open it – reduced further by the umask, as for any new file. The name contains the PID, so two runs (or two students on the same server) do not collide in /tmp. If the name already exists, mkfifo fails with EEXIST.

The important difference from a file: open() blocks. Opening for reading waits until someone opens it for writing, and the other way round. That is the synchronisation point – the two processes meet at open().

/* child – writer */
int fd = open(pipename, O_WRONLY);     /* blocks until someone opens it for reading */
write(fd, "Test value\n", 12);
close(fd);

/* parent – reader */
int fd = open(pipename, O_RDONLY);     /* blocks until someone opens it for writing */
ssize_t n = read(fd, pufferchar, sizeof(pufferchar) - 1);
pufferchar[n > 0 ? n : 0] = '\0';
close(fd);
unlink(pipename);                       /* remove the name – the FIFO file stays otherwise */

The FIFO's name stays in the file system after the processes exit, so remove it with unlink() when you are done (the open descriptors keep working until closed). From the shell you can try the same thing with two commands:

$ mkfifo /tmp/demo
$ ls -l /tmp/demo
prw-r--r-- 1 student student 0 … /tmp/demo     <- type p, size 0: nothing is stored on disk
$ cat /tmp/demo &                              <- the reader waits
$ echo hello > /tmp/demo                       <- the writer: cat prints hello
$ rm /tmp/demo

System V message queues

A message queue is a list of messages kept by the kernel. Unlike a pipe it keeps message boundaries – one msgsnd is one msgrcv – and every message carries a type, a positive long that the receiver can select by (Fig. 8).

Creating or opening a queue

A queue is identified by a numeric key. ftok(path, id) makes one from an existing file's identity and a non-zero project number – every process that calls it with the same path and number gets the same key. msgget(key, 0600 | IPC_CREAT) creates the queue if it does not exist yet and returns its ID (msqid); the 0600 are the permissions, as for a file.

key_t key = ftok(".", 1);             /* an existing path + a non-zero project id */
if (key == -1) { perror("ftok"); exit(EXIT_FAILURE); }

int queue = msgget(key, 0600 | IPC_CREAT);   /* create it, or open the existing one */
if (queue < 0) { perror("msgget"); exit(EXIT_FAILURE); }

The message and its size

The message is your own struct, with one rule: it starts with a long mtype. After that comes the payload – text, an array, anything. The size you pass to msgsnd and msgrcv counts only the payload, never the long in front of it (Fig. 9).

#define MSG_SIZE 1024

struct message {
    long mtype;               /* must be > 0 – the receiver selects by it */
    char mtext[MSG_SIZE];     /* the payload: any bytes, any layout you like */
};

struct message ms;
ms.mtype = 5;
strcpy(ms.mtext, "This is a test sentence.");

/* size = the payload only, WITHOUT the long mtype; +1 for the closing '\0' */
if (msgsnd(queue, &ms, strlen(ms.mtext) + 1, 0) < 0) {    /* 0: block if the queue is full */
    perror("msgsnd");
    exit(EXIT_FAILURE);
}

mtype must be greater than 0, otherwise msgsnd fails with EINVAL. The last argument is the flag: 0 blocks while the queue is full, IPC_NOWAIT returns -1 with EAGAIN instead.

Receiving by type

struct message ms;
/* the first message of type 5; blocks until there is one */
ssize_t len = msgrcv(queue, &ms, MSG_SIZE, 5, 0);
if (len < 0)
    perror("msgrcv");
else
    printf("type %ld: %s\n", ms.mtype, ms.mtext);

msgrcv blocks until a matching message exists (unless IPC_NOWAIT), and returns the number of payload bytes. If the message is longer than the size you gave, it fails with E2BIG and the message stays in the queue. With types, one queue can serve several conversations: in msg-queue2.c the parent sends type 5 and listens for type 4, the child the other way round, so neither reads back its own message. The payload there is an int array:

struct uzenet2 {
    long mtype;
    int  marr[5];
};
struct uzenet2 uz2 = {5, {1, 2, 3, 4, 5}};

msgsnd(queue, &uz2, sizeof(uz2.marr), 0);                  /* 20 bytes: the array only */
msgrcv(queue, &uz2, sizeof(uz2.marr), 5, 0);               /* in the other process */

Removing the queue

A queue is not tied to a process. It stays in the kernel – with any unread messages – after every process has exited, until someone removes it or the machine restarts. A stale queue from a crashed run is found again by the next run with the same key, old messages included. Remove it when the work is done:

/* after every process is done with it – usually the parent, after waitpid */
if (msgctl(queue, IPC_RMID, NULL) < 0)
    perror("msgctl");

$ ipcs -q                       # your message queues
------ Message Queues --------
key        msqid      owner      perms      used-bytes   messages
0x0118063d 3          student    600        25           1
$ ipcrm -q 3                    # remove the one with msqid 3

Which one to use

A rule of thumb: a pipe for a stream from one process to another, two pipes for a question and an answer, a message queue when several kinds of messages share a channel or a receiver must pick a particular one. Shared memory, the fastest of all, comes in the next session.

pipe

A one-way byte buffer in the kernel with a read end and a write end, shared by inheritance.

EOF

End of file: read() returns 0 because every write end is closed.

framing

Telling the reader where one piece of data ends – fixed sizes or a length prefix.

FIFO

A named pipe: a pipe with a name in the file system, created with mkfifo.

mtype

The positive long at the start of a message; msgrcv selects messages by it.

IPC_RMID

The msgctl command that deletes a message queue.

Practice

Parent says: 1
Child says: 1
Parent says: 2
Child says: 2
Parent says: 3
Child says: 3

enum command { COUNT, STOP };

/* parent */
for (int i = 1; i <= 3; i++) {
    printf("Parent says: %d\n", i);
    fflush(stdout);
    enum command c = COUNT;
    write(to_child[1], &c, sizeof(c));
    read(from_child[0], &c, sizeof(c));      /* wait for "done" – stricter than sleep(1) */
}

$ ./child_communications
3
Child_1 read message: 3
Child_2 read message: 3
Child_2 read random number: 749
Child_2 read random number: 489
Child_2 read random number: 529

/* child2: first the count, then exactly that many numbers */
int n;
read_full(p12[0], &n, sizeof(n));
for (int i = 0; i < n; i++) {
    int value;
    if (read_full(p12[0], &value, sizeof(value)) == -1) break;
    printf("Child_2 read random number: %d\n", value);
}

Child: the job Disinfection is done by 3 workers.
Received: Id: 1 and Name: Bela and Address: Budapest
Received: Id: 2 and Name: Janos and Address: Miskolc
Received: Id: 3 and Name: Mark and Address: Szeged

char job[100];
int count;
read_full(fd, job, sizeof(job));                    /* fixed-size name: both sides agree on 100 */
read_full(fd, &count, sizeof(count));

struct worker *list = malloc(count * sizeof(struct worker));
read_full(fd, list, count * sizeof(struct worker));
/* … print them … */
free(list);

Summary

• —pipe(fd): fd[0] reads, fd[1] writes; create it before fork() so the child inherits it.
• —Right after fork(), every process closes the ends it does not use. read() returns 0 only when every write end is closed.
• —One pipe, one direction: two-way talk needs two pipes; child-to-child pipes are created by the parent before both forks.
• —A pipe forgets write boundaries. Use fixed sizes or a length prefix, check what read() returned, and loop until you have it all. sizeof of a pointer is not the length of the string.
• —A FIFO is a pipe with a name: mkfifo, open() blocks until the other side opens, unlink() at the end.
• —Message queues keep boundaries and types: ftok → msgget → msgsnd/msgrcv (size without mtype, type > 0) → msgctl(IPC_RMID).
• —fflush(stdout) before fork(), and wait for every child – no IPC replaces waitpid.