ThreadedFileWriter.cpp

Go to the documentation of this file.

// ANSI C headers
#include <cstdio>
#include <cstdlib>
#include <cerrno>

// Unix C headers
#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>
#include <signal.h>
#include <fcntl.h>

// Qt headers
#include <qstring.h>
#include <qdeepcopy.h>

// MythTV headers
#include "ThreadedFileWriter.h"
#include "mythcontext.h"
#include "compat.h"

#if defined(_POSIX_SYNCHRONIZED_IO) && _POSIX_SYNCHRONIZED_IO > 0
#define HAVE_FDATASYNC
#endif

#define LOC QString("TFW: ")
#define LOC_ERR QString("TFW, Error: ")

const uint ThreadedFileWriter::TFW_DEF_BUF_SIZE   = 2*1024*1024;
const uint ThreadedFileWriter::TFW_MAX_WRITE_SIZE = TFW_DEF_BUF_SIZE / 4;
const uint ThreadedFileWriter::TFW_MIN_WRITE_SIZE = TFW_DEF_BUF_SIZE / 32;

static uint safe_write(int fd, const void *data, uint sz, bool &ok)
{
    int ret;
    uint tot = 0;
    uint errcnt = 0;

    while (tot < sz)
    {
        ret = write(fd, (char *)data + tot, sz - tot);
        if (ret < 0)
        {
            if (errno == EAGAIN)
            {
                VERBOSE(VB_IMPORTANT, LOC + "safe_write(): Got EAGAIN.");
                continue;
            }

            errcnt++;
            VERBOSE(VB_IMPORTANT, LOC_ERR + "safe_write(): File I/O " +
                    QString(" errcnt: %1").arg(errcnt) + ENO);

            if (errcnt == 3)
                break;
        }
        else
        {
            tot += ret;
        }

        if (tot < sz)
        {
            VERBOSE(VB_IMPORTANT, LOC + "safe_write(): funky usleep");
            usleep(1000);
        }
    }
    ok = (errcnt < 3);
    return tot;
}

void *ThreadedFileWriter::boot_writer(void *wotsit)
{
#ifndef USING_MINGW
    signal(SIGXFSZ, SIG_IGN);
#endif
    ThreadedFileWriter *fw = (ThreadedFileWriter *)wotsit;
    fw->DiskLoop();
    return NULL;
}

void *ThreadedFileWriter::boot_syncer(void *wotsit)
{
    ThreadedFileWriter *fw = (ThreadedFileWriter *)wotsit;
    fw->SyncLoop();
    return NULL;
}

ThreadedFileWriter::ThreadedFileWriter(const QString &fname,
                                       int pflags, mode_t pmode) :
    // file stuff
    filename(QDeepCopy<QString>(fname)), flags(pflags),
    mode(pmode),                         fd(-1),
    // state
    no_writes(false),                    flush(false),
    in_dtor(false),                      ignore_writes(false),
    tfw_min_write_size(0),
    // buffer position state
    rpos(0),                             wpos(0),
    written(0),
    // buffer
    buf(NULL),                           tfw_buf_size(0)
{
}

bool ThreadedFileWriter::Open(void)
{
    ignore_writes = false;
    fd = open(filename.ascii(), flags, mode);

    if (fd < 0)
    {
        VERBOSE(VB_IMPORTANT, LOC_ERR + 
                QString("Opening file '%1'.").arg(filename) + ENO);
        return false;
    }
    else
    {
        buf = new char[TFW_DEF_BUF_SIZE + 1024];
        bzero(buf, TFW_DEF_BUF_SIZE + 64);

        tfw_buf_size = TFW_DEF_BUF_SIZE;
        tfw_min_write_size = TFW_MIN_WRITE_SIZE;
        pthread_create(&writer, NULL, boot_writer, this);
        pthread_create(&syncer, NULL, boot_syncer, this);
        return true;
    }
}

ThreadedFileWriter::~ThreadedFileWriter()
{
    no_writes = true;

    if (fd >= 0)
    {
        Flush();
        in_dtor = true; /* tells child thread to exit */

        bufferSyncWait.wakeAll();
        pthread_join(syncer, NULL);

        bufferHasData.wakeAll();
        pthread_join(writer, NULL);
        close(fd);
        fd = -1;
    }

    if (buf)
    {
        delete [] buf;
        buf = NULL;
    }
}

uint ThreadedFileWriter::Write(const void *data, uint count)
{
    if (count == 0)
        return 0;

    bool first = true;

    while (count > BufFree())
    {
        if (first)
        {
            VERBOSE(VB_IMPORTANT, LOC_ERR + "Write() -- IOBOUND begin " + 
                    QString("cnt(%1) free(%2)").arg(count).arg(BufFree()));
            first = false;
        }

        bufferWroteData.wait(100);
    }
    if (!first)
        VERBOSE(VB_IMPORTANT, LOC_ERR + "Write() -- IOBOUND end");

    if (no_writes)
        return 0;

    if ((wpos + count) > tfw_buf_size)
    {
        int first_chunk_size = tfw_buf_size - wpos;
        int second_chunk_size = count - first_chunk_size;
        memcpy(buf + wpos, data, first_chunk_size );
        memcpy(buf, (char *)data + first_chunk_size, second_chunk_size );
    }
    else
    {
        memcpy(buf + wpos, data, count);
    }

    buflock.lock();
    wpos = (wpos + count) % tfw_buf_size;
    buflock.unlock();

    bufferHasData.wakeAll();

    return count;
}

long long ThreadedFileWriter::Seek(long long pos, int whence)
{
    Flush();

    return lseek(fd, pos, whence);
}

void ThreadedFileWriter::Flush(void)
{
    flush = true;
    while (BufUsed() > 0)
    {
        if (!bufferEmpty.wait(2000))
            VERBOSE(VB_IMPORTANT, LOC + "Taking a long time to flush..");
    }
    flush = false;
}

void ThreadedFileWriter::Sync(void)
{
    if (fd >= 0)
    {
#ifdef HAVE_FDATASYNC
        fdatasync(fd);
#else
        fsync(fd);
#endif
    }
}

void ThreadedFileWriter::SetWriteBufferSize(uint newSize)
{
    if (newSize <= 0)
        return;

    Flush();

    buflock.lock();
    delete [] buf;
    rpos = wpos = 0;
    buf = new char[newSize + 1024];
    bzero(buf, newSize + 64);
    tfw_buf_size = newSize;
    buflock.unlock();
}

void ThreadedFileWriter::SetWriteBufferMinWriteSize(uint newMinSize)
{
    if (newMinSize <= 0)
        return;

    tfw_min_write_size = newMinSize;
}

void ThreadedFileWriter::SyncLoop(void)
{
    while (!in_dtor)
    {
        bufferSyncWait.wait(written > tfw_min_write_size ? 1000 : 100);
        Sync();
    }
}

void ThreadedFileWriter::DiskLoop(void)
{
    uint size = 0;
    written = 0;

    while (!in_dtor || BufUsed() > 0)
    {
        size = BufUsed();

        if (size == 0)
            bufferEmpty.wakeAll();

        if (!size || (!in_dtor && !flush &&
            ((size < tfw_min_write_size) &&
             (written >= tfw_min_write_size))))
        {
            bufferHasData.wait(100);
            continue;
        }

        /* cap the max. write size. Prevents the situation where 90% of the
           buffer is valid, and we try to write all of it at once which
           takes a long time. During this time, the other thread fills up
           the 10% that was free... */
        size = (size > TFW_MAX_WRITE_SIZE) ? TFW_MAX_WRITE_SIZE : size;

        bool write_ok;
        if (ignore_writes)
            ;
        else if ((rpos + size) > tfw_buf_size)
        {
            int first_chunk_size  = tfw_buf_size - rpos;
            int second_chunk_size = size - first_chunk_size;
            size = safe_write(fd, buf+rpos, first_chunk_size, write_ok);
            if ((int)size == first_chunk_size && write_ok)
                size += safe_write(fd, buf, second_chunk_size, write_ok);
        }
        else
        {
            size = safe_write(fd, buf+rpos, size, write_ok);
        }


        if (!ignore_writes && !write_ok && (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));
            ignore_writes = true;
        }

        if (written <= tfw_min_write_size)
        {
            written += size;
        }

        buflock.lock();
        rpos = (rpos + size) % tfw_buf_size;
        buflock.unlock();

        bufferWroteData.wakeAll();
    }
}

uint ThreadedFileWriter::BufUsed(void)
{
    QMutexLocker locker(&buflock);
    return (wpos >= rpos) ? wpos - rpos : tfw_buf_size - rpos + wpos;
}

uint ThreadedFileWriter::BufFree(void)
{
    QMutexLocker locker(&buflock);
    return ((wpos >= rpos) ? (rpos + tfw_buf_size) : rpos) - wpos - 1;
}

---