Embedded HandbookApps137. Communication between processes: pipes, sockets, queues, shared memory
Each process has its own memory, and that is a protection: a failure in one does not touch
the others. But they do have to exchange data, and Unix has
accumulated many ways to do it — from the | pipe in the shell to shared memory and passing
open files through a socket. All of them are run here in stm-sys:1
(_verify/sys137/), with numbers.
Terms used in the documentation
Term Meaning inter-process communication (IPC) exchanging data between processes pipe a byte stream from the write end to the read end named pipe, FIFO a pipe with a name in the file system UNIX domain socket a socket within one machine, named by a file ancillary data, SCM_RIGHTSauxiliary socket data; passing descriptors message queue a queue of typed messages in the kernel semaphore a counter you can wait on shared memory memory mapped into several processes XSI / System V IPC the old queues, semaphores and memory with ftokkeysPOSIX IPC POSIX queues, semaphores and shared memory with names like /name— newer than XSImemory-mapped file a file mapped into memory ring buffer a circular buffer
137.1 The pipe
pipe creates two descriptors: fd[0] for reading, fd[1] for writing. On its own
a pipe is useless — it is meant to be used with fork: the child inherits both
ends, and each side closes the one it does not need:
int fd[2];
if (pipe(fd) < 0) { perror("pipe"); return 1; }
pid_t p = fork();
if (p == 0) {
close(fd[0]); /* потомок только пишет */
const char msg[] = "привет из потомка";
write(fd[1], msg, sizeof msg);
close(fd[1]);
_exit(0);
}
close(fd[1]); /* родитель только читает; иначе read не увидит конец */
char buf[64];
ssize_t n = read(fd[0], buf, sizeof buf);
ёмкость канала 65536 байт, PIPE_BUF 4096 (запись не длиннее — атомарна)
родитель прочитал 33 байт: привет из потомка
второй read: 0 (конец: пишущих концов нет)
Two numbers define the behavior. The capacity is 65536 bytes (16 pages):
a writer that has filled the pipe blocks until the reader takes the data.
PIPE_BUF is 4096: a write no longer than this many bytes goes into the pipe
whole, without being interleaved with the writes of other writers; a longer one may
be split. read returns 0 only when all the
write ends are closed — so the parent must close its own fd[1], otherwise
it will wait for itself. Writing to a pipe with no readers gives SIGPIPE.
popen does all of this in one line — fork, exec sh -c, the pipe —
and returns a FILE*: popen("printf '3\n1\n2\n' | sort -n | tr '\n' ' '", "r")
reads as 1 2 3.
A FIFO is the same pipe, but with a name: mkfifo("/tmp/demo.fifo", 0600)
creates a file for which S_ISFIFO is true. Unrelated processes
communicate through it; open for reading blocks until someone
opens it for writing.
137.2 The UNIX socket: a two-way pipe and passing a descriptor
A socket of the AF_UNIX family is like a network socket (section 78), but without a network: the address is
a path in the file system, and socketpair gives an already connected pair at once.
It is bidirectional, it supports both streams (SOCK_STREAM) and messages
(SOCK_DGRAM, SOCK_SEQPACKET), and one thing found nowhere else:
it can pass open descriptors. The child opens a file and
sends the descriptor in an SCM_RIGHTS ancillary message:
struct cmsghdr *c = CMSG_FIRSTHDR(&m);
c->cmsg_level = SOL_SOCKET; c->cmsg_type = SCM_RIGHTS; c->cmsg_len = CMSG_LEN(sizeof f);
memcpy(CMSG_DATA(c), &f, sizeof f);
sendmsg(sv[1], &m, 0);
The parent receives its own descriptor number, pointing to the same
open file (/etc/debian_version — the system version in the container):
родитель получил дескриптор 4, прочитано: 12.15
This is how servers hand connections from the main process to workers, and how access to devices goes through a privileged broker.
137.3 XSI: queue, semaphore, shared memory
The old set is three kinds of objects with integer identifiers that live in
the kernel independently of the processes. The key is made from a path and a letter (ftok),
or IPC_PRIVATE is used for related processes:
key_t key = ftok("/tmp", 'q');
int q = msgget(IPC_PRIVATE, IPC_CREAT | 0600);
int s = semget(IPC_PRIVATE, 1, IPC_CREAT | 0600);
int m = shmget(IPC_PRIVATE, 4096, IPC_CREAT | 0600);
The key from ftok is only printed here: related processes are fine with IPC_PRIVATE.
Unrelated processes need a common key —
msgget(ftok("/var/run/myapp", 'q'), IPC_CREAT | 0600). Use a path of your own,
not the shared /tmp, otherwise the key will collide with another program's key.
The child writes a number to the shared memory, raises the semaphore and puts two messages of different types into the queue; the parent waits on the semaphore and reads:
struct sembuf down = { 0, -1, 0 };
semop(s, &down, 1); /* ждём, пока потомок не поднимет */
printf("из общей памяти: %d\n", shared[0]);
struct msg mm;
msgrcv(q, &mm, sizeof mm.text, 2, 0); /* сначала тип 2 — очередь выбирает по типу */
из общей памяти: 12345
msgrcv тип 2: сообщение типа 2
msgrcv любой: сообщение типа 1
сегмент 4096 байт, подключений 1
ресурсы удалены: без IPC_RMID они пережили бы процесс
A queue is not a stream but messages with a type: msgrcv with type 2 takes
the second message before the first. The kernel limits: a message up to 8192
bytes (msgmax), a queue up to 16384 (msgmnb).
The main trouble with XSI objects is that they are not removed together with the processes: a process
crashed — the queue, the semaphore and the segment stay in the kernel (the segment with zero
attachments), and they are removed with ipcrm or with IPC_RMID from the program.
Named POSIX objects behave the same way — they stay in /dev/shm until
shm_unlink/sem_unlink or a reboot, but at least you can see them as files
and remove them with rm. To avoid leftovers, the name is removed as soon as all the
participants have opened the object, and related processes are fine with
MAP_SHARED | MAP_ANONYMOUS or an unnamed memfd. XSI semaphores
also work as arrays with a union semun that you have to declare
yourself. For new code, use the POSIX set.
137.4 POSIX: shm_open, sem_open, mmap
POSIX objects are named like files (/demo137) and in Linux they live in
/dev/shm. Shared memory is a file that is mapped with mmap:
int fd = shm_open("/demo137", O_CREAT | O_RDWR, 0600);
ftruncate(fd, 4096);
long *cnt = mmap(NULL, 4096, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
sem_t *sem = sem_open("/demo137", O_CREAT, 0600, 1);
Four processes increment a shared counter 100000 times each:
4 процесса × 100000 под семафором: 400000 (ожидалось 400000)
без семафора: 180980 — потеряно 219020 приращений
объект лежит в /dev/shm: да
Without the semaphore part of the increments were lost — 55 % in this run: (*cnt)++
is a read, an add and a write, and another process squeezes in between them. Shared memory
is the only way to exchange data without copying through the kernel, and the only one
where synchronization is entirely up to the program. The number of lost increments varies
from run to run — that is exactly what a race is.
shm_unlink and sem_unlink remove the name, like unlink for a file; the object itself
lives as long as someone still has it open.
137.5 mmap: a file as memory and a ring buffer
mmap maps a file into the address space: reading is a memory
access, and so is writing; msync flushes to disk:
char *p = mmap(NULL, page, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
strcpy(p, "записано через mmap");
msync(p, page, MS_SYNC);
страница 4096 байт; из файла прочитано: записано через mmap
MAP_SHARED — changes are visible to others and go to the file; MAP_PRIVATE —
copy-on-write, the file does not change; MAP_ANONYMOUS — no file, this is how
malloc gets large blocks (section 135). mprotect changes the permissions of
pages, madvise hints to the kernel how the memory will be read.
A trick with two mappings solves the eternal ring-buffer problem —
writing across the edge. The same physical memory is mapped twice
in a row, and the byte at ring[i] is also visible at ring[i + page]:
int mfd = memfd_create("ring", 0);
ftruncate(mfd, page);
char *ring = mmap(NULL, 2 * page, PROT_NONE, MAP_PRIVATE | MAP_ANONYMOUS, -1, 0);
mmap(ring, page, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_FIXED, mfd, 0);
mmap(ring + page, page, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_FIXED, mfd, 0);
ring[0] = A, ring[page] = A — один и тот же байт по двум адресам
через край: конец abc начало def
A memcpy of six bytes at address page − 3 put abc at the end of the buffer and
def at the start, without a single boundary check. Audio buffers and
network queues are built this way; a microcontroller has nothing like this — there the edge
of the buffer is checked by hand (section 20).
137.6 What to choose
| Task | Tool |
|---|---|
| a byte stream to or from a child | pipe, popen |
| a stream between unrelated processes | FIFO or UNIX socket |
| messages with boundaries, both ways | UNIX socket SOCK_SEQPACKET |
| pass an open file or connection | UNIX socket, SCM_RIGHTS |
| a large volume without copying | shm_open + mmap + semaphore |
| just wait for an event | semaphore sem_open, eventfd |
| one instance of a process per machine | fcntl F_SETLK on a file (section 138) |
| the same machine, but over the network too | TCP socket (section 78) |
Pipes and sockets copy the data through the kernel twice and give you synchronization for free; shared memory does not copy and does not give synchronization. The XSI set is for reading other people's code, not for new code.
137.7 Pitfalls
| Symptom | Cause |
|---|---|
read from a pipe never returns 0 |
the write end is open in the reader itself — close it after fork |
| the writer hangs | the pipe is full (65536 bytes), and the reader is not reading |
| two writers mixed up their lines | writes longer than PIPE_BUF (4096) are not atomic |
the process is killed by SIGPIPE |
a write to a pipe or socket with no reader |
open of a FIFO for reading hangs |
there is not a single writer; O_NONBLOCK, or open it for writing yourself |
after the program crashed, ipcs shows leftovers |
XSI objects are not removed together with the processes — ipcrm, IPC_RMID |
after a crash, files are left in /dev/shm |
named POSIX objects live until shm_unlink/sem_unlink — remove the name right after opening |
| a counter in shared memory counts wrong | no semaphore: increments are lost — more than half in the experiments |
changes in mmap did not reach the file |
MAP_PRIVATE instead of MAP_SHARED, or no msync before the check |
the process is killed by SIGBUS when accessing a mapping |
the file is shorter than the mapping — mmap itself does not check this; ftruncate to the needed size before mmap |
137.8 Summary
- A pipe: a capacity of 65536 bytes,
PIPE_BUFof 4096 is atomic,readgives 0 only when the write ends are closed;popen, FIFO. - A UNIX socket works both ways and can pass descriptors (
SCM_RIGHTS). - XSI (
msgget,semget,shmget) uses keys; POSIX (shm_open,sem_open) uses names in/dev/shm. Both outlive a crashed process until they are removed (IPC_RMID,shm_unlink). - Shared memory without a semaphore loses increments (55 % in the run).
mmap: a file as memory; two mappings in a row give a ring buffer without an edge check.
Full sources
pipe_fifo.c — a pipe between parent and child, the pipe capacity, PIPE_BUF, popen and FIFO.
/* Раздел 137: канал между родителем и потомком; ёмкость канала и PIPE_BUF; popen; FIFO. */
#define _GNU_SOURCE
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <fcntl.h>
#include <limits.h>
#include <sys/wait.h>
#include <sys/stat.h>
int main(void)
{
int fd[2];
if (pipe(fd) < 0) { perror("pipe"); return 1; }
printf("ёмкость канала %d байт, PIPE_BUF %d (запись не длиннее — атомарна)\n",
fcntl(fd[0], F_GETPIPE_SZ), PIPE_BUF);
pid_t p = fork();
if (p == 0) {
close(fd[0]); /* потомок только пишет */
const char msg[] = "привет из потомка";
write(fd[1], msg, sizeof msg);
close(fd[1]);
_exit(0);
}
close(fd[1]); /* родитель только читает; иначе read не увидит конец */
char buf[64];
ssize_t n = read(fd[0], buf, sizeof buf);
printf("родитель прочитал %zd байт: %s\n", n, buf);
n = read(fd[0], buf, sizeof buf); /* все пишущие концы закрыты — конец файла */
printf("второй read: %zd (конец: пишущих концов нет)\n", n);
close(fd[0]); wait(NULL);
/* popen: канал к команде */
FILE *f = popen("printf '3\\n1\\n2\\n' | sort -n | tr '\\n' ' '", "r");
if (fgets(buf, sizeof buf, f)) printf("popen: %s\n", buf);
pclose(f);
/* FIFO: именованный канал в файловой системе */
unlink("/tmp/demo.fifo");
mkfifo("/tmp/demo.fifo", 0600);
struct stat st; stat("/tmp/demo.fifo", &st);
printf("FIFO создан: S_ISFIFO = %d\n", S_ISFIFO(st.st_mode));
if (fork() == 0) { FILE *w = fopen("/tmp/demo.fifo", "w"); fputs("через FIFO\n", w); fclose(w); _exit(0); }
FILE *r = fopen("/tmp/demo.fifo", "r"); /* open блокируется, пока нет писателя */
if (fgets(buf, sizeof buf, r)) printf("прочитано из FIFO: %s", buf);
fclose(r); wait(NULL);
unlink("/tmp/demo.fifo");
return 0;
}
unix_sock.c — a pair of UNIX sockets and passing a descriptor through SCM_RIGHTS.
/* Раздел 137: пара UNIX-сокетов и передача открытого дескриптора через SCM_RIGHTS. */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <fcntl.h>
#include <sys/socket.h>
#include <sys/wait.h>
int main(void)
{
int sv[2];
socketpair(AF_UNIX, SOCK_STREAM, 0, sv);
if (fork() == 0) {
close(sv[0]);
int f = open("/etc/debian_version", O_RDONLY);
char dummy = 'F';
struct iovec io = { &dummy, 1 };
char cbuf[CMSG_SPACE(sizeof f)];
struct msghdr m = { .msg_iov = &io, .msg_iovlen = 1, .msg_control = cbuf, .msg_controllen = sizeof cbuf };
struct cmsghdr *c = CMSG_FIRSTHDR(&m);
c->cmsg_level = SOL_SOCKET; c->cmsg_type = SCM_RIGHTS; c->cmsg_len = CMSG_LEN(sizeof f);
memcpy(CMSG_DATA(c), &f, sizeof f);
sendmsg(sv[1], &m, 0);
_exit(0);
}
close(sv[1]);
char dummy; struct iovec io = { &dummy, 1 };
char cbuf[CMSG_SPACE(sizeof(int))];
struct msghdr m = { .msg_iov = &io, .msg_iovlen = 1, .msg_control = cbuf, .msg_controllen = sizeof cbuf };
recvmsg(sv[0], &m, 0);
int f; memcpy(&f, CMSG_DATA(CMSG_FIRSTHDR(&m)), sizeof f);
char buf[64]; ssize_t n = read(f, buf, sizeof buf - 1); buf[n > 0 ? n : 0] = 0;
printf("родитель получил дескриптор %d, прочитано: %s", f, buf);
wait(NULL);
return 0;
}
sysv_ipc.c — an XSI message queue, semaphore and shared memory.
/* Раздел 137: очередь сообщений, семафор и разделяемая память XSI (System V). */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <sys/sem.h>
#include <sys/shm.h>
#include <sys/wait.h>
struct msg { long mtype; char text[64]; };
int main(void)
{
key_t key = ftok("/tmp", 'q');
int q = msgget(IPC_PRIVATE, IPC_CREAT | 0600);
int s = semget(IPC_PRIVATE, 1, IPC_CREAT | 0600);
int m = shmget(IPC_PRIVATE, 4096, IPC_CREAT | 0600);
printf("ключ ftok %#lx; ids: очередь %d, семафор %d, память %d\n", (long)key, q, s, m);
semctl(s, 0, SETVAL, 0); /* семафор = 0: «данных нет» */
int *shared = shmat(m, NULL, 0);
if (fork() == 0) {
int *sh = shmat(m, NULL, 0);
sh[0] = 12345; /* пишем в общую память */
struct sembuf up = { 0, +1, 0 };
semop(s, &up, 1); /* сигналим: готово */
struct msg mm = { 1, "сообщение типа 1" };
msgsnd(q, &mm, sizeof mm.text, 0);
mm.mtype = 2; strcpy(mm.text, "сообщение типа 2");
msgsnd(q, &mm, sizeof mm.text, 0);
shmdt(sh); _exit(0);
}
struct sembuf down = { 0, -1, 0 };
semop(s, &down, 1); /* ждём, пока потомок не поднимет */
printf("из общей памяти: %d\n", shared[0]);
struct msg mm;
msgrcv(q, &mm, sizeof mm.text, 2, 0); /* сначала тип 2 — очередь выбирает по типу */
printf("msgrcv тип 2: %s\n", mm.text);
msgrcv(q, &mm, sizeof mm.text, 0, 0);
printf("msgrcv любой: %s\n", mm.text);
wait(NULL); /* потомок отключился — считаем подключения */
struct shmid_ds ds; shmctl(m, IPC_STAT, &ds);
printf("сегмент %zu байт, подключений %ld\n", ds.shm_segsz, (long)ds.shm_nattch);
shmdt(shared); shmctl(m, IPC_RMID, NULL); semctl(s, 0, IPC_RMID); msgctl(q, IPC_RMID, NULL);
printf("ресурсы удалены: без IPC_RMID они пережили бы процесс\n");
return 0;
}
posix_shm.c — shm_open, mmap and sem_open: a counter in shared memory with and without a semaphore.
/* Раздел 137: POSIX — shm_open + mmap + sem_open; счётчик в общей памяти под семафором и без. */
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <fcntl.h>
#include <semaphore.h>
#include <sys/mman.h>
#include <sys/wait.h>
#define N 100000
int main(void)
{
shm_unlink("/demo137"); sem_unlink("/demo137");
int fd = shm_open("/demo137", O_CREAT | O_RDWR, 0600);
ftruncate(fd, 4096);
long *cnt = mmap(NULL, 4096, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
sem_t *sem = sem_open("/demo137", O_CREAT, 0600, 1);
*cnt = 0;
for (int k = 0; k < 4; k++)
if (fork() == 0) {
for (int i = 0; i < N; i++) { sem_wait(sem); (*cnt)++; sem_post(sem); }
_exit(0);
}
for (int k = 0; k < 4; k++) wait(NULL);
printf("4 процесса × %d под семафором: %ld (ожидалось %d)\n", N, *cnt, 4 * N);
*cnt = 0;
int go[2];
pipe(go); /* старт по сигналу: иначе дети могут отработать по очереди */
for (int k = 0; k < 4; k++)
if (fork() == 0) {
char c;
close(go[1]);
if (read(go[0], &c, 1) < 0) _exit(1); /* ждём, пока родитель закроет канал */
for (int i = 0; i < N; i++) (*cnt)++;
_exit(0);
}
close(go[1]); /* все четверо стартуют разом */
for (int k = 0; k < 4; k++) wait(NULL);
close(go[0]);
printf("без семафора: %ld — потеряно %ld приращений\n", *cnt, 4L * N - *cnt);
printf("объект лежит в /dev/shm: %s\n", access("/dev/shm/demo137", F_OK) == 0 ? "да" : "нет");
munmap(cnt, 4096); close(fd); sem_close(sem);
shm_unlink("/demo137"); sem_unlink("/demo137");
return 0;
}
mmap_ring.c — mmap of a file and a ring buffer made of two adjacent mappings.
/* Раздел 137: mmap файла и кольцевой буфер из двух соседних отображений одной памяти —
запись через край буфера не требует разбиения на два куска. */
#define _GNU_SOURCE
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <fcntl.h>
#include <sys/mman.h>
int main(void)
{
long page = sysconf(_SC_PAGESIZE);
/* файл через mmap: запись в память = запись в файл после msync */
int fd = open("/tmp/mmap137.bin", O_RDWR | O_CREAT | O_TRUNC, 0600);
ftruncate(fd, page);
char *p = mmap(NULL, page, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
strcpy(p, "записано через mmap");
msync(p, page, MS_SYNC);
munmap(p, page); close(fd);
char buf[64] = {0};
fd = open("/tmp/mmap137.bin", O_RDONLY); read(fd, buf, sizeof buf - 1); close(fd);
printf("страница %ld байт; из файла прочитано: %s\n", page, buf);
/* кольцевой буфер: резервируем 2 страницы адресов, отображаем одну и ту же память дважды */
int mfd = memfd_create("ring", 0);
ftruncate(mfd, page);
char *ring = mmap(NULL, 2 * page, PROT_NONE, MAP_PRIVATE | MAP_ANONYMOUS, -1, 0);
mmap(ring, page, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_FIXED, mfd, 0);
mmap(ring + page, page, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_FIXED, mfd, 0);
ring[0] = 'A';
printf("ring[0] = %c, ring[page] = %c — один и тот же байт по двум адресам\n", ring[0], ring[page]);
memcpy(ring + page - 3, "abcdef", 6); /* запись «через край» одним memcpy */
printf("через край: конец %.3s начало %.3s\n", ring + page - 3, ring);
return 0;
}
Next: section 138 — a daemon, its log and post-mortem debugging.