input.c

mpeg 解复用

/*
 * ISO 13818 stream multiplexer
 * Copyright (C) 2001 Convergence Integrated Media GmbH Berlin
 * Author: Oskar Schirmer ([e-mail])
 *
 * 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 2 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, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 */

/*
 * Module:  Input
 * Purpose: Data Acquisition from the various open files,
 *          promotion to the single split-functions.
 *
 * This module holds two main data structures, an unsorted list of
 * open input files, and an unsorted list of open input streams
 * (extracted from these files).
 * Further provided are functions to open and close files and streams,
 * to detect states of the input buffers, and to read data into the
 * (raw) file input buffers.
 */

#include "global.h"
#include "error.h"
#include "pes.h"
#include "splitpes.h"
#include "splitps.h"
#include "splitts.h"
#include "splice.h"
#include "input.h"
#include "descref.h"
#include "ts.h"

/* index of files in use, containing i.a. the raw input data buffers:
 */
static file_descr *inf [MAX_INFILE];
static int in_files;
static int in_openfiles[number_ct];

/* index of streams in use, containing i.a. the input pes buffers:
 */
static stream_descr *ins [MAX_INSTREAM];
static int in_streams;
static int in_openstreams[number_sd];

static t_msec trigger_msec_input;

boolean input_init (void)
{
  in_files = 0;
  memset (in_openfiles, 0, sizeof (in_openfiles));
  in_streams = 0;
  memset (in_openstreams, 0, sizeof (in_openstreams));
  trigger_msec_input = TRIGGER_MSEC_INPUT;
  return (TRUE);
}

#ifdef DEBUG_TIMEPOLL
int deb_inraw_free (int f)
{
  register int r;
  r = (f < in_files) ? list_free (inf[f]->data) : 0;
  return (r);
}

int deb_instr_free (int s)
{
  register int r;
  r = (s < in_streams) ? list_free (ins[s]->data) : 0;
  return (r);
}
#endif

/* Determine whether data is expected as input.
 * Return: TRUE, if any valuable file is open, FALSE otherwise
 */
boolean input_expected (void)
{
  return ((in_files > 0)
       && ((in_openfiles[ct_transport] != in_files)
        || (in_openstreams[sd_unparsedsi] != in_streams)));
}

/* Set trigger timing value.
 */
void input_settriggertiming (t_msec time)
{
  trigger_msec_input = time;
}

/* Determine whether input data is acceptable, i.e. there is space in buffers.
 * If so, set the poll struct accordingly for each file in question.
 * Check the streams for time stamps and set the timeout^ accordingly,
 * if the streams might be responsible alone for blocking pipes.
 * Return: TRUE, if at least one buffer has enough space to accept new data.
 *         FALSE otherwise.
 */
boolean input_acceptable (unsigned int *nfds,
    struct pollfd *ufds,
    t_msec *timeout,
    boolean outnotfull)
{
  boolean accept = FALSE;
  int i;
  t_msec t, now;
  file_descr *f;
  stream_descr *s;
  i = in_files;
  while (--i >= 0) {
    f = inf[i];
    warn (LDEB,"Acceptable",EINP,2,1,i);
    warn (LDEB,"Free Raw",EINP,2,2,list_free (f->data));
    if ((list_free (f->data) >= HIGHWATER_RAW)
     && (f->handle >= 0)) {
      ufds->fd = f->handle;
      ufds->events = POLLIN;
      *nfds += 1;
      accept = TRUE;
      f->ufds = ufds++;
    } else {
      f->ufds = NULL;
    }
  }
  if (outnotfull) {
    now = msec_now ();
    i = in_streams;
    while (--i >= 0) {
      s = ins[i];
      if (s->streamdata == sd_data) {
        if (!list_empty (s->ctrl)) {
          if (s->u.d.trigger) {
            t = s->ctrl.ptr[s->ctrl.out].msecpush - now + s->u.d.delta;
          } else {
            t = s->ctrl.ptr[s->ctrl.out].msecread - now + trigger_msec_input;
          }
          if ((t > 0)
           && ((*timeout < 0)
            || (*timeout > t))) {
            *timeout = t;
          }
        }
      }
    }
  }
  return (accept);
}

