Ticket #712: drb-v1.patch

libs/libmythtv/dvbrecorder.cpp

@@ -72 +72 @@
 const int DVBRecorder::POLL_INTERVAL        =  50; // msec
 const int DVBRecorder::POLL_WARNING_TIMEOUT = 500; // msec
 
+#define USE_DRB
+#ifndef USE_DRB
+static ssize_t safe_read(int fd, unsigned char *buf, size_t bufsize);
+#endif // USE_DRB
+
 #define LOC      QString("DVBRec(%1): ").arg(_card_number_option)
 #define LOC_WARN QString("DVBRec(%1) Warning: ").arg(_card_number_option)
 #define LOC_ERR  QString("DVBRec(%1) Error: ").arg(_card_number_option)
 
 DVBRecorder::DVBRecorder(TVRec *rec, DVBChannel* advbchannel)
     : DTVRecorder(rec, "DVBRecorder"),
+      _drb(NULL),
       // Options set in SetOption()
       _card_number_option(0), _record_transport_stream_option(false),
       _hw_decoder_option(false),
       // DVB stuff
       dvbchannel(advbchannel),
+      _reset_pid_filters(true),
+      _pid_lock(true),
       // PS recorder stuff
       _ps_rec_audio_id(0xC0), _ps_rec_video_id(0xE0),
       // Output stream info
       _pat(NULL), _pmt(NULL), _next_pmt_version(0),
       _ts_packets_until_psip_sync(0),
-      // Input Misc
-      _reset_pid_filters(true),
-      // Locking
-      _pid_lock(true),
       // Statistics
       _continuity_error_count(0), _stream_overflow_count(0),
       _bad_packet_count(0)
 {
     bzero(_ps_rec_buf, sizeof(unsigned char) * 3);
 
-    bzero(&_polls, sizeof(struct pollfd));
-    _polls.fd = _stream_fd;
+#ifdef USE_DRB
+    _drb = new DeviceReadBuffer(this);
+    _buffer_size = (1024*1024 / TSPacket::SIZE) * TSPacket::SIZE;
+#else
+    _buffer_size = (4*1024*1024 / TSPacket::SIZE) * TSPacket::SIZE;
+#endif
 
-    _buffer_size = (4*1024*1024 / MPEG_TS_PKT_SIZE) * MPEG_TS_PKT_SIZE;
     _buffer = new unsigned char[_buffer_size];
     bzero(_buffer, _buffer_size);
 }
@@ -113 +120 @@
 
 void DVBRecorder::TeardownAll(void)
 {
-    if (_stream_fd >= 0)
+    // Make SURE that the device read thread is cleaned up -- John Poet
+    StopRecording();
+
+    if (IsOpen())
         Close();
 
     if (_buffer)
@@ -177 +187 @@
 
 bool DVBRecorder::Open(void)
 {
-    if (_stream_fd >= 0)
+    if (IsOpen())
     {
         VERBOSE(VB_GENERAL, LOC_WARN + "Card already open");
         return true;
@@ -185 +195 @@
 
     _stream_fd = open(dvbdevice(DVB_DEV_DVR,_card_number_option),
                       O_RDONLY | O_NONBLOCK);
-    if (_stream_fd < 0)
+    if (!IsOpen())
     {
         VERBOSE(VB_IMPORTANT, LOC_ERR + "Failed to open DVB device" + ENO);
         return false;
     }
+#ifdef USE_DRB
+    if (_drb)
+        _drb->Reset(videodevice, _stream_fd);
+#endif
 
-    _polls.fd = _stream_fd;
-    _polls.events = POLLIN;
-    _polls.revents = 0;
-
     connect(dvbchannel, SIGNAL(UpdatePMTObject(const PMTObject *)),
             this, SLOT(SetPMTObject(const PMTObject *)));
 
-    VERBOSE(VB_RECORD, LOC + 
-            QString("Card opened successfully (using %1 mode).")
+    VERBOSE(VB_RECORD, LOC + QString("Card opened successfully fd(%1)")
+            .arg(_stream_fd) + QString(" (using %2 mode).")
             .arg(_record_transport_stream_option ? "TS" : "PS"));
 
     dvbchannel->RecorderStarted();
@@ -211 +221 @@
 {
     VERBOSE(VB_RECORD, LOC + "Close() fd("<<_stream_fd<<") -- begin");
 
-    if (_stream_fd < 0)
-        return;
-
-    CloseFilters();
-
-    if (_stream_fd >= 0)
+    if (IsOpen())
+    {
+        CloseFilters();
         close(_stream_fd);
+        _stream_fd = -1;
+    }
 
-    _stream_fd = -1;
-    _polls.fd = -1;
-
     VERBOSE(VB_RECORD, LOC + "Close() fd("<<_stream_fd<<") -- end");
 }
 
 void DVBRecorder::CloseFilters(void)
 {
     QMutexLocker change_lock(&_pid_lock);
-    for(unsigned int i=0; i<_pid_filters.size(); i++)
-        if (_pid_filters[i] >= 0)
-            close(_pid_filters[i]);
-    _pid_filters.clear();
 
-    pid_ipack_t::iterator iter = _ps_rec_pid_ipack.begin();
-    for (;iter != _ps_rec_pid_ipack.end(); iter++)
+    PIDInfoMap::iterator it;
+    for (it = _pid_infos.begin(); it != _pid_infos.end(); ++it)
     {
-        if ((*iter).second != NULL)
-        {
-            free_ipack((*iter).second);
-            free((void*)(*iter).second);
-        }
+        (*it)->Close();
+        delete *it;
     }
-    _ps_rec_pid_ipack.clear();
+    _pid_infos.clear();
 }
 
 void DVBRecorder::OpenFilter(uint           pid,
@@ -284 +283 @@
         usecs /= 2;
     }
     VERBOSE(VB_RECORD, LOC + "Set demux buffer size for " +
-            QString("pid 0x%1 to %2,\n\t\t\t which gives us a %3 msec buffer.")
+            QString("pid 0x%1 to %2,\n\t\t\twhich gives us a %3 msec buffer.")
             .arg(pid,0,16).arg(sz).arg(usecs/1000));
 
     // Set the filter type
@@ -303 +302 @@
         return;
     }
 
+    PIDInfo *info = new PIDInfo();
+    // Set isVideo based on stream type
+    info->isVideo = StreamID::IsVideo(stream_type);
     // Add the file descriptor to the filter list
-    QMutexLocker change_lock(&_pid_lock);
-    _pid_filters.push_back(fd_tmp);
-
-    // Initialize continuity count
-    _continuity_count[pid] = -1;
-    if (_record_transport_stream_option)
-    {
-        //Set the TS->PES converter to NULL
-        _ps_rec_pid_ipack[pid] = NULL;
-        return;
-    }
-
+    info->filter_fd = fd_tmp;
     // If we are in legacy PES mode, initialize TS->PES converter
-    ipack* ip = (ipack*)malloc(sizeof(ipack));
-    assert(ip);
-    switch (type)
-    {
-        case ES_TYPE_VIDEO_MPEG1:
-        case ES_TYPE_VIDEO_MPEG2:
-            init_ipack(ip, 2048, ProcessDataPS);
-            ip->replaceid = _ps_rec_video_id++;
-            break;
+    if (!_record_transport_stream_option)
+        info->ip = CreateIPack(type);
 
-        case ES_TYPE_AUDIO_MPEG1:
-        case ES_TYPE_AUDIO_MPEG2:
-            init_ipack(ip, 65535, ProcessDataPS); /* don't repack PES */
-            ip->replaceid = _ps_rec_audio_id++;
-            break;
-
-        case ES_TYPE_AUDIO_AC3:
-            init_ipack(ip, 65535, ProcessDataPS); /* don't repack PES */
-            ip->priv_type = PRIV_TS_AC3;
-            break;
-
-        case ES_TYPE_SUBTITLE:
-        case ES_TYPE_TELETEXT:
-            init_ipack(ip, 65535, ProcessDataPS); /* don't repack PES */
-            ip->priv_type = PRIV_DVB_SUB;
-            break;
-
-        default:
-            init_ipack(ip, 2048, ProcessDataPS);
-            break;
-    }
-    ip->data = (void*)this;
-    _ps_rec_pid_ipack[pid] = ip;
+    // Add the new info to the map
+    QMutexLocker change_lock(&_pid_lock);    
+    _pid_infos[pid] = info;
 }
 
-bool DVBRecorder::SetDemuxFilters(void)
+bool DVBRecorder::OpenFilters(void)
 {
-    QMutexLocker change_lock(&_pid_lock);
     CloseFilters();
 
-    _continuity_count.clear();
-    _encrypted_pid.clear();
-    _payload_start_seen.clear();
-    data_found = false;
+    QMutexLocker change_lock(&_pid_lock);
+
     _wait_for_keyframe = _wait_for_keyframe_option;
-    keyframe_found = false;
 
     _ps_rec_audio_id = 0xC0;
     _ps_rec_video_id = 0xE0;
@@ -427 +387 @@
                     (*es).Orig_Type);
 
 
-    if (_pid_filters.size() == 0 && _ps_rec_pid_ipack.size() == 0)
+    if (_pid_infos.empty())
     {        
         VERBOSE(VB_GENERAL, LOC_WARN +
                 "Recording will not commence until a PID is set.");
@@ -436 +396 @@
     return true;
 }
 
-/*
- *  Process PMT and decide which components should be recorded
+/** \fn DVBRecorder::AutoPID(void)
+ *  \brief Process PMT and decide which components should be recorded
+ *
+ *  This is particularly for the hardware decoders which don't like
+ *  to see more than one audio or video stream.
  */
 void DVBRecorder::AutoPID(void)
 {
     QMutexLocker change_lock(&_pid_lock);
-    _videoPID.clear();
 
     VERBOSE(VB_RECORD, LOC +
             QString("AutoPID for MPEG Program Number(%1), PCR PID(0x%2)")
             .arg(_input_pmt.ServiceID).arg(((uint)_input_pmt.PCRPID),0,16));
 
-    // Wanted languages:
-    //QStringList Languages = iso639_get_language_list();
-
     // Wanted stream types:
     QValueList<ES_Type> StreamTypes;
     StreamTypes += ES_TYPE_VIDEO_MPEG1;
@@ -467 +426 @@
         StreamTypes += ES_TYPE_SUBTITLE;
     }
 
-    QMap<ES_Type, bool> flagged;
+    uint has_video = 0;
+    uint has_audio = 0;
+
     QValueList<ElementaryPIDObject>::Iterator es;
     for (es = _input_pmt.Components.begin();
          es != _input_pmt.Components.end(); ++es)
     {
+        // Check if this is a stream we know about
         if (!StreamTypes.contains((*es).Type))
-        {
-            // Type not wanted
             continue;
-        }
 
-        if (((*es).Type == ES_TYPE_AUDIO_MPEG1) ||
-            ((*es).Type == ES_TYPE_AUDIO_MPEG2) ||
-            ((*es).Type == ES_TYPE_AUDIO_AC3))
-        {
-            bool ignore = false;
-
-            // Check for audio descriptors
-            DescriptorList::Iterator dit;
-            for (dit = (*es).Descriptors.begin();
-                 dit != (*es).Descriptors.end(); ++dit)
-            {
-                // Check for "Hearing impaired" or 
-                // "Visual impaired commentary" stream
-                if (((*dit).Data[0] == 0x0A) &&
-                    ((*dit).Data[5] & 0xFE == 0x02))
-                {
-                    ignore = true;
-                    break;
-                }
-            }
-
-            if (ignore)
-                continue; // Ignore this stream
-        }
-
         if (_hw_decoder_option)
         {
-            // Limit hardware decoders to one A/V stream
-            switch ((*es).Type)
-            {
-                case ES_TYPE_VIDEO_MPEG1:
-                case ES_TYPE_VIDEO_MPEG2:
-                    if (flagged.contains(ES_TYPE_VIDEO_MPEG1) ||
-                        flagged.contains(ES_TYPE_VIDEO_MPEG2))
-                    {
-                        continue; // Ignore this stream
-                    }
-                    break;
+            has_video += StreamID::IsVideo((*es).Orig_Type) ? 1 : 0;
+            has_audio += StreamID::IsAudio((*es).Orig_Type) ? 1 : 0;
 
-                case ES_TYPE_AUDIO_MPEG1:
-                case ES_TYPE_AUDIO_MPEG2:
-                    if (flagged.contains(ES_TYPE_AUDIO_MPEG1) ||
-                        flagged.contains(ES_TYPE_AUDIO_MPEG2))
-                    {
-                        continue; // Ignore this stream
-                    }
-                    break;
-
-                default:
-                    break;
-            }
+            // Limit hardware decoders streams to one video stream
+            if (StreamID::IsVideo((*es).Orig_Type) && has_video > 1)
+                continue;
+            // Limit hardware decoders streams to one audio stream
+            if (StreamID::IsAudio((*es).Orig_Type) && has_audio > 1)
+                continue;
         }
 
-        //if (Languages.isEmpty() // No specific language wanted
-        //    || (*es).Language.isEmpty() // Component has no language
-        //    || Languages.contains((*es).Language)) // This language is wanted!
-        {
-            (*es).Record = true;
-            flagged[(*es).Type] = true;
-        }
+        (*es).Record = true;
     }
 
+    if (!(print_verbose_messages|VB_RECORD))
+        return;
+
+    // print out some debugging info
     for (es = _input_pmt.Components.begin();
          es != _input_pmt.Components.end(); ++es)
     {
-        if (StreamTypes.contains((*es).Type) && !flagged.contains((*es).Type))
-        {
-            // We want this stream type but no component was flagged
-            (*es).Record = true;
-        }
-
-        if ((*es).Record)
-        {
-            VERBOSE(VB_RECORD, LOC +
-                    QString("AutoPID selecting PID 0x%1, %2")
-                    .arg((*es).PID,0,16).arg((*es).Description));
-
-            switch ((*es).Type)
-            {
-                case ES_TYPE_VIDEO_MPEG1:
-                case ES_TYPE_VIDEO_MPEG2:
-                    _videoPID[(*es).PID] = true;
-                    break;
-
-                default:
-                    // Do nothing
-                    break;
-            }
-        }
-        else
-        {
-            VERBOSE(VB_RECORD, LOC +
-                    QString("AutoPID skipping PID 0x%1, %2")
-                    .arg((*es).PID,0,16).arg((*es).Description));
-        }
+        QString msg = ((*es).Record) ? "recording" : "skipping";
+        VERBOSE(VB_RECORD, LOC +
+                QString("AutoPID %1 PID 0x%2, %3")
+                .arg(msg).arg((*es).PID,0,16).arg((*es).Description));
     }
 
     VERBOSE(VB_RECORD, LOC + "AutoPID Complete - PAT/PMT Loaded for service");
 
     QString msg = (_input_pmt.FTA()) ? "unencrypted" : "ENCRYPTED";
-    VERBOSE(VB_RECORD, LOC + "A/V Stream is " + msg);
+    VERBOSE(VB_RECORD, LOC + "AutoPID A/V Stream is " + msg);
 }
 
 void DVBRecorder::StartRecording(void)
@@ -600 +491 @@
     SetPMT(NULL);
     _ts_packets_until_psip_sync = 0;
 
-    MythTimer t;
-    t.start();
+#ifdef USE_DRB
+    bool ok = _drb->Setup(videodevice, _stream_fd);
+    if (!ok)
+    {
+        VERBOSE(VB_IMPORTANT, LOC_ERR + "Failed to allocate DRB buffer");
+        Close();
+        _error = true;
+        return;
+    }
+    _drb->Start();
+#else
+    _poll_timer.start();
+#endif // USE_DRB
+
     while (_request_recording && !_error)
     {
         if (PauseAndWait())
             continue;
 
-        if (_stream_fd < 0)
+        if (!IsOpen())
         {
-            usleep(50);
+            usleep(2000);
             continue;
         }
 
         if (_reset_pid_filters)
         {
+            _reset_pid_filters = false;
             VERBOSE(VB_RECORD, LOC + "Resetting Demux Filters");
-            if (SetDemuxFilters())
+            if (OpenFilters())
             {
                 CreatePAT();
                 CreatePMT();
             }
-            _reset_pid_filters = false;
         }
 
-        int ret;
-        do
-            ret = poll(&_polls, 1, POLL_INTERVAL);
-        while (!request_pause && (_stream_fd >= 0) &&
-               ((-1 == ret) && ((EAGAIN == errno) || (EINTR == errno))));
+        if (Poll())
+        {
+#ifndef USE_DRB
+            ssize_t len = safe_read(_stream_fd, _buffer, _buffer_size);
+#else
+            ssize_t len = _drb->Read(_buffer, _buffer_size);
+#endif // !USE_DRB
+            if (len > 0)
+                ProcessDataTS(_buffer, len);
+        }
 
-        if (request_pause || _stream_fd < 0)
-            continue;
-
-        if (ret == 0 && t.elapsed() > POLL_WARNING_TIMEOUT)
+        // Check for DRB errors
+        if (_drb->IsErrored())
         {
-            VERBOSE(VB_GENERAL, LOC_WARN +
-                    QString("No data from card in %1 milliseconds.")
-                    .arg(t.elapsed()));
+            VERBOSE(VB_IMPORTANT, LOC_ERR + "Device error detected");
+            _error = true;
         }
-        else if (ret == 1 && _polls.revents & POLLIN)
+
+        if (_drb->IsEOF())
         {
-            int msec = t.restart();
-            if (msec >= POLL_WARNING_TIMEOUT)
-            {
-                VERBOSE(VB_IMPORTANT, LOC_WARN +
-                        QString("Got data from card after %1 ms. (>%2)")
-                        .arg(msec).arg(POLL_WARNING_TIMEOUT));
-            }
-
-            ReadFromDMX();
-            if (t.elapsed() >= 20)
-            {
-                VERBOSE(VB_RECORD, LOC_WARN +
-                        QString("ReadFromDMX took %1 ms").arg(t.elapsed()));
-            }
+            VERBOSE(VB_IMPORTANT, LOC_ERR + "Device EOF detected");
+            _error = true;
         }
-        else if ((ret < 0) || (ret == 1 && _polls.revents & POLLERR))
-            VERBOSE(VB_IMPORTANT, LOC_ERR +
-                    "Poll failed while waiting for data." + ENO);
     }
 
+#ifdef USE_DRB
+    if (_drb)
+        _drb->Stop();
+#endif        
+
     Close();
 
     FinishRecording();
@@ -668 +564 @@
     _recording = false;
 }
 
