/* ============================================================================== UDP -> HLS -- FINAL COMPLETE BUILD (in-process transcoder, v25) ============================================================================== SIMPLE-COPY mode (default, no deps beyond openssl): gcc -O2 -pthread -o udp_hls udp_hls_final.c transcode.c \ $(pkg-config --cflags --libs libavcodec libavformat libavutil libswresample) \ -lssl -lcrypto TRANSCODER mode (audio_aac=1 or video_h264=1): No ffmpeg subprocess. Demux/decode/encode/mux all happen in-process via libavformat/libavcodec/libswresample (transcode.c), linked directly into this binary. Video encodes through libx264; audio encodes through libfdk_aac when available (falls back to the native "aac" encoder automatically if libfdk_aac isn't linked into your libavcodec build). The transcoder hands back finished 188-byte MPEG-TS packets one at a time, which feed into the SAME pkt_process()/seg_open()/seg_close()/write_m3u8() pipeline that simple-copy mode uses -- segmentation, AES, and playlist writing are unified across both modes instead of duplicated. VIDEO ENCODE mode (video_h264=1): x264 GOP size is sized to fps * SEGMENT_SECS with sc_threshold=0, matching the old -g 50 -keyint_min 25 -sc_threshold 0 recipe, so every segment boundary still lands on an IDR frame. See open_x264_encoder() in transcode.c. AUDIO-ONLY / VIDEO-COPY mode (audio_aac=1 only): Video stream is passed through unmodified (decode/encode skipped entirely for video); segment cuts rely on the source's own IDR placement exactly as before. channels.conf: mcast port dir name iface keyfile keyuri iv_hex start_seq [opts] Options (transcoder mode -- in-process, no subprocess): audio_aac=1 convert audio to AAC audio_map=0 specific audio stream (0-based); -1 (default) = ALL audio streams video_h264=1 transcode video to H264 (libx264) aac_mode=vbr3 VBR quality vbr1..vbr5 or cbr (libfdk_aac only; native aac uses a fixed bitrate) audio_bitrate=128k AAC CBR bitrate video_preset=ultrafast x264 preset video_bitrate=2000k x264 CBR (ignored if video_crf set) video_crf=23 x264 CRF 0-51 (overrides video_bitrate) video_tune=zerolatency x264 tune (~15%% CPU saving) deinterlace=1 yadif-equivalent deinterlace filter (576i->progressive) video_threads=2 limit x264 threads Options (zero-CPU, no transcoder, simple-copy mode): fix_mp2=1 patch PMT audio type 0x03->0x0F (LG MP2 fix) fix_interlace=1 patch H264 SPS progressive flag (LG 576i fix) pts_reset=1 reset PTS/DTS to near-zero (LG high-PTS fix) ENV: HLS_DELETE_THRESHOLD=N extra segments on disk [default 5] AAC_DEBUG=1 show transcoder lifecycle/error diagnostics on stderr ============================================================================== */ #define _GNU_SOURCE #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include "transcode.h" /* -- tunables --------------------------------------------------- */ #define SEGMENT_SECS 2 #define MAX_SEGMENTS 15 #define HLS_DELETE_THRESHOLD 5 #define MAX_CHANNELS 512 #define MAX_NICS 16 #define WRITE_BATCH 128 #define RCVBUF (32<<20) #define DUR_SLOTS 256 #define PIPE_BUF_SZ (4<<20) /* 4MB pipe: ~22000 TS pkts, ~3s HD buffer */ #define MAX_PKTS_PER_CHAN 7 /* max datagrams drained per channel per poll */ #define READ_CHUNK 65536 /* -- constants -------------------------------------------------- */ #define TS_SZ 188 #define AES_BLK 16 #define UDP_MAX 65536 #define PCR_HZ 90000ULL #define PCR_WRAP 0x200000000ULL static int g_del = HLS_DELETE_THRESHOLD; static int g_aac_debug = 0; /* -- wall-clock timestamp helper ----------------------------- */ static void ts_now(char *buf){ /* buf must be =9 bytes */ time_t t = time(NULL); struct tm *tm = localtime(&t); snprintf(buf,9,"%02d:%02d:%02d",tm->tm_hour,tm->tm_min,tm->tm_sec); } /* -------------------------------------------------------------- LOCK-FREE EVENT RING — zero I/O in the hot path -------------------------------------------------------------- The recv/process thread calls event_push() which does: - one bounds check - one struct copy (=32 bytes) - one __atomic_store on the write index No printf, no mutex, no syscall. The stats thread calls event_drain() which reads and prints every pending event. Single-producer (recv thread) / single-consumer (stats thread) per channel — no ABA risk. Ring size 256 — at worst one event per segment (every 2s) so 256 slots = 512s of headroom before wrap. */ #define EVT_RING_SZ 256 /* must be power of 2 */ #define EVT_RING_MASK (EVT_RING_SZ-1) typedef enum { EVT_NONE = 0, EVT_PAT, /* PAT acquired */ EVT_PMT, /* PMT acquired / updated */ EVT_CLEAN_START, /* first segment started */ EVT_SEG_OPEN, /* segment file opened */ EVT_SEG_CLOSE, /* segment file closed */ EVT_CC_DROP, /* CC discontinuity detected */ EVT_RECOVERED, /* recovery complete */ EVT_STALE, /* no input for >2×seg_secs */ EVT_OPEN_GOP, /* stream detected open-GOP */ EVT_WRITE_ERR, /* disk write failed */ EVT_FFMPEG_EOF, /* ffmpeg process exited */ } EvtType; typedef struct { EvtType type; time_t when; /* time(NULL) at event */ uint64_t seq; /* seg_seq at event */ uint32_t u32a; /* type-specific (see below) */ uint32_t u32b; uint16_t u16a; uint8_t u8a, u8b; /* EVT_SEG_CLOSE: u32a=elapsed_ms seq=seg closed u32b=kbps */ /* EVT_SEG_OPEN: seq=seg opened */ /* EVT_CC_DROP: u16a=pid u8a=exp_cc u8b=got_cc u32a=gap u32b=total */ /* EVT_CLEAN_START/RECOVERED: u16a=vid_pid u32a=total_lost */ /* EVT_PAT: u16a=pmt_pid */ /* EVT_PMT: u16a=vid_pid u32a=aud_pid u32b=pcr_pid */ /* EVT_STALE: u32a=idle_ms */ } Evt; typedef struct { Evt ring[EVT_RING_SZ]; unsigned wr; /* written by recv thread, atomic store/load */ unsigned rd; /* read by stats thread only */ } EvtRing; /* push one event — called from hot path, zero blocking */ static inline void event_push(EvtRing *r, const Evt *e){ unsigned w = __atomic_load_n(&r->wr, __ATOMIC_RELAXED); r->ring[w & EVT_RING_MASK] = *e; __atomic_store_n(&r->wr, w+1, __ATOMIC_RELEASE); } /* drain and print all pending events — called from stats thread */ static void event_drain(EvtRing *r, const char *chname){ unsigned w = __atomic_load_n(&r->wr, __ATOMIC_ACQUIRE); while(r->rd != w){ Evt *e = &r->ring[r->rd & EVT_RING_MASK]; r->rd++; /* format wall time */ struct tm *tm = localtime(&e->when); char ts[9]; snprintf(ts,9,"%02d:%02d:%02d",tm->tm_hour,tm->tm_min,tm->tm_sec); switch(e->type){ case EVT_PAT: printf(" [%s][%s] PAT — pmt_pid=0x%04X\n", ts, chname, e->u16a); break; case EVT_PMT: printf(" [%s][%s] PMT — vid=0x%04X aud=0x%04X pcr=0x%04X\n", ts, chname, e->u16a, e->u32a, e->u32b); break; case EVT_CLEAN_START: printf(" [%s][%s] CLEAN START — vid=0x%04X\n", ts, chname, e->u16a); break; case EVT_SEG_OPEN: printf(" [%s][%s] opened index%llu.ts\n", ts, chname, (unsigned long long)e->seq); break; case EVT_SEG_CLOSE: printf(" [%s][%s] closed index%llu.ts %u.%03us %u kbps\n", ts, chname, (unsigned long long)e->seq, e->u32a/1000, e->u32a%1000, e->u32b); break; case EVT_CC_DROP: printf(" [%s][%s] !! CC DROP pid=0x%04X" " exp=0x%X got=0x%X ~%u lost (total=%u)\n", ts, chname, e->u16a, e->u8a, e->u8b, e->u32a, e->u32b); break; case EVT_RECOVERED: printf(" [%s][%s] RECOVERED — new seg started" " (session_total_lost=%u)\n", ts, chname, e->u32a); break; case EVT_STALE: printf(" [%s][%s] !! STALE — no input for %u.%03us\n", ts, chname, e->u32a/1000, e->u32a%1000); break; case EVT_OPEN_GOP: printf(" [%s][%s] open-GOP stream detected\n", ts, chname); break; case EVT_WRITE_ERR: printf(" [%s][%s] !! WRITE ERROR on segment disk\n", ts, chname); break; case EVT_FFMPEG_EOF: printf(" [%s][%s] !! FFMPEG EOF/EXIT\n", ts, chname); break; default: break; } } } /* -------------------------------------------------------------- AES-128-CBC -------------------------------------------------------------- */ typedef struct { AES_KEY enc; uint8_t iv0[AES_BLK], iv[AES_BLK]; uint8_t buf[WRITE_BATCH*TS_SZ + AES_BLK*2]; int n; } Aes; static void aes_init(Aes *a, const uint8_t *k, const uint8_t *iv){ AES_set_encrypt_key(k, 128, &a->enc); memcpy(a->iv0, iv, AES_BLK); memcpy(a->iv, iv, AES_BLK); a->n = 0; } static void aes_reset(Aes *a){ memcpy(a->iv, a->iv0, AES_BLK); a->n = 0; } static void aes_flush(Aes *a, int fd, int fin){ int n = a->n; if(fin){ int p = AES_BLK-(n%AES_BLK); if(!p) p = AES_BLK; memset(a->buf+n, p, p); n += p; } else n = (n/AES_BLK)*AES_BLK; if(n <= 0) return; uint8_t out[sizeof(a->buf)]; AES_cbc_encrypt(a->buf, out, n, &a->enc, a->iv, AES_ENCRYPT); ssize_t r = write(fd, out, n); (void)r; int rem = fin ? 0 : a->n - n; if(rem > 0) memmove(a->buf, a->buf+n, rem); a->n = rem; } static void aes_push(Aes *a, int fd, const uint8_t *p){ memcpy(a->buf+a->n, p, TS_SZ); a->n += TS_SZ; if(a->n >= WRITE_BATCH*TS_SZ) aes_flush(a, fd, 0); } /* for writer thread: push arbitrary bytes (not aligned to TS_SZ) */ static void aes_push_bytes(Aes *a, int fd, const uint8_t *data, int len){ int off = 0; while(off < len){ int sp = (int)sizeof(a->buf) - a->n - AES_BLK*2; if(sp <= 0){ aes_flush(a, fd, 0); continue; } int cp = len - off; if(cp > sp) cp = sp; memcpy(a->buf+a->n, data+off, cp); a->n += cp; off += cp; if(a->n >= WRITE_BATCH*TS_SZ) aes_flush(a, fd, 0); } } /* -------------------------------------------------------------- CHANNEL OPTIONS (ffmpeg pipe) -------------------------------------------------------------- */ typedef struct { int audio_aac; int audio_map; /* -1 = all streams */ int video_h264; char audio_bitrate[32]; char aac_mode[16]; char video_bitrate[32]; char video_preset[32]; int video_crf; /* 0 = not set; 1-51 = CRF value; overrides video_bitrate */ char video_tune[32]; /* x264 -tune: zerolatency, film, animation, etc. */ int deinterlace; /* 1 = add -vf yadif=mode=0 (fast field deinterlace) */ int video_threads; /* 0 = auto; N = limit x264 threads (reduces CPU spikes) */ /* zero-CPU stream fix options (simple-copy mode) */ int fix_mp2; /* patch PMT audio stream_type 0x03->0x0F, recalc CRC */ int fix_interlace; /* patch H264 SPS frame_mbs_only_flag 0->1 */ int pts_reset; /* subtract first PTS from all PES timestamps */ } ChOpts; /* -------------------------------------------------------------- IN-PROCESS TRANSCODER STATE (replaces FfPipe / ffmpeg subprocess) -------------------------------------------------------------- */ /* (Transcoder is an opaque type from transcode.h; Ch just holds a * pointer plus a drop counter mirroring the old pipe_drops stat.) */ /* -------------------------------------------------------------- CHANNEL STATE -------------------------------------------------------------- */ typedef struct { /* config */ char mcast[64], dir[256], name[64], iface[64]; char keyfile[256], keyuri[512], iv_hex[33]; int port; uint64_t start_seq; ChOpts opts; /* socket (owned by NIC recv thread) */ int fd; int rtp_ok, rtp_off; /* PSI */ uint8_t pat[TS_SZ], pmt[TS_SZ]; int pat_ok, pmt_ok; uint16_t pmt_pid, vid_pid, aud_pid, pcr_pid; /* in-process transcoder (replaces ffmpeg subprocess) */ int use_ffmpeg; /* name kept for minimal diff; means "use transcoder" now */ Transcoder *tc; uint64_t tc_drops; /* ring-buffer overflow drops, mirrors old pipe_drops stat */ /* segment */ int seg_fd; uint64_t seg_seq; double dur[DUR_SLOTS]; uint64_t seg_size[DUR_SLOTS]; /* AES */ int aes_on; Aes aes; /* write batch (plain copy mode) */ uint8_t wb[WRITE_BATCH*TS_SZ]; int wb_n; /* Wall-clock segment timing */ struct timespec seg_start; int seg_timing_ok; /* startup gate */ int started; /* CC drop detection (on the active output stream -- raw input in * simple-copy mode, or the transcoder's re-muxed output in * transcoder mode; drives segment recovery via pkt_process) */ uint8_t cc_last[8192]; int cc_init; uint32_t cc_errors; /* recovery after CC drop */ int recovering; /* raw-input-side CC/source-health stats (transcoder mode only). * Tracked independently of cc_last/cc_errors above so transcoded * mode doesn't conflate "is the multicast source healthy" with * "is our re-muxed output healthy" -- they're different signals * once a transcoder sits in between. Simple-copy mode doesn't use * these; cc_errors above already covers it directly. */ uint8_t in_cc_last[8192]; int in_cc_init; uint32_t in_cc_errors; /* PTS/DTS reset state (pts_reset=1 option) */ int pts_base_ok; /* 1 once we have captured the first PTS */ int64_t pts_base; /* first PTS value seen — subtracted from all PTS/DTS */ /* IDR tracking (simple-copy only) */ int idr_found; int idr_slice; int idr_scan; int idr_off; int idr_misses; struct timespec idr_wait_start; int idr_waiting; /* stats */ uint64_t dgrams, pkts, segs; struct timespec last_pkt; /* per-interval snapshot for bitrate/pkt-rate in stats */ uint64_t snap_pkts; /* pkts at last stats snapshot */ uint64_t snap_bytes; /* TS bytes received since last snapshot */ uint64_t snap_cc_errors; /* cc_errors at last snapshot */ struct timespec snap_time; /* wall time of last snapshot */ /* lock-free event ring — recv thread writes, stats thread reads */ EvtRing evtring; } Ch; static volatile sig_atomic_t g_stop = 0; static Ch g_ch[MAX_CHANNELS]; static int g_nch = 0; static void onsig(int s){ (void)s; g_stop = 1; } /* -------------------------------------------------------------- TS HELPERS -------------------------------------------------------------- */ static inline uint16_t ts_pid(const uint8_t *p) { return (uint16_t)(((p[1]&0x1F)<<8)|p[2]); } static inline int ts_pusi(const uint8_t *p){ return (p[1]>>6)&1; } static inline int ts_afc (const uint8_t *p){ return (p[3]>>4)&3; } static inline int ts_poff(const uint8_t *p){ int a = ts_afc(p), o; if(a==1) o = 4; else if(a==3) o = 5+(int)p[4]; else return -1; return o < TS_SZ ? o : -1; } static inline double wall_el(const struct timespec *a, const struct timespec *b){ return (double)(b->tv_sec - a->tv_sec) + (double)(b->tv_nsec - a->tv_nsec)/1e9; } static uint16_t pat_parse(const uint8_t *p){ int o = ts_poff(p); if(o<0||o>=TS_SZ) return 0; int b = o+1+p[o]; if(b+8>=TS_SZ||p[b]!=0) return 0; int e = b+3+(((p[b+1]&0xF)<<8)|p[b+2])-4; for(int q=b+8; q+3<=e&&q+3=TS_SZ) return 0; int b = o+1+p[o]; if(b+12>=TS_SZ||p[b]!=0x02) return 0; *pcr = (uint16_t)(((p[b+8]&0x1F)<<8)|p[b+9]); int pil = ((p[b+10]&0xF)<<8)|p[b+11]; int sec = ((p[b+1]&0xF)<<8)|p[b+2]; int es = b+12+pil, ee = b+3+sec-4; *vid = *aud = 0; while(es+4 0; } /* scan H264 NAL units — returns 1=IDR, 0=non-IDR slice, -1=no slice */ static int scan_nal(const uint8_t *es, int len){ for(int i=0; i+2=len) continue; uint8_t t = es[n]&0x1F; if(t==5) return 1; if(t==1||t==2||t==3||t==4) return 0; if(t==20||t==21) return 1; i = n; } return -1; } static int rtp_detect(const uint8_t *b, int n){ if(n<13||(b[0]>>6)!=2) return 0; int h = 12+(b[0]&0xF)*4; if((b[0]>>4)&1){ if(n<=h+4) return 0; h += 4+((b[h+2]<<8)|b[h+3])*4; } return (hpmt[] cached copy; called once after PMT seen. Returns 1 if a patch was made, 0 if nothing changed. */ static int pmt_patch_mp2(uint8_t *pkt, uint16_t aud_pid){ int o = ts_poff(pkt); if(o<0||o+1>=TS_SZ) return 0; int b = o+1+pkt[o]; if(b+12>=TS_SZ||pkt[b]!=0x02) return 0; int sec_len = ((pkt[b+1]&0x0F)<<8)|pkt[b+2]; /* section_length */ int pil = ((pkt[b+10]&0x0F)<<8)|pkt[b+11]; /* program_info_length */ int es = b+12+pil; int ee = b+3+sec_len-4; /* -4 = exclude CRC32 */ int patched = 0; while(es+4 < TS_SZ && es+4 <= ee){ uint8_t st = pkt[es]; uint16_t sp = (uint16_t)(((pkt[es+1]&0x1F)<<8)|pkt[es+2]); int el = ((pkt[es+3]&0x0F)<<8)|pkt[es+4]; if(sp == aud_pid && st == 0x03){ pkt[es] = 0x0F; /* MP2 -> ADTS AAC stream type */ patched = 1; } es += 5+el; } if(!patched) return 0; /* recalculate CRC32 over the entire section (from table_id to end-4) */ int crc_off = b+3+sec_len-4; /* byte offset of CRC32 in pkt[] */ if(crc_off+4 > TS_SZ) return 1; /* safety: CRC outside packet, skip */ uint32_t crc = crc32_mpeg(pkt+b, sec_len-1); /* section minus CRC field */ pkt[crc_off+0] = (crc>>24)&0xFF; pkt[crc_off+1] = (crc>>16)&0xFF; pkt[crc_off+2] = (crc>> 8)&0xFF; pkt[crc_off+3] = (crc )&0xFF; return 1; } /* -- H264 SPS frame_mbs_only_flag patch ------------------------ Scans a TS packet payload for a raw H264 SPS (NAL type 7) and sets frame_mbs_only_flag=1 (progressive). H264 SPS bit layout (after the 3-byte NAL header): profile_idc 8 bits constraint_flags 8 bits level_idc 8 bits seq_parameter_set_id exp-golomb ... (profile-dependent fields) ... frame_mbs_only_flag 1 bit — position depends on profile Rather than parsing the full Exp-Golomb SPS we use a reliable heuristic: find the SPS start code, then locate the first byte whose bit 0 is 0 in a position consistent with the interlace flag location for the common CAVLC/CABAC profiles (Baseline, Main, High). For SD broadcast (Main profile, level 3.0) the frame_mbs_only_flag sits reliably at a known byte offset. Safe approach: only patch if we can verify profile_idc == 77 (Main) or 66 (Baseline) and the flag byte is 0 — avoids corrupting streams that are already progressive or use extended profiles. Returns 1 if patched, 0 otherwise. */ static int sps_patch_interlace(uint8_t *pkt){ int o = ts_poff(pkt); if(o<0) return 0; const uint8_t *pay = pkt+o; int plen = TS_SZ-o; /* find SPS start code in payload */ for(int i=0; i+4=plen) break; uint8_t nal_type = pay[n]&0x1F; if(nal_type != 7){ i=n; continue; } /* not SPS, keep scanning */ /* NAL unit: [forbidden_zero(1) nal_ref_idc(2) nal_unit_type(5)] */ /* SPS starts at n+1 in the payload */ int s = n+1; /* s = first SPS byte = profile_idc */ if(s+4 >= plen) break; uint8_t profile = pay[s]; /* profile_idc */ /* uint8_t constraints = pay[s+1]; */ /* uint8_t level = pay[s+2]; */ /* seq_parameter_set_id: exp-golomb, usually 1 byte (0x00 = id 0) */ /* For Baseline (66) and Main (77) without chroma/bit-depth ext: frame_mbs_only_flag is at a fixed offset after the 3-byte header + id. Most SD broadcast uses Main profile with id=0, putting the flag at byte offset s+4. We verify the byte is 0 before patching. */ if(profile != 66 && profile != 77 && profile != 88) break; /* offset to frame_mbs_only_flag for Baseline/Main/Extended, seq_param_set_id=0 (single-byte Exp-Golomb = 0x00) */ int flag_byte_off = s+4; /* within pay[] */ if(flag_byte_off >= plen) break; if(pay[flag_byte_off] & 0x80){ /* frame_mbs_only_flag is the MSB of this byte */ if(pay[flag_byte_off] & 0x80) break; /* already 1 = progressive */ /* bit is 0 = interlaced — set it */ pkt[o + flag_byte_off] |= 0x80; return 1; } break; } return 0; } /* -- PES PTS/DTS reset ----------------------------------------- Reads PTS (and DTS if present) from a PES header, subtracts pts_base, and writes the adjusted values back. PTS encoding in PES (ISO 13818-1 §2.4.3.7): byte 9 : 0010 PTS[32..30] 1 byte 10 : PTS[29..22] byte 11 : PTS[21..15] 1 byte 12 : PTS[14..7] byte 13 : PTS[6..0] 1 DTS follows immediately in the same 5-byte format. We operate on the raw TS packet so we need to skip the TS header (ts_poff) to reach the PES header. */ static int64_t pes_read_pts(const uint8_t *p){ return (int64_t)(((uint64_t)(p[0]&0x0E)<<29)| ((uint64_t) p[1] <<22)| ((uint64_t)(p[2]&0xFE)<<14)| ((uint64_t) p[3] << 7)| ((uint64_t)(p[4]&0xFE)>> 1)); } static void pes_write_pts(uint8_t *p, int64_t v, uint8_t marker4){ /* marker4 = 0x21 for PTS-only, 0x31 for PTS when DTS present */ p[0] = (uint8_t)(marker4 | ((v>>29)&0x0E)); p[1] = (uint8_t)((v>>22)&0xFF); p[2] = (uint8_t)(0x01 | ((v>>14)&0xFE)); p[3] = (uint8_t)((v>> 7)&0xFF); p[4] = (uint8_t)(0x01 | ((v<< 1)&0xFE)); } /* Patch PTS/DTS in a TS packet that carries a PES header (PUSI=1). pts_base is subtracted; result is clamped to >= 0. Returns 1 if pts_base was freshly captured, 0 otherwise. */ static int pes_patch_pts(uint8_t *pkt, int64_t *pts_base, int *pts_base_ok){ if(!ts_pusi(pkt)) return 0; int o = ts_poff(pkt); if(o<0||o+8>=TS_SZ) return 0; const uint8_t *pes = pkt+o; /* PES start code: 00 00 01 */ if(pes[0]!=0||pes[1]!=0||pes[2]!=1) return 0; /* PES header flags byte */ if(o+9 >= TS_SZ) return 0; uint8_t pts_dts_flags = (pes[7]>>6)&0x03; if(pts_dts_flags == 0) return 0; /* no PTS present */ int pts_off = o+9; /* first byte of PTS field within pkt[] */ if(pts_off+5 > TS_SZ) return 0; int64_t pts = pes_read_pts(pkt+pts_off); if(!*pts_base_ok){ *pts_base = pts; *pts_base_ok = 1; } int64_t new_pts = pts - *pts_base; if(new_pts < 0) new_pts = 0; uint8_t m4_pts = (pts_dts_flags==3) ? 0x31 : 0x21; pes_write_pts(pkt+pts_off, new_pts, m4_pts); if(pts_dts_flags == 3){ int dts_off = pts_off+5; if(dts_off+5 > TS_SZ) return 0; int64_t dts = pes_read_pts(pkt+dts_off); int64_t new_dts = dts - *pts_base; if(new_dts < 0) new_dts = 0; pes_write_pts(pkt+dts_off, new_dts, 0x11); } return 0; } /* forward decl -- pkt_process is defined later in this file (shared by * both simple-copy mode and the transcoder output bridge below) */ static void pkt_process(Ch *c, const uint8_t *p, const struct timespec *now); /* -------------------------------------------------------------- TRANSCODER START / STOP / OUTPUT BRIDGE (replaces FFMPEG COMMAND BUILDER + FFMPEG PIPE START/STOP/WRITE) Design: the transcoder (transcode.c, linked against libavformat/ libavcodec/libswresample, no ffmpeg subprocess) demuxes/decodes/ re-encodes/re-muxes entirely in-process and hands back finished 188-byte MPEG-TS packets one at a time via tc_output_cb(). Those packets are fed straight into pkt_process() -- the SAME function simple-copy mode uses -- so segment/playlist/AES/IDR-cut logic is shared by both modes instead of duplicated. pkt_process() re-learns PAT/PMT/PIDs from whatever bytes it is given, so feeding it the transcoder's OUTPUT stream (which has its own PAT/PMT/PIDs, assigned by libavformat's mpegts muxer) works correctly as long as the ORIGINAL input is never also fed to it for the same channel (that would corrupt c->vid_pid/c->pmt_pid with two unrelated PID spaces). See the recv loop: in transcoder mode, raw input goes ONLY to tc_feed(); only the transcoder's own output reaches pkt_process(). -------------------------------------------------------------- */ static void tc_output_cb(void *user, const uint8_t *ts188){ Ch *c = (Ch*)user; struct timespec now; clock_gettime(CLOCK_MONOTONIC, &now); pkt_process(c, ts188, &now); } /* Build a TcOpts from this channel's ChOpts. Field-for-field mirror of * the old build_ffmpeg_cmd()'s option handling, minus shell-string * formatting -- these are passed directly as a struct, no command * line / no subprocess. */ static void build_tc_opts(Ch *c, TcOpts *o){ memset(o, 0, sizeof *o); o->video_h264 = c->opts.video_h264; o->video_crf = c->opts.video_crf; snprintf(o->video_bitrate, sizeof o->video_bitrate, "%s", c->opts.video_bitrate); snprintf(o->video_preset, sizeof o->video_preset, "%s", c->opts.video_preset); snprintf(o->video_tune, sizeof o->video_tune, "%s", c->opts.video_tune); o->video_threads = c->opts.video_threads; o->deinterlace = c->opts.deinterlace; o->audio_aac = c->opts.audio_aac; o->audio_map = c->opts.audio_map; snprintf(o->audio_bitrate, sizeof o->audio_bitrate, "%s", c->opts.audio_bitrate); snprintf(o->aac_mode, sizeof o->aac_mode, "%s", c->opts.aac_mode); o->prefer_fdk = 1; /* always try libfdk_aac first; falls back to native aac automatically inside transcode.c if not linked/available */ o->segment_secs = SEGMENT_SECS; o->verbose = g_aac_debug; /* AAC_DEBUG=1 env var also controls transcoder logging now */ } static int tc_start_for_channel(Ch *c){ TcOpts o; build_tc_opts(c, &o); c->tc = tc_start(&o, tc_output_cb, c, c->name); if(!c->tc){ fprintf(stderr,"[%s] transcoder failed to start\n", c->name); return 0; } printf("[%s] transcoder live (in-process, no ffmpeg subprocess)\n", c->name); return 1; } static void tc_stop_for_channel(Ch *c){ if(!c->tc) return; c->tc_drops = tc_drop_count(c->tc); /* snapshot before free */ tc_stop(c->tc); c->tc = NULL; } /* write one chunk of raw input TS bytes into the transcoder. Mirrors * the old ffp_write()'s non-blocking contract: never blocks the NIC * recv thread -- tc_write() only pushes into a ring buffer guarded by * its own lock and returns immediately; the actual demux/decode/ * encode work happens on the transcoder's dedicated pump thread. */ static void tc_feed(Ch *c, const uint8_t *p, int len){ if(!c->tc) return; tc_write(c->tc, p, len); c->tc_drops = tc_drop_count(c->tc); /* keep stats live during operation */ } /* -------------------------------------------------------------- WRITE HELPERS (simple-copy mode) -------------------------------------------------------------- */ static void wb_flush(Ch *c){ if(!c->wb_n || c->seg_fd<0) return; ssize_t r = write(c->seg_fd, c->wb, (size_t)c->wb_n * TS_SZ); /* write errors logged via event ring in pkt_write callers if needed */ (void)r; c->wb_n = 0; } static void pkt_write(Ch *c, const uint8_t *p){ if(c->seg_fd < 0) return; if(c->aes_on) aes_push(&c->aes, c->seg_fd, p); else { memcpy(c->wb+c->wb_n*TS_SZ, p, TS_SZ); if(++c->wb_n >= WRITE_BATCH) wb_flush(c); } } /* -------------------------------------------------------------- PLAYLIST -------------------------------------------------------------- */ static void write_m3u8(Ch *c){ char tmp[512], fin[512]; snprintf(fin, 512, "%s/%s.m3u8", c->dir, c->name); snprintf(tmp, 512, "%s/%s.m3u8.tmp", c->dir, c->name); FILE *f = fopen(tmp,"w"); if(!f) return; uint64_t tot = c->seg_seq - c->start_seq; uint64_t win = tot > (uint64_t)MAX_SEGMENTS ? (uint64_t)MAX_SEGMENTS : tot; uint64_t seq0 = c->seg_seq - win; double md = (double)SEGMENT_SECS; for(uint64_t s=seq0; sseg_seq; s++){ double d = c->dur[s%DUR_SLOTS]; if(d>md) md = d; } int target_dur = (int)md + 1; fprintf(f,"#EXTM3U\n#EXT-X-VERSION:3\n" "#EXT-X-TARGETDURATION:%d\n" "#EXT-X-MEDIA-SEQUENCE:%llu\n", target_dur, (unsigned long long)seq0); if(c->keyuri[0]) fprintf(f,"#EXT-X-KEY:METHOD=AES-128,URI=\"%s\",IV=0x%s\n", c->keyuri, c->iv_hex); for(uint64_t s=seq0; sseg_seq; s++){ double d = c->dur[s%DUR_SLOTS]; if(d<=0) d = (double)SEGMENT_SECS; if(c->seg_size[s%DUR_SLOTS]>0 && d>0){ int kbps = (int)((double)c->seg_size[s%DUR_SLOTS]*8/d/1000); fprintf(f,"#EXTINF:%.3f,\n#EXT-X-BITRATE:%d\nindex%llu.ts\n", d, kbps, (unsigned long long)s); } else fprintf(f,"#EXTINF:%.3f,\nindex%llu.ts\n", d, (unsigned long long)s); } fflush(f); fclose(f); rename(tmp, fin); } /* -------------------------------------------------------------- SEGMENT OPEN / CLOSE (shared by both modes) -------------------------------------------------------------- */ static void seg_close(Ch *c, const struct timespec *now){ if(c->seg_fd < 0) return; if(c->aes_on) aes_flush(&c->aes, c->seg_fd, 1); else wb_flush(c); uint64_t fsize = 0; { struct stat st; if(fstat(c->seg_fd, &st)==0) fsize = (uint64_t)st.st_size; } close(c->seg_fd); c->seg_fd = -1; double elapsed = c->seg_timing_ok ? wall_el(&c->seg_start, now) : 0.0; uint64_t cl = c->seg_seq - 1; c->dur[cl%DUR_SLOTS] = elapsed; c->seg_size[cl%DUR_SLOTS] = fsize; c->segs++; { uint32_t elapsed_ms = (uint32_t)(elapsed * 1000); uint32_t kbps = (elapsed > 0) ? (uint32_t)(fsize*8/elapsed/1000) : 0; Evt e = {EVT_SEG_CLOSE, time(NULL), cl, elapsed_ms, kbps, 0, 0, 0}; event_push(&c->evtring, &e); } } static void seg_open(Ch *c, const struct timespec *now){ int keep = MAX_SEGMENTS + g_del; if(c->seg_seq >= (uint64_t)keep + c->start_seq){ char dp[512]; snprintf(dp, 512, "%s/index%llu.ts", c->dir, (unsigned long long)(c->seg_seq-(uint64_t)keep)); unlink(dp); } char path[512]; snprintf(path, 512, "%s/index%llu.ts", c->dir, (unsigned long long)c->seg_seq); c->seg_fd = open(path, O_WRONLY|O_CREAT|O_TRUNC, 0644); if(c->seg_fd < 0){ Evt e = {EVT_WRITE_ERR, time(NULL), c->seg_seq, 0,0,0,0,0}; event_push(&c->evtring, &e); return; } if(c->aes_on) aes_reset(&c->aes); c->wb_n = 0; c->seg_seq++; c->seg_start = *now; c->seg_timing_ok = 1; if(c->pat_ok) pkt_write(c, c->pat); if(c->pmt_ok) pkt_write(c, c->pmt); write_m3u8(c); { Evt e = {EVT_SEG_OPEN, time(NULL), c->seg_seq-1, 0,0,0,0,0}; event_push(&c->evtring, &e); } } /* NOTE: the old FFMPEG WATCHDOG THREAD (writer_run) is removed -- it * existed to detect the ffmpeg subprocess exiting unexpectedly. The * in-process transcoder has no subprocess to watch; tc_stop() already * joins its pump thread cleanly, and any internal failure is reported * via stderr from transcode.c and reflected in c->use_ffmpeg falling * back to simple-copy at startup if tc_start_for_channel() fails. */ /* -------------------------------------------------------------- PACKET PROCESSOR (shared by simple-copy mode AND the transcoder output bridge -- see tc_output_cb above. Always runs full segment/ IDR-cut/playlist logic; there is no more a separate "ffmpeg owns HLS output" branch, since the transcoder hands back raw TS bytes for THIS function to segment, exactly like simple-copy mode.) -------------------------------------------------------------- */ static void pkt_process(Ch *c, const uint8_t *p, const struct timespec *now){ if(p[0] != 0x47) return; uint16_t pid = ts_pid(p); if(pid == 0x1FFF) return; c->pkts++; c->snap_bytes += TS_SZ; /* PAT */ if(pid == 0x0000){ if(!c->pat_ok){ uint16_t pp = pat_parse(p); if(pp){ c->pmt_pid=pp; c->pat_ok=1; Evt _e={EVT_PAT,time(NULL),0,0,0,pp,0,0}; event_push(&c->evtring,&_e); } } if(c->pat_ok) memcpy(c->pat, p, TS_SZ); } /* PMT */ if(c->pat_ok && pid == c->pmt_pid){ uint16_t vp=0, ap=0, cp=0; if(pmt_parse(p,&cp,&vp,&ap) && cp){ if(!c->pmt_ok || cp!=c->pcr_pid || vp!=c->vid_pid){ c->pcr_pid=cp; c->vid_pid=vp; c->aud_pid=ap; c->pmt_ok=1; { Evt _e={EVT_PMT,time(NULL),0,(uint32_t)ap,(uint32_t)cp,vp,0,0}; event_push(&c->evtring,&_e); } } } memcpy(c->pmt, p, TS_SZ); /* fix_mp2: patch audio stream_type in cached PMT once */ if(c->opts.fix_mp2 && c->aud_pid){ pmt_patch_mp2(c->pmt, c->aud_pid); } } /* CC drop detection -- only reliable drop indicator */ { int afc = ts_afc(p); if((afc==1||afc==3) && pid!=0x1FFF && pid!=0x0000 && pid!=c->pmt_pid){ uint8_t cc = (uint8_t)(p[3]&0x0F); uint8_t prev = c->cc_last[pid&0x1FFF]; if(prev != 0xFF){ uint8_t exp = (uint8_t)((prev+1)&0x0F); if(cc!=exp && cc!=prev){ uint8_t gap = (uint8_t)((cc-exp)&0x0F); c->cc_errors += gap ? gap : 1; if(c->started && !c->recovering){ Evt _e={EVT_CC_DROP,time(NULL),0,gap?(uint32_t)gap:1u,c->cc_errors,(uint16_t)(pid&0x1FFF),exp,cc}; event_push(&c->evtring,&_e); seg_close(c, now); c->recovering = 1; c->seg_timing_ok = 0; } } } c->cc_last[pid&0x1FFF] = cc; } } /* STATE 0: startup gate */ if(!c->started){ if(c->pmt_ok && c->vid_pid && pid==c->vid_pid && ts_pusi(p)){ c->started = 1; { Evt _e={EVT_CLEAN_START,time(NULL),0,0,0,c->vid_pid,0,0}; event_push(&c->evtring,&_e); } seg_open(c, now); pkt_write(c, p); } return; } /* STATE 1: CC recovery -- wait for clean video PUSI */ if(c->recovering){ if(c->vid_pid && pid==c->vid_pid && ts_pusi(p)){ c->recovering = 0; c->idr_found=0; c->idr_slice=0; c->idr_scan=0; c->idr_waiting=0; { Evt _e={EVT_RECOVERED,time(NULL),0,c->cc_errors,0,0,0,0}; event_push(&c->evtring,&_e); } seg_open(c, now); pkt_write(c, p); } return; } /* STATE 2: IDR-aware cut (same as v18) */ if(c->vid_pid && pid==c->vid_pid){ if(!c->idr_off){ int o = ts_poff(p); if(o >= 0){ const uint8_t *pay = p+o; int plen = TS_SZ-o; if(ts_pusi(p)){ c->idr_scan = 1; int rf = -1; if(plen>=9 && pay[0]==0 && pay[1]==0 && pay[2]==1){ int hdr = 9+(int)pay[8]; if(hdr < plen) rf = scan_nal(pay+hdr, plen-hdr); if(rf<0 && 6idr_found = (rf==1); c->idr_slice = (rf>=0); if(c->seg_timing_ok){ double el = wall_el(&c->seg_start, now); if(el >= (double)SEGMENT_SECS){ int do_cut = 0; if(c->idr_off){ do_cut = 1; } else if(c->idr_found){ do_cut=1; c->idr_misses=0; c->idr_waiting=0; } else if(!c->idr_slice){ do_cut=1; c->idr_misses=0; } else if(!c->idr_waiting){ c->idr_waiting=1; c->idr_wait_start=*now; } else { double waited = wall_el(&c->idr_wait_start, now); if(waited >= 0.5){ c->idr_misses++; if(c->idr_misses >= 5){ c->idr_off=1; { Evt _e={EVT_OPEN_GOP,time(NULL),0,0,0,0,0,0}; event_push(&c->evtring,&_e); } } do_cut=1; c->idr_waiting=0; } } if(do_cut){ seg_close(c, now); c->idr_found=0; c->idr_slice=0; seg_open(c, now); pkt_write(c, p); return; } } } if(!c->seg_timing_ok){ c->seg_start=*now; c->seg_timing_ok=1; } } else if(c->idr_scan && !c->idr_slice){ int r = scan_nal(pay, plen); if(r==1){ c->idr_found=1; c->idr_slice=1; } else if(r==0){ c->idr_slice=1; } } } } else if(ts_pusi(p)){ /* idr_off: open-GOP, cut on any PUSI */ if(c->seg_timing_ok){ double el = wall_el(&c->seg_start, now); if(el >= (double)SEGMENT_SECS){ seg_close(c, now); seg_open(c, now); pkt_write(c, p); return; } } if(!c->seg_timing_ok){ c->seg_start=*now; c->seg_timing_ok=1; } } } /* -- zero-CPU stream patches ------------------------------ * Operate on a mutable copy so the original recv buffer is * never modified (other channels sharing this NIC use buf). * NOTE: when transcoding, these patches (pts_reset/fix_interlace) * apply to the TRANSCODER's output stream, not the original * input -- generally unnecessary since transcoding already * normalizes timestamps and SPS, but left available in case a * downstream player still needs them on the re-muxed output. */ if(c->opts.pts_reset || c->opts.fix_interlace){ uint8_t tmp[TS_SZ]; memcpy(tmp, p, TS_SZ); /* PTS reset: apply to video and audio PES PUSI packets */ if(c->opts.pts_reset && (pid==c->vid_pid || pid==c->aud_pid) && ts_pusi(tmp)){ pes_patch_pts(tmp, &c->pts_base, &c->pts_base_ok); } /* SPS interlace fix: apply to video PUSI packets only */ if(c->opts.fix_interlace && pid==c->vid_pid && ts_pusi(tmp)){ sps_patch_interlace(tmp); } pkt_write(c, tmp); return; } pkt_write(c, p); } /* -------------------------------------------------------------- RAW INPUT STATS (transcoder mode only) Lightweight CC-error tracking on the ORIGINAL multicast feed, kept entirely separate from pkt_process()'s cc_last/cc_errors (which track the transcoder's re-muxed OUTPUT once transcoding is active). This answers "is the source itself healthy" independently of "is our re-encoded output healthy" -- useful since a transcoder can mask source drops (decoder error concealment) or, conversely, a perfectly healthy source can still show output-side gaps if the encoder/mux stalls. Does not touch segment state, PAT/PMT, or any field pkt_process() relies on. -------------------------------------------------------------- */ static void raw_input_cc_stats(Ch *c, const uint8_t *p){ if(p[0] != 0x47) return; uint16_t pid = ts_pid(p); if(pid == 0x1FFF) return; int afc = ts_afc(p); if(!(afc==1||afc==3)) return; uint8_t cc = (uint8_t)(p[3]&0x0F); uint8_t prev = c->in_cc_last[pid&0x1FFF]; if(prev != 0xFF){ uint8_t exp = (uint8_t)((prev+1)&0x0F); if(cc!=exp && cc!=prev){ uint8_t gap = (uint8_t)((cc-exp)&0x0F); c->in_cc_errors += gap ? gap : 1; } } c->in_cc_last[pid&0x1FFF] = cc; } /* -------------------------------------------------------------- NIC RECEIVE THREAD (one per NIC group — v18 architecture) -------------------------------------------------------------- */ typedef struct { Ch **chs; int nch; char iface[64]; } NicGroup; static void auto_iface(char *out, size_t len){ struct ifaddrs *l=NULL; getifaddrs(&l); for(struct ifaddrs *i=l; i; i=i->ifa_next){ if(!i->ifa_addr||i->ifa_addr->sa_family!=AF_INET) continue; struct sockaddr_in *sa=(struct sockaddr_in*)i->ifa_addr; if((ntohl(sa->sin_addr.s_addr)>>24)==127) continue; inet_ntop(AF_INET,&sa->sin_addr,out,(socklen_t)len); break; } if(l) freeifaddrs(l); } static int sock_open_ch(Ch *c){ uint32_t grp = inet_addr(c->mcast); int mc = ((ntohl(grp)>>28)==0xE); char iface[64]={0}; if(c->iface[0]) strncpy(iface,c->iface,63); else if(mc) auto_iface(iface,sizeof iface); int fd = socket(AF_INET,SOCK_DGRAM,IPPROTO_UDP); if(fd<0){ perror("socket"); return -1; } int one=1,zero=0; setsockopt(fd,SOL_SOCKET,SO_REUSEADDR,&one,sizeof one); setsockopt(fd,SOL_SOCKET,SO_REUSEPORT,&one,sizeof one); int rb = RCVBUF; int force_ok = (setsockopt(fd,SOL_SOCKET,SO_RCVBUFFORCE,&rb,sizeof rb)==0); if(!force_ok) setsockopt(fd,SOL_SOCKET,SO_RCVBUF,&rb,sizeof rb); { int actual=0; socklen_t sl=sizeof actual; getsockopt(fd,SOL_SOCKET,SO_RCVBUF,&actual,&sl); if(actual/2 < rb/2) fprintf(stderr,"[WARN] socket buf: requested %dMB got %dMB " "(%s)\n", rb>>20, actual>>20, force_ok?"FORCE ok":"FORCE failed, used SO_RCVBUF"); } /* SO_PRIORITY=6: hint to kernel qdisc to prefer these sockets. * Reduces kernel-side drop probability under NIC congestion. * Silently ignored if process lacks CAP_NET_ADMIN — harmless. */ int prio = 6; setsockopt(fd, SOL_SOCKET, SO_PRIORITY, &prio, sizeof prio); int fl = fcntl(fd,F_GETFL,0); fcntl(fd,F_SETFL,fl|O_NONBLOCK); struct sockaddr_in sa={0}; sa.sin_family=AF_INET; sa.sin_port=htons((uint16_t)c->port); if(mc){ sa.sin_addr.s_addr=grp; if(bind(fd,(struct sockaddr*)&sa,sizeof sa)<0){ sa.sin_addr.s_addr=htonl(INADDR_ANY); bind(fd,(struct sockaddr*)&sa,sizeof sa); } struct ip_mreq mr={0}; mr.imr_multiaddr.s_addr=grp; mr.imr_interface.s_addr=iface[0]?inet_addr(iface):htonl(INADDR_ANY); if(setsockopt(fd,IPPROTO_IP,IP_ADD_MEMBERSHIP,&mr,sizeof mr)<0){ perror("IP_ADD_MEMBERSHIP"); close(fd); return -1; } setsockopt(fd,IPPROTO_IP,IP_MULTICAST_ALL,&zero,sizeof zero); } else { sa.sin_addr.s_addr=htonl(INADDR_ANY); bind(fd,(struct sockaddr*)&sa,sizeof sa); } return fd; } static void mkdirp(const char *p){ char t[512]; snprintf(t,512,"%s",p); for(char *s=t+1;*s;s++) if(*s=='/'){ *s=0; mkdir(t,0755); *s='/'; } mkdir(t,0755); } static void *nic_recv_thread(void *arg){ NicGroup *g = (NicGroup *)arg; int nch = g->nch; struct pollfd *pfds = calloc(nch, sizeof(struct pollfd)); for(int i=0; ichs[i]; mkdirp(c->dir); if(c->seg_seq==0) c->seg_seq = c->start_seq; c->seg_fd=-1; c->rtp_ok=0; c->rtp_off=0; c->pat_ok=0; c->pmt_ok=0; c->seg_timing_ok=0; c->started=0; c->recovering=0; c->wb_n=0; c->aes_on=0; memset(c->cc_last,0xFF,sizeof c->cc_last); c->cc_init=0; memset(c->in_cc_last,0xFF,sizeof c->in_cc_last); c->in_cc_init=0; c->in_cc_errors=0; c->idr_found=0; c->idr_slice=0; c->idr_scan=0; c->idr_off=0; c->idr_misses=0; c->idr_waiting=0; c->pts_base_ok=0; c->pts_base=0; c->snap_pkts=0; c->snap_bytes=0; c->snap_cc_errors=0; clock_gettime(CLOCK_MONOTONIC,&c->snap_time); memset(&c->evtring, 0, sizeof c->evtring); c->tc = NULL; c->tc_drops = 0; clock_gettime(CLOCK_MONOTONIC,&c->last_pkt); /* determine mode */ c->use_ffmpeg = (c->opts.audio_aac || c->opts.audio_map>=0 || c->opts.video_h264); /* AES key */ if(c->keyfile[0] && strcmp(c->keyfile,"-")!=0){ FILE *kf = fopen(c->keyfile,"rb"); if(kf){ uint8_t key[16]; size_t n=fread(key,1,16,kf); fclose(kf); if(n==16){ uint8_t iv[16]={0}; int ok=1; if(c->iv_hex[0]&&strcmp(c->iv_hex,"-")!=0) ok=parse_iv(c->iv_hex,iv); if(ok){ aes_init(&c->aes,key,iv); c->aes_on=1; printf("[%s] AES ready\n", c->name); }}} } /* start in-process transcoder if needed. No subprocess, no pipe * readiness wait, no SPS gate: the transcoder's own demuxer * (avformat_open_input/find_stream_info) handles starting * mid-stream/mid-GOP internally, the same way any standard * MPEG-TS player would. */ if(c->use_ffmpeg){ if(!tc_start_for_channel(c)){ fprintf(stderr,"[%s] transcoder failed, falling back to simple-copy\n",c->name); c->use_ffmpeg = 0; } } if(!c->use_ffmpeg) printf("[%s] simple-copy mode\n", c->name); c->fd = sock_open_ch(c); if(c->fd < 0){ pfds[i].fd=-1; pfds[i].events=0; continue; } pfds[i].fd=c->fd; pfds[i].events=POLLIN; printf("[%s] joined %s:%d iface=%s mode=%s\n", c->name, c->mcast, c->port, c->iface[0]?c->iface:"auto", c->use_ffmpeg?"transcoder":"simple-copy"); } printf("[NIC:%s] watching %d channels\n", g->iface[0]?g->iface:"auto", nch); fflush(stdout); static uint8_t buf[UDP_MAX]; while(!g_stop){ int ready = poll(pfds, nch, 10); if(ready<0){ if(errno==EINTR) continue; break; } struct timespec now; clock_gettime(CLOCK_MONOTONIC,&now); for(int i=0; ichs[i]; if(pfds[i].fd < 0) continue; /* stale: no data for 2× segment duration */ if(c->started && c->seg_fd>=0){ double age = wall_el(&c->last_pkt, &now); if(age >= (double)(SEGMENT_SECS*2)){ { uint32_t age_ms=(uint32_t)(age*1000); Evt _e={EVT_STALE,time(NULL),0,age_ms,0,0,0,0}; event_push(&c->evtring,&_e); } seg_close(c, &now); c->started=0; c->recovering=0; c->seg_timing_ok=0; } } if(!(pfds[i].revents & POLLIN)) continue; /* Drain at most MAX_PKTS_PER_CHAN datagrams then move to the * next channel. Draining to EAGAIN here starves sibling * channels sharing this NIC thread ? their socket buffers fill * ? input drops even on a clean source (BUG 2 in v20). */ for(int pkt_limit = 0; pkt_limit < MAX_PKTS_PER_CHAN && !g_stop; pkt_limit++){ ssize_t n = recv(c->fd, buf, sizeof buf, 0); if(n<=0){ if(errno==EAGAIN||errno==EWOULDBLOCK) break; if(errno==EINTR) continue; break; } c->dgrams++; clock_gettime(CLOCK_MONOTONIC,&c->last_pkt); if(c->dgrams==1){ printf("[%s] first dgram len=%zd\n", c->name, n); if(buf[0]!=0x47){ int o=rtp_detect(buf,(int)n); if(o>0){ c->rtp_off=o; printf("[%s] RTP+%d\n",c->name,o); } } else printf("[%s] plain TS\n", c->name); } const uint8_t *ts = buf + c->rtp_off; int np = (int)(n - c->rtp_off) / TS_SZ; for(int j=0; juse_ffmpeg){ /* Track raw-input source health (CC errors on the * ORIGINAL multicast feed) independently of the * transcoder's output -- see in_cc_* fields. This * does NOT touch c->pat_ok/vid_pid/segment state, * which belong exclusively to the transcoder's * output stream (fed via tc_output_cb -> pkt_process * below, on a different PID space entirely). */ raw_input_cc_stats(c, ts); /* Hand raw bytes to the transcoder. No SPS/IDR gate * needed here: libavformat's own mpegts demuxer * (avformat_open_input/find_stream_info inside * transcode.c) already handles starting mid-GOP / * mid-stream the same way any standard TS player * does, so there's nothing to gate on at this layer. */ tc_feed(c, ts, TS_SZ); } else pkt_process(c, ts, &now); /* direct copy */ } } } } /* -- shutdown -------------------------------------------- */ for(int i=0; ichs[i]; if(c->use_ffmpeg){ /* tc_stop_for_channel() flushes the encoders, writes the * mpegts trailer, and joins the pump thread -- any final * buffered frames are emitted through tc_output_cb() -> * pkt_process() during this call, which may itself call * seg_close()/pkt_write() on c->seg_fd. So stop the * transcoder BEFORE closing the segment fd below, the same * ordering simple-copy mode already relies on. */ tc_stop_for_channel(c); } if(c->seg_fd >= 0){ struct timespec now; clock_gettime(CLOCK_MONOTONIC,&now); if(c->aes_on) aes_flush(&c->aes, c->seg_fd, 1); else wb_flush(c); close(c->seg_fd); printf("[%s] final\n", c->name); } if(c->fd >= 0) close(c->fd); } free(pfds); return NULL; } /* -------------------------------------------------------------- OPTIONS PARSER -------------------------------------------------------------- */ static void parse_opts(ChOpts *o, const char *toks[], int ntok){ memset(o, 0, sizeof *o); o->audio_map = -1; for(int i=0; iaudio_map=-1; else o->audio_map=atoi(t+10); } else if(!strncmp(t,"video_h264=",11)) o->video_h264 = atoi(t+11); else if(!strncmp(t,"audio_bitrate=",14)) snprintf(o->audio_bitrate,sizeof o->audio_bitrate,"%s",t+14); else if(!strncmp(t,"aac_mode=",9)) snprintf(o->aac_mode,sizeof o->aac_mode,"%s",t+9); else if(!strncmp(t,"video_bitrate=",14)) snprintf(o->video_bitrate,sizeof o->video_bitrate,"%s",t+14); else if(!strncmp(t,"video_preset=",13)) snprintf(o->video_preset,sizeof o->video_preset,"%s",t+13); else if(!strncmp(t,"video_crf=",10)) o->video_crf = atoi(t+10); else if(!strncmp(t,"video_tune=",11)) snprintf(o->video_tune,sizeof o->video_tune,"%s",t+11); else if(!strncmp(t,"deinterlace=",12)) o->deinterlace = atoi(t+12); else if(!strncmp(t,"video_threads=",14)) o->video_threads = atoi(t+14); else if(!strncmp(t,"fix_mp2=",8)) o->fix_mp2 = atoi(t+8); else if(!strncmp(t,"fix_interlace=",14)) o->fix_interlace = atoi(t+14); else if(!strncmp(t,"pts_reset=",10)) o->pts_reset = atoi(t+10); } } /* -------------------------------------------------------------- CONFIG LOADER -------------------------------------------------------------- */ static int conf_load(const char *path){ FILE *f = fopen(path,"r"); if(!f){ perror(path); return 0; } int n=0; char line[2048]; while(fgets(line,sizeof line,f) && n0&&(line[l-1]==' '||line[l-1]=='\t'))line[--l]=0; if(!line[0]) continue; Ch *c=&g_ch[n]; memset(c,0,sizeof *c); c->opts.audio_map=-1; unsigned long long seq=0; char opt_buf[1024]={0}; int r=sscanf(line, "%63s %d %255s %63s %63s %255s %511s %32s %llu %1023[^\n]", c->mcast,&c->port,c->dir,c->name, c->iface,c->keyfile,c->keyuri,c->iv_hex,&seq,opt_buf); if(r<4) continue; c->start_seq=(uint64_t)seq; if(c->iface[0] &&!strcmp(c->iface, "-")) c->iface[0]=0; if(c->keyfile[0]&&!strcmp(c->keyfile,"-")) c->keyfile[0]=0; if(c->keyuri[0] &&!strcmp(c->keyuri, "-")) c->keyuri[0]=0; if(c->iv_hex[0] &&!strcmp(c->iv_hex, "-")) c->iv_hex[0]=0; if(opt_buf[0]){ const char *toks[32]; int ntok=0; char *sv=NULL; char *tok=strtok_r(opt_buf," \t",&sv); while(tok&&ntok<32){ toks[ntok++]=tok; tok=strtok_r(NULL," \t",&sv); } parse_opts(&c->opts, toks, ntok); } const char *mode = (c->opts.audio_aac||c->opts.audio_map>=0||c->opts.video_h264) ? "transcoder" : "simple-copy"; printf("[conf] %s:%d -> %s/%s mode=%s\n", c->mcast,c->port,c->dir,c->name,mode); n++; } fclose(f); return n; } /* -------------------------------------------------------------- STATS THREAD -------------------------------------------------------------- */ static void *stats_run(void *a){ (void)a; const int INTERVAL = 10; while(!g_stop){ sleep(INTERVAL); if(g_stop) break; struct timespec now; clock_gettime(CLOCK_MONOTONIC, &now); char ts[9]; ts_now(ts); uint64_t total_segs=0, total_pkts=0; int stalled=0, recovering=0, no_signal=0; printf("\n[%s] -- stats (%ds interval) --\n", ts, INTERVAL); for(int i=0; ievtring, c->name); /* time since last packet */ double idle = wall_el(&c->last_pkt, &now); /* interval delta */ double dt = wall_el(&c->snap_time, &now); if(dt < 0.001) dt = 0.001; /* guard /0 */ uint64_t dpkts = c->pkts - c->snap_pkts; uint64_t dbytes = c->snap_bytes; /* reset each interval */ uint64_t dcc = c->cc_errors - c->snap_cc_errors; double pps = (double)dpkts / dt; /* packets/sec */ double kbps = (double)dbytes * 8.0 / dt / 1000.0; /* kbps */ /* update snapshots */ c->snap_pkts = c->pkts; c->snap_bytes = 0; c->snap_cc_errors = c->cc_errors; c->snap_time = now; total_segs += c->segs; total_pkts += c->pkts; /* classify channel state */ const char *state; if(!c->started) { state="NO_SIGNAL"; no_signal++; } else if(c->recovering) { state="RECOVERING"; recovering++; } else if(idle > SEGMENT_SECS*3) { state="STALLED"; stalled++; } else { state="OK"; } /* always print every channel -- full picture every interval */ printf(" [%-14s] %s segs=%-6llu %6.1f pkt/s %7.1f kbps" " idle=%5.1fs cc_err_new=%-3llu cc_total=%-5u" " drops=%-4llu mode=%s\n", c->name, state, (unsigned long long)c->segs, pps, kbps, idle, (unsigned long long)dcc, c->cc_errors, (unsigned long long)c->tc_drops, c->use_ffmpeg ? "transcoder" : "simple-copy"); /* -- per-channel problem detail -- */ if(!c->started){ printf(" !! NO_SIGNAL: PAT=%s PMT=%s vid_pid=0x%04X" " — check multicast source/iface\n", c->pat_ok?"ok":"missing", c->pmt_ok?"ok":"missing", c->vid_pid); } if(c->recovering){ printf(" !! RECOVERING: waiting for clean IDR after CC drop" " (total_lost=%u)\n", c->cc_errors); } if(idle > SEGMENT_SECS*3 && c->started){ printf(" !! STALLED: no input for %.1fs" " — source dropped or NIC issue\n", idle); } if(dcc > 0){ printf(" !! CC ERRORS this interval: %llu new drops" " (total=%u) — check network/switch/igmp\n", (unsigned long long)dcc, c->cc_errors); } if(c->tc_drops > 0 && c->use_ffmpeg){ printf(" !! TRANSCODER DROPS: %llu bytes total -- encoder too" " slow for input bitrate, try lower preset or crf\n", (unsigned long long)c->tc_drops); } if(c->in_cc_errors > 0 && c->use_ffmpeg){ printf(" !! SOURCE CC ERRORS (pre-transcode): %u total --" " original multicast feed is dropping packets," " independent of transcoder output health\n", c->in_cc_errors); } if(c->idr_off){ printf(" ~~ open-GOP stream: cutting on any PUSI\n"); } if(kbps < 100 && c->started && idle < 5.0){ printf(" ~~ LOW BITRATE: %.1f kbps — possible partial stream\n", kbps); } } printf(" -- TOTAL: segs=%llu pkts=%llu" " stalled=%d recovering=%d no_signal=%d / %d channels\n", (unsigned long long)total_segs, (unsigned long long)total_pkts, stalled, recovering, no_signal, g_nch); printf("[%s] ----------------------------------\n\n", ts); fflush(stdout); } return NULL; } /* -------------------------------------------------------------- MAIN -------------------------------------------------------------- */ int main(int argc, char *argv[]){ signal(SIGINT, onsig); signal(SIGTERM, onsig); signal(SIGPIPE, SIG_IGN); signal(SIGCHLD, SIG_DFL); memset(g_ch, 0, sizeof g_ch); const char *e; if((e=getenv("HLS_DELETE_THRESHOLD"))){ int t=atoi(e); if(t>=0) g_del=t; } if(getenv("AAC_DEBUG")) g_aac_debug=1; if(argc==2){ g_nch=conf_load(argv[1]); if(!g_nch){ fprintf(stderr,"no channels\n"); return 1; } } else if(argc>=5){ Ch *c=&g_ch[0]; memset(c,0,sizeof *c); c->opts.audio_map=-1; strncpy(c->mcast, argv[1],63); c->port=atoi(argv[2]); strncpy(c->dir, argv[3],255); strncpy(c->name, argv[4],63); if(argc>5) strncpy(c->iface, argv[5],63); if(argc>6) strncpy(c->keyfile, argv[6],255); if(argc>7) strncpy(c->keyuri, argv[7],511); if(argc>8) strncpy(c->iv_hex, argv[8],32); if(argc>9) c->start_seq=(uint64_t)atoll(argv[9]); if(argc>10){ const char *toks[32]; int ntok=0; for(int i=10; iopts, toks, ntok); } g_nch=1; } else { fprintf(stderr, "UDP->HLS v25 (in-process transcoder: libx264/libfdk_aac via libavformat, no ffmpeg subprocess)\n\n" "Usage:\n" " %s channels.conf\n" " %s [iface] [keyfile] [keyuri]" " [iv_hex] [seq] [opts...]\n\n" "Options (transcoder mode -- in-process, no subprocess):\n" " audio_aac=1 convert audio to AAC (libfdk_aac if available, else native aac)\n" " audio_map=0 use one specific audio stream (0-based); -1 (default) = ALL streams\n" " video_h264=1 transcode video to H264 (libx264)\n" " aac_mode=vbr3 VBR quality vbr1..vbr5 or cbr (libfdk_aac only)\n" " audio_bitrate=128k AAC CBR bitrate\n" " video_preset=ultrafast x264 preset (ultrafast/superfast/veryfast/fast)\n" " video_bitrate=2000k x264 CBR bitrate (ignored if video_crf set)\n" " video_crf=23 x264 CRF quality 0-51 (overrides video_bitrate)\n" " video_tune=zerolatency x264 tune (zerolatency saves ~15%% CPU)\n" " deinterlace=1 add yadif-equivalent deinterlace filter (576i->progressive)\n" " video_threads=2 limit x264 threads (reduces CPU, default=auto)\n" "Zero-CPU fix options (simple-copy, no re-encode):\n" " fix_mp2=1 patch PMT audio type 0x03->0x0F (LG MP2 fix)\n" " fix_interlace=1 patch H264 SPS progressive flag (LG 576i fix)\n" " pts_reset=1 reset PTS/DTS to near-zero (LG high-PTS fix)\n\n" "Env:\n" " HLS_DELETE_THRESHOLD=N extra segments on disk [default %d]\n" " AAC_DEBUG=1 show transcoder lifecycle/error diagnostics\n", argv[0], argv[0], HLS_DELETE_THRESHOLD); return 1; } NicGroup groups[MAX_NICS]; int ngroups=0; memset(groups, 0, sizeof groups); for(int i=0; iiface[0]?c->iface:""; int gidx=-1; for(int j=0; jHLS final ch=%d groups=%d seg=%ds ===\n\n", g_nch, ngroups, SEGMENT_SECS); pthread_t gtids[MAX_NICS], stid; pthread_create(&stid, NULL, stats_run, NULL); for(int i=0; i