outputHandlerDaosFs.cpp

Go to the documentation of this file.

#include <OptionsChrono.h>
#include "outputHandlerDaosFs.h"
#include "block.h"
#include "copyRequestTypes.h"
#include "ewmscp.h"
#include "timer.h"
#include "throwcall.h"
#include "errMsgQueue.h"
#include <fcntl.h>
#include <libgen.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/statvfs.h>
#include <unistd.h>
#include <vector>
 
 
namespace outputHandler {
    decltype(daosFs::factory) daosFs::factory("daosFs");
    std::unique_ptr<base::writer> daosFs::newWriter(const std::string& path,
            bool mightAppend,
            size_t sourceSize,
            size_t readBlockSize,
            copyRequest::stateType &state,
            bool noWrite,
            std::unique_ptr<ioHandle::attrDataType> attrData,
            std::unique_ptr<acl::list> aclData) {
        return std::unique_ptr<base::writer>(new writerDaosFs(path, mightAppend,
                                             sourceSize, readBlockSize,
                                             state, noWrite,
                                             std::move(attrData),
                                                              std::move(aclData),
                                                              *this));
    }
 
    std::unique_ptr<base::writer> daosFs::newTmpWriter(std::string& path,
            size_t /*sourceSize*/,
            bool noWrite,
            std::unique_ptr<ioHandle::attrDataType> attrData,
            std::unique_ptr<acl::list> aclData) {
        return std::unique_ptr<base::writer>(new writerDaosFs(path,
                                                                 noWrite,
                                                                 std::move(attrData),
                                                                 std::move(aclData),
                                                                 *this));
    }
 
 
    static options::single<timer::clock_type::duration> slownessThreshold('\0',"daosFsSlownessThreshold",
        "duration after which file operations are considered slow",
        std::chrono::milliseconds(10));
 
    void daosFs::ensureParentDirs(const std::string& path,
                                     const std::string& srcPath,
                                     inputHandler::base* InputHandler) {
        std::vector<std::remove_reference<decltype(path)>::type::value_type> disposable_buffer(path.c_str(), path.c_str() + path.size() + 1);
        std::string dir(dirname(disposable_buffer.data()));
        struct stat statbuf;
 
        if (stat(dir.c_str(), &statbuf)) {
            if (errno == ENOENT || errno == ENOTDIR) {
                ensureParentDirs(dir, srcPath, InputHandler);
                static timer::anchor A("mkdir");
                timer::instanceUnscoped I(A);
                auto result = mkdir(dir.c_str(), 0777u);
                I.stop(slownessThreshold, dir);
 
                if (result != 0 && errno == EEXIST) {
                    return; // we lost a race with another copy process...
                }
 
                throwcall::good0(result, "can't create directory ", dir);
 
                if (gid != -1 || uid != -1) {
                    timerInst(chown);
                    throwcall::good0(chown(dir.c_str(), uid, gid), "can't set owner/group (", uid, ",", gid, ") on ", dir);
                }
            }
        } else if (! S_ISDIR(statbuf.st_mode)) {
            errMsg::emit(errMsg::level::info, errMsg::location(),
                         dir, "ensure parents",
                         "is not a directory (st_mode is ", statbuf.st_mode, ") but should be");
        }
        if (requiredFsid != 0) {
            struct statvfs fsstat;
            timerInst(statvfs);
            throwcall::good0(statvfs(dir.c_str(), &fsstat), "can't statvfs ", dir);
            if ((fsstat.f_fsid & fsidMask) != requiredFsid) {
                throw fatalException("wrong fs id 0x", std::hex, fsstat.f_fsid, " instead of ", requiredFsid);
            }
        }
    }
 
 
    void daosFs::remove(const std::string& path, copyRequest::stateType& state) {
        struct stat dsttat;
        {
            timerInst(lstat);
            auto retval = lstat(path.c_str(), &dsttat);
 
            if (retval && errno == ENOENT) {
                state |= copyRequest::stateBitType::vanished;
                return;
            }
        }
        if (S_ISDIR(dsttat.st_mode)) {
            timerInst(rmdir);
            auto retval = rmdir(path.c_str());
            if (retval) {
                switch (errno) {
                    case ENOENT:
                        state |= copyRequest::stateBitType::vanished;
                        return;
                    case ENOTEMPTY:
                    case EEXIST: // POSIX.1 allows this also
                        errMsg::emit(errMsg::level::info, errMsg::location(),
                                     path, "remove", "directory not empty but should be removed");
                        state |= copyRequest::stateBitType::failed;
                        return;
                }
                throwcall::good0(retval, "can't remove directory ", path);
            } else {
                state |= copyRequest::stateBitType::done;
                return;
            }
        } else {
            timerInst(unlink);
            auto retval = unlink(path.c_str());
 
            if (retval && errno == ENOENT) {
                state |= copyRequest::stateBitType::vanished;
            } else {
                throwcall::good0(retval, "can't unlink ", path);
            }
            state |= copyRequest::stateBitType::done;
        }
    }
 
