The ultimate guide to implementing inter-process communication (IPC) in C++

// Example of verification process memory isolation#include <iostream>
#include <>

int global_var = 100;  // Global variables
int main() {
    pid_t pid = fork();
    if (pid == 0) {    // Subprocess        global_var = 200;
        std::cout << "Child global_var: " << global_var 
                  << " Address: " << &global_var << std::endl;
    } else {           // Parent process        sleep(1);      // Make sure the child process executes first        std::cout << "Parent global_var: " << global_var 
                  << " Address: " << &global_var << std::endl;
    }
    return 0;
}

// Server process#include <>
#include <sys/>
#include <>

int main() {
    const char* fifo_path = "/tmp/myfifo";
    
    // Create a named pipe    mkfifo(fifo_path, 0666);
    
    int fd = open(fifo_path, O_WRONLY);
    const char* msg = "Server message";
    write(fd, msg, strlen(msg)+1);
    close(fd);
    
    unlink(fifo_path); // Clean pipeline files    return 0;
}

// Client process#include <>
#include <iostream>

​​​​​​​int main() {
    const char* fifo_path = "/tmp/myfifo";
    int fd = open(fifo_path, O_RDONLY);
    
    char buffer[256];
    read(fd, buffer, sizeof(buffer));
    std::cout << "Received: " << buffer << std::endl;
    
    close(fd);
    return 0;
}

#include <boost/interprocess/managed_shared_memory.hpp>
#include <boost/interprocess/sync/named_mutex.hpp>

using namespace boost::interprocess;

struct HighFrequencyData {
    uint64_t timestamp;
    double   price;
    uint32_t volume;
};

void shm_writer() {
    managed_shared_memory segment(open_or_create, "StockData", 1024*1024);
    auto data = segment.find_or_construct<HighFrequencyData>("HFData")();
    named_mutex mutex(open_or_create, "shm_mutex");
    
    while(running) {
        ();
        // Update market data        data->timestamp = get_timestamp();
        data->price = get_latest_price();
        data->volume = get_trade_volume();
        ();
        std::this_thread::sleep_for(1us);
    }
}

void shm_reader() {
    managed_shared_memory segment(open_only, "StockData");
    auto data = <HighFrequencyData>("HFData").first;
    named_mutex mutex(open_only, "shm_mutex");
    
    while(running) {
        ();
        process_data(*data);
        ();
        std::this_thread::yield();
    }
}

#include <>
#include <iostream>

struct TradeOrder {
    long   order_id;
    char   symbol;
    double price;
    int    quantity;
};

int main() {
    mq_attr attr = {
        .mq_flags = 0,
        .mq_maxmsg = 1000,       // Maximum number of messages        .mq_msgsize = sizeof(TradeOrder),
        .mq_curmsgs = 0
    };
    
    mqd_t mq = mq_open("/order_queue", O_CREAT | O_RDWR, 0644, &attr);
    if(mq == -1) {
        perror("mq_open");
        exit(EXIT_FAILURE);
    }

    // Producer thread    auto producer = []{
        TradeOrder order{/*...*/};
        for(int i=0; i<1000; ++i) {
            mq_send(mq, (char*)&order, sizeof(order), 0);
        }
    };
    
    // Consumer thread    auto consumer = []{
        TradeOrder order;
        while(true) {
            ssize_t bytes = mq_receive(mq, (char*)&order, sizeof(order), nullptr);
            if(bytes == -1) break;
            process_order(order);
        }
    };
    
    // Start the thread...    mq_close(mq);
    mq_unlink("/order_queue");
    return 0;
}

Copy Code
class ReactorServer {
    int epoll_fd;
    std::atomic<bool> running{true};
    
    void start_epoll() {
        epoll_event events[MAX_EVENTS];
        while(running) {
            int n = epoll_wait(epoll_fd, events, MAX_EVENTS, -1);
            for(int i=0; i<n; ++i) {
                if(events[i].events & EPOLLIN) {
                    handle_io(events[i].);
                }
            }
        }
    }
    
public:
    void start() {
        epoll_fd = epoll_create1(0);
        // Add a listening socket to epoll        // Start the worker thread pool        // Start the timer thread    }
};

// Use splice to implement file transfervoid send_file(int out_fd, int in_fd, off_t offset, size_t size) {
    loff_t in_offset = offset;
    while(size > 0) {
        ssize_t transferred = splice(in_fd, &in_offset,
                                     out_fd, nullptr,
                                     size, SPLICE_F_MOVE);
        if(transferred <= 0) break;
        size -= transferred;
    }
}

