Ticket #6330: locklesstfw-v5.3.diff

mythtv/libs/libmythtv/libmythtv.pro

@@ -156 +156 @@
 HEADERS += transporteditor.h        listingsources.h
 HEADERS += myth_imgconvert.h
 HEADERS += channelgroup.h           channelgroupsettings.h
+HEADERS += ThreadedFileWriterLockless.h
 
 # Remove when everything is switched to MythUI
 HEADERS += proglist_qt.h
@@ -181 +182 @@
 SOURCES += transporteditor.cpp
 SOURCES += channelgroup.cpp         channelgroupsettings.cpp
 SOURCES += myth_imgconvert.cpp
+SOURCES += ThreadedFileWriterLockless.cpp
 
 # Remove when everything is switched to MythUI
 SOURCES += proglist_qt.cpp

mythtv/libs/libmythtv/RingBuffer.cpp

@@ -24 +24 @@
 #include "RingBuffer.h"
 #include "remotefile.h"
 #include "remoteencoder.h"
-#include "ThreadedFileWriter.h"
+#include "ThreadedFileWriterLockless.h"
 #include "livetvchain.h"
 #include "DVDRingBuffer.h"
 #include "util.h"
@@ -169 +169 @@
         }
         else
         {
-            tfw = new ThreadedFileWriter(
+            tfw = new ThreadedFileWriterLockless(
                 filename, O_WRONLY|O_TRUNC|O_CREAT|O_LARGEFILE, 0644);
 
             if (!tfw->Open())

mythtv/libs/libmythtv/RingBuffer.h

@@ -16 +16 @@
 
 class RemoteFile;
 class RemoteEncoder;
-class ThreadedFileWriter;
+class ThreadedFileWriterLockless;
 class DVDRingBufferPriv;
 class LiveTVChain;
 
@@ -120 +120 @@
     QString filename;
     QString subtitlefilename;
 
-    ThreadedFileWriter *tfw;
+    ThreadedFileWriterLockless *tfw;
     int fd2;
 
     bool writemode;

mythtv/libs/libmythtv/ThreadedFileWriterLockless.cpp

@@ +1 @@
+// ANSI C headers
+#include <cerrno>
+#include <cstring>
+
+// Unix C headers
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <fcntl.h>
+#include <unistd.h>
+#include <signal.h>
+
+// Qt headers
+#include <QString>
+
+// MythTV headers
+#include "ThreadedFileWriterLockless.h"
+#include "mythcontext.h"
+#include "compat.h"
+#include "mythverbose.h"
+#include "mythconfig.h"
+
+#define LOC QString("TFWL: ")
+#define LOC_ERR QString("TFWL, Error: ")
+
+const int ThreadedFileWriterLockless::DEFAULT_BUFFER_SIZE = 2*1024*1024;
+const int ThreadedFileWriterLockless::MAX_WRITE_SIZE      = DEFAULT_BUFFER_SIZE / 4;
+const int ThreadedFileWriterLockless::MIN_WRITE_SIZE      = DEFAULT_BUFFER_SIZE / 32;
+const int ThreadedFileWriterLockless::BUFFER_CUSHION      = 512;
+const int ThreadedFileWriterLockless::MAX_WRITE_ERRORS    = 10;
+
+/** \class ThreadedFileWriterLockless
+ *  \brief A (almost) lock-free threaded disk writer that buffers data written
+ *         to it and writes it to the disk in a seperate thread.
+ *
+ *  This class implements a threaded ring buffer. All data written via ::Write()
+ *  is inserted into the ring buffer while a seperate thread reads the data from
+ *  the buffer and writes it to disk.
+ *
+ *  The implementation is intentionally not thread-safe but reentrant.
+ */
+
+/** \fn ThreadedFileWriterLockless::safe_write(int fd, const void *data, uint count, bool &ok)
+ *  \brief Writes all data to disk and retries in case of errors.
+ *
+ *  The standard POSIX write() can return early with no or just a portion of
+ *  the desired data written to disk. This function tries to make sure that
+ *  all data gets written to disk by...
+ *
+ *    1) endlessly retrying in case of EAGAIN or EINTR errors
+ *    2) retrying up to MAX_WRITE_ERRORS times in case of other errors
+ *    3) writing out the rest amount of data until no data is left
+ *
+ *  \param fd    File descriptor
+ *  \param data  Pointer to data to write
+ *  \param count Size of data to write in bytes
+ *
+ *  \return Number of written bytes
+ */
+uint ThreadedFileWriterLockless::safe_write(int fd, const void *data,
+                                            uint count, bool &ok)
+{
+    uint bytesWritten = 0;
+    int errors        = 0;
+
+    while (bytesWritten < count && errors < MAX_WRITE_ERRORS)
+    {
+        int ret = write(fd, (char*)data + bytesWritten, count - bytesWritten);
+
+        if (ret >= 0)
+            bytesWritten += ret;
+        else if (errno != EAGAIN || errno != EINTR)
+            ++errors;
+    }
+
+    ok = (errors < MAX_WRITE_ERRORS);
+
+    return bytesWritten;
+}
+
+/** \fn ThreadedFileWriterLockless::BootWriter(void *tfw)
+ *  \brief Helper function which simply runs the DiskWriterLoop().
+ *
+ *  \param tfw Pointer to ThreadedFileWriterLockless instance
+ */
+void *ThreadedFileWriterLockless::BootWriter(void *tfw)
+{
+#ifndef USING_MINGW
+    signal(SIGXFSZ, SIG_IGN);
+#endif
+    ((ThreadedFileWriterLockless *)tfw)->DiskWriterLoop();
+    return NULL;
+}
+
+/** \fn ThreadedFileWriterLockless::BootFileAdviser(void *tfw)
+ *  \brief Helper function which simply runs the FileAdviserLoop().
+ *
+ *  \param tfw Pointer to ThreadedFileWriterLockless instance
+ */
+void *ThreadedFileWriterLockless::BootFileAdviser(void *tfw)
+{
+    ((ThreadedFileWriterLockless *)tfw)->FileAdviserLoop();
+    return NULL;
+}
+
+/** \fn ThreadedFileWriterLockless::ThreadedFileWriterLockless(const QString &fname, int flags, mode_t mode)
+ *  \brief Constructor
+ *
+ *  The constructor sets the filename, the access/creation/status flags and
+ *  the file permissions. It does not create the file nor does it start the
+ *  writer thread.
+ *
+ *  \param fname Filename
+ *  \param flags Access/Creation/Status flags (see POSIX open())
+ *  \param mode  File permissions (see POSIX open())
+ */
+ThreadedFileWriterLockless::ThreadedFileWriterLockless(const QString &fname,
+                                                       int flags, mode_t mode) :
+    // file handling related
+    fileName(fname), fileFlags(flags), fileMode(mode), fileFD(-1),
+    fileMinWriteSize(MIN_WRITE_SIZE), fileBytesWritten(0),
+    // state flags
+    isFlushing(0), isWriterRunning(0), isFileAdviserRunning(0), isInDestructor(0),
+    isIgnoringWrites(0),
+    // buffer and related
+    buffer(NULL), bufferSize(0), bufferReadPos(0), bufferWritePos(0)
+{
+    fileName.detach();
+}
+
+/** \fn ThreadedFileWriterLockless::Open(void)
+ *  \brief Opens the file we will be writing to and starts the DiskWriter thread.
+ *
+ *  \return TRUE if file was opened successfully and DiskWriter thread started
+ */
+bool ThreadedFileWriterLockless::Open(void)
+{
+    if (isWriterRunning)
+    {
+        VERBOSE(VB_IMPORTANT, LOC_ERR +
+                QString("Opening file '%1' failed. Writer already running")
+                .arg(fileName));
+
+        return false;
+    }
+
+    // those do _not_ need to be atomic since the writer is not running
+    isFlushing       = 0;
+    isInDestructor   = 0;
+    isIgnoringWrites = 0;
+
+    bool result = false;
+
+    if (fileName == "-")
+        fileFD = fileno(stdout);
+    else
+        fileFD = open(fileName.toAscii().constData(), fileFlags, fileMode);
+
+    if (fileFD < 0)
+    {
+        VERBOSE(VB_IMPORTANT, LOC_ERR + 
+                QString("Opening file '%1'.").arg(fileName) + ENO);
+
+        result = false;
+    }
+    else
+    {
+        SetupBuffer(DEFAULT_BUFFER_SIZE);
+        fileMinWriteSize = MIN_WRITE_SIZE;
+
+        pthread_create(&writer, NULL, BootWriter, this);
+        pthread_create(&adviser, NULL, BootFileAdviser, this);
+/*
+        if(gContext->GetNumSetting("RealtimePriority", 1))
+        {
+            // thread priority in respect to SCHED_RR: 1
+            struct sched_param sp = {2};
+
+            if(!pthread_setschedparam(writer, SCHED_RR, &sp));
+            {
+                VERBOSE(VB_GENERAL, LOC +
+                        "Using realtime priority.");
+            }
+            else
+            {
+                VERBOSE(VB_IMPORTANT, LOC_ERR +
+                        "Realtime priority requires sufficient user priviledges.");
+            }
+        }
+*/
+        result = true;
+    }
+
+    return result;
+}
+
+/** \fn ThreadedFileWriterLockless::~ThreadedFileWriterLockless()
+ *  \brief Destructor which commits all writes, ends the DiskWriter thread
+ *         and closes the file.
+ */
+ThreadedFileWriterLockless::~ThreadedFileWriterLockless()
+{
+    isInDestructor.fetchAndStoreOrdered(1);
+
+    if (fileFD >= 0)
+    {
+        // wake everyone
+        bufferHasData.wakeAll();
+        bufferEmpty.wakeAll();
+
+        // if there is still data, flush it to disk
+        Flush();
+
+        pthread_join(writer, NULL);
+        pthread_join(adviser, NULL);
+
+        close(fileFD);
+    }
+
+    if (buffer)
+        delete [] (buffer - BUFFER_CUSHION);
+}
+
+/** \fn ThreadedFileWriterLockless::Write(const void *data, uint count)
+ *  \brief Writes data to the internal buffer.
+ *
+ *  This function writes data to the internal buffer.
+ *
+ *  \param data  Pointer to data
+ *  \param count Size of data in bytes
+ *
+ *  \return Returns the number of bytes written
+ */
+uint ThreadedFileWriterLockless::Write(const void *data, uint count)
+{
+    if (count == 0 || isIgnoringWrites)
+        return 0;
+
+    uint  totalWritten = 0;
+    char* dataPtr      = (char*)data;
+
+    while (totalWritten < count)
+    {
+        // using local copies for efficiency reasons: a) doing arithmetic on
+        // instances of QAtomicInt is expensive and b) the architecture MythTV
+        // is currently running on could probably not have native/wait-free
+        // support for the atomic operation fetchAndStore
+        int localWritePos = bufferWritePos;
+
+        uint writeFwdFree = WriteFwdFree(bufferReadPos, localWritePos);
+        while (writeFwdFree == 0)
+        {
+            bufferWriterLock.lock();
+            bufferWroteData.wait(&bufferWriterLock, 100);
+            bufferWriterLock.unlock();
+
+            writeFwdFree = WriteFwdFree(bufferReadPos, localWritePos);
+        }
+
+        int chunkSize = (count-totalWritten < writeFwdFree) ? count-totalWritten
+                                                            : writeFwdFree;
+
+        if (localWritePos + chunkSize > bufferSize)
+        {
+            uint partSize = bufferSize - localWritePos;
+            memcpy(buffer + localWritePos, dataPtr, partSize);
+
+            totalWritten += partSize;
+            chunkSize    -= partSize;
+            dataPtr      += partSize;
+            localWritePos = 0;
+        }
+        memcpy(buffer + localWritePos, dataPtr, chunkSize);
+
+        totalWritten  += chunkSize;
+        localWritePos  = (localWritePos + chunkSize) % bufferSize;
+        dataPtr       += chunkSize;
+
+        bufferWritePos.fetchAndStoreOrdered(localWritePos);
+
+        bufferHasData.wakeAll();
+    }
+
+    return totalWritten;
+}
+
+/** \fn ThreadedFileWriterLockless::Seek(long long pos, int whence)
+ *  \brief Seek to a position within the disk file
+ *
+ *  \param pos    New file offset (with regards to whence)
+ *  \param whence Specifies how to interpret offset (see lseek())
+ *
+ *  \return Resulting offset
+ */
+long long ThreadedFileWriterLockless::Seek(long long pos, int whence)
+{
+    Flush();
+
+    return lseek(fileFD, pos, whence);
+}
+
+/** \fn ThreadedFileWriterLockless::Flush(void)
+ *  \brief Instructs the DiskWriter thread to flush all remaining data in the
+ *         ring buffer to the disk. If the DiskWriter thread is currently ignoring
+ *         writes to the disk because of e.g. previous errors, the buffer just
+ *         gets discarded.
+ *
+ *  \warning This call is expensive!
+ */
+void ThreadedFileWriterLockless::Flush(void)
+{
+    isFlushing.fetchAndStoreOrdered(1);
+
+    // wake up DiskWriter thread
+    bufferHasData.wakeAll();
+
+    diskWriterLock.lock();
+    // loop as long as there is still data in the buffer or an ongoing write()
+    // is registered which could potentially fill up the buffer
+    while (BufUsed() > 0)
+    {
+        if (!bufferEmpty.wait(&diskWriterLock, 2000))
+           VERBOSE(VB_IMPORTANT, LOC + "Taking a long time to flush...");
+    }
+    diskWriterLock.unlock();
+
+    isFlushing.fetchAndStoreOrdered(0);
+}
+
+/** \fn ThreadedFileWriterLockless::Sync(void)
+ *  \brief Instructs the OS to sync all file data associated data to the disk.
+ *
+ *  \warning This is a very expensive call which blocks until the data is on
+ *           the disk. Some filesystems (e.g. ext3) sync the entire filesystem
+ *           data to disk, not only the file associated ones. This can cause
+ *           high latency and delays if used extensively. Use with care.
+ */
+void ThreadedFileWriterLockless::Sync(void)
+{
+    if (fileFD >= 0)
+    {
+#if defined(_POSIX_SYNCHRONIZED_IO) && _POSIX_SYNCHRONIZED_IO > 0
+        fdatasync(fileFD);
+#else
+        fsync(fileFD);
+#endif
+    }
+}
+
+/** \fn ThreadedFileWriterLockless::SetWriteBufferSize(uint newSize)
+ *  \brief Sets the total size of the internal ringbuffer.
+ *
+ *  \warning This is an expensive call because all data currently in the ring
+ *           buffer is flushed to the disk.
+ *
+ *  \param newSize New ringbuffer size in bytes
+ */
+void ThreadedFileWriterLockless::SetWriteBufferSize(uint newSize)
+{
+    if (newSize <= 0)
+        return;
+
+    Flush();
+    SetupBuffer(newSize);
+}
+
+/** \fn ThreadedFileWriterLockless::SetWriteBufferMinWriteSize(uint newMinSize)
+ *  \brief Sets the minumum number of bytes to write to disk in a single write.
+ *
+ *  \param newMinSize New minimum size in bytes for a single write
+ */
+void ThreadedFileWriterLockless::SetWriteBufferMinWriteSize(uint newMinSize)
+{
+    fileMinWriteSize.fetchAndStoreOrdered(newMinSize);
+}
+
+/** \fn ThreadedFileWriterLockless::DiskLoop(void)
+ *  \brief DiskWriter thread which reads data from the ring buffer and writes it
+ *         out to the disk.
+ *
+ *  The DiskWriterLoop() tries to free space as soon as possible but is limited
+ *  by the minimum and maximum write size for a single write. Setting those
+ *  higher can increase throughput but also latency.
+ */
+void ThreadedFileWriterLockless::DiskWriterLoop(void)
+{
+    isWriterRunning.fetchAndStoreOrdered(1);
+
+    int readFwdFree = 1;
+    while (!isInDestructor || ReadFwdFree(bufferReadPos, bufferWritePos) > 0)
+    {
+        readFwdFree = ReadFwdFree(bufferReadPos, bufferWritePos);
+
+        if (!readFwdFree)
+            bufferEmpty.wakeAll();
+
+        if (!readFwdFree || (!isInDestructor && !isFlushing &&
+                             readFwdFree < fileMinWriteSize
+                             && fileBytesWritten >= fileMinWriteSize))
+        {
+            diskWriterLock.lock();
+            bufferHasData.wait(&diskWriterLock, 250);
+            diskWriterLock.unlock();
+            continue;
+        }
+
+        int  localReadPos = bufferReadPos;
+        bool wasWriteOk   = true;
+        uint chunkSize    = 0;
+
+        while (wasWriteOk && (readFwdFree >= fileMinWriteSize
+                              || (isFlushing && readFwdFree > 0)))
+        {
+            if(localReadPos + readFwdFree > bufferSize)
+            {
+                int remainder = bufferSize - localReadPos;
+                chunkSize = (remainder <= MAX_WRITE_SIZE || isFlushing)
+                             ? remainder : MAX_WRITE_SIZE;
+            }
+            else
+            {
+                chunkSize = (readFwdFree <= MAX_WRITE_SIZE || isFlushing)
+                             ? readFwdFree : MAX_WRITE_SIZE;
+            }
+
+            if (!isIgnoringWrites)
+                chunkSize = safe_write(fileFD, buffer + localReadPos, chunkSize, wasWriteOk);
+
+            fileBytesWritten += chunkSize;
+            localReadPos      = (localReadPos + chunkSize) % bufferSize;
+
+            bufferReadPos.fetchAndStoreOrdered(localReadPos);
+            bufferWroteData.wakeAll();
+
+            readFwdFree = ReadFwdFree(localReadPos, bufferWritePos);
+        }
+
+        if (!isIgnoringWrites && !wasWriteOk && (EFBIG == errno))
+        {
+            QString msg =
+                "Maximum file size exceeded by '%1'"
+                "\n\t\t\t"
+                "You must either change the process ulimits, configure"
+                "\n\t\t\t"
+                "your operating system with \"Large File\" support, or use"
+                "\n\t\t\t"
+                "a filesystem which supports 64-bit or 128-bit files."
+                "\n\t\t\t"
+                "HINT: FAT32 is a 32-bit filesystem.";
+
+            VERBOSE(VB_IMPORTANT, msg.arg(fileName));
+            isIgnoringWrites.fetchAndStoreOrdered(1);
+        }
+    }
+
+    isWriterRunning.fetchAndStoreOrdered(0);
+}
+
+/** \fn ThreadedFileWriterLockless::FileAdviserLoop(void)
+ *  \brief Periodically frees related system file cache.
+ *
+ *  This function periodically calls posix_fadvise() to free related system
+ *  file cache for the active file descriptor. Since the system call can be
+ *  expensive and cause delay spikes, it was put in a seperate thread.
+ */
+void ThreadedFileWriterLockless::FileAdviserLoop(void)
+{
+    isFileAdviserRunning.fetchAndStoreOrdered(1);
+
+    while (!isInDestructor)
+    {
+        if (fileFD >= 0)
+            posix_fadvise(fileFD, 0, 0, POSIX_FADV_DONTNEED);
+
+        sleep(2);
+    }
+
+    isFileAdviserRunning.fetchAndStoreOrdered(0);
+}
+
+/** \fn ThreadedFileWriterLockless::BufUsed(void) const
+ *  \brief Gets the buffer usage in bytes which are queued for disk write
+ *
+ *  \return Buffer usage in bytes
+ */
+uint ThreadedFileWriterLockless::BufUsed(void) const
+{
+    return (bufferSize-1) - BufFree();
+}
+
+/** \fn ThreadedFileWriterLockless::BufFree(void) const
+ *  \brief Gets the free buffer space in bytes
+ *
+ *  \return Free buffer space in bytes
+ */
+uint ThreadedFileWriterLockless::BufFree(void) const
+{
+    return WriteFwdFree(bufferReadPos, bufferWritePos);
+}
+
+/** \fn ThreadedFileWriterLockless::WriteFwdFree(void) const
+ *  \brief Gets the number of bytes available for write to the ring buffer
+ *         from the write pointer with regards to the passed pointers.
+ *
+ *  \return Free buffer space in bytes for write.
+ */
+inline uint ThreadedFileWriterLockless::WriteFwdFree(int readPos, int writePos) const
+{
+    return (readPos + bufferSize - 1 - writePos) % bufferSize;
+}
+
+/** \fn ThreadedFileWriterLockless::ReadFwdFree(void) const
+ *  \brief Gets the number of bytes available for read from the ring buffer
+ *         from the read pointer with regards to the passed pointers.
+ *
+ *  \return Bytes available for read from the buffer
+ */
+inline uint ThreadedFileWriterLockless::ReadFwdFree(int readPos, int writePos) const
+{
+    return (writePos + bufferSize - readPos) % bufferSize;
+}
+
+/** \fn ThreadedFileWriterLockless::SetupBuffer(void)
+ *  \brief Helper function which sets up the buffer and all necessary pointers.
+ *
+ *  \param reqBufferSize Requested buffer size
+ */
+void ThreadedFileWriterLockless::SetupBuffer(uint reqBufferSize)
+{
+    if (buffer)
+        delete [] (buffer - BUFFER_CUSHION);
+
+    buffer = new char[1 + reqBufferSize + 2 * BUFFER_CUSHION];
+    bzero(buffer, 1 + reqBufferSize + 2 * BUFFER_CUSHION);
+
+    buffer    += BUFFER_CUSHION;
+    bufferSize = reqBufferSize;
+
+    bufferWritePos.fetchAndStoreOrdered(0);
+    bufferReadPos.fetchAndStoreOrdered(0);
+}

mythtv/libs/libmythtv/ThreadedFileWriterLockless.h

@@ +1 @@
+#ifndef TFW_LOCKLESS_H_
+#define TFW_LOCKLESS_H_
+
+#include <pthread.h>
+
+#include <QAtomicInt>
+#include <QMutex>
+#include <QWaitCondition>
+#include <QString>
+
+class ThreadedFileWriterLockless
+{
+  public:
+    ThreadedFileWriterLockless(const QString &fname, int flags, mode_t mode);
+    ~ThreadedFileWriterLockless();
+
+    bool Open(void);
+    long long Seek(long long pos, int whence);
+    uint Write(const void *data, uint count);
+
+    void SetWriteBufferSize(uint newSize = DEFAULT_BUFFER_SIZE);
+    void SetWriteBufferMinWriteSize(uint newMinSize = MIN_WRITE_SIZE);
+
+    uint BufUsed(void) const;
+    uint BufFree(void) const;
+
+    void Sync(void);
+    void Flush(void);
+
+    static uint safe_write(int fd, const void *data, uint count, bool &ok);
+
+  protected:
+    static void *BootWriter(void *);
+    static void *BootFileAdviser(void *);
+    void DiskWriterLoop(void);
+    void FileAdviserLoop(void);
+
+    uint WriteFwdFree(int readPos, int writePos) const;
+    uint ReadFwdFree(int readPos, int writePos) const;
+
+    void SetupBuffer(uint reqBufferSize);
+
+  private:
+    // file handling related
+    QString    fileName;
+    int        fileFlags;
+    mode_t     fileMode;
+    int        fileFD;
+    QAtomicInt fileMinWriteSize;
+    long long  fileBytesWritten;
+
+    // state flags
+    QAtomicInt isFlushing;
+    QAtomicInt isWriterRunning;
+    QAtomicInt isFileAdviserRunning;
+    QAtomicInt isInDestructor;
+    QAtomicInt isIgnoringWrites;
+
+    // locks and wait conditions
+    mutable QMutex bufferWriterLock;
+    mutable QMutex diskWriterLock;
+
+    QWaitCondition bufferEmpty;
+    QWaitCondition bufferHasData;
+    QWaitCondition bufferWroteData;
+
+    // buffer and related
+    char       *buffer;
+    int         bufferSize;
+    QAtomicInt  bufferReadPos;
+    QAtomicInt  bufferWritePos;
+
+    // threads
+    pthread_t writer;
+    pthread_t adviser;
+
+  private:
+    // constants
+    /// default buffer size
+    static const int DEFAULT_BUFFER_SIZE;
+    /// maximum chunk size to write to disk in one go
+    static const int MAX_WRITE_SIZE;
+    /// minimum chunk size to write to disk in one go (except when flushing)
+    static const int MIN_WRITE_SIZE;
+    /// number of extra bytes to reserve before and after the actual buffer
+    static const int BUFFER_CUSHION;
+    /// maximum write errors for a single call to safe_write before giving up
+    static const int MAX_WRITE_ERRORS;
+};
+#endif