mpegts: begun restructuring, need some more thought

Still haven't dealt with the whole multiple active muxes per input
This commit is contained in:
Adam Sutton 2013-04-27 20:32:39 +01:00
parent 4b73a651cf
commit 024d68416e
5 changed files with 250 additions and 90 deletions

View file

@ -73,7 +73,7 @@ struct mpegts_table
int mt_fd;
LIST_ENTRY(mpegts_table) mt_link;
mpegts_mux_instance_t *mt_mux;
mpegts_mux_t *mt_mux;
char *mt_name;
@ -106,6 +106,7 @@ struct mpegts_table
struct mpegts_table_feed {
TAILQ_ENTRY(mpegts_table_feed) mtf_link;
uint8_t mtf_tsb[188];
mpegts_mux_t *mtf_mux;
};
@ -198,13 +199,15 @@ struct mpegts_mux
int mm_num_tables;
LIST_HEAD(, mpegts_table) mm_tables;
TAILQ_HEAD(, mpegts_table) mm_table_queue;
uint8_t mm_table_filter;
uint8_t mm_table_filter[8192];
/*
* Functions
*/
int (*mm_start) ( mpegts_mux_t *mm, const char *reason, int weight );
int (*mm_start) ( mpegts_mux_t *mm, const char *reason, int weight );
void (*mm_open_table) (mpegts_mux_t*,mpegts_table_t*);
void (*mm_close_table) (mpegts_mux_t*,mpegts_table_t*);
#if 0
dvb_mux_conf_t dm_conf;
@ -326,6 +329,8 @@ struct mpegts_mux_instance
LIST_ENTRY(mpegts_mux_instance) mmi_mux_link;
LIST_ENTRY(mpegts_mux_instance) mmi_active_link;
streaming_pad_t mmi_streaming_pad;
mpegts_mux_t *mmi_mux;
mpegts_input_t *mmi_input;
@ -345,7 +350,7 @@ struct mpegts_input
mpegts_network_t *mi_network; // TODO: this may need altering for DVB-S
mpegts_mux_instance_t *mi_mux_current;
//mpegts_mux_instance_t *mi_mux_current;
/*
* Input processing
@ -358,9 +363,6 @@ struct mpegts_input
int mi_bytes;
// Full mux streaming, protected via the delivery mutex
streaming_pad_t mi_streaming_pad;
struct mpegts_table_feed_queue mi_table_feed;
@ -378,8 +380,6 @@ struct mpegts_input
void (*mi_stop_mux) (mpegts_input_t*);
void (*mi_open_service) (mpegts_input_t*,mpegts_service_t*);
void (*mi_close_service) (mpegts_input_t*,mpegts_service_t*);
void (*mi_open_table) (mpegts_input_t*,mpegts_table_t*);
void (*mi_close_table) (mpegts_input_t*,mpegts_table_t*);
};
#endif /* __TVH_MPEGTS_H__ */
@ -406,9 +406,11 @@ mpegts_mux_instance_t *mpegts_mux_instance_create0
void mpegts_mux_initial_scan_done ( mpegts_mux_t *mm );
size_t mpegts_input_recv_packets
(mpegts_input_t *mi, uint8_t *tsb, size_t len,
(mpegts_input_t *mi, mpegts_mux_instance_t *mmi, uint8_t *tsb, size_t len,
int64_t *pcr, uint16_t *pcr_pid);
void *mpegts_input_table_thread ( void *aux );
void mpegts_table_dispatch
(mpegts_table_t *mt, const uint8_t *sec, int r);
void mpegts_table_release

View file

@ -23,6 +23,7 @@
#include "atomic.h"
#include <pthread.h>
#include <assert.h>
const idclass_t mpegts_input_class =
{
@ -34,11 +35,14 @@ const idclass_t mpegts_input_class =
size_t
mpegts_input_recv_packets
( mpegts_input_t *mi, uint8_t *tsb, size_t l,
int64_t *pcr, uint16_t *pcr_pid )
( mpegts_input_t *mi, mpegts_mux_instance_t *mmi,
uint8_t *tsb, size_t l, int64_t *pcr, uint16_t *pcr_pid )
{
int len = l; // TODO: fix ts_resync() to remove this
int i = 0, table_wakeup = 0;
mpegts_mux_t *mm = mmi->mmi_mux;
assert(mmi->mmi_input == mi);
assert(mm != NULL);
tvhtrace("mpegts", "recv_packets tsb=%p, len=%d, pcr=%p, pcr_pid=%p",
tsb, (int)len, pcr, pcr_pid);
@ -49,13 +53,13 @@ mpegts_input_recv_packets
pthread_mutex_lock(&mi->mi_delivery_mutex);
/* Raw stream */
if (LIST_FIRST(&mi->mi_streaming_pad.sp_targets) != NULL) {
if (LIST_FIRST(&mmi->mmi_streaming_pad.sp_targets) != NULL) {
streaming_message_t sm;
pktbuf_t *pb = pktbuf_alloc(tsb, len);
memset(&sm, 0, sizeof(sm));
sm.sm_type = SMT_MPEGTS;
sm.sm_data = pb;
streaming_pad_deliver(&mi->mi_streaming_pad, &sm);
streaming_pad_deliver(&mmi->mmi_streaming_pad, &sm);
pktbuf_ref_dec(pb);
}
@ -68,11 +72,12 @@ mpegts_input_recv_packets
int pid = ((tsb[i+1] & 0x1f) << 8) | tsb[i+2];
/* SI data */
if (mi->mi_table_filter[pid]) {
if (mm->mm_table_filter[pid]) {
printf("pid = %04X\n", pid);
if (!(tsb[i+1] & 0x80)) {
mpegts_table_feed_t *mtf = malloc(sizeof(mpegts_table_feed_t));
memcpy(mtf->mtf_tsb, tsb+i, 188);
mtf->mtf_mux = mm;
TAILQ_INSERT_TAIL(&mi->mi_table_feed, mtf, mtf_link);
table_wakeup = 1;
}
@ -123,6 +128,55 @@ mpegts_input_recv_packets
return len;
}
static void
mpegts_input_table_dispatch ( mpegts_mux_t *mm, mpegts_table_feed_t *mtf )
{
int i = 0;
int len = mm->mm_num_tables;
uint16_t pid = ((mtf->mtf_tsb[1] & 0x1f) << 8) | mtf->mtf_tsb[2];
mpegts_table_t *mt, *vec[len];
/* Collate - tables may be removed during callbacks */
LIST_FOREACH(mt, &mm->mm_tables, mt_link) {
vec[i++] = mt;
mt->mt_refcount++;
}
assert(i == len);
/* Process */
for (i = 0; i < len; i++) {
mt = vec[i];
if (!mt->mt_destroyed && mt->mt_pid == pid)
psi_section_reassemble(&mt->mt_sect, mtf->mtf_tsb, 0, NULL, mt);
mpegts_table_release(mt);
}
}
void *
mpegts_input_table_thread ( void *aux )
{
mpegts_table_feed_t *mtf;
mpegts_input_t *mi = aux;
while (1) {
/* Wait for data */
pthread_mutex_lock(&mi->mi_delivery_mutex);
while(!(mtf = TAILQ_FIRST(&mi->mi_table_feed)))
pthread_cond_wait(&mi->mi_table_feed_cond, &mi->mi_delivery_mutex);
TAILQ_REMOVE(&mi->mi_table_feed, mtf, mtf_link);
pthread_mutex_unlock(&mi->mi_delivery_mutex);
/* Process */
pthread_mutex_lock(&global_lock);
// TODO: should we check the mux is active
mpegts_input_table_dispatch(mtf->mtf_mux, mtf);
pthread_mutex_unlock(&global_lock);
free(mtf);
}
return NULL;
}
mpegts_input_t*
mpegts_input_create0 ( const char *uuid )
{

View file

@ -31,11 +31,13 @@ const idclass_t mpegts_mux_instance_class =
}
};
#if 0
static int
mpegts_mux_instance_weight ( mpegts_mux_instance_t *mmi )
{
return 0;
}
#endif
mpegts_mux_instance_t *
mpegts_mux_instance_create0
@ -121,8 +123,8 @@ mpegts_mux_start ( mpegts_mux_t *mm, const char *reason, int weight )
/* Find free input */
LIST_FOREACH(mmi, &mm->mm_instances, mmi_mux_link)
if (!mmi->mmi_tune_failed &&
!mmi->mmi_input->mi_mux_current)
if (!mmi->mmi_tune_failed /*TODO&&
!mmi->mmi_input->mi_mux_current*/)
break;
printf("free input = %p\n", mmi);
@ -135,7 +137,7 @@ mpegts_mux_start ( mpegts_mux_t *mm, const char *reason, int weight )
continue;
/* Found */
if (mpegts_mux_instance_weight(mmi->mmi_input->mi_mux_current) < weight)
if (100 < weight)//TODO:mpegts_mux_instance_weight(mmi->mmi_input->mi_mux_current) < weight)
break;
}
@ -160,6 +162,23 @@ mpegts_mux_start ( mpegts_mux_t *mm, const char *reason, int weight )
return 0;
}
static void
mpegts_mux_open_table ( mpegts_mux_t *mm, mpegts_table_t *mt )
{
if (mt->mt_pid >= 0x2000)
return;
mm->mm_table_filter[mt->mt_pid] = 1;
printf("table opened %04X\n", mt->mt_pid);
}
static void
mpegts_mux_close_table ( mpegts_mux_t *mm, mpegts_table_t *mt )
{
if (mt->mt_pid >= 0x2000)
return;
mm->mm_table_filter[mt->mt_pid] = 0;
}
mpegts_mux_t *
mpegts_mux_create0
( const char *uuid, mpegts_network_t *net, uint16_t onid, uint16_t tsid )
@ -175,6 +194,11 @@ mpegts_mux_create0
mm->mm_start = mpegts_mux_start;
mpegts_mux_initial_scan_link(mm);
/* Table processing */
mm->mm_open_table = mpegts_mux_open_table;
mm->mm_close_table = mpegts_mux_close_table;
TAILQ_INIT(&mm->mm_table_queue);
return mm;
}

View file

@ -0,0 +1,145 @@
/*
* MPEGTS table support
* Copyright (C) 2013 Andreas Öman
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "tvheadend.h"
#include "input/mpegts.h"
#include <assert.h>
void
mpegts_table_dispatch
( mpegts_table_t *mt, const uint8_t *sec, int r )
{
int tid, len, ret;
int chkcrc = mt->mt_flags & MT_CRC;
if(mt->mt_destroyed)
return;
/* It seems some hardware (or is it the dvb API?) does not
honour the DMX_CHECK_CRC flag, so we check it again */
if(chkcrc && tvh_crc32(sec, r, 0xffffffff))
return;
/* Table info */
tid = sec[0];
len = ((sec[1] & 0x0f) << 8) | sec[2];
/* Not enough data */
if(len < r - 3)
return;
/* Check table mask */
if((tid & mt->mt_mask) != mt->mt_table)
return;
/* Strip trailing CRC */
if(chkcrc)
len -= 4;
/* Pass with tableid / len in data */
if (mt->mt_flags & MT_FULL)
ret = mt->mt_callback(mt, sec, len, tid);
/* Pass w/out tableid/len in data */
else
ret = mt->mt_callback(mt, sec+3, len-3, tid);
/* Good */
if(ret == 0)
mt->mt_count++;
/* TODO
if(mt->mt_flags & TDT_QUICKREQ)
dvb_table_fastswitch(tdmi);
*/
}
void
mpegts_table_release ( mpegts_table_t *mt )
{
if(--mt->mt_refcount == 0) {
free(mt->mt_name);
free(mt);
}
}
static void
mpegts_table_destroy ( mpegts_table_t *mt )
{
LIST_REMOVE(mt, mt_link);
mt->mt_destroyed = 1;
mt->mt_mux->mm_num_tables--;
mt->mt_mux->mm_close_table(mt->mt_mux, mt);
mpegts_table_release(mt);
}
/**
* Add a new DVB table
*/
void
mpegts_table_add
( mpegts_mux_t *mm, int tableid, int mask,
mpegts_table_callback callback, void *opaque,
const char *name, int flags, int pid )
{
mpegts_table_t *mt;
/* Check for existing */
LIST_FOREACH(mt, &mm->mm_tables, mt_link)
if ( mt->mt_pid == pid &&
mt->mt_callback == callback &&
mt->mt_opaque == opaque )
return;
/* Create */
mt = calloc(1, sizeof(mpegts_table_t));
mt->mt_refcount = 1;
mt->mt_name = strdup(name);
mt->mt_callback = callback;
mt->mt_opaque = opaque;
mt->mt_pid = pid;
mt->mt_flags = flags;
mt->mt_table = tableid;
mt->mt_mask = mask;
mt->mt_mux = mm;
mt->mt_fd = -1;
LIST_INSERT_HEAD(&mm->mm_tables, mt, mt_link);
mm->mm_num_tables++;
/* Open table */
mm->mm_open_table(mm, mt);
}
/**
*
*/
void
mpegts_table_flush_all ( mpegts_mux_t *mm )
{
mpegts_table_t *mt;
while ((mt = LIST_FIRST(&mm->mm_tables)))
mpegts_table_destroy(mt);
}
/******************************************************************************
* Editor Configuration
*
* vim:sts=2:ts=2:sw=2:et
*****************************************************************************/