/* Set the trigger on a stream, enabling the data to be spliced now.
 * Set the trigger for all streams that correspond thru the target program, too
 * Precondition: s!=NULL
 */
static void set_trigger (stream_descr *s,
    t_msec now)
{
  int q, i;
  prog_descr *p;
  if (!list_empty (s->data)) {
    s->u.d.lasttime = now;
    s->u.d.delta =
      now - s->ctrl.ptr[s->ctrl.out].msecpush;
    warn (LDEB,"Set Trigger",EINP,8,s->u.d.pid,s->u.d.delta);
    s->u.d.trigger = TRUE;
    s->u.d.mention = TRUE;
    q = s->u.d.progs;
    while (--q >= 0) {
      p = s->u.d.pdescr[q];
      p->unchanged = TRUE;
      i = p->streams;
      while (--i >= 0) {
        if (!p->stream[i]->u.d.trigger) {
          set_trigger (p->stream[i],now);
        }
      }
    }
  }
}

/* Clear the trigger on a stream, clear it for all corresponding streams, too
 * Precondition: s!=NULL
 */
static void clear_trigger (stream_descr *s)
{
  int q, i;
  prog_descr *p;
  warn (LDEB,"Clear Trigger",EINP,13,s->u.d.pid,s->u.d.delta);
  s->u.d.discontinuity = TRUE;
  s->u.d.trigger = FALSE;
  q = s->u.d.progs;
  while (--q >= 0) {
    p = s->u.d.pdescr[q];
    i = p->streams;
    while (--i >= 0) {
      if (p->stream[i]->u.d.trigger) {
        clear_trigger (p->stream[i]);
      }
    }
  }
}

/* Check if mapstream provides prominent data.
 * Precondition: d!=NULL, !list_empty(d->ctrl)
 * Return: TRUE, if mapstream has prominent data, FALSE otherwise
 */
static boolean preceding_sequence (stream_descr *d,
    stream_descr *m)
{
  if (m != NULL) {
    if (!list_empty (m->ctrl)) {
      if (m->ctrl.ptr[m->ctrl.out].sequence
        - d->ctrl.ptr[d->ctrl.out].sequence <= 0) {
        return (TRUE);
      }
    }
  }
  return (FALSE);
}

/* Check for every stream whether data is available to be spliced.
 * Unparsed SI from an otherwise unused TS has priority.
 * If the stream with the lowest time stamp has a corresponding map stream,
 * that provides data to be spliced first, the map stream is returned.
 * Prior, check if a stream is empty and end it, if necessary; check if a
 * stream is ready but not yet triggered, so trigger it.
 * Return: stream to be spliced next.
 */
