inputHandlerDavix.cpp

Go to the documentation of this file.

#include "inputHandlerDavix.h"
#include "timer.h"
 
#include <memory>
#include "block.h"
#include "ewmscp.h"
 
#include <errMsgQueue.h>
 
namespace inputHandler {
    decltype(davix::factory) davix::factory("davix");
    std::unique_ptr<base::reader> davix::newReader(const std::string& aPath,
            copyRequest::stateType &state,
            const genericStat& inititalStat) {
        return std::unique_ptr<reader>(new readerDavix(*this, aPath, state, inititalStat));
    }
 
    static davixConfigObject inputConfig("davixIn");
 
    davix::davix():
        davixCommon(inputConfig) {
    };
 
 
    std::unique_ptr<base::Directory> davix::getDirectory(const std::string& path) {
        return std::unique_ptr<Directory>(new DavixDirectory(path, *this));
    }
    davix::DavixDirectory::DavixDirectory(const std::string& aPath, davixCommon& aHandler):
        Directory(aPath),
        handler(aHandler) {
        errorReport report(__func__, "opendirpp", path);
        dir = handler.posix.opendirpp(&handler.params, path, report);
        if (dir == nullptr) {
            report.throwUp();
        }
    }
    davix::DavixDirectory::~DavixDirectory() noexcept(false) {
        errorReport report(__func__, "closedirpp", path);
        if (handler.posix.closedirpp(dir, report)) {
            if (isUnwinding()) {
                errMsg::emit(errMsg::level::debug, errMsg::location(),
                             path, "close directory during unwind ",
                             report.getMessage());
            } else {
                report.throwUp();
            }
        }
    }
    std::unique_ptr<base::Directory::Entry> davix::DavixDirectory::getNextEntry(bool /*ignoreMissing*/) {
        errorReport report(__func__, "readdirpp", path);
        struct stat statbuf;
        auto entry = handler.posix.readdirpp(dir, &statbuf, report);
        if (entry == nullptr) {
            return nullptr;
        }
        auto genstat = std::unique_ptr<const genericStat>(new genericStat(statbuf, std::chrono::seconds(1)));
        return std::unique_ptr<Entry>(new Entry(entry->d_name, genstat));
    }
 
    davix::readerDavix::readerDavix(davixCommon& aHandler,
                                    const std::string& aPath,
                                    copyRequest::stateType &state,
                                    const genericStat& inititalStat):
        davixIoCommon(aPath, aHandler),
        reader(inititalStat) {
        errorReport report(__func__, "open", path);
        fd = handler.posix.open(&handler.params, path.c_str(), O_RDONLY, report);
        if (fd == nullptr) {
            if (report->getStatus() == Davix::StatusCode::FileNotFound) {
                state |= copyRequest::stateBitType::vanished;
            }
            report.throwUp();
        }
        blockSize = readInitialStat.blksize;
    }
    davix::readerDavix::~readerDavix() noexcept(false) {
        if (fd) {
            errorReport report(__func__, "close", path);
            if (handler.posix.close(fd, report) != 0) {
                if (isUnwinding()) {
                    errMsg::emit(errMsg::level::debug, errMsg::location(),
                                 path, "close during unwind ",
                                 report.getMessage());
                    return;
                }
                report.throwUp();
            }
        }
    }
 
 
    bool davix::readerDavix::readBlock(block &b) {
        b.clear(totalBytesRead);
        bool lastblock = false;
        while (b.size() + blockSize <= b.max_size()) {
            readRateLimit.wait();
            errorReport report(__func__, "read", path);
            timerInst(read);
            auto bytes_read = handler.posix.read(fd, b.bufferAt(b.size()), blockSize, report);
            if (bytes_read < 0) {
                report.throwUp();
            }
            if (bytes_read == 0) {
                lastblock = true;
                if (totalBytesRead < readInitialStat.size) {
                    throw delayAdvisingError(path + " has shrunk while reading, (" +
                                             std::to_string(readInitialStat.size) +
                                             " -> " +
                                             std::to_string(totalBytesRead) +
                                             ")");
                }
                break;
            }
            readRateLimit.update(bytes_read);
            totalBytesRead += bytes_read;
            if (totalBytesRead > readInitialStat.size) {
                throw delayAdvisingError(path + " has grown while reading, (" +
                                         std::to_string(readInitialStat.size) +
                                         " -> " +
                                         std::to_string(totalBytesRead) +
                                         ")");
            }
            b.bump_size(bytes_read);
        }
 
        return lastblock;
    };
 
    bool davix::readerDavix::parallelizable() const {
        return true;
    }
 
    void davix::readerDavix::readBlockP(block& b, size_t bytesToRead, off_t offset) {
        b.clear(offset);
        while (b.size() + blockSize <= b.max_size()) {
            errorReport report(__func__, "pread", path);
            timerInst(pread);
            auto bytes_read = handler.posix.pread64(fd, b.bufferAt(b.size()), blockSize, offset + b.size(),report);
            if (bytes_read < 0)
                report.throwUp();
            if (bytes_read == 0) {
                break;
            }
            b.bump_size(bytes_read);
            if (b.size() > bytesToRead) {
                throw delayAdvisingError(path + " has grown while reading");
            }
        }
        if (b.size() < bytesToRead) {
            throw delayAdvisingError(path + " has shrunk while reading "
                                     + std::to_string(bytesToRead)
                                     + " "
                                     + std::to_string(b.size()));
        }
    }
 
    
    void davix::readerDavix::checkUnchangedness() {
    };
} // end namespace inputHandler