sectionfilter.c

/*
 * sectionfilter.c: IPTV plugin for the Video Disk Recorder
 *
 * See the README file for copyright information and how to reach the author.
 *
 */

#include "config.h"
#include "log.h"
#include "sectionfilter.h"

cIptvSectionFilter::cIptvSectionFilter(int deviceIndexP, uint16_t pidP, uint8_t tidP, uint8_t maskP)
: pusiSeenM(0),
  feedCcM(0),
  doneqM(0),
  secBufM(NULL),
  secBufpM(0),
  secLenM(0),
  tsFeedpM(0),
  pidM(pidP),
  ringBufferM(new cRingBufferFrame(eDmxMaxSectionCount * eDmxMaxSectionSize)),
  deviceIndexM(deviceIndexP)
{
  debug16("%s (%d, %d)", __PRETTY_FUNCTION__, deviceIndexM, pidM);
  int i;

  memset(secBufBaseM,     0, sizeof(secBufBaseM));
  memset(filterValueM,    0, sizeof(filterValueM));
  memset(filterMaskM,     0, sizeof(filterMaskM));
  memset(filterModeM,     0, sizeof(filterModeM));
  memset(maskAndModeM,    0, sizeof(maskAndModeM));
  memset(maskAndNotModeM, 0, sizeof(maskAndNotModeM));

  filterValueM[0] = tidP;
  filterMaskM[0] = maskP;

  // Invert the filter
  for (i = 0; i < eDmxMaxFilterSize; ++i)
      filterValueM[i] ^= 0xFF;

  uint8_t mask, mode, doneq = 0;
  for (i = 0; i < eDmxMaxFilterSize; ++i) {
      mode = filterModeM[i];
      mask = filterMaskM[i];
      maskAndModeM[i] = (uint8_t)(mask & mode);
      maskAndNotModeM[i] = (uint8_t)(mask & ~mode);
      doneq |= maskAndNotModeM[i];
      }
  doneqM = doneq ? 1 : 0;

  // Create sockets
  socketM[0] = socketM[1] = -1;
  if (socketpair(AF_UNIX, SOCK_SEQPACKET, 0, socketM) != 0) {
     char tmp[64];
     error("Opening section filter sockets failed (device=%d pid=%d): %s", deviceIndexM, pidM, strerror_r(errno, tmp, sizeof(tmp)));
     }
  else if ((fcntl(socketM[0], F_SETFL, O_NONBLOCK) != 0) || (fcntl(socketM[1], F_SETFL, O_NONBLOCK) != 0)) {
     char tmp[64];
     error("Setting section filter socket to non-blocking mode failed (device=%d pid=%d): %s", deviceIndexM, pidM, strerror_r(errno, tmp, sizeof(tmp)));
     }
}

cIptvSectionFilter::~cIptvSectionFilter()
{
  debug16("%s (%d, %d)", __PRETTY_FUNCTION__, deviceIndexM, pidM);
  int tmp = socketM[1];
  socketM[1] = -1;
  if (tmp >= 0)
     close(tmp);
  tmp = socketM[0];
  socketM[0] = -1;
  if (tmp >= 0)
     close(tmp);
  secBufM = NULL;
  DELETENULL(ringBufferM);
}

inline uint16_t cIptvSectionFilter::GetLength(const uint8_t *dataP)
{
  return (uint16_t)(3 + ((dataP[1] & 0x0f) << 8) + dataP[2]);
}

void cIptvSectionFilter::New(void)
{
  tsFeedpM = secBufpM = secLenM = 0;
  secBufM = secBufBaseM;
}

int cIptvSectionFilter::Filter(void)
{
  if (secBufM) {
     int i;
     uint8_t neq = 0;

     for (i = 0; i < eDmxMaxFilterSize; ++i) {
         uint8_t calcxor = (uint8_t)(filterValueM[i] ^ secBufM[i]);
         if (maskAndModeM[i] & calcxor)
            return 0;
         neq |= (maskAndNotModeM[i] & calcxor);
         }

     if (doneqM && !neq)
        return 0;

     if (ringBufferM && (secLenM > 0))
        ringBufferM->Put(new cFrame(secBufM, secLenM));
     }
  return 0;
}

