Index: libs/libmythtv/RingBuffer.cpp =================================================================== --- libs/libmythtv/RingBuffer.cpp (revision 25788) +++ libs/libmythtv/RingBuffer.cpp (working copy) @@ -19,8 +19,8 @@ using namespace std; +#include "mythcontext.h" // for VERBOSE #include "mythconfig.h" - #include "exitcodes.h" #include "RingBuffer.h" #include "remotefile.h" @@ -30,7 +30,6 @@ #include "BDRingBuffer.h" #include "util.h" #include "compat.h" -#include "mythverbose.h" #ifndef O_STREAMING #define O_STREAMING 0 @@ -65,9 +64,7 @@ /* Locking relations: - rwlock->readAheadLock - ->readsAllowedWaitMutex->readAheadRunningCondLock - ->availWaitMutex + rwlock->poslock A child should never lock any of the parents without locking the parent lock before the child lock. @@ -113,17 +110,18 @@ RingBuffer::RingBuffer(const QString &lfilename, bool write, bool readahead, uint read_retries) - : filename(lfilename), subtitlefilename(QString::null), + : readpos(0), writepos(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), + readaheadrunning(false), + request_pause(false), paused(false), rbrpos(0), rbwpos(0), internalreadpos(0), ateof(false), - readsallowed(false), wantseek(false), setswitchtonext(false), + readsallowed(false), + setswitchtonext(false), rawbitrate(4000), playspeed(1.0f), fill_threshold(65536), fill_min(-1), readblocksize(CHUNK), wanttoread(0), @@ -273,6 +271,8 @@ VERBOSE(VB_PLAYBACK, LOC + QString("OpenFile(%1, %2)") .arg(lfilename).arg(retryCount)); + rwlock.lockForWrite(); + uint openAttempts = retryCount + 1; filename = lfilename; @@ -414,16 +414,12 @@ 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 @@ -478,7 +474,10 @@ commserror = false; numfailures = 0; - UpdateRawBitrate(4000); + rawbitrate = 4000; + CalcReadAheadThresh(); + + rwlock.unlock(); } /** \fn RingBuffer::IsOpen(void) const @@ -486,12 +485,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) @@ -535,27 +538,14 @@ 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; rwlock.lockForWrite(); - wantseek = false; + poslock.lockForWrite(); + numfailures = 0; commserror = false; setswitchtonext = false; @@ -579,6 +569,7 @@ if (resetInternal) internalreadpos = readpos; + poslock.unlock(); rwlock.unlock(); } @@ -695,31 +686,6 @@ 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) @@ -743,7 +709,6 @@ { uint estbitrate = 0; - wantseek = false; readsallowed = false; readblocksize = CHUNK; @@ -785,23 +750,46 @@ .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; } -/** \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; + return (rbwpos >= rbrpos) ? rbwpos - rbrpos : kBufferSize - rbrpos + rbwpos; } /** \fn RingBuffer::ResetReadAhead(long long) @@ -817,7 +805,6 @@ */ void RingBuffer::ResetReadAhead(long long newinternal) { - readAheadLock.lock(); readblocksize = CHUNK; rbrpos = 0; rbwpos = 0; @@ -825,22 +812,55 @@ ateof = false; readsallowed = false; setswitchtonext = false; - readAheadLock.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; + VERBOSE(VB_IMPORTANT, LOC + "Start)"); - readAheadRunningCondLock.lock(); + bool do_start = true; + + 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; + } + QThread::start(); - readAheadRunningCond.wait(&readAheadRunningCondLock); - readAheadRunningCondLock.unlock(); + readAheadRunningWait.wait(&rwlock); + rwlock.unlock(); } /** \fn RingBuffer::KillReadAheadThread(void) @@ -848,11 +868,18 @@ */ void RingBuffer::KillReadAheadThread(void) { - if (!readaheadrunning) - return; + rwlock.lockForWrite(); + bool do_wait = readaheadrunning; + readaheadrunning = false; + readsAllowedWait.wakeAll(); + pauseWait.wakeAll(); + unpauseWait.wakeAll(); + availWait.wakeAll(); + readsAllowedWait.wakeAll(); + rwlock.unlock(); - readaheadrunning = false; - QThread::wait(); + if (do_wait) + QThread::wait(); } /** \fn RingBuffer::StopReads(void) @@ -880,8 +907,11 @@ */ void RingBuffer::Pause(void) { - pausereadthread = true; StopReads(); + + rwlock.lockForWrite(); + request_pause = true; + rwlock.unlock(); } /** \fn RingBuffer::Unpause(void) @@ -891,28 +921,85 @@ void RingBuffer::Unpause(void) { StartReads(); - pausereadthread = false; + + rwlock.lockForWrite(); + request_pause = false; + unpauseWait.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(); + pauseWait.wait(&rwlock, 1000); + if (readaheadrunning && !paused && request_pause) + { + VERBOSE(VB_IMPORTANT, LOC + + 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; + pauseWait.wakeAll(); + } + + rwlock.unlock(); + rwlock.lockForRead(); + } + + if (request_pause && paused && readaheadrunning) + unpauseWait.wait(&rwlock, timeout); } + + if (!request_pause && paused) + { + rwlock.unlock(); + rwlock.lockForWrite(); + + if (!request_pause) + { + paused = false; + unpauseWait.wakeAll(); + } + + rwlock.unlock(); + rwlock.lockForRead(); + } + + return request_pause || paused; } void RingBuffer::run(void) @@ -920,7 +1007,6 @@ long long totfree = 0; int ret = -1; int used = 0; - int loops = 0; struct timeval lastread, now; gettimeofday(&lastread, NULL); @@ -928,52 +1014,36 @@ int readtimeavg = 300; int readinterval; - pausereadthread = false; - + rwlock.lockForWrite(); + request_pause = false; readAheadBuffer = new char[kBufferSize + KB640]; - - rwlock.lockForWrite(); ResetReadAhead(0); + readaheadrunning = true; + readAheadRunningWait.wakeAll(); 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(); while (readaheadrunning) { - if (pausereadthread || writemode) - { - readaheadpaused = true; - pauseWait.wakeAll(); - usleep(5000); - totfree = ReadBufFree(); + if (PauseAndWait()) continue; - } - if (readaheadpaused) - { - totfree = ReadBufFree(); - readaheadpaused = false; - } + totfree = ReadBufFree(); - totfree = ReadBufFree(); - if (totfree < GetReadBlockSize()) + if (totfree < readblocksize) { + rwlock.unlock(); 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; + rwlock.lockForRead(); + continue; } - loops = 0; - rwlock.lockForRead(); - if (totfree > readblocksize && !commserror && !ateof && !setswitchtonext) + if (totfree > readblocksize && !commserror && + !ateof && !setswitchtonext) { // limit the read size totfree = readblocksize; @@ -1036,11 +1106,15 @@ internalreadpos += ret; } - readAheadLock.lock(); + rwlock.unlock(); + + rwlock.lockForWrite(); if (ret > 0 ) rbwpos = (rbwpos + ret) % kBufferSize; - readAheadLock.unlock(); + rwlock.unlock(); + rwlock.lockForRead(); + if (ret == 0 && !stopreads) { if (livetvchain) @@ -1053,10 +1127,19 @@ } } else + { + rwlock.unlock(); + rwlock.lockForWrite(); ateof = true; + rwlock.unlock(); + rwlock.lockForRead(); + } } } + rwlock.unlock(); + rwlock.lockForWrite(); + if (numfailures > 5) commserror = true; @@ -1087,163 +1170,112 @@ // .arg(fill_min)); } - readsAllowedWaitMutex.lock(); - if (readsallowed || stopreads) + if (readsallowed || stopreads || commserror) readsAllowedWait.wakeAll(); - readsAllowedWaitMutex.unlock(); - availWaitMutex.lock(); if (commserror || ateof || stopreads || setswitchtonext || (wanttoread <= used && wanttoread > 0)) { availWait.wakeAll(); } - availWaitMutex.unlock(); + bool do_yield = + ((used >= fill_threshold || ateof || setswitchtonext) && + !request_pause); + rwlock.unlock(); - if ((used >= fill_threshold || wantseek || ateof || setswitchtonext) && - !pausereadthread) - { - usleep(500); - } + if (do_yield) + usleep(5000); + + rwlock.lockForRead(); } + rwlock.unlock(); + delete [] readAheadBuffer; readAheadBuffer = NULL; + + rwlock.lockForWrite(); rbrpos = 0; rbwpos = 0; + rwlock.unlock(); } long long RingBuffer::SetAdjustFilesize(void) { + rwlock.lockForWrite(); readAdjust += internalreadpos; + 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); - } - + rwlock.lockForWrite(); + int ret = ReadPriv(buf, count, true); if (ret != count) { VERBOSE(VB_IMPORTANT, LOC_WARN + QString("Peek() requested %1 bytes, but only returning %2") .arg(count).arg(ret)); } + rwlock.unlock(); 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); + readsAllowedWait.wait(&rwlock, 500); + if (!readsallowed) + { + VERBOSE(VB_IMPORTANT, + LOC + "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)) + availWait.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 > 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) + @@ -1270,28 +1302,133 @@ 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; - availWaitMutex.unlock(); + wanttoread = 0; - avail = ReadBufAvail(); - if ((ateof || setswitchtonext) && avail < count) - count = avail; + return avail >= count; +} - if (commserror) - return 0; +int RingBuffer::ReadDirect(void *buf, int count, bool peek) +{ + long long old_pos = 0; + if (peek) + { + poslock.lockForRead(); + old_pos = readpos; + poslock.unlock(); } - if ((ateof || stopreads) && avail < count) - count = avail; + 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; + } + if (peek && ret > 0) + { +#ifdef USING_FRONTEND + if (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 (bdPriv) + { + // No idea if this will work. + bdPriv->Seek(0); + } + else +#endif // USING_FRONTEND + { + rwlock.unlock(); + long long new_pos = Seek(-ret, SEEK_CUR); + rwlock.lockForWrite(); + 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)); + } + } + } + + return ret; +} + +/** \brief When possible reads from the read-ahead buffer, + * otherwise reads directly from the device. + * + * WARNING: Must be called with rwlock in write lock state. + * + * \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) +{ + if (writemode) + { + VERBOSE(VB_IMPORTANT, LOC_ERR + + "Attempt to read from a write only file"); + errno = EBADF; + return -1; + } + + if (commserror) + { + VERBOSE(VB_IMPORTANT, LOC_ERR + + "Attempt to read after commserror set"); + errno = EIO; + return -1; + } + + if (request_pause || stopreads || !readaheadrunning) + { + return ReadDirect(buf, count, peek); + } + + if (!WaitForReadsAllowed()) + { + VERBOSE(VB_FILE, LOC + "!WaitForReadsAllowed()"); + wanttoread = 0; + stopreads = true; + return 0; + } + + if (!WaitForAvail(count)) + { + VERBOSE(VB_FILE, LOC + "!WaitForAvail()"); + ateof = true; + wanttoread = 0; + stopreads = true; + return 0; + } + + count = min(ReadBufAvail(), count); + + if (count <= 0) + return count; + if (rbrpos + count > (int) kBufferSize) { int firstsize = kBufferSize - rbrpos; @@ -1304,18 +1441,8 @@ memcpy(buf, readAheadBuffer + rbrpos, count); if (!peek) - { - readAheadLock.lock(); rbrpos = (rbrpos + count) % kBufferSize; - readAheadLock.unlock(); - } - if (readone) - { - Pause(); - WaitForPause(); - } - return count; } @@ -1329,53 +1456,20 @@ */ int RingBuffer::Read(void *buf, int count) { - int ret = -1; - if (writemode) + rwlock.lockForWrite(); + 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 { @@ -1404,25 +1498,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; } @@ -1431,8 +1536,10 @@ */ void RingBuffer::Sync(void) { + rwlock.lockForRead(); if (tfw) tfw->Sync(); + rwlock.unlock(); } /** \fn RingBuffer::Seek(long long, int) @@ -1440,26 +1547,38 @@ */ long long RingBuffer::Seek(long long pos, int whence) { + long long ret = -1; + + // lockForWrite takes priority over lockForRead, so this will + // take priority over the lockForRead in the read ahead thread. + rwlock.lockForWrite(); + if (writemode) + { + rwlock.unlock(); return WriterSeek(pos, whence); + } - wantseek = true; - rwlock.lockForWrite(); - wantseek = false; + poslock.lockForRead(); // optimize nop seeks if ((whence == SEEK_SET && pos == readpos) || (whence == SEEK_CUR && pos == 0)) { + ret = readpos; + rwlock.unlock(); - return readpos; + poslock.unlock(); + + return ret; } errno = 0; // clear errno, in case of remotefile error - long long ret = -1; if (remotefile) + { ret = remotefile->Seek(pos, whence, readpos); + } #ifdef USING_FRONTEND else if (dvdPriv) { @@ -1491,6 +1610,10 @@ } } + poslock.unlock(); + + poslock.lockForWrite(); + if (ret >= 0) { if (whence == SEEK_SET) @@ -1511,6 +1634,7 @@ VERBOSE(VB_IMPORTANT, LOC_ERR + cmd + " Failed" + ENO); } + poslock.unlock(); rwlock.unlock(); return ret; @@ -1523,12 +1647,18 @@ { long long ret = -1; + rwlock.lockForRead(); + poslock.lockForWrite(); + if (tfw) { ret = tfw->Seek(pos, whence); writepos = ret; } + poslock.unlock(); + rwlock.unlock(); + return ret; } @@ -1538,11 +1668,13 @@ */ void RingBuffer::WriterFlush(void) { + rwlock.lockForRead(); if (tfw) { tfw->Flush(); tfw->Sync(); } + rwlock.unlock(); } /** \fn RingBuffer::SetWriteBufferSize(int) @@ -1550,8 +1682,10 @@ */ void RingBuffer::SetWriteBufferSize(int newSize) { + rwlock.lockForRead(); if (tfw) tfw->SetWriteBufferSize(newSize); + rwlock.unlock(); } /** \fn RingBuffer::SetWriteBufferMinWriteSize(int) @@ -1559,8 +1693,10 @@ */ void RingBuffer::SetWriteBufferMinWriteSize(int newMinSize) { + rwlock.lockForRead(); if (tfw) tfw->SetWriteBufferMinWriteSize(newMinSize); + rwlock.unlock(); } /** \fn RingBuffer::GetReadPosition(void) const @@ -1568,14 +1704,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 @@ -1583,7 +1723,10 @@ */ long long RingBuffer::GetWritePosition(void) const { - return writepos; + poslock.lockForRead(); + long long ret = writepos; + poslock.unlock(); + return ret; } /** \fn RingBuffer::GetRealFileSize(void) const @@ -1592,11 +1735,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 @@ -1605,7 +1751,10 @@ */ bool RingBuffer::LiveMode(void) const { - return (livetvchain); + rwlock.lockForRead(); + bool ret = (livetvchain); + rwlock.unlock(); + return ret; } /** \fn RingBuffer::SetLiveMode(LiveTVChain*) @@ -1614,16 +1763,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 25788) +++ libs/libmythtv/RingBuffer.h (working copy) @@ -1,7 +1,7 @@ // -*- Mode: c++ -*- -#ifndef RINGBUFFER -#define RINGBUFFER +#ifndef _RINGBUFFER_H_ +#define _RINGBUFFER_H_ #include #include @@ -38,23 +38,29 @@ // 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*/); @@ -82,7 +88,12 @@ 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); @@ -93,8 +104,18 @@ // 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 @@ -108,92 +129,86 @@ 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 - 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 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 request_pause; // protected by rwlock + bool paused; // protected by rwlock + int rbrpos; // protected by rwlock + int rbwpos; // protected by rwlock + long long internalreadpos; // protected by rwlock (see note 1) + bool ateof; // protected by rwlock + bool readsallowed; // protected by rwlock + bool setswitchtonext; // 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: internalreadpos and numfailures are 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 these two variables. - 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 readAheadRunningWait;// protected by rwlock + QWaitCondition pauseWait; // protected by rwlock + QWaitCondition unpauseWait; // protected by rwlock + QWaitCondition availWait; // protected by rwlock + QWaitCondition readsAllowedWait; // protected by rwlock public: static QMutex subExtLock; @@ -206,4 +221,4 @@ static const uint kReadTestSize; }; -#endif +#endif // _RINGBUFFER_H_ Index: libs/libmythtv/mythplayer.cpp =================================================================== --- libs/libmythtv/mythplayer.cpp (revision 25787) +++ libs/libmythtv/mythplayer.cpp (working copy) @@ -3190,29 +3190,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.