dragonflight/services/framecache/client/fc_pipe.c

/**
 * fc_pipe.c — Framecache slot → stdout pipe adapter.
 *
 * FRAME-COUPLED AUDIO (FC_VERSION 2):
 *   Each framecache ring entry carries the VIDEO frame AND that frame's
 *   SDI-embedded AUDIO together (written by the JOINED bridge from one slot).
 *   fc_pipe reads ONE entry per loop iteration.
 *
 *   TWO OUTPUT MODES:
 *
 *   1) AVI MODE (default when audio is wanted; selected with --avi or by giving
 *      an arg of "avi"):  fc_pipe writes a SINGLE streaming AVI container to
 *      stdout — video and audio INTERLEAVED in one byte stream. ffmpeg reads it
 *      as ONE input:
 *          ffmpeg -f avi -i pipe:0 -map 0:v ... -map 0:a ...
 *      This eliminates the two-live-pipe deadlock: when ffmpeg was given a raw
 *      video pipe AND a separate audio FIFO it stalled forever probing input 0.
 *      The AVI muxer writes its header once, then for each ring entry emits a
 *      '00dc' video chunk followed by a '01wb' audio chunk — frame-coupled by
 *      construction (both come from the same ring entry in the same iteration).
 *
 *   2) RAW MODE (legacy, video-only): if no audio FIFO / avi flag is given,
 *      fc_pipe writes raw UYVY422 video bytes to stdout as before.
 *
 *   The old split video-stdout / audio-FIFO design is REMOVED — it was the
 *   source of the ffmpeg deadlock.
 *
 * Usage:  fc_pipe <slot_id> [wait_ms] [mode]
 *   mode: "--avi" | "avi"  → single streaming AVI (video+audio) on stdout.
 *         omitted | "-"    → raw UYVY422 video-only on stdout.
 *
 * Terminates on: SIGTERM/SIGINT, stdout EPIPE (ffmpeg exited), slot gone.
 */

#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>
#include <math.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;
    }
    return 0;
}

/* ── Little-endian byte emitters into a caller buffer ────────────────────────── */
static inline void put_u16(uint8_t **pp, uint16_t v) {
    uint8_t *p = *pp; p[0] = (uint8_t)(v & 0xff); p[1] = (uint8_t)((v >> 8) & 0xff); *pp = p + 2;
}
static inline void put_u32(uint8_t **pp, uint32_t v) {
    uint8_t *p = *pp;
    p[0] = (uint8_t)(v & 0xff); p[1] = (uint8_t)((v >> 8) & 0xff);
    p[2] = (uint8_t)((v >> 16) & 0xff); p[3] = (uint8_t)((v >> 24) & 0xff);
    *pp = p + 4;
}
static inline void put_fourcc(uint8_t **pp, const char *cc) {
    uint8_t *p = *pp; p[0] = (uint8_t)cc[0]; p[1] = (uint8_t)cc[1];
    p[2] = (uint8_t)cc[2]; p[3] = (uint8_t)cc[3]; *pp = p + 4;
}

/* ── Streaming AVI header ─────────────────────────────────────────────────────
 * Builds RIFF('AVI ') + LIST('hdrl'){ avih + strl(vids) + strl(auds) } +
 * LIST('movi'). For a streaming AVI over a pipe we cannot seek back to patch
 * the RIFF and movi sizes, so we set them to 0x7FFFFFFF; ffmpeg's AVI demuxer
 * reads the strf headers and the 00dc/01wb chunk stream regardless. The hdrl
 * LIST size IS fixed/known, so it is written correctly. dwFlags is 0 — we do
 * NOT set AVIF_HASINDEX / AVIF_MUSTUSEINDEX (there is no index in a stream).
 *
 * Writes the header to *out and returns its length. Buffer must be >= 512. */