-void DVBRecorder::ReadFromDMX(void)
+void DVBRecorder::StopRecording(void)
 {
-    int cardnum = _card_number_option;
-    int readsz = 1;
-    unsigned char *pktbuf;
+#ifdef USE_DRB
+    TVRec *rec = tvrec;
+    tvrec = NULL; // don't notify of pause..
 
-    while (readsz > 0)
+    bool ok = true;
+    if (!IsPaused())
     {
-        readsz = read(_stream_fd, _buffer, _buffer_size);
-        if (readsz < 0)
-        {
-            if (errno == EOVERFLOW)
-            {
-                ++_stream_overflow_count;
-                VERBOSE(VB_RECORD, LOC +
-                        "DVB Buffer overflow error detected on read");
-                break;
-            }
+        Pause();
+        ok = WaitForPause(250);
+    }
 
-            if (errno == EAGAIN)
-                break;
-            VERBOSE(VB_IMPORTANT, LOC_ERR +
-                    "Error reading from DVB device." + ENO);
-            break;
-        } else if (readsz == 0)
-            break;
+    _request_recording = false;
 
-        if (readsz % MPEG_TS_PKT_SIZE)
-        {
-            VERBOSE(VB_IMPORTANT, LOC_ERR + "Incomplete packet received.");
-            readsz = readsz - (readsz % MPEG_TS_PKT_SIZE);
-        }
+    _drb->Stop();
 
-        int pkts = readsz / MPEG_TS_PKT_SIZE;
-        int curpkt = 0;
+    if (ok)
+        _drb->Teardown();
+    else // Better to have a memory leak, than a segfault? -- John Poet
+        VERBOSE(VB_IMPORTANT, LOC_ERR + "DeviceReadBuffer not cleaned up!");
 
-        _pid_lock.lock();
-        while (curpkt < pkts)
-        {
-            if (data_found == false)
-            {
-                GENERAL("Data read from DMX - "
-                        "This is for debugging with transform.c");
-                data_found = true;
-            }
+    while (_recording)
+        usleep(2000);
 
-            pktbuf = _buffer + (curpkt * MPEG_TS_PKT_SIZE);
-            curpkt++;
+    tvrec = rec;
+#else
+    DTVRecorder::StopRecording();
+#endif
+}
 
-            int pes_offset = 0;
-            int pid = ((pktbuf[1]&0x1f) << 8) | pktbuf[2];
-            uint8_t scrambling = (pktbuf[3] >> 6) & 0x03;
-            uint8_t cc = pktbuf[3] & 0xf;
-            uint8_t content = (pktbuf[3] & 0x30) >> 4;
+void DVBRecorder::ReaderPaused(int /*fd*/)
+{
+#ifdef USE_DRB
+    pauseWait.wakeAll();
+    if (tvrec)
+        tvrec->RecorderPaused();
+#endif // USE_DRB
+}
 
-            if (pktbuf[1] & 0x80)
-            {
-                VERBOSE(VB_RECORD, LOC +
-                        "Packet dropped due to uncorrectable error.");
-                ++_bad_packet_count;
-                continue;
-            }
+bool DVBRecorder::PauseAndWait(int timeout)
+{
+#ifdef USE_DRB
+    if (request_pause)
+    {
+        paused = true;
+        if (!_drb->IsPaused())
+            _drb->SetRequestPause(true);
 
-            if (scrambling)
-            {
-                if (!_encrypted_pid[pid])
-                {
-                    VERBOSE(VB_RECORD, LOC +
-                            QString("PID 0x%1 is encrypted, ignoring")
-                            .arg(pid,0,16));
-                    _encrypted_pid[pid] = true;
-                }
-                continue; // Drop encrypted TS packet
-            }
+        unpauseWait.wait(timeout);
+    }
+    else if (_drb->IsPaused())
+    {
+        _drb->SetRequestPause(false);
+        _drb->WaitForUnpause(timeout);
+        paused = _drb->IsPaused();
+    }
+    else
+    {
+        paused = false;
+    }
+    return paused;
+#else // if !USE_DRB
+    return RecorderBase::PauseAndWait(timeout);
+#endif // !USE_DRB
+}
 
-            if (_encrypted_pid[pid])
-            {
-                VERBOSE(VB_RECORD, LOC +
-                        QString("PID 0x%1 is no longer encrypted")
-                        .arg(pid,0,16));
-                _encrypted_pid[pid] = false;
-            }
-            if (content & 0x1)
-            {
-                 if (_continuity_count[pid] < 0)
-                     _continuity_count[pid] = cc;
-                 else
-                 {
-                     _continuity_count[pid] = (_continuity_count[pid]+1) & 0xf;
-                     if (_continuity_count[pid] != cc)
-                     {
-                         VERBOSE(VB_RECORD, LOC +
-                                 QString("PID 0x%1 _continuity_count %2 cc %3")
-                                 .arg(pid,0,16).arg(_continuity_count[pid])
-                                 .arg(cc));
-                         _continuity_count[pid] = cc;
-                         ++_continuity_error_count;
-                     }
-                 }
-            }
+uint DVBRecorder::ProcessDataTS(unsigned char *buffer, uint len)
+{
+    if (len % TSPacket::SIZE)
+    {
+        VERBOSE(VB_IMPORTANT, LOC_ERR + "Incomplete packet received.");
+        len = len - (len % TSPacket::SIZE);
+    }
+    if (len < TSPacket::SIZE)
+        return len;
 
-            if (_record_transport_stream_option)
-            {   // handle TS recording
-                MythTimer t;
-                t.start();
-                if (_videoPID[pid])
-                {
-                    // Check for keyframe
-                    const TSPacket *pkt =
-                        reinterpret_cast<const TSPacket*>(pktbuf);
-                    FindKeyframes(pkt);
-                }
-                if (t.elapsed() > 10)
-                {
-                    VERBOSE(VB_RECORD, LOC_WARN +
-                            QString("Find keyframes took %1 ms!")
-                            .arg(t.elapsed()));
-                }
+    uint pos = 0;
+    uint end = len - TSPacket::SIZE;
+    while (pos <= end)
+    {
+        const TSPacket *pkt = reinterpret_cast<const TSPacket*>(&buffer[pos]);
+        ProcessTSPacket(*pkt);
+        pos += TSPacket::SIZE;
+    }
+    return len - pos;
+}
 
-                // Sync recording start to first keyframe
-                if (_wait_for_keyframe && !_keyframe_seen)
-                    continue;
-                if (!keyframe_found)
-                {
-                    keyframe_found = true;
-                    VERBOSE(VB_RECORD, QString("Found first keyframe"));
-                }
+bool DVBRecorder::ProcessTSPacket(const TSPacket& tspacket)
+{
+    if (tspacket.TransportError())
+    {
+        VERBOSE(VB_RECORD, LOC + "Packet dropped due to uncorrectable error.");
+        ++_bad_packet_count;
+        return false; // Drop bad TS packets...
+    }
 
-                // Sync streams to the first Payload Unit Start Indicator
-                // _after_ first keyframe iff _wait_for_keyframe is true
-                if (!_payload_start_seen[pid])
-                {
-                    if ((pktbuf[1] & 0x40) == 0)
-                        continue; // not payload start - drop packet
+    const uint pid = tspacket.PID();
 
-                    VERBOSE(VB_RECORD, QString("Found Payload Start for PID %1").arg(pid));
-                    _payload_start_seen[pid] = true;
-                }
+    QMutexLocker locker(&_pid_lock);
+    PIDInfo *info = _pid_infos[pid];
+    if (!info)
+       info = _pid_infos[pid] = new PIDInfo();
 
-                t.start();
-                LocalProcessDataTS(pktbuf, MPEG_TS_PKT_SIZE);
-                if (t.elapsed() > 10)
-                {
-                    VERBOSE(VB_RECORD, LOC_WARN +
-                            QString("TS packet write took %1 ms!")
-                            .arg(t.elapsed()));
-                }
-            }
-            else
-            {   // handle PS recording
-                ipack *ip = _ps_rec_pid_ipack[pid];
-                if (ip == NULL)
-                    continue;
-
-                ip->ps = 1;
-
-                if ((pktbuf[1] & 0x40) && (ip->plength == MMAX_PLENGTH-6))
-                {
-                    ip->plength = ip->found-6;
-                    ip->found = 0;
-                    send_ipack(ip);
-                    reset_ipack(ip);
-                }
-
-                if (content & 0x2)
-                    pes_offset = pktbuf[4] + 1;
-
-                if (pes_offset > 183)
-                    continue;
-
-                instant_repack(pktbuf + 4 + pes_offset,
-                               MPEG_TS_PKT_SIZE - 4 - pes_offset, ip);
-            }
+    // track scrambled pids
+    if (tspacket.ScramplingControl())
+    {
+        if (!info->isEncrypted)
+        {
+            VERBOSE(VB_RECORD, LOC +
+                    QString("PID 0x%1 is encrypted, ignoring").arg(pid,0,16));
+            info->isEncrypted = true;
         }
-        _pid_lock.unlock();
+        return true; // Drop encrypted TS packets...
     }
-}
+    else if (info->isEncrypted)
+    {
+        VERBOSE(VB_RECORD, LOC +
+                QString("PID 0x%1 is no longer encrypted").arg(pid,0,16));
+        info->isEncrypted = false;
+    }
 
-#define SEQ_START     0x000001B3
-#define GOP_START     0x000001B8
-#define PICTURE_START 0x00000100
-#define SLICE_MIN     0x00000101
-#define SLICE_MAX     0x000001af
-
-void DVBRecorder::ProcessDataPS(unsigned char *buffer, int len, void *priv)
-{
-    ((DVBRecorder*)priv)->LocalProcessDataPS(buffer, len);
-}
-
-void DVBRecorder::LocalProcessDataPS(unsigned char *buffer, int len)
-{
-    if (buffer[0] != 0x00 || buffer[1] != 0x00 || buffer[2] != 0x01)
+    // Check continuity counter
+    if (tspacket.HasPayload())
     {
-        if (!_wait_for_keyframe)
-            ringBuffer->Write(buffer, len);
-        return;
+        if (!info->CheckCC(tspacket.ContinuityCounter()))
+        {
+            VERBOSE(VB_RECORD, LOC +
+                    QString("PID 0x%1 discontinuity detected").arg(pid,0,16));
+            _continuity_error_count++;
+        }
     }
 
-    if (buffer[3] >= VIDEO_STREAM_S && buffer[3] <= VIDEO_STREAM_E)
+    if (_record_transport_stream_option)
     {
-        int pos = 8 + buffer[8];
-        int datalen = len - pos;
+        // handle TS recording
 
-        unsigned char *bufptr = &buffer[pos];
-        uint state = 0xFFFFFFFF;
-        uint state_byte = 0;
-        int prvcount = -1;
+        // Check for keyframes and count frames
+        if (info->isVideo)
+            FindKeyframes(&tspacket);
 
-        while (bufptr < &buffer[pos] + datalen)
+        // Sync recording start to first keyframe
+        if (_wait_for_keyframe_option && !_keyframe_seen)
+            return true;
+
+        // Sync streams to the first Payload Unit Start Indicator
+        // _after_ first keyframe iff _wait_for_keyframe_option is true
+        if (!info->payloadStartSeen)
         {
-            if (++prvcount < 3)
-                state_byte = _ps_rec_buf[prvcount];
-            else
-                state_byte = *bufptr++;
+            if (!tspacket.PayloadStart())
+                return true; // not payload start - drop packet
 
-            if (state != 0x000001)
-            {
-                state = ((state << 8) | state_byte) & 0xFFFFFF;
-                continue;
-            }
+            VERBOSE(VB_RECORD,
+                    QString("PID 0x%1 Found Payload Start").arg(pid,0,16));
+            info->payloadStartSeen = true;
+        }
 
-            state = ((state << 8) | state_byte) & 0xFFFFFF;
-            if (state >= SLICE_MIN && state <= SLICE_MAX)
-                continue;
+        // Write PAT & PMT tables occasionally
+        WritePATPMT();
 
-            if (state == SEQ_START)
-                _wait_for_keyframe = false;
+        // Write Data
+        ringBuffer->Write(tspacket.data(), TSPacket::SIZE);
+    }
+    else
+    {
+        // handle PS recording
 
-            if (GOP_START == state)
-            {
-                long long startpos = ringBuffer->GetWritePosition();
-                        
-                if (!_position_map.contains(_frames_written_count))
-                {
-                    _position_map_delta[_frames_written_count] = startpos;
-                    _position_map[_frames_written_count] = startpos;
+        ipack *ip = info->ip;
+        if (ip == NULL)
+            return true;
 
-                    if (curRecording &&
-                        ((_position_map_delta.size() % 30) == 0))
-                    {
-                        curRecording->SetPositionMapDelta(
-                            _position_map_delta,
-                            MARK_GOP_BYFRAME);
-                        curRecording->SetFilesize(startpos);
-                        _position_map_delta.clear();
-                    }
-                }
-            }
-            else if (PICTURE_START == state)
-                _frames_written_count++;
+        ip->ps = 1;
+
+        if (tspacket.PayloadStart() && (ip->plength == MMAX_PLENGTH-6))
+        {
+            ip->plength = ip->found-6;
+            ip->found = 0;
+            send_ipack(ip);
+            reset_ipack(ip);
         }
+
+        uint afc_offset = tspacket.AFCOffset();
+        if (afc_offset > 187)
+            return true;
+
+        instant_repack((uint8_t*)tspacket.data() + afc_offset,
+                       TSPacket::SIZE - afc_offset, ip);
     }
-    memcpy(_ps_rec_buf, &buffer[len - 3], 3);
-
-    if (!_wait_for_keyframe)
-        ringBuffer->Write(buffer, len);
+    return true;
 }
 
-void DVBRecorder::LocalProcessDataTS(unsigned char *buffer, int len)
+void DVBRecorder::WritePATPMT(void)
 {
-    QMutexLocker read_lock(&_pid_lock);
     if (_ts_packets_until_psip_sync == 0)
     {
+        QMutexLocker read_lock(&_pid_lock);
         if (_pat && _pmt)
         {
             ringBuffer->Write(_pat->tsheader()->data(), TSPacket::SIZE);
@@ -940 +769 @@
     }
     else
         _ts_packets_until_psip_sync--;
-
-    ringBuffer->Write(buffer,len);
 }
 
 void DVBRecorder::Reset(void)
@@ -977 +804 @@
 void DVBRecorder::CreatePAT(void)
 {
     QMutexLocker read_lock(&_pid_lock);
-    int next_cc = 0;
+    uint next_cc = 0;
     if (_pat)
         next_cc = (_pat->tsheader()->ContinuityCounter() + 1) & 0x0F;
 
@@ -989 +816 @@
     SetPAT(ProgramAssociationTable::Create(tsid, next_cc, pnum, pid));
 }
 
+//#define USE_OLD_CREATE_PMT
+#ifndef USE_OLD_CREATE_PMT
+static void DescList_to_desc_list(DescriptorList &list, desc_list_t &vec)
+{
+    vec.clear();
+    for (DescriptorList::iterator it = list.begin(); it != list.end(); ++it)
+        vec.push_back((*it).Data);
+}
+
 void DVBRecorder::CreatePMT(void)
 {
     QMutexLocker read_lock(&_pid_lock);
-    int pmt_cc = 0;
+
+    // Figure out what goes into the PMT
+    uint programNumber = 1; // MPEG Program Number
+    desc_list_t gdesc;
+    vector<uint> pids;
+    vector<uint> types;
+    vector<desc_list_t> pdesc;
+    QValueList<ElementaryPIDObject>::iterator it;
+
+    DescList_to_desc_list(_input_pmt.Descriptors, gdesc);
+
+    it = _input_pmt.Components.begin();
+    for (; it != _input_pmt.Components.end(); ++it)
+    {
+        if ((*it).Record)
+        {
+            pids.push_back((*it).PID);
+            types.push_back((*it).Orig_Type);
+            pdesc.resize(pdesc.size()+1);
+            DescList_to_desc_list((*it).Descriptors, pdesc.back());
+        }
+    }
+
+    // Create the PMT
+    ProgramMapTable *pmt = ProgramMapTable::Create(
+        programNumber, PMT_PID, _input_pmt.PCRPID,
+        _next_pmt_version, gdesc,
+        pids, types, pdesc);
+
+    // Increment the continuity counter...
+    uint pmt_cc = 0;
     if (_pmt)
         pmt_cc = (_pmt->tsheader()->ContinuityCounter() + 1) & 0x0F;
+    pmt->tsheader()->SetContinuityCounter(pmt_cc);
 
-    TSPacket pkt;
-    uint8_t *ts_packet = pkt.data();
-    memset(ts_packet, 0xFF, TSPacket::SIZE);
+    SetPMT(pmt);
+}
+#endif    
 
-    ts_packet[0] = 0x47;                            // sync byte
-    ts_packet[1] = 0x40 | ((PMT_PID >> 8) & 0x1F);  // payload start & PID
-    ts_packet[2] = PMT_PID & 0xFF;                  // PID
-    // scrambling, adaptation & continuity counter
-    ts_packet[3] = 0x10 | pmt_cc; 
-    ts_packet[4] = 0x00;                            // pointer field
+////////////////////////////////////////////////////////////
+// Stuff below this comment will be phased out after 0.20 //
+////////////////////////////////////////////////////////////
 
-    ++pmt_cc &= 0x0F;   // inc. continuity counter
-    uint8_t *pmt_data = ts_packet + 5;
-    int p = 0;
+#ifdef USE_OLD_CREATE_PMT
+void DVBRecorder::CreatePMT(void)
+{
+    QMutexLocker read_lock(&_pid_lock);
+    uint pmt_cc = 0;
+    if (_pmt)
+        pmt_cc = (_pmt->tsheader()->ContinuityCounter() + 1) & 0x0F;
 
-    pmt_data[p++] = PMT_TID; // table ID
-    pmt_data[p++] = 0xB0;    // section syntax indicator
-    p++;                // section length (set later)
-    pmt_data[p++] = 0;       // program number (ServiceID)
-    pmt_data[p++] = 1;       // program number (ServiceID)
-    pmt_data[p++] = 0xC1 + (_next_pmt_version << 1); // Version + Current/Next
-    pmt_data[p++] = 0;       // Current Section
-    pmt_data[p++] = 0;       // Last Section
-    pmt_data[p++] = (_input_pmt.PCRPID >> 8) & 0x1F;
-    pmt_data[p++] = _input_pmt.PCRPID & 0xFF;
+    ProgramMapTable *pmt = ProgramMapTable::CreateBlank();
+    pmt->tsheader()->SetPID(PMT_PID);
+    pmt->tsheader()->SetContinuityCounter(pmt_cc);
 
+    pmt->SetProgramNumber(1);
+    pmt->SetPCRPID(_input_pmt.PCRPID);
+    pmt->SetVersionNumber(_next_pmt_version);
+    uint8_t *pmt_data = pmt->tsheader()->data() + 5;
+    uint p = 10;
+
     // Write descriptors
     int program_info_length = 0;
     DescriptorList::Iterator dit;
@@ -1111 +978 @@
     pmt_data[p++] = (crc >> 16) & 0xFF;
     pmt_data[p++] = (crc >> 8) & 0xFF;
     pmt_data[p++] = crc & 0xFF;
-    
-    SetPMT(new ProgramMapTable(PSIPTable(pkt)));
+
+    SetPMT(pmt);
 }
+#endif    
 
-void DVBRecorder::DebugTSHeader(unsigned char* buffer, int len)
+bool DVBRecorder::Poll(void) const
 {
-    (void) len;
+#ifndef USE_DRB
+    struct pollfd polls;
+    polls.fd      = _stream_fd;
+    polls.events  = POLLIN;
+    polls.revents = 0;
 
-    uint8_t sync = buffer[0];
-    uint8_t transport_error = (buffer[1] & 0x80) >> 7;
-    uint8_t payload_start = (buffer[1] & 0x40) >> 6;
-    uint16_t pid = (buffer[1] & 0x1F) << 8 | buffer[2];
-    uint8_t transport_scrambled = (buffer[3] & 0xB0) >> 6;
-    uint8_t adaptation_control = (buffer[3] & 0x30) >> 4;
-    uint8_t counter = buffer[3] & 0x0F;
+    int ret;
+    do
+        ret = poll(&polls, 1, POLL_INTERVAL);
+    while (!request_pause && IsOpen() &&
+           ((-1 == ret) && ((EAGAIN == errno) || (EINTR == errno))));
 
-    int pos=4;
-    if (adaptation_control == 2 || adaptation_control == 3)
+    if (request_pause || !IsOpen())
+        return false;
+
+    if (ret > 0 && polls.revents & POLLIN)
     {
-        unsigned char adaptation_length;
-        adaptation_length = buffer[pos++];
-        pos += adaptation_length;
+        if (_poll_timer.elapsed() >= POLL_WARNING_TIMEOUT)
+        {
+            VERBOSE(VB_IMPORTANT, LOC_WARN +
+                    QString("Got data from card after %1 ms. (>%2)")
+                    .arg(_poll_timer.elapsed()).arg(POLL_WARNING_TIMEOUT));
+        }
+        _poll_timer.start();
+        return true;
     }
 
-    QString debugmsg =
-        QString("sync: %1 err: %2 paystart: %3 "
-                "pid: %4 enc: %5 adaptation: %6 counter: %7")
-        .arg(sync, 2, 16)
-        .arg(transport_error)
-        .arg(payload_start)
-        .arg(pid)
-        .arg(transport_scrambled)
-        .arg(adaptation_control)
-        .arg(counter);
+    if (ret == 0 && _poll_timer.elapsed() > POLL_WARNING_TIMEOUT)
+    {
+        VERBOSE(VB_GENERAL, LOC_WARN +
+                QString("No data from card in %1 ms.")
+                .arg(_poll_timer.elapsed()));
+    }
 
-    const TSPacket *pkt = reinterpret_cast<const TSPacket*>(&buffer[0]);
-    FindKeyframes(pkt);
+    if (ret < 0 && (EOVERFLOW == errno))
+    {
+        _stream_overflow_count++;
+        VERBOSE(VB_RECORD, LOC_ERR + "Driver buffer overflow detected.");
+    }
 
-    int cardnum = _card_number_option;
-    GENERAL(debugmsg);
+    if ((ret < 0) || (ret > 0 && polls.revents & POLLERR))
+    {
+        VERBOSE(VB_IMPORTANT, LOC_ERR +
+                "Poll failed while waiting for data." + ENO);
+    }
+
+    return false;
+#else // if USE_DRB
+    return true;
+#endif // USE_DRB
 }
+
+void DVBRecorder::ProcessDataPS(unsigned char *buffer, uint len)
+{
+    if (len < 4)
+        return;
+
+    if (buffer[0] != 0x00 || buffer[1] != 0x00 || buffer[2] != 0x01)
+    {
+        if (!_wait_for_keyframe_option || _keyframe_seen)
+            ringBuffer->Write(buffer, len);
+        return;
+    }
+
+    uint stream_id = buffer[3];
+    if ((stream_id >= PESStreamID::MPEGVideoStreamBegin) &&
+        (stream_id <= PESStreamID::MPEGVideoStreamEnd))
+    {
+        uint pos = 8 + buffer[8];
+        uint datalen = len - pos;
+
+        unsigned char *bufptr = &buffer[pos];
+        uint state = 0xFFFFFFFF;
+        uint state_byte = 0;
+        int prvcount = -1;
+
+        while (bufptr < &buffer[pos] + datalen)
+        {
+            state_byte  = (++prvcount < 3) ? _ps_rec_buf[prvcount] : *bufptr++;
+            uint last   = state;
+            state       = ((state << 8) | state_byte) & 0xFFFFFF;
+            stream_id   = state_byte;
+
+            // Skip non-prefixed stream id's and skip slice PES stream id's
+            if ((last != 0x000001) ||
+                ((stream_id >= PESStreamID::SliceStartCodeBegin) &&
+                 (stream_id <= PESStreamID::SliceStartCodeEnd)))
+            {
+                continue;
+            }
+
+            // Now process the stream id's we care about
+            if (PESStreamID::PictureStartCode == stream_id)
+                _frames_written_count++;
+            else if (PESStreamID::SequenceStartCode == stream_id)
+                _keyframe_seen = true;
+            else if (PESStreamID::GOPStartCode == stream_id)
+            {
+                _position_map_lock.lock();
+                bool save_map = false;
+                if (!_position_map.contains(_frames_written_count))
+                {
+                    long long startpos = ringBuffer->GetWritePosition();
+                    _position_map_delta[_frames_written_count] = startpos;
+                    _position_map[_frames_written_count] = startpos;
+                    save_map = true;
+                }
+                _position_map_lock.unlock();
+                if (save_map)
+                    SavePositionMap(false);
+            }
+        }
+    }
+    memcpy(_ps_rec_buf, &buffer[len - 3], 3);
+
+    if (!_wait_for_keyframe_option || _keyframe_seen)
+        ringBuffer->Write(buffer, len);
+}
+
+void DVBRecorder::process_data_ps_cb(unsigned char *buffer,
+                                     int len, void *priv)
+{
+    ((DVBRecorder*)priv)->ProcessDataPS(buffer, (uint)len);
+}
+
+ipack *DVBRecorder::CreateIPack(ES_Type type)
+{
+    ipack* ip = (ipack*)malloc(sizeof(ipack));
+    assert(ip);
+    switch (type)
+    {
+        case ES_TYPE_VIDEO_MPEG1:
+        case ES_TYPE_VIDEO_MPEG2:
+            init_ipack(ip, 2048, process_data_ps_cb);
+            ip->replaceid = _ps_rec_video_id++;
+            break;
+
+        case ES_TYPE_AUDIO_MPEG1:
+        case ES_TYPE_AUDIO_MPEG2:
+            init_ipack(ip, 65535, process_data_ps_cb); /* don't repack PES */
+            ip->replaceid = _ps_rec_audio_id++;
+            break;
+
+        case ES_TYPE_AUDIO_AC3:
+            init_ipack(ip, 65535, process_data_ps_cb); /* don't repack PES */
+            ip->priv_type = PRIV_TS_AC3;
+            break;
+
+        case ES_TYPE_SUBTITLE:
+        case ES_TYPE_TELETEXT:
+            init_ipack(ip, 65535, process_data_ps_cb); /* don't repack PES */
+            ip->priv_type = PRIV_DVB_SUB;
+            break;
+
+        default:
+            init_ipack(ip, 2048, process_data_ps_cb);
+            break;
+    }
+    ip->data = (void*)this;
+    return ip;
+}
+
+#ifndef USE_DRB
+static ssize_t safe_read(int fd, unsigned char *buf, size_t bufsize)
+{
+    ssize_t size = read(fd, buf, bufsize);
+
+    if ((size < 0) &&
+        (errno != EAGAIN) && (errno != EOVERFLOW) && (EINTR != errno))
+    {
+        VERBOSE(VB_IMPORTANT, "DVB:safe_read(): "
+                "Error reading from DVB device." + ENO);
+    }
+
+    return size;
+}
+#endif // USE_DRB

libs/libmythtv/DeviceReadBuffer.h

@@ +1 @@
+// -*- Mode: c++ -*-
+/* Device Buffer written by John Poet */
+
+#ifndef _DEVICEREADBUFFER_H_
+#define _DEVICEREADBUFFER_H_
+
+#include <unistd.h>
+#include <pthread.h>
+#include <sys/poll.h>
+
+#include <qmutex.h>
+#include <qwaitcondition.h>
+#include <qstring.h>
+
+#include "util.h"
+
+class ReaderPausedCB
+{
+  public:
+    virtual void ReaderPaused(int fd) = 0;
+};
+
+/** \class DeviceReadBuffer
+ *  \brief Buffers reads from device files.
+ *
+ *  This allows us to read the device regularly even in the presence
+ *  of long blocking conditions on writing to disk or accessing the
+ *  database.
+ */
+class DeviceReadBuffer
+{
+  public:
+    DeviceReadBuffer(ReaderPausedCB *callback, bool use_poll = true);
+   ~DeviceReadBuffer();
+
+    bool Setup(const QString &streamName, int streamfd);
+    void Teardown(void);
+
+    void Start(void);
+    void Reset(const QString &streamName, int streamfd);
+    void Stop(void);
+
+    void SetRequestPause(bool request);
+    bool IsPaused(void) const;
+    bool WaitForUnpause(int timeout);
+    
+    bool IsErrored(void) const { return error; }
+    bool IsEOF(void)     const { return eof;   }
+
+    uint Read(unsigned char *buf, uint count);
+
+  private:
+    static void *boot_ringbuffer(void *);
+    void fill_ringbuffer(void);
+
+    void SetPaused(bool);
+    void IncrWritePointer(uint len);
+    void IncrReadPointer(uint len);
+
+    bool HandlePausing(void);
+    bool Poll(void) const;
+    uint WaitForUnused(uint bytes_needed) const;
+    uint WaitForUsed  (uint bytes_needed) const;
+
+    bool IsPauseRequested(void) const;
+    bool IsOpen(void) const { return _stream_fd >= 0; }
+    uint GetUnused(void) const;
+    uint GetUsed(void) const;
+    uint GetContiguousUnused(void) const;
+
+    bool CheckForErrors(ssize_t read_len, uint &err_cnt);
+    void ReportStats(void);
+
+    QString          videodevice;
+    int              _stream_fd;
+
+    ReaderPausedCB  *readerPausedCB;
+    pthread_t        thread;
+
+    // Data for managing the device ringbuffer
+    mutable QMutex   lock;
+    bool             run;
+    bool             running;
+    bool             eof;
+    bool             error;
+    bool             request_pause;
+    bool             paused;
+    bool             using_poll;
+
+    size_t           size;
+    size_t           used;
+    size_t           dev_read_size;
+    size_t           min_read;
+    unsigned char   *buffer;
+    unsigned char   *readPtr;
+    unsigned char   *writePtr;
+    unsigned char   *endPtr;
+
+    QWaitCondition   pauseWait;
+    QWaitCondition   unpauseWait;
+
+    // statistics
+    size_t           max_used;
+    size_t           avg_used;
+    size_t           avg_cnt;
+    MythTimer        lastReport;
+};
+
+#endif // _DEVICEREADBUFFER_H_

libs/libmythtv/hdtvrecorder.h

@@ -3 +3 @@
  *  HDTVRecorder
  *  Copyright (c) 2003-2004 by Brandon Beattie, Doug Larrick, 
  *    Jason Hoos, and Daniel Thor Kristjansson
- *  Device ringbuffer added by John Poet
  *  Distributed as part of MythTV under GPL v2 and later.
  */
 
@@ -12 +11 @@
 
 #include "dtvrecorder.h"
 #include "tsstats.h"
+#include "DeviceReadBuffer.h"
 
 struct AVFormatContext;
 struct AVPacket;
@@ -28 +28 @@
  *
  *  \sa DTVRecorder, DVBRecorder
  */
-class HDTVRecorder : public DTVRecorder
+class HDTVRecorder : public DTVRecorder, private ReaderPausedCB
 {
     Q_OBJECT
     friend class ATSCStreamData;
     friend class TSPacketProcessor;
   public:
-    enum {report_loops = 20000};
-
     HDTVRecorder(TVRec *rec);
    ~HDTVRecorder();
 
@@ -47 +45 @@
     void StartRecording(void);
     void StopRecording(void);
 
-    void Pause(bool clear = false);
-    bool IsPaused(void) const;
-
     void Reset(void);
 
     bool Open(void);
@@ -61 +56 @@
     void deleteLater(void);
 
   private:
+    bool IsOpen(void) const { return _stream_fd >= 0; }
+    bool Close(void);
+
     void TeardownAll(void);
-    int ProcessData(unsigned char *buffer, int len);
+    uint ProcessDataTS(unsigned char *buffer, uint len);
     bool ProcessTSPacket(const TSPacket& tspacket);
     void HandleVideo(const TSPacket* tspacket);
     void HandleAudio(const TSPacket* tspacket);
 
     int ResyncStream(unsigned char *buffer, int curr_pos, int len);
 
-    static void *boot_ringbuffer(void *);
-    void fill_ringbuffer(void);
-    int ringbuf_read(unsigned char *buffer, size_t count);
+    void ReaderPaused(int fd);
+    bool PauseAndWait(int timeout = 100);
 
- private slots:
+    bool readchan(int chanfd, unsigned char* buffer, int dlen);
+    bool syncchan(int chanfd, int dlen, int keepsync);
+
+  private slots:
     void WritePAT(ProgramAssociationTable*);
     void WritePMT(ProgramMapTable*);
     void ProcessMGT(const MasterGuideTable*);
     void ProcessVCT(uint, const VirtualChannelTable*);
- private:
-    ATSCStreamData* _atsc_stream_data;
 
+  private:
+    ATSCStreamData   *_atsc_stream_data;
+    DeviceReadBuffer *_drb;
+
     // statistics
-    TSStats _ts_stats;
-    long long _resync_count;
-    size_t loop;
+    TSStats           _ts_stats;
+    long long         _resync_count;
 
-    // Data for managing the device ringbuffer
-    struct {
-        pthread_t        thread;
-        mutable pthread_mutex_t lock;
-        mutable pthread_mutex_t lock_stats;
-
-        bool             run;
-        bool             eof;
-        bool             error;
-        bool             request_pause;
-        bool             paused;
-        size_t           size;
-        size_t           used;
-        size_t           max_used;
-        size_t           avg_used;
-        size_t           avg_cnt;
-        size_t           dev_read_size;
-        size_t           min_read;
-        unsigned char  * buffer;
-        unsigned char  * readPtr;
-        unsigned char  * writePtr;
-        unsigned char  * endPtr;
-    } ringbuf;
+    /// unsynced packets to look at before giving up initially
+    static const uint INIT_SYNC_WINDOW_SIZE;
+    /// synced packets to require before starting recording
+    static const uint INIT_MIN_NUM_SYNC_PACKETS;
 };
 
 #endif

libs/libmythtv/hdtvrecorder.cpp

@@ -84 +84 @@
 #include "atsctables.h"
 #include "atscstreamdata.h"
 #include "tv_rec.h"
+#include "DeviceReadBuffer.h"
 
 // AVLib/FFMPEG includes
 #include "../libavcodec/avcodec.h"
 #include "../libavformat/avformat.h"
 #include "../libavformat/mpegts.h"
 
-#define REPORT_RING_STATS 1
+#define LOC QString("HDTVRec(%1):").arg(videodevice)
+#define LOC_ERR QString("HDTVRec(%1) Error:").arg(videodevice)
 
 #define DEFAULT_SUBCHANNEL 1
 
@@ -109 +111 @@
         };
 #endif
 
+const uint HDTVRecorder::INIT_SYNC_WINDOW_SIZE     = 50;
+const uint HDTVRecorder::INIT_MIN_NUM_SYNC_PACKETS = 10;
+
 HDTVRecorder::HDTVRecorder(TVRec *rec)
     : DTVRecorder(rec, "HDTVRecorder"), _atsc_stream_data(0), _resync_count(0)
 {
@@ -116 +121 @@
     connect(_atsc_stream_data, SIGNAL(UpdatePATSingleProgram(
                                           ProgramAssociationTable*)),
             this, SLOT(WritePAT(ProgramAssociationTable*)));
-    connect(_atsc_stream_data, SIGNAL(UpdatePMTSingleProgram(ProgramMapTable*)),
-            this, SLOT(WritePMT(ProgramMapTable*)));
+    connect(_atsc_stream_data,
+            SIGNAL(UpdatePMTSingleProgram(ProgramMapTable*)),
+            this,
+            SLOT(WritePMT(ProgramMapTable*)));
     connect(_atsc_stream_data, SIGNAL(UpdateMGT(const MasterGuideTable*)),
             this, SLOT(ProcessMGT(const MasterGuideTable*)));
     connect(_atsc_stream_data,
@@ -125 +132 @@
             this, SLOT(ProcessVCT(uint, const VirtualChannelTable*)));
 
     _buffer_size = TSPacket::SIZE * 1500;
-    if ((_buffer = new unsigned char[_buffer_size])) {
-        // make valgrind happy, initialize buffer memory
+    _buffer = new unsigned char[_buffer_size];
+
+    // make valgrind happy, initialize buffer memory
+    if (_buffer)
         memset(_buffer, 0xFF, _buffer_size);
-    }
 
-    VERBOSE(VB_RECORD, QString("HD buffer size %1 KB").arg(_buffer_size/1024));
+    VERBOSE(VB_RECORD, LOC +
+            QString("buffer size %1 KB").arg(_buffer_size/1024));
 
-    ringbuf.run = false;
-    ringbuf.buffer = 0;
-    pthread_mutex_init(&ringbuf.lock, NULL);
-    pthread_mutex_init(&ringbuf.lock_stats, NULL);
-    loop = random() % (report_loops / 2);
+    _drb = new DeviceReadBuffer(this);
 }
 
 void HDTVRecorder::TeardownAll(void)
 {
-    // Make SURE that the ringbuffer thread is cleaned up
+    // Make SURE that the device read thread is cleaned up -- John Poet
     StopRecording();
 
-    if (_stream_fd >= 0)
-    {
-        close(_stream_fd);
-        _stream_fd = -1;
-    }
+    Close();
+
     if (_atsc_stream_data)
     {
         delete _atsc_stream_data;
@@ -164 +166 @@
 HDTVRecorder::~HDTVRecorder()
 {
     TeardownAll();
-    pthread_mutex_destroy(&ringbuf.lock);
-    pthread_mutex_destroy(&ringbuf.lock_stats);
+    delete _drb;
 }
 
 void HDTVRecorder::deleteLater(void)
@@ -189 +190 @@
     SetOption("vbiformat", gContext->GetSetting("VbiFormat"));
 }
 
-bool HDTVRecorder::Open()
+bool HDTVRecorder::Open(void)
 {
     if (!_atsc_stream_data || !_buffer)
         return false;
 
 #if FAKE_VIDEO
     // open file instead of device
-    if (_stream_fd >=0 && close(_stream_fd))
-    {
-        VERBOSE(VB_IMPORTANT,
-                QString("HDTVRecorder::Open(): Error, failed to close "
-                        "existing fd (%1)").arg(strerror(errno)));
-        return false;
-    }
 
+    Close(); // close old video file
     _stream_fd = open(FAKE_VIDEO_FILES[fake_video_index], O_RDWR);
-    VERBOSE(VB_IMPORTANT, QString("Opened fake video source %1").arg(FAKE_VIDEO_FILES[fake_video_index]));
-    fake_video_index = (fake_video_index+1)%FAKE_VIDEO_NUM;
+
+    VERBOSE(VB_IMPORTANT, LOC_ERR + QString("Opened fake video source '%1'")
+            .arg(FAKE_VIDEO_FILES[fake_video_index]) + ENO);
+
+    fake_video_index = (fake_video_index + 1) % FAKE_VIDEO_NUM;
+
 #else
-    if (_stream_fd <= 0)
+    if (!IsOpen())
         _stream_fd = open(videodevice.ascii(), O_RDWR);
 #endif
-    if (_stream_fd <= 0)
+
+    if (!IsOpen())
     {
-        VERBOSE(VB_IMPORTANT, QString("Can't open video device: %1 chanfd = %2")
-                .arg(videodevice).arg(_stream_fd));
-        perror("open video:");
+        VERBOSE(VB_IMPORTANT, LOC_ERR +
+                QString("Couldn't open video device: '%1'")
+                .arg(videodevice) + ENO);
     }
-    return (_stream_fd>0);
+
+    return IsOpen();
 }
 
+bool HDTVRecorder::Close(void)
+{
+    if (IsOpen() && (0 != close(_stream_fd)))
+    {
+        VERBOSE(VB_IMPORTANT, LOC_ERR +
+                "Failed to close file descriptor." + ENO);
+        return false;
+    }
+    _stream_fd = -1;
+    return true;
+}
+
 void HDTVRecorder::SetStreamData(ATSCStreamData *stream_data)
 {
     if (stream_data == _atsc_stream_data)
@@ -231 +244 @@
         delete old_data;
 }
 
-bool readchan(int chanfd, unsigned char* buffer, int dlen) {
+bool HDTVRecorder::readchan(int chanfd, unsigned char* buffer, int dlen)
+{
     int len = read(chanfd, buffer, dlen); // read next byte
     if (dlen != len)
     {
         if (len < 0)
         {
-            VERBOSE(VB_IMPORTANT, QString("HD1 error reading from device"));
-            perror("read");
+            VERBOSE(VB_IMPORTANT, LOC_ERR +
+                    "Reading from device failed" + ENO);
         }
         else if (len == 0)
-            VERBOSE(VB_IMPORTANT, QString("HD2 end of file found in packet"));
+            VERBOSE(VB_IMPORTANT, LOC_ERR + "EOF found in TS packet");
         else 
-            VERBOSE(VB_IMPORTANT, QString("HD3 partial read. This shouldn't happen!"));
+            VERBOSE(VB_IMPORTANT, LOC_ERR +
+                    "Partial read during initial TS sync phase");
     }
     return (dlen == len);
 }
 
-bool syncchan(int chanfd, int dlen, int keepsync) {
+bool HDTVRecorder::syncchan(int chanfd, int dlen, int keepsync)
+{
     unsigned char b[188];
     int i, j;
-    for (i=0; i<dlen; i++) {
+    for (i=0; i<dlen; i++)
+    {
         if (!readchan(chanfd, b, 1))
             break;
         if (SYNC_BYTE == b[0])
         {
-            if (readchan(chanfd, &b[1], TSPacket::SIZE-1)) {
+            if (readchan(chanfd, &b[1], TSPacket::SIZE-1))
+            {
                 i += (TSPacket::SIZE - 1);
                 for (j=0; j<keepsync; j++)
                 {
@@ -268 +286 @@
                 }
                 if (j==keepsync)
                 {
-                    VERBOSE(VB_RECORD,
-                            QString("HD4 obtained device stream sync after reading %1 bytes").
-                            arg(dlen));
+                    VERBOSE(VB_RECORD, LOC + "Obtained TS sync, "+
+                            QString("after reading %1 bytes").arg(dlen));
                     return true;
                 }
                 continue;
@@ -278 +295 @@
             break;
         }
     }
-    VERBOSE(VB_IMPORTANT, QString("HD5 Error: could not obtain sync"));
+    VERBOSE(VB_IMPORTANT, LOC_ERR + "Could not obtain TS sync");
     return false;
 }
 
-void * HDTVRecorder::boot_ringbuffer(void * arg)
-{
-    HDTVRecorder *dtv = (HDTVRecorder *)arg;
-    dtv->fill_ringbuffer();
-    return NULL;
-}
+#define SR_CHK(MSG) \
+    if (!ok) { VERBOSE(VB_IMPORTANT, MSG); _error = true; return; }
 
-void HDTVRecorder::fill_ringbuffer(void)
+void HDTVRecorder::StartRecording(void)
 {
-    int       errcnt = 0;
-    int       len;
-    size_t    unused, used;
-    size_t    contiguous;
-    size_t    read_size;
-    bool      run, request_pause, paused;
+    bool ok        = true;
+    uint len       = 0;
+    uint remainder = 0;
 
-    pthread_mutex_lock(&ringbuf.lock);
-    ringbuf.run = true;
-    pthread_mutex_unlock(&ringbuf.lock);
+    VERBOSE(VB_RECORD, LOC + "StartRecording()");
 
-    for (;;)
-    {
-        pthread_mutex_lock(&ringbuf.lock);
-        run = ringbuf.run;
-        unused = ringbuf.size - ringbuf.used;
-        request_pause = ringbuf.request_pause;
-        paused = ringbuf.paused;
-        pthread_mutex_unlock(&ringbuf.lock);
+    ok = Open();
+    SR_CHK("Failed to open device.");
 
-        if (!run)
-            break;
+    ok = _drb->Setup(videodevice, _stream_fd);
+    SR_CHK("Failed to allocate device read buffer.");
 
-        if (request_pause)
-        {
-            pthread_mutex_lock(&ringbuf.lock);
-            ringbuf.paused = true;
-            pthread_mutex_unlock(&ringbuf.lock);
+    ok = syncchan(_stream_fd,
+                  INIT_SYNC_WINDOW_SIZE * TSPacket::SIZE,
+                  INIT_MIN_NUM_SYNC_PACKETS);
+    SR_CHK("Failed to sync to transport stream to valid packet.");
 
-            pauseWait.wakeAll();
-            if (tvrec)
-                tvrec->RecorderPaused();
+    _drb->Start();
 
-            usleep(1000);
-            continue;
-        }
-        else if (paused)
-        {
-            pthread_mutex_lock(&ringbuf.lock);
-            ringbuf.writePtr = ringbuf.readPtr = ringbuf.buffer;
-            ringbuf.used = 0;
-            ringbuf.paused = false;
-            pthread_mutex_unlock(&ringbuf.lock);
-        }
-
-        contiguous = ringbuf.endPtr - ringbuf.writePtr;
-
-        while (unused < TSPacket::SIZE && contiguous > TSPacket::SIZE)
-        {
-            usleep(500);
-
-            pthread_mutex_lock(&ringbuf.lock);
-            unused = ringbuf.size - ringbuf.used;
-            request_pause = ringbuf.request_pause;
-            pthread_mutex_unlock(&ringbuf.lock);
-
-            if (request_pause)
-                break;
-        }
-        if (request_pause)
-            continue;
-
-        read_size = unused > contiguous ? contiguous : unused;
-        if (read_size > ringbuf.dev_read_size)
-            read_size = ringbuf.dev_read_size;
-
-        len = read(_stream_fd, ringbuf.writePtr, read_size);
-
-        if (len < 0)
-        {
-            if (errno == EINTR)
-                continue;
-
-            VERBOSE(VB_IMPORTANT, QString("HD7 error reading from %1")
-                    .arg(videodevice));
-            perror("read");
-            if (++errcnt > 5)
-            {
-                pthread_mutex_lock(&ringbuf.lock);
-                ringbuf.error = true;
-                pthread_mutex_unlock(&ringbuf.lock);
-
-                break;
-            }
-
-            usleep(500);
-            continue;
-        }
-        else if (len == 0)
-        {
-            if (++errcnt > 5)
-            {
-                VERBOSE(VB_IMPORTANT, QString("HD8 %1 end of file found.")
-                        .arg(videodevice));
-
-                pthread_mutex_lock(&ringbuf.lock);
-                ringbuf.eof = true;
-                pthread_mutex_unlock(&ringbuf.lock);
-
-                break;
-            }
-            usleep(500);
-            continue;
-        }
-
-        errcnt = 0;
-
-        pthread_mutex_lock(&ringbuf.lock);
-        ringbuf.used += len;
-        used = ringbuf.used;
-        ringbuf.writePtr += len;
-        pthread_mutex_unlock(&ringbuf.lock);
-
-#ifdef REPORT_RING_STATS
-        pthread_mutex_lock(&ringbuf.lock_stats);
-
-        if (ringbuf.max_used < used)
-            ringbuf.max_used = used;
-
-        ringbuf.avg_used = ((ringbuf.avg_used * ringbuf.avg_cnt) + used)
-                           / ++ringbuf.avg_cnt;
-        pthread_mutex_unlock(&ringbuf.lock_stats);
-#endif
-
-        if (ringbuf.writePtr == ringbuf.endPtr)
-            ringbuf.writePtr = ringbuf.buffer;
-    }
-
-    close(_stream_fd);
-    _stream_fd = -1;
-}
-
-/* read count bytes from ring into buffer */
-int HDTVRecorder::ringbuf_read(unsigned char *buffer, size_t count)
-{
-    size_t          avail;
-    size_t          cnt = count;
-    size_t          min_read;
-    unsigned char  *cPtr = buffer;
-
-    bool            dev_error = false;
-    bool            dev_eof = false;
-
-    pthread_mutex_lock(&ringbuf.lock);
-    avail = ringbuf.used;
-    pthread_mutex_unlock(&ringbuf.lock);
-
-    min_read = cnt < ringbuf.min_read ? cnt : ringbuf.min_read;
-
-    while (min_read > avail)
-    {
-        usleep(50000);
-
-        if (request_pause || dev_error || dev_eof)
-            return 0;
-
-        pthread_mutex_lock(&ringbuf.lock);
-        dev_error = ringbuf.error;
-        dev_eof = ringbuf.eof;
-        avail = ringbuf.used;
-        pthread_mutex_unlock(&ringbuf.lock);
-    }
-    if (cnt > avail)
-        cnt = avail;
-
-    if (ringbuf.readPtr + cnt > ringbuf.endPtr)
-    {
-        size_t      len;
-
-        // Process as two pieces
-        len = ringbuf.endPtr - ringbuf.readPtr;
-        memcpy(cPtr, ringbuf.readPtr, len);
-        cPtr += len;
-        len = cnt - len;
-
-        // Wrap arround to begining of buffer
-        ringbuf.readPtr = ringbuf.buffer;
-        memcpy(cPtr, ringbuf.readPtr, len);
-        ringbuf.readPtr += len;
-    }
-    else
-    {
-        memcpy(cPtr, ringbuf.readPtr, cnt);
-        ringbuf.readPtr += cnt;
-    }
-
-    pthread_mutex_lock(&ringbuf.lock);
-    ringbuf.used -= cnt;
-    pthread_mutex_unlock(&ringbuf.lock);
-
-    if (ringbuf.readPtr == ringbuf.endPtr)
-        ringbuf.readPtr = ringbuf.buffer;
-    else
-    {
-#ifdef REPORT_RING_STATS
-        size_t samples, avg, max;
-
-        if (++loop == report_loops)
-        {
-            loop = 0;
-            pthread_mutex_lock(&ringbuf.lock_stats);
-            avg = ringbuf.avg_used;
-            samples = ringbuf.avg_cnt;
-            max = ringbuf.max_used;
-            ringbuf.avg_used = 0;
-            ringbuf.avg_cnt = 0;
-            ringbuf.max_used = 0;
-            pthread_mutex_unlock(&ringbuf.lock_stats);
-
-            VERBOSE(VB_IMPORTANT, QString("%1 ringbuf avg %2% max %3%"
-                                          " samples %4")
-                    .arg(videodevice)
-                    .arg((static_cast<double>(avg)
-                          / ringbuf.size) * 100.0)
-                    .arg((static_cast<double>(max)
-                          / ringbuf.size) * 100.0)
-                    .arg(samples));
-        }
-        else
-#endif
-            usleep(25);
-    }
-
-    return cnt;
-}
-
-void HDTVRecorder::StartRecording(void)
-{
-    bool            pause;
-    bool            dev_error, dev_eof;
-    int             len;
-
-    const int unsyncpackets = 50; // unsynced packets to look at before giving up
-    const int syncpackets   = 10; // synced packets to require before starting recording
-
-    VERBOSE(VB_RECORD, QString("StartRecording"));
-
-    if (!Open())
-    {
-        _error = true;        
-        return;
-    }
-
     _request_recording = true;
-    _recording = true;
+    _recording         = true;
 
-    // Setup device ringbuffer
-    delete[] ringbuf.buffer;
-
-//    ringbuf.size = 60 * 1024 * TSPacket::SIZE;
-    ringbuf.size = gContext->GetNumSetting("HDRingbufferSize", 50*188);
-    ringbuf.size *= 1024;
-
-    if ((ringbuf.buffer =
-         new unsigned char[ringbuf.size + TSPacket::SIZE]) == NULL)
+    // Process packets while recording is requested
+    while (_request_recording && !_error)
     {
-        VERBOSE(VB_IMPORTANT, "Failed to allocate HDTVRecorder ring buffer.");
-        _error = true;
-        return;
-    }
-
-    memset(ringbuf.buffer, 0xFF, ringbuf.size + TSPacket::SIZE);
-    ringbuf.endPtr = ringbuf.buffer + ringbuf.size;
-    ringbuf.readPtr = ringbuf.writePtr = ringbuf.buffer;
-    ringbuf.dev_read_size = TSPacket::SIZE * 48;
-    ringbuf.min_read = TSPacket::SIZE * 4;
-    ringbuf.used = 0;
-    ringbuf.max_used = 0;
-    ringbuf.avg_used = 0;
-    ringbuf.avg_cnt = 0;
-    ringbuf.request_pause = false;
-    ringbuf.paused = false;
-    ringbuf.error = false;
-    ringbuf.eof = false;
-
-    VERBOSE(VB_RECORD, QString("HD ring buffer size %1 KB")
-            .arg(ringbuf.size/1024));
-
-    // sync device stream so it starts with a valid ts packet
-    if (!syncchan(_stream_fd, TSPacket::SIZE*unsyncpackets, syncpackets))
-    {
-        _error = true;
-        return;
-    }
-
-    // create thread to fill the ringbuffer
-    pthread_create(&ringbuf.thread, NULL, boot_ringbuffer,
-                   reinterpret_cast<void *>(this));
-
-    int remainder = 0;
-    // TRANSFER DATA
-    while (_request_recording) 
-    {
-        pthread_mutex_lock(&ringbuf.lock);
-        dev_error = ringbuf.error;
-        dev_eof = ringbuf.eof;
-        pause = ringbuf.paused;
-        pthread_mutex_unlock(&ringbuf.lock);
-
-        if (request_pause)
-        {
-            pthread_mutex_lock(&ringbuf.lock);
-            ringbuf.request_pause = true;
-            pthread_mutex_unlock(&ringbuf.lock);
-
-            usleep(1000);
+        if (PauseAndWait())
             continue;
-        }
-        else if (pause)
-        {
-            pthread_mutex_lock(&ringbuf.lock);
-            ringbuf.request_pause = false;
-            pthread_mutex_unlock(&ringbuf.lock);
 
-            usleep(1500);
-            continue;
-        }
+        len = _drb->Read(&(_buffer[remainder]), _buffer_size - remainder);
 
-        if (dev_error)
-        {
-            VERBOSE(VB_IMPORTANT, "HDTV: device error detected");
-            _error = true;
-            break;
-        }
-
-        if (dev_eof)
-            break;
-
-        len = ringbuf_read(&(_buffer[remainder]), _buffer_size - remainder);
-
         if (len == 0)
             continue;
 
         len += remainder;
-        remainder = ProcessData(_buffer, len);
+        remainder = ProcessDataTS(_buffer, len);
         if (remainder > 0) // leftover bytes
             memmove(_buffer, &(_buffer[_buffer_size - remainder]),
                     remainder);
+
+        // Check for DRB errors
+        if (_drb->IsErrored())
+        {
+            VERBOSE(VB_IMPORTANT, LOC_ERR + "Device error detected");
+            _error = true;
+        }
+
+        if (_drb->IsEOF())
+        {
+            VERBOSE(VB_IMPORTANT, LOC_ERR + "Device EOF detected");
+            _error = true;
+        }
     }
 
     FinishRecording();
     _recording = false;
 }
+#undef SR_CHK
 
 void HDTVRecorder::StopRecording(void)
 {
@@ -649 +376 @@
 
     _request_recording = false;
 
-    pthread_mutex_lock(&ringbuf.lock);
-    bool run = ringbuf.run;
-    ringbuf.run = false;
-    pthread_mutex_unlock(&ringbuf.lock);
+    _drb->Stop();
 
-    if (run)
-        pthread_join(ringbuf.thread, NULL);
+    if (ok)
+        _drb->Teardown();
+    else // Better to have a memory leak, than a segfault? -- John Poet
+        VERBOSE(VB_IMPORTANT, LOC_ERR + "DeviceReadBuffer not cleaned up!");
 
-    if (!ok)
-    {
-        // Better to have a memory leak, then a segfault?
-        VERBOSE(VB_IMPORTANT, "DTV ringbuffer not cleaned up!\n");
-    }
-    else
-    {
-        delete[] ringbuf.buffer;
-        ringbuf.buffer = 0;
-    }
+    while (_recording)
+        usleep(2000);
+
     tvrec = rec;
 }
 
-void HDTVRecorder::Pause(bool /*clear*/)
+void HDTVRecorder::ReaderPaused(int /*fd*/)
 {
-    pthread_mutex_lock(&ringbuf.lock);
-    ringbuf.paused = false;
-    pthread_mutex_unlock(&ringbuf.lock);
-    request_pause = true;
+    pauseWait.wakeAll();
+    if (tvrec)
+        tvrec->RecorderPaused();
 }
 
-bool HDTVRecorder::IsPaused(void) const
+bool HDTVRecorder::PauseAndWait(int timeout)
 {
-    pthread_mutex_lock(&ringbuf.lock);
-    bool paused = ringbuf.paused;
-    pthread_mutex_unlock(&ringbuf.lock);
+#ifdef USE_DRB
+    if (request_pause)
+    {
+        paused = true;
+        if (!_drb->IsPaused())
+            _drb->SetRequestPause(true);
 
+        unpauseWait.wait(timeout);
+    }
+    else if (_drb->IsPaused())
+    {
+        _drb->SetRequestPause(false);
+        _drb->WaitForUnpause(timeout);
+        paused = _drb->IsPaused();
+    }
+    else
+    {
+        paused = false;
+    }
     return paused;
+#else // if !USE_DRB
+    return RecorderBase::PauseAndWait(timeout);
+#endif // !USE_DRB
 }
 
 int HDTVRecorder::ResyncStream(unsigned char *buffer, int curr_pos, int len)
@@ -695 +431 @@
     if (nextpos >= len)
         return -1; // not enough bytes; caller should try again
     
-    while (buffer[pos] != SYNC_BYTE || buffer[nextpos] != SYNC_BYTE) {
+    while (buffer[pos] != SYNC_BYTE || buffer[nextpos] != SYNC_BYTE)
+    {
         pos++;
         nextpos++;
         if (nextpos == len)
@@ -707 +444 @@
 
 void HDTVRecorder::WritePAT(ProgramAssociationTable *pat)
 {
+    if (!pat)
+        return;
+
     int next = (pat->tsheader()->ContinuityCounter()+1)&0xf;
     pat->tsheader()->SetContinuityCounter(next);
     ringBuffer->Write(pat->tsheader()->data(), TSPacket::SIZE);
 }
 
-#if WHACK_A_BUG_VIDEO
-static int WABV_base_pid     = 0x100;
-#define WABV_WAIT 60
-static int WABV_wait_a_while = WABV_WAIT;
-bool WABV_started = false;
-#endif
-
-#if WHACK_A_BUG_AUDIO
-static int WABA_base_pid     = 0x200;
-#define WABA_WAIT 60
-static int WABA_wait_a_while = WABA_WAIT;
-bool WABA_started = false;
-#endif
-
 void HDTVRecorder::WritePMT(ProgramMapTable* pmt)
 {
-    if (pmt) {
-        int next = (pmt->tsheader()->ContinuityCounter()+1)&0xf;
-        pmt->tsheader()->SetContinuityCounter(next);
+    if (!pmt)
+        return;
 
-#if WHACK_A_BUG_VIDEO
-        WABV_wait_a_while--;
-        if (WABV_wait_a_while<=0) {
-            WABV_started = true;
-            WABV_wait_a_while = WABV_WAIT;
-            WABV_base_pid = (((WABV_base_pid-0x100)+1)%32)+0x100;
-            if (StreamID::MPEG2Video != StreamData()->PMT()->StreamType(0))
-            {
-                VERBOSE(VB_IMPORTANT, "HDTVRecorder::WritePMT(): Error,"
-                        "Whack a Bug can not rewrite PMT, wrong stream type");
-            }
-            else
-            {
-                VERBOSE(VB_IMPORTANT, QString("Whack a Bug: new video pid %1").
-                        arg(WABV_base_pid));
-                // rewrite video pid
-                const uint old_video_pid=StreamData()->PMT()->StreamPID(0);
-                StreamData()->PMT()->SetStreamPID(0, WABV_base_pid);
-                if (StreamData()->PMT()->PCRPID() == old_video_pid)
-                    StreamData()->PMT()->SetPCRPID(WABV_base_pid);
-                StreamData()->PMT()->SetCRC(StreamData()->PMT()->CalcCRC());
-                VERBOSE(VB_IMPORTANT, StreamData()->PMT()->toString());
-            }
-        }
-#endif
-#if WHACK_A_BUG_AUDIO
-        WABA_wait_a_while--;
-        if (WABA_wait_a_while<=0) {
-            WABA_started = true;
-            WABA_wait_a_while = WABA_WAIT;
-            WABA_base_pid = (((WABA_base_pid-0x200)+1)%32)+0x200;
-            VERBOSE(VB_IMPORTANT, QString("Whack a Bug: new audio BASE pid %1").arg(WABA_base_pid));
-            // rewrite audio pids
-            for (uint i=0; i<StreamData()->PMT()->StreamCount(); i++) {
-                if (StreamID::MPEG2Audio == StreamData()->PMT()->StreamType(i) ||
-                    StreamID::MPEG2Audio == StreamData()->PMT()->StreamType(i)) {
-                    const uint old_audio_pid = StreamData()->PMT()->StreamPID(i);
-                    const uint new_audio_pid = WABA_base_pid + old_audio_pid;
-                    StreamData()->PMT()->SetStreamPID(i, new_audio_pid);
-                    if (StreamData()->PMT()->PCRPID() == old_audio_pid)
-                        StreamData()->PMT()->SetPCRPID(new_audio_pid);
-                    StreamData()->PMT()->SetCRC(StreamData()->PMT()->CalcCRC());
-                    VERBOSE(VB_IMPORTANT, StreamData()->PMT()->toString());
-                }
-            }
-        }
-#endif
-
-        ringBuffer->Write(pmt->tsheader()->data(), TSPacket::SIZE);
-    }
+    int next = (pmt->tsheader()->ContinuityCounter()+1)&0xf;
+    pmt->tsheader()->SetContinuityCounter(next);
+    ringBuffer->Write(pmt->tsheader()->data(), TSPacket::SIZE);
 }
 
 /** \fn HDTVRecorder::ProcessMGT(const MasterGuideTable*)
@@ -820 +498 @@
         {
             if (vct->ProgramNumber(i) != (uint)StreamData()->DesiredProgram())
             {
-                VERBOSE(VB_RECORD, 
-                        QString("Resetting desired program from %1"
-                                " to %2")
+                VERBOSE(VB_RECORD, LOC_ERR +
+                        QString("Resetting desired program from %1 to %2")
                         .arg(StreamData()->DesiredProgram())
                         .arg(vct->ProgramNumber(i)));
                 // Do a (partial?) reset here if old desired
@@ -834 +511 @@
     }
     if (!found)
     {
-        VERBOSE(VB_IMPORTANT, 
+        VERBOSE(VB_IMPORTANT, LOC_ERR + 
                 QString("Desired channel %1_%2 not found;"
                         " using %3_%4 instead.")
                 .arg(StreamData()->DesiredMajorChannel())
                 .arg(StreamData()->DesiredMinorChannel())
                 .arg(vct->MajorChannel(0))
-                .arg(vct->MinorChannel(0)));
-        VERBOSE(VB_IMPORTANT, vct->toString());
+                .arg(vct->MinorChannel(0)) + "\n" + vct->toString());
+
         StreamData()->SetDesiredProgram(vct->ProgramNumber(0));
     }
 }
@@ -854 +531 @@
     if (_wait_for_keyframe && !_keyframe_seen)
         return;
 
-#if WHACK_A_BUG_VIDEO
-    if (WABV_started)
-        ((TSPacket*)(tspacket))->SetPID(WABV_base_pid);
-#endif
-
     ringBuffer->Write(tspacket->data(), TSPacket::SIZE);
 }
 
@@ -868 +540 @@
     if (_wait_for_keyframe && !_keyframe_seen)
         return;
 
-#if WHACK_A_BUG_AUDIO
-    if (WABA_started)
-        ((TSPacket*)(tspacket))->SetPID(WABA_base_pid+tspacket->PID());
-#endif
-
     ringBuffer->Write(tspacket->data(), TSPacket::SIZE);
 }
 
@@ -902 +569 @@
     return ok;
 }
 
-int HDTVRecorder::ProcessData(unsigned char *buffer, int len)
+uint HDTVRecorder::ProcessDataTS(unsigned char *buffer, uint len)
 {
-    int pos = 0;
+    if (len < TSPacket::SIZE)
+        return len;
 
-    while (pos + 187 < len) // while we have a whole packet left
+    uint pos = 0;
+    uint end = len - TSPacket::SIZE;
+    while (pos <= end) // while we have a whole packet left
     {
         if (buffer[pos] != SYNC_BYTE)
         {
             _resync_count++;
-            if (25 == _resync_count) 
-                VERBOSE(VB_RECORD, QString("Resyncing many of times, suppressing error messages"));
+
+            if (25 == _resync_count)
+            {
+                VERBOSE(VB_RECORD, LOC + "Resyncing many of times, "
+                        "suppressing error messages");
+            }
             else if (25 > _resync_count)
-                VERBOSE(VB_RECORD, QString("Resyncing"));
+            {
+                VERBOSE(VB_RECORD, LOC + "Resyncing");
+            }
+
             int newpos = ResyncStream(buffer, pos, len);
             if (newpos == -1)
                 return len - pos;
@@ -925 +602 @@
         }
 
         const TSPacket *pkt = reinterpret_cast<const TSPacket*>(&buffer[pos]);
-        if (ProcessTSPacket(*pkt)) {
-            pos += TSPacket::SIZE; // Advance to next TS packet
+        if (ProcessTSPacket(*pkt))
+        {
+            // Advance to next TS packet
+            pos += TSPacket::SIZE;
+
+            // Take care of statistics
             _ts_stats.IncrTSPacketCount();
-            if (0 == _ts_stats.TSPacketCount()%1000000)
-                VERBOSE(VB_RECORD, _ts_stats.toString());
-        } else // Let it resync in case of dropped bytes
-            buffer[pos] = SYNC_BYTE+1;
+            if (0 == _ts_stats.TSPacketCount() % 1000000)
+                VERBOSE(VB_RECORD, LOC + "\n" + _ts_stats.toString());
+
+        }
+        else
+        {
+            pos++; // Resync on invalid packet, in case of dropped bytes...
+        }
     }
 
     return len - pos;
@@ -939 +624 @@
 
 void HDTVRecorder::Reset(void)
 {
-    VERBOSE(VB_RECORD, "HDTVRecorder::Reset(void)");
+    VERBOSE(VB_RECORD, LOC + "Reset(void)");
     DTVRecorder::Reset();
 
     _error = false;
@@ -947 +632 @@
     _ts_stats.Reset();
 
     if (curRecording)
-    {
         curRecording->ClearPositionMap(MARK_GOP_BYFRAME);
+
+    if (!IsOpen())
+        return /* true */;
+
+    if (!IsPaused())
+    {
+        Pause();
+        WaitForPause();
     }
 
-    if (_stream_fd >= 0) 
+    if (!Close())
+        return /* false */;
+
+    if (Open())
     {
-        if (!IsPaused())
-        {
-            Pause();
-            WaitForPause();
-        }
-        int ret = close(_stream_fd);
-        if (ret < 0) 
-        {
-            perror("close");
-            return;
-        }
-#if FAKE_VIDEO
-        // open file instead of device
-        _stream_fd = open(FAKE_VIDEO_FILES[fake_video_index], O_RDWR);
-        VERBOSE(VB_IMPORTANT, QString("Opened fake video source %1").arg(FAKE_VIDEO_FILES[fake_video_index]));
-        fake_video_index = (fake_video_index+1)%FAKE_VIDEO_NUM;
-#else
-        _stream_fd = open(videodevice.ascii(), O_RDWR);
-#endif
-        if (_stream_fd < 0)
-        {
-            VERBOSE(VB_IMPORTANT, QString("HD1 Can't open video device: %1 chanfd = %2").
-                    arg(videodevice).arg(_stream_fd));
-            perror("open video");
-            return;
-        }
-        else
-        {
-            pthread_mutex_lock(&ringbuf.lock);
-            ringbuf.used = 0;
-            ringbuf.max_used = 0;
-            ringbuf.readPtr = ringbuf.writePtr = ringbuf.buffer;
-            pthread_mutex_unlock(&ringbuf.lock);
-        }
+        _drb->Reset(videodevice, _stream_fd);
         Unpause();
+        return /* true */;
     }
+
+    VERBOSE(VB_IMPORTANT, LOC_ERR + "Couldn't open video device: " +
+            QString("'%1'").arg(videodevice) + ENO);
+    return /* false */;
 }

libs/libmythtv/libmythtv.pro

@@ -206 +206 @@
 
     # TVRec & Recorder base classes
     HEADERS += tv_rec.h
-    HEADERS += recorderbase.h
+    HEADERS += recorderbase.h              DeviceReadBuffer.h
     HEADERS += dtvrecorder.h               dummydtvrecorder.h
     SOURCES += tv_rec.cpp
-    SOURCES += recorderbase.cpp
+    SOURCES += recorderbase.cpp            DeviceReadBuffer.cpp
     SOURCES += dtvrecorder.cpp             dummydtvrecorder.cpp
 
     # MPEG parsing stuff

libs/libmythtv/dvbrecorder.h

@@ +1 @@
+// -*- Mode: c++ -*-
 /*
  *  Copyright (C) Kenneth Aafloy 2003
  *  
@@ -17 +18 @@
 #include "dtvrecorder.h"
 #include "tspacket.h"
 #include "transform.h"
+#include "DeviceReadBuffer.h"
 
 #include "dvbtypes.h"
 #include "dvbchannel.h"
@@ -25 +27 @@
 class ProgramAssociationTable;
 class ProgramMapTable;
 
+class PIDInfo
+{
+  public:
+    PIDInfo() :
+        filter_fd(-1),  continuityCount(0xFF), ip(NULL),
+        isVideo(false), isEncrypted(false),    payloadStartSeen(false) {;}
+
+    int    filter_fd;         ///< Input filter file descriptor
+    uint   continuityCount;   ///< last Continuity Count (sentinel 0xFF)
+    ipack *ip;                ///< TS->PES converter
+    bool   isVideo;
+    bool   isEncrypted;       ///< true if PID is marked as encrypted
+    bool   payloadStartSeen;  ///< true if payload start packet seen on PID
+
+    inline void Close(void);
+    inline bool CheckCC(uint cc);
+};
+typedef QMap<uint,PIDInfo*> PIDInfoMap;
+
 /** \class DVBRecorder
  *  \brief This is a specialization of DTVRecorder used to
  *         handle streams from DVB drivers.
  *
  *  \sa DTVRecorder, HDTVRecorder
  */
-class DVBRecorder: public DTVRecorder
+class DVBRecorder: public DTVRecorder, private ReaderPausedCB
 {
     Q_OBJECT
   public:
@@ -47 +68 @@
 
     void StartRecording(void);
     void Reset(void);
+    void StopRecording(void);
 
     bool Open(void);
+    bool IsOpen(void) const { return _stream_fd >= 0; }
     void Close(void);
 
     bool RecordsTransportStream(void) const
@@ -63 +86 @@
 
   private:
     void TeardownAll(void);
-    void ReadFromDMX(void);
-    static void ProcessDataPS(unsigned char *buffer, int len, void *priv);
-    void LocalProcessDataPS(unsigned char *buffer, int len);
-    void LocalProcessDataTS(unsigned char *buffer, int len);
 
+    bool Poll(void) const;
+
+    uint ProcessDataTS(unsigned char *buffer, uint len);
+    bool ProcessTSPacket(const TSPacket& tspacket);
+
+    void AutoPID(void);
+    bool OpenFilters(void);
     void CloseFilters(void);
     void OpenFilter(uint pid, ES_Type type, dmx_pes_type_t pes_type,
                     uint mpeg_stream_type);
-    bool SetDemuxFilters(void);
-    void AutoPID(void);
 
     void SetPAT(ProgramAssociationTable*);
     void SetPMT(ProgramMapTable*);
 
     void CreatePAT(void);
     void CreatePMT(void);
+    void WritePATPMT(void);
 
     void DebugTSHeader(unsigned char* buffer, int len);
 
+    void ReaderPaused(int fd);
+    bool PauseAndWait(int timeout = 100);
+
+    ipack *CreateIPack(ES_Type type);
+    void ProcessDataPS(unsigned char *buffer, uint len);
+    static void process_data_ps_cb(unsigned char*,int,void*);
+
+    DeviceReadBuffer *_drb;
+
     // Options set in SetOption()
     int             _card_number_option;
     bool            _record_transport_stream_option;
@@ -90 +124 @@
     // DVB stuff
     DVBChannel*     dvbchannel;
 
+    // general recorder stuff
+    /// Set when we want to generate a new filter set
+    bool            _reset_pid_filters;
+    QMutex          _pid_lock;
+    PIDInfoMap      _pid_infos;
+
     // PS recorder stuff
     int             _ps_rec_audio_id;
     int             _ps_rec_video_id;
     unsigned char   _ps_rec_buf[3];
-    pid_ipack_t     _ps_rec_pid_ipack;
 
     // TS recorder stuff
     ProgramAssociationTable *_pat;
     ProgramMapTable         *_pmt;
     uint            _next_pmt_version;
     uint            _ts_packets_until_psip_sync;
-    QMap<uint,bool> _payload_start_seen;
-    QMap<uint,bool> _videoPID;
 
     // Input Misc
     /// PMT on input side
     PMTObject       _input_pmt;
-    /// Input filter file descriptors
-    vector<int>     _pid_filters;
-    /// Input polling structure for _stream_fd
-    struct pollfd   _polls;
-    /// Set when we want to generate a new filter set
-    bool            _reset_pid_filters;
-    /// Encrypted PID, so we can drop these
-    QMap<uint,bool> _encrypted_pid;
 
-    // locking
-    QMutex          _pid_lock;
-
     // Statistics
-    uint            _continuity_error_count;
-    uint            _stream_overflow_count;
-    uint            _bad_packet_count;
-    QMap<uint,int>  _continuity_count;
+    mutable uint        _continuity_error_count;
+    mutable uint        _stream_overflow_count;
+    mutable uint        _bad_packet_count;
+    mutable MythTimer   _poll_timer;
 
-    // For debugging
-    bool data_found; ///< debugging variable used by transform.c
-    bool keyframe_found;
-
     // Constants
     static const int PMT_PID;
     static const int TSPACKETS_BETWEEN_PSIP_SYNC;
@@ -136 +158 @@
     static const int POLL_WARNING_TIMEOUT;
 };
 
+inline void PIDInfo::Close(void)
+{
+    if (filter_fd >= 0)
+        close(filter_fd);
+
+    if (ip)
+    {
+        free_ipack(ip);
+        free(ip);
+    }
+}
+
+inline bool PIDInfo::CheckCC(uint new_cnt)
+{
+    if (continuityCount == 0xFF)
+    {
+        continuityCount = new_cnt;
+        return true;
+    }
+
+    continuityCount = (continuityCount+1) & 0xf;
+    if (continuityCount == new_cnt)
+        return true;
+    
+    continuityCount = new_cnt;
+    return false;
+}
+
 #endif

libs/libmythtv/DeviceReadBuffer.cpp

@@ +1 @@
+#include <algorithm>
+#include <cassert>
+
+#include "DeviceReadBuffer.h"
+#include "mythcontext.h"
+#include "tspacket.h"
+
+#define REPORT_RING_STATS 1
+
+#define LOC QString("DevRdB(%1): ").arg(videodevice)
+#define LOC_ERR QString("DevRdB(%1) Error: ").arg(videodevice)
+
+DeviceReadBuffer::DeviceReadBuffer(ReaderPausedCB *cb, bool use_poll)
+    : videodevice(QString::null),   _stream_fd(-1),
+      readerPausedCB(cb),
+
+      // Data for managing the device ringbuffer
+      run(false),                   running(false),
+      eof(false),                   error(false),
+      request_pause(false),         paused(false),
+      using_poll(use_poll),
+
+      size(0),                      used(0),
+      dev_read_size(0),             min_read(0),
+
+      buffer(NULL),                 readPtr(NULL),
+      writePtr(NULL),               endPtr(NULL),
+
+      // statistics
+      max_used(0),                  avg_used(0),
+      avg_cnt(0)
+{
+}
+
+DeviceReadBuffer::~DeviceReadBuffer()
+{
+}
+
+bool DeviceReadBuffer::Setup(const QString &streamName, int streamfd)
+{
+    QMutexLocker locker(&lock);
+
+    if (buffer)
+        delete[] buffer;
+
+    videodevice   = streamName;
+    _stream_fd    = streamfd;
+
+    // Setup device ringbuffer
+    eof           = false;
+    error         = false;
+    request_pause = false;
+    paused        = false;
+
+    size          = gContext->GetNumSetting("HDRingbufferSize",
+                                            50 * TSPacket::SIZE) * 1024;
+    used          = 0;
+    dev_read_size = TSPacket::SIZE * (using_poll ? 256 : 48);
+    min_read      = TSPacket::SIZE * 4;
+
+    buffer        = new unsigned char[size + TSPacket::SIZE];
+    readPtr       = buffer;
+    writePtr      = buffer;
+    endPtr        = buffer + size;
+
+    // Initialize buffer, if it exists
+    if (!buffer)
+        return false;
+    memset(buffer, 0xFF, size + TSPacket::SIZE);
+
+    // Initialize statistics
+    max_used      = 0;
+    avg_used      = 0;
+    avg_cnt       = 0;
+    lastReport.start();
+
+    VERBOSE(VB_RECORD, LOC + QString("buffer size %1 KB").arg(size/1024));
+
+    return true;
+}
+
+void DeviceReadBuffer::Teardown(void)
+{
+    if (buffer)
+        delete[] buffer;
+    buffer = NULL;
+}
+
+void DeviceReadBuffer::Start(void)
+{
+    lock.lock();
+    bool was_running = running;
+    lock.unlock();
+    if (was_running)
+    {
+        VERBOSE(VB_IMPORTANT, LOC_ERR + "Start(): Already running.");
+        SetRequestPause(false);
+        return;
+    }
+
+    pthread_create(&thread, NULL, boot_ringbuffer, this);
+}
+
+void DeviceReadBuffer::Reset(const QString &streamName, int streamfd)
+{
+    QMutexLocker locker(&lock);
+
+    videodevice   = streamName;
+    _stream_fd    = streamfd;
+
+    used          = 0;
+    readPtr       = buffer;
+    writePtr      = buffer;
+}
+
+void DeviceReadBuffer::Stop(void)
+{
+    lock.lock();
+    bool was_running = running;
+    run = false;
+    lock.unlock();
+
+    if (!was_running)
+    {
+        VERBOSE(VB_IMPORTANT, LOC_ERR + "Stop(): Not running.");
+        return;
+    }
+
+    pthread_join(thread, NULL);
+}
+
+void DeviceReadBuffer::SetRequestPause(bool req)
+{
+    QMutexLocker locker(&lock);
+    request_pause = req;
+}
+
+void DeviceReadBuffer::SetPaused(bool val)
+{
+    lock.lock();
+    paused = val;
+    lock.unlock();
+    if (val)
+        pauseWait.wakeAll();
+    else
+        unpauseWait.wakeAll();
+}
+
+bool DeviceReadBuffer::IsPaused(void) const
+{
+    QMutexLocker locker(&lock);
+    return paused;
+}
+
+bool DeviceReadBuffer::WaitForUnpause(int timeout)
+{
+    if (IsPaused())
+        unpauseWait.wait(timeout);
+    return IsPaused();
+}
+
+bool DeviceReadBuffer::IsPauseRequested(void) const
+{
+    QMutexLocker locker(&lock);
+    return request_pause;
+}
+
+uint DeviceReadBuffer::GetUnused(void) const
+{
+    QMutexLocker locker(&lock);
+    return size - used;
+}
+
+uint DeviceReadBuffer::GetUsed(void) const
+{
+    QMutexLocker locker(&lock);
+    return used;
+}
+
+uint DeviceReadBuffer::GetContiguousUnused(void) const
+{
+    QMutexLocker locker(&lock);
+    return endPtr - writePtr;
+}
+
+void DeviceReadBuffer::IncrWritePointer(uint len)
+{
+    QMutexLocker locker(&lock);
+    used     += len;
+    writePtr += len;
+    writePtr  = (writePtr == endPtr) ? buffer : writePtr;
+#ifdef REPORT_RING_STATS
+    max_used = max(used, max_used);
+    avg_used = ((avg_used * avg_cnt) + used) / ++avg_cnt;
+#endif
+}
+
+void DeviceReadBuffer::IncrReadPointer(uint len)
+{
+    QMutexLocker locker(&lock);
+    used    -= len;
+    readPtr += len;
+    readPtr  = (readPtr == endPtr) ? buffer : readPtr;
+    assert(readPtr <= endPtr);
+}
+
+void *DeviceReadBuffer::boot_ringbuffer(void *arg)
+{
+    ((DeviceReadBuffer*) arg)->fill_ringbuffer();
+    return NULL;
+}
+
+void DeviceReadBuffer::fill_ringbuffer(void)
+{
+    uint      errcnt = 0;
+
+    lock.lock();
+    run     = true;
+    running = true;
+    lock.unlock();
+
+    while (run)
+    {
+        if (!HandlePausing())
+            continue;
+
+        if (!IsOpen())
+        {
+            usleep(10000);
+            continue;
+        }
+
+        if (using_poll && !Poll())
+            continue;
+
+        // Limit read size for faster return from read
+        size_t read_size = min(dev_read_size, WaitForUnused(TSPacket::SIZE));
+
+        // if read_size > 0 do the read...
+        if (read_size)
+        {
+            ssize_t len = read(_stream_fd, writePtr, read_size);
+            if (!CheckForErrors(len, errcnt))
+            {
+                if (errcnt > 5)
+                    break;
+                else
+                    continue;
+            }
+            errcnt = 0;
+            IncrWritePointer(len);
+        }
+    }
+
+    lock.lock();
+    running = false;
+    lock.unlock();
+}
+
+bool DeviceReadBuffer::HandlePausing(void)
+{
+    if (IsPauseRequested())
+    {
+        SetPaused(true);
+
+        if (readerPausedCB)
+            readerPausedCB->ReaderPaused(_stream_fd);
+
+        usleep(5000);
+        return false;
+    }
+    else if (IsPaused())
+    {
+        Reset(videodevice, _stream_fd);
+        SetPaused(false);
+    }
+    return true;
+}
+
+bool DeviceReadBuffer::Poll(void) const
+{
+    bool retval = true;
+    while (true)
+    {
+        struct pollfd polls;
+        polls.fd      = _stream_fd;
+        polls.events  = POLLIN;
+        polls.revents = 0;
+
+        int ret = poll(&polls, 1 /*number of polls*/, 10 /*msec*/);
+        if (IsPauseRequested() || !IsOpen() || !run)
+        {
+            retval = false;
+            break; // are we supposed to pause, stop, etc.
+        }
+
+        if (ret > 0)
+            break; // we have data to read :)
+        if ((-1 == ret) && (EOVERFLOW == errno))
+            break; // we have an error to handle
+
+        if ((-1 == ret) && ((EAGAIN == errno) || (EINTR  == errno)))
+            continue; // errors that tell you to try again
+        if (ret == 0)
+            continue; // timed out, try again
+
+        usleep(2500);
+    }
+    return retval;
+}
+
+bool DeviceReadBuffer::CheckForErrors(ssize_t len, uint &errcnt)
+{
+    if (len < 0)
+    {
+        if (EINTR == errno)
+            return false;
+        if (EAGAIN == errno)
+        {
+            usleep(2500);
+            return false;
+        }
+        if (EOVERFLOW == errno)
+        {
+            VERBOSE(VB_IMPORTANT, LOC_ERR + "Driver buffers overflowed");
+            return false;
+        }
+
+        VERBOSE(VB_IMPORTANT, LOC_ERR +
+                QString("Problem reading fd(%1)").arg(_stream_fd) + ENO);
+
+        if (++errcnt > 5)
+        {
+            lock.lock();
+            error = true;
+            lock.unlock();
+            return false;
+        }
+
+        usleep(500);
+        return false;
+    }
+    else if (len == 0)
+    {
+        if (++errcnt > 5)
+        {
+            VERBOSE(VB_IMPORTANT, LOC +
+                    QString("End-Of-File? fd(%1)").arg(_stream_fd));
+
+            lock.lock();
+            eof = true;
+            lock.unlock();
+
+            return false;
+        }
+        usleep(500);
+        return false;
+    }
+    return true;
+}
+
+/** \fn DeviceReadBuffer::Read(unsigned char*, uint)
+ *  \brief Try to Read count bytes from into buffer
+ *  \param buffer Buffer to put data in
+ *  \param count  Number of bytes to attempt to read
+ *  \return number of bytes actually read
+ */
+uint DeviceReadBuffer::Read(unsigned char *buf, const uint count)
+{
+    uint avail = WaitForUsed(min(count, min_read));
+    size_t cnt = min(count, avail);
+
+    if (!cnt)
+        return 0;
+
+    if (readPtr + cnt > endPtr)
+    {
+        // Process as two pieces
+        size_t len = endPtr - readPtr;
+        if (len)
+        {
+            memcpy(buf, readPtr, len);
+            buf += len;
+            IncrReadPointer(len);
+        }
+        if (cnt > len)
+        {
+            len = cnt - len;
+            memcpy(buf, readPtr, len);
+            IncrReadPointer(len);
+        }
+    }
+    else
+    {
+        memcpy(buf, readPtr, cnt);
+        IncrReadPointer(cnt);
+    }
+
+    ReportStats();
+
+    return cnt;
+}
+
+/// \return bytes available for writing
+uint DeviceReadBuffer::WaitForUnused(uint needed) const
+{
+    size_t unused = GetUnused();
+    size_t contig = GetContiguousUnused();
+
+    if (contig > TSPacket::SIZE)
+    {
+        while (unused < needed)
+        {
+            unused = GetUnused();
+            if (IsPauseRequested() || !IsOpen() || !run)
+                return 0;
+            usleep(5000);
+        }
+        if (IsPauseRequested() || !IsOpen() || !run)
+            return 0;
+        contig = GetContiguousUnused();
+    }
+
+    return min(contig, unused);
+}
+
+/// \return bytes available for reading
+uint DeviceReadBuffer::WaitForUsed(uint needed) const
+{
+    size_t avail = GetUsed();
+    while (needed > avail)
+    {
+        {
+            QMutexLocker locker(&lock);
+            avail = used;
+            if (request_pause || error || eof)
+                return 0;
+        }
+        usleep(5000);
+    }
+    return avail;
+}
+
+void DeviceReadBuffer::ReportStats(void)
+{
+#ifdef REPORT_RING_STATS
+    if (lastReport.elapsed() > 20*1000 /* msg every 20 seconds */)
+    {
+        QMutexLocker locker(&lock);
+        double rsize = 100.0 / size;
+        QString msg  = QString("fill avg(%1%) ").arg(avg_used*rsize,3,'f',0);
+        msg         += QString("fill max(%2%) ").arg(max_used*rsize,3,'f',0);
+        msg         += QString("samples(%3)").arg(avg_cnt);
+
+        avg_used    = 0;
+        avg_cnt     = 0;
+        max_used    = 0;
+        lastReport.start();
+
+        VERBOSE(VB_IMPORTANT, LOC + msg);
+    }
+#endif
+}