Ticket #6330: locklesstfw-v1.diff

mythtv/libs/libmythtv/ThreadedFileWriterLockless.h

@@ +1 @@
+#ifndef TFW_LOCKLESS_H_
+#define TFW_LOCKLESS_H_
+
+#include <pthread.h>
+
+#include <QAtomicInt>
+#include <QAtomicPointer>
+#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 *);
+    void DiskWriterLoop(void);
+
+    long long WriteFwdFree(const char* readPtr, const char* writePtr) const;
+    long long ReadFwdFree(const char* readPtr, const char* writePtr) const;
+    long long WriteFwdContinuousFree(const char* readPtr, const char* writePtr) const;
+    long long ReadFwdContinuousFree(const char* readPtr, const char* writePtr) const;
+
+    void SetupBuffer(uint reqBufferSize);
+
+  private:
+    // file handling related
+    QString         fileName;
+    int             fileFlags;
+    mode_t          fileMode;
+    int             fileFD;
+    QAtomicInt      fileMinWriteSize; // IMPORTANT CHANGE: kilobytes
+    long long       fileBytesWritten;
+
+    // state flags
+    QAtomicInt  isFlushing;
+    QAtomicInt  isWriterRunning;
+    QAtomicInt  isResettingBuffer;
+    QAtomicInt  isInDestructor;
+    QAtomicInt  isIgnoringWritesToBuffer;
+    QAtomicInt  isIgnoringWritesToDisk;
+
+    // locks and wait conditions
+    mutable QMutex  bufferWriterLock;
+    mutable QMutex  diskWriterLock;
+
+    QWaitCondition  bufferEmpty;
+    QWaitCondition  bufferHasData;
+    QWaitCondition  bufferWroteData;
+
+    // buffer and related
+    char                *buffer;
+    char                *bufferEnd;
+    unsigned long        bufferSize;
+    QAtomicPointer<char> bufferReadPtr;
+    QAtomicPointer<char> bufferWritePtr;
+    QAtomicPointer<char> bufferAheadWritePtr;
+    QAtomicInt           bufferReservationSlot;
+    QAtomicInt           bufferReservationNextActiveSlot;
+
+    // threads
+    pthread_t       writer;
+
+  private:
+    // constants
+    /// default buffer size
+    static const uint DEFAULT_BUFFER_SIZE;
+    /// maximum chunk size to write to disk in one go
+    static const uint MAX_WRITE_SIZE;
+    /// minimum chunk size to write to disk in one go (except when flushing)
+    static const uint MIN_WRITE_SIZE;
+    /// number of extra bytes to reserve before and after the actual buffer
+    static const uint BUFFER_CUSHION;
+    /// maximum write errors for a single call to safe_write before giving up
+    static const uint MAX_WRITE_ERRORS;
+};
+#endif

mythtv/libs/libmythtv/libmythtv.pro

@@ -247 +247 @@
 SOURCES += dtvmultiplex.cpp
 SOURCES += dtvconfparser.cpp        dtvconfparserhelpers.cpp
 
