diff --git a/mythtv/libs/libmythbase/threadedfilewriter.cpp b/mythtv/libs/libmythbase/threadedfilewriter.cpp index aa2fd9d..deec603 100644 --- a/mythtv/libs/libmythbase/threadedfilewriter.cpp +++ b/mythtv/libs/libmythbase/threadedfilewriter.cpp @@ -32,14 +32,6 @@ void TFWWriteThread::run(void) RunEpilog(); } -/// \brief Runs ThreadedFileWriter::SyncLoop(void) -void TFWSyncThread::run(void) -{ - RunProlog(); - m_parent->SyncLoop(); - RunEpilog(); -} - const uint ThreadedFileWriter::kMaxBufferSize = 8 * 1024 * 1024; const uint ThreadedFileWriter::kMinWriteSize = 64 * 1024; const uint ThreadedFileWriter::kMaxBlockSize = 1 * 1024 * 1024; @@ -50,7 +42,7 @@ const uint ThreadedFileWriter::kMaxBlockSize = 1 * 1024 * 1024; * This class allows us manage the buffering when writing to * disk. We write to the kernel image of the disk using one * thread, and sync the kernel's image of the disk to hardware - * using another thread. The goal here so to block as little as + * at most every seconds. The goal here so to block as little as * possible when the classes using this class want to add data * to the stream. */ @@ -68,7 +60,7 @@ ThreadedFileWriter::ThreadedFileWriter(const QString &fname, ignore_writes(false), tfw_min_write_size(kMinWriteSize), totalBufferUse(0), // threads - writeThread(NULL), syncThread(NULL), + writeThread(NULL), m_warned(false), m_blocking(false) { filename.detach(); @@ -88,6 +80,7 @@ bool ThreadedFileWriter::ReOpen(QString newFilename) if (fd >= 0) { + Sync(); close(fd); fd = -1; } @@ -135,12 +128,6 @@ bool ThreadedFileWriter::Open(void) writeThread->start(); } - if (!syncThread) - { - syncThread = new TFWSyncThread(this); - syncThread->start(); - } - return true; } } @@ -155,7 +142,6 @@ ThreadedFileWriter::~ThreadedFileWriter() { /* tell child threads to exit */ QMutexLocker locker(&buflock); in_dtor = true; - bufferSyncWait.wakeAll(); bufferHasData.wakeAll(); } @@ -178,15 +164,9 @@ ThreadedFileWriter::~ThreadedFileWriter() emptyBuffers.pop_front(); } - if (syncThread) - { - syncThread->wait(); - delete syncThread; - syncThread = NULL; - } - if (fd >= 0) { + Sync(); close(fd); fd = -1; } @@ -365,6 +345,8 @@ void ThreadedFileWriter::Flush(void) */ void ThreadedFileWriter::Sync(void) { + LOG(VB_FILE, LOG_INFO, LOC + "Sync() called"); + if (fd >= 0) { #if defined(_POSIX_SYNCHRONIZED_IO) && _POSIX_SYNCHRONIZED_IO > 0 @@ -390,23 +372,6 @@ void ThreadedFileWriter::SetWriteBufferMinWriteSize(uint newMinSize) bufferHasData.wakeAll(); } -/** \fn ThreadedFileWriter::SyncLoop(void) - * \brief The thread run method that calls Sync(void). - */ -void ThreadedFileWriter::SyncLoop(void) -{ - QMutexLocker locker(&buflock); - while (!in_dtor) - { - locker.unlock(); - - Sync(); - - locker.relock(); - bufferSyncWait.wait(&buflock, 1000); - } -} - /** \fn ThreadedFileWriter::DiskLoop(void) * \brief The thread run method that actually calls writes to disk. */ @@ -422,8 +387,9 @@ void ThreadedFileWriter::DiskLoop(void) // Even if the bytes buffered is less than the minimum write // size we do want to write to the OS buffers periodically. // This timer makes sure we do. - MythTimer minWriteTimer; + MythTimer minWriteTimer, lastSyncTimer; minWriteTimer.start(); + lastSyncTimer.start(); while (!in_dtor) { @@ -482,7 +448,7 @@ void ThreadedFileWriter::DiskLoop(void) uint tot = 0; uint errcnt = 0; - LOG(VB_FILE, LOG_DEBUG, LOC + QString("write(%1) cnt %2 total %3") + LOG(VB_FILE, LOG_INFO, LOC + QString("write(%1) cnt %2 total %3") .arg(sz).arg(writeBuffers.size()) .arg(totalBufferUse)); @@ -526,6 +492,14 @@ void ThreadedFileWriter::DiskLoop(void) bufferHasData.wait(locker.mutex(), 50); } + if (lastSyncTimer.elapsed() >= 1000) + { + locker.unlock(); + Sync(); + locker.relock(); + lastSyncTimer.restart(); + } + ////////////////////////////////////////// buf->lastUsed = MythDate::current(); @@ -604,4 +578,4 @@ bool ThreadedFileWriter::SetBlocking(bool block) bool old = m_blocking; m_blocking = block; return old; -} \ No newline at end of file +} diff --git a/mythtv/libs/libmythbase/threadedfilewriter.h b/mythtv/libs/libmythbase/threadedfilewriter.h index 1142d27..e06fc29 100644 --- a/mythtv/libs/libmythbase/threadedfilewriter.h +++ b/mythtv/libs/libmythbase/threadedfilewriter.h @@ -28,20 +28,10 @@ class TFWWriteThread : public MThread ThreadedFileWriter *m_parent; }; -class TFWSyncThread : public MThread -{ - public: - TFWSyncThread(ThreadedFileWriter *p) : MThread("TFWSync"), m_parent(p) {} - virtual ~TFWSyncThread() { wait(); m_parent = NULL; } - virtual void run(void); - private: - ThreadedFileWriter *m_parent; -}; - class MBASE_PUBLIC ThreadedFileWriter { friend class TFWWriteThread; - friend class TFWSyncThread; + public: ThreadedFileWriter(const QString &fname, int flags, mode_t mode); ~ThreadedFileWriter(); @@ -60,7 +50,6 @@ class MBASE_PUBLIC ThreadedFileWriter protected: void DiskLoop(void); - void SyncLoop(void); void TrimEmptyBuffers(void); private: @@ -90,12 +79,10 @@ class MBASE_PUBLIC ThreadedFileWriter // threads TFWWriteThread *writeThread; - TFWSyncThread *syncThread; // wait conditions QWaitCondition bufferEmpty; QWaitCondition bufferHasData; - QWaitCondition bufferSyncWait; QWaitCondition bufferWasFreed; // constants diff --git a/mythtv/libs/libmythtv/fileringbuffer.cpp b/mythtv/libs/libmythtv/fileringbuffer.cpp index 2548703..3c7cc92 100644 --- a/mythtv/libs/libmythtv/fileringbuffer.cpp +++ b/mythtv/libs/libmythtv/fileringbuffer.cpp @@ -454,7 +454,12 @@ int FileRingBuffer::safe_read(int fd, void *data, uint sz) while (tot < sz) { + LOG(VB_FILE, LOG_DEBUG, LOC + + QString("read(%1) -- begin").arg(sz)); ret = read(fd2, (char *)data + tot, sz - tot); + LOG(VB_FILE, LOG_DEBUG, LOC + + QString("read(%1) -> %2 end").arg(sz).arg(ret)); + if (ret < 0) { if (errno == EAGAIN) diff --git a/mythtv/libs/libmythtv/ringbuffer.cpp b/mythtv/libs/libmythtv/ringbuffer.cpp index cbca8a6..5dbd48f 100644 --- a/mythtv/libs/libmythtv/ringbuffer.cpp +++ b/mythtv/libs/libmythtv/ringbuffer.cpp @@ -348,6 +348,8 @@ void RingBuffer::CalcReadAheadThresh(void) uint estbitrate = 0; readsallowed = false; + LOG(VB_FILE, LOG_INFO, LOC + + "CalcReadAheadThresh: readsallowed set to false."); readblocksize = max(readblocksize, CHUNK); // loop without sleeping if the buffered data is less than this @@ -484,6 +486,8 @@ void RingBuffer::ResetReadAhead(long long newinternal) internalreadpos = newinternal; ateof = false; readsallowed = false; + LOG(VB_FILE, LOG_INFO, LOC + + "ResetReadAhead: readsallowed set to false."); setswitchtonext = false; generalWait.wakeAll(); @@ -762,13 +766,14 @@ void RingBuffer::run(void) LOG(VB_FILE, LOG_INFO, LOC + QString("Initial readblocksize %1K & fill_min %2K") - .arg(readblocksize/1024).arg(fill_min/1024)); + .arg(readblocksize/1024).arg(fill_min/1024)); while (readaheadrunning) { if (PauseAndWait()) { ignore_for_read_timing = true; + LOG(VB_FILE, LOG_DEBUG, LOC + "run: PauseAndWait Not reading continuing"); continue; } @@ -783,6 +788,7 @@ void RingBuffer::run(void) ignore_for_read_timing |= (ignorereadpos >= 0) || commserror || stopreads; generalWait.wait(&rwlock, (stopreads) ? 50 : 1000); + LOG(VB_FILE, LOG_DEBUG, LOC + "run: Not reading continuing"); continue; } @@ -843,11 +849,13 @@ void RingBuffer::run(void) ignore_for_read_timing = (totfree < readblocksize) ? true : false; lastread = now; + LOG(VB_FILE, LOG_INFO, LOC + "run(): rbwlock.lockForRead() start"); rbwlock.lockForRead(); + LOG(VB_FILE, LOG_INFO, LOC + "run(): rbwlock.lockForRead() done"); if (rbwpos + totfree > bufferSize) { totfree = bufferSize - rbwpos; - LOG(VB_FILE, LOG_DEBUG, LOC + + LOG(VB_FILE, LOG_INFO, LOC + "Shrinking read, near end of buffer"); } @@ -858,7 +866,7 @@ void RingBuffer::run(void) "Reading enough data to start playback"); } - LOG(VB_FILE, LOG_DEBUG, LOC + + LOG(VB_FILE, LOG_INFO, LOC + QString("safe_read(...@%1, %2) -- begin") .arg(rbwpos).arg(totfree)); @@ -896,10 +904,18 @@ void RingBuffer::run(void) LOC + QString("rbwpos += %1K requested %2K in read") .arg(read_return/1024,3).arg(totfree/1024,3)); } + else + { + LOG(VB_FILE, LOG_INFO, LOC + "rbwpos != rbwposcopy"); + } rbwlock.unlock(); poslock.unlock(); } } + else + { + LOG(VB_FILE, LOG_INFO, LOC + QString("We are not reading anything (totfree: %1 commserror:%2 ateof:%3 setswitchtonext:%4").arg(totfree).arg(commserror).arg(ateof).arg(setswitchtonext)); + } int used = bufferSize - ReadBufFree(); @@ -945,7 +961,7 @@ void RingBuffer::run(void) used = bufferSize - ReadBufFree(); } - LOG(VB_FILE, LOG_DEBUG, LOC + "@ end of read ahead loop"); + LOG(VB_FILE, LOG_INFO, LOC + "@ end of read ahead loop"); if (!readsallowed || commserror || ateof || setswitchtonext || (wanttoread <= used && wanttoread > 0)) @@ -1084,18 +1100,18 @@ bool RingBuffer::WaitForAvail(int count) int elapsed = t.elapsed(); if (elapsed > 500 && low_buffers && avail >= fill_min) count = avail; - else if (((elapsed > 250) && (elapsed < 500)) || - ((elapsed > 500) && (elapsed < 750)) || + else if (((elapsed > 500) && (elapsed < 750)) || ((elapsed > 1000) && (elapsed < 1250)) || ((elapsed > 2000) && (elapsed < 2250)) || ((elapsed > 4000) && (elapsed < 4250)) || ((elapsed > 8000) && (elapsed < 8250)) || ((elapsed > 9000))) { + LOG(VB_FILE, LOG_INFO, LOC + QString("used = %1").arg(bufferSize - ReadBufFree())); LOG(VB_GENERAL, LOG_INFO, LOC + "Waited " + 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)); + QString(" %2 < %3") .arg(avail).arg(count) + QString(" thread:%1").arg(QThread::currentThreadId())); } if (elapsed > 16000)