dragonflight/services/framecache/client/fc_pipe.c

/**
 * fc_pipe.c — Framecache slot → stdout(video) + FIFO(audio) pipe adapter.
 *
 * FRAME-COUPLED AUDIO (FC_VERSION 2):
 *   Each framecache ring entry carries the VIDEO frame AND that frame's
 *   SDI-embedded AUDIO together. fc_pipe reads ONE entry per loop iteration
 *   and writes:
 *       - the video bytes  → stdout      (ffmpeg rawvideo input 0 / pipe:0)
 *       - the audio bytes  → audio FIFO  (ffmpeg s16le input 1)
 *   IN LOCKSTEP from the SAME cursor read. Because both come out of the same
 *   ring entry in the same iteration, audio can never drift ahead of (or
 *   behind) its video frame — there is no second independent buffer/transport
 *   to race. This eliminates the constant "audio ahead of video" offset at the
 *   root.
 *
 * capture-manager.js spawns this process, pipes its stdout to ffmpeg input 0,
 * and passes the audio FIFO path it created as argv[3]; ffmpeg reads that FIFO
 * as input 1.
 *
 * Each consumer instance has its own independent read cursor, so multiple
 * fc_pipe processes reading from the same slot never interfere with each other
 * (growing + proxy + HLS all read the same SDI signal simultaneously).
 *
 * Usage:
 *   fc_pipe <slot_id> [wait_ms] [audio_fifo_path]
 *
 *   audio_fifo_path optional: if omitted (or "-"), audio is NOT emitted and
 *   fc_pipe behaves video-only (legacy / network video-only sources).
 *
 * Audio FIFO lifecycle:
 *   - Opened O_WRONLY|O_NONBLOCK with retry; until a reader (ffmpeg input 1)
 *     attaches the open returns ENXIO. We keep delivering VIDEO meanwhile so
 *     ffmpeg makes progress and opens the FIFO. Audio for those very first
 *     pre-roll frames is dropped (sub-frame startup gap inside pre-roll).
 *   - Once open we switch the fd to blocking and write each frame's audio with
 *     the same write_all() as video, keeping them coupled. If the signal has
 *     no embedded audio on a frame (audio_size 0) we synthesize exactly that
 *     frame's worth of silence so ffmpeg input 1 never starves and the audio
 *     timeline length always equals the video timeline length (no drift).
 *   - On audio FIFO EPIPE (ffmpeg input 1 reader died, e.g. session restart),
 *     we close and re-open the FIFO; VIDEO delivery is unaffected.
 *
 * Terminates on:
 *   - SIGTERM / SIGINT (clean stop from capture-manager)
 *   - stdout EPIPE (ffmpeg exited)
 *   - Slot disappears (bridge stopped)
 *
 * Exit codes:
 *   0  clean stop (SIGTERM)
 *   1  slot not found within wait_ms
 *   2  stdout write error (EPIPE)
 */

#include "../src/slot.h"
#include "fc_client.h"

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <signal.h>
#include <stdint.h>
#include <unistd.h>
#include <errno.h>
#include <time.h>
#include <fcntl.h>

static volatile int g_stop = 0;
static void on_signal(int s) { (void)s; g_stop = 1; }

/* Write all bytes to fd (blocking). Returns 0 on success, -1 on EPIPE/error. */
static int write_all_fd(int fd, const void *buf, size_t len) {
    const uint8_t *p = (const uint8_t *)buf;
    size_t off = 0;
    while (off < len) {
        ssize_t n = write(fd, p + off, len - off);
        if (n > 0) { off += (size_t)n; continue; }
        if (n < 0 && errno == EINTR) continue;
        return -1; /* EPIPE or other fatal error */
    }
    return 0;
}

