Index: libs/libmythtv/RingBuffer.cpp =================================================================== --- libs/libmythtv/RingBuffer.cpp (revision 25873) +++ libs/libmythtv/RingBuffer.cpp (working copy) @@ -5,6 +5,7 @@ #include // POSIX C headers +#define _LARGEFILE53_SOURCE #include #include #include @@ -19,8 +20,8 @@ using namespace std; +#include "mythcontext.h" // for VERBOSE #include "mythconfig.h" - #include "exitcodes.h" #include "RingBuffer.h" #include "remotefile.h" @@ -30,7 +31,6 @@ #include "BDRingBuffer.h" #include "util.h" #include "compat.h" -#include "mythverbose.h" #ifndef O_STREAMING #define O_STREAMING 0 @@ -44,7 +44,8 @@ #define O_BINARY 0 #endif -const uint RingBuffer::kBufferSize = 3 * 1024 * 1024; +// about one second at 35mbit +const uint RingBuffer::kBufferSize = 4 * 1024 * 1024; #define CHUNK 32768 /* readblocksize increments */ @@ -65,23 +66,25 @@ /* Locking relations: - rwlock->readAheadLock - ->readsAllowedWaitMutex->readAheadRunningCondLock - ->availWaitMutex + rwlock->poslock->rbrlock->rbwlock A child should never lock any of the parents without locking the parent lock before the child lock. void RingBuffer::Example1() { - QMutexLocker locker1(&readAheadRunningCondLock); - QMutexLocker locker2(&readsAllowedWaitMutex); // error! + poslock.lockForWrite(); + rwlock.lockForRead(); // error! blah(); // <- does not implicitly aquire any locks + rwlock.unlock(); + poslock.unlock(); } void RingBuffer::Example2() { - QMutexLocker locker1(&readsAllowedWaitMutex); - QMutexLocker locker2(&readAheadRunningCondLock); // ok! + rwlock.lockForRead(); + rbrlock.lockForWrite(); // ok! blah(); // <- does not implicitly aquire any locks + rbrlock.unlock(); + rwlock.unlock(); } */ @@ -113,19 +116,19 @@ RingBuffer::RingBuffer(const QString &lfilename, bool write, bool readahead, uint read_retries) - : filename(lfilename), subtitlefilename(QString::null), + : readpos(0), writepos(0), + internalreadpos(0), ignorereadpos(-1), + rbrpos(0), rbwpos(0), + stopreads(false), + filename(lfilename), subtitlefilename(QString::null), tfw(NULL), fd2(-1), - writemode(false), - readpos(0), writepos(0), - stopreads(false), remotefile(NULL), + writemode(false), remotefile(NULL), startreadahead(readahead),readAheadBuffer(NULL), - readaheadrunning(false), readaheadpaused(false), - pausereadthread(false), - rbrpos(0), rbwpos(0), - internalreadpos(0), ateof(false), - readsallowed(false), wantseek(false), setswitchtonext(false), - streamOnly(false), - rawbitrate(4000), playspeed(1.0f), + readaheadrunning(false), reallyrunning(false), + request_pause(false), paused(false), + ateof(false), readsallowed(false), + setswitchtonext(false), streamOnly(false), + rawbitrate(8000), playspeed(1.0f), fill_threshold(65536), fill_min(-1), readblocksize(CHUNK), wanttoread(0), numfailures(0), commserror(false), @@ -263,18 +266,17 @@ return QString::null; } -/** \fn RingBuffer::OpenFile(const QString&, uint) - * \brief Opens a file for reading. +/** \brief Opens a file for reading. * * \param lfilename Name of file to read - * \param retryCount How often to retry reading the file before giving up + * \param retryTime How many ms to retry reading the file before giving up */ -void RingBuffer::OpenFile(const QString &lfilename, uint retryCount) +void RingBuffer::OpenFile(const QString &lfilename, uint retry_ms) { - VERBOSE(VB_PLAYBACK, LOC + QString("OpenFile(%1, %2)") - .arg(lfilename).arg(retryCount)); + VERBOSE(VB_PLAYBACK, LOC + QString("OpenFile(%1, %2 ms)") + .arg(lfilename).arg(retry_ms)); - uint openAttempts = retryCount + 1; + rwlock.lockForWrite(); filename = lfilename; @@ -365,14 +367,15 @@ if (is_local) { char buf[kReadTestSize]; - int timetowait = 500 * openAttempts; int lasterror = 0; MythTimer openTimer; openTimer.start(); - while (openTimer.elapsed() < timetowait) + uint openAttempts = 0; + while ((uint)openTimer.elapsed() < retry_ms) { + openAttempts++; lasterror = 0; QByteArray fname = filename.toLocal8Bit(); fd2 = open(fname.constData(), @@ -381,10 +384,13 @@ if (fd2 < 0) { if (!check_permissions(filename)) + { + lasterror = 3; break; + } lasterror = 1; - usleep(1000); + usleep(10 * 1000); } else { @@ -394,61 +400,66 @@ lasterror = 2; close(fd2); fd2 = -1; - usleep(1000); + usleep(10 * 1000); } else { - lseek(fd2, 0, SEEK_SET); + if (0 == lseek(fd2, 0, SEEK_SET)) + { #if HAVE_POSIX_FADVISE - posix_fadvise(fd2, 0, 0, POSIX_FADV_SEQUENTIAL); + posix_fadvise(fd2, 0, 0, POSIX_FADV_SEQUENTIAL); #endif - openAttempts = 0; - break; + lasterror = 0; + break; + } + close(fd2); + fd2 = -1; } } } switch (lasterror) { + case 0: + { + QFileInfo fi(filename); + oldfile = fi.lastModified() + .secsTo(QDateTime::currentDateTime()) > 60; + QString extension = fi.completeSuffix().toLower(); + if (is_subtitle_possible(extension)) + subtitlefilename = local_sub_filename(fi); + break; + } case 1: - VERBOSE(VB_IMPORTANT, LOC + - QString("Could not open %1.").arg(filename)); + VERBOSE(VB_IMPORTANT, LOC_ERR + + QString("OpenFile(): Could not open.")); break; case 2: - VERBOSE(VB_IMPORTANT, LOC + - QString("Invalid file (fd %1) when opening '%2'.") - .arg(fd2).arg(filename)); + VERBOSE(VB_IMPORTANT, LOC_ERR + + QString("OpenFile(): File too small (%1B).") + .arg(QFileInfo(filename).size())); break; + case 3: + VERBOSE(VB_IMPORTANT, LOC_ERR + + "OpenFile(): Improper permissions."); + break; default: break; } + VERBOSE(VB_FILE, LOC + QString("OpenFile() made %1 attempts in %2 ms") + .arg(openAttempts).arg(openTimer.elapsed())); - - QFileInfo fileInfo(filename); - if (fileInfo.lastModified().secsTo(QDateTime::currentDateTime()) > - 30 * 60) - { - oldfile = true; - } - - QString extension = fileInfo.completeSuffix().toLower(); - if (is_subtitle_possible(extension)) - subtitlefilename = local_sub_filename(fileInfo); } #ifdef USING_FRONTEND else if (is_dvd) { dvdPriv->OpenFile(filename); - rwlock.lockForWrite(); readblocksize = DVD_BLOCK_SIZE * 62; - rwlock.unlock(); } else if (is_bd) { bdPriv->OpenFile(filename); - rwlock.lockForWrite(); readblocksize = BD_BLOCK_SIZE * 62; - rwlock.unlock(); } #endif // USING_FRONTEND else @@ -484,7 +495,7 @@ remotefile = new RemoteFile(filename, false, true, -1, &auxFiles); if (!remotefile->isOpen()) { - VERBOSE(VB_IMPORTANT, + VERBOSE(VB_IMPORTANT, LOC_ERR + QString("RingBuffer::RingBuffer(): Failed to open remote " "file (%1)").arg(filename)); delete remotefile; @@ -503,7 +514,10 @@ commserror = false; numfailures = 0; - UpdateRawBitrate(4000); + rawbitrate = 8000; + CalcReadAheadThresh(); + + rwlock.unlock(); } /** \fn RingBuffer::IsOpen(void) const @@ -511,12 +525,16 @@ */ bool RingBuffer::IsOpen(void) const { + rwlock.lockForRead(); + bool ret; #ifdef USING_FRONTEND - return tfw || (fd2 > -1) || remotefile || (dvdPriv && dvdPriv->IsOpen()) || + ret = tfw || (fd2 > -1) || remotefile || (dvdPriv && dvdPriv->IsOpen()) || (bdPriv && bdPriv->IsOpen()); #else // if !USING_FRONTEND - return tfw || (fd2 > -1) || remotefile; + ret = tfw || (fd2 > -1) || remotefile; #endif // !USING_FRONTEND + rwlock.unlock(); + return ret; } /** \fn RingBuffer::~RingBuffer(void) @@ -560,27 +578,17 @@ rwlock.unlock(); } -/** \fn RingBuffer::Start(void) - * \brief Starts the read-ahead thread. - * - * If this RingBuffer is not in write-mode, the RingBuffer constructor - * was called with a usereadahead of true, and the read-ahead thread - * is not already running. - */ -void RingBuffer::Start(void) -{ - if (!writemode && !readaheadrunning && startreadahead) - StartupReadAheadThread(); -} - /** \fn RingBuffer::Reset(bool, bool, bool) * \brief Resets the read-ahead thread and our position in the file */ void RingBuffer::Reset(bool full, bool toAdjust, bool resetInternal) { - wantseek = true; + VERBOSE(VB_FILE, LOC + QString("Reset(%1,%2,%3)") + .arg(full).arg(toAdjust).arg(resetInternal)); + rwlock.lockForWrite(); - wantseek = false; + poslock.lockForWrite(); + numfailures = 0; commserror = false; setswitchtonext = false; @@ -590,7 +598,7 @@ if (readpos != 0) { - VERBOSE(VB_IMPORTANT, QString( + VERBOSE(VB_IMPORTANT, LOC + QString( "RingBuffer::Reset() nonzero readpos. toAdjust: %1 readpos: %2" " readAdjust: %3").arg(toAdjust).arg(readpos).arg(readAdjust)); } @@ -604,6 +612,8 @@ if (resetInternal) internalreadpos = readpos; + generalWait.wakeAll(); + poslock.unlock(); rwlock.unlock(); } @@ -625,7 +635,7 @@ unsigned errcnt = 0; unsigned zerocnt = 0; - if (fd < 0) + if (fd2 < 0) { VERBOSE(VB_IMPORTANT, LOC_ERR + "Invalid file descriptor in 'safe_read()'"); @@ -637,7 +647,7 @@ while (tot < sz) { - ret = read(fd, (char *)data + tot, sz - tot); + ret = read(fd2, (char *)data + tot, sz - tot); if (ret < 0) { if (errno == EAGAIN) @@ -699,7 +709,9 @@ VERBOSE(VB_IMPORTANT, LOC_ERR + "RingBuffer::safe_read(RemoteFile* ...): read failed"); + poslock.lockForRead(); rf->Seek(internalreadpos - readAdjust, SEEK_SET); + poslock.unlock(); ret = 0; numfailures++; } @@ -714,37 +726,22 @@ */ void RingBuffer::UpdateRawBitrate(uint raw_bitrate) { + VERBOSE(VB_FILE, LOC + QString("UpdateRawBitrate(%1Kb)").arg(raw_bitrate)); + if (raw_bitrate < 2500) + { + VERBOSE(VB_FILE, LOC + + QString("UpdateRawBitrate(%1Kb) - ignoring bitrate,") + .arg(raw_bitrate) + + "\n\t\t\tappears to be abnormally low."); + return; + } + rwlock.lockForWrite(); rawbitrate = raw_bitrate; CalcReadAheadThresh(); rwlock.unlock(); } -/** \fn RingBuffer::GetBitrate(void) const - * \brief Returns effective bits per second (in thousands). - * - * NOTE: This is reported in telecom kilobytes, to get - * the bits per second multiply by 1000, not 1024. - */ -uint RingBuffer::GetBitrate(void) const -{ - rwlock.lockForRead(); - uint tmp = (uint) max(abs(rawbitrate * playspeed), 0.5f * rawbitrate); - rwlock.unlock(); - return min(rawbitrate * 3, tmp); -} - -/** \fn RingBuffer::GetReadBlockSize(void) const - * \brief Returns size of each disk read made by read ahead thread (in bytes). - */ -uint RingBuffer::GetReadBlockSize(void) const -{ - rwlock.lockForRead(); - uint tmp = readblocksize; - rwlock.unlock(); - return tmp; -} - /** \fn RingBuffer::UpdatePlaySpeed(float) * \brief Set the play speed, to allow RingBuffer adjust effective bitrate. * \param play_speed Speed to set. (1.0 for normal speed) @@ -768,65 +765,91 @@ { uint estbitrate = 0; - wantseek = false; readsallowed = false; - readblocksize = CHUNK; + readblocksize = max(readblocksize, CHUNK); // loop without sleeping if the buffered data is less than this - fill_threshold = CHUNK * 2; - fill_min = 1; + fill_threshold = kBufferSize / 8; -#ifdef USING_FRONTEND - if (dvdPriv || bdPriv) - { - const uint KB32 = 32*1024; - const uint KB64 = 64*1024; - const uint KB128 = 128*1024; - const uint KB256 = 256*1024; - const uint KB512 = 512*1024; + const uint KB32 = 32*1024; + const uint KB64 = 64*1024; + const uint KB128 = 128*1024; + const uint KB256 = 256*1024; + const uint KB512 = 512*1024; - estbitrate = (uint) max(abs(rawbitrate * playspeed), - 0.5f * rawbitrate); - estbitrate = min(rawbitrate * 3, estbitrate); - readblocksize = (estbitrate > 2500) ? KB64 : KB32; - readblocksize = (estbitrate > 5000) ? KB128 : readblocksize; - readblocksize = (estbitrate > 9000) ? KB256 : readblocksize; - readblocksize = (estbitrate > 18000) ? KB512 : readblocksize; + estbitrate = (uint) max(abs(rawbitrate * playspeed), + 0.5f * rawbitrate); + estbitrate = min(rawbitrate * 3, estbitrate); + int rbs = (estbitrate > 2500) ? KB64 : KB32; + rbs = (estbitrate > 5000) ? KB128 : rbs; + rbs = (estbitrate > 9000) ? KB256 : rbs; + rbs = (estbitrate > 18000) ? KB512 : rbs; + readblocksize = max(rbs,readblocksize); - // minumum seconds of buffering before allowing read - float secs_min = 0.1; + // minumum seconds of buffering before allowing read + float secs_min = 0.25; + // set the minimum buffering before allowing ffmpeg read + fill_min = (uint) ((estbitrate * secs_min) * 0.125f); + // make this a multiple of ffmpeg block size.. + fill_min = ((fill_min / KB32) + 1) * KB32; - // set the minimum buffering before allowing ffmpeg read - fill_min = (uint) ((estbitrate * secs_min) * 0.125f); - // make this a multiple of ffmpeg block size.. - fill_min = ((fill_min / KB32) + 1) * KB32; - } -#endif // USING_FRONTEND - - VERBOSE(VB_PLAYBACK, LOC + - QString("CalcReadAheadThresh(%1 KB)\n\t\t\t -> " + VERBOSE(VB_FILE, LOC + + QString("CalcReadAheadThresh(%1 Kb)\n\t\t\t -> " "threshhold(%2 KB) min read(%3 KB) blk size(%4 KB)") .arg(estbitrate).arg(fill_threshold/1024) .arg(fill_min/1024).arg(readblocksize/1024)); } -/** \fn RingBuffer::ReadBufFree(void) const - * \brief Returns number of bytes available for reading into buffer. - */ +bool RingBuffer::IsNearEnd(double fps, uint vvf) const +{ + rwlock.lockForRead(); + int sz = ReadBufAvail(); + uint rbs = readblocksize; + // telecom kilobytes (i.e. 1000 per k not 1024) + uint tmp = (uint) max(abs(rawbitrate * playspeed), 0.5f * rawbitrate); + uint kbits_per_sec = min(rawbitrate * 3, tmp); + rwlock.unlock(); + + // WARNING: readahead_frames can greatly overestimate or underestimate + // the number of frames available in the read ahead buffer + // when rh_frames is less than the keyframe distance. + double bytes_per_frame = kbits_per_sec * (1000.0/8.0) / fps; + double readahead_frames = sz / bytes_per_frame; + + bool near_end = ((vvf + readahead_frames) < 10.0) || (sz < rbs*1.5); + + VERBOSE(VB_PLAYBACK, LOC + "IsReallyNearEnd()" + <<" br("<<(kbits_per_sec/8)<<"KB)" + <<" sz("<<(sz / 1000)<<"KB)" + <<" vfl("<= rbrpos) ? rbrpos + kBufferSize : rbrpos) - rbwpos - 1; + rbrlock.lockForRead(); + rbwlock.lockForRead(); + int ret = ((rbwpos >= rbrpos) ? rbrpos + kBufferSize : rbrpos) - rbwpos - 1; + rbwlock.unlock(); + rbrlock.unlock(); + return ret; } -/** \fn RingBuffer::ReadBufAvail(void) const - * \brief Returns number of bytes available for reading from buffer. - */ +/// \brief Returns number of bytes available for reading from buffer. +/// WARNING: Must be called with rwlock in locked state. int RingBuffer::ReadBufAvail(void) const { - QMutexLocker locker(&readAheadLock); - return (rbwpos >= rbrpos) ? - rbwpos - rbrpos : kBufferSize - rbrpos + rbwpos; + rbrlock.lockForRead(); + rbwlock.lockForRead(); + int ret = (rbwpos >= rbrpos) ? rbwpos - rbrpos : kBufferSize - rbrpos + rbwpos; + rbwlock.unlock(); + rbrlock.unlock(); + return ret; } /** \fn RingBuffer::ResetReadAhead(long long) @@ -836,36 +859,81 @@ * buffer doesn't contain any stale data, and so that it will read * any new data from the new position in the file. * - * WARNING: Must be called with rwlock in write lock state. + * WARNING: Must be called with rwlock and poslock in write lock state. * * \param newinternal Position in file to start reading data from */ void RingBuffer::ResetReadAhead(long long newinternal) { - readAheadLock.lock(); - readblocksize = CHUNK; + VERBOSE(VB_FILE, LOC + QString("ResetReadAhead(internalreadpos = %1->%2)") + .arg(internalreadpos).arg(newinternal)); + + rbrlock.lockForWrite(); + rbwlock.lockForWrite(); + + CalcReadAheadThresh(); rbrpos = 0; rbwpos = 0; internalreadpos = newinternal; ateof = false; readsallowed = false; setswitchtonext = false; - readAheadLock.unlock(); + generalWait.wakeAll(); + + rbwlock.unlock(); + rbrlock.unlock(); } -/** \fn RingBuffer::StartupReadAheadThread(void) - * \brief Creates the read-ahead thread, and waits for it to start. +/** + * \brief Starts the read-ahead thread. * - * \sa Start(void). + * If the RingBuffer constructor was not called with a usereadahead + * of true of if this was reset to false because we're dealing with + * a DVD the read ahead thread will not be started. + * + * If this RingBuffer is in write-mode a warning will be printed and + * the read ahead thread will not be started. + * + * If the read ahead thread is already running a warning will be printed + * and the read ahead thread will not be started. + * */ -void RingBuffer::StartupReadAheadThread(void) +void RingBuffer::Start(void) { - readaheadrunning = false; + bool do_start = true; - readAheadRunningCondLock.lock(); + rwlock.lockForWrite(); + if (!startreadahead) + { + do_start = false; + } + else if (writemode) + { + VERBOSE(VB_IMPORTANT, LOC_WARN + "Not starting read ahead thread, " + "this is a write only RingBuffer"); + do_start = false; + } + else if (readaheadrunning) + { + VERBOSE(VB_IMPORTANT, LOC_WARN + "Not starting read ahead thread, " + "already running"); + do_start = false; + } + + if (!do_start) + { + rwlock.unlock(); + return; + } + + StartReads(); + QThread::start(); - readAheadRunningCond.wait(&readAheadRunningCondLock); - readAheadRunningCondLock.unlock(); + + while (readaheadrunning && !reallyrunning) + generalWait.wait(&rwlock); + + rwlock.unlock(); } /** \fn RingBuffer::KillReadAheadThread(void) @@ -873,11 +941,15 @@ */ void RingBuffer::KillReadAheadThread(void) { - if (!readaheadrunning) - return; + rwlock.lockForWrite(); + bool do_wait = readaheadrunning; + readaheadrunning = false; + StopReads(); + generalWait.wakeAll(); + rwlock.unlock(); - readaheadrunning = false; - QThread::wait(); + if (do_wait) + QThread::wait(); } /** \fn RingBuffer::StopReads(void) @@ -887,7 +959,7 @@ void RingBuffer::StopReads(void) { stopreads = true; - availWait.wakeAll(); + generalWait.wakeAll(); } /** \fn RingBuffer::StartReads(void) @@ -905,8 +977,11 @@ */ void RingBuffer::Pause(void) { - pausereadthread = true; StopReads(); + + rwlock.lockForWrite(); + request_pause = true; + rwlock.unlock(); } /** \fn RingBuffer::Unpause(void) @@ -916,157 +991,278 @@ void RingBuffer::Unpause(void) { StartReads(); - pausereadthread = false; + + rwlock.lockForWrite(); + request_pause = false; + generalWait.wakeAll(); + rwlock.unlock(); } +/// Returns false iff read-ahead is not running and read-ahead is not paused. +bool RingBuffer::isPaused(void) const +{ + rwlock.lockForRead(); + bool ret = !readaheadrunning || paused; + rwlock.unlock(); + return ret; +} + /** \fn RingBuffer::WaitForPause(void) * \brief Waits for Pause(void) to take effect. */ void RingBuffer::WaitForPause(void) { - if (!readaheadrunning) - return; + MythTimer t; + t.start(); - if (!readaheadpaused) + rwlock.lockForRead(); + while (readaheadrunning && !paused && request_pause) { - // Qt4 requires a QMutex as a parameter... - // not sure if this is the best solution. Mutex Must be locked before wait. - QMutex mutex; - mutex.lock(); + generalWait.wait(&rwlock, 1000); + if (readaheadrunning && !paused && request_pause && t.elapsed() > 1000) + { + VERBOSE(VB_IMPORTANT, LOC_WARN + + QString("Waited %1 ms for ringbuffer pause..") + .arg(t.elapsed())); + } + } + rwlock.unlock(); +} - while (!pauseWait.wait(&mutex, 1000)) - VERBOSE(VB_IMPORTANT, - LOC + "Waited too long for ringbuffer pause.."); +bool RingBuffer::PauseAndWait(void) +{ + const uint timeout = 500; // ms + + if (request_pause) + { + if (!paused) + { + rwlock.unlock(); + rwlock.lockForWrite(); + + if (request_pause) + { + paused = true; + generalWait.wakeAll(); + } + + rwlock.unlock(); + rwlock.lockForRead(); + } + + if (request_pause && paused && readaheadrunning) + generalWait.wait(&rwlock, timeout); } + + if (!request_pause && paused) + { + rwlock.unlock(); + rwlock.lockForWrite(); + + if (!request_pause) + { + paused = false; + generalWait.wakeAll(); + } + + rwlock.unlock(); + rwlock.lockForRead(); + } + + return request_pause || paused; } void RingBuffer::run(void) { - long long totfree = 0; - int ret = -1; - int used = 0; - int loops = 0; - + // These variables are used to adjust the read block size struct timeval lastread, now; - gettimeofday(&lastread, NULL); - const int KB640 = 640*1024; int readtimeavg = 300; - int readinterval; + bool ignore_for_read_timing = true; - pausereadthread = false; + gettimeofday(&lastread, NULL); // this is just to keep gcc happy - readAheadBuffer = new char[kBufferSize + KB640]; - rwlock.lockForWrite(); + poslock.lockForWrite(); + request_pause = false; + readAheadBuffer = new char[kBufferSize + 1024]; ResetReadAhead(0); + readaheadrunning = true; + reallyrunning = true; + generalWait.wakeAll(); + poslock.unlock(); rwlock.unlock(); - totfree = ReadBufFree(); + // NOTE: this must loop at some point hold only + // a read lock on rwlock, so that other functions + // such as reset and seek can take priority. - readaheadrunning = true; - readAheadRunningCondLock.lock(); - readAheadRunningCond.wakeAll(); - readAheadRunningCondLock.unlock(); + rwlock.lockForRead(); + + VERBOSE(VB_FILE, LOC + QString("Initial readblocksize %1K & fill_min %2K") + .arg(readblocksize/1024).arg(fill_min/1024)); + while (readaheadrunning) { - if (pausereadthread || writemode) + if (PauseAndWait()) { - readaheadpaused = true; - pauseWait.wakeAll(); - usleep(5000); - totfree = ReadBufFree(); + ignore_for_read_timing = true; continue; } - if (readaheadpaused) + long long totfree = ReadBufFree(); + + if (totfree < readblocksize) { - totfree = ReadBufFree(); - readaheadpaused = false; + generalWait.wait(&rwlock, 1000); + continue; } - totfree = ReadBufFree(); - if (totfree < GetReadBlockSize()) + if (ignorereadpos >= 0) { - usleep(50000); - totfree = ReadBufFree(); - ++loops; - // break out if we've spent lots of time here, just in case things - // are waiting on a wait condition that never got triggered. - if (readsallowed && loops < 10) - continue; + generalWait.wait(&rwlock, 1000); + continue; } - loops = 0; - rwlock.lockForRead(); - if (totfree > readblocksize && !commserror && !ateof && !setswitchtonext) + // No point in trying to read if stopreads is set but do sleep + // a little so we don't get stuck in a tight busy loop here.. + if (stopreads) { + ignore_for_read_timing = true; + generalWait.wait(&rwlock, 10); + continue; + } + + int read_return = -1; + if (totfree >= readblocksize && !commserror && + !ateof && !setswitchtonext) + { // limit the read size totfree = readblocksize; // adapt blocksize gettimeofday(&now, NULL); - readinterval = (now.tv_sec - lastread.tv_sec ) * 1000 + - (now.tv_usec - lastread.tv_usec) / 1000; + if (!ignore_for_read_timing) + { + int readinterval = (now.tv_sec - lastread.tv_sec ) * 1000 + + (now.tv_usec - lastread.tv_usec) / 1000; + readtimeavg = (readtimeavg * 9 + readinterval) / 10; - readtimeavg = (readtimeavg * 9 + readinterval) / 10; - - if (readtimeavg < 200 && readblocksize < KB640) - { - readblocksize += CHUNK; - //VERBOSE(VB_PLAYBACK, - // QString("Avg read interval was %1 msec. %2K block size") - // .arg(readtimeavg).arg(readblocksize/1024)); - readtimeavg = 300; + if (readtimeavg < 150 && (uint)readblocksize < (kBufferSize>>2)) + { + int old_block_size = readblocksize; + readblocksize = 3 * readblocksize / 2; + readblocksize = ((readblocksize+CHUNK-1) / CHUNK) * CHUNK; + VERBOSE(VB_FILE, LOC + + QString("Avg read interval was %1 msec. " + "%2K -> %3K block size") + .arg(readtimeavg) + .arg(old_block_size/1024) + .arg(readblocksize/1024)); + readtimeavg = 225; + } + else if (readtimeavg > 300 && readblocksize > CHUNK) + { + readblocksize -= CHUNK; + VERBOSE(VB_FILE, LOC + + QString("Avg read interval was %1 msec. " + "%2K -> %3K block size") + .arg(readtimeavg) + .arg((readblocksize+CHUNK)/1024) + .arg(readblocksize/1024)); + readtimeavg = 225; + } } - else if (readtimeavg > 400 && readblocksize > CHUNK) - { - readblocksize -= CHUNK; - //VERBOSE(VB_PLAYBACK, - // QString("Avg read interval was %1 msec. %2K block size") - // .arg(readtimeavg).arg(readblocksize/1024)); - readtimeavg = 300; - } + ignore_for_read_timing = false; lastread = now; + rbwlock.lockForRead(); if (rbwpos + totfree > kBufferSize) + { totfree = kBufferSize - rbwpos; + VERBOSE(VB_FILE|VB_EXTRA, LOC + + "Shrinking read, near end of buffer"); + } if (internalreadpos == 0) - totfree = fill_min; + { + totfree = max(fill_min, readblocksize); + VERBOSE(VB_FILE|VB_EXTRA, LOC + + "Reading enough data to start playback"); + } + VERBOSE(VB_FILE|VB_EXTRA, + LOC + QString("safe_read(...@%1, %2) -- begin") + .arg(rbwpos).arg(totfree)); if (remotefile) { if (livetvchain && livetvchain->HasNext()) remotefile->SetTimeout(true); - - ret = safe_read(remotefile, readAheadBuffer + rbwpos, - totfree); - internalreadpos += ret; + read_return = safe_read( + remotefile, readAheadBuffer + rbwpos, totfree); } #ifdef USING_FRONTEND else if (dvdPriv) { - ret = dvdPriv->safe_read(readAheadBuffer + rbwpos, totfree); - internalreadpos += ret; + read_return = dvdPriv->safe_read( + readAheadBuffer + rbwpos, totfree); } else if (bdPriv) { - ret = bdPriv->safe_read(readAheadBuffer + rbwpos, totfree); - internalreadpos += ret; + read_return = bdPriv->safe_read( + readAheadBuffer + rbwpos, totfree); } #endif // USING_FRONTEND else { - ret = safe_read(fd2, readAheadBuffer + rbwpos, totfree); - internalreadpos += ret; + read_return = safe_read(fd2, readAheadBuffer + rbwpos, totfree); } + VERBOSE(VB_FILE|VB_EXTRA, LOC + + QString("safe_read(...@%1, %2) -> %3") + .arg(rbwpos).arg(totfree).arg(read_return)); + rbwlock.unlock(); + } - readAheadLock.lock(); - if (ret > 0 ) - rbwpos = (rbwpos + ret) % kBufferSize; - readAheadLock.unlock(); + // If stopreads is set we toss the read data since we're probably + // doing a seek or opening a new file anyway. + if (stopreads) + { + VERBOSE(VB_FILE|VB_EXTRA, LOC + + "Saw stopreads.. dropping recently read data."); + continue; + } - if (ret == 0 && !stopreads) + if (read_return >= 0) + { + poslock.lockForWrite(); + rbwlock.lockForWrite(); + internalreadpos += read_return; + rbwpos = (rbwpos + read_return) % kBufferSize; + VERBOSE(VB_FILE|VB_EXTRA, + LOC + QString("rbwpos += %1K requested %2K in read") + .arg(read_return/1024,3).arg(totfree/1024,3)); + rbwlock.unlock(); + poslock.unlock(); + } + + int used = kBufferSize - ReadBufFree(); + + if ((0 == read_return) || (numfailures > 5) || + (readsallowed != (used >= fill_min || ateof || + setswitchtonext || commserror))) + { + // If readpos changes while the lock is released + // we should not handle the 0 read_return now. + long long old_readpos = readpos; + + rwlock.unlock(); + rwlock.lockForWrite(); + + commserror |= (numfailures > 5); + + readsallowed = used >= fill_min || ateof || + setswitchtonext || commserror; + + if (0 == read_return && old_readpos == readpos) { if (livetvchain) { @@ -1078,116 +1274,71 @@ } } else + { ateof = true; + } } - } - if (numfailures > 5) - commserror = true; - - totfree = ReadBufFree(); - used = kBufferSize - totfree; - - if (ateof || commserror) - { - readsallowed = true; - totfree = 0; + rwlock.unlock(); + rwlock.lockForRead(); + used = kBufferSize - ReadBufFree(); } - if (!readsallowed && (used >= fill_min || setswitchtonext)) - { - readsallowed = true; - //VERBOSE(VB_PLAYBACK, QString("reads allowed (%1 %2)").arg(used) - // .arg(fill_min)); - } - //else if (!readsallowed) - // VERBOSE(VB_PLAYBACK, QString("buffering (%1 %2 %3)").arg(used) - // .arg(fill_min) - // .arg(ret)); + VERBOSE(VB_FILE|VB_EXTRA, LOC + "@ end of read ahead loop"); - if (readsallowed && used < fill_min && !ateof && !setswitchtonext) - { - readsallowed = false; - //VERBOSE(VB_GENERAL, QString ("rebuffering (%1 %2)").arg(used) - // .arg(fill_min)); - } - - readsAllowedWaitMutex.lock(); - if (readsallowed || stopreads) - readsAllowedWait.wakeAll(); - readsAllowedWaitMutex.unlock(); - - availWaitMutex.lock(); - if (commserror || ateof || stopreads || setswitchtonext || + if (readsallowed || commserror || ateof || setswitchtonext || (wanttoread <= used && wanttoread > 0)) { - availWait.wakeAll(); + // To give other threads a good chance to handle these + // conditions, even if they are only requesting a read lock + // like us, yield (currently implemented with short usleep). + generalWait.wakeAll(); + rwlock.unlock(); + usleep(5 * 1000); + rwlock.lockForRead(); } - availWaitMutex.unlock(); - - rwlock.unlock(); - - if ((used >= fill_threshold || wantseek || ateof || setswitchtonext) && - !pausereadthread) + else { - usleep(500); + // yield if we have nothing to do... + if (!request_pause && + (used >= fill_threshold || ateof || setswitchtonext)) + { + generalWait.wait(&rwlock, 100); + } } } + rwlock.unlock(); + + rwlock.lockForWrite(); + rbrlock.lockForWrite(); + rbwlock.lockForWrite(); + + rbrpos = 0; + rbwpos = 0; + reallyrunning = false; + readsallowed = false; delete [] readAheadBuffer; + readAheadBuffer = NULL; - rbrpos = 0; - rbwpos = 0; + rbwlock.unlock(); + rbrlock.unlock(); + rwlock.unlock(); } long long RingBuffer::SetAdjustFilesize(void) { + rwlock.lockForWrite(); + poslock.lockForRead(); readAdjust += internalreadpos; + poslock.unlock(); + rwlock.unlock(); return readAdjust; } int RingBuffer::Peek(void *buf, int count) { - long long ret = -1; - - if (!readaheadrunning) - { - long long old_pos = Seek(0, SEEK_CUR); - - ret = Read(buf, count); -#ifdef USING_FRONTEND - if (ret > 0 && dvdPriv) - { - // This is technically incorrect it we should seek - // back to exactly where we were, but we can't do - // that with the DVDRingBuffer - dvdPriv->NormalSeek(0); - } - else if (ret > 0 && bdPriv) - { - // No idea if this will work. - bdPriv->Seek(0); - } - else -#endif // USING_FRONTEND - if (ret > 0) - { - long long new_pos = Seek(-ret, SEEK_CUR); - if (new_pos != old_pos) - { - VERBOSE(VB_IMPORTANT, LOC_ERR + - QString("Peek() Failed to return from new " - "position %1 to old position %2, now " - "at position %3") - .arg(old_pos - ret).arg(old_pos).arg(new_pos)); - } - } - } - else - { - ret = ReadFromBuf(buf, count, true); - } - + int ret = ReadPriv(buf, count, true); if (ret != count) { VERBOSE(VB_IMPORTANT, LOC_WARN + @@ -1197,82 +1348,65 @@ return ret; } -/** - * \brief Reads from the read-ahead buffer, this is called by - * Read(void*, int) when the read-ahead thread is running. - * \param buf Pointer to where data will be written - * \param count Number of bytes to read - * \param peek If true, don't increment read count - * \return Returns number of bytes read - */ -int RingBuffer::ReadFromBuf(void *buf, int count, bool peek) +bool RingBuffer::WaitForReadsAllowed(void) { - if (commserror) - return 0; + MythTimer t; + t.start(); - bool readone = false; - int readErr = 0; - - if (readaheadpaused && stopreads) + while (!readsallowed && !stopreads && + !request_pause && !commserror && readaheadrunning) { - readone = true; - Unpause(); - } - else - { - QMutexLocker locker(&readsAllowedWaitMutex); + generalWait.wait(&rwlock, 1000); + if (!readsallowed && t.elapsed() > 1000) + { + VERBOSE(VB_IMPORTANT, LOC_WARN + + "Taking too long to be allowed to read.."); - while (!readsallowed && !stopreads) - { - if (!readsAllowedWait.wait(&readsAllowedWaitMutex, 1000)) + if (t.elapsed() > 10000) { - VERBOSE(VB_IMPORTANT, - LOC + "Taking too long to be allowed to read.."); - readErr++; - - // HACK Sometimes the readhead thread gets borked on startup. - if ((readErr > 4 && readErr % 2) && (rbrpos ==0)) - { - VERBOSE(VB_IMPORTANT, "restarting readhead thread.."); - KillReadAheadThread(); - StartupReadAheadThread(); - } - - if (readErr > 10) - { - VERBOSE(VB_IMPORTANT, LOC_ERR + "Took more than " - "10 seconds to be allowed to read, aborting."); - wanttoread = 0; - stopreads = true; - return 0; - } + VERBOSE(VB_IMPORTANT, LOC_ERR + "Took more than " + "10 seconds to be allowed to read, aborting."); + return false; } } } + return readsallowed; +} + +bool RingBuffer::WaitForAvail(int count) +{ int avail = ReadBufAvail(); + count = (ateof && avail < count) ? avail : count; - if (ateof && avail < count) - count = avail; - MythTimer t; t.start(); - while (avail < count && !stopreads) + while ((avail < count) && !stopreads && + !request_pause && !commserror && readaheadrunning) { - availWaitMutex.lock(); wanttoread = count; - if (!availWait.wait(&availWaitMutex, 250)) + generalWait.wait(&rwlock, 250); + avail = ReadBufAvail(); + + if ((ateof || setswitchtonext) && avail < count) + count = avail; + + if (avail < count) { int elapsed = t.elapsed(); - if (/*((elapsed > 500) && (elapsed < 750)) ||*/ + if (((elapsed > 250) && (elapsed < 500)) || + ((elapsed > 500) && (elapsed < 750)) || ((elapsed > 1000) && (elapsed < 1250)) || ((elapsed > 2000) && (elapsed < 2250)) || ((elapsed > 4000) && (elapsed < 4250)) || - ((elapsed > 8000) && (elapsed < 8250))) + ((elapsed > 8000) && (elapsed < 8250)) || + ((elapsed > 9000))) { VERBOSE(VB_IMPORTANT, LOC + "Waited " + - QString("%1").arg((elapsed / 500) * 0.5f, 3, 'f', 1) + - " seconds for data to become available..."); + QString("%1").arg((elapsed / 250) * 0.25f, 3, 'f', 1) + + " seconds for data \n\t\t\tto become available..." + + QString(" %2 < %3") + .arg(avail).arg(count)); if (livetvchain) { VERBOSE(VB_IMPORTANT, "Checking to see if there's a " @@ -1295,28 +1429,198 @@ VERBOSE(VB_IMPORTANT, LOC + "Timing out wait due to " "impending livetv switch."); - ateof = true; - wanttoread = 0; - stopreads = true; - availWaitMutex.unlock(); - return 0; + return false; } } + } + wanttoread = 0; + + return avail >= count; +} + +int RingBuffer::ReadDirect(void *buf, int count, bool peek) +{ + long long old_pos = 0; + if (peek) + { + poslock.lockForRead(); + old_pos = (ignorereadpos >= 0) ? ignorereadpos : readpos; + poslock.unlock(); + } + + int ret; + if (remotefile) + ret = safe_read(remotefile, buf, count); +#ifdef USING_FRONTEND + else if (dvdPriv) + ret = dvdPriv->safe_read(buf, count); + else if (bdPriv) + ret = bdPriv->safe_read(buf, count); +#endif // USING_FRONTEND + else if (fd2 >= 0) + ret = safe_read(fd2, buf, count); + else + { + ret = -1; + errno = EBADF; + } + + poslock.lockForWrite(); + if (ignorereadpos >= 0 && ret > 0) + { + if (peek) + { + // seek should always succeed since we were at this position + if (remotefile) + remotefile->Seek(old_pos, SEEK_SET); + else + lseek64(fd2, old_pos, SEEK_SET); + } + else + { + ignorereadpos += ret; + } + poslock.unlock(); + return ret; + } + poslock.unlock(); + + if (peek && ret > 0) + { + if (!dvdPriv && !bdPriv) + { + long long new_pos = Seek(old_pos, SEEK_SET, true); + if (new_pos != old_pos) + { + VERBOSE(VB_IMPORTANT, LOC_ERR + + QString("Peek() Failed to return from new " + "position %1 to old position %2, now " + "at position %3") + .arg(old_pos - ret).arg(old_pos).arg(new_pos)); + } + } + else if (old_pos != 0) + { + VERBOSE(VB_IMPORTANT, LOC_ERR + + "DVD and Blu-Ray do not support arbitrary " + "peeks except when read-ahead is enabled." + "\n\t\t\tWill seek to beginning of video."); + } +#ifdef USING_FRONTEND + if (dvdPriv) + { + dvdPriv->NormalSeek(0); + } + else if (bdPriv) + { + bdPriv->Seek(0); + } +#endif // USING_FRONTEND + } + + return ret; +} + +/** \brief When possible reads from the read-ahead buffer, + * otherwise reads directly from the device. + * + * \param buf Pointer to where data will be written + * \param count Number of bytes to read + * \param peek If true, don't increment read count + * \return Returns number of bytes read + */ +int RingBuffer::ReadPriv(void *buf, int count, bool peek) +{ + VERBOSE(VB_FILE|VB_EXTRA, LOC + + QString("ReadPriv(*, %1, %2) @%3 -- begin") + .arg(count).arg(peek?"peek":"normal").arg(rbrpos)); + + rwlock.lockForRead(); + if (writemode) + { + VERBOSE(VB_IMPORTANT, LOC_ERR + + "Attempt to read from a write only file"); + errno = EBADF; + rwlock.unlock(); + return -1; + } + + if (commserror) + { + VERBOSE(VB_IMPORTANT, LOC_ERR + + "Attempt to read after commserror set"); + errno = EIO; + rwlock.unlock(); + return -1; + } + + if (request_pause || stopreads || !readaheadrunning || (ignorereadpos>=0)) + { + rwlock.unlock(); + rwlock.lockForWrite(); + // we need a write lock so the read-ahead thread + // can't start mucking with the read position. + // If the read ahead thread was started while we + // didn't hold the lock, we proceed with a normal + // read from the buffer, otherwise we read directly. + if (request_pause || stopreads || + !readaheadrunning || (ignorereadpos >= 0)) + { + int ret = ReadDirect(buf, count, peek); + VERBOSE(VB_FILE|VB_EXTRA, LOC + + QString("ReadPriv(*, %1, %2) @%3 -- RD checksum %4") + .arg(count).arg(peek?"peek":"normal").arg(rbrpos) + .arg(qChecksum((char*)buf,count))); + rwlock.unlock(); + return ret; + } + rwlock.unlock(); + rwlock.lockForRead(); + } + + if (!WaitForReadsAllowed()) + { + VERBOSE(VB_FILE, LOC + "!WaitForReadsAllowed()"); + rwlock.unlock(); + rwlock.lockForWrite(); wanttoread = 0; - availWaitMutex.unlock(); + stopreads = true; + rwlock.unlock(); + return 0; + } - avail = ReadBufAvail(); - if ((ateof || setswitchtonext) && avail < count) - count = avail; + if (!WaitForAvail(count)) + { + VERBOSE(VB_FILE, LOC + "!WaitForAvail()"); + rwlock.unlock(); + rwlock.lockForWrite(); + ateof = true; + wanttoread = 0; + stopreads = true; + rwlock.unlock(); + return 0; + } - if (commserror) - return 0; + count = min(ReadBufAvail(), count); + + if (count <= 0) + { + // this can happen under a few conditions but the most + // notable is an exit from the read ahead thread. + rwlock.unlock(); + return count; } - if ((ateof || stopreads) && avail < count) - count = avail; + if (peek) + rbrlock.lockForRead(); + else + rbrlock.lockForWrite(); + VERBOSE(VB_FILE|VB_EXTRA, LOC + + QString("ReadPriv(*, %1, %2) @%3 -- copying data") + .arg(count).arg(peek?"peek":"normal").arg(rbrpos)); + if (rbrpos + count > (int) kBufferSize) { int firstsize = kBufferSize - rbrpos; @@ -1326,21 +1630,22 @@ memcpy((char *)buf + firstsize, readAheadBuffer, secondsize); } else + { memcpy(buf, readAheadBuffer + rbrpos, count); + VERBOSE(VB_FILE|VB_EXTRA, LOC + + QString("ReadPriv(*, %1, %2) @%3 -- checksum %4") + .arg(count).arg(peek?"peek":"normal").arg(rbrpos) + .arg(qChecksum(readAheadBuffer+rbrpos,count))); + } if (!peek) { - readAheadLock.lock(); rbrpos = (rbrpos + count) % kBufferSize; - readAheadLock.unlock(); + generalWait.wakeAll(); } + rbrlock.unlock(); + rwlock.unlock(); - if (readone) - { - Pause(); - WaitForPause(); - } - return count; } @@ -1354,53 +1659,18 @@ */ int RingBuffer::Read(void *buf, int count) { - int ret = -1; - if (writemode) + int ret = ReadPriv(buf, count, false); + if (ret > 0) { - VERBOSE(VB_IMPORTANT, LOC_ERR + - "Attempt to read from a write only file"); - return ret; - } - - rwlock.lockForRead(); - - if (!readaheadrunning) - { - if (remotefile) - { - ret = safe_read(remotefile, buf, count); - readpos += ret; - } -#ifdef USING_FRONTEND - else if (dvdPriv) - { - ret = dvdPriv->safe_read(buf, count); - readpos += ret; - } - else if (bdPriv) - { - ret = bdPriv->safe_read(buf, count); - readpos += ret; - } -#endif // USING_FRONTEND - else - { - ret = safe_read(fd2, buf, count); - readpos += ret; - } - } - else - { - ret = ReadFromBuf(buf, count); + poslock.lockForWrite(); readpos += ret; + poslock.unlock(); } - - rwlock.unlock(); return ret; } /** \fn RingBuffer::IsIOBound(void) const - * \brief Returns true if a RingBuffer::Read(void*,int) is likely to block. + * \brief Returns true if a RingBuffer::Write(void*,int) is likely to block. */ bool RingBuffer::IsIOBound(void) const { @@ -1429,25 +1699,36 @@ */ int RingBuffer::Write(const void *buf, uint count) { - int ret = -1; + rwlock.lockForRead(); + if (!writemode) { VERBOSE(VB_IMPORTANT, LOC_ERR + "Tried to write to a read only file."); - return ret; + rwlock.unlock(); + return -1; } if (!tfw && !remotefile) - return ret; + { + rwlock.unlock(); + return -1; + } - rwlock.lockForRead(); - + int ret = -1; if (tfw) ret = tfw->Write(buf, count); else ret = remotefile->Write(buf, count); - writepos += ret; + if (ret > 0) + { + poslock.lockForWrite(); + writepos += ret; + poslock.unlock(); + } + rwlock.unlock(); + return ret; } @@ -1456,69 +1737,241 @@ */ void RingBuffer::Sync(void) { + rwlock.lockForRead(); if (tfw) tfw->Sync(); + rwlock.unlock(); } -/** \fn RingBuffer::Seek(long long, int) - * \brief Seeks to a particular position in the file. +/** \brief Seeks to a particular position in the file. */ -long long RingBuffer::Seek(long long pos, int whence) +long long RingBuffer::Seek(long long pos, int whence, bool has_lock) { + VERBOSE(VB_FILE, LOC + QString("Seek(%1,%2,%3)") + .arg(pos).arg((SEEK_SET==whence)?"SEEK_SET": + ((SEEK_CUR==whence)?"SEEK_CUR":"SEEK_END")) + .arg(has_lock?"locked":"unlocked")); + + long long ret = -1; + + stopreads = true; + + // lockForWrite takes priority over lockForRead, so this will + // take priority over the lockForRead in the read ahead thread. + if (!has_lock) + rwlock.lockForWrite(); + + stopreads = false; + if (writemode) - return WriterSeek(pos, whence); + { + ret = WriterSeek(pos, whence, true); + if (!has_lock) + rwlock.unlock(); + return ret; + } - wantseek = true; - rwlock.lockForWrite(); - wantseek = false; + poslock.lockForWrite(); - // optimize nop seeks - if ((whence == SEEK_SET && pos == readpos) || - (whence == SEEK_CUR && pos == 0)) + // Optimize no-op seeks + if (readaheadrunning && + ((whence == SEEK_SET && pos == readpos) || + (whence == SEEK_CUR && pos == 0))) { - rwlock.unlock(); - return readpos; + ret = readpos; + + poslock.unlock(); + if (!has_lock) + rwlock.unlock(); + + return ret; } - errno = 0; // clear errno, in case of remotefile error + // only valid for SEEK_SET & SEEK_CUR + long long new_pos = (SEEK_SET==whence) ? pos : readpos + pos; - long long ret = -1; + // Optimize short seeks where the data for + // them is in our ringbuffer already. + if (readaheadrunning && + (SEEK_SET==whence || SEEK_CUR==whence)) + { + rbrlock.lockForWrite(); + rbwlock.lockForRead(); + VERBOSE(VB_FILE, LOC + + QString("Seek(): rbrpos: %1 rbwpos: %2" + "\n\t\t\treadpos: %3 internalreadpos: %4") + .arg(rbrpos).arg(rbwpos) + .arg(readpos).arg(internalreadpos)); + bool used_opt = false; + if ((new_pos < readpos)) + { + int min_safety = max(fill_min, readblocksize); + int free = ((rbwpos >= rbrpos) ? + rbrpos + kBufferSize : rbrpos) - rbwpos; + int internal_backbuf = + (rbwpos >= rbrpos) ? rbrpos : rbrpos - rbwpos; + internal_backbuf = min(internal_backbuf, free - min_safety); + long long sba = readpos - new_pos; + VERBOSE(VB_FILE, LOC + + QString("Seek(): internal_backbuf: %1 sba: %2") + .arg(internal_backbuf).arg(sba)); + if (internal_backbuf >= sba) + { + rbrpos = (rbrpos>=sba) ? rbrpos - sba : + kBufferSize + rbrpos - sba; + used_opt = true; + VERBOSE(VB_FILE, LOC + + QString("Seek(): OPT1 rbrpos: %1 rbwpos: %2" + "\n\t\t\treadpos: %3 internalreadpos: %4") + .arg(rbrpos).arg(rbwpos) + .arg(new_pos).arg(internalreadpos)); + } + } + else if ((new_pos >= readpos) && (new_pos <= internalreadpos)) + { + rbrpos = (rbrpos + (new_pos - readpos)) % kBufferSize; + used_opt = true; + VERBOSE(VB_IMPORTANT, LOC + + QString("Seek(): OPT2 rbrpos: %1 sba: %2") + .arg(rbrpos).arg(readpos - new_pos)); + } + rbwlock.unlock(); + rbrlock.unlock(); + + if (used_opt) + { + if (ignorereadpos >= 0) + { + // seek should always succeed since we were at this position + if (remotefile) + remotefile->Seek(internalreadpos, SEEK_SET); + else + lseek64(fd2, internalreadpos, SEEK_SET); + ignorereadpos = -1; + } + readpos = new_pos; + poslock.unlock(); + generalWait.wakeAll(); + if (!has_lock) + rwlock.unlock(); + return new_pos; + } + } + + // This optimizes the seek end-250000, read, seek 0, read portion + // of the pattern ffmpeg performs at the start of playback to + // determine the pts. + // If the seek is a SEEK_END or is a seek where the position + // changes over 100 MB we check the file size and if the + // destination point is within 300000 bytes of the end of + // the file we enter a special mode where the read ahead + // buffer stops reading data and all reads are made directly + // until another seek is performed. The point of all this is + // to avoid flushing out the buffer that still contains all + // the data the final seek 0, read will need just to read the + // last 250000 bytes. A further optimization would be to buffer + // the 250000 byte read, which is currently performed in 32KB + // blocks (inefficient with RemoteFile). + if ((remotefile || fd2 >= 0) && (ignorereadpos < 0)) + { + long long off_end = 0xDEADBEEF; + if (SEEK_END == whence) + { + off_end = pos; + if (remotefile) + { + new_pos = remotefile->GetFileSize() - off_end; + } + else + { + QFileInfo fi(filename); + new_pos = fi.size() - off_end; + } + } + else if (abs(new_pos-readpos) > 100000000) + { + if (remotefile) + { + off_end = remotefile->GetFileSize() - new_pos; + } + else + { + QFileInfo fi(filename); + off_end = fi.size() - new_pos; + } + } + if (off_end < 300000) + { + VERBOSE(VB_FILE, LOC + + QString("Seek(): offset from end: %1").arg(off_end) + + "\n\t\t\t -- ignoring read ahead thread until next seek."); + + ignorereadpos = new_pos; + errno = EINVAL; + int ret; + if (remotefile) + ret = remotefile->Seek(ignorereadpos, SEEK_SET); + else + ret = lseek64(fd2, ignorereadpos, SEEK_SET); + + if (ret < 0) + { + ignorereadpos = -1; + QString cmd = QString("Seek(%1, %2)").arg(pos) + .arg((SEEK_SET == whence) ? "SEEK_SET" : + ((SEEK_CUR == whence) ?"SEEK_CUR" : "SEEK_END")); + VERBOSE(VB_IMPORTANT, LOC_ERR + cmd + " Failed" + ENO); + } + + rbwlock.unlock(); + rbrlock.unlock(); + poslock.unlock(); + + generalWait.wakeAll(); + + if (!has_lock) + rwlock.unlock(); + + return ret; + } + } + + // Here we perform a normal seek. When successful we + // need to call ResetReadAhead(). A reset means we will + // need to refill the buffer, which takes some time. if (remotefile) + { ret = remotefile->Seek(pos, whence, readpos); + if (ret<0) + errno = EINVAL; + } #ifdef USING_FRONTEND + else if ((dvdPriv || bdPriv) && (SEEK_END == whence)) + { + errno = EINVAL; + ret = -1; + } else if (dvdPriv) { - dvdPriv->NormalSeek(pos); - ret = pos; + dvdPriv->NormalSeek(new_pos); + ret = new_pos; } else if (bdPriv) { - bdPriv->Seek(pos); - ret = pos; + bdPriv->Seek(new_pos); + ret = new_pos; } #endif // USING_FRONTEND else { - if ((whence == SEEK_SET) || (whence == SEEK_END)) -#ifdef USING_MINGW - ret = lseek64(fd2, pos, whence); -#else - ret = lseek(fd2, pos, whence); -#endif - else - { - long long realseek = readpos + pos; -#ifdef USING_MINGW - ret = lseek64(fd2, realseek, SEEK_SET); -#else - ret = lseek(fd2, realseek, SEEK_SET); -#endif - } + ret = lseek64(fd2, pos, whence); } if (ret >= 0) { readpos = ret; + + ignorereadpos = -1; if (readaheadrunning) ResetReadAhead(readpos); @@ -1533,24 +1986,38 @@ VERBOSE(VB_IMPORTANT, LOC_ERR + cmd + " Failed" + ENO); } - rwlock.unlock(); + poslock.unlock(); + generalWait.wakeAll(); + + if (!has_lock) + rwlock.unlock(); + return ret; } -/** \fn RingBuffer::WriterSeek(long long, int) - * \brief Calls ThreadedFileWriter::Seek(long long,int). +/** \brief Calls ThreadedFileWriter::Seek(long long,int). */ -long long RingBuffer::WriterSeek(long long pos, int whence) +long long RingBuffer::WriterSeek(long long pos, int whence, bool has_lock) { long long ret = -1; + if (!has_lock) + rwlock.lockForRead(); + + poslock.lockForWrite(); + if (tfw) { ret = tfw->Seek(pos, whence); writepos = ret; } + poslock.unlock(); + + if (!has_lock) + rwlock.unlock(); + return ret; } @@ -1560,11 +2027,13 @@ */ void RingBuffer::WriterFlush(void) { + rwlock.lockForRead(); if (tfw) { tfw->Flush(); tfw->Sync(); } + rwlock.unlock(); } /** \fn RingBuffer::SetWriteBufferSize(int) @@ -1572,8 +2041,10 @@ */ void RingBuffer::SetWriteBufferSize(int newSize) { + rwlock.lockForRead(); if (tfw) tfw->SetWriteBufferSize(newSize); + rwlock.unlock(); } /** \fn RingBuffer::SetWriteBufferMinWriteSize(int) @@ -1581,8 +2052,10 @@ */ void RingBuffer::SetWriteBufferMinWriteSize(int newMinSize) { + rwlock.lockForRead(); if (tfw) tfw->SetWriteBufferMinWriteSize(newMinSize); + rwlock.unlock(); } /** \fn RingBuffer::GetReadPosition(void) const @@ -1590,14 +2063,18 @@ */ long long RingBuffer::GetReadPosition(void) const { + rwlock.lockForRead(); + poslock.lockForRead(); + long long ret = readpos; #ifdef USING_FRONTEND if (dvdPriv) - return dvdPriv->GetReadPosition(); + ret = dvdPriv->GetReadPosition(); else if (bdPriv) - return bdPriv->GetReadPosition(); + ret = bdPriv->GetReadPosition(); #endif // USING_FRONTEND - - return readpos; + poslock.unlock(); + rwlock.unlock(); + return ret; } /** \fn RingBuffer::GetWritePosition(void) const @@ -1605,7 +2082,10 @@ */ long long RingBuffer::GetWritePosition(void) const { - return writepos; + poslock.lockForRead(); + long long ret = writepos; + poslock.unlock(); + return ret; } /** \fn RingBuffer::GetRealFileSize(void) const @@ -1614,11 +2094,14 @@ */ long long RingBuffer::GetRealFileSize(void) const { + rwlock.lockForRead(); + long long ret = -1; if (remotefile) - return remotefile->GetFileSize(); - - QFileInfo info(filename); - return info.size(); + ret = remotefile->GetFileSize(); + else + ret = QFileInfo(filename).size(); + rwlock.unlock(); + return ret; } /** \fn RingBuffer::LiveMode(void) const @@ -1627,7 +2110,10 @@ */ bool RingBuffer::LiveMode(void) const { - return (livetvchain); + rwlock.lockForRead(); + bool ret = (livetvchain); + rwlock.unlock(); + return ret; } /** \fn RingBuffer::SetLiveMode(LiveTVChain*) @@ -1636,16 +2122,21 @@ */ void RingBuffer::SetLiveMode(LiveTVChain *chain) { + rwlock.lockForWrite(); livetvchain = chain; + rwlock.unlock(); } bool RingBuffer::InDVDMenuOrStillFrame(void) { + rwlock.lockForRead(); + bool ret = false; #ifdef USING_FRONTEND if (dvdPriv) - return (dvdPriv->IsInMenu() || dvdPriv->InStillFrame()); + ret = (dvdPriv->IsInMenu() || dvdPriv->InStillFrame()); #endif // USING_FRONTEND - return false; + rwlock.unlock(); + return ret; } /* vim: set expandtab tabstop=4 shiftwidth=4: */ Index: libs/libmythtv/RingBuffer.h =================================================================== --- libs/libmythtv/RingBuffer.h (revision 25873) +++ libs/libmythtv/RingBuffer.h (working copy) @@ -1,7 +1,7 @@ // -*- Mode: c++ -*- -#ifndef RINGBUFFER -#define RINGBUFFER +#ifndef _RINGBUFFER_H_ +#define _RINGBUFFER_H_ #include #include @@ -35,30 +35,41 @@ void SetWriteBufferMinWriteSize(int newMinSize); void UpdateRawBitrate(uint rawbitrate); void UpdatePlaySpeed(float playspeed); - void SetStreamOnly(bool stream) { streamOnly = stream; } + void SetStreamOnly(bool stream) + { + rwlock.lockForWrite(); + streamOnly = stream; + rwlock.unlock(); + } // Gets /// Returns name of file used by this RingBuffer - QString GetFilename(void) const { return filename; } - QString GetSubtitleFilename(void) const { return subtitlefilename; } - /// Returns ReadBufAvail(void) - int DataInReadAhead(void) const { return ReadBufAvail(); } + QString GetFilename(void) const + { + rwlock.lockForRead(); + QString tmp = filename; tmp.detach(); + rwlock.unlock(); + return tmp; + } + QString GetSubtitleFilename(void) const + { + rwlock.lockForRead(); + QString tmp = subtitlefilename; tmp.detach(); + rwlock.unlock(); + return tmp; + } /// Returns value of stopreads /// \sa StartReads(void), StopReads(void) bool GetStopReads(void) const { return stopreads; } - /// Returns false iff read-ahead is not - /// running and read-ahead is not paused. - bool isPaused(void) const - { return (!readaheadrunning) ? true : readaheadpaused; } + bool isPaused(void) const; long long GetReadPosition(void) const; long long GetWritePosition(void) const; long long GetRealFileSize(void) const; - uint GetBitrate(void) const; - uint GetReadBlockSize(void) const; bool IsOpen(void) const; + bool IsNearEnd(double fps, uint vvf) const; // General Commands - void OpenFile(const QString &lfilename, uint retryCount = 12/*4*/); + void OpenFile(const QString &lfilename, uint retry_ms = 2000); int Read(void *buf, int count); int Peek(void *buf, int count); // only works with readahead @@ -67,7 +78,7 @@ bool resetInternal = false); // Seeks - long long Seek(long long pos, int whence); + long long Seek(long long pos, int whence, bool has_lock = false); // Pause commands void Pause(void); @@ -83,19 +94,34 @@ bool LiveMode(void) const; void SetLiveMode(LiveTVChain *chain); /// Tells RingBuffer whether to igonre the end-of-file - void IgnoreLiveEOF(bool ignore) { ignoreliveeof = ignore; } + void IgnoreLiveEOF(bool ignore) + { + rwlock.lockForWrite(); + ignoreliveeof = ignore; + rwlock.unlock(); + } // ThreadedFileWriter proxies int Write(const void *buf, uint count); bool IsIOBound(void) const; void WriterFlush(void); void Sync(void); - long long WriterSeek(long long pos, int whence); + long long WriterSeek(long long pos, int whence, bool has_lock = false); // DVDRingBuffer proxies /// Returns true if this is a DVD backed RingBuffer. - inline bool isDVD(void) const { return dvdPriv; } - DVDRingBufferPriv *DVD() { return dvdPriv; } + bool isDVD(void) const + { + rwlock.lockForRead(); + bool ret = dvdPriv; + rwlock.unlock(); + return ret; + } + /// Illicitly manipulating privates is ill advised! + DVDRingBufferPriv *DVD() + { + return dvdPriv; + } bool InDVDMenuOrStillFrame(void); // BDRingBuffer proxies @@ -109,93 +135,88 @@ protected: void run(void); // QThread void CalcReadAheadThresh(void); + bool PauseAndWait(void); int safe_read_bd(void *data, uint sz); int safe_read_dvd(void *data, uint sz); int safe_read(int fd, void *data, uint sz); int safe_read(RemoteFile *rf, void *data, uint sz); - int ReadFromBuf(void *buf, int count, bool peek = false); + int ReadPriv(void *buf, int count, bool peek); + int ReadDirect(void *buf, int count, bool peek); + bool WaitForReadsAllowed(void); + bool WaitForAvail(int count); int ReadBufFree(void) const; int ReadBufAvail(void) const; - void StartupReadAheadThread(void); void ResetReadAhead(long long newinternal); void KillReadAheadThread(void); private: - // NR == trivial risk, not protected, but only modified in single thread - // LR == low risk, not protected, but only modified on Open,Close,ctor,dtor - // HR == high risk, likely to cause unexpected behaviour - // MR == medium risk, unsafe methods unlikely to be called at wrong moment + mutable QReadWriteLock poslock; + long long readpos; // protected by poslock + long long writepos; // protected by poslock + long long internalreadpos; // protected by poslock + long long ignorereadpos; // protected by poslock + mutable QReadWriteLock rbrlock; + int rbrpos; // protected by rbrlock + mutable QReadWriteLock rbwlock; + int rbwpos; // protected by rbwlock - QString filename; // not protected by a lock LR - QString subtitlefilename; // not protected by a lock LR + // note should not go under rwlock.. + // this is used to break out of read_safe where rwlock is held + volatile bool stopreads; - ThreadedFileWriter *tfw; // not protected by a lock LR - int fd2; // not protected by a lock LR - - bool writemode; // not protected by a lock LR - - long long readpos; // not protected by a lock HR - long long writepos; // not protected by a lock HR - - bool stopreads; // not protected by a lock HR - mutable QReadWriteLock rwlock; - RemoteFile *remotefile; // not protected by a lock LR + QString filename; // protected by rwlock + QString subtitlefilename; // protected by rwlock - // this lock does not consistently protect anything, - // but seems to be intented to protect rbrpos & rbwpos - mutable QMutex readAheadLock; + ThreadedFileWriter *tfw; // protected by rwlock + int fd2; // protected by rwlock - bool startreadahead; // not protected by a lock HR - char *readAheadBuffer; // not protected by a lock MR - bool readaheadrunning; // not protected by a lock HR - bool readaheadpaused; // not protected by a lock HR - bool pausereadthread; // not protected by a lock HR - int rbrpos; // not protected by a lock HR - int rbwpos; // not protected by a lock HR - long long internalreadpos; // not protected by a lock HR - bool ateof; // not protected by a lock HR - bool readsallowed; // not protected by a lock HR - volatile bool wantseek; // not protected by a lock HR - bool setswitchtonext; // protected by rwlock - bool streamOnly; + bool writemode; // protected by rwlock - uint rawbitrate; // protected by rwlock - float playspeed; // protected by rwlock - int fill_threshold;// not protected by a lock HR - int fill_min; // protected by rwlock - int readblocksize; // protected by rwlock + RemoteFile *remotefile; // protected by rwlock - QWaitCondition pauseWait; // not protected by a lock HR + bool startreadahead; // protected by rwlock + char *readAheadBuffer; // protected by rwlock + bool readaheadrunning; // protected by rwlock + bool reallyrunning; // protected by rwlock + bool request_pause; // protected by rwlock + bool paused; // protected by rwlock + bool ateof; // protected by rwlock + bool readsallowed; // protected by rwlock + bool setswitchtonext; // protected by rwlock + bool streamOnly; // protected by rwlock + bool ignorereadahead; // protected by rwlock + uint rawbitrate; // protected by rwlock + float playspeed; // protected by rwlock + int fill_threshold; // protected by rwlock + int fill_min; // protected by rwlock + int readblocksize; // protected by rwlock + int wanttoread; // protected by rwlock + int numfailures; // protected by rwlock (see note 1) + bool commserror; // protected by rwlock - int wanttoread; // not protected by a lock HR - QWaitCondition availWait; // protected by availWaitMutex - QMutex availWaitMutex; + DVDRingBufferPriv *dvdPriv; // not protected by rwlock, when DVD() is used + BDRingBufferPriv *bdPriv; // protected by rwlock - QWaitCondition readsAllowedWait;// protected by readsAllowedWaitMutex - QMutex readsAllowedWaitMutex; + bool oldfile; // protected by rwlock - int numfailures; // not protected by a lock MR + LiveTVChain *livetvchain; // protected by rwlock + bool ignoreliveeof; // protected by rwlock - bool commserror; // not protected by a lock MR + long long readAdjust; // protected by rwlock - DVDRingBufferPriv *dvdPriv; // not protected by a lock LR - BDRingBufferPriv *bdPriv; // not protected by a lock LR + // note 1: numfailures is modified with only a read lock in the + // read ahead thread, but this is safe since all other places + // that use it are protected by a write lock. But this is a + // fragile state of affairs and care must be taken when modifying + // code or locking around this variable. - bool oldfile; // not protected by a lock LR - - LiveTVChain *livetvchain; // not protected by a lock HR - bool ignoreliveeof; // not protected by a lock HR - - long long readAdjust; // not protected by a lock HR - /// Condition to signal that the read ahead thread is running - QWaitCondition readAheadRunningCond;//protected by readAheadRunningCondLock - QMutex readAheadRunningCondLock; + QWaitCondition generalWait; // protected by rwlock public: static QMutex subExtLock; @@ -208,4 +229,4 @@ static const uint kReadTestSize; }; -#endif +#endif // _RINGBUFFER_H_ Index: libs/libmythtv/mythplayer.cpp =================================================================== --- libs/libmythtv/mythplayer.cpp (revision 25873) +++ libs/libmythtv/mythplayer.cpp (working copy) @@ -893,6 +893,8 @@ void MythPlayer::OpenDummy(void) { + VERBOSE(VB_IMPORTANT, LOC + "OpenDummy()"); + isDummy = true; float displayAspect = @@ -936,14 +938,24 @@ SetDecoder(NULL); int testreadsize = 2048; + MythTimer bigTimer; bigTimer.start(); + int timeout = (retries + 1) * 500; while (testreadsize <= kDecoderProbeBufferSize) { - if (player_ctx->buffer->Peek(testbuf, testreadsize) != testreadsize) + MythTimer peekTimer; peekTimer.start(); + while (player_ctx->buffer->Peek(testbuf, testreadsize) != testreadsize) { - VERBOSE(VB_IMPORTANT, LOC_ERR + - QString("OpenFile(): Error, couldn't read file: %1") - .arg(player_ctx->buffer->GetFilename())); - return -1; + if (peekTimer.elapsed() > 1000 || bigTimer.elapsed() > timeout) + { + VERBOSE(VB_IMPORTANT, LOC_ERR + + QString("OpenFile(): Could not read " + "first %1 bytes of '%2'") + .arg(testreadsize) + .arg(player_ctx->buffer->GetFilename())); + return -1; + } + VERBOSE(VB_IMPORTANT, LOC_WARN + "OpenFile() waiting on data"); + usleep(50 * 1000); } player_ctx->LockPlayingInfo(__FILE__, __LINE__); @@ -960,7 +972,7 @@ player_ctx->GetSpecialDecode())); } player_ctx->UnlockPlayingInfo(__FILE__, __LINE__); - if (GetDecoder()) + if (GetDecoder() || (bigTimer.elapsed() > timeout)) break; testreadsize <<= 1; } @@ -2168,12 +2180,14 @@ return; } - uint retries = 10; // about 5 seconds of retries - player_ctx->buffer->OpenFile(pginfo->GetPlaybackURL(), retries); + player_ctx->buffer->OpenFile(pginfo->GetPlaybackURL(), 2000 /*ms*/); if (!player_ctx->buffer->IsOpen()) { - VERBOSE(VB_IMPORTANT, LOC_ERR + "SwitchToProgram's OpenFile failed."); + VERBOSE(VB_IMPORTANT, LOC_ERR + "SwitchToProgram's OpenFile failed " + + QString("(card type: %1).") + .arg(player_ctx->tvchain->GetCardType(newid))); + VERBOSE(VB_IMPORTANT, QString("\n") + player_ctx->tvchain->toString()); eof = true; SetErrored(QObject::tr("Error opening switch program buffer")); delete pginfo; @@ -2290,10 +2304,14 @@ SendMythSystemPlayEvent("PLAY_CHANGED", pginfo); - player_ctx->buffer->OpenFile(pginfo->GetPlaybackURL()); + player_ctx->buffer->OpenFile(pginfo->GetPlaybackURL(), 2000 /*ms*/); if (!player_ctx->buffer->IsOpen()) { - VERBOSE(VB_IMPORTANT, LOC_ERR + "JumpToProgram's OpenFile failed."); + VERBOSE(VB_IMPORTANT, LOC_ERR + "JumpToProgram's OpenFile failed " + + QString("(card type: %1).") + .arg(player_ctx->tvchain->GetCardType(newid))); + VERBOSE(VB_IMPORTANT, QString("\n") + player_ctx->tvchain->toString()); + eof = true; SetErrored(QObject::tr("Error opening jump program file buffer")); delete pginfo; @@ -3190,29 +3208,9 @@ if (!videoOutput) return false; - int sz = player_ctx->buffer->DataInReadAhead(); - uint rbs = player_ctx->buffer->GetReadBlockSize(); - uint kbits_per_sec = player_ctx->buffer->GetBitrate(); - uint vvf = videoOutput->ValidVideoFrames(); - double inv_fps = 1.0 / GetDecoder()->GetFPS(); - double bytes_per_frame = kbits_per_sec * (1000.0/8.0) * inv_fps; - double rh_frames = sz / bytes_per_frame; - - // WARNING: rh_frames can greatly overestimate or underestimate - // the number of frames available in the read ahead buffer - // when rh_frames is less than the keyframe distance. - - bool near_end = ((vvf + rh_frames) < 10.0) || (sz < rbs*1.5); - - VERBOSE(VB_PLAYBACK, LOC + "IsReallyNearEnd()" - <<" br("<<(kbits_per_sec/8)<<"KB)" - <<" fps("<<((uint)(1.0/inv_fps))<<")" - <<" sz("<<(sz / 1000)<<"KB)" - <<" vfl("<buffer->IsNearEnd( + GetDecoder()->GetFPS(), + videoOutput->ValidVideoFrames()); } /** \brief Returns true iff near end of recording.