    bool daosFs::renameSimple(const std::string& fromPath,
                                 const std::string& toPath) {
        timerInstTO(rename,slownessThreshold,fromPath);
        auto retval = std::rename(fromPath.c_str(), toPath.c_str());
        if (retval && (errno == ENOENT || errno == ENOTDIR)) {
            return false;
        }
        throwcall::good0(retval, "can't rename '", fromPath, "' to '", toPath, "'");
        return true;
    }
 
 
    base::renameRetvalType daosFs::rename(const std::string& fromPath,
            const std::unique_ptr<const genericStat>& readInitialStat,
            const std::string& toPath,
            copyRequest::stateType& state) {
        struct stat fromPathStatBuf;
        static timer::anchor A("stat");
        timer::instanceUnscoped I(A);
        auto statRetVal = stat(fromPath.c_str(), &fromPathStatBuf);
        I.stop();
        if (statRetVal && (errno == ENOENT || errno == ENOTDIR)) {
            if (readInitialStat && readInitialStat->isDir()) { // no need to act, dirs are created on the fly
                state |= copyRequest::stateBitType::ignore;
                return base::renameRetvalType::ok;
            }
            return base::renameRetvalType::fileVanished;
        } else if (statRetVal) { // curious error
            throwcall::good0(statRetVal, "can't stat '", fromPath, "'");
            return base::renameRetvalType::cantHappen; // never reached
        }
        genericStat fromPathStat(fromPathStatBuf, std::chrono::nanoseconds(1));
        if (!readInitialStat || (! readInitialStat->isDir() &&
                                 (fromPathStat.size != readInitialStat->size
                                  || !fromPathStat.isSameMtimeAs(*readInitialStat)))) {
            // the copy on fromPath is not up to date, create a new copy
            if (readInitialStat) {
                std::string initialTime;
                std::string finalTime;
                readInitialStat->getMtime(initialTime);
                fromPathStat.getMtime(finalTime);
                errMsg::emit(errMsg::level::info, errMsg::location(),
                             fromPath, "move"
                             "file (", toPath, ") changed unexpectedly ("
                             , initialTime, " -> ", finalTime, ", "
                             , readInitialStat->size, " -> ", fromPathStat.size
                             , ", doing fresh copy");
            } else {
                errMsg::emit(errMsg::level::info, errMsg::location(),
                             fromPath, "move",
                             "file (", toPath, ") has no initial stat, doing fresh copy");
            }
            return base::renameRetvalType::fileChanged;
        } else { // try to move the existing copy
            timerInst(rename);
            auto retval = std::rename(fromPath.c_str(), toPath.c_str());
 
            if (retval && (errno == ENOENT || errno == ENOTDIR)) {
                if (readInitialStat->isDir()) { // no need to act, dirs are created on the fly
                    state |= copyRequest::stateBitType::ignore;
                    return base::renameRetvalType::ok;
                }
                errMsg::emit(errMsg::level::info, errMsg::location(),
                             fromPath, "move",
                             "vanished unexpectedly, doing fresh copy");
                // file was probably not copied yet, try to copy it from the move target
                return base::renameRetvalType::fileVanished;
            } else {
                throwcall::good0(retval, "can't rename '", fromPath, "' to '", toPath, "'");
                state |= copyRequest::stateBitType::done;
                return base::renameRetvalType::ok;
            }
        }
    }
 
    void daosFs::createSymlink(const std::vector<char>& target,
                                  const std::string& path,
                                  uid_t uid, gid_t gid) {
        timerInst(symlink);
        auto retval = symlink(target.data(), path.c_str());
        if (retval != 0 && errno != EEXIST) {
            throwcall::good0(retval, "can't create link at ", path);
        }
        if (static_cast<std::make_signed<decltype(gid)>::type>(gid) != -1
                || static_cast<std::make_signed<decltype(uid)>::type>(uid) != -1) {
            throwcall::good0(lchown(path.c_str(), uid, gid), "can't set owner/group (", uid, ",", gid, ") on ", path);
        }
    }
 
