Designing a Lock-Free Ring Buffer for IPC with Shared Memory and POSIX Semaphores in C — code snippets

snippet-1.txt

#pragma once

#include <stdint.h>
#include <stdbool.h>
#include <stdatomic.h>
#include <semaphore.h>
#include <sys/types.h>

#define RING_CAPACITY 4096  // Must be a power of two
#define MSG_PAYLOAD_SIZE 1024
#define CACHE_LINE_SIZE 64

typedef struct {
    uint32_t length;
    uint8_t payload[MSG_PAYLOAD_SIZE];
} ipc_msg_t;

typedef struct {
    // Control metadata region
    volatile pid_t producer_pid;
    volatile pid_t consumer_pid;
    volatile bool is_active;
    
    sem_t sem_empty; // Signals space is available (producer waits on this)
    sem_t sem_fill;  // Signals data is available (consumer waits on this)
    
    // Pad control block to isolate it from the hot paths
    char pad0[CACHE_LINE_SIZE - ((sizeof(pid_t)*2 + sizeof(bool) + sizeof(sem_t)*2) % CACHE_LINE_SIZE)];

    // Producer Hot Path: modified by producer, read by consumer
    _Atomic uint64_t write_idx __attribute__((aligned(CACHE_LINE_SIZE)));
    _Atomic bool producer_sleeping;
    uint64_t cached_read_idx; 

    // Consumer Hot Path: modified by consumer, read by producer
    _Atomic uint64_t read_idx __attribute__((aligned(CACHE_LINE_SIZE)));
    _Atomic bool consumer_sleeping;
    uint64_t cached_write_idx; 

    // Data payload region starts on its own cache line
    ipc_msg_t buffer[RING_CAPACITY] __attribute__((aligned(CACHE_LINE_SIZE)));
} shm_ring_buffer_t;

snippet-2.txt

#include <stdio.h>
#include <stdlib.h>
#include <fcntl.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <unistd.h>
#include <string.h>
#include <errno.h>
#include <time.h>
#include "ipc_ring_buffer.h"

shm_ring_buffer_t* ipc_shm_init(const char* shm_name, bool* created_new) {
    int fd = shm_open(shm_name, O_CREAT | O_EXCL | O_RDWR, S_IRUSR | S_IWUSR);
    *created_new = false;

    if (fd == -1) {
        if (errno == EEXIST) {
            fd = shm_open(shm_name, O_RDWR, S_IRUSR | S_IWUSR);
            if (fd == -1) {
                perror("shm_open (existing) failed");
                return NULL;
            }
        } else {
            perror("shm_open failed");
            return NULL;
        }
    } else {
        *created_new = true;
    }

    if (*created_new) {
        if (ftruncate(fd, sizeof(shm_ring_buffer_t)) == -1) {
            perror("ftruncate failed");
            close(fd);
            return NULL;
        }
    }

    shm_ring_buffer_t* shm = mmap(NULL, sizeof(shm_ring_buffer_t),
                                  PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
    close(fd);

    if (shm == MAP_FAILED) {
        perror("mmap failed");
        return NULL;
    }

    if (*created_new) {
        memset((void*)shm, 0, sizeof(shm_ring_buffer_t));
        
        // pshared = 1 permits process-shared semaphore synchronization
        if (sem_init(&shm->sem_empty, 1, RING_CAPACITY) == -1 ||
            sem_init(&shm->sem_fill, 1, 0) == -1) {
            perror("sem_init failed");
            munmap(shm, sizeof(shm_ring_buffer_t));
            return NULL;
        }
        
        atomic_store_explicit(&shm->write_idx, 0, memory_order_relaxed);
        atomic_store_explicit(&shm->read_idx, 0, memory_order_relaxed);
        shm->producer_pid = getpid();
        
        // Finalize initialization by raising active flag
        shm->is_active = true;
    } else {
        // Wait for creator to finish initialization
        int timeout_ms = 1000;
        struct timespec sleep_time = { .tv_sec = 0, .tv_nsec = 1000000 }; // 1ms
        while (!shm->is_active && timeout_ms-- > 0) {
            nanosleep(&sleep_time, NULL);
        }
        if (!shm->is_active) {
            fprintf(stderr, "Timeout waiting for shared memory initialization\n");
            munmap(shm, sizeof(shm_ring_buffer_t));
            return NULL;
        }
    }

    return shm;
}

snippet-3.txt

#include <string.h>
#include <errno.h>
#include "ipc_ring_buffer.h"

bool ipc_ring_buffer_enqueue(shm_ring_buffer_t* shm, const ipc_msg_t* msg) {
    uint64_t write_idx = atomic_load_explicit(&shm->write_idx, memory_order_relaxed);
    
    // Check if buffer is full using cached read index
    if (write_idx - shm->cached_read_idx == RING_CAPACITY) {
        shm->cached_read_idx = atomic_load_explicit(&shm->read_idx, memory_order_acquire);
        
        if (write_idx - shm->cached_read_idx == RING_CAPACITY) {
            // Buffer is full. Set sleeping flag and double-check before blocking.
            atomic_store_explicit(&shm->producer_sleeping, true, memory_order_seq_cst);
            
            // Re-read read_idx after setting producer_sleeping to avoid lost-wakeup race
            shm->cached_read_idx = atomic_load_explicit(&shm->read_idx, memory_order_acquire);
            
            while (write_idx - shm->cached_read_idx == RING_CAPACITY) {
                if (sem_wait(&shm->sem_empty) == -1) {
                    if (errno == EINTR) {
                        continue; 
                    }
                    atomic_store_explicit(&shm->producer_sleeping, false, memory_order_relaxed);
                    return false;
                }
                shm->cached_read_idx = atomic_load_explicit(&shm->read_idx, memory_order_acquire);
            }
            atomic_store_explicit(&shm->producer_sleeping, false, memory_order_relaxed);
        }
    }

    // Write message into the ring slot (index wraps naturally due to power-of-two size)
    uint64_t slot = write_idx & (RING_CAPACITY - 1);
    memcpy(&shm->buffer[slot], msg, sizeof(ipc_msg_t));

    // Release store guarantees data write is visible before write_idx is updated
    atomic_store_explicit(&shm->write_idx, write_idx + 1, memory_order_release);

    // If consumer is asleep, wake it up
    if (atomic_load_explicit(&shm->consumer_sleeping, memory_order_seq_cst)) {
        sem_post(&shm->sem_fill);
    }

    return true;
}

snippet-4.txt

#include <string.h>
#include <errno.h>
#include "ipc_ring_buffer.h"

bool ipc_ring_buffer_dequeue(shm_ring_buffer_t* shm, ipc_msg_t* msg_out) {
    uint64_t read_idx = atomic_load_explicit(&shm->read_idx, memory_order_relaxed);

    // Check if buffer is empty using cached write index
    if (read_idx == shm->cached_write_idx) {
        shm->cached_write_idx = atomic_load_explicit(&shm->write_idx, memory_order_acquire);

        if (read_idx == shm->cached_write_idx) {
            // Buffer is empty. Set sleeping flag and double-check before blocking.
            atomic_store_explicit(&shm->consumer_sleeping, true, memory_order_seq_cst);

            // Re-read write_idx after setting consumer_sleeping to avoid lost-wakeup race
            shm->cached_write_idx = atomic_load_explicit(&shm->write_idx, memory_order_acquire);

            while (read_idx == shm->cached_write_idx) {
                if (sem_wait(&shm->sem_fill) == -1) {
                    if (errno == EINTR) {
                        continue;
                    }
                    atomic_store_explicit(&shm->consumer_sleeping, false, memory_order_relaxed);
                    return false;
                }
                shm->cached_write_idx = atomic_load_explicit(&shm->write_idx, memory_order_acquire);
            }
            atomic_store_explicit(&shm->consumer_sleeping, false, memory_order_relaxed);
        }
    }

    // Read message from the ring slot
    uint64_t slot = read_idx & (RING_CAPACITY - 1);
    memcpy(msg_out, &shm->buffer[slot], sizeof(ipc_msg_t));

    // Release store to let producer know this slot has been read
    atomic_store_explicit(&shm->read_idx, read_idx + 1, memory_order_release);

    // If producer is asleep, wake it up
    if (atomic_load_explicit(&shm->producer_sleeping, memory_order_seq_cst)) {
        sem_post(&shm->sem_empty);
    }

    return true;
}

snippet-5.txt

#include <stdio.h>
#include <signal.h>
#include <sys/types.h>
#include <unistd.h>
#include <errno.h>
#include "ipc_ring_buffer.h"

bool is_process_alive(pid_t pid) {
    if (pid <= 0) return false;
    if (kill(pid, 0) == 0) return true;
    return errno != ESRCH;
}

shm_ring_buffer_t* ipc_shm_robust_init(const char* shm_name, bool is_producer) {
    bool created_new = false;
    shm_ring_buffer_t* shm = ipc_shm_init(shm_name, &created_new);

    if (shm && !created_new) {
        // Detached check: Verify if the peer owner is still alive
        pid_t peer_pid = is_producer ? shm->consumer_pid : shm->producer_pid;
        if (peer_pid > 0 && !is_process_alive(peer_pid)) {
            fprintf(stderr, "[SHM] Peer process %d is dead. Re-initializing segment.\n", peer_pid);
            munmap(shm, sizeof(shm_ring_buffer_t));
            shm_unlink(shm_name);
            shm = ipc_shm_init(shm_name, &created_new);
        }
    }

    if (shm) {
        if (is_producer) {
            shm->producer_pid = getpid();
        } else {
            shm->consumer_pid = getpid();
        }
    }
    return shm;
}

void ipc_shm_cleanup(const char* shm_name, shm_ring_buffer_t* shm) {
    if (shm) {
        shm->is_active = false;
        // Break any blocked peer threads out of sem_wait loops
        sem_post(&shm->sem_empty);
        sem_post(&shm->sem_fill);
        munmap(shm, sizeof(shm_ring_buffer_t));
    }
    shm_unlink(shm_name);
}

snippet-6.txt

#define _GNU_SOURCE
#include <pthread.h>
#include <stdio.h>

bool pin_thread_to_core(int core_id) {
    cpu_set_t cpuset;
    CPU_ZERO(&cpuset);
    CPU_SET(core_id, &cpuset);
    
    pthread_t current_thread = pthread_self();
    if (pthread_setaffinity_np(current_thread, sizeof(cpu_set_t), &cpuset) != 0) {
        perror("pthread_setaffinity_np failed");
        return false;
    }
    return true;
}