inputHandlerDaosFs.cpp

Go to the documentation of this file.

#include "inputHandlerDaosFs.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(daosFs::factory) daosFs::factory("daosFs");
 
    static daosFsCommon::daosOptions inputConfig("daosFsIn");
 
    daosFs::daosFs(): daosFsCommon(inputConfig, false){};
    
    std::unique_ptr<base::reader> daosFs::newReader(const std::string& aPath,
            copyRequest::stateType &state,
            const genericStat& inititalStat) {
        return std::unique_ptr<base::reader>(new readerDaosFs(aPath,
                                             state,
                                                              inititalStat,
                                                              *this));
    }
 
 
    bool daosFs::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;
    }
 
 
 
    daosFs::readerDaosFs::readerDaosFs(const std::string& aPath,
            copyRequest::stateType &state,
            const genericStat&inititalStat,
                         daosFsCommon& aHandler):
        daosFsIoCommon(aPath, aHandler),
        reader(inititalStat) {
        int retval;
        {
            timerInst(open);
            retval = dfs_open(handler.dfs, handler.getDirObj(path), path.c_str(), 0444, O_RDONLY,
                              0, readInitialStat.blksize, nullptr, &obj);
        }
        if (retval) {
            if (errno == ENOENT
                    || (errno == ENOTDIR && !S_ISDIR(readInitialStat.mode))) {
                state |= copyRequest::stateBitType::vanished;
            }
        }
        throwcall::good0(retval, "can't open ", path, " for reading");
        blockSize = readInitialStat.blksize;
    }
 
    daosFs::readerDaosFs::~readerDaosFs() {
        if (obj) {
            if (isUnwinding()) {
                timerInst(release);
                if (dfs_release(obj) != 0) {
                    errMsg::emit(errMsg::level::debug, errMsg::location(),
                                 path, "release during unwind ",
                                 std::system_category().default_error_condition(errno).message());
                }
                return;
            }
            {
                timerInst(release);
                throwcall::good0(dfs_release(obj), "can't release ", path, " after reading");
            }
        }
    }
 
 
 
 
    bool daosFs::readerDaosFs::parallelizable() const {
        return true;
    }
    bool daosFs::readerDaosFs::readBlock(block& b) {
        b.clear(totalBytesRead);
        bool lastblock = false;
 
        auto bytesToRead = b.max_size();
 
        while (b.size() + blockSize <= bytesToRead) {
            readRateLimit.wait();
            d_iov_t iov;
            iov.iov_buf =  b.bufferAt(b.size());
            iov.iov_buf_len = blockSize;
            d_sg_list_t sg_list;
            sg_list.sg_nr=1;
            sg_list.sg_iovs = &iov;
            timerInst(read);
            daos_size_t bytes_read;
            throwcall::good0(dfs_read(handler.dfs, obj, &sg_list, totalBytesRead, &bytes_read, nullptr),
                             "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 daosFs::readerDaosFs::readBlockP(block& b, size_t bytesToRead, off_t offset) {
        b.clear(offset);
        while (b.size() + blockSize <= b.max_size()) {
            timerInst(pread);
            d_iov_t iov;
            iov.iov_buf =  b.bufferAt(b.size());
            iov.iov_buf_len = blockSize;
            d_sg_list_t sg_list;
            sg_list.sg_nr=1;
            sg_list.sg_iovs = &iov;
            timerInst(read);
            daos_size_t bytes_read;
            throwcall::good0(dfs_read(handler.dfs, obj, &sg_list, offset + b.size() , &bytes_read, nullptr),
                             "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 daosFs::readerDaosFs::checkUnchangedness() {
        struct stat readFinalStatBuf;
        {
            timerInst(fstat);
            throwcall::good0(dfs_ostat(handler.dfs, obj, &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");
        }
    }
    daosFs::DaosFsDirectory::DaosFsDirectory(daosFs& aHandler, const std::string& aPath): Directory(aPath), handler(aHandler) {
        timerInst(opendir);
        throwcall::good0(dfs_open(handler.dfs, handler.getDirObj(path), path.c_str(), 0666, O_RDONLY, 0, 0, nullptr, &dir),
                         "can't open direcory ", path);
    }
    daosFs::DaosFsDirectory::~DaosFsDirectory() noexcept(false) {
        if (isUnwinding()) {
            if (dfs_release(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(dfs_release(dir), "can't close directory ", path);
        }
    }
    std::unique_ptr<base::Directory::Entry> daosFs::DaosFsDirectory::getNextEntry(bool ignoreMissing) {
        while (true) {
            struct dirent entry;
            uint32_t n=1;
            throwcall::good0(dfs_readdir(handler.dfs, dir, &anchor, &n, &entry),"can't read dir enry");
            if (n==1) {
                if (entry.d_name[entry.d_name[0] != '.' ? 0 : entry.d_name[1] != '.' ? 1 : 2] == '\0') {
                    continue; // skip . .. and empty strings
                }
                struct stat statbuf;
                {
                    timerInst(fstatat);
                    auto result = dfs_stat(handler.dfs, dir, entry.d_name, &statbuf);
                    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));
            } else {
                break;
            }
        }
        return nullptr;
    }
    std::unique_ptr<base::Directory> daosFs::getDirectory(const std::string& path) {
        return std::unique_ptr<Directory>(new DaosFsDirectory(*this, path));
    }
 
} //end namespace inputHandler