inline int cIptvSectionFilter::Feed(void)
{
  if (Filter() < 0)
     return -1;
  secLenM = 0;
  return 0;
}

int cIptvSectionFilter::CopyDump(const uint8_t *bufP, uint8_t lenP)
{
  uint16_t limit, seclen, n;

  if (tsFeedpM >= eDmxMaxSectionFeedSize)
     return 0;

  if (tsFeedpM + lenP > eDmxMaxSectionFeedSize)
     lenP = (uint8_t)(eDmxMaxSectionFeedSize - tsFeedpM);

  if (lenP <= 0)
     return 0;

  memcpy(secBufBaseM + tsFeedpM, bufP, lenP);
  tsFeedpM = uint16_t(tsFeedpM + lenP);

  limit = tsFeedpM;
  if (limit > eDmxMaxSectionFeedSize)
     return -1; // internal error should never happen

  // Always set secbuf
  secBufM = secBufBaseM + secBufpM;

  for (n = 0; secBufpM + 2 < limit; ++n) {
      seclen = GetLength(secBufM);
      if ((seclen <= 0) || (seclen > eDmxMaxSectionSize) || ((seclen + secBufpM) > limit))
         return 0;
      secLenM = seclen;
      if (pusiSeenM)
         Feed();
      secBufpM = uint16_t(secBufpM + seclen);
      secBufM += seclen;
      }
  return 0;
}

void cIptvSectionFilter::Process(const uint8_t* dataP)
{
  if (dataP[0] != TS_SYNC_BYTE)
     return;

  // Stop if not the PID this filter is looking for
  if (ts_pid(dataP) != pidM)
     return;

  uint8_t count = payload(dataP);

  // Check if no payload or out of range
  if (count == 0)
     return;

  // Payload start
  uint8_t p = (uint8_t)(TS_SIZE - count);

  uint8_t cc = (uint8_t)(dataP[3] & 0x0f);
  int ccok = ((feedCcM + 1) & 0x0f) == cc;
  feedCcM = cc;

  int dc_i = 0;
  if (dataP[3] & 0x20) {
     // Adaption field present, check for discontinuity_indicator
     if ((dataP[4] > 0) && (dataP[5] & 0x80))
        dc_i = 1;
     }

  if (!ccok || dc_i) {
     // Discontinuity detected. Reset pusiSeenM = 0 to
     // stop feeding of suspicious data until next PUSI=1 arrives
     pusiSeenM = 0;
     New();
     }

  if (dataP[1] & 0x40) {
     // PUSI=1 (is set), section boundary is here
     if (count > 1 && dataP[p] < count) {
        const uint8_t *before = &dataP[p + 1];
        uint8_t before_len = dataP[p];
        const uint8_t *after = &before[before_len];
        uint8_t after_len = (uint8_t)(count - 1 - before_len);
        CopyDump(before, before_len);

        // Before start of new section, set pusiSeenM = 1
        pusiSeenM = 1;
        New();
        CopyDump(after, after_len);
        }
     }
  else {
     // PUSI=0 (is not set), no section boundary
     CopyDump(&dataP[p], count);
     }
}

bool cIptvSectionFilter::Send(void)
{
  bool result = false;
  cFrame *section = ringBufferM->Get();
  if (section) {
     uchar *data = section->Data();
     int count = section->Count();
     if (data && (count > 0) && (socketM[1] >= 0) && (socketM[0] >= 0)) {
        ssize_t len = send(socketM[1], data, count, MSG_EOR);
        ERROR_IF(len < 0 && errno != EAGAIN, "send()");
        if (len > 0) {
           ringBufferM->Drop(section);
           result = !!ringBufferM->Available();
           // Update statistics
           AddSectionStatistic(len, 1);
           }
        }
     }
  return result;
}


