Межпроцессное взаимодействие в C++: рабочие приёмы

Задача звучит просто: два процесса на одной машине должны обмениваться данными. Один пишет, другой читает. На собеседовании тут говорят «ну, через сокет» — и получают систему, которая сериализует и копирует мегабайты там, где хватило бы указателя. Или наоборот: берут разделяемую память ради двух сообщений в секунду и потом неделю ловят гонку на неинициализированном мьютексе.

Ниже — четыре механизма IPC для Linux и C++17/20, у каждого своя цена: скорость, сложность синхронизации и то, насколько больно чинить, когда один из процессов упал в середине записи. Речь про обмен внутри одной машины; сетевое взаимодействие между хостами — отдельная тема, здесь её не трогаем.

Четыре механизма и когда какой

Разберём то, что реально применяется в проде под Linux: анонимные и именованные каналы (pipe/FIFO), POSIX message queue, разделяемую память (POSIX shm) и локальный сокет (Unix domain socket). Windows-примитивы вроде именованных каналов и CreateFileMapping устроены иначе — держите в голове, что код ниже про POSIX.

Каналы: когда данных немного и порядок важен

Анонимный pipe между родителем и потомком после fork — самый простой и надёжный вариант, если поток однонаправленный. Ядро само разруливает конкуренцию: атомарность записи гарантируется для блоков до PIPE_BUF байт (на Linux это 4096). Пишете больше — данные могут перемешаться с чужой записью, если писателей несколько.

// Компиляция: g++ -std=c++17 pipe_demo.cpp -o pipe_demo
// Linux, glibc
#include <unistd.h>
#include <sys/wait.h>
#include <cstring>
#include <cstdio>

int main() {
    int fds[2];                 // fds[0] — чтение, fds[1] — запись
    if (pipe(fds) == -1) { perror("pipe"); return 1; }

    pid_t pid = fork();
    if (pid == 0) {             // потомок читает
        close(fds[1]);          // закрываем неиспользуемый конец записи
        char buf[256];
        ssize_t n = read(fds[0], buf, sizeof(buf) - 1);
        if (n > 0) { buf[n] = '\0'; printf("child got: %s\n", buf); }
        close(fds[0]);
        return 0;
    }

    close(fds[0]);              // родитель пишет
    const char* msg = "hello from parent";
    write(fds[1], msg, std::strlen(msg));
    close(fds[1]);             // закрытие даёт читателю EOF
    waitpid(pid, nullptr, nullptr);
    return 0;
}

Ключевой момент, на котором спотыкаются: закрывайте лишние концы в каждом процессе. Если родитель не закроет fds[1], читатель никогда не получит EOF и read будет висеть вечно — в pipe остаётся открытый writer.

Для связи неродственных процессов вместо анонимного pipe берут именованный (mkfifo) — файл в ФС, который открывают оба.

POSIX message queue: когда нужны границы сообщений

Через сокет типа SOCK_STREAM или через pipe вы получаете поток байт — где кончается одно сообщение, решаете сами (длина префиксом, разделитель). POSIX mq хранит именно сообщения: одно mq_send — одно mq_receive, с приоритетом. Плюс — можно задать максимальный размер сообщения и глубину очереди при создании.

// Компиляция: g++ -std=c++17 mq_demo.cpp -o mq_demo -lrt
#include <mqueue.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <cstdio>
#include <cstring>

int main() {
    mq_attr attr{};
    attr.mq_maxmsg = 10;        // глубина очереди
    attr.mq_msgsize = 256;      // макс. размер одного сообщения

    // Имя очереди обязано начинаться со слэша
    mqd_t mq = mq_open("/demo_queue", O_CREAT | O_RDWR, 0644, &attr);
    if (mq == (mqd_t)-1) { perror("mq_open"); return 1; }

    const char* msg = "task-42";
    // последний аргумент — приоритет
    if (mq_send(mq, msg, std::strlen(msg), 1) == -1) perror("mq_send");

    char buf[256];
    unsigned prio = 0;
    ssize_t n = mq_receive(mq, buf, sizeof(buf), &prio);
    if (n > 0) { buf[n] = '\0'; printf("got %s prio=%u\n", buf, prio); }

    mq_close(mq);
    mq_unlink("/demo_queue");    // удаляет очередь из системы
    return 0;
}

Грабли номер один: буфер в mq_receive должен быть не меньше mq_msgsize, иначе получите EMSGSIZE и ничего не прочитаете. Размер берите из mq_getattr, а не угадывайте. Очередь живёт в системе до mq_unlink или перезагрузки — забудете отцепить, и на следующем запуске mq_open без O_EXCL подхватит старую очередь со старыми атрибутами.

Разделяемая память: ноль копирований, но синхронизация на вас