View file

@ -42,12 +42,13 @@ tsfile_input_thread ( void *aux )
int64_t pcr, pcr_last = PTS_UNSET, pcr_last_realtime = 0;
uint16_t pcr_pid = 0;
mpegts_input_t *mi = aux;
tsfile_mux_instance_t *mmi;
mpegts_mux_instance_t *mmi;
/* Open file */
printf("waiting for lock..\n");
pthread_mutex_lock(&global_lock);
printf("got lock\n");
#if 0
printf("cur mux = %p\n", mi->mi_mux_current);
if (mi->mi_mux_current) {
mmi = (tsfile_mux_instance_t*)mi->mi_mux_current;
@ -57,6 +58,7 @@ printf("cur mux = %p\n", mi->mi_mux_current);
mmi->mmi_tsfile_path, errno, strerror(errno));
}
}
#endif
pthread_mutex_unlock(&global_lock);
if (fd == -1) return NULL;
printf("file opened = %d\n", fd);
@ -153,8 +155,10 @@ tsfile_input_stop_mux ( mpegts_input_t *mi )
tvhtrace("tsfile", "adapter %d stopped thread", mi->mi_instance);
}
#if 0
mi->mi_mux_current->mmi_mux->mm_active = NULL;
mi->mi_mux_current = NULL;
#endif
}
static int
@ -187,7 +191,7 @@ tsfile_input_start_mux ( mpegts_input_t *mi, mpegts_mux_instance_t *t )
}
/* Current */
mi->mi_mux_current = mmi->mmi_mux->mm_active = t;
//mi->mi_mux_current = mmi->mmi_mux->mm_active = t;
/* Install table handlers */
mpegts_table_add(mm, 0x0, 0xff, psi_pat_callback, NULL, "pat",
@ -212,73 +216,6 @@ tsfile_input_close_service ( mpegts_input_t *mi, mpegts_service_t *t )
{
}
static void
tsfile_input_open_table ( mpegts_input_t *mi, mpegts_table_t *mt )
{
if (mt->mt_pid >= 0x2000)
return;
mi->mi_table_filter[mt->mt_pid] = 1;
printf("table opened %04X\n", mt->mt_pid);
}
static void
tsfile_input_close_table ( mpegts_input_t *mi, mpegts_table_t *mt )
{
if (mt->mt_pid >= 0x2000)
return;
mi->mi_table_filter[mt->mt_pid] = 0;
}
static void
tsfile_table_dispatch ( mpegts_mux_t *mm, mpegts_table_feed_t *mtf )
{
int i = 0;
int len = mm->mm_num_tables;
uint16_t pid = ((mtf->mtf_tsb[1] & 0x1f) << 8) | mtf->mtf_tsb[2];
mpegts_table_t *mt, *vec[len];
/* Collate - tables may be removed during callbacks */
LIST_FOREACH(mt, &mm->mm_tables, mt_link) {
vec[i++] = mt;
mt->mt_refcount++;
}
assert(i == len);
/* Process */
for (i = 0; i < len; i++) {
mt = vec[i];
if (!mt->mt_destroyed && mt->mt_pid == pid)
psi_section_reassemble(&mt->mt_sect, mtf->mtf_tsb, 0, NULL, mt);
mpegts_table_release(mt);
}
}
static void *
tsfile_table_thread ( void *aux )
{
mpegts_table_feed_t *mtf;
mpegts_mux_instance_t *mmi;
mpegts_input_t *mi = aux;
while (1) {
/* Wait for data */
pthread_mutex_lock(&mi->mi_delivery_mutex);
while(!(mtf = TAILQ_FIRST(&mi->mi_table_feed)))
pthread_cond_wait(&mi->mi_table_feed_cond, &mi->mi_delivery_mutex);
TAILQ_REMOVE(&mi->mi_table_feed, mtf, mtf_link);
pthread_mutex_unlock(&mi->mi_delivery_mutex);
/* Process */
pthread_mutex_lock(&global_lock);
if ((mmi = mi->mi_mux_current))
tsfile_table_dispatch(mmi->mmi_mux, mtf);
pthread_mutex_unlock(&global_lock);
free(mtf);
}
return NULL;
}
mpegts_input_t *
tsfile_input_create ( void )
{
@ -289,13 +226,11 @@ tsfile_input_create ( void )
mi = mpegts_input_create0(NULL);
mi->mi_start_mux = tsfile_input_start_mux;
mi->mi_stop_mux = tsfile_input_stop_mux;
mi->mi_open_table = tsfile_input_open_table;
mi->mi_close_table = tsfile_input_close_table;
mi->mi_open_service = tsfile_input_open_service;
mi->mi_close_service = tsfile_input_close_service;
/* Start table thread */
pthread_create(&tid, NULL, tsfile_table_thread, mi);
pthread_create(&tid, NULL, mpegts_input_table_thread, mi);
return mi;
}