int main(int argc, char *argv[]) {
    if (argc < 2) {
        fprintf(stderr, "Usage: %s <slot_id> [wait_ms] [audio_fifo_path]\n", argv[0]);
        return 1;
    }
    const char *slot_id  = argv[1];
    uint64_t    wait_ms  = argc >= 3 ? (uint64_t)atoll(argv[2]) : 30000;
    const char *audio_fifo = (argc >= 4 && strcmp(argv[3], "-") != 0) ? argv[3] : NULL;

    signal(SIGTERM, on_signal);
    signal(SIGINT,  on_signal);
    signal(SIGPIPE, SIG_IGN);   /* detect EPIPE via write() return value */

    /* Set stdout to binary/blocking mode — no newline translation */
    fcntl(STDOUT_FILENO, F_SETFL,
          fcntl(STDOUT_FILENO, F_GETFL, 0) & ~O_NONBLOCK);

    fprintf(stderr, "[fc_pipe] opening slot '%s' (wait %llums) audio_fifo=%s\n",
            slot_id, (unsigned long long)wait_ms,
            audio_fifo ? audio_fifo : "(none)");

    fc_consumer_t *c = fc_consumer_open(slot_id, wait_ms);
    if (!c) {
        fprintf(stderr, "[fc_pipe] slot '%s' not found within %llums\n",
                slot_id, (unsigned long long)wait_ms);
        return 1;
    }

    fprintf(stderr, "[fc_pipe] slot open, streaming video→stdout%s\n",
            audio_fifo ? " + audio→FIFO (frame-coupled)" : "");

    int    afd        = -1;   /* audio FIFO fd, -1 until a reader attaches */
    int    logged_aud = 0;

    uint64_t frames_out    = 0;
    uint64_t total_dropped = 0;
    uint64_t audio_bytes   = 0;
    uint64_t audio_gaps    = 0;   /* frames where embedded audio was absent (silence filled) */

    while (!g_stop) {
        fc_frame_ref_t ref;
        int rc = fc_consumer_read(c, &ref, 2000 /* 2s timeout */);

        if (rc == FC_TIMEOUT) continue;
        if (rc == FC_ERROR)   break;

        if (rc == FC_LAPPED) {
            total_dropped = fc_consumer_dropped(c);
            fprintf(stderr, "[fc_pipe] WARNING: frame lapped mid-read — total dropped: %llu\n",
                    (unsigned long long)total_dropped);
            continue;
        }

        if (rc == FC_DROPPED) {
            total_dropped = fc_consumer_dropped(c);
            fprintf(stderr, "[fc_pipe] WARNING: consumer fell behind — total dropped: %llu\n",
                    (unsigned long long)total_dropped);
        }

        /* ── Try to (re)attach the audio FIFO without stalling video ──────────
         * O_WRONLY|O_NONBLOCK returns ENXIO until ffmpeg input 1 opens the read
         * end. We keep delivering video so ffmpeg progresses and opens it. */
        if (audio_fifo && afd < 0) {
            int fd = open(audio_fifo, O_WRONLY | O_NONBLOCK);
            if (fd >= 0) {
                /* Switch to blocking for coupled writes. */
                int fl = fcntl(fd, F_GETFL, 0);
                if (fl >= 0) fcntl(fd, F_SETFL, fl & ~O_NONBLOCK);
                afd = fd;
                if (!logged_aud) {
                    fprintf(stderr, "[fc_pipe] audio FIFO reader attached — coupled audio live\n");
                    logged_aud = 1;
                }
            }
            /* else: not ready yet (ENXIO) — deliver video, retry next frame. */
        }

        /* ── Write VIDEO to stdout ──────────────────────────────────────────── */
        if (write_all_fd(STDOUT_FILENO, ref.data, ref.size) < 0) {
            if (!g_stop)
                fprintf(stderr, "[fc_pipe] stdout EPIPE — ffmpeg exited\n");
            break;
        }

        /* ── Write THIS frame's AUDIO to the FIFO, in lockstep ────────────────
         * Same ring entry, same iteration ⇒ frame-coupled. The bridge writer
         * guarantees every entry carries one frame-interval of audio (real
         * embedded PCM, or silence when the signal has none), so ref.audio_size
         * is non-zero in steady state and the audio timeline length always
         * tracks the video timeline length (no drift). A 0-size entry (only at
         * the very first frame, or a video-only net source) contributes nothing
         * and is harmless because ffmpeg derives audio PTS from sample count. */
        if (afd >= 0) {
            if (ref.audio_size > 0 && ref.audio) {
                if (write_all_fd(afd, ref.audio, ref.audio_size) < 0) {
                    fprintf(stderr, "[fc_pipe] audio FIFO EPIPE — will reattach\n");
                    close(afd); afd = -1;
                } else {
                    audio_bytes += ref.audio_size;
                }
            } else {
                audio_gaps++;   /* diagnostics: frame without embedded audio */
            }
        }

        frames_out++;
        if (frames_out % 300 == 0) {
            fprintf(stderr, "[fc_pipe] frames=%llu dropped=%llu audio_bytes=%llu gaps=%llu\n",
                    (unsigned long long)frames_out,
                    (unsigned long long)total_dropped,
                    (unsigned long long)audio_bytes,
                    (unsigned long long)audio_gaps);
        }
    }

    if (afd >= 0) close(afd);
    fc_consumer_close(c);
    fprintf(stderr, "[fc_pipe] done frames=%llu dropped=%llu audio_bytes=%llu gaps=%llu\n",
            (unsigned long long)frames_out,
            (unsigned long long)total_dropped,
            (unsigned long long)audio_bytes,
            (unsigned long long)audio_gaps);
    return 0;
}