frodovdr/vdr-plugin-streamdev
1
0
Fork 0
mirror of https://projects.vdr-developer.org/git/vdr-plugin-streamdev.git synced 2023-10-10 17:16:51 +00:00
Code Issues Releases Wiki Activity

- added namespace to remuxers

Browse Source 
- increased WRITERBUFSIZE - buffer was too small for high bandwidth content
- removed cStreamdevStreamer::m_Running
- eliminated potential busy waits in remuxers
- updated cTSRemux static helpers to code of their VDR 1.6.0 counterparts
- use a copy of VDR 1.6.0's cRemux for TS to PES remuxing.
- make sure that only complete TS packets are written to ringbuffers
- use signaling instead of sleeps when writing to ringbuffers
- optimized cStreamdevPatFilter PAT packet initialization
- fixed cStreamdevPatFilter not processing PATs with length > TS_SIZE - 5
- use a small ringbuffer for cStreamdevPatFilter instead of writing to
  cStreamdevStreamers SendBuffer as two threads mustn't write to the same
  ringbuffer
Modified Files:
 Tag: v0_4
	CONTRIBUTORS HISTORY Makefile streamdev-server.c
	libdvbmpeg/transform.h remux/extern.c remux/extern.h
	remux/ts2es.c remux/ts2es.h remux/ts2ps.c remux/ts2ps.h
	remux/tsremux.c remux/tsremux.h server/livestreamer.c
	server/livestreamer.h server/streamer.c server/streamer.h
Added Files:
 Tag: v0_4
	remux/ts2pes.c remux/ts2pes.h
This commit is contained in:
schmirl
2009-06-29 06:25:27 +00:00
parent cacd4b73d5
commit 412c6982b6
19 changed files with 2313 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files

128
server/livestreamer.c
View File

