From 5e30711bfdb28085234a5ef6da4f4e44305ac3e4 Mon Sep 17 00:00:00 2001 From: Frank Schmirler Date: Thu, 2 Dec 2010 08:53:01 +0100 Subject: Snapshot 2007-03-20 --- remux/extern.c | 157 +++++++++++++++++++++++++++++++++++++++++ remux/extern.h | 25 +++++++ remux/ts2es.c | 140 +++++++++++++++++++++++++++++++++++++ remux/ts2es.h | 25 +++++++ remux/ts2ps.c | 211 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++ remux/ts2ps.h | 29 ++++++++ remux/tsremux.c | 59 ++++++++++++++++ remux/tsremux.h | 33 +++++++++ 8 files changed, 679 insertions(+) create mode 100644 remux/extern.c create mode 100644 remux/extern.h create mode 100644 remux/ts2es.c create mode 100644 remux/ts2es.h create mode 100644 remux/ts2ps.c create mode 100644 remux/ts2ps.h create mode 100644 remux/tsremux.c create mode 100644 remux/tsremux.h (limited to 'remux') diff --git a/remux/extern.c b/remux/extern.c new file mode 100644 index 0000000..51c353c --- /dev/null +++ b/remux/extern.c @@ -0,0 +1,157 @@ +#include "remux/extern.h" +#include "server/streamer.h" +#include +#include +#include +#include +#include + +const char *g_ExternRemux = "/root/externremux.sh"; + +class cTSExt: public cThread { +private: + cRingBufferLinear *m_ResultBuffer; + bool m_Active; + int m_Process; + int m_Inpipe, m_Outpipe; + +protected: + virtual void Action(void); + +public: + cTSExt(cRingBufferLinear *ResultBuffer); + virtual ~cTSExt(); + + void Put(const uchar *Data, int Count); +}; + +cTSExt::cTSExt(cRingBufferLinear *ResultBuffer): + m_ResultBuffer(ResultBuffer), + m_Active(false), + m_Process(0), + m_Inpipe(0), + m_Outpipe(0) +{ + int inpipe[2]; + int outpipe[2]; + + if (pipe(inpipe) == -1) { + LOG_ERROR_STR("pipe failed"); + return; + } + + if (pipe(outpipe) == -1) { + LOG_ERROR_STR("pipe failed"); + close(inpipe[0]); + close(inpipe[1]); + return; + } + + if ((m_Process = fork()) == -1) { + LOG_ERROR_STR("fork failed"); + close(inpipe[0]); + close(inpipe[1]); + close(outpipe[0]); + close(outpipe[1]); + return; + } + + if (m_Process == 0) { + // child process + dup2(inpipe[0], STDIN_FILENO); + close(inpipe[1]); + dup2(outpipe[1], STDOUT_FILENO); + close(outpipe[0]); + + int MaxPossibleFileDescriptors = getdtablesize(); + for (int i = STDERR_FILENO + 1; i < MaxPossibleFileDescriptors; i++) + close(i); //close all dup'ed filedescriptors + + //printf("starting externremux.sh\n"); + execl("/bin/sh", "sh", "-c", g_ExternRemux, NULL); + //printf("failed externremux.sh\n"); + _exit(-1); + } + + close(inpipe[0]); + close(outpipe[1]); + m_Inpipe = inpipe[1]; + m_Outpipe = outpipe[0]; + Start(); +} + +cTSExt::~cTSExt() +{ + m_Active = false; + Cancel(3); + if (m_Process > 0) { + close(m_Outpipe); + close(m_Inpipe); + kill(m_Process, SIGTERM); + for (int i = 0; waitpid(m_Process, NULL, WNOHANG) == 0; i++) { + if (i == 20) { + esyslog("streamdev-server: externremux process won't stop - killing it"); + kill(m_Process, SIGKILL); + } + cCondWait::SleepMs(100); + } + } +} + +void cTSExt::Action(void) +{ + m_Active = true; + while (m_Active) { + fd_set rfds; + struct timeval tv; + + FD_ZERO(&rfds); + FD_SET(m_Outpipe, &rfds); + + while (FD_ISSET(m_Outpipe, &rfds)) { + tv.tv_sec = 2; + tv.tv_usec = 0; + if (select(m_Outpipe + 1, &rfds, NULL, NULL, &tv) == -1) { + LOG_ERROR_STR("poll failed"); + break;; + } + + if (FD_ISSET(m_Outpipe, &rfds)) { + int result; + if ((result = m_ResultBuffer->Read(m_Outpipe)) == -1) { + LOG_ERROR_STR("read failed"); + break; + } + } + } + } + m_Active = false; +} + + +void cTSExt::Put(const uchar *Data, int Count) +{ + if (safe_write(m_Inpipe, Data, Count) == -1) { + LOG_ERROR_STR("write failed"); + return; + } +} + +cExternRemux::cExternRemux(int VPid, const int *APids, const int *Dpids, const int *SPids): + m_ResultBuffer(new cRingBufferLinear(WRITERBUFSIZE, TS_SIZE * 2)), + m_Remux(new cTSExt(m_ResultBuffer)) +{ + m_ResultBuffer->SetTimeouts(0, 100); +} + +cExternRemux::~cExternRemux() +{ + delete m_Remux; + delete m_ResultBuffer; +} + +int cExternRemux::Put(const uchar *Data, int Count) +{ + m_Remux->Put(Data, Count); + return Count; +} diff --git a/remux/extern.h b/remux/extern.h new file mode 100644 index 0000000..ae055ac --- /dev/null +++ b/remux/extern.h @@ -0,0 +1,25 @@ +#ifndef VDR_STREAMDEV_EXTERNREMUX_H +#define VDR_STREAMDEV_EXTERNREMUX_H + +#include "remux/tsremux.h" +#include + +extern const char *g_ExternRemux; + +class cTSExt; + +class cExternRemux: public cTSRemux { +private: + cRingBufferLinear *m_ResultBuffer; + cTSExt *m_Remux; + +public: + cExternRemux(int VPid, const int *APids, const int *Dpids, const int *SPids); + virtual ~cExternRemux(); + + int Put(const uchar *Data, int Count); + uchar *Get(int &Count) { return m_ResultBuffer->Get(Count); } + void Del(int Count) { m_ResultBuffer->Del(Count); } +}; + +#endif // VDR_STREAMDEV_EXTERNREMUX_H diff --git a/remux/ts2es.c b/remux/ts2es.c new file mode 100644 index 0000000..3476e24 --- /dev/null +++ b/remux/ts2es.c @@ -0,0 +1,140 @@ +#include "remux/ts2es.h" +#include "server/streamer.h" +#include "libdvbmpeg/transform.h" +#include "common.h" +#include + +// from VDR's remux.c +#define MAXNONUSEFULDATA (10*1024*1024) + +class cTS2ES: public ipack { + friend void PutES(uint8_t *Buffer, int Size, void *Data); + +private: + cRingBufferLinear *m_ResultBuffer; + +public: + cTS2ES(cRingBufferLinear *ResultBuffer); + ~cTS2ES(); + + void PutTSPacket(const uint8_t *Buffer); +}; + +void PutES(uint8_t *Buffer, int Size, void *Data) +{ + cTS2ES *This = (cTS2ES*)Data; + uint8_t payl = Buffer[8] + 9 + This->start - 1; + int count = Size - payl; + + int n = This->m_ResultBuffer->Put(Buffer + payl, count); + if (n != count) + esyslog("ERROR: result buffer overflow, dropped %d out of %d byte", count - n, count); + This->start = 1; +} + +cTS2ES::cTS2ES(cRingBufferLinear *ResultBuffer) +{ + m_ResultBuffer = ResultBuffer; + + init_ipack(this, IPACKS, PutES, 0); + data = (void*)this; +} + +cTS2ES::~cTS2ES() +{ + free_ipack(this); +} + +void cTS2ES::PutTSPacket(const uint8_t *Buffer) { + if (!Buffer) + return; + + if (Buffer[1] & 0x80) { // ts error + // TODO + } + + if (Buffer[1] & 0x40) { // payload start + if (plength == MMAX_PLENGTH - 6) { + plength = found - 6; + found = 0; + send_ipack(this); + reset_ipack(this); + } + } + + uint8_t off = 0; + + if (Buffer[3] & 0x20) { // adaptation field? + off = Buffer[4] + 1; + if (off + 4 > TS_SIZE - 1) + return; + } + + instant_repack((uint8_t*)(Buffer + 4 + off), TS_SIZE - 4 - off, this); +} + +cTS2ESRemux::cTS2ESRemux(int Pid): + m_Pid(Pid), + m_ResultBuffer(new cRingBufferLinear(WRITERBUFSIZE, IPACKS)), + m_Remux(new cTS2ES(m_ResultBuffer)) +{ + m_ResultBuffer->SetTimeouts(0, 100); +} + +cTS2ESRemux::~cTS2ESRemux() +{ + delete m_Remux; + delete m_ResultBuffer; +} + +int cTS2ESRemux::Put(const uchar *Data, int Count) +{ + int used = 0; + + // Make sure we are looking at a TS packet: + + while (Count > TS_SIZE) { + if (Data[0] == TS_SYNC_BYTE && Data[TS_SIZE] == TS_SYNC_BYTE) + break; + Data++; + Count--; + used++; + } + + if (used) + esyslog("ERROR: skipped %d byte to sync on TS packet", used); + + // Convert incoming TS data into ES: + + for (int i = 0; i < Count; i += TS_SIZE) { + if (Count - i < TS_SIZE) + break; + if (Data[i] != TS_SYNC_BYTE) + break; + if (m_ResultBuffer->Free() < 2 * IPACKS) + break; // A cTS2ES might write one full packet and also a small rest + int pid = cTSRemux::GetPid(Data + i + 1); + if (Data[i + 3] & 0x10) { // got payload + if (m_Pid == pid) + m_Remux->PutTSPacket(Data + i); + } + used += TS_SIZE; + } + +/* + // Check if we're getting anywhere here: + if (!synced && skipped >= 0) { + if (skipped > MAXNONUSEFULDATA) { + esyslog("ERROR: no useful data seen within %d byte of video stream", skipped); + skipped = -1; + if (exitOnFailure) + cThread::EmergencyExit(true); + } + else + skipped += used; + } +*/ + + return used; +} + diff --git a/remux/ts2es.h b/remux/ts2es.h new file mode 100644 index 0000000..551df1d --- /dev/null +++ b/remux/ts2es.h @@ -0,0 +1,25 @@ +#ifndef VDR_STREAMDEV_TS2ESREMUX_H +#define VDR_STREAMDEV_TS2ESREMUX_H + +#include "remux/tsremux.h" +#include + +class cTS2ES; +class cRingBufferLinear; + +class cTS2ESRemux: public cTSRemux { +private: + int m_Pid; + cRingBufferLinear *m_ResultBuffer; + cTS2ES *m_Remux; + +public: + cTS2ESRemux(int Pid); + virtual ~cTS2ESRemux(); + + int Put(const uchar *Data, int Count); + uchar *Get(int &Count) { return m_ResultBuffer->Get(Count); } + void Del(int Count) { m_ResultBuffer->Del(Count); } +}; + +#endif // VDR_STREAMDEV_TS2ESREMUX_H diff --git a/remux/ts2ps.c b/remux/ts2ps.c new file mode 100644 index 0000000..d0d08cf --- /dev/null +++ b/remux/ts2ps.c @@ -0,0 +1,211 @@ +#include "remux/ts2ps.h" +#include "server/streamer.h" +#include +#include + +class cTS2PS { + friend void PutPES(uint8_t *Buffer, int Size, void *Data); + +private: + ipack m_Ipack; + int m_Pid; + cRingBufferLinear *m_ResultBuffer; + +public: + cTS2PS(cRingBufferLinear *ResultBuffer, int Pid, uint8_t AudioCid = 0x00); + ~cTS2PS(); + + void PutTSPacket(const uint8_t *Buffer); + + int Pid(void) const { return m_Pid; } +}; + +void PutPES(uint8_t *Buffer, int Size, void *Data) +{ + cTS2PS *This = (cTS2PS*)Data; + int n = This->m_ResultBuffer->Put(Buffer, Size); + if (n != Size) + esyslog("ERROR: result buffer overflow, dropped %d out of %d byte", Size - n, Size); +} + +cTS2PS::cTS2PS(cRingBufferLinear *ResultBuffer, int Pid, uint8_t AudioCid) +{ + m_ResultBuffer = ResultBuffer; + m_Pid = Pid; + + init_ipack(&m_Ipack, IPACKS, PutPES, false); + m_Ipack.cid = AudioCid; + m_Ipack.data = (void*)this; +} + +cTS2PS::~cTS2PS() +{ + free_ipack(&m_Ipack); +} + +void cTS2PS::PutTSPacket(const uint8_t *Buffer) +{ + if (!Buffer) + return; + + if (Buffer[1] & 0x80) { // ts error + // TODO + } + + if (Buffer[1] & 0x40) { // payload start + if (m_Ipack.plength == MMAX_PLENGTH - 6 && m_Ipack.found > 6) { + m_Ipack.plength = m_Ipack.found - 6; + m_Ipack.found = 0; + send_ipack(&m_Ipack); + reset_ipack(&m_Ipack); + } + } + + uint8_t off = 0; + + if (Buffer[3] & 0x20) { // adaptation field? + off = Buffer[4] + 1; + if (off + 4 > TS_SIZE - 1) + return; + } + + instant_repack((uint8_t*)(Buffer + 4 + off), TS_SIZE - 4 - off, &m_Ipack); +} + +cTS2PSRemux::cTS2PSRemux(int VPid, const int *APids, const int *DPids, const int *SPids): + m_NumTracks(0), + m_ResultBuffer(new cRingBufferLinear(WRITERBUFSIZE, IPACKS)), + m_ResultSkipped(0), + m_Skipped(0), + m_Synced(false), + m_IsRadio(VPid == 0 || VPid == 1 || VPid == 0x1FFF) +{ + m_ResultBuffer->SetTimeouts(0, 100); + + if (VPid) + m_Remux[m_NumTracks++] = new cTS2PS(m_ResultBuffer, VPid); + if (APids) { + int n = 0; + while (*APids && m_NumTracks < MAXTRACKS && n < MAXAPIDS) + m_Remux[m_NumTracks++] = new cTS2PS(m_ResultBuffer, *APids++, 0xC0 + n++); + } + if (DPids) { + int n = 0; + while (*DPids && m_NumTracks < MAXTRACKS && n < MAXDPIDS) + m_Remux[m_NumTracks++] = new cTS2PS(m_ResultBuffer, *DPids++, 0x80 + n++); + } +} + +cTS2PSRemux::~cTS2PSRemux() { + for (int i = 0; i < m_NumTracks; ++i) + delete m_Remux[i]; + delete m_ResultBuffer; +} + +int cTS2PSRemux::Put(const uchar *Data, int Count) +{ + int used = 0; + + // Make sure we are looking at a TS packet: + while (Count > TS_SIZE) { + if (Data[0] == TS_SYNC_BYTE && Data[TS_SIZE] == TS_SYNC_BYTE) + break; + Data++; + Count--; + used++; + } + if (used) + esyslog("ERROR: m_Skipped %d byte to sync on TS packet", used); + + // Convert incoming TS data into multiplexed PS: + + for (int i = 0; i < Count; i += TS_SIZE) { + if (Count - i < TS_SIZE) + break; + if (Data[i] != TS_SYNC_BYTE) + break; + if (m_ResultBuffer->Free() < 2 * IPACKS) + break; // A cTS2PS might write one full packet and also a small rest + int pid = GetPid(Data + i + 1); + if (Data[i + 3] & 0x10) { // got payload + for (int t = 0; t < m_NumTracks; t++) { + if (m_Remux[t]->Pid() == pid) { + m_Remux[t]->PutTSPacket(Data + i); + break; + } + } + } + used += TS_SIZE; + } + + // Check if we're getting anywhere here: + if (!m_Synced && m_Skipped >= 0) + m_Skipped += used; + + return used; +} + +uchar *cTS2PSRemux::Get(int &Count) +{ + // Remove any previously skipped data from the result buffer: + + if (m_ResultSkipped > 0) { + m_ResultBuffer->Del(m_ResultSkipped); + m_ResultSkipped = 0; + } + + // Special VPID case to enable recording radio channels: + if (m_IsRadio) { + // Force syncing of radio channels to avoid "no useful data" error + m_Synced = true; + return m_ResultBuffer->Get(Count); + } + + // Check for frame borders: + Count = 0; + uchar *resultData = NULL; + int resultCount = 0; + uchar *data = m_ResultBuffer->Get(resultCount); + if (data) { + for (int i = 0; i < resultCount - 3; i++) { + if (data[i] == 0 && data[i + 1] == 0 && data[i + 2] == 1) { + int l = 0; + uchar StreamType = data[i + 3]; + if (VIDEO_STREAM_S <= StreamType && StreamType <= VIDEO_STREAM_E) { + uchar pt = NO_PICTURE; + l = ScanVideoPacket(data, resultCount, i, pt); + if (l < 0) + return resultData; + if (pt != NO_PICTURE) { + if (pt < I_FRAME || B_FRAME < pt) { + esyslog("ERROR: unknown picture type '%d'", pt); + } + else if (!m_Synced) { + if (pt == I_FRAME) { + m_ResultSkipped = i; // will drop everything before this position + SetBrokenLink(data + i, l); + m_Synced = true; + } + } + else if (Count) + return resultData; + } + } else { + l = GetPacketLength(data, resultCount, i); + if (l < 0) + return resultData; + } + if (m_Synced) { + if (!Count) + resultData = data + i; + Count += l; + } else + m_ResultSkipped = i + l; + if (l > 0) + i += l - 1; // the loop increments, too + } + } + } + return resultData; +} + diff --git a/remux/ts2ps.h b/remux/ts2ps.h new file mode 100644 index 0000000..334215a --- /dev/null +++ b/remux/ts2ps.h @@ -0,0 +1,29 @@ +#ifndef VDR_STREAMDEV_TS2PESREMUX_H +#define VDR_STREAMDEV_TS2PESREMUX_H + +#include "remux/tsremux.h" +#include +#include + +class cTS2PS; + +class cTS2PSRemux: public cTSRemux { +private: + int m_NumTracks; + cTS2PS *m_Remux[MAXTRACKS]; + cRingBufferLinear *m_ResultBuffer; + int m_ResultSkipped; + int m_Skipped; + bool m_Synced; + bool m_IsRadio; + +public: + cTS2PSRemux(int VPid, const int *Apids, const int *Dpids, const int *Spids); + virtual ~cTS2PSRemux(); + + int Put(const uchar *Data, int Count); + uchar *Get(int &Count); + void Del(int Count) { m_ResultBuffer->Del(Count); } +}; + +#endif // VDR_STREAMDEV_TS2PESREMUX_H diff --git a/remux/tsremux.c b/remux/tsremux.c new file mode 100644 index 0000000..6be5245 --- /dev/null +++ b/remux/tsremux.c @@ -0,0 +1,59 @@ +#include "remux/tsremux.h" + +#define SC_PICTURE 0x00 // "picture header" + +void cTSRemux::SetBrokenLink(uchar *Data, int Length) +{ + if (Length > 9 && Data[0] == 0 && Data[1] == 0 && Data[2] == 1 && (Data[3] & 0xF0) == VIDEO_STREAM_S) { + for (int i = Data[8] + 9; i < Length - 7; i++) { // +9 to skip video packet header + if (Data[i] == 0 && Data[i + 1] == 0 && Data[i + 2] == 1 && Data[i + 3] == 0xB8) { + if (!(Data[i + 7] & 0x40)) // set flag only if GOP is not closed + Data[i + 7] |= 0x20; + return; + } + } + dsyslog("SetBrokenLink: no GOP header found in video packet"); + } + else + dsyslog("SetBrokenLink: no video packet in frame"); +} + +int cTSRemux::GetPid(const uchar *Data) +{ + return (((uint16_t)Data[0] & PID_MASK_HI) << 8) | (Data[1] & 0xFF); +} + +int cTSRemux::GetPacketLength(const uchar *Data, int Count, int Offset) +{ + // Returns the length of the packet starting at Offset, or -1 if Count is + // too small to contain the entire packet. + int Length = (Offset + 5 < Count) ? (Data[Offset + 4] << 8) + Data[Offset + 5] + 6 : -1; + if (Length > 0 && Offset + Length <= Count) + return Length; + return -1; +} + +int cTSRemux::ScanVideoPacket(const uchar *Data, int Count, int Offset, uchar &PictureType) +{ + // Scans the video packet starting at Offset and returns its length. + // If the return value is -1 the packet was not completely in the buffer. + int Length = GetPacketLength(Data, Count, Offset); + if (Length > 0) { + if (Length >= 8) { + int i = Offset + 8; // the minimum length of the video packet header + i += Data[i] + 1; // possible additional header bytes + for (; i < Offset + Length - 5; i++) { + if (Data[i] == 0 && Data[i + 1] == 0 && Data[i + 2] == 1) { + switch (Data[i + 3]) { + case SC_PICTURE: PictureType = (Data[i + 5] >> 3) & 0x07; + return Length; + } + } + } + } + PictureType = NO_PICTURE; + return Length; + } + return -1; +} + diff --git a/remux/tsremux.h b/remux/tsremux.h new file mode 100644 index 0000000..f7e4e09 --- /dev/null +++ b/remux/tsremux.h @@ -0,0 +1,33 @@ +#ifndef VDR_STREAMDEV_TSREMUX_H +#define VDR_STREAMDEV_TSREMUX_H + +#include "libdvbmpeg/transform.h" +#include + +#define RESULTBUFFERSIZE KILOBYTE(256) + +class cTSRemux { +protected: + /*uchar m_ResultBuffer[RESULTBUFFERSIZE]; + int m_ResultCount; + int m_ResultDelivered; + int m_Synced; + int m_Skipped; + int m_Sync; + + + virtual void PutTSPacket(int Pid, const uint8_t *Data) = 0; + +public: + cTSRemux(bool Sync = true); + virtual ~cTSRemux(); + + virtual uchar *Process(const uchar *Data, int &Count, int &Result);*/ + + static void SetBrokenLink(uchar *Data, int Length); + static int GetPid(const uchar *Data); + static int GetPacketLength(const uchar *Data, int Count, int Offset); + static int ScanVideoPacket(const uchar *Data, int Count, int Offset, uchar &PictureType); +}; + +#endif // VDR_STREAMDEV_TSREMUX_H -- cgit v1.2.3