+exists( $$[QT_INSTALL_HEADERS]/QtCore/QAtomicPointer ):exists( $$[QT_INSTALL_HEADERS]/QtCore/QAtomicInt ) {
+    HEADERS -= ThreadedFileWriter.h
+    HEADERS += ThreadedFileWriterLockless.h
+    SOURCES -= ThreadedFileWriter.cpp
+    SOURCES += ThreadedFileWriterLockless.cpp
+}
+
 using_frontend {
     # Recording profile stuff
     HEADERS += profilegroup.h

mythtv/libs/libmythtv/ThreadedFileWriterLockless.cpp

@@ +1 @@
+// ANSI C headers
+#include <cstdio>
+#include <cstdlib>
+#include <cerrno>
+#include <ctime>
+#include <climits>
+
+// Unix C headers
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <unistd.h>
+#include <signal.h>
+#include <fcntl.h>
+
+// MythTV headers
+#include "ThreadedFileWriterLockless.h"
+#include "mythcontext.h"
+#include "compat.h"
+#include "mythverbose.h"
+
+#if defined(_POSIX_SYNCHRONIZED_IO) && _POSIX_SYNCHRONIZED_IO > 0
+#define HAVE_FDATASYNC
+#endif
+
+#define LOC QString("TFWL: ")
+#define LOC_ERR QString("TFWL, Error: ")
+
+const uint ThreadedFileWriterLockless::DEFAULT_BUFFER_SIZE = 2*1024*1024;
+const uint ThreadedFileWriterLockless::MAX_WRITE_SIZE      = DEFAULT_BUFFER_SIZE / 4;
+const uint ThreadedFileWriterLockless::MIN_WRITE_SIZE      = DEFAULT_BUFFER_SIZE / 1024 / 32;
+const uint ThreadedFileWriterLockless::BUFFER_CUSHION      = 512;
+const uint 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 ringbuffer. All data written via ::Write()
+ *  is inserted into the ringbuffer while a seperate thread reads the data from
+ *  the buffer and writes it to disk.
+ *
+ *  The implementation is thread-safe and thus also 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;
+    uint 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;
+    }
+
+    if(errors < MAX_WRITE_ERRORS)
+        ok = true;
+    else
+        ok = false;
+
+    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::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(0), fileBytesWritten(0),
+    // state flags
+    isFlushing(0), isWriterRunning(0), isResettingBuffer(0),  isInDestructor(0),
+    isIgnoringWritesToBuffer(0), isIgnoringWritesToDisk(0),
+    // buffer and related
+    buffer(NULL), bufferEnd(NULL), bufferSize(0), bufferReadPtr(NULL),
+    bufferWritePtr(NULL), bufferAheadWritePtr(NULL),
+    bufferReservationSlot(0), bufferReservationNextActiveSlot(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)
+{
+    // TODO: don't fail with an error, just flush the buffer and end the writer
+    if(isWriterRunning)
+    {
+        VERBOSE(VB_IMPORTANT, LOC_ERR +
+                QString("Opening file '%1' failed. Writer already running")
+                .arg(fileName));
+
+        return false;
+    }
+
+    isFlushing               = 0;
+    isResettingBuffer        = 0;
+    isInDestructor           = 0;
+    isIgnoringWritesToBuffer = 0;
+    isIgnoringWritesToDisk   = 0;
+
+    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);
+        return false;
+    }
+    else
+    {
+        SetupBuffer(DEFAULT_BUFFER_SIZE);
+        fileMinWriteSize = MIN_WRITE_SIZE;
+
+        pthread_create(&writer, NULL, BootWriter, 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.");
+            }
+        }
+*/
+        return true;
+    }
+}
+
+/** \fn ThreadedFileWriterLockless::~ThreadedFileWriterLockless()
+ *  \brief Destructor which commits all writes, ends the DiskWriter thread
+ *         and closes the file.
+ */
+ThreadedFileWriterLockless::~ThreadedFileWriterLockless()
+{
+    isIgnoringWritesToBuffer.fetchAndStoreOrdered(1);
+
+    if (fileFD >= 0)
+    {
+        Flush();
+        isInDestructor.fetchAndStoreOrdered(1); /* tells child thread to exit */
+
+        // TODO: check if this is enough... or still required.
+        bufferHasData.wakeAll();
+        bufferEmpty.wakeAll();
+        pthread_join(writer, 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. It is guaranteed that
+ *  the data won't get fragmented if there are parallel calls to this function.
+ *  However in order to keep the internal data structures sane and prevent
+ *  deadlocks and similar problems, the following limitations exist:
+ *
+ *   - if there concurrent calls and the ringbuffer is filled up to the point
+ *     that there is not enough free space, a fallback to a FIFO-behaviour
+ *     happens until all calls are served
+ *
+ *   - all concurrent calls will return one after one in the order their data
+ *     were written to the internal ringbuffer. Up to that point tough, they
+ *     are fully concurrent.
+ *
+ *  NOTE: Currently it is not supported to write data larger than the current
+ *        ringbuffer size.
+ *
+ *  \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 || isIgnoringWritesToBuffer || isResettingBuffer
+                  || isInDestructor)
+        return 0;
+
+    // TODO: implement proper handling of data chunks larger than buffer size
+    if (count >= bufferSize)
+    {
+        VERBOSE(VB_IMPORTANT, LOC_ERR +
+                QString("ERROR: Unable to buffer data larger than buffer size: "
+                        "%1 > %2.").arg(count).arg(bufferSize));
+
+        return 0;
+    }
+
+    bool hasSlot     = false;
+    bool forcedQueue = (bufferReservationNextActiveSlot != bufferReservationSlot);
+    int  slotNumber  = 0;
+
+    char* localReadPtr           = NULL;
+    char* localWritePtr          = NULL;
+    char* newBufferAheadWritePtr = NULL;
+    do
+    {
+        bool locked = false;
+
+        if(forcedQueue)
+        {
+            if(!hasSlot)
+            {
+                hasSlot    = true;
+                slotNumber = bufferReservationSlot.fetchAndAddOrdered(-1);
+            }
+
+            bufferWriterLock.lock();
+            locked = true;
+
+            do
+            {
+                bufferWroteData.wait(&bufferWriterLock, 100);
+            }
+            while(bufferReservationNextActiveSlot != slotNumber);
+
+            forcedQueue = false;
+        }
+
+        localReadPtr = static_cast<char*>(bufferReadPtr);
+        localWritePtr= static_cast<char*>(bufferAheadWritePtr);
+
+        if(!forcedQueue && WriteFwdFree(localReadPtr, localWritePtr) >= count)
+        {
+            uint contSize = WriteFwdContinuousFree(localReadPtr, localWritePtr);
+
+            if(contSize >= count)
+            {
+                newBufferAheadWritePtr = localWritePtr + count;
+            }
+            else
+            {
+                newBufferAheadWritePtr = buffer + (count - contSize);
+            }
+        }
+        else
+        {
+            forcedQueue = true;
+        }
+
+        if(locked)
+        {
+            locked = false;
+            bufferWriterLock.unlock();
+        }
+    }
+    while(forcedQueue || !bufferAheadWritePtr.testAndSetOrdered(localWritePtr, newBufferAheadWritePtr));
+
+    if(hasSlot)
+        bufferReservationNextActiveSlot.fetchAndAddOrdered(-1);
+
+    long long offset  = 0;
+    uint totalWritten = 0;
+    uint dataOffset   = 0;
+    if(WriteFwdContinuousFree(localReadPtr, localWritePtr) < count)
+    {
+        uint chunkSize = WriteFwdContinuousFree(localReadPtr, localWritePtr);
+        memcpy(localWritePtr, data, chunkSize);
+
+        count          -= chunkSize;
+        totalWritten   += chunkSize;
+        offset          = chunkSize;
+        dataOffset      = chunkSize;
+
+        if(localWritePtr + offset > bufferEnd)
+            offset = -(localWritePtr - buffer);
+    }
+    memcpy(localWritePtr + offset, (char*)data + dataOffset, count);
+    totalWritten += count;
+
+    while(!bufferWritePtr.testAndSetOrdered(localWritePtr, newBufferAheadWritePtr))
+    {
+        struct timespec sleepTime = {0, 10};
+        nanosleep(&sleepTime, NULL);
+    }
+
+    bufferHasData.wakeAll();
+
+    return totalWritten;
+}
+
+/** \fn ThreadedFileWriterLockless::Seek(long long pos, int whence)
+ *  \brief Seek to a position within the disk file
+ *
+ *  \warning This method has not been tested yet!
+ *
+ *  \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)
+{
+    isIgnoringWritesToBuffer.fetchAndStoreOrdered(1);
+
+    Flush();
+
+    long long result = lseek(fileFD, pos, whence);
+
+    isIgnoringWritesToBuffer.fetchAndStoreOrdered(0);
+
+    return result;
+}
+
+/** \fn ThreadedFileWriterLockless::Flush(void)
+ *  \brief Instructs the DiskWriter thread to flush all remaining data in the
+ *         ringbuffer 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 and no concurrent buffer writes are accepted.
+ */
+void ThreadedFileWriterLockless::Flush(void)
+{
+    bool resetIgnoreBufferWrites = true;
+
+    if(!isIgnoringWritesToBuffer.testAndSetOrdered(0, 1))
+        resetIgnoreBufferWrites = false;
+
+    isFlushing.fetchAndStoreOrdered(1);
+
+    // TODO: check isIgnoringWritesTo*, otherwise we might wait forever
+
+    bufferHasData.wakeAll();
+
+    diskWriterLock.lock();
+    while (BufUsed() > 0)
+    {
+        if (!bufferEmpty.wait(&diskWriterLock, 2000))
+           VERBOSE(VB_IMPORTANT, LOC + "Taking a long time to flush...");
+    }
+    diskWriterLock.unlock();
+
+    isFlushing.fetchAndStoreOrdered(0);
+
+    if(resetIgnoreBufferWrites)
+        isIgnoringWritesToBuffer.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)
+    {
+#ifdef HAVE_FDATASYNC
+        fdatasync(fileFD);
+#else
+        fsync(fileFD);
+#endif
+    }
+}
+
+/** \fn ThreadedFileWriterLockless::SetWriteBufferSize(uint newSize)
+ *  \brief Sets the total size of the internal ringbuffer.
+ *
+ *  All data currently in the ringbuffer is flushed to the disk. During the
+ *  time of the resetting, no writes to the buffer are accepted.
+ *
+ *  \warning This is an expensive call.
+ *
+ *  \param newSize New ringbuffer size in bytes
+ */
+void ThreadedFileWriterLockless::SetWriteBufferSize(uint newSize)
+{
+    if (newSize <= 0)
+        return;
+
+    // TODO: spinlock
+    if(!isResettingBuffer.testAndSetOrdered(0, 1))
+        return;
+
+    isIgnoringWritesToBuffer.fetchAndStoreOrdered(1);
+
+    Flush();
+    SetupBuffer(newSize);
+
+    isIgnoringWritesToBuffer.fetchAndStoreOrdered(0);
+    isResettingBuffer.fetchAndStoreOrdered(0);
+}
+
+/** \fn ThreadedFileWriterLockless::SetWriteBufferMinWriteSize(uint newMinSize)
+ *  \brief Sets the minumum number of bytes to write to disk in a single write.
+ *
+ *  \warning The size gets stored in kilobytes internally and is thus
+*            susceptible to rounding effects.
+ *
+ *  \param newMinSize New minimum size in bytes for a single write
+ */
+void ThreadedFileWriterLockless::SetWriteBufferMinWriteSize(uint newMinSize)
+{
+    if (newMinSize <= 0)
+        return;
+
+    newMinSize /= 1024;
+    if(newMinSize > INT_MAX)
+        newMinSize = INT_MAX;
+
+    fileMinWriteSize.fetchAndStoreOrdered(newMinSize);
+}
+
+/** \fn ThreadedFileWriterLockless::DiskLoop(void)
+ *  \brief DiskWriter thread which reads data from the ringbuffer 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)
+{
+    while (!isInDestructor || BufUsed() > 0)
+    {
+        uint bufferUsage = BufUsed();
+
+        if (!bufferUsage)
+            bufferEmpty.wakeAll();
+
+        if (!bufferUsage || isResettingBuffer || (!isInDestructor && !isFlushing &&
+             (((uint)(fileMinWriteSize * 1024) > bufferUsage) &&
+             ((fileMinWriteSize * 1024) <= fileBytesWritten))))
+        {
+            diskWriterLock.lock();
+            bufferHasData.wait(&diskWriterLock, 250);
+            diskWriterLock.unlock();
+            continue;
+        }
+
+        long long  currWriteSize = (bufferUsage > MAX_WRITE_SIZE && !isFlushing) ?
+                                    MAX_WRITE_SIZE : bufferUsage;
+
+        uint  currWritten   = 0;
+        char* localReadPtr  = bufferReadPtr;
+        char* localWritePtr = bufferWritePtr;
+        bool  wasWriteOk    = true;
+        uint  chunkSize     = 0;
+
+        while(wasWriteOk && currWriteSize > 0 && ReadFwdFree(localReadPtr, localWritePtr) > 0)
+        {
+            long long continuousSize = ReadFwdContinuousFree(localReadPtr, localWritePtr);
+
+            chunkSize = (currWriteSize <= continuousSize) ? currWriteSize : continuousSize;
+
+            if(!isIgnoringWritesToDisk)
+                chunkSize = safe_write(fileFD, localReadPtr, chunkSize, wasWriteOk);
+
+            currWriteSize -= chunkSize;
+            currWritten   += chunkSize;
+            localReadPtr  += chunkSize;
+
+            if(localReadPtr > bufferEnd)
+                localReadPtr = buffer;
+        }
+
+        bufferReadPtr.fetchAndStoreOrdered(localReadPtr);
+
+        bufferWroteData.wakeAll();
+
+        if (!isIgnoringWritesToDisk && !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));
+            isIgnoringWritesToBuffer.fetchAndStoreOrdered(1);
+            isIgnoringWritesToDisk.fetchAndStoreOrdered(1);
+        }
+
+        // if we wrote anything, make sure it gets uncached
+        if(!isIgnoringWritesToDisk && currWritten > 0)
+        {
+            posix_fadvise(fileFD, fileBytesWritten, currWritten, POSIX_FADV_DONTNEED);
+        }
+
+        if(!isIgnoringWritesToDisk)
+            fileBytesWritten += currWritten;
+    }
+}
+
+/** \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(bufferReadPtr, bufferWritePtr);
+}
+
+/** \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.
+ */
+long long ThreadedFileWriterLockless::WriteFwdFree(const char* readPtr, const char* writePtr) const
+{
+    return (readPtr - writePtr) + ((readPtr <= writePtr) ? bufferSize : 0) - 1;
+}
+
+/** \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
+ */
+long long ThreadedFileWriterLockless::ReadFwdFree(const char *readPtr,
+                                                  const char *writePtr) const
+{
+    return (writePtr - readPtr) + ((readPtr <= writePtr) ? 0 : bufferSize);
+}
+
+/** \fn ThreadedFileWriterLockless::WriteFwdContinuousFree(void) const
+ *  \brief Gets the number of connected 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.
+ */
+long long ThreadedFileWriterLockless::WriteFwdContinuousFree(const char* readPtr,
+                                                             const char* writePtr) const
+{
+    uint offset = (readPtr > buffer) ? 1 : 0;
+
+    return (readPtr <= writePtr) ? (bufferEnd - writePtr) + offset
+                          : (readPtr - writePtr) - 1;
+}
+
+/** \fn ThreadedFileWriterLockless::ReadFwdContinuousFree(void) const
+ *  \brief Gets the number of connected 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
+ */
+long long ThreadedFileWriterLockless::ReadFwdContinuousFree(const char* readPtr,
+                                                            const char* writePtr) const
+{
+    return ((readPtr <= writePtr) ? writePtr : (bufferEnd + 1)) - readPtr;
+}
+
+/** \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, reqBufferSize + 2 * BUFFER_CUSHION);
+
+    buffer    += BUFFER_CUSHION;
+    bufferSize = reqBufferSize;
+    bufferEnd  = buffer + (reqBufferSize-1);
+
+    bufferWritePtr      = buffer;
+    bufferAheadWritePtr = buffer;
+    bufferReadPtr       = buffer;
+
+    bufferReservationSlot           = 0;
+    bufferReservationNextActiveSlot = 0;
+}

mythtv/libs/libmythtv/RingBuffer.cpp

@@ -23 +23 @@
 #include "RingBuffer.h"
 #include "remotefile.h"
 #include "remoteencoder.h"
+
+#if QT_VERSION < 0x040400
 #include "ThreadedFileWriter.h"
+#else
+#include "ThreadedFileWriterLockless.h"
+#endif
+
 #include "livetvchain.h"
 #include "DVDRingBuffer.h"
 #include "util.h"
@@ -132 +138 @@
 
     if (write)
     {
+#if QT_VERSION < 0x040400
         tfw = new ThreadedFileWriter(
+#else
+        tfw = new ThreadedFileWriterLockless(
+#endif
             filename, O_WRONLY|O_TRUNC|O_CREAT|O_LARGEFILE, 0644);
 
         if (!tfw->Open())

mythtv/libs/libmythtv/RingBuffer.h

@@ -16 +16 @@
 
 class RemoteFile;
 class RemoteEncoder;
+#if QT_VERSION < 0x040400
 class ThreadedFileWriter;
+#else
+class ThreadedFileWriterLockless;
+#endif
 class DVDRingBufferPriv;
 class LiveTVChain;
 
@@ -118 +122 @@
   private:
     QString filename;
 
+#if QT_VERSION < 0x040400
     ThreadedFileWriter *tfw;
+#else
+    ThreadedFileWriterLockless *tfw;
+#endif
     int fd2;
 
     bool writemode;