inputHandlerPosixFile.cpp

Go to the documentation of this file.

#include "inputHandlerPosixFile.h"
#include "block.h"
#include "copyRequestTypes.h"
#include "ewmscp.h"
#include "timer.h"
 
#include <fcntl.h>
#include <errMsgQueue.h>
#include <throwcall.h>
#include <unistd.h>
 
#include <memory>
 
namespace inputHandler {
    decltype(posixFile::factory) posixFile::factory("posixFile");
    decltype(posixFile::preserveAtime) posixFile::preserveAtime('\0', "preserveAtime",
            "preserve atime of inpout files", false);
    std::unique_ptr<base::reader> posixFile::newReader(const std::string& aPath,
            copyRequest::stateType &state,
            const genericStat& inititalStat) {
        return std::unique_ptr<base::reader>(new readerPosixFile(aPath,
                                             state,
                                             inititalStat));
    }
 
 
    bool posixFile::readLinkTarget(const std::string& path,
                                   std::vector<char>& target) {
        timerInst(readlink);
        auto linklength = readlink(path.c_str(), target.data(), target.size());
        if (linklength == -1 && errno == ENOENT) {
            return false;
        }
        throwcall::badval(linklength, -1, "could not read link ", path);
        if (linklength >= static_cast<ssize_t>(target.size())) {
            throw std::runtime_error("link size increased after stat for " + path);
        }
        target[linklength] = '\0';
        return true;
    }
 
 
 
    posixFile::readerPosixFile::readerPosixFile(const std::string& aPath,
            copyRequest::stateType &state,
            const genericStat&inititalStat):
        posixFileIoCommon(aPath),
        reader(inititalStat) {
        {
            timerInst(open);
            fd = open(path.c_str(), O_RDONLY);
        }
        if (fd == -1) {
            if (errno == ENOENT
                    || (errno == ENOTDIR && !S_ISDIR(readInitialStat.mode))) {
                state |= copyRequest::stateBitType::vanished;
            }
        }
        throwcall::badval(fd, -1, "can't open ", path, " for reading");
        blockSize = readInitialStat.blksize;
        {
            timerInst(posix_fadvise);
            throwcall::good0(posix_fadvise(fd, 0, 0, POSIX_FADV_SEQUENTIAL),
                             "can't advise ", path, " as sequential");
        }
        {
            timerInst(posix_fadvise);
            throwcall::good0(posix_fadvise(fd, 0, 0, POSIX_FADV_NOREUSE),
                             "can't advise ", path, " as use only once");
        }
    }
 
    posixFile::readerPosixFile::~readerPosixFile() {
        if (fd != -1) {
            if (isUnwinding()) {
                timerInst(close);
                if (close(fd) != 0) {
                    errMsg::emit(errMsg::level::debug, errMsg::location(),
                                 path, "close during unwind ",
                                 std::system_category().default_error_condition(errno).message());
                }
                return;
            }
            if (preserveAtime) {
                struct timespec times[2];
                readInitialStat.getAtime(times[0]);
                times[1].tv_nsec = UTIME_OMIT; // leave mtime unchanged
                timerInst(futimens);
                throwcall::good0(futimens(fd, times), "can't set time stamps on ", path, " to old values");
            }
            {
                timerInst(close);
                throwcall::good0(close(fd), "can't close ", path, " after reading");
            }
        }
    }
 
 
 
