outputHandlerGpfs.cpp

Go to the documentation of this file.

#include "outputHandlerGpfs.h"
#include "block.h"
#include "copyRequestTypes.h"
#include "gpfsCommon.h"
#include "ewmscp.h"
#include "timer.h"
 
#include "throwcall.h"
#include <fcntl.h>
#include <libgen.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <unistd.h>
#include <vector>
#include <gpfs.h>
 
namespace outputHandler {
    decltype(Gpfs::factory) Gpfs::factory("Gpfs");
 
    std::unique_ptr<base::writer> Gpfs::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 writerGpfs(path, mightAppend,
                                             sourceSize, readBlockSize,
                                             state, noWrite,
                                             std::move(attrData),
                                             std::move(aclData)));
    }
    std::unique_ptr<base::writer> Gpfs::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 writerGpfs(path,
                                             sourceSize,
                                             noWrite,
                                             std::move(attrData),
                                                            std::move(aclData),
                                             *this));
    }
    decltype(Gpfs::writerGpfs::usePreallocate) Gpfs::writerGpfs::usePreallocate('\0', "useGpfsPreallocate", "use gpfs_prealloc for output files");
    static options::single<bool> useRange('\0', "gpfsOutUseRange", "use range hint in gpfs output", false);
    static options::single<bool> freeRange('\0', "gpfsOutFreeRange", "use range free in gpfs output", false);
    static options::single<bool> doClearCache('\0', "gpfsOutClearCache", "clear cache in gpfs output", false);
 
    void Gpfs::writerGpfs::doSpecialFdHandling(size_t sourceSize) {
        if (fd != -1) {
            if (usePreallocate) {
                //calculate size which uses entire blocks
                auto truncatedSize = (sourceSize / getBlockSize()) * getBlockSize();
                if (truncatedSize > 0) {
                    timerInst(gpfs_prealloc);
                    throwcall::good0(gpfs_prealloc(fd, 0, truncatedSize),
                                     "can't gpfs_prealloc ", truncatedSize, " bytes for ", path);
                }
            }
            if (useRange) {
                gpfs_fcntl_handler fcntl;
                auto& range = fcntl.buffer.add<gpfsAccessRange_t>(GPFS_ACCESS_RANGE);
                range.start = 0;
                range.length = sourceSize;
                range.isWrite = 1;
                timerInst(gpfsAccessRange);
                fcntl.call(fd, path);
            }
        }
    }
 
    Gpfs::writerGpfs::writerGpfs(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):
        posixFileIoCommon(aPath),
        gpfsIoCommon(aPath),
        writerPosixFile(aPath, mightAppend, sourceSize, readBlockSize, state, noWrite,
                        std::move(aAttrData),
                        std::move(aAclData)) {
        doSpecialFdHandling(sourceSize);
    };
 
    Gpfs::writerGpfs::writerGpfs(std::string& aPath,
                                 size_t sourceSize,
                                 bool noWrite,
                                 std::unique_ptr<ioHandle::attrDataType> aAttrData,
                                 std::unique_ptr<acl::list> aAclData,
                                 base& handler):
        posixFileIoCommon(aPath),
        gpfsIoCommon(aPath),
        writerPosixFile(aPath, noWrite, std::move(aAttrData), std::move(aAclData),handler) {
        doSpecialFdHandling(sourceSize);
    };
 
    Gpfs::writerGpfs::~writerGpfs() noexcept(false) {
        if (isUnwinding()) {
            return;
        }
        if (fd != -1) {
            if (aclData) {
                setAclData();
            }
            if (doClearCache) {
                gpfs_fcntl_handler fcntl;
                fcntl.buffer.add<gpfsClearFileCache_t>(GPFS_CLEAR_FILE_CACHE);
                timerInst(gpfsClearFileCache);
                fcntl.call(fd, path);
            }
        }
    }
 
    void Gpfs::writerGpfs::writeBlock(const block& b) {
        writerPosixFile::writeBlock(b);
        leastRecentlyAccessedBlock = b.offset();
        if (freeRange) {
            releaseUsedBlocks(b.max_size());
        }
    }
 
    void Gpfs::writerGpfs::writeBlockP(const block& b) {
        writerPosixFile::writeBlockP(b);
        if (freeRange) {
            handleParallelUsedBlocks(b.max_size(), b.offset());
        }
    }
 
    void Gpfs::writerGpfs::setAclData() {
        gpfs_acl_handler aclHandler(aclData->entries.size());
 
        auto gpfs_ace = aclHandler.begin();
        for (const auto& rich_ace : aclData->entries) {
            gpfs_ace->aceType = rich_ace.isAllowedType() ? ACE4_TYPE_ALLOW : ACE4_TYPE_DENY;
            gpfs_ace->aceFlags =    rich_ace.getFlagBits() & (ACE4_FLAG_FILE_INHERIT |
                                    ACE4_FLAG_DIR_INHERIT |
                                    ACE4_FLAG_NO_PROPAGATE |
                                    ACE4_FLAG_INHERIT_ONLY |
                                    ACE4_FLAG_GROUP_ID |
                                    ACE4_FLAG_INHERITED);
            gpfs_ace->aceIFlags = 0;
            gpfs_ace->aceMask = rich_ace.getMaskBits();
            if (rich_ace.isFlagSet(acl::list::entryType::flagsType::special_who)) {
                gpfs_ace->aceIFlags = ACE4_IFLAG_SPECIAL_ID;
                switch (rich_ace.e_id) {
                    case acl::list::entryType::specialIdType::owner_special_id:
                        gpfs_ace->aceWho = ACE4_SPECIAL_OWNER;
                        break;
                    case acl::list::entryType::specialIdType::group_special_id:
                        gpfs_ace->aceWho = ACE4_SPECIAL_GROUP;
                        break;
                    case acl::list::entryType::specialIdType::everyone_special_id:
                        gpfs_ace->aceWho = ACE4_SPECIAL_EVERYONE;
                        break;
                    default:
                        throw std::runtime_error("illegal special id");
                }
            } else {
                gpfs_ace->aceWho = rich_ace.e_id;
            }
            gpfs_ace++;
        }
        aclHandler.set(fd, path);
    }
 
    base::renameRetvalType Gpfs::rename(const std::string& fromPath,
                                        const std::unique_ptr<const genericStat>& readInitialStat,
                                        const std::string& toPath,
                                        copyRequest::stateType& state) {
        auto retval = posixFile::rename(fromPath, readInitialStat, toPath, state);
        if (retval == base::renameRetvalType::ok && state & copyRequest::stateBitType::done) {
            auto lastSlash = fromPath.find_last_of('/');
            auto fromDir = lastSlash == std::string::npos ? "." : fromPath.substr(0, lastSlash);
            lastSlash =  toPath.find_last_of('/');
            auto toDir = lastSlash == std::string::npos ? "." : toPath.substr(0, lastSlash);
            if (fromDir != toDir) { // we move from one dir to another with possibly different acls
                char aclBufferSpace[0x10000];
                auto aclBuffer = reinterpret_cast<gpfs_opaque_acl_t*>(aclBufferSpace);
                aclBuffer->acl_buffer_len = sizeof(aclBufferSpace);
                aclBuffer->acl_version = 0;
                aclBuffer->acl_type = GPFS_ACL_TYPE_ACCESS;
                static const std::string pattern("/.aclgetXXXXXX");
                std::vector<char> tmpName;
                tmpName.reserve(toDir.size() + pattern.size());
                tmpName.insert(tmpName.end(), toDir.cbegin(), toDir.cend());
                tmpName.insert(tmpName.end(), pattern.cbegin(), pattern.cend());
                tmpName.push_back('\0');
                if (readInitialStat->isDir()) {
                    timerInst(mkdtemp);
                    throwcall::badval(mkdtemp(tmpName.data()), nullptr, "can't create tmp dir '", tmpName.data(), "'");
                } else {
                    static timer::anchor a("mkstemp");
                    timer::instanceUnscoped i(a);
                    auto fd = throwcall::badval(mkstemp(tmpName.data()), -1, "can't create tmp file '", tmpName.data(), "'");
                    i.stop();
                    timerInst(close);
                    throwcall::good0(close(fd), "can't close ", tmpName.data());
                }
                {
                    timerInst(gpfs_getacl);
                    throwcall::good0(gpfs_getacl(tmpName.data(), 0, aclBuffer), "can't get acl for '", tmpName.data(), "'");
                }
                if (readInitialStat->isDir()) {
                    timerInst(rmdir);
                    throwcall::good0(rmdir(tmpName.data()), "can't remove '", tmpName.data(), "'");
                } else {
                    timerInst(unlink);
                    throwcall::good0(unlink(tmpName.data()), "can't remove '", tmpName.data(), "'");
                }
                if (aclBuffer->acl_buffer_len > 0) {
                    timerInst(gpfs_putacl);
                    throwcall::good0(gpfs_putacl(toPath.c_str(), 0, aclBuffer), "can't set acl for '", toPath, "'");
                }
            }
        }
        return retval;
    }
}; // end namespace outputHandler