dataloader.cpp

Go to the documentation of this file.

/*
        Description: async stream reader, used to load repository data on startup

        Author: Marco Costalba (C) 2005-2006

        Copyright: See COPYING file that comes with this distribution

*/
#include <stdlib.h> // rand()
#include <time.h> // time()
#include <qcstring.h>
#include <qdir.h>
#include <qprocess.h>
#include "git.h"
#include "dataloader.h"

#define GUI_UPDATE_INTERVAL 500
#define READ_BLOCK_SIZE     65535

DataLoader::DataLoader(Git* g, FileHistory* f) : QObject(g), git(g), fh(f) {

        canceling = parsing = false;
        isProcExited = true;
        halfChunk = NULL;
        proc = NULL;
        loadedBytes = 0;

        connect(git, SIGNAL(cancelAllProcesses()), this, SLOT(on_cancel()));
        connect(&guiUpdateTimer, SIGNAL(timeout()), this, SLOT(on_timeout()));
}

void DataLoader::on_cancel(const FileHistory* f) {

        if (f == fh)
                on_cancel();
}

bool DataLoader::start(SCList args, SCRef wd) {

        if (!isProcExited) {
                dbs("ASSERT in DataLoader::start, called while processing");
                return false;
        }
        isProcExited = false;
        if (!doStart(args, wd)) {
                deleteLater();
                return false;
        }
        loadTime.start();
        guiUpdateTimer.start(GUI_UPDATE_INTERVAL, true);
        return true;
}

void DataLoader::on_eof() {

        isProcExited = true;

        if (parsing && guiUpdateTimer.isActive())
                dbs("ASSERT in DataLoader: timer active while parsing");

        if (parsing == guiUpdateTimer.isActive() && !canceling)
                dbs("ASSERT in DataLoader: inconsistent timer");

        if (guiUpdateTimer.isActive()) // no need to wait anymore
                guiUpdateTimer.start(1, true);
}

void DataLoader::on_timeout() {

        if (canceling) {
                deleteLater();
                return; // we leave with guiUpdateTimer not active
        }
        parsing = true;

        // process could exit while we are processing so save the flag now
        bool lastBuffer = isProcExited;
        loadedBytes += readNewData(lastBuffer);
        emit newDataReady(fh); // inserting in list view is about 3% of total time

        if (lastBuffer) {
                emit loaded(fh, loadedBytes, loadTime.elapsed(), true, "", "");
                deleteLater();

        } else if (isProcExited) { // exited while parsing
                dbs("Exited while parsing!!!!");
                guiUpdateTimer.start(1, true);
        } else
                guiUpdateTimer.start(GUI_UPDATE_INTERVAL, true);

        parsing = false;
}

void DataLoader::parseSingleBuffer(const QByteArray& ba) {

        if (ba.size() == 0 || canceling)
                return;

        int ofs = 0, newOfs, bz = ba.size();
        while (bz - ofs > 0) {

                if (!halfChunk) {

                        newOfs = git->addChunk(fh, ba, ofs);
                        if (newOfs == -1)
                                break; // half chunk detected

                        ofs = newOfs;

                } else { // less then 1% of cases with READ_BLOCK_SIZE = 64KB

                        int end = ba.find('\0', 0);
                        if (end == -1) // consecutives half chunks
                                break;

                        ofs = end + 1;
                        QGit::baAppend(&halfChunk, &(ba[0]), ofs);
                        fh->rowData.append(halfChunk);
                        addSplittedChunks(halfChunk);
                        halfChunk = NULL;
                }
        }
        // save any remaining half chunk
        if (bz - ofs > 0)
                QGit::baAppend(&halfChunk, &(ba[ofs]),  bz - ofs);
}

void DataLoader::addSplittedChunks(const QByteArray* hc) {

        if (hc->at(hc->size() - 1) != 0) {
                dbs("ASSERT in DataLoader, bad half chunk");
                return;
        }
        // do not assume we have only one chunk in hc
        int ofs = 0;
        while (ofs != -1 && ofs != (int)hc->size())
                ofs = git->addChunk(fh, *hc, ofs);
}

// *************** git interface facility dependant code *****************************

#ifdef USE_QPROCESS

DataLoader::~DataLoader() {}

void DataLoader::on_cancel() {

        if (!canceling) { // just once
                canceling = true;
                if (proc)
                        proc->tryTerminate();
        }
}

bool DataLoader::doStart(SCList args, SCRef wd) {

        proc = new QProcess(args, this);
        proc->setWorkingDirectory(wd);
        if (!proc->start())
                return false;

        connect(proc, SIGNAL(processExited()), this, SLOT(on_eof()));
        // signal readyReadStdout() is not connected, read is timeout based. Faster.
        return true;
}