// protobuf message definitionmessage RpcRequest {
    uint64 request_id = 1;
    string method_name = 2;
    bytes  parameters = 3;
}

message RpcResponse {
    uint64 request_id = 1;
    StatusCode code = 2;
    bytes  result = 3;
}

# perf flame diagram generation processperf record -F 99 -g -- ./my_program
perf script |  > 
  > 

#include <cppcoro/>
using namespace cppcoro;

task<> handle_client(io_service& ios, tcp_socket socket) {
    char buffer[1024];
    for(;;) {
        auto bytes_read = co_await (buffer, sizeof(buffer));
        if(bytes_read == 0) break;
        co_await (buffer, bytes_read);
    }
}

task<> server(io_service& ios, int port) {
    auto listener = tcp_listener::create(ios, tcp_endpoint{ipv4_address::any(), port});
    for(;;) {
        auto socket = co_await ();
        handle_client(ios, std::move(socket));
    }
}

template<typename T, size_t Size>
class LockFreeRingBuffer {
    std::atomic<size_t> write_idx{0};
    std::atomic<size_t> read_idx{0};
    T buffer[Size];
    
public:
    bool push(const T& item) {
        size_t current = write_idx.load(std::memory_order_relaxed);
        size_t next = (current + 1) % Size;
        if(next == read_idx.load(std::memory_order_acquire)) 
            return false;
        buffer[current] = item;
        write_idx.store(next, std::memory_order_release);
        return true;
    }
    
    bool pop(T& item) {
        size_t current = read_idx.load(std::memory_order_relaxed);
        if(current == write_idx.load(std::memory_order_acquire))
            return false;
        item = buffer[current];
        read_idx.store((current + 1) % Size, std::memory_order_release);
        return true;
    }
};

#include <openssl/>

​​​​​​​SSL_CTX* init_ssl_ctx() {
    SSL_library_init();
    SSL_CTX* ctx = SSL_CTX_new(TLS_server_method());
    SSL_CTX_use_certificate_file(ctx, "", SSL_FILETYPE_PEM);
    SSL_CTX_use_PrivateKey_file(ctx, "", SSL_FILETYPE_PEM);
    return ctx;
}

void ssl_echo_server(SSL_CTX* ctx, int port) {
    int sock = create_server_socket(port);
    while(true) {
        int client = accept(sock, nullptr, nullptr);
        SSL* ssl = SSL_new(ctx);
        SSL_set_fd(ssl, client);
        SSL_accept(ssl);
        
        char buf[1024];
        int bytes = SSL_read(ssl, buf, sizeof(buf));
        SSL_write(ssl, buf, bytes);
        
        SSL_shutdown(ssl);
        SSL_free(ssl);
        close(client);
    }
}

template<typename T>
struct Validator {
    bool operator()(const T& data) {
        static_assert(has_validate_method<T>::value, 
                     "Type must implement validate()");
        return ();
    }
};

class SafeShmBuffer {
    void* mapping;
    size_t size;
    
public:
    SafeShmBuffer(const char* name, size_t size) 
        : mapping(mmap(..., PROT_READ | PROT_WRITE, ...)),
          size(size)
    {
        mprotect(mapping, size, PROT_READ); // Read only by default    }
    
    void enable_write() {
        mprotect(mapping, size, PROT_READ | PROT_WRITE);
    }
};

# Docker Compose Network Configuration Exampleservices:
  producer:
    image: ipc-producer
    networks:
      - ipc-net
      
  consumer:
    image: ipc-consumer
    networks:
      - ipc-net

​​​​​​​networks:
  ipc-net:
    driver: bridge
    attachable: true

// Envoy Filter Examplefunc onData(buffer []byte)  {
    if isSensitive(buffer) {
        ("Detected sensitive data")
        return 
    }
    return 
}

# Test command exampletaskset -c 0,1 ./ipc_bench \
    --protocol=shm \
    --threads=32 \
    --message-size=256 \
    --duration=60 \
    --warmup=10

# Generate core dumpulimit -c unlimited
./my_program
gdb ./my_program core.<pid>

# Commonly used GDB commands(gdb) bt full      # Full stack backtracking(gdb) info threads # Check thread status(gdb) p *mutex     # Check the status of mutexes

# Use bpftrace to monitor shmget callsbpftrace -e 'tracepoint:syscalls:sys_enter_shmget {
    @[comm] = count();
} interval:s:5 {
    print(@);
    clear(@);
}'