137. Обмен между процессами: каналы, сокеты, очереди, общая память

У каждого процесса своя память, и это защита: сбой одного не трогает другие. Но обмениваться данными им приходится, и способов в Unix накопилось много — от канала | в оболочке до общей памяти и передачи открытых файлов через сокет. Все они здесь прогнаны в stm-sys:1 (_verify/sys137/), с числами.

Как это называется в документации

Оригинал	Перевод
inter-process communication (IPC)	обмен между процессами
pipe	канал: поток байтов от пишущего конца к читающему
named pipe, FIFO	именованный канал: канал с именем в файловой системе
UNIX domain socket	сокет внутри одной машины, с именем-файлом
ancillary data, SCM_RIGHTS	служебные данные сокета; передача дескрипторов
message queue	очередь сообщений
semaphore	семафор: счётчик с ожиданием
shared memory	общая память
XSI / System V IPC	старые очереди, семафоры и память с ключами ftok
POSIX IPC	очереди, семафоры и общая память POSIX с именами вида /имя — новее, чем XSI
memory-mapped file	файл, отображённый в память
ring buffer	кольцевой буфер

137.1 Канал

pipe создаёт два дескриптора: fd[0] читать, fd[1] писать. Сам по себе канал бесполезен — он нужен вместе с fork: потомок наследует оба конца, и каждая сторона закрывает ненужный ей:

    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 (конец: пишущих концов нет)

Два числа определяют поведение. Ёмкость — 65 536 байт (16 страниц): писатель, заполнивший канал, блокируется, пока читатель не заберёт данные. PIPE_BUF — 4096: запись не длиннее этого числа байт попадает в канал целиком, не перемешиваясь с записями других писателей; длиннее — может быть разрезана. read возвращает 0, только когда закрыты все пишущие концы, — поэтому родитель обязан закрыть свой fd[1], иначе будет ждать себя самого. Запись в канал без читателей — SIGPIPE.

popen делает всё это за одну строку — fork, exec sh -c, канал — и возвращает FILE*: popen("printf '3\n1\n2\n' | sort -n | tr '\n' ' '", "r") читается как 1 2 3.

FIFO — тот же канал, но с именем: mkfifo("/tmp/demo.fifo", 0600) создаёт файл, у которого S_ISFIFO истинно. Через него общаются процессы без родства; open на чтение блокируется, пока кто-нибудь не откроет на запись.

137.2 UNIX-сокет: канал в обе стороны и передача дескриптора

Сокет семейства AF_UNIX — как сетевой (раздел 78), но без сети: адрес — путь в файловой системе, а socketpair даёт сразу соединённую пару. Он двунаправленный, умеет и поток (SOCK_STREAM), и сообщения (SOCK_DGRAM, SOCK_SEQPACKET), и одну вещь, которой нет больше нигде: по нему передают открытые дескрипторы. Потомок открывает файл и отправляет дескриптор служебным сообщением SCM_RIGHTS:

        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);

Родитель получает свой номер дескриптора, указывающий на тот же открытый файл (/etc/debian_version — версия системы в контейнере):

родитель получил дескриптор 4, прочитано: 12.15

Так устроена передача соединений от главного процесса рабочим в серверах и доступ к устройствам через привилегированный посредник.

137.3 XSI: очередь, семафор, общая память

Старый набор — три объекта с целыми идентификаторами, которые живут в ядре независимо от процессов. Ключ делается из пути и буквы (ftok), либо IPC_PRIVATE для родственников:

    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);

Ключ из ftok здесь только печатается: родственникам хватает IPC_PRIVATE. Неродственным процессам нужен общий ключ — msgget(ftok("/var/run/myapp", 'q'), IPC_CREAT | 0600). Путь берут свой, а не общий /tmp, иначе ключ совпадёт с ключом чужой программы.

Потомок пишет число в общую память, поднимает семафор и кладёт в очередь два сообщения разных типов; родитель ждёт семафора и читает:

    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 они пережили бы процесс

