Index: libs/libmythtv/dvbrecorder.cpp =================================================================== --- libs/libmythtv/dvbrecorder.cpp (revision 8093) +++ libs/libmythtv/dvbrecorder.cpp (working copy) @@ -72,36 +72,43 @@ 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,7 +120,10 @@ 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,7 +187,7 @@ bool DVBRecorder::Open(void) { - if (_stream_fd >= 0) + if (IsOpen()) { VERBOSE(VB_GENERAL, LOC_WARN + "Card already open"); return true; @@ -185,21 +195,21 @@ _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,38 +221,27 @@ { 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,7 +283,7 @@ 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,66 +302,27 @@ 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,7 +387,7 @@ (*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,21 +396,20 @@ 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 StreamTypes; StreamTypes += ES_TYPE_VIDEO_MPEG1; @@ -467,118 +426,50 @@ StreamTypes += ES_TYPE_SUBTITLE; } - QMap flagged; + uint has_video = 0; + uint has_audio = 0; + QValueList::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,67 +491,72 @@ 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,269 +564,202 @@ _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(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(&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,8 +769,6 @@ } else _ts_packets_until_psip_sync--; - - ringBuffer->Write(buffer,len); } void DVBRecorder::Reset(void) @@ -977,7 +804,7 @@ 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,39 +816,79 @@ 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 pids; + vector types; + vector pdesc; + QValueList::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,44 +978,187 @@ 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(&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 Index: libs/libmythtv/DeviceReadBuffer.h =================================================================== --- libs/libmythtv/DeviceReadBuffer.h (revision 0) +++ libs/libmythtv/DeviceReadBuffer.h (revision 0) @@ -0,0 +1,109 @@ +// -*- Mode: c++ -*- +/* Device Buffer written by John Poet */ + +#ifndef _DEVICEREADBUFFER_H_ +#define _DEVICEREADBUFFER_H_ + +#include +#include +#include + +#include +#include +#include + +#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_ Index: libs/libmythtv/hdtvrecorder.h =================================================================== --- libs/libmythtv/hdtvrecorder.h (revision 8093) +++ libs/libmythtv/hdtvrecorder.h (working copy) @@ -3,7 +3,6 @@ * 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,6 +11,7 @@ #include "dtvrecorder.h" #include "tsstats.h" +#include "DeviceReadBuffer.h" struct AVFormatContext; struct AVPacket; @@ -28,14 +28,12 @@ * * \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,9 +45,6 @@ void StartRecording(void); void StopRecording(void); - void Pause(bool clear = false); - bool IsPaused(void) const; - void Reset(void); bool Open(void); @@ -61,54 +56,41 @@ 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 Index: libs/libmythtv/hdtvrecorder.cpp =================================================================== --- libs/libmythtv/hdtvrecorder.cpp (revision 8093) +++ libs/libmythtv/hdtvrecorder.cpp (working copy) @@ -84,13 +84,15 @@ #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,6 +111,9 @@ }; #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,8 +121,10 @@ 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,30 +132,25 @@ 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,8 +166,7 @@ HDTVRecorder::~HDTVRecorder() { TeardownAll(); - pthread_mutex_destroy(&ringbuf.lock); - pthread_mutex_destroy(&ringbuf.lock_stats); + delete _drb; } void HDTVRecorder::deleteLater(void) @@ -189,37 +190,49 @@ 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,32 +244,37 @@ 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; ifill_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(avg) - / ringbuf.size) * 100.0) - .arg((static_cast(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(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,42 +376,51 @@ _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,7 +431,8 @@ 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,81 +444,22 @@ 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; iPMT()->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,9 +498,8 @@ { 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,14 +511,14 @@ } 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,11 +531,6 @@ 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,11 +540,6 @@ 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,19 +569,29 @@ 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,13 +602,21 @@ } const TSPacket *pkt = reinterpret_cast(&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,7 +624,7 @@ void HDTVRecorder::Reset(void) { - VERBOSE(VB_RECORD, "HDTVRecorder::Reset(void)"); + VERBOSE(VB_RECORD, LOC + "Reset(void)"); DTVRecorder::Reset(); _error = false; @@ -947,46 +632,28 @@ _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 */; } Index: libs/libmythtv/libmythtv.pro =================================================================== --- libs/libmythtv/libmythtv.pro (revision 8093) +++ libs/libmythtv/libmythtv.pro (working copy) @@ -206,10 +206,10 @@ # 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 Index: libs/libmythtv/dvbrecorder.h =================================================================== --- libs/libmythtv/dvbrecorder.h (revision 8093) +++ libs/libmythtv/dvbrecorder.h (working copy) @@ -1,3 +1,4 @@ +// -*- Mode: c++ -*- /* * Copyright (C) Kenneth Aafloy 2003 * @@ -17,6 +18,7 @@ #include "dtvrecorder.h" #include "tspacket.h" #include "transform.h" +#include "DeviceReadBuffer.h" #include "dvbtypes.h" #include "dvbchannel.h" @@ -25,13 +27,32 @@ 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 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,8 +68,10 @@ 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,25 +86,36 @@ 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,45 +124,33 @@ // 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 _payload_start_seen; - QMap _videoPID; // Input Misc /// PMT on input side PMTObject _input_pmt; - /// Input filter file descriptors - vector _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 _encrypted_pid; - // locking - QMutex _pid_lock; - // Statistics - uint _continuity_error_count; - uint _stream_overflow_count; - uint _bad_packet_count; - QMap _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,4 +158,32 @@ 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 Index: libs/libmythtv/DeviceReadBuffer.cpp =================================================================== --- libs/libmythtv/DeviceReadBuffer.cpp (revision 0) +++ libs/libmythtv/DeviceReadBuffer.cpp (revision 0) @@ -0,0 +1,463 @@ +#include +#include + +#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 +}