void DataLoader::on_procDataReady(const QByteArray&) { /* timeout based */ }

ulong DataLoader::readNewData(bool) {

        /*
           QByteArray copy c'tor uses shallow copy, but there is a deep copy in
           QProcess::readStdout(), from an internal buffers list to return value.

           Qt uses a select() to detect new data is ready, copies immediately the
           data to the heap with a read() and stores the pointer to new data in a
           pointer list, from qprocess_unix.cpp:

                const int basize = 4096;
                QByteArray *ba = new QByteArray( basize );
                n = ::read( fd, ba->data(), basize );
                buffer->append( ba ); // added to a QPtrList<QByteArray> pointer list

           When we call QProcess::readStdout() data from buffers pointed by the
           pointer list is memcpy() to the function return value, from qprocess.cpp:

                ....
                return buf->readAll(); // memcpy() here
        */
        QByteArray* ba = new QByteArray(proc->readStdout());
        if (ba->size() == 0) {
                delete ba;
                return 0;
        }
        fh->rowData.append(ba);
        parseSingleBuffer(*ba);
        return ba->size();
}

#else // temporary file as data exchange facility

DataLoader::~DataLoader() {

        if (dataFile.isOpen())
                dataFile.close();

        QDir dir;
        dir.remove(dataFileName);
        dir.remove(scriptFileName);
}

void DataLoader::on_cancel() {

        if (!canceling) { // just once
                canceling = true;
                git->cancelProcess(proc);
                if (!procPID.isEmpty()) {
                        git->errorReportingEnabled = false; // hide error when already terminated
                        git->run("kill " + procPID.stripWhiteSpace());
                        git->errorReportingEnabled = true;
                }
        }
}

bool DataLoader::doStart(SCList args, SCRef wd) {

        // create a script to redirect 'git rev-list' stdout to dataFile
        const QString tmpfsDir("/tmp");
        QDir dir(tmpfsDir); // use a tmpfs mounted filesystem if available
        bool useTmpfsDir = (dir.exists() && dir.isReadable());
        bool noexec;
        srand (time(NULL));
        do {
                do {
                        // ensure unique names for our DataLoader instance file
                        QString t = QString::number(rand(), 16);
                        dataFileName = "/qgit_" + t + ".txt";
                        scriptFileName = "/qgit_" + t + ".sh";
                        dataFileName.prepend(useTmpfsDir ? tmpfsDir : wd);
                        scriptFileName.prepend(useTmpfsDir ? tmpfsDir : wd);
                } while (dir.exists(dataFileName) || dir.exists(scriptFileName));

                /* in case we use '/tmp' be sure is writable and executable */
                noexec = ( useTmpfsDir &&
                          !QGit::writeToFile(scriptFileName, "test exec bit", true));
                if (noexec)
                        useTmpfsDir = false;
        } while (noexec);

        dataFile.setName(dataFileName);
        QString runCmd;
        FOREACH_SL (it, args)
                if ((*it).contains(' '))
                        runCmd.append("\"" + *it + "\" ");
                else
                        runCmd.append(*it + " ");

        runCmd.append("> " +  dataFileName);
        runCmd.append(" &\necho $!\nwait"); // we want to read git-rev-list PID
        runCmd.prepend("cd " + wd + "\n");
        if (!QGit::writeToFile(scriptFileName, runCmd, true))
                return false;

        proc = git->runAsync(scriptFileName, this, "");
        return (proc != NULL);
}

void DataLoader::on_procDataReady(const QByteArray& data) {
// the script sends pid of launched git-rev-list, to be used for canceling

        procPID.append(data);
}

ulong DataLoader::readNewData(bool lastBuffer) {

        bool ok =     dataFile.isOpen()
                  || (dataFile.exists() && dataFile.open(IO_Raw | IO_ReadOnly));
        if (!ok)
                return 0;

        ulong cnt = 0;
        while (!dataFile.atEnd()) {

                QByteArray* ba = new QByteArray(READ_BLOCK_SIZE);

                // this is the ONLY deep copy involved in the whole loading
                // QFile::readBlock() calls standard C read() function when
                // file is open with IO_Raw flag, or fread() otherwise
                uint len = dataFile.readBlock(ba->data(), READ_BLOCK_SIZE);
                if (len <= 0) {
                        delete ba;
                        break;

                } else if (len < READ_BLOCK_SIZE) // unlikely
                        ba->resize(len, QGArray::SpeedOptim);

                cnt += len;
                fh->rowData.append(ba);
                parseSingleBuffer(*ba);

                // avoid reading small chunks if data producer is still running
                if (len < READ_BLOCK_SIZE && !lastBuffer)
                        break;
        }
        return cnt;
}

#endif // USE_QPROCESS

---