@@ -8,11 +8,12 @@
#include "server/livestreamer.h"
#include "server/livefilter.h"
#include "remux/ts2ps.h"
#include "remux/ts2pes.h"
#include "remux/ts2es.h"
#include "remux/extern.h"
#include "common.h"
#define TSPATREPACKER
using namespace Streamdev;
// --- cStreamdevLiveReceiver -------------------------------------------------
@@ -73,6 +74,8 @@ private:
int pmtPid;
int pmtSid;
int pmtVersion;
uchar tspat_buf[TS_SIZE];
cStreamdevBuffer siBuffer;
const cChannel *m_Channel;
cStreamdevLiveStreamer *m_Streamer;
@@ -82,9 +85,11 @@ private:
int GetPid(SI::PMT::Stream& stream);
public:
cStreamdevPatFilter(cStreamdevLiveStreamer *Streamer, const cChannel *Channel);
uchar* Get(int &Count) { return siBuffer.Get(Count); }
void Del(int Count) { return siBuffer.Del(Count); }
};
cStreamdevPatFilter::cStreamdevPatFilter(cStreamdevLiveStreamer *Streamer, const cChannel *Channel)
cStreamdevPatFilter::cStreamdevPatFilter(cStreamdevLiveStreamer *Streamer, const cChannel *Channel): siBuffer(10 * TS_SIZE, TS_SIZE)
{
Dprintf("cStreamdevPatFilter(\"%s\")\n", Channel->Name());
assert(Streamer);
@@ -94,6 +99,29 @@ cStreamdevPatFilter::cStreamdevPatFilter(cStreamdevLiveStreamer *Streamer, const
pmtSid = 0;
pmtVersion = -1;
Set(0x00, 0x00); // PAT
// initialize PAT buffer. Only some values are dynamic (see comments)
memset(tspat_buf, 0xff, TS_SIZE);
tspat_buf[0] = TS_SYNC_BYTE; // Transport packet header sunchronization byte (1000011 = 0x47h)
tspat_buf[1] = 0x40; // Set payload unit start indicator bit
tspat_buf[2] = 0x0; // PID
tspat_buf[3] = 0x10; // Set payload flag, DYNAMIC: Continuity counter
tspat_buf[4] = 0x0; // SI pointer field
tspat_buf[5] = 0x0; // PAT table id
tspat_buf[6] = 0xb0; // Section syntax indicator bit and reserved bits set
tspat_buf[7] = 12 + 1; // Section length (12 bit): PAT_TABLE_LEN + 1
tspat_buf[8] = 0; // DYNAMIC: Transport stream ID (bits 8-15)
tspat_buf[9] = 0; // DYNAMIC: Transport stream ID (bits 0-7)
tspat_buf[10] = 0xc0; // Reserved, DYNAMIC: Version number, DYNAMIC: Current next indicator
tspat_buf[11] = 0x0; // Section number
tspat_buf[12] = 0x0; // Last section number
tspat_buf[13] = 0; // DYNAMIC: Program number (bits 8-15)
tspat_buf[14] = 0; // DYNAMIC: Program number (bits 0-7)
tspat_buf[15] = 0xe0; // Reserved, DYNAMIC: Network ID (bits 8-12)
tspat_buf[16] = 0; // DYNAMIC: Network ID (bits 0-7)
tspat_buf[17] = 0; // DYNAMIC: Checksum
tspat_buf[18] = 0; // DYNAMIC: Checksum
tspat_buf[19] = 0; // DYNAMIC: Checksum
tspat_buf[20] = 0; // DYNAMIC: Checksum
}
static const char * const psStreamTypes[] = {
@@ -233,54 +261,37 @@ void cStreamdevPatFilter::Process(u_short Pid, u_char Tid, const u_char *Data, i
if (0 != (pmtPid = assoc.getPid())) {
Dprintf("cStreamdevPatFilter: PMT pid for channel %s: %d\n", Channel->Name(), pmtPid);
pmtSid = assoc.getServiceId();
if (Length < TS_SIZE-5) {
// repack PAT to TS frame and send to client
#ifndef TSPATREPACKER
uint8_t pat_ts[TS_SIZE] = {TS_SYNC_BYTE, 0x40 /* pusi=1 */, 0 /* pid=0 */, 0x10 /* adaption=1 */, 0 /* pointer */};
memcpy(pat_ts + 5, Data, Length);
m_Streamer->Put(pat_ts, TS_SIZE);
#else
int ts_id;
unsigned int crc, i, len;
uint8_t *tmp, tspat_buf[TS_SIZE];
static uint8_t ccounter = 0;
ccounter = (ccounter + 1) % 16;
memset(tspat_buf, 0xff, TS_SIZE);
ts_id = Channel->Tid(); // Get transport stream id of the channel
tspat_buf[0] = TS_SYNC_BYTE; // Transport packet header sunchronization byte (1000011 = 0x47h)
tspat_buf[1] = 0x40; // Set payload unit start indicator bit
tspat_buf[2] = 0x0; // PID
tspat_buf[3] = 0x10 | ccounter; // Set payload flag, Continuity counter
tspat_buf[4] = 0x0; // SI pointer field
tspat_buf[5] = 0x0; // PAT table id
tspat_buf[6] = 0xb0; // Section syntax indicator bit and reserved bits set
tspat_buf[7] = 12 + 1; // Section length (12 bit): PAT_TABLE_LEN + 1
tspat_buf[8] = (ts_id >> 8); // Transport stream ID (bits 8-15)
tspat_buf[9] = (ts_id & 0xff); // Transport stream ID (bits 0-7)
tspat_buf[10] = 0xc0 | ((pat.getVersionNumber() << 1) & 0x3e) |
pat.getCurrentNextIndicator();// Version number, Current next indicator
tspat_buf[11] = 0x0; // Section number
tspat_buf[12] = 0x0; // Last section number
tspat_buf[13] = (pmtSid >> 8); // Program number (bits 8-15)
tspat_buf[14] = (pmtSid & 0xff); // Program number (bits 0-7)
tspat_buf[15] = 0xe0 | (pmtPid >> 8); // Network ID (bits 8-12)
tspat_buf[16] = (pmtPid & 0xff); // Network ID (bits 0-7)
crc = 0xffffffff;
len = 12; // PAT_TABLE_LEN
tmp = &tspat_buf[4 + 1]; // TS_HDR_LEN + 1
while (len--) {
crc ^= *tmp++ << 24;
for (i = 0; i < 8; i++)
crc = (crc << 1) ^ ((crc & 0x80000000) ? 0x04c11db7 : 0); // CRC32POLY
}
tspat_buf[17] = crc >> 24 & 0xff; // Checksum
tspat_buf[18] = crc >> 16 & 0xff; // Checksum
tspat_buf[19] = crc >> 8 & 0xff; // Checksum
tspat_buf[20] = crc & 0xff; // Checksum
m_Streamer->Put(tspat_buf, TS_SIZE);
#endif
} else
isyslog("cStreamdevPatFilter: PAT size %d too large to fit in one TS", Length);
// repack PAT to TS frame and send to client
int ts_id;
unsigned int crc, i, len;
uint8_t *tmp;
static uint8_t ccounter = 0;
ccounter = (ccounter + 1) % 16;
ts_id = Channel->Tid(); // Get transport stream id of the channel
tspat_buf[3] = 0x10 | ccounter; // Set payload flag, Continuity counter
tspat_buf[8] = (ts_id >> 8); // Transport stream ID (bits 8-15)
tspat_buf[9] = (ts_id & 0xff); // Transport stream ID (bits 0-7)
tspat_buf[10] = 0xc0 | ((pat.getVersionNumber() << 1) & 0x3e) |
pat.getCurrentNextIndicator();// Version number, Current next indicator
tspat_buf[13] = (pmtSid >> 8); // Program number (bits 8-15)
tspat_buf[14] = (pmtSid & 0xff); // Program number (bits 0-7)
tspat_buf[15] = 0xe0 | (pmtPid >> 8); // Network ID (bits 8-12)
tspat_buf[16] = (pmtPid & 0xff); // Network ID (bits 0-7)
crc = 0xffffffff;
len = 12; // PAT_TABLE_LEN
tmp = &tspat_buf[4 + 1]; // TS_HDR_LEN + 1
while (len--) {
crc ^= *tmp++ << 24;
for (i = 0; i < 8; i++)
crc = (crc << 1) ^ ((crc & 0x80000000) ? 0x04c11db7 : 0); // CRC32POLY
}
tspat_buf[17] = crc >> 24 & 0xff; // Checksum
tspat_buf[18] = crc >> 16 & 0xff; // Checksum
tspat_buf[19] = crc >> 8 & 0xff; // Checksum
tspat_buf[20] = crc & 0xff; // Checksum
int written = siBuffer.PutTS(tspat_buf, TS_SIZE);
if (written != TS_SIZE)
siBuffer.ReportOverflow(TS_SIZE - written);
if (pmtPid != prevPmtPid) {
m_Streamer->SetPids(pmtPid);
Add(pmtPid, 0x02);
@@ -301,7 +312,7 @@ void cStreamdevPatFilter::Process(u_short Pid, u_char Tid, const u_char *Data, i
if (pmtVersion != -1) {
if (pmtVersion != pmt.getVersionNumber()) {
Dprintf("cStreamdevPatFilter: PMT version changed, detaching all pids\n");
Del(pmtPid, 0x02);
cFilter::Del(pmtPid, 0x02);
pmtPid = 0; // this triggers PAT scan
}
return;
@@ -473,8 +484,8 @@ bool cStreamdevLiveStreamer::SetChannel(const cChannel *Channel, eStreamType Str
}
case stPES:
m_PESRemux = new cRemux(m_Channel->Vpid(), m_Channel->Apids(), m_Channel->Dpids(),
m_Channel->Spids(), false);
m_PESRemux = new cTS2PESRemux(m_Channel->Vpid(), m_Channel->Apids(), m_Channel->Dpids(),
m_Channel->Spids());
return SetPids(m_Channel->Vpid(), Apids, Dpids, m_Channel->Spids());
case stPS:
@@ -512,6 +523,17 @@ int cStreamdevLiveStreamer::Put(const uchar *Data, int Count)
{
switch (m_StreamType) {
case stTS:
// insert si data
if (m_PatFilter) {
int got;
uchar *si = m_PatFilter->Get(got);
if (si) {
int count = cStreamdevStreamer::Put(si, got);
if (count)
m_PatFilter->Del(count);
}
}
// fall through
case stTSPIDS:
return cStreamdevStreamer::Put(Data, Count);

18
server/livestreamer.h
View File

@@ -7,10 +7,12 @@
#include "server/streamer.h"
#include "common.h"
class cTS2PSRemux;
class cTS2ESRemux;
class cExternRemux;
class cRemux;
namespace Streamdev {
class cTS2PSRemux;
class cTS2ESRemux;
class cExternRemux;
class cTS2PESRemux;
}
class cStreamdevPatFilter;
class cStreamdevLiveReceiver;
@@ -27,10 +29,10 @@ private:
cDevice *m_Device;
cStreamdevLiveReceiver *m_Receiver;
cStreamdevPatFilter *m_PatFilter;
cRemux *m_PESRemux;
cTS2ESRemux *m_ESRemux;
cTS2PSRemux *m_PSRemux;
cExternRemux *m_ExtRemux;
Streamdev::cTS2PESRemux *m_PESRemux;
Streamdev::cTS2ESRemux *m_ESRemux;
Streamdev::cTS2PSRemux *m_PSRemux;
Streamdev::cExternRemux *m_ExtRemux;
void StartReceiver(void);
bool HasPid(int Pid);

22
server/streamer.c
View File

@@ -1,5 +1,5 @@
/*
* $Id: streamer.c,v 1.16.2.2 2009/02/13 10:39:42 schmirl Exp $
* $Id: streamer.c,v 1.16.2.3 2009/06/29 06:25:30 schmirl Exp $
*/
#include <vdr/ringbuffer.h>
@@ -14,6 +14,13 @@
#include "tools/select.h"
#include "common.h"
// --- cStreamdevBuffer -------------------------------------------------------
cStreamdevBuffer::cStreamdevBuffer(int Size, int Margin, bool Statistics, const char *Description):
cRingBufferLinear(Size, Margin, Statistics, Description)
{
}
// --- cStreamdevWriter -------------------------------------------------------
cStreamdevWriter::cStreamdevWriter(cTBSocket *Socket,
@@ -95,14 +102,13 @@ void cStreamdevWriter::Action(void)
cStreamdevStreamer::cStreamdevStreamer(const char *Name):
cThread(Name),
m_Running(false),
m_Writer(NULL),
m_RingBuffer(new cRingBufferLinear(STREAMERBUFSIZE, TS_SIZE * 2,
m_RingBuffer(new cStreamdevBuffer(STREAMERBUFSIZE, TS_SIZE * 2,
true, "streamdev-streamer")),
m_SendBuffer(new cRingBufferLinear(WRITERBUFSIZE, TS_SIZE * 2))
m_SendBuffer(new cStreamdevBuffer(WRITERBUFSIZE, TS_SIZE * 2))
{
m_RingBuffer->SetTimeouts(0, 100);
m_SendBuffer->SetTimeouts(0, 100);
m_SendBuffer->SetTimeouts(100, 100);
}
cStreamdevStreamer::~cStreamdevStreamer()
@@ -116,7 +122,6 @@ void cStreamdevStreamer::Start(cTBSocket *Socket)
{
Dprintf("start streamer\n");
m_Writer = new cStreamdevWriter(Socket, this);
m_Running = true;
Attach();
}
@@ -135,9 +140,8 @@ void cStreamdevStreamer::Stop(void)
Dprintf("stopping streamer\n");
Cancel(3);
}
if (m_Running) {
if (m_Writer) {
Detach();
m_Running = false;
DELETENULL(m_Writer);
}
}
@@ -152,8 +156,6 @@ void cStreamdevStreamer::Action(void)
int count = Put(block, got);
if (count)
m_RingBuffer->Del(count);
else
cCondWait::SleepMs(100);
}
}
}

43
server/streamer.h
View File

@@ -1,5 +1,5 @@
/*
* $Id: streamer.h,v 1.8.2.2 2009/02/13 10:39:42 schmirl Exp $
* $Id: streamer.h,v 1.8.2.3 2009/06/29 06:25:30 schmirl Exp $
*/
#ifndef VDR_STREAMDEV_STREAMER_H
@@ -16,8 +16,34 @@ class cStreamdevStreamer;
#define TS_SIZE 188
#endif
#define STREAMERBUFSIZE MEGABYTE(4)
#define WRITERBUFSIZE KILOBYTE(256)
#define STREAMERBUFSIZE (20000 * TS_SIZE)
#define WRITERBUFSIZE (5000 * TS_SIZE)
// --- cStreamdevBuffer -------------------------------------------------------
class cStreamdevBuffer: public cRingBufferLinear {
public:
// make public
void WaitForPut(void) { cRingBuffer::WaitForPut(); }
// Always write complete TS packets
// (assumes Count is a multiple of TS_SIZE)
int PutTS(const uchar *Data, int Count);
cStreamdevBuffer(int Size, int Margin = 0, bool Statistics = false, const char *Description = NULL);
};
inline int cStreamdevBuffer::PutTS(const uchar *Data, int Count)
{
int free = Free();
if (free < Count)
Count = free;
Count -= Count % TS_SIZE;
if (Count)
Count = Put(Data, Count);
else
WaitForPut();
return Count;
}
// --- cStreamdevWriter -------------------------------------------------------
@@ -38,15 +64,14 @@ public:
class cStreamdevStreamer: public cThread {
private:
bool m_Running;
cStreamdevWriter *m_Writer;
cRingBufferLinear *m_RingBuffer;
cRingBufferLinear *m_SendBuffer;
cStreamdevBuffer *m_RingBuffer;
cStreamdevBuffer *m_SendBuffer;
protected:
virtual void Action(void);
bool IsRunning(void) const { return m_Running; }
bool IsRunning(void) const { return m_Writer; }
public:
cStreamdevStreamer(const char *Name);
@@ -57,10 +82,10 @@ public:
bool Abort(void);
void Activate(bool On);
int Receive(uchar *Data, int Length) { return m_RingBuffer->Put(Data, Length); }
int Receive(uchar *Data, int Length) { return m_RingBuffer->PutTS(Data, Length); }
void ReportOverflow(int Bytes) { m_RingBuffer->ReportOverflow(Bytes); }
virtual int Put(const uchar *Data, int Count) { return m_SendBuffer->Put(Data, Count); }
virtual int Put(const uchar *Data, int Count) { return m_SendBuffer->PutTS(Data, Count); }
virtual uchar *Get(int &Count) { return m_SendBuffer->Get(Count); }
virtual void Del(int Count) { m_SendBuffer->Del(Count); }