Очередь — не поток, а сообщения с типом: msgrcv с типом 2 достаёт второе сообщение раньше первого. Пределы в ядре: сообщение до 8192 байт (msgmax), очередь до 16 384 (msgmnb).

Главная беда XSI-объектов — они не удаляются вместе с процессами: процесс упал — очередь, семафор и сегмент остались в ядре (у сегмента — ноль подключений), и удаляют их ipcrm или IPC_RMID из программы. Именованные POSIX-объекты ведут себя так же — остаются в /dev/shm до shm_unlink/sem_unlink или перезагрузки, но их хотя бы видно как файлы и можно удалить rm. Чтобы мусора не было, имя удаляют сразу, как все участники открыли объект, а родственным процессам хватает MAP_SHARED | MAP_ANONYMOUS или memfd без имени. Семафоры XSI ещё и работают массивами с union semun, который нужно объявлять самому. Для нового кода берут POSIX-набор.

137.4 POSIX: shm_open, sem_open, mmap

POSIX-объекты называются как файлы (/demo137) и в Linux лежат в /dev/shm. Общая память — это файл, который отображают 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);

Четыре процесса по 100 000 раз увеличивают общий счётчик:

4 процесса × 100000 под семафором: 400000 (ожидалось 400000)
без семафора: 180980 — потеряно 219020 приращений
объект лежит в /dev/shm: да

Без семафора часть приращений пропала — в этом прогоне 55 %: (*cnt)++ — это чтение, сложение и запись, и между ними вклинивается другой процесс. Общая память — единственный способ обмена без копирования через ядро, и единственный, где синхронизация целиком на программе. Число потерянных приращений от прогона к прогону разное — это и есть гонка.

shm_unlink и sem_unlink удаляют имя, как unlink у файла; сам объект живёт, пока открыт хоть у кого-то.

137.5 mmap: файл как память и кольцевой буфер

mmap отображает файл в адресное пространство: чтение — обращение к памяти, запись — тоже, msync сбрасывает на диск:

    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 — изменения видны другим и уходят в файл; MAP_PRIVATE — копия при записи, файл не меняется; MAP_ANONYMOUS — без файла, так malloc берёт большие блоки (раздел 135). mprotect меняет права страниц, madvise подсказывает ядру, как память будут читать.

Приём с двумя отображениями решает вечную задачу кольцевого буфера — запись через край. Одну и ту же физическую память отображают дважды подряд, и байт по адресу ring[i] виден и по 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

memcpy шести байт по адресу page − 3 положил abc в конец буфера и def — в начало, без единой проверки границы. Так делают буферы звука и сетевые очереди; на микроконтроллере такого нет — там край буфера проверяют руками (раздел 20).

137.6 Что выбрать

Каналы и сокеты копируют данные через ядро дважды и дают синхронизацию даром; общая память не копирует и синхронизацию не даёт. XSI-набор — для чтения чужого кода, не для нового.

137.7 Грабли

137.8 Итоги

• Канал: 65 536 байт ёмкости, PIPE_BUF 4096 атомарно, read даёт 0 только при закрытых пишущих концах; popen, FIFO.
• UNIX-сокет — в обе стороны и с передачей дескрипторов (SCM_RIGHTS).
• XSI (msgget, semget, shmget) — с ключами; POSIX (shm_open, sem_open) — с именами в /dev/shm. И те и другие переживают упавший процесс, пока их не удалят (IPC_RMID, shm_unlink).
• Общая память без семафора теряет приращения (в прогоне — 55 %).
• mmap: файл как память; два отображения подряд — кольцевой буфер без проверки края.

Полные исходники

pipe_fifo.c — канал между родителем и потомком, ёмкость канала, PIPE_BUF, popen и 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 — пара UNIX-сокетов и передача дескриптора через 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 — очередь сообщений, семафор и разделяемая память XSI.

/* Раздел 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 и sem_open: счётчик в общей памяти под семафором и без.

/* Раздел 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 файла и кольцевой буфер из двух соседних отображений.

/* Раздел 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;
}

Дальше: раздел 138 — демон, его журнал и посмертная отладка.