    bool posixFile::readerPosixFile::setupSparseRegions(const std::string& sparseHandling) {
        if (readInitialStat.size < readInitialStat.blksize) { // cannot be sparse
            return false;
        }
        if (sparseHandling == "auto" &&
            readInitialStat.size <= readInitialStat.sizeOnDisk) {
            return false;
        } else if (sparseHandling == "never") {
            return false;
        }
        auto what = SEEK_HOLE;
        for (std::remove_const<decltype(readInitialStat.size)>::type pos = 0;
                pos < readInitialStat.size;
                what = what == SEEK_HOLE ? SEEK_DATA : SEEK_HOLE) {
            timerInst(lseek);
            size_t next;
            auto result = lseek(fd, pos, what);
            if (result == -1 && errno == ENXIO) { // we found a hole at the end of the file
                next = readInitialStat.size;
            } else {
                throwcall::badval(result, -1,
                                  "seek ", path, " for ", what == SEEK_HOLE ? "hole" : "data",
                                  ", from offset ", pos);
                next = result;
            }
            if (next != pos) {
                regions.emplace_back(pos, next, what == SEEK_DATA);
                pos = next;
            }
        }
        timerInst(lseek);
        throwcall::good0(lseek(fd, 0, SEEK_SET), "re-seek to start of ", path);
        return !regions.empty();
    }
 
 
    void posixFile::readerPosixFile::seek(size_t pos) {
        throwcall::badval(lseek(fd, pos, SEEK_SET),
                          -1, "can't seek ", path, " to ", pos);
        if (! regions.empty()) {
            while (regions.front().getEnd() < pos) {
                regions.pop_front();
                if (regions.empty()) {
                    break;
                }
            }
        }
    }
    bool posixFile::readerPosixFile::parallelizable() const {
        return regions.empty();
    }
    bool posixFile::readerPosixFile::readBlock(block& b) {
        b.clear(totalBytesRead);
        bool lastblock = false;
 
        auto bytesToRead = b.max_size();
        if (!regions.empty()) {
            auto region = regions.front();
            if (region.isHole()) {
                b.setHoleState(true);
                b.bump_size(region.size());
                timerInst(lseek);
                size_t nextData;
                auto result = lseek(fd, totalBytesRead, SEEK_DATA);
                if (result == -1 && errno == ENXIO) { // this is the final hole
                    nextData = region.getEnd();
                } else {
                    throwcall::badval(result, -1, "seek in ", path, " to data after ", totalBytesRead);
                    nextData = result;
                }
                totalBytesRead += region.size(); // hole counts as total size...
                if (nextData != totalBytesRead) {
                    throw delayAdvisingError(path + " hole moved druring reading, (" +
                                             std::to_string(totalBytesRead) +
                                             " -> " +
                                             std::to_string(nextData) +
                                             ")");
                }
                regions.pop_front();
                return regions.empty();
            } else { // we are somewhere in a data region...
                if (region.getEnd() < totalBytesRead + bytesToRead) {
                    bytesToRead = region.getEnd() - totalBytesRead; // read only up to end of region
                }
                if (region.getEnd() == totalBytesRead + bytesToRead) { // we finish the region in this block
                    regions.pop_front();
                }
            }
        }
 
        while (b.size() + blockSize <= bytesToRead) {
            readRateLimit.wait();
            timerInst(read);
            auto bytes_read = throwcall::badval(read(fd, b.bufferAt(b.size()), blockSize),
                                                -1, "read failed on ", path);
            readRateLimit.update(bytes_read);
            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;
            }
            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;
    }
 
 
    void posixFile::readerPosixFile::readBlockP(block& b, size_t bytesToRead, off_t offset) {
        if (!regions.empty()) {
            throw std::logic_error("parallel read attempted on sparse file");
        }
        b.clear(offset);
        while (b.size() + blockSize <= b.max_size()) {
            timerInst(pread);
            auto bytes_read = throwcall::badval(pread(fd, b.bufferAt(b.size()), blockSize, offset + b.size()),
                                                -1, "read failed on ", path);
            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 posixFile::readerPosixFile::checkUnchangedness() {
        struct stat readFinalStatBuf;
        {
            timerInst(fstat);
            throwcall::good0(fstat(fd, &readFinalStatBuf), "can't stat path file ", path);
        }
        genericStat readFinalStat(readFinalStatBuf, std::chrono::nanoseconds(1));
        if (readFinalStat.size != readInitialStat.size) {
            throw delayAdvisingError("file size has changed (" +
                                     std::to_string(readInitialStat.size) +
                                     " -> " +
                                     std::to_string(readFinalStat.size) +
                                     ") during reading on " + path);
        }
 
        if (!readFinalStat.isSameMtimeAs(readInitialStat)) {
            throw delayAdvisingError("file " + path + " was modified (" +
                                     std::to_string(std::chrono::duration_cast<std::chrono::duration<double>>(readFinalStat.getMtime() - readInitialStat.getMtime()).count()) +
                                     "s different mtime) during reading");
        }
    }
    posixFile::PosixDirectory::PosixDirectory(const std::string& aPath): Directory(aPath) {
        timerInst(opendir);
        dir = throwcall::badval(opendir(path.c_str()), nullptr, "can't open directory ", path);
    }
    posixFile::PosixDirectory::~PosixDirectory() noexcept(false) {
        if (isUnwinding()) {
            if (closedir(dir) != 0) {
                errMsg::emit(errMsg::level::debug, errMsg::location(),
                             "", "close directory during unwind ",
                             std::system_category().default_error_condition(errno).message());
            }
        } else {
            timerInst(closedir);
            throwcall::good0(closedir(dir), "can't close directory ", path);
        }
    }
    std::unique_ptr<base::Directory::Entry> posixFile::PosixDirectory::getNextEntry(bool ignoreMissing) {
        static timer::anchor a("readdir");
        timer::instanceUnscoped i(a);
        while (auto entry = readdir(dir)) {
            i.stop();
            if (entry->d_name[entry->d_name[0] != '.' ? 0 : entry->d_name[1] != '.' ? 1 : 2] == '\0') {
                continue; // skip . .. and empty strings
                i.restart();
            }
            struct stat statbuf;
            {
                timerInst(fstatat);
                auto result = fstatat(dirfd(dir), entry->d_name, &statbuf, AT_SYMLINK_NOFOLLOW);
                if (result != 0 && errno == ENOENT && ignoreMissing) {
                    continue;
                }
                throwcall::good0(result, "can't stat ", entry->d_name);
            }
            auto genStat = std::unique_ptr<const genericStat>(new genericStat(statbuf, std::chrono::nanoseconds(1)));
            return std::unique_ptr<Entry>(new Entry(entry->d_name, genStat));
        }
        return nullptr;
    }
    std::unique_ptr<base::Directory> posixFile::getDirectory(const std::string& path) {
        return std::unique_ptr<Directory>(new PosixDirectory(path));
    }
 
} //end namespace inputHandler