cIptvSectionFilterHandler::cIptvSectionFilterHandler(int deviceIndexP, unsigned int bufferLenP)
: cThread("IPTV section handler"),
  ringBufferM(new cRingBufferLinear(bufferLenP, TS_SIZE, false, *cString::sprintf("IPTV SECTION HANDLER %d", deviceIndexP))),
  mutexM(),
  deviceIndexM(deviceIndexP)
{
  debug1("%s (%d, %d) [device %d]", __PRETTY_FUNCTION__, deviceIndexP, bufferLenP, deviceIndexM);

  // Initialize filter pointers
  memset(filtersM, 0, sizeof(filtersM));

  // Create input buffer
  if (ringBufferM) {
     ringBufferM->SetTimeouts(100, 100);
     ringBufferM->SetIoThrottle();
     }
  else
     error("Failed to allocate buffer for section filter handler (device=%d): ", deviceIndexM);

  Start();
}

cIptvSectionFilterHandler::~cIptvSectionFilterHandler()
{
  debug1("%s [device %d]", __PRETTY_FUNCTION__, deviceIndexM);
  // Stop thread
  if (Running())
     Cancel(3);
  DELETE_POINTER(ringBufferM);

  // Destroy all filters
  cMutexLock MutexLock(&mutexM);
  for (int i = 0; i < eMaxSecFilterCount; ++i)
      Delete(i);
}

void cIptvSectionFilterHandler::Action(void)
{
  debug1("%s Entering [device %d]", __PRETTY_FUNCTION__, deviceIndexM);
  bool processed = false;
  // Do the thread loop
  while (Running()) {
        // Send demuxed section packets through all filters
        bool retry = false;
        mutexM.Lock();
        for (unsigned int i = 0; i < eMaxSecFilterCount; ++i) {
            if (filtersM[i] && filtersM[i]->Send())
               retry = true;
            }
        mutexM.Unlock();
        if (retry)
           continue;
        // Read one TS packet
        if (ringBufferM) {
           int len = 0;
           if (processed) {
              ringBufferM->Del(TS_SIZE);
              processed = false;
              }
           uchar *p = ringBufferM->Get(len);
           if (p && (len >= TS_SIZE)) {
              if (*p != TS_SYNC_BYTE) {
                 for (int i = 1; i < len; ++i) {
                     if (p[i] == TS_SYNC_BYTE) {
                        len = i;
                        break;
                        }
                     }
                 ringBufferM->Del(len);
                 debug1("%s Skipped %d bytes to sync on TS packet [device %d]]", __PRETTY_FUNCTION__, len, deviceIndexM);
                 continue;
                 }
              // Process TS packet through all filters
              mutexM.Lock();
              for (unsigned int i = 0; i < eMaxSecFilterCount; ++i) {
                  if (filtersM[i])
                     filtersM[i]->Process(p);
                  }
              mutexM.Unlock();
              processed = true;
              continue;
              }
           }
        cCondWait::SleepMs(10); // to avoid busy loop and reduce cpu load
        }
  debug1("%s Exiting [device %d]", __PRETTY_FUNCTION__, deviceIndexM);
}

cString cIptvSectionFilterHandler::GetInformation(void)
{
  debug16("%s [device %d]", __PRETTY_FUNCTION__, deviceIndexM);
  // loop through active section filters
  cMutexLock MutexLock(&mutexM);
  cString s = "";
  unsigned int count = 0;
  for (unsigned int i = 0; i < eMaxSecFilterCount; ++i) {
      if (filtersM[i]) {
         s = cString::sprintf("%sFilter %d: %s Pid=0x%02X (%s)\n", *s, i,
                              *filtersM[i]->GetSectionStatistic(), filtersM[i]->GetPid(),
                              id_pid(filtersM[i]->GetPid()));
         if (++count > IPTV_STATS_ACTIVE_FILTERS_COUNT)
            break;
         }
      }
  return s;
}

