Ticket #712: drb-v2.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)
@@ -122 +132 @@
         _buffer = NULL;
     }
 
+    if (_drb)
+    {
+        delete _drb;
+        _drb = NULL;
+    }
+
     SetPAT(NULL);
     SetPMT(NULL);
 }
@@ -177 +193 @@
 
 bool DVBRecorder::Open(void)
 {
-    if (_stream_fd >= 0)
+    if (IsOpen())
     {
         VERBOSE(VB_GENERAL, LOC_WARN + "Card already open");
         return true;
@@ -185 +201 @@
 
     _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;
     }
 
-    _polls.fd = _stream_fd;
-    _polls.events = POLLIN;
-    _polls.revents = 0;
+#ifdef USE_DRB
+    if (_drb)
+        _drb->Reset(videodevice, _stream_fd);
+#endif
 
     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 +228 @@
 {
     VERBOSE(VB_RECORD, LOC + "Close() fd("<<_stream_fd<<") -- begin");
 
-    if (_stream_fd < 0)
-        return;
+    if (IsOpen())
+    {
 
-    CloseFilters();
+#ifdef USE_DRB
+    if (_drb)
+        _drb->Reset(videodevice, -1);
+#endif
 
-    if (_stream_fd >= 0)
+        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 +296 @@
         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 +315 @@
         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);
+    info->filter_fd = 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;
-    }
-
     // 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 +401 @@
                     (*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 +410 @@
     return true;
 }
 
-/*
- *  Process PMT and decide which components should be recorded
- */
-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 stream types:
-    QValueList<ES_Type> StreamTypes;
-    StreamTypes += ES_TYPE_VIDEO_MPEG1;
-    StreamTypes += ES_TYPE_VIDEO_MPEG2;
-    StreamTypes += ES_TYPE_AUDIO_MPEG1;
-    StreamTypes += ES_TYPE_AUDIO_MPEG2;
-    StreamTypes += ES_TYPE_AUDIO_AC3;
-    if (_record_transport_stream_option)
-    {
-        // The following types are only supported with TS recording
-        StreamTypes += ES_TYPE_AUDIO_DTS;
-        StreamTypes += ES_TYPE_AUDIO_AAC;
-        StreamTypes += ES_TYPE_TELETEXT;
-        StreamTypes += ES_TYPE_SUBTITLE;
-    }
-
-    QMap<ES_Type, bool> flagged;
-    QValueList<ElementaryPIDObject>::Iterator es;
-
-    if (!_hw_decoder_option)
-    {
-        for (es = _input_pmt.Components.begin();
-             es != _input_pmt.Components.end(); ++es)
-        {
-            (*es).Record = true;
-            flagged[(*es).Type] = true;
-        }
-    }
-    else
-    {
-        // Wanted languages:
-        QStringList Languages = iso639_get_language_list();
-
-        for (es = _input_pmt.Components.begin();
-             es != _input_pmt.Components.end(); ++es)
-        {
-            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
-            }
-
-            // 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;
-
-                case ES_TYPE_AUDIO_MPEG1: 
-                case ES_TYPE_AUDIO_MPEG2:
-                case ES_TYPE_AUDIO_AC3:
-                case ES_TYPE_AUDIO_DTS:
-                case ES_TYPE_AUDIO_AAC:
-                    if (flagged.contains(ES_TYPE_AUDIO_MPEG1) ||
-                        flagged.contains(ES_TYPE_AUDIO_MPEG2) ||
-                        flagged.contains(ES_TYPE_AUDIO_AC3)   ||
-                        flagged.contains(ES_TYPE_AUDIO_DTS)   ||
-                        flagged.contains(ES_TYPE_AUDIO_AAC))
-                    {
-                        continue; // Ignore this stream
-                    }
-                    break;
-
-                default:
-                    break;
-            }
-            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;
-            }
-        }
-    }
-
-    for (es = _input_pmt.Components.begin();
-         es != _input_pmt.Components.end(); ++es)
-    {
-        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));
-        }
-    }
-
-    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);
-}
-
 void DVBRecorder::StartRecording(void)
 {
     if (!Open())
@@ -608 +429 @@
     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(5000);
             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->IsRunning())