static size_t build_avi_header(uint8_t *out,
                               uint32_t width, uint32_t height,
                               uint32_t fps_num, uint32_t fps_den,
                               uint32_t video_bytes,
                               uint32_t audio_rate, uint32_t audio_channels,
                               uint32_t audio_sample_bytes) {
    const uint32_t STREAMING = 0x7FFFFFFFu;

    const uint16_t bits_per_sample = (uint16_t)(audio_sample_bytes * 8u);
    const uint16_t block_align     = (uint16_t)(audio_channels * audio_sample_bytes);
    const uint32_t avg_bytes_sec   = audio_rate * block_align;
    /* dwMicroSecPerFrame = 1e6 * fps_den / fps_num */
    const uint32_t usec_per_frame  =
        (uint32_t)((1000000.0 * (double)fps_den / (double)fps_num) + 0.5);

    /* Fixed sub-sizes (data bytes only, excluding the 8-byte ckID+ckSize). */
    const uint32_t AVIH_DATA = 56;   /* MainAVIHeader */
    const uint32_t STRH_DATA = 56;   /* AVISTREAMHEADER */
    const uint32_t BIH_DATA  = 40;   /* BITMAPINFOHEADER */
    const uint32_t WFX_DATA  = 18;   /* WAVEFORMATEX (cbSize=0) */

    /* LIST('strl') sizes = 4 (the 'strl' fourcc) + contained chunks. */
    const uint32_t vstrl_size = 4 + (8 + STRH_DATA) + (8 + BIH_DATA);   /* 4+64+48 = 116 */
    const uint32_t astrl_size = 4 + (8 + STRH_DATA) + (8 + WFX_DATA);   /* 4+64+26 = 94  */
    /* LIST('hdrl') size = 4 (the 'hdrl' fourcc) + avih chunk + both strl LISTs. */
    const uint32_t hdrl_size  = 4 + (8 + AVIH_DATA) + (8 + vstrl_size) + (8 + astrl_size);

    uint8_t *p = out;

    /* RIFF 'AVI ' (size unseekable → streaming sentinel) */
    put_fourcc(&p, "RIFF");
    put_u32(&p, STREAMING);
    put_fourcc(&p, "AVI ");

    /* LIST 'hdrl' */
    put_fourcc(&p, "LIST");
    put_u32(&p, hdrl_size);
    put_fourcc(&p, "hdrl");

    /* avih — MainAVIHeader (56 bytes) */
    put_fourcc(&p, "avih");
    put_u32(&p, AVIH_DATA);
    put_u32(&p, usec_per_frame);     /* dwMicroSecPerFrame */
    put_u32(&p, 0);                  /* dwMaxBytesPerSec */
    put_u32(&p, 0);                  /* dwPaddingGranularity */
    put_u32(&p, 0);                  /* dwFlags — NO index flags */
    put_u32(&p, 0);                  /* dwTotalFrames (unknown in stream) */
    put_u32(&p, 0);                  /* dwInitialFrames */
    put_u32(&p, 2);                  /* dwStreams (video + audio) */
    put_u32(&p, 0);                  /* dwSuggestedBufferSize */
    put_u32(&p, width);              /* dwWidth */
    put_u32(&p, height);             /* dwHeight */
    put_u32(&p, 0); put_u32(&p, 0); put_u32(&p, 0); put_u32(&p, 0); /* dwReserved[4] */

    /* LIST 'strl' — VIDEO */
    put_fourcc(&p, "LIST");
    put_u32(&p, vstrl_size);
    put_fourcc(&p, "strl");

    /* strh — AVISTREAMHEADER 'vids' (56 bytes) */
    put_fourcc(&p, "strh");
    put_u32(&p, STRH_DATA);
    put_fourcc(&p, "vids");          /* fccType */
    put_fourcc(&p, "UYVY");          /* fccHandler */
    put_u32(&p, 0);                  /* dwFlags */
    put_u16(&p, 0);                  /* wPriority */
    put_u16(&p, 0);                  /* wLanguage */
    put_u32(&p, 0);                  /* dwInitialFrames */
    put_u32(&p, fps_den);            /* dwScale = 1001 */
    put_u32(&p, fps_num);            /* dwRate  = 60000 */
    put_u32(&p, 0);                  /* dwStart */
    put_u32(&p, 0);                  /* dwLength (unknown) */
    put_u32(&p, video_bytes);        /* dwSuggestedBufferSize */
    put_u32(&p, 0xFFFFFFFFu);        /* dwQuality (-1 default) */
    put_u32(&p, video_bytes);        /* dwSampleSize (fixed for uncompressed) */
    put_u16(&p, 0); put_u16(&p, 0);  /* rcFrame.left, top */
    put_u16(&p, (uint16_t)width);    /* rcFrame.right */
    put_u16(&p, (uint16_t)height);   /* rcFrame.bottom */

    /* strf — BITMAPINFOHEADER (40 bytes) */
    put_fourcc(&p, "strf");
    put_u32(&p, BIH_DATA);
    put_u32(&p, 40);                 /* biSize */
    put_u32(&p, width);              /* biWidth */
    put_u32(&p, height);             /* biHeight */
    put_u16(&p, 1);                  /* biPlanes */
    put_u16(&p, 16);                 /* biBitCount (UYVY422 = 16bpp) */
    put_fourcc(&p, "UYVY");          /* biCompression fourcc */
    put_u32(&p, video_bytes);        /* biSizeImage = W*H*2 */
    put_u32(&p, 0);                  /* biXPelsPerMeter */
    put_u32(&p, 0);                  /* biYPelsPerMeter */
    put_u32(&p, 0);                  /* biClrUsed */
    put_u32(&p, 0);                  /* biClrImportant */

    /* LIST 'strl' — AUDIO */
    put_fourcc(&p, "LIST");
    put_u32(&p, astrl_size);
    put_fourcc(&p, "strl");

    /* strh — AVISTREAMHEADER 'auds' (56 bytes) */
    put_fourcc(&p, "strh");
    put_u32(&p, STRH_DATA);
    put_fourcc(&p, "auds");          /* fccType */
    put_u32(&p, 0);                  /* fccHandler (none for PCM) */
    put_u32(&p, 0);                  /* dwFlags */
    put_u16(&p, 0);                  /* wPriority */
    put_u16(&p, 0);                  /* wLanguage */
    put_u32(&p, 0);                  /* dwInitialFrames */
    put_u32(&p, block_align);        /* dwScale = nBlockAlign */
    put_u32(&p, avg_bytes_sec);      /* dwRate  = nAvgBytesPerSec */
    put_u32(&p, 0);                  /* dwStart */
    put_u32(&p, 0);                  /* dwLength (unknown) */
    put_u32(&p, avg_bytes_sec);      /* dwSuggestedBufferSize (~1s) */
    put_u32(&p, 0xFFFFFFFFu);        /* dwQuality */
    put_u32(&p, block_align);        /* dwSampleSize = nBlockAlign */
    put_u16(&p, 0); put_u16(&p, 0); put_u16(&p, 0); put_u16(&p, 0); /* rcFrame */

    /* strf — WAVEFORMATEX (18 bytes) */
    put_fourcc(&p, "strf");
    put_u32(&p, WFX_DATA);
    put_u16(&p, 1);                  /* wFormatTag = WAVE_FORMAT_PCM */
    put_u16(&p, (uint16_t)audio_channels);  /* nChannels */
    put_u32(&p, audio_rate);         /* nSamplesPerSec */
    put_u32(&p, avg_bytes_sec);      /* nAvgBytesPerSec */
    put_u16(&p, block_align);        /* nBlockAlign */
    put_u16(&p, bits_per_sample);    /* wBitsPerSample */
    put_u16(&p, 0);                  /* cbSize */

    /* LIST 'movi' — frames follow. Size unseekable → streaming sentinel. */
    put_fourcc(&p, "LIST");
    put_u32(&p, STREAMING);
    put_fourcc(&p, "movi");

    return (size_t)(p - out);
}