bool cIptvSectionFilterHandler::Delete(unsigned int indexP)
{
  debug16("%s (%d) [device %d]", __PRETTY_FUNCTION__, indexP, deviceIndexM);
  if ((indexP < eMaxSecFilterCount) && filtersM[indexP]) {
     debug16("%s (%d) Found [device %d]", __PRETTY_FUNCTION__, indexP, deviceIndexM);
     cIptvSectionFilter *tmp = filtersM[indexP];
     filtersM[indexP] = NULL;
     delete tmp;
     return true;
     }
  return false;
}

bool cIptvSectionFilterHandler::IsBlackListed(u_short pidP, u_char tidP, u_char maskP) const
{
  debug16("%s (%d, %02X, %02X) [device %d]", __PRETTY_FUNCTION__, pidP, tidP, maskP, deviceIndexM);
  // loop through section filter table
  for (int i = 0; i < SECTION_FILTER_TABLE_SIZE; ++i) {
      int index = IptvConfig.GetDisabledFilters(i);
      // Check if matches
      if ((index >= 0) && (index < SECTION_FILTER_TABLE_SIZE) &&
          (section_filter_table[index].pid == pidP) && (section_filter_table[index].tid == tidP) &&
          (section_filter_table[index].mask == maskP)) {
         debug16("%s (%d, %02X, %02X) Found %s [device %d]", __PRETTY_FUNCTION__, pidP, tidP, maskP, section_filter_table[index].description, deviceIndexM);
         return true;
         }
      }
  return false;
}

int cIptvSectionFilterHandler::Open(u_short pidP, u_char tidP, u_char maskP)
{
  // Lock
  cMutexLock MutexLock(&mutexM);
  // Blacklist check, refuse certain filters
  if (IsBlackListed(pidP, tidP, maskP))
     return -1;
  // Search the next free filter slot
  for (unsigned int i = 0; i < eMaxSecFilterCount; ++i) {
      if (!filtersM[i]) {
         filtersM[i] = new cIptvSectionFilter(deviceIndexM, pidP, tidP, maskP);
         debug16("%s (%d, %02X, %02X) handle=%d index=%u [device %d]", __PRETTY_FUNCTION__, pidP, tidP, maskP, filtersM[i]->GetFd(), i, deviceIndexM);
         return filtersM[i]->GetFd();
         }
      }
  // No free filter slot found
  return -1;
}

void cIptvSectionFilterHandler::Close(int handleP)
{
  // Lock
  cMutexLock MutexLock(&mutexM);
  // Search the filter for deletion
  for (unsigned int i = 0; i < eMaxSecFilterCount; ++i) {
      if (filtersM[i] && (handleP == filtersM[i]->GetFd())) {
         debug1("%s (%d) pid=%d handle=%d index=%d [device %d]", __PRETTY_FUNCTION__, handleP, filtersM[i]->GetPid(), filtersM[i]->GetFd(), i, deviceIndexM);
         Delete(i);
         break;
         }
      }
}

int cIptvSectionFilterHandler::GetPid(int handleP)
{
  // Lock
  cMutexLock MutexLock(&mutexM);
  // Search the filter for data
  for (unsigned int i = 0; i < eMaxSecFilterCount; ++i) {
      if (filtersM[i] && (handleP == filtersM[i]->GetFd())) {
         debug1("%s (%d) pid=%d handle=%d index=%d [device %d]", __PRETTY_FUNCTION__, handleP, filtersM[i]->GetPid(), filtersM[i]->GetFd(), i, deviceIndexM);
         return filtersM[i]->GetPid();
         }
      }
  return -1;
}

void cIptvSectionFilterHandler::Write(uchar *bufferP, int lengthP)
{
  debug16("%s (, %d) [device %d]", __PRETTY_FUNCTION__, lengthP, deviceIndexM);
  // Fill up the buffer
  if (ringBufferM) {
     int len = ringBufferM->Put(bufferP, lengthP);
     if (len != lengthP)
        ringBufferM->ReportOverflow(lengthP - len);
     }
}