+        _drb->Stop();
+#endif        
+
     Close();
 
     FinishRecording();
@@ -676 +502 @@
     _recording = false;
 }
 
+#if 0
 void DVBRecorder::ReadFromDMX(void)
 {
     int cardnum = _card_number_option;
@@ -958 +785 @@
 
     ringBuffer->Write(buffer,len);
 }
+#endif
 
 void DVBRecorder::Reset(void)
 {
@@ -1064 +892 @@
     SetPMT(pmt);
 }
 
+void DVBRecorder::WritePATPMT(void)
+{
+    if (_ts_packets_until_psip_sync <= 0)
+    {
+        QMutexLocker read_lock(&_pid_lock);
+        uint next_cc;
+        if (_pat && _pmt && ringBuffer)
+        {
+            next_cc = (_pat->tsheader()->ContinuityCounter()+1)&0xf;
+            _pat->tsheader()->SetContinuityCounter(next_cc);
+            ringBuffer->Write(_pat->tsheader()->data(), TSPacket::SIZE);
+
+            next_cc = (_pmt->tsheader()->ContinuityCounter()+1)&0xf;
+            _pmt->tsheader()->SetContinuityCounter(next_cc);
+            ringBuffer->Write(_pmt->tsheader()->data(), TSPacket::SIZE);
+
+            _ts_packets_until_psip_sync = TSPACKETS_BETWEEN_PSIP_SYNC;
+        }
+    }
+    else
+        _ts_packets_until_psip_sync--;
+}
+
+void DVBRecorder::StopRecording(void)
+{
+#ifdef USE_DRB
+    TVRec *rec = tvrec;
+    tvrec = NULL; // don't notify of pause..
+  
+#if 0
+    bool ok = true;
+    if (!IsPaused())
+    {
+        VERBOSE(VB_IMPORTANT, "_recording: "<<_recording);
+        if (_recording)
+        {
+            Pause();
+            ok = WaitForPause(250);
+        }
+        else if (_drb && !_drb->IsPaused() && _drb->IsRunning())
+        {
+            VERBOSE(VB_IMPORTANT, "pausing DeviceReadBuffer");
+            _drb->SetRequestPause(true);
+            MythTimer t;
+            t.start();
+            while (!_drb->IsPaused() && t.elapsed() < 250);
+            ok = _drb->IsPaused();            
+        }
+    }
+#endif
+
+    _request_recording = false;
+  
+    if (_drb && _drb->IsRunning())
+        _drb->Stop();
+
+    while (_recording)
+        usleep(2000);
+
+    tvrec = rec;
+#else
+    DTVRecorder::StopRecording();
+#endif
+}
+
+void DVBRecorder::ReaderPaused(int /*fd*/)
+{
+#ifdef USE_DRB
+    pauseWait.wakeAll();
+    if (tvrec)
+        tvrec->RecorderPaused();
+#endif // USE_DRB
+}
+
+bool DVBRecorder::PauseAndWait(int timeout)
+{
+#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
+}
+
+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;
+
+    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;
+}
+
+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...
+    }
+
+    const uint pid = tspacket.PID();
+
+    QMutexLocker locker(&_pid_lock);
+    PIDInfo *info = _pid_infos[pid];
+    if (!info)
+       info = _pid_infos[pid] = new PIDInfo();
+
+    // 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;
+        }
+        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;
+    }
+
+    // Check continuity counter
+    if (tspacket.HasPayload())
+    {
+        if (!info->CheckCC(tspacket.ContinuityCounter()))
+        {
+            VERBOSE(VB_RECORD, LOC +
+                    QString("PID 0x%1 discontinuity detected").arg(pid,0,16));
+            _continuity_error_count++;
+        }
+    }
+
+    // handle legacy PES recording mode
+    if (!_record_transport_stream_option)
+    {
+        // handle PS recording
+
+        ipack *ip = info->ip;
+        if (ip == NULL)
+            return true;
+
+        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);
+        return true;
+    }
+
+    // handle TS recording
+
+    // Check for keyframes and count frames
+    if (info->isVideo)
+        FindKeyframes(&tspacket);
+
+    // 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 (!tspacket.PayloadStart())
+            return true; // not payload start - drop packet
+
+        VERBOSE(VB_RECORD,
+                QString("PID 0x%1 Found Payload Start").arg(pid,0,16));
+        info->payloadStartSeen = true;
+    }
+
+    // Write PAT & PMT tables occasionally
+    WritePATPMT();
+
+    // Write Data
+    if (ringBuffer)
+        ringBuffer->Write(tspacket.data(), TSPacket::SIZE);
+    return true;
+}
+
+////////////////////////////////////////////////////////////
+// Stuff below this comment will be phased out after 0.20 //
+////////////////////////////////////////////////////////////
+
+static void print_pmt_info(
+    QString pre, const QValueList<ElementaryPIDObject> &pmt_list, bool fta)
+{
+    if (!(print_verbose_messages|VB_RECORD))
+        return;
+
+    QValueList<ElementaryPIDObject>::const_iterator it;
+    for (it = pmt_list.begin(); it != pmt_list.end(); ++it)
+    {
+        VERBOSE(VB_RECORD, pre +
+                QString("AutoPID %1 PID 0x%2, %3")
+                .arg(((*it).Record) ? "recording" : "skipping")
+                .arg((*it).PID,0,16).arg((*it).Description));
+    }
+
+    VERBOSE(VB_RECORD, pre + "AutoPID Complete - PAT/PMT Loaded for service\n"
+            "\t\t\tA/V Streams are " + ((fta ? "unencrypted" : "ENCRYPTED")));
+}
+
+/** \fn DVBRecorder::AutoPID(void)
+ *  \brief Process PMT and decide which components should be recorded
+ *
+ *   This is particularly for hardware decoders which don't
+ *   like to see more than one audio or one video stream.
+ *
+ *   When the hardware decoder is not specified all components
+ *   are recorded.
+ */
+void DVBRecorder::AutoPID(void)
+{
+    QMutexLocker change_lock(&_pid_lock);
+
+    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));
+
+    // If this is not for a hardware decoder, record everything.
+    if (!_hw_decoder_option)
+    {
+        QValueList<ElementaryPIDObject> &pmt_list = _input_pmt.Components;
+
+        QValueList<ElementaryPIDObject>::iterator it;
+        for (it = pmt_list.begin(); it != pmt_list.end(); ++it)
+            (*it).Record = true;
+
+        print_pmt_info(LOC, pmt_list, _input_pmt.FTA());
+        return;
+    }
+
+    // Wanted stream types:
+    QValueList<ES_Type> StreamTypes;
+    StreamTypes += ES_TYPE_VIDEO_MPEG1;
+    StreamTypes += ES_TYPE_VIDEO_MPEG2;
+    StreamTypes += ES_TYPE_AUDIO_MPEG1;
+    StreamTypes += ES_TYPE_AUDIO_MPEG2;
+    StreamTypes += ES_TYPE_AUDIO_AC3;
+    if (_record_transport_stream_option)
+    {
+        // The following types are only supported with TS recording
+        StreamTypes += ES_TYPE_AUDIO_DTS;
+        StreamTypes += ES_TYPE_AUDIO_AAC;
+        StreamTypes += ES_TYPE_TELETEXT;
+        StreamTypes += ES_TYPE_SUBTITLE;
+    }
+
+    QMap<ES_Type, bool> flagged;
+    // Wanted languages:
+    QStringList Languages = iso639_get_language_list();
+
+    QValueList<ElementaryPIDObject>::iterator es;
+    for (es = _input_pmt.Components.begin();
+         es != _input_pmt.Components.end(); ++es)
+    {
+        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
+        }
+
+        // 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;
+
+            case ES_TYPE_AUDIO_MPEG1: 
+            case ES_TYPE_AUDIO_MPEG2:
+            case ES_TYPE_AUDIO_AC3:
+            case ES_TYPE_AUDIO_DTS:
+            case ES_TYPE_AUDIO_AAC:
+                if (flagged.contains(ES_TYPE_AUDIO_MPEG1) ||
+                    flagged.contains(ES_TYPE_AUDIO_MPEG2) ||
+                    flagged.contains(ES_TYPE_AUDIO_AC3)   ||
+                    flagged.contains(ES_TYPE_AUDIO_DTS)   ||
+                    flagged.contains(ES_TYPE_AUDIO_AAC))
+                {
+                    continue; // Ignore this stream
+                }
+                break;
+
+            default:
+                break;
+        }
+        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;
+        }
+    }
+
+    print_pmt_info(LOC, _input_pmt.Components, _input_pmt.FTA());
+}
+
+bool DVBRecorder::Poll(void) const
+{
+#ifndef USE_DRB
+    struct pollfd polls;
+    polls.fd      = _stream_fd;
+    polls.events  = POLLIN;
+    polls.revents = 0;
+
+    int ret;
+    do
+        ret = poll(&polls, 1, POLL_INTERVAL);
+    while (!request_pause && IsOpen() &&
+           ((-1 == ret) && ((EAGAIN == errno) || (EINTR == errno))));
+
+    if (request_pause || !IsOpen())
+        return false;
+
+    if (ret > 0 && polls.revents & POLLIN)
+    {
+        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;
+    }
+
+    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()));
+    }
+
+    if (ret < 0 && (EOVERFLOW == errno))
+    {
+        _stream_overflow_count++;
+        VERBOSE(VB_RECORD, LOC_ERR + "Driver buffer overflow detected.");
+    }
+
+    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
+
 void DVBRecorder::DebugTSHeader(unsigned char* buffer, int len)
 {
     (void) len;

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 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;   }
+    bool IsRunning(void) const;
+
+    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);
+    if (_drb && _drb->IsRunning())
+        _drb->Stop();
 