stream_descr *input_available (void)
{
  int i, s, q;
  t_msec t, u, now;
  stream_descr *d, *e;
  ctrl_buffer *c;
  file_descr *f;
  now = msec_now ();
  i = in_files;
  while (--i >= 0) {
    f = inf[i];
    if (f->content == ct_transport) {
      d = ts_file_stream (f,TS_UNPARSED_SI);
      if (d != NULL) {
        if (((f->openstreams[sd_data] == 0)
          && (!list_empty (d->ctrl)))
         || (list_full (d->ctrl))) {
          return (d);
        }
      }
    }
  }
  i = in_streams;
  while (--i >= 0) {
    d = ins[i];
    if (d->streamdata == sd_data) {
      if (list_empty (d->ctrl)) {
        switch (d->endaction) {
          case ENDSTR_CLOSE:
            input_endstream (d);
            if (i > in_streams) {
              i = in_streams;
            }
            break;
          case ENDSTR_KILL:
            input_endstreamkill (d);
            if (i > in_streams) {
              i = in_streams;
            }
            break;
          case ENDSTR_WAIT:
            break;
          default:
            warn (LERR,"End Action",EINP,3,1,d->endaction);
            break;
        }
        /* trigger:=false if empty? no ! */
      } else {
        if (!d->u.d.trigger) {
          if (list_full (d->ctrl)
           || list_partialfull (d->data)
        /* || (list_free (d->fdescr->data) < HIGHWATER_IN) */
           || (d->endaction == ENDSTR_CLOSE)
           || (d->endaction == ENDSTR_KILL)
           || ((now - d->ctrl.ptr[d->ctrl.out].msecread)
                 >= trigger_msec_input)) {
            set_trigger (d,now);
          }
        }
      }
    }
  }
  d = NULL;
  i = in_streams;
  while (--i >= 0) {
    e = ins[i];
    if ((e->streamdata == sd_data)
     && (e->u.d.trigger)) {
      if (!list_empty (e->ctrl)) {
        warn (LDEB,"Available",EINP,3,2,i);
        c = &(e->ctrl.ptr[e->ctrl.out]);
        t = c->msecpush + e->u.d.delta;
        if (t - e->u.d.lasttime < 0) {
          warn (LWAR,"Time Decrease",EINP,3,3,t - e->u.d.lasttime);
          clear_trigger (e);
        } else {
          e->u.d.lasttime = t; 
          t -= now;
          if ((t > MAX_MSEC_PUSHJTTR)
           || (t < -MAX_MSEC_PUSHJTTR)) {
            warn (LWAR,"Time Jumpness",EINP,3,4,t);
            clear_trigger (e);
          } else {
            q = c->sequence;
            if ((t <= 0)
             && ((d == NULL)
              || (t < u)
              || ((t == u) && (q - s < 0)))) {
              u = t;
              s = q;
              d = e;
            }
          }
        }
      }
    }
  }
  if (d != NULL) {
    switch (d->fdescr->content) {
      case ct_transport:
        e = d;
        if (preceding_sequence (d, ts_file_stream (d->fdescr,0))) {
          d = ts_file_stream (d->fdescr,0);
        } else {
          if (preceding_sequence (d, d->u.d.mapstream)) {
            d = d->u.d.mapstream;
          } else {
            if (preceding_sequence (d,
                  ts_file_stream (d->fdescr,TS_UNPARSED_SI))) {
              d = ts_file_stream (d->fdescr,TS_UNPARSED_SI);
            }
          }
        }
        break;
      case ct_program:
        if (preceding_sequence (d,d->u.d.mapstream)) {
          d = d->u.d.mapstream;
        }
        break;
      default:
        break;
    }
  }
  return (d);
}

/* Check all files for a given filerefnum.
 * Precondition: filerefnum>=0
 * Return: filename, if filerefnum matches, NULL otherwise
 */
char *input_filerefername (int filerefnum)
{
  int i;
  file_descr *f;
  i = in_files;
  while (--i >= 0) {
    f = inf[i];
    if (f->filerefnum == filerefnum) {
      return (f->name);
    }
    if ((f->append_name != NULL)
     && (f->append_filerefnum == filerefnum)) {
      return (f->append_name);
    }
  }
  return (NULL);
}

/* Open a file. Allocate and initialize it.
 * Precondition: name!=NULL
 * Return: file, if successful, NULL otherwise
 */