Единственный механизм из четырёх, где данные не идут через ядро при каждой передаче: оба процесса маппят один и тот же объект в своё адресное пространство и работают с ним как с обычным указателем. Отсюда и цена — ядро больше ничего не сериализует, любую синхронизацию вы пишете руками.

Разделяемая память сама по себе не передаёт данные — она даёт общий буфер. Всё, ради чего люди берут IPC (кто когда что записал), вы строите поверх неё сами: мьютекс, семафор, условная переменная в этом же буфере.

// Компиляция: g++ -std=c++17 shm_demo.cpp -o shm_demo -lrt -lpthread
#include <sys/mman.h>
#include <fcntl.h>
#include <unistd.h>
#include <pthread.h>
#include <cstring>
#include <cstdio>

struct SharedBlock {
    pthread_mutex_t mtx;        // процессно-разделяемый мьютекс
    int counter;
};

int main() {
    int fd = shm_open("/demo_shm", O_CREAT | O_RDWR, 0644);
    if (fd == -1) { perror("shm_open"); return 1; }
    if (ftruncate(fd, sizeof(SharedBlock)) == -1) { perror("ftruncate"); return 1; }

    void* addr = mmap(nullptr, sizeof(SharedBlock),
                      PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
    if (addr == MAP_FAILED) { perror("mmap"); return 1; }
    auto* block = static_cast<SharedBlock*>(addr);

    // Инициализация мьютекса с атрибутом PTHREAD_PROCESS_SHARED —
    // без него мьютекс не работает между процессами
    pthread_mutexattr_t mattr;
    pthread_mutexattr_init(&mattr);
    pthread_mutexattr_setpshared(&mattr, PTHREAD_PROCESS_SHARED);
    pthread_mutex_init(&block->mtx, &mattr);
    pthread_mutexattr_destroy(&mattr);

    pthread_mutex_lock(&block->mtx);
    block->counter += 1;
    pthread_mutex_unlock(&block->mtx);

    munmap(addr, sizeof(SharedBlock));
    close(fd);
    // shm_unlink("/demo_shm"); — вызывает тот, кто отвечает за жизненный цикл
    return 0;
}

Про PTHREAD_PROCESS_SHARED забывают чаще всего. Обычный мьютекс, инициализированный по умолчанию, между процессами не синхронизирует — работает как обычная переменная в разделяемом буфере, то есть никак. И вторая мина: если процесс упал, держа pthread_mutex_lock, второй встанет намертво. Лечится атрибутом PTHREAD_MUTEX_ROBUST и обработкой EOWNERDEAD — иначе получаете вечный залипший лок.

Unix domain socket: сокетный API без сети

Если вам привычен сокетный код и нужна двусторонняя связь неродственных процессов — берите Unix socket. Скорость близка к pipe, семантика — как у сети, но без стека TCP/IP. Уникальная фича: через SCM_RIGHTS можно передать другому процессу открытый файловый дескриптор. Так, например, воркеру отдают уже принятое соединение.

Что ломается на практике

• Зомби-объекты в /dev/shm и списке mq. shm_open и mq_open создают именованные объекты, которые переживают процесс. Упали, не сделав shm_unlink — на следующем запуске подхватите чужой мусор. Проверяйте ls /dev/shm.
• SIGPIPE при записи в закрытый pipe/socket. Читатель отвалился, вы пишете — процесс по умолчанию получает SIGPIPE и умирает. Игнорируйте сигнал (signal(SIGPIPE, SIG_IGN)) или пишите с флагом MSG_NOSIGNAL и разбирайте EPIPE.
• Указатели и виртуальные функции в shared memory. Объект с указателями внутри бесполезен: адрес, валидный в одном процессе, в другом указывает в никуда, потому что базовый адрес маппинга разный. Только POD-типы, смещения вместо указателей, никаких std::string и виртуальных таблиц.
• Частичное чтение/запись. read и write могут вернуть меньше запрошенного. Оборачивайте в цикл, пока не прочитаете/запишете нужное число байт, либо обрабатывайте частичный результат.
• Права доступа. Если процессы работают под разными пользователями, режим 0644 на shm или mq не даст второму записать. Считайте маску заранее.

Что настроить дальше

Если пишете shared memory всерьёз — начните с robust-мьютексов и продумайте протокол восстановления после падения писателя, это дороже самого обмена. Для быстрого старта без ручного менеджмента примитивов посмотрите boost::interprocess: он оборачивает shm, именованные мьютексы и очереди в RAII-обёртки и сам чистит объекты. Меряйте не «сообщений в секунду» абстрактно, а латентность round-trip на вашем размере сообщения: для мелких частых сообщений выигрыш shared memory над сокетом может не окупить сложность синхронизации.