-    if (run)
-        pthread_join(ringbuf.thread, NULL);
+    while (_recording)
+        usleep(2000);
 
-    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;
-    }
     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 +427 @@
     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 +440 @@
 
 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 +494 @@
         {
             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 +507 @@
     }
     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 +527 @@
     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 +536 @@
     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 +565 @@
     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 +598 @@
         }
 
         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 +620 @@
 
 void HDTVRecorder::Reset(void)
 {
-    VERBOSE(VB_RECORD, "HDTVRecorder::Reset(void)");
+    VERBOSE(VB_RECORD, LOC + "Reset(void)");
     DTVRecorder::Reset();
 
     _error = false;
@@ -947 +628 @@
     _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

@@ -18 +18 @@
 #include "dtvrecorder.h"
 #include "tspacket.h"
 #include "transform.h"
+#include "DeviceReadBuffer.h"
 
 #include "dvbtypes.h"
 #include "dvbchannel.h"
@@ -26 +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:
@@ -48 +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
@@ -64 +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;
@@ -91 +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;
@@ -137 +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()
+{
+    if (buffer)
+        delete[] buffer;
+}
+
+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::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)
+{
+    bool was_running = IsRunning();
+    lock.lock();
+    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;
+}
+
+bool DeviceReadBuffer::IsRunning(void) const
+{
+    QMutexLocker locker(&lock);
+    return running;
+}
+
+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(5000);
+            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) && running)
+    {
+        {
+            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
+}