file_descr* input_openfile (char *name,
    int filerefnum,
    content_type content,
    boolean automatic,
    int programnb)
{
  file_descr *f;
  struct stat stat;
  warn (LIMP,"Create file",EINP,4,automatic,content);
  warn (LIMP,name,EINP,4,4,programnb);
  if (in_files < MAX_INFILE) {
    switch (content) {
      case ct_packetized:
        f = unionalloc (file_descr,pes);
        break;
      case ct_program:
        f = unionalloc (file_descr,ps);
        break;
      case ct_transport:
        f = unionalloc (file_descr,ts);
        break;
      default:
        warn (LERR,"Unknown contents",EINP,4,7,0);
        f = NULL;
        break;
    }
    if (f != NULL) {
      if ((f->name = malloc (strlen(name) + 1)) != NULL) {
        if (list_create (f->data,MAX_DATA_RAWB)) {
          if ((f->handle = open (name,O_RDONLY)) >= 0) {
            if (fstat (f->handle,&stat) == 0) {
              f->st_mode = stat.st_mode;
              if (!S_ISREG (f->st_mode)) {
                timed_io = TRUE;
              }
              strcpy (f->name,name);
              f->filerefnum = filerefnum;
              f->skipped = 0;
              f->payload = 0;
              f->total = 0;
              f->sequence = 0;
              memset (f->openstreams,0,sizeof(f->openstreams));
              f->append_name = NULL;
              f->repeatitions = 0;
              f->auto_programnb = programnb;
              f->automatic = automatic;
              f->stopfile = FALSE;
              f->content = content;
              switch (content) {
                case ct_packetized:
                  f->u.pes.stream = NULL;
                  in_openfiles[content] += 1;
                  inf[in_files++] = f;
                  return (f);
                  break;
                case ct_program:
                  memset (f->u.ps.stream,0,sizeof(f->u.ps.stream));
                  f->u.ps.stream[0] = input_openstream (f,0,0,0,sd_map,NULL);
                  if (f->u.ps.stream[0] != NULL) {
                    in_openfiles[content] += 1;
                    inf[in_files++] = f;
                    return (f);
                  }
                  break;
                case ct_transport:
                  f->u.ts.pat_version = 0xFF;
                  f->u.ts.newpat_version = 0xFF;
                  f->u.ts.pat = NULL;
                  f->u.ts.newpat = NULL;
                  f->u.ts.tsauto = NULL;
                  f->u.ts.tssi = NULL;
                  memset (f->u.ts.stream,0,sizeof(f->u.ts.stream));
                  ts_file_stream (f,0) = input_openstream (f,0,0,0,sd_map,NULL);
                  if (ts_file_stream (f,0) != NULL) {
                    in_openfiles[content] += 1;
                    inf[in_files++] = f;
                    return (f);
                  }
                  break;
                default:
                  break;
              }
            } else {
              warn (LERR,"FStat fail",EINP,4,6,0);
            }
            close (f->handle);
          } else {
            warn (LERR,"Open fail",EINP,4,5,f->handle);
          }
          list_release (f->data);
        }
        free (f->name);
      } else {
        warn (LERR,"Alloc fail",EINP,4,8,in_files);
      }
      free (f);
    } else {
      warn (LERR,"Alloc fail",EINP,4,2,in_files);
    }
  } else {
    warn (LERR,"Max file open",EINP,4,3,in_files);
  }
  return (NULL);
}

/* Check if a file with a given name is yet open.
 * The file name comparision is purely textual.
 * Precondition: name!=NULL
 * Return: file if found, NULL otherwise.
 */
file_descr* input_existfile (char *name)
{
  int i;
  i = in_files;
  while (--i >= 0) {
    if (!strcmp (name,inf[i]->name)) {
      return (inf[i]);
    }
  }
  return (NULL);
}

/* Mark all streams in a file to end soon, close the file itself
 * Precondition: f!=NULL
 */
static void input_endfile (file_descr *f)
{
  int i;
  stream_descr *s;
  i = in_streams;
  while (--i >= 0) {
    s = ins[i];
    if (s->fdescr == f) {
      s->endaction = ENDSTR_CLOSE;
    }
  }
  if (f->handle >= 0) {
    close (f->handle);
    f->handle = -1;
  }
  input_closefileifunused (f);
}

/* Close a file and release all corresponding data structures
 * Precondition: f!=NULL, f->u.*.stream[*]==NULL