/* Write a single AVI chunk: 4-byte fourcc + u32 LE size + data (+ pad byte if
 * the size is odd, per the RIFF even-alignment rule). Returns 0 / -1. */
static int write_avi_chunk(int fd, const char *cc,
                           const uint8_t *data, uint32_t size) {
    uint8_t hdr[8];
    uint8_t *p = hdr;
    put_fourcc(&p, cc);
    put_u32(&p, size);
    if (write_all_fd(fd, hdr, 8) < 0) return -1;
    if (size && write_all_fd(fd, data, size) < 0) return -1;
    if (size & 1u) {
        uint8_t pad = 0;
        if (write_all_fd(fd, &pad, 1) < 0) return -1;
    }
    return 0;
}

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

    /* AVI mode is selected by an explicit flag in argv[3]. Anything that is not
     * "--avi"/"avi" (including "-" or omitted) → legacy raw video-only mode. */
    int avi_mode = 0;
    if (argc >= 4) {
        const char *m = argv[3];
        if (strcmp(m, "--avi") == 0 || strcmp(m, "avi") == 0) avi_mode = 1;
    }

    signal(SIGTERM, on_signal);
    signal(SIGINT,  on_signal);
    signal(SIGPIPE, SIG_IGN);

    fcntl(STDOUT_FILENO, F_SETFL,
          fcntl(STDOUT_FILENO, F_GETFL, 0) & ~O_NONBLOCK);

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

    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;
    }

    /* Pull stream format from the slot header for the AVI header. */
    fc_stream_info_t si;
    if (fc_consumer_info(c, &si) != 0 || si.width == 0 || si.height == 0) {
        fprintf(stderr, "[fc_pipe] failed to read slot stream info\n");
        fc_consumer_close(c);
        return 1;
    }
    if (si.fps_num == 0) { si.fps_num = 60000; si.fps_den = 1001; }
    if (si.fps_den == 0) si.fps_den = 1;
    if (si.audio_rate == 0)         si.audio_rate = 48000;
    if (si.audio_channels == 0)     si.audio_channels = 2;
    if (si.audio_sample_bytes == 0) si.audio_sample_bytes = 2;

    const uint32_t video_bytes = si.frame_size ? si.frame_size
                                               : si.width * si.height * 2u;
    const uint32_t a_blockalign = si.audio_channels * si.audio_sample_bytes;
    /* Samples per video frame for synthesized silence when a frame has no audio:
     * round(audio_rate * fps_den / fps_num). Bytes = samples * blockalign. */
    uint32_t silence_bytes = 0;
    {
        double spf = (double)si.audio_rate * (double)si.fps_den / (double)si.fps_num;
        uint32_t samples = (uint32_t)(spf + 0.5);
        silence_bytes = samples * a_blockalign;
    }
    uint8_t *silence = NULL;
    if (avi_mode && silence_bytes) {
        silence = (uint8_t *)calloc(1, silence_bytes);
        if (!silence) silence_bytes = 0;
    }

    if (avi_mode) {
        uint8_t hdr[512];
        size_t  hlen = build_avi_header(hdr, si.width, si.height,
                                        si.fps_num, si.fps_den, video_bytes,
                                        si.audio_rate, si.audio_channels,
                                        si.audio_sample_bytes);
        if (write_all_fd(STDOUT_FILENO, hdr, hlen) < 0) {
            fprintf(stderr, "[fc_pipe] stdout EPIPE writing AVI header\n");
            fc_consumer_close(c); free(silence);
            return 1;
        }
        fprintf(stderr,
                "[fc_pipe] slot open, streaming AVI(video+audio) → stdout "
                "(%ux%u %u/%u, %ub/frame, audio %uHz %uch s%ule, silence=%uB/frame)\n",
                si.width, si.height, si.fps_num, si.fps_den, video_bytes,
                si.audio_rate, si.audio_channels, si.audio_sample_bytes * 8u,
                silence_bytes);
    } else {
        fprintf(stderr, "[fc_pipe] slot open, streaming raw video → stdout (%ux%u)\n",
                si.width, si.height);
    }

    uint64_t frames_out    = 0;
    uint64_t total_dropped = 0;
    uint64_t audio_bytes   = 0;
    uint64_t audio_gaps    = 0;

    while (!g_stop) {
        fc_frame_ref_t ref;
        int rc = fc_consumer_read(c, &ref, 2000);
        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);
        }

        if (avi_mode) {
            /* Interleave THIS frame's video + audio in one stream. Both are
             * sourced from the SAME ring entry ⇒ frame-coupled by construction.
             * Video first (00dc), then audio (01wb). */
            if (write_avi_chunk(STDOUT_FILENO, "00dc", ref.data, ref.size) < 0) {
                if (!g_stop)
                    fprintf(stderr, "[fc_pipe] stdout EPIPE (video) — ffmpeg exited\n");
                break;
            }
            if (ref.audio_size > 0 && ref.audio) {
                if (write_avi_chunk(STDOUT_FILENO, "01wb", ref.audio, ref.audio_size) < 0) {
                    if (!g_stop)
                        fprintf(stderr, "[fc_pipe] stdout EPIPE (audio) — ffmpeg exited\n");
                    break;
                }
                audio_bytes += ref.audio_size;
            } else {
                /* No embedded audio this frame: emit one frame-interval of
                 * silence so the audio stream length tracks the video and
                 * ffmpeg never starves on the audio demuxer. */
                if (silence_bytes &&
                    write_avi_chunk(STDOUT_FILENO, "01wb", silence, silence_bytes) < 0) {
                    if (!g_stop)
                        fprintf(stderr, "[fc_pipe] stdout EPIPE (silence) — ffmpeg exited\n");
                    break;
                }
                audio_bytes += silence_bytes;
                audio_gaps++;
            }
        } else {
            /* Legacy raw video-only: write the UYVY422 bytes straight 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;
            }
        }

        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);
        }
    }

    free(silence);
    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;
}