    daosFs::writerDaosFs::writerDaosFs(const std::string& aPath,
            bool mightAppend,
            size_t sourceSize,
            size_t readBlockSize,
            copyRequest::stateType &state,
            bool noWrite,
                                       std::unique_ptr<ioHandle::attrDataType> aAttrData,
                                       std::unique_ptr<acl::list> aAclData,
                                       daosFs& aHandler):
        daosFsIoCommon(aPath, aHandler),
        attrData(std::move(aAttrData)),
        aclData(std::move(aAclData)) {
        bool haveWriteStat = false;
        if (noWrite) {
            obj = nullptr;
            blockSize = 4096;
        } else {
            auto openMode = O_CREAT | O_TRUNC | O_WRONLY;
            state.clear(copyRequest::stateBitType::append); // clear any old append bits, may be left over from former attempts
            timerInstTO(open,slownessThreshold,path);
            fd = throwcall::badval(open(path.c_str(), openMode, S_IWUSR),
                                   -1, "can't open ", path, " for writing");
            if (!haveWriteStat) {
                throwcall::good0(fstat(fd, &writeInitialStat), "can't stat destination file ", path);
            }
            blockSize = writeInitialStat.st_blksize;
            throwcall::good0(posix_fadvise(fd, 0, 0, POSIX_FADV_SEQUENTIAL),
                             "can't advise ", path, " as sequential");
            throwcall::good0(posix_fadvise(fd, 0, 0, POSIX_FADV_NOREUSE),
                             "can't advise ", path, " as use only once");
        }
    };
    daosFs::writerDaosFs::writerDaosFs(std::string& aPath,
            bool noWrite,
            std::unique_ptr<ioHandle::attrDataType> aAttrData,
                                                std::unique_ptr<acl::list> aAclData,
                                                base& handler):
        daosFsIoCommon(aPath),
        attrData(std::move(aAttrData)),
        aclData(std::move(aAclData)) {
        if (noWrite) {
            fd = -1;
            blockSize = 4096;
        } else {
            static const std::string pattern("XXXXXX.tmp");
            std::vector<char> tmpName;
            tmpName.reserve(path.size() + pattern.size() + 1);
            tmpName.insert(tmpName.end(), path.cbegin(), path.cend());
            handler.shortenNameToMax(path,tmpName,pattern);
            tmpName.insert(tmpName.end(), pattern.cbegin(), pattern.cend());
            tmpName.push_back('\0');
            timerInstTO(mkstemps,slownessThreshold,tmpName);
            fd = throwcall::badval(mkstemps(tmpName.data(), 4),
                                   -1, "can't open ", tmpName.data(), " for writing");
            aPath.replace(0, path.size(), tmpName.data(), tmpName.size());
            throwcall::good0(fstat(fd, &writeInitialStat), "can't stat destination file ", path);
            blockSize = writeInitialStat.st_blksize;
            throwcall::good0(posix_fadvise(fd, 0, 0, POSIX_FADV_SEQUENTIAL),
                             "can't advise ", path, " as sequential");
            throwcall::good0(posix_fadvise(fd, 0, 0, POSIX_FADV_NOREUSE),
                             "can't advise ", path, " as use only once");
        }
    };
    void daosFs::writerDaosFs::closeAndRemoveBadCopy() {
        if (close(fd) != 0) {
            errMsg::emit(errMsg::level::debug, errMsg::location(),
                         path, "close during unwind ",
                         std::system_category().default_error_condition(errno).message());
        }
        errMsg::emit(errMsg::level::debug, errMsg::location(),
                     path, "unlink failed copy", "due to exception");
        auto retval = unlink(path.c_str());
        if (retval && errno != ENOENT) {
            errMsg::emit(errMsg::level::err, errMsg::location(),
                         path, "can't remove bad copy ",
                         std::system_category().default_error_condition(errno).message());
        }
    }
 
    daosFs::writerDaosFs::~writerDaosFs() noexcept(false) {
        if (fd != -1) {
            if (isUnwinding()) {
                closeAndRemoveBadCopy();
                return;
            } else {
                if (attrData) {
                    try {
                        attrData->set(this);
                    } catch (const std::exception& e) {
                        errMsg::emit(errMsg::level::debug, errMsg::location(),
                                     path, "set attr at close ",
                                     e.what());
                        closeAndRemoveBadCopy();
                        throw;
                    }
                }
                timerInstTO(close,slownessThreshold,path);
                throwcall::good0(close(fd), "can't close ", path);
            }
        }
    };
 
    void daosFs::writerDaosFs::seek(size_t position) {
        timerInst(lseek);
        throwcall::badval(lseek(fd, position, SEEK_SET),
                          -1, "can't seek ", path, " to ", position);
    }
    bool daosFs::writerDaosFs::parallelizable() const {
        return writeInitialStat.st_size == 0;
    }
    size_t daosFs::writerDaosFs::getSize() const {
        return writeInitialStat.st_size;
    }
 
    void daosFs::writerDaosFs::writeBlock(const block& b) {
        if (b.isHole()) { // just write a hole...
            static timer::anchor a("lseek");
            timer::instanceUnscoped i(a);
            auto currentPosition = throwcall::badval(lseek(fd, 0, SEEK_CUR), -1, "lseek to determine position on", path);
            i.stop();
            auto holeEnd  = currentPosition + b.size();
            {
                timerInst(ftruncate);
                throwcall::good0(ftruncate(fd, holeEnd),
                                 "fruncate ", path, " to ", holeEnd);
            }
            timerInst(lseek);
            throwcall::badval(lseek(fd, holeEnd, SEEK_SET), -1,
                              "can't seek to hole end (", holeEnd, ") in ", path);
            return;
        }
 
        size_t bytes_writen_so_far = 0;
 
        while (bytes_writen_so_far < b.size()) {
            auto count = b.size() - bytes_writen_so_far;
 
            if (count > blockSize) {
                count = blockSize;
            }
            writeRateLimit.wait();
            timerInst(write);
            auto bytes_writen = throwcall::badval(write(fd, b.bufferAt(bytes_writen_so_far), count),
                                                  -1, "write failed on ", path);
            writeRateLimit.update(bytes_writen);
            bytes_writen_so_far += bytes_writen;
        }
    }
 
 
    void daosFs::writerDaosFs::writeBlockP(const block& b) {
        size_t bytes_writen_so_far = 0;
 
        while (bytes_writen_so_far < b.size()) {
            auto count = b.size() - bytes_writen_so_far;
 
            if (count > blockSize) {
                count = blockSize;
            }
            timerInst(pwrite);
            auto bytes_writen = throwcall::badval(pwrite(fd, b.bufferAt(bytes_writen_so_far), count, b.offset() + bytes_writen_so_far),
                                                  -1, "write failed on ", path);
            bytes_writen_so_far += bytes_writen;
        }
    }
 
    void daosFs::writerDaosFs::doAttributePreservations(const genericStat& readInitialStat) {
        if (preserve.timestamps) {
            struct timespec times[2];
            readInitialStat.getAtime(times[0]);
            readInitialStat.getMtime(times[1]);
            timerInst(futimens);
            throwcall::good0(futimens(fd, times), "can't set time stamp of ", path);
        }
 
        if (preserve.mode) {
            timerInst(fchmod);
            throwcall::good0(fchmod(fd, readInitialStat.mode), "can't set mode of ", path);
        } else {
            auto oldmask = umask(0);
            timerInst(fchmod);
            throwcall::good0(fchmod(fd, modeBits & ~oldmask), "can't set mode of ", path);
            umask(oldmask);
        }
 
        if (gid != -1 || uid != -1) {
            timerInst(fchown);
            throwcall::good0(fchown(fd, uid, gid), "can't set owner/group (", uid, ",", gid, ") of ", path);
        } else if (preserve.ownership) {
            timerInst(fchown);
            throwcall::good0(fchown(fd, readInitialStat.ownerUid, readInitialStat.ownerGid),
                             "can't set owner/group (", readInitialStat.ownerUid, ",", readInitialStat.ownerGid, ") of ",
                             path);
        }
    }
 
    void daosFs::doAttributePreservations(const std::string& path,
            const genericStat& stat) {
        if (preserve.timestamps) {
            struct timespec times[2];
            times[0].tv_nsec = UTIME_OMIT; // leave atime unchanged
            stat.getMtime(times[1]);
            timerInst(utimensat);
            throwcall::good0(utimensat(0, path.c_str(), times, 0), "can't set mtime on ", path);
        }
        if (preserve.mode) {
            timerInst(chmod);
            throwcall::good0(chmod(path.c_str(), stat.mode), "can't set mode of ", path);
        } else {
            auto oldmask = umask(0);
            timerInst(chmod);
            throwcall::good0(chmod(path.c_str(), modeBits & ~oldmask), "can't set mode of ", path);
            umask(oldmask);
        }
 
        if (gid != -1 || uid != -1) {
            timerInst(chown);
            throwcall::good0(chown(path.c_str(), uid, gid), "can't set owner/group (", uid, ",", gid, ") of ", path);
        } else if (preserve.ownership) {
            timerInst(chown);
            throwcall::good0(chown(path.c_str(), stat.ownerUid, stat.ownerGid),
                             "can't set owner/group (", stat.ownerUid, ",", stat.ownerGid, ") of ",
                             path);
        }
    }
 
    size_t daosFs::getMaxNameLength(const std::string& dirPath) {
        errno = 0;
        timerInst(pathconf);
        return throwcall::badval(pathconf(dirPath.c_str(), _PC_NAME_MAX), -1,"can't get max name lenght on", dirPath);
    }
 
 
    void daosFs::writerDaosFs::sync() {
        timerInst(fsync);
        throwcall::good0(fsync(fd), "can't sync ", path);
    }
 
}; // end namespace outputHandler