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>
#include <string.h>

// Qt headers
#include <QString>

// MythTV headers
#include "ThreadedFileWriter.h"
#include "mythlogging.h"

#include "mythtimer.h"
#include "compat.h"

#define LOC QString("TFW(%1:%2): ").arg(filename).arg(fd)

void TFWWriteThread::run(void)
{
    RunProlog();
    m_parent->DiskLoop();
    RunEpilog();
}

void TFWSyncThread::run(void)
{
    RunProlog();
    m_parent->SyncLoop();
    RunEpilog();
}

const uint ThreadedFileWriter::kMaxBufferSize = 128 * 1024 * 1024;
const uint ThreadedFileWriter::kMinWriteSize = 64 * 1024;

ThreadedFileWriter::ThreadedFileWriter(const QString &fname,
                                       int pflags, mode_t pmode) :
    // file stuff
    filename(fname),                     flags(pflags),
    mode(pmode),                         fd(-1),
    // state
    flush(false),                        in_dtor(false),
    ignore_writes(false),                tfw_min_write_size(kMinWriteSize),
    totalBufferUse(0),
    // threads
    writeThread(NULL),                   syncThread(NULL)
{
    filename.detach();
}

bool ThreadedFileWriter::ReOpen(QString newFilename)
{
    Flush();

    buflock.lock();

    if (fd >= 0)
    {
        close(fd);
        fd = -1;
    }

    if (!newFilename.isEmpty())
        filename = newFilename;

    buflock.unlock();

    return Open();
}

bool ThreadedFileWriter::Open(void)
{
    ignore_writes = false;

    if (filename == "-")
        fd = fileno(stdout);
    else
    {
        QByteArray fname = filename.toLocal8Bit();
        fd = open(fname.constData(), flags, mode);
    }

    if (fd < 0)
    {
        LOG(VB_GENERAL, LOG_ERR, LOC +
            QString("Opening file '%1'.").arg(filename) + ENO);
        return false;
    }
    else
    {
        LOG(VB_FILE, LOG_INFO, LOC + "Open() successful");

#ifdef USING_MINGW
        _setmode(fd, _O_BINARY);
#endif
        if (!writeThread)
        {
            writeThread = new TFWWriteThread(this);
            writeThread->start();
        }

        if (!syncThread)
        {
            syncThread = new TFWSyncThread(this);
            syncThread->start();
        }

        return true;
    }
}

ThreadedFileWriter::~ThreadedFileWriter()
{
    Flush();

    {  /* tell child threads to exit */
        QMutexLocker locker(&buflock);
        in_dtor = true;
        bufferSyncWait.wakeAll();
        bufferHasData.wakeAll();
    }

    if (writeThread)
    {
        writeThread->wait();
        delete writeThread;
        writeThread = NULL;
    }

    while (!writeBuffers.empty())
    {
        delete writeBuffers.front();
        writeBuffers.pop_front();
    }

    while (!emptyBuffers.empty())
    {
        delete emptyBuffers.front();
        emptyBuffers.pop_front();
    }

    if (syncThread)
    {
        syncThread->wait();
        delete syncThread;
        syncThread = NULL;
    }

    if (fd >= 0)
    {
        close(fd);
        fd = -1;
    }
}

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

    QMutexLocker locker(&buflock);

    if (ignore_writes)
        return count;

    if (totalBufferUse + count > kMaxBufferSize)
    {
        LOG(VB_GENERAL, LOG_ERR, LOC +
                "Maximum buffer size exceeded."
                "\n\t\t\tfile will be truncated, no further writing "
                "will be done."
                "\n\t\t\tThis generally indicates your disk performance "
                "\n\t\t\tis insufficient to deal with the number of on-going "
                "\n\t\t\trecordings, or you have a disk failure.");
        ignore_writes = true;
        return count;
    }

    TFWBuffer *buf = NULL;

    if (!writeBuffers.empty() &&
        (writeBuffers.back()->data.size() + count) < kMinWriteSize)
    {
        buf = writeBuffers.back();
        writeBuffers.pop_back();
    }
    else
    {
        if (!emptyBuffers.empty())
        {
            buf = emptyBuffers.front();
            emptyBuffers.pop_front();
            buf->data.clear();
        }
        else
        {
            buf = new TFWBuffer();
        }
    }

    totalBufferUse += count;
    const char *cdata = (const char*) data;
    buf->data.insert(buf->data.end(), cdata, cdata+count);
    buf->lastUsed = QDateTime::currentDateTime();

    writeBuffers.push_back(buf);

    bufferHasData.wakeAll();

    LOG(VB_FILE, LOG_DEBUG, LOC + QString("Write(*, %1) total %2 cnt %3")
            .arg(count,4).arg(totalBufferUse).arg(writeBuffers.size()));

    return count;
}

long long ThreadedFileWriter::Seek(long long pos, int whence)
{
    QMutexLocker locker(&buflock);
    flush = true;
    while (!writeBuffers.empty())
    {
        bufferHasData.wakeAll();
        if (!bufferEmpty.wait(locker.mutex(), 2000))
        {
            LOG(VB_GENERAL, LOG_WARNING, LOC +
                QString("Taking a long time to flush.. buffer size %1")
                    .arg(totalBufferUse));
        }
    }
    flush = false;
    return lseek(fd, pos, whence);
}

void ThreadedFileWriter::Flush(void)
{
    QMutexLocker locker(&buflock);
    flush = true;
    while (!writeBuffers.empty())
    {
        bufferHasData.wakeAll();
        if (!bufferEmpty.wait(locker.mutex(), 2000))
        {
            LOG(VB_GENERAL, LOG_WARNING, LOC +
                QString("Taking a long time to flush.. buffer size %1")
                    .arg(totalBufferUse));
        }
    }
    flush = false;
}

void ThreadedFileWriter::Sync(void)
{
    if (fd >= 0)
    {
#if defined(_POSIX_SYNCHRONIZED_IO) && _POSIX_SYNCHRONIZED_IO > 0
        // fdatasync tries to avoid updating metadata, but will in
        // practice always update metadata if any data is written
        // as the file will usually have grown.
        fdatasync(fd);
#else
        fsync(fd);
#endif
    }
}

void ThreadedFileWriter::SetWriteBufferMinWriteSize(uint newMinSize)
{
    QMutexLocker locker(&buflock);
    if (newMinSize > 0)
        tfw_min_write_size = newMinSize;
    bufferHasData.wakeAll();
}

void ThreadedFileWriter::SyncLoop(void)
{
    QMutexLocker locker(&buflock);
    while (!in_dtor)
    {
        locker.unlock();

        Sync();

        locker.relock();
        bufferSyncWait.wait(&buflock, 1000);
    }
}

void ThreadedFileWriter::DiskLoop(void)
{
#ifndef USING_MINGW
    // don't exit program if file gets larger than quota limit..
    signal(SIGXFSZ, SIG_IGN);
#endif

    QMutexLocker locker(&buflock);

    // 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;
    minWriteTimer.start();

    while (!in_dtor)
    {
        if (ignore_writes)
        {
            while (!writeBuffers.empty())
            {
                delete writeBuffers.front();
                writeBuffers.pop_front();
            }
            while (!emptyBuffers.empty())
            {
                delete emptyBuffers.front();
                emptyBuffers.pop_front();
            }
            bufferEmpty.wakeAll();
            bufferHasData.wait(locker.mutex());
            continue;
        }

        if (writeBuffers.empty())
        {
            bufferEmpty.wakeAll();
            bufferHasData.wait(locker.mutex(), 1000);
            TrimEmptyBuffers();
            continue;
        }

        int mwte = minWriteTimer.elapsed();
        if (!flush && (mwte < 250) && (totalBufferUse < kMinWriteSize))
        {
            bufferHasData.wait(locker.mutex(), 250 - mwte);
            TrimEmptyBuffers();
            continue;
        }

        if (fd == -1)
        {
            bufferHasData.wait(locker.mutex(), 200);
            TrimEmptyBuffers();
            continue;
        }

        TFWBuffer *buf = writeBuffers.front();
        writeBuffers.pop_front();
        totalBufferUse -= buf->data.size();
        minWriteTimer.start();


        const void *data = &(buf->data[0]);
        uint sz = buf->data.size();

        bool write_ok = true;
        uint tot = 0;
        uint errcnt = 0;

        LOG(VB_FILE, LOG_DEBUG, LOC + QString("write(%1) cnt %2 total %3")
                .arg(sz).arg(writeBuffers.size())
                .arg(totalBufferUse));

        MythTimer writeTimer;
        writeTimer.start();

        while ((tot < sz) && !in_dtor)
        {
            locker.unlock();

            int ret = write(fd, (char *)data + tot, sz - tot);

            if (ret < 0)
            {
                if (errno == EAGAIN)
                {
                    LOG(VB_GENERAL, LOG_WARNING, LOC + "Got EAGAIN.");
                }
                else
                {
                    errcnt++;
                    LOG(VB_GENERAL, LOG_ERR, LOC + "File I/O " +
                        QString(" errcnt: %1").arg(errcnt) + ENO);
                }

                if ((errcnt >= 3) || (ENOSPC == errno) || (EFBIG == errno))
                {
                    locker.relock();
                    write_ok = false;
                    break;
                }
            }
            else
            {
                tot += ret;
            }

            locker.relock();

            if (!in_dtor)
                bufferHasData.wait(locker.mutex(), 50);
        }


        buf->lastUsed = QDateTime::currentDateTime();
        emptyBuffers.push_back(buf);

        if (writeTimer.elapsed() > 1000)
        {
            LOG(VB_GENERAL, LOG_WARNING, LOC +
                QString("write(%1) cnt %2 total %3 -- took a long time, %4 ms")
                    .arg(sz).arg(writeBuffers.size())
                    .arg(totalBufferUse).arg(writeTimer.elapsed()));
        }

        if (!write_ok && ((EFBIG == errno) || (ENOSPC == errno)))
        {
            QString msg;
            switch (errno)
            {
                case EFBIG:
                    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.";
                    break;
                case ENOSPC:
                    msg =
                        "No space left on the device for file '%1'"
                        "\n\t\t\t"
                        "file will be truncated, no further writing "
                        "will be done.";
                    break;
            }

            LOG(VB_GENERAL, LOG_ERR, LOC + msg.arg(filename));
            ignore_writes = true;
        }
    }
}

void ThreadedFileWriter::TrimEmptyBuffers(void)
{
    QDateTime cur = QDateTime::currentDateTime();
    QDateTime cur_m_60 = cur.addSecs(-60);

    QList<TFWBuffer*>::iterator it = emptyBuffers.begin();
    while (it != emptyBuffers.end())
    {
        if (((*it)->lastUsed < cur_m_60) ||
            ((*it)->data.capacity() > 3 * (*it)->data.size() &&
             (*it)->data.capacity() > 64 * 1024))
        {
            delete *it;
            it = emptyBuffers.erase(it);
            continue;
        }
        ++it;
    }
}