dragonflight/services/framecache/client/fc_client.c

/**
 * fc_client.c — Consumer-side framecache client implementation.
 */
#include "fc_client.h"
#include "../src/slot.h"

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

#define SHM_DIR    "/dev/shm/framecache"
#define SEM_PREFIX "/framecache-"
#define SEM_SUFFIX "-write"

struct fc_consumer {
    int      shm_fd;
    void    *base;
    size_t   shm_size;
    sem_t   *sem;
    uint64_t read_cursor;   /* consumer's own position in the ring */
    uint64_t local_dropped; /* frames skipped by this consumer */
    char     slot_id[FC_MAX_SLOT_ID];
};

static uint64_t now_us(void)
{
    struct timespec ts;
    clock_gettime(CLOCK_REALTIME, &ts);
    return (uint64_t)ts.tv_sec * 1000000ULL + ts.tv_nsec / 1000ULL;
}

fc_consumer_t *fc_consumer_open(const char *slot_id, uint64_t wait_ms)
{
    char shm_path[128], sem_name[128];
    snprintf(shm_path, sizeof shm_path, "%s/%s", SHM_DIR, slot_id);
    snprintf(sem_name, sizeof sem_name, "%s%s%s", SEM_PREFIX, slot_id, SEM_SUFFIX);

    uint64_t deadline = now_us() + wait_ms * 1000ULL;
    int fd = -1;
    while (1) {
        fd = open(shm_path, O_RDONLY);
        if (fd >= 0) break;
        if (now_us() >= deadline) return NULL;
        struct timespec ts = { .tv_nsec = 100000000 }; /* 100ms */
        nanosleep(&ts, NULL);
    }

    /* Read header to get frame_size */
    fc_header_t hdr;
    if (pread(fd, &hdr, sizeof hdr, 0) != sizeof hdr || hdr.magic != FC_MAGIC) {
        close(fd); return NULL;
    }
    size_t total = fc_slot_shm_size(hdr.frame_size);

    void *base = mmap(NULL, total, PROT_READ, MAP_SHARED, fd, 0);
    if (base == MAP_FAILED) { close(fd); return NULL; }

    sem_t *sem = sem_open(sem_name, 0);
    if (sem == SEM_FAILED) { munmap(base, total); close(fd); return NULL; }

    fc_consumer_t *c = calloc(1, sizeof *c);
    if (!c) { sem_close(sem); munmap(base, total); close(fd); return NULL; }

    c->shm_fd  = fd;
    c->base    = base;
    c->shm_size = total;
    c->sem     = sem;
    /* Start reading from the current write position so we don't replay old frames */
    c->read_cursor  = atomic_load_explicit(
        &((fc_header_t *)base)->write_cursor, memory_order_acquire);
    c->local_dropped = 0;
    strncpy(c->slot_id, slot_id, FC_MAX_SLOT_ID - 1);
    return c;
}

int fc_consumer_read(fc_consumer_t *c, fc_frame_ref_t *ref, uint64_t timeout_ms)
{
    fc_header_t *hdr = (fc_header_t *)c->base;

    /* Wait for a new frame via semaphore */
    struct timespec abs_ts;
    clock_gettime(CLOCK_REALTIME, &abs_ts);
    abs_ts.tv_sec  += (time_t)(timeout_ms / 1000);
    abs_ts.tv_nsec += (long)((timeout_ms % 1000) * 1000000L);
    if (abs_ts.tv_nsec >= 1000000000L) {
        abs_ts.tv_sec++;
        abs_ts.tv_nsec -= 1000000000L;
    }

    while (sem_timedwait(c->sem, &abs_ts) != 0) {
        if (errno == ETIMEDOUT) return FC_TIMEOUT;
        if (errno == EINTR)     continue;
        return FC_ERROR;
    }

    uint64_t write_cur = atomic_load_explicit(&hdr->write_cursor,
                                               memory_order_acquire);
    int dropped = 0;

    /* Check if consumer fell behind by more than ring_depth */
    if (write_cur > c->read_cursor + hdr->ring_depth) {
        uint64_t skipped = write_cur - c->read_cursor - hdr->ring_depth;
        c->read_cursor   = write_cur - hdr->ring_depth;
        c->local_dropped += skipped;
        atomic_fetch_add(&hdr->dropped_frames, skipped);
        dropped = 1;
    }

    if (c->read_cursor >= write_cur) {
        /* Semaphore posted but nothing new yet (spurious) */
        return FC_TIMEOUT;
    }

    fc_frame_t *frame = fc_frame_at(c->base, hdr->frame_size, c->read_cursor);
    ref->data    = frame->data;
    ref->size    = frame->size;
    ref->pts_us  = frame->pts_us;
    ref->wall_us = frame->wall_us;
    ref->seq     = c->read_cursor;

    c->read_cursor++;
    return dropped ? FC_DROPPED : FC_OK;
}

void fc_consumer_close(fc_consumer_t *c)
{
    if (!c) return;
    sem_close(c->sem);
    munmap(c->base, c->shm_size);
    close(c->shm_fd);
    free(c);
}

uint64_t fc_consumer_write_cursor(fc_consumer_t *c)
{
    fc_header_t *hdr = (fc_header_t *)c->base;
    return atomic_load(&hdr->write_cursor);
}

uint64_t fc_consumer_dropped(fc_consumer_t *c)
{
    return c->local_dropped;
}