copyRequest.cpp

Go to the documentation of this file.

/*
  ewmscp: copy program with extended functionality
    Copyright (C) 2018  Juergen Hannappel
 
    This program is free software: you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation, either version 3 of the License, or
    (at your option) any later version.
 
    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.
 
    You should have received a copy of the GNU General Public License
    along with this program.  If not, see <https://www.gnu.org/licenses/>.
*/
#include "copyRequest.h"
#include "inputHandler.h"
#include "md5sum.h"
#include "scoped.h"
#include "syslogstream.h"
#include "throwcall.h"
#include "escapism.h"
#include "git-rev.h"
#include "errMsgQueue.h"
#include <string.h>
#include "timer.h"
#ifdef WithJsonCpp
#include <json/json.h>
#endif
 
std::ostream& operator<< (std::ostream& stream,
                          const copyRequest::clock_type::time_point& t) {
    stream << std::chrono::duration_cast<std::chrono::duration<double>>(t.time_since_epoch()).count();
    return stream;
}
 
 
 
defineStatic(copyRequest::base::success_source_attrs, '\0', "success-source-attrs",
             "xattrs to set on source file on success");
defineStatic(copyRequest::base::failure_source_attrs, '\0', "failure-source-attrs",
             "xattrs to set on source file on failure");
defineStatic(copyRequest::base::process_source_attrs, '\0', "process-source-attrs",
             "xattrs to set on source file during copy, removed ad end of copy");
defineStatic(copyRequest::base::start_source_attrs, '\0', "start-source-attrs",
             "xattrs to set on source file on start of copy");
defineStatic(copyRequest::base::check_source_attrs, '\0', "check-source-attrs",
             "xattrs to check on source file after copy (or hashing)");
defineStatic(copyRequest::base::success_dest_attrs, '\0', "success-dest-attrs",
             "xattrs to set on destination file on success");
defineStatic(copyRequest::base::maxRetries, '\0', "retries", "max number of retries", 5);
 
defineStatic(copyRequest::base::prohibitive_attrs,
             '\0', "prohibitive-attrs",
             "xattrs that prohibit copying when present on source");
defineStatic(copyRequest::base::append_attrs,
             '\0', "append-attrs",
             "xattrs that trigger append mode when present on source");
 
defineStatic(copyRequest::base::appendableFiles,
             '\0', "appendable-files",
             "regexp for files that may be appended to");
 
defineStatic(copyRequest::base::ignoreExisting,
             '\0', "ignoreExisting",
             "ignore files where destination exists and has same stat values");
defineStatic(copyRequest::base::changeAttrsOnIgnoreExisting,
             '\0', "changeAttrsOnIgnoreExisting",
             "update preserved attrs on files ignored by ignoreExisting");
 
defineStatic(copyRequest::base::linkBaseMap, '\0', "linkBaseMap",
             "base dir that  gets removed from absolute links");
defineStatic(copyRequest::base::sparse, '\0', "sparse",
             "control creation of sparse files.", "auto", {"auto", "never", "always"});
 
defineStatic(copyRequest::base::backupModeName,
[](options::single<std::string>& me) {
    backupMode = getBackupModeNameMap().at(me);
},
'\0', "backupMode", "backup mode", "none", [] {std::vector<std::string> v; for (const auto& i : getBackupModeNameMap()) {
v.push_back(i.first);
}
return v;
                                              }());
 
defineStatic(copyRequest::base::backupMode, backupModeType::none);
 
defineStatic(copyRequest::base::backupSuffix, '\0', "backupSuffix", "suffix to append to backup files");
defineStatic(copyRequest::base::syncWrittenFiles, '\0', "sync",
             "sync new files after write", false);
defineStatic(copyRequest::base::setAttributesAfterClose, '\0', "setAttributesAfterClose",
             "set attributes after close() (needed for dCache)");
 
defineStatic(copyRequest::base::perFileThreads, '\0', "perFileThreads",
             "extra write/checksum thread per file", false);
defineStatic(copyRequest::base::nSumThreads, '\0', "nSumThreads",
             "number of threads for parallelizable checksums per file",
             3);
defineStatic(copyRequest::base::nReadThreads, '\0', "nReadThreads",
             "number of threads for parallelizable reads per file",
             3);
defineStatic(copyRequest::base::forceParallelRead, '\0', "forceParallelRead",
             "do parallel read even on loaded system");
defineStatic(copyRequest::base::nWriteThreads, '\0', "nWriteThreads",
             "number of threads for parallelizable writes per file",
             3);
defineStatic(copyRequest::base::userReadBlockSize, '\0', "readBlockSize",
             "read block size, 0: from fs",
             0);
defineStatic(copyRequest::base::userWriteBlockSize, '\0', "writeBlockSize",
             "write block size, 0: from fs",
             0);
 
 
defineStatic(copyRequest::base::maxMemoryBlockSize, '\0', "maxMemoryBlockSize",
             "max block size of memory blocks",
             16 * 1024 * 1024);
defineStatic(copyRequest::base::minMemoryBlockSize, '\0', "minMemoryBlockSize",
             "min block size of memory blocks",
             1 * 1024 * 1024);
 
defineStatic(copyRequest::base::logFields, '\0', "logField",
             "field in the log lines");
defineStatic(copyRequest::base::sourcePrefixRegex, '\0', "sourcePrefixRegex",
             "regexp to determine log prefix from source path");
defineStatic(copyRequest::base::destinationPrefixRegex, '\0', "destinationPrefixRegex",
             "regexp to determine log prefix from destination path");
 
#ifdef WithMagic
defineStatic(copyRequest::base::doMagic, '\0', "doMagic",
             "determine file type via libmagic");
#endif
 
 
defineStatic(copyRequest::base::tTotalStat, "time to finish");
defineStatic(copyRequest::base::tTotal2Stat, "time modify to finish");
defineStatic(copyRequest::base::tInotifyStat, "time to notify");
defineStatic(copyRequest::base::tPipeStat, "time in pipe");
defineStatic(copyRequest::base::tWaitStat, "time in queue");
defineStatic(copyRequest::base::tCopyStat, "time to copy");
defineStatic(copyRequest::base::tEnqueueStat, "time to enqueue");
defineStatic(copyRequest::base::bytesStat, "file size", "B");
defineStatic(copyRequest::base::speedStat, "copy speed", "MiB/s");
defineStatic(copyRequest::base::tProgramStart, clock_type::now());
 
defineStatic(copyRequest::base::nConcurrentProcesses, 0);
defineStaticNoArg(copyRequest::base::processMultiplicities);
defineStatic(copyRequest::base::maxDelayTime, '\0', "maxDelayTime",
             "max seconds to delay files", 10);
defineStaticNoArg(copyRequest::base::advisoryWaitMap);
defineStaticNoArg(copyRequest::base::advisoryWaitMapMutex);
 
defineStatic(copyRequest::base::statPrinter, copyRequest::base::printStats);
 
 
defineStaticNoArg(copyRequest::base::objectCountMutex);
defineStaticNoArg(copyRequest::base::objectCountCondVar);
defineStatic(copyRequest::base::objectCount, 0);
 
copyRequest::perThreadData::perThreadData() {
    InputHandler = inputHandler::base::newHandler(inputHandlerName);
    OutputHandler = outputHandler::base::newHandler(outputHandlerName);
    #ifdef WithMagic
    magicCookie = throwcall::badval(magic_open(MAGIC_NONE), nullptr, "can't init libmagic");
    if (magic_load(magicCookie, nullptr) != 0) {
        throw std::runtime_error(magic_error(magicCookie));
    }
    #endif
}
 
 
void copyRequest::exceptionList::add(std::exception_ptr e) {
    std::unique_lock<decltype(listMutex)> lock(listMutex);
    list.push_front(e);
}
bool copyRequest::exceptionList::empty() const {
    // for test on emptyness lock is not needed
    return list.empty();
}
size_t copyRequest::exceptionList::size() const {
    std::unique_lock<decltype(listMutex)> lock(listMutex);
    return std::distance(list.cbegin(), list.cend());
}
std::exception_ptr  copyRequest::exceptionList::front() {
    std::unique_lock<decltype(listMutex)> lock(listMutex);
    return list.front();
}
 
 
const std::map<std::string, copyRequest::backupModeType>& copyRequest::base::getBackupModeNameMap() {
    static std::map<std::string, backupModeType> backupModeNameMap = {
        {"none", copyRequest::backupModeType::none},
        {"before", copyRequest::backupModeType::before},
        {"during", copyRequest::backupModeType::during},
        {"after", copyRequest::backupModeType::after},
        {"remove", copyRequest::backupModeType::remove}
    };
    return backupModeNameMap;
}
 
std::map<std::string, void (copyRequest::base::*)(std::string&) const>& copyRequest::base::keywordMap() {
    static std::map<std::string, void (copyRequest::base::*)(std::string&) const> map;
    return map;
}
// /}
#define defKeyword(kw) static copyRequest::base::registerme kw_##kw("%" #kw,&copyRequest::base::kw_##kw); \
    void copyRequest::base::kw_##kw(std::string& value) const
defKeyword(size) {
    if (readInitialStat == nullptr) {
        value = "NA";
        return;
    }
    value = std::to_string(readInitialStat->size);
}
 
defKeyword(mtime) {
    if (readInitialStat == nullptr) {
        value = "NA";
        return;
    }
    readInitialStat->getMtime(value);
}
defKeyword(inow) {
    value = std::to_string(std::chrono::duration_cast<std::chrono::seconds>(std::chrono::system_clock::now().time_since_epoch()).count());
}
 
defKeyword(now) {
    value = std::to_string(std::chrono::duration_cast<std::chrono::duration<double>>(std::chrono::system_clock::now().time_since_epoch()).count());
}
 
defKeyword(version) {
    value = EWMSCP_VERSION;
}
 
defKeyword(commit) {
    value = EWMSCP_COMMIT;
}
 
defKeyword(prefix) {
    value = prefix;
}
 
 
defKeyword(reason) {
    if (errorMessage.empty()) {
        value = "no error message found";
    } else {
        value = errorMessage;
    }
}
 
defKeyword(state) {
    value = "";
    static struct {
        stateBitType bit;
        const char *name;
    } bitNames[] = {{stateBitType::done, "done"},
        {stateBitType::append, "append"},
        {stateBitType::linkToBeMade, "link"},
        {stateBitType::failed, "failed"},
        {stateBitType::ignore, "ignored"},
        {stateBitType::trucated, "trucated"},
        {stateBitType::inWork, "inWork"}
    };
 
    bool needComma = false;
    for (const auto& bitName : bitNames) {
        if (state & bitName.bit) {
            if (needComma) {
                value += ", ";
            }
 
            value += bitName.name;
            needComma = true;
        }
    }
}
 
defKeyword(finishtime) {
    char buffer[64];
    sprintf(buffer, "%.6f", std::chrono::duration_cast<std::chrono::duration<double>>(tWorkDone - tInotify).count());
    value = buffer;
}
 
#ifdef WithMagic
defKeyword(magic) {
    value = fileMagic;
}
#endif
 
bool copyRequest::base::expandAttrValue(std::string& value) {
    auto it = keywordMap().find(value);
    if (it != keywordMap().end()) {
        auto f = it->second;
        (this ->* f)(value);
        return true;
    }
 
    if (value.size() > 1 && value.at(0) == '%') {
        auto keyword = value.substr(1);
        for (const auto& cksum : checkSums) {
            if (cksum->getName() == keyword) {
                value = cksum->getResult();
                return true;
            }
        }
    }
 
    return false; // no change...
}
void copyRequest::base::attrset(ioHandle& writeHandle,
                                const std::map<std::string, std::string>& attrs) {
    for (const auto& attr : attrs) {
        auto name = attr.first;
        auto value = attr.second;
 
        if (!expandAttrValue(value)) {
            if (value == "%count") { // special expansion, only useful here
                unsigned int count = 0;
                auto retval = writeHandle.getXattr(name);
                if (!retval.empty()) {
                    try {
                        count = std::stoul(retval);
                    } catch (...) {
                        errMsg::emit(errMsg::level::debug, errMsg::location(),
                                     source, "xattr count",
                                     "strange value '", retval, "'");
                        count = 0;
                    }
                }
                value = std::to_string(count + 1);
            } else if (value == "%sums") { // special expansion only useful here
                for (const auto& item : checkSums) {
                    std::string fullName(name);
                    fullName += item->getName();
                    writeHandle.setXattr(fullName, item->getResult());
                }
                continue;
            }
        }
        writeHandle.setXattr(name, value);
    }
}
 
 
void copyRequest::base::attrdel(ioHandle& handle, const std::map<std::string, std::string>& attrs) {
    for (const auto& attr : attrs) {
        handle.removeXattr(attr.first);
    }
}
void copyRequest::base::checkAttributes(ioHandle& handle, const std::map<std::string, std::string>& attrs) {
    for (const auto& attr : attrs) {
        auto name = attr.first;
        auto value = attr.second;
        expandAttrValue(value);
        auto fileValue = handle.getXattr(name);
 
        if (fileValue.empty()) {
            state |= stateBitType::attributeMismatch;
            errorMessage += "attr " + name + " missing ";
            continue;
        }
 
        if (value != fileValue) {
            state |= stateBitType::attributeMismatch;
            errorMessage += "attr " + name + " mismatch " + value + " i.o. " + fileValue;
        }
    }
 
    if (state & stateBitType::attributeMismatch) {
        throw std::runtime_error(errorMessage);
    }
}
 
void copyRequest::base::readWorker(inputHandler::base::reader& input,
                                   blockReadRequestQueue& blockRequests,
                                   blockQueue& freeBlocks,
                                   blockQueue& readBlocks,
                                   exceptionList& exceptions) {
    try {
        while (auto request = blockRequests.dequeue()) {
            auto b = freeBlocks.dequeue(readBlocks.size() < 3, memoryBlockSize);
            if (b == nullptr) {
                throw std::runtime_error("got no free block in readWorker");
            }
            input.readBlockP(*b, request->size, request->offset);
            readBlocks.enqueue(b);
        }
    } catch (...) {
        exceptions.add(std::current_exception());
    }
}
 
 
void copyRequest::base::reader(inputHandler::base::reader& input,
                               blockQueue& freeBlocks,
                               blockQueue& readBlocks,
                               bool mayParallelize,
                               exceptionList& exceptions) {
    try {
        unsigned nBlocks = readInitialStat->size / memoryBlockSize;
        if (mayParallelize
                && input.parallelizable()
            && (nConcurrentProcesses < nThreads / nReadThreads + 1 // don't load loaded systems
                || forceParallelRead)
                && nReadThreads > 1
                && nBlocks >= nReadThreads) {
            blockReadRequestQueue readRequests;
            for (size_t o = 0; o < readInitialStat->size; o += memoryBlockSize) {
                auto s = readInitialStat->size - o;
                s = std::min(s, memoryBlockSize);
                readRequests.emplace(s, o);
            }
            readRequests.signalDone();
            std::vector<std::thread> workers;
            {
                timerInst(readThreadSetup);
                workers.resize(nReadThreads);
                for (auto& worker : workers) {
                    worker = std::thread(&copyRequest::base::readWorker, this,
                                         std::ref(input),
                                         std::ref(readRequests),
                                         std::ref(freeBlocks),
                                         std::ref(readBlocks),
                                         std::ref(exceptions));
                }
            }
            for (auto& worker : workers) {
                worker.join();
            }
        } else {
            while (true) {
                auto b = freeBlocks.dequeue(readBlocks.size() < 3, memoryBlockSize);
                if (b == nullptr) {
                    throw std::runtime_error("got no free block in reader");
                }
                auto lastblock = input.readBlock(*b);
 
                if (b->empty()) { // we are done, recycle the block in the free block list
                    freeBlocks.enqueue(b);
                } else {
                    readBlocks.enqueue(b);
                }
 
                if (lastblock) {
                    break;
                }
            }
        }
    } catch (...) {
        exceptions.add(std::current_exception());
    }
    readBlocks.signalDone();
}
 
 
void copyRequest::base::writer(std::unique_ptr<outputHandler::base::writer>& writeHandle,
                               blockQueue& blocksToWrite,
                               blockQueue& writtenBlocks,
                               exceptionList& exceptions) {
    try {
        size_t lastOffset = 0;
        while (auto b = blocksToWrite.dequeue()) {
            if (b->offset() < lastOffset) {
                throw std::logic_error("blocks not in order");
            }
            writeHandle->writeBlock(*b);
            lastOffset = b->offset();
            writtenBlocks.enqueue(b);
        }
    } catch (...) {
        exceptions.add(std::current_exception());
    }
}
void copyRequest::base::writeWorker(std::unique_ptr<outputHandler::base::writer>& writeHandle,
                                    blockQueue& blocksToWrite,
                                    blockQueue& writtenBlocks,
                                    exceptionList& exceptions) {
    try {
        while (auto b = blocksToWrite.dequeue()) {
            writeHandle->writeBlockP(*b);
            writtenBlocks.enqueue(b);
        }
    } catch (...) {
        exceptions.add(std::current_exception());
    }
}
 
 
copyRequest::base::writeActor::writeActor(std::unique_ptr<outputHandler::base::writer>& writeHandle,
        blockQueue& blocksToWrite,
        blockQueue& writtenBlocks,
        bool mayParallelize,
        base* request,
        exceptionList& exceptions):
    outputQueue(writtenBlocks) {
    if (noCopy) { // do nothing
        return;
    }
    unsigned nBlocks = request->readInitialStat->size / request->memoryBlockSize;
    if (mayParallelize
            && request->nConcurrentProcesses < nThreads / nWriteThreads + 1
            && writeHandle->parallelizable()
            && nWriteThreads > 1
            && nBlocks >= nWriteThreads) {
        workers.resize(nWriteThreads);
        for (auto& worker : workers) {
            worker = std::thread(&copyRequest::base::writeWorker, request,
                                 std::ref(writeHandle),
                                 std::ref(blocksToWrite),
                                 std::ref(writtenBlocks),
                                 std::ref(exceptions));
        }
    } else {
        workers.push_back(std::thread(&copyRequest::base::writer, request,
                                      std::ref(writeHandle),
                                      std::ref(blocksToWrite),
                                      std::ref(writtenBlocks),
                                      std::ref(exceptions)));
    }
}
copyRequest::base::writeActor::~writeActor() noexcept (false) {
    if (workers.empty()) {
        return;
    }
    for (auto& worker : workers) {
        worker.join();
    }
    outputQueue.signalDone();
}
bool copyRequest::base::writeActor::joinable() const {
    return !workers.empty();
}
copyRequest::base::hashCalculator::hashCalculator(blockQueue& blocksToHash,
        blockQueue& hashedBlocks,
        base* request,
        bool mayParallelize,
        exceptionList& exceptions):
    outputQueue(hashedBlocks) {
    if (request->checkSums.empty() || (request->state & stateBitType::append)) {
        return;
    }
    unsigned nBlocks = request->readInitialStat->size / request->memoryBlockSize;
    if (request->checkSums.size() == 1
            && request->checkSums.front()->parallelizable()
            && nSumThreads > 1
            && mayParallelize
            && nBlocks >= nSumThreads ) {
        workers.resize(std::min(nSumThreads.fGetValue(), nBlocks));
        parallelSum = dynamic_cast<checksum::parallel*>(request->checkSums.front());
        for (auto& worker : workers) {
            worker = std::thread(&copyRequest::base::hash_worker, request,
                                 parallelSum,
                                 std::ref(blocksToHash),
                                 std::ref(outputQueue),
                                 std::ref(exceptions));
        }
    } else { // use one thread to calculate all checksums
        workers.emplace_back(std::thread(&copyRequest::base::hasher, request,
                                         std::ref(blocksToHash),
                                         std::ref(outputQueue),
                                         std::ref(exceptions)));
    }
}
 
copyRequest::base::hashCalculator::~hashCalculator() noexcept (false) {
    if (workers.empty()) {
        return;
    }
    for (auto& worker : workers) {
        worker.join();
    }
    if (workers.size() > 1) {
        parallelSum->parallelFinish();
    }
    outputQueue.signalDone();
}
bool copyRequest::base::hashCalculator::joinable() const {
    return ! workers.empty();
}
bool copyRequest::base::hashCalculator::parallelized() const {
    return workers.size() != 1; // we also do nothing in parallel
}
 
void copyRequest::base::hash_worker(checksum::parallel* sum,
                                    blockQueue& blocksToHash,
                                    blockQueue& hashedBlocks,
                                    exceptionList& exceptions) {
    try {
        while (auto b = blocksToHash.dequeue()) {
            timerInst(hash);
            if (b->isHole()) {
                sum->update(b->size(), b->offset());
            } else {
                sum->update(b->bufferAt(0), b->size(), b->offset());
            }
            hashedBlocks.enqueue(b);
        }
    } catch (...) {
        exceptions.add(std::current_exception());
    }
}
void copyRequest::base::hasher(blockQueue& blocksToHash,
                               blockQueue& hashedBlocks,
                               exceptionList& exceptions) {
    try {
        while (auto b = blocksToHash.dequeue()) {
            timerInst(hash);
            if (b->isHole()) {
                for (auto sum : checkSums) {
                    sum->update(b->size());
                }
            } else {
                for (auto sum : checkSums) {
                    sum->update(b->bufferAt(0), b->size());
                }
            }
            hashedBlocks.enqueue(b);
        }
 
        for (auto sum : checkSums) {
            sum->finish();
        }
    } catch (...) {
        exceptions.add(std::current_exception());
    }
}
 
#ifdef WithMagic
void copyRequest::base::innerMagic(const block& b) {
    if (b.offset() == 0) { // do magic only for the first block
        timerInst(magic);
        auto result = magic_buffer(threadData->magicCookie, b.bufferAt(0), b.size());
        if (result == nullptr) {
            result = magic_error(threadData->magicCookie);
        }
        if (result) {
            fileMagic = result;
        } else {
            fileMagic = "unknown";
        }
    }
}
void copyRequest::base::magician(blockQueue& inputQueue,
                                 blockQueue& outputQueue,
                                 exceptionList& exceptions) {
    try {
        while (auto b = inputQueue.dequeue()) {
            innerMagic(*b);
            outputQueue.enqueue(b);
        }
    } catch (...) {
        exceptions.add(std::current_exception());
    }
}
copyRequest::base::magicCalculator::magicCalculator(blockQueue& inputQueue,
        blockQueue& aOutputQueue,
        base* request,
        exceptionList& exceptions):
    outputQueue(aOutputQueue) {
    if (doMagic) {
        worker = std::thread(&copyRequest::base::magician, request,
                             std::ref(inputQueue),
                             std::ref(outputQueue),
                             std::ref(exceptions));
    }
}
copyRequest::base::magicCalculator::~magicCalculator() noexcept (false) {
    if (worker.joinable()) {
        worker.join();
        outputQueue.signalDone();
    }
}
bool copyRequest::base::magicCalculator::joinable() const {
    return worker.joinable();
}
#endif
void copyRequest::base::doThreadedCopy(inputHandler::base::reader& input,
                                       std::unique_ptr<outputHandler::base::writer>& writeHandle) {
    blockQueue blocksToWrite;
    blockQueue blocksToHash;
    #ifdef WithMagic
    blockQueue blocksForMagic;
    #endif
    exceptionList exceptions;
 
    scoped::generic < decltype(threadData->freeBlocks) > freeBlockReseter(threadData->freeBlocks,
    [](decltype(threadData->freeBlocks)& freeBlocks) {
        freeBlocks.resetDone();
    });
    {
        #ifdef WithMagic
        magicCalculator magicThread(blocksForMagic, threadData->freeBlocks, this, exceptions);
        auto& lastQueue = magicThread.joinable() ?
                          blocksForMagic : // with magic thread
                          threadData->freeBlocks; // without magic thread
        #else
        auto& lastQueue = threadData->freeBlocks;
        #endif
        hashCalculator hashThread(blocksToHash, lastQueue, this,
                                  (writeHandle->parallelizable() && nWriteThreads > 1) || noCopy, exceptions);
        auto& writerOutputQueue = hashThread.joinable() ?
                                  blocksToHash : // with hash thread
                                  lastQueue; // without hash thread
        writeActor writeThread(writeHandle,
                               blocksToWrite,
                               writerOutputQueue,
                               hashThread.parallelized(),
                               this,
                               exceptions);
        auto& readerOutputQueue = writeThread.joinable() ? // with write thread
                                  blocksToWrite :
                                  ( // without write thread
                                      hashThread.joinable() ? // with hash thred
                                      blocksToHash : // we read and hash
                                      lastQueue // apparently we just read and discard the result
                                  );
 
        reader(input, threadData->freeBlocks, readerOutputQueue,
               hashThread.parallelized() && (writeHandle->parallelizable() || noCopy), exceptions);
    } // hashThread and writeThread go out of scope and are joined
 
    if (!exceptions.empty()) {
        auto n = exceptions.size();
        if (n > 1) {
            errMsg::emit(errMsg::level::warning, errMsg::location(),
                         source, "threaded copy", "we have ", n, " exceptions at once");
        }
        std::rethrow_exception(exceptions.front());
    }
}
 
void copyRequest::base::doUnthreadedCopy(inputHandler::base::reader& input, std::unique_ptr<outputHandler::base::writer>& writeHandle) {
    auto& freeBlocks = threadData->freeBlocks;
    auto b = freeBlocks.dequeue(freeBlocks.empty(), memoryBlockSize);
    if (b == nullptr) {
        throw std::runtime_error("got no block for unthreaded copy");
    }
    for (;;) {
        auto lastblock = input.readBlock(*b);
 
        if (!checkSums.empty() && !(state & stateBitType::append)) {
            timerInst(checksum);
            if (b->isHole()) {
                for (auto sum : checkSums) {
                    sum->update(b->size());
                }
            } else {
                for (auto sum : checkSums) {
                    sum->update(b->bufferAt(0), b->size());
                }
            }
        }
 
        if (!noCopy) {
            writeHandle->writeBlock(*b);
        }
        #ifdef WithMagic
        if (doMagic) {
            innerMagic(*b);
        }
        #endif
        if (lastblock) {
            break;
        }
    }
 
    for (auto sum : checkSums) {
        sum->finish();
    }
    if (b != nullptr) {
        freeBlocks.enqueue(b);
    } else {
        errMsg::emit(errMsg::level::crit, errMsg::location(),
                     source, "unthreaded copy", "block ptr is nullptr after copy");
    }
}
 
void copyRequest::base::removeFileOrDirectory(inputHandler::base* InputHandler,
        outputHandler::base* OutputHandler) {
    if (InputHandler->pathExists(source)) {// source really vanished
        errorMessage = "source " + source + " still exists, not deleting " + destination;
        state |= stateBitType::failed;
    } else {
        OutputHandler->remove(destination, state);
        if (state & stateBitType::vanished) {
            errorMessage = "vanished before processing";
        } else if (state & stateBitType::failed) {
            errorMessage = "directory not empty";
        }
        state |= stateBitType::done;
    }
}
 
 
bool copyRequest::base::makeSymLink(inputHandler::base* InputHandler,
                                    outputHandler::base* OutputHandler) {
    if (readInitialStat == nullptr) {
        throw std::runtime_error("makeSymlink: no readInitialStat for " + source);
    }
    std::vector<char> link(readInitialStat->size + 1);
    auto linkExists = InputHandler->readLinkTarget(source, link);
 
    if (! linkExists) {
        state |= copyRequest::stateBitType::vanished;
        return true;
    }
 
    for (const auto& item : linkBaseMap) {
        if ((link.size() - 1 > item.first.size()) &&
                (item.first.compare(0, item.first.size(), link.data(), item.first.size()) == 0)) { // link is relative to (source) work dir
            link.erase(link.begin(), link.begin() + item.first.size());
            while (link.front() == '/') {
                link.erase(link.begin());
            }
            auto searchStart(item.second.empty() ? mapEntry.second.size() + 1 : item.second.size() + 1);
 
            // source, i.e. the link itself is relative to the workDIr of the inotify watch
            // prepend as many ../ to the link dest as needed to make it the rest start at
            // that directory
            for (auto slashPosition = source.find_first_of('/', searchStart);
                    slashPosition != std::remove_reference<decltype(source)>::type::npos;
                    slashPosition = source.find_first_of('/', slashPosition + 1)) {
                if (slashPosition > 0 &&
                        source[slashPosition - 1] == '.' &&
                        (slashPosition == 1 || source[slashPosition - 2] != '.')) { // skip ./
                    continue;
                }
 
                link.insert(link.begin(), {'.', '.', '/'});
            }
            break; // we consider only the first match
        }
    }
    OutputHandler->createSymlink(link, destination, uid, gid);
 
 
    return true;
}
 
 
bool copyRequest::base::adviseDelay(clock_type::duration dt,
                                    const std::string& suffix) {
    if (suffix.empty() || dt < clock_type::duration::zero()) {
        return false;
    }
    if (std::chrono::duration_cast<std::chrono::duration<double>>(dt).count() > maxDelayTime) {
        dt = std::chrono::duration_cast<decltype(dt)>(std::chrono::duration<double>(maxDelayTime));
    }
    bool delayChanged = false;
    {
        std::unique_lock<decltype(advisoryWaitMapMutex)> lock(advisoryWaitMapMutex);
        auto result = advisoryWaitMap.emplace(suffix, dt);
        if (result.second == false) { // we already have an entry
            auto it = result.first;
            if (it->second < dt) { // apparently we must wait longer
                it->second = dt;
                delayChanged = true;
            }
        } else {
            delayChanged = true;
        }
    }
    if (delayChanged) {
        errMsg::emit(errMsg::level::notice, errMsg::location(),
                     "-- no path --", "delay change",
                     "set wait time for '", suffix, "' files to ",
                     std::fixed, std::chrono::duration_cast<std::chrono::duration<double>>(dt).count(), "s");
    }
    return delayChanged;
}
 
 
void copyRequest::base::adviseDelay(clock_type::duration dt) {
    if (dt == decltype(dt)::zero()) {
        dt = clock_type::now() - tInotify;
    }
 
    if (adviseDelay(dt, getSuffix())) {
        if (filesInWorkIterator == fileInWork::filesInWork.end()) {
            return;
        }
        filesInWorkIterator->second.setWaitTime(dt);
    }
}
copyRequest::clock_type::duration copyRequest::base::getAdvisedDelay() const {
    auto it = advisoryWaitMap.find(getSuffix());
    if (it == advisoryWaitMap.end()) {
        return clock_type::duration::zero();
    }
    return it->second;
}
 
 
 
 
 
 
void copyRequest::base::changeRequestType(stateBitType newType) {
    fileInWork::typeChanger changer(*this, newType);
}
 
 
std::string copyRequest::base::getBackupSuffix() {
    if (backupSuffix.empty()) {
        throw std::logic_error("backup suffix must not be empty");
    }
    std::string bkgSuffix;
    decltype(backupSuffix.find("bla")) index = 0;
    while (index != decltype(backupSuffix)::npos) {
        auto newIndex = backupSuffix.find_first_of('%', index);
        bkgSuffix += backupSuffix.substr(index, newIndex - index);
        if (newIndex != decltype(backupSuffix)::npos) {
            newIndex++;
            if (newIndex < backupSuffix.size()) {
                if (backupSuffix.at(newIndex) == '{') {
                    newIndex++;
                    auto endmarker = backupSuffix.find_first_of('}', newIndex);
                    auto part = "%" + backupSuffix.substr(newIndex, endmarker - newIndex);
                    expandAttrValue(part);
                    bkgSuffix += part;
                    newIndex = endmarker + 1;
                } else {
                    auto part = backupSuffix.substr(newIndex - 1);
                    expandAttrValue(part);
                    bkgSuffix += part;
                    newIndex = decltype(backupSuffix)::npos;
                }
            }
        }
        index = newIndex;
    }
    return bkgSuffix;
}
 
void copyRequest::base::process(perThreadData& aThreadData) {
    processMultiplicities->at(++nConcurrentProcesses)++;
    scoped::generic<decltype(*this)> concurrencyCount(*this,
    [](decltype(*this)& ) {
        --copyRequest::base::nConcurrentProcesses;
    });
 
    tWorkStart = clock_type::now();
    scoped::generic<decltype(tWorkDone)> timer(tWorkDone,
    [](decltype(tWorkDone)& workDone) {
        workDone = clock_type::now();
    });
 
    state |= stateBitType::inWork;
    scoped::generic<decltype(state)> stateReset(state,
    [](decltype(state)& State) {
        State.clear(stateBitType::inWork);
    });
 
    threadData = &aThreadData;
    scoped::generic<decltype(threadData)> unsetThreadData(threadData,
    [](decltype(threadData)& ThreadData) {
        ThreadData = nullptr;
    });
 
    errorMessage.clear(); // forget messages from previous attempts
    retries++;
 
    // do this before check for noAction needed to not be fooled by vanished bit
    if (state & stateBitType::fileToBeRemoved) {
        removeFileOrDirectory(threadData->InputHandler, threadData->OutputHandler);
        return; // no more things to do for stuff to remove
    }
 
    if (state & stateBitType::noActionNeeded) {
        if (state & stateBitType::ignore) {
            errorMessage = "ignored";
        } else if (state & stateBitType::vanished) {
            errorMessage = "vanished before action";
        } else {
            errorMessage = "oops";
        }
        return;
    }
 
    try {
        if (!noCopy) {
            threadData->OutputHandler->ensureParentDirs(destination, source, threadData->InputHandler);
        }
        if (state & stateBitType::linkToBeMade && !noCopy) {
            if (makeSymLink(threadData->InputHandler, threadData->OutputHandler)) {
                if (state & stateBitType::vanished) {
                    errorMessage = "vanished before action";
                } else {
                    state |= stateBitType::done;
                }
                return;
            }
            changeRequestType(stateBitType::fileToBeCopied);
        }
        
        if (state & stateBitType::fileToBeRenamed) {
            auto result = threadData->OutputHandler->rename(moveSource, readInitialStat, destination, state);
            switch (result) {
            case outputHandler::base::renameRetvalType::ok:
                if (state & stateBitType::ignore) {
                    errorMessage = "directory move ignored";
                }
                return;
            case outputHandler::base::renameRetvalType::fileChanged: {
                if (threadData->InputHandler->getStat(origSource) == nullptr) {
                    stateType dummyState;
                    errMsg::emit(errMsg::level::notice, errMsg::location(),
                                 moveSource, "remove obsolete mv source",
                                 "can't find source '", origSource, "'");
                    threadData->OutputHandler->remove(moveSource, dummyState);
                } else {
                    errMsg::emit(errMsg::level::notice, errMsg::location(),
                                 moveSource, "rename",
                                 "littering with obsolete mv source");
                }
            }
#if __GNUC__ >= 7
                __attribute__ ((fallthrough));
#endif
            case outputHandler::base::renameRetvalType::fileVanished:
                // file was probably not copied yet, try to copy it from the move target
                changeRequestType(stateBitType::fileToBeCopied);
                break;
            default:
                throw std::runtime_error("impossible rename retval");
            }
        }
    } catch (const fatalException&) { // always throw, never return
        throw;
    } catch (const std::exception& e) {
        errorMessage = e.what();
        state |= stateBitType::failed;
        if (continueOnError) {
            return;
        } else {
            throw;
        }
    }
    
    try { // try scope for input file setup/cleanup
        if  (readInitialStat == nullptr ||// may happen with failed rename events
                retries > 1 ||               // for second or worse attempts always refresh stat
                clock_type::now() - tEnqueue > std::chrono::seconds(1) ||
             (! readInitialStat->isRegularFile())) {
            readInitialStat = threadData->InputHandler->getStat(source);
            if (readInitialStat == nullptr) {
                state |= stateBitType::vanished;
                errorMessage = "vanished in action";
                return;
            }
        }
        if (! readInitialStat->isRegularFile()) {
            state |= stateBitType::ignore;
            errorMessage = "ignored non regular file";
            return;
        }
 
        if (ignoreExisting) { // check if copy exists and has same stat values
            auto destStat = threadData->OutputHandler->getStat(destination);
            if (destStat) { // destination exists, check stat values
                if (readInitialStat->size == destStat->size &&
                        readInitialStat->isSameMtimeAs(*destStat)) {
                    state |= stateBitType::ignore;
                    errorMessage = "ignored due to existing copy with same stat";
                    if (changeAttrsOnIgnoreExisting) {
                        if ((preserve.mode && readInitialStat->mode == destStat->mode)
                                || (preserve.ownership && readInitialStat->ownerUid == destStat->ownerUid
                                    && readInitialStat->ownerGid == destStat->ownerGid)) {
                            errorMessage += ", preserved attrs are already fine";
                        } else {
                            threadData->OutputHandler->doAttributePreservations(destination, *readInitialStat);
                            errorMessage += ", attributes updated";
                        }
                    }
                    return;
                }
            }
        }
 
        auto input = threadData->InputHandler->newReader(source, state, *readInitialStat);
        std::unique_ptr<ioHandle::attrDataType> attrData;
        if (preserve.attrs) {
            attrData = input->getAttrData(threadData->OutputHandler);
        }
        std::unique_ptr<acl::list> aclData;
        if (preserve.acls) {
            aclData = input->getAclData();
        }
 
        attrset(*input, process_source_attrs);
        attrset(*input, start_source_attrs);
        try {
            std::string destNameDuringWrite(destination);
            std::string backupFile;
            switch (backupMode) {
                case backupModeType::none:
                    break;
                case backupModeType::before:
                case backupModeType::during: {
                    backupFile = destination + getBackupSuffix();
                    threadData->OutputHandler->renameSimple(destination, backupFile);
                    break;
                }
                case backupModeType::after: {
                    if (!backupSuffix.empty()) {
                        destNameDuringWrite += getBackupSuffix();
                    }
                    break;
                }
                case backupModeType::remove: {
                    stateType dummy;
                    threadData->OutputHandler->remove(destNameDuringWrite, dummy);
                    break;
                }
                default:
                    throw std::runtime_error("illegal backupModeType");
            }
            {
                // scope for writeHandle
                std::unique_ptr<outputHandler::base::writer> writeHandle;
                if (backupSuffix.empty() && backupMode == backupModeType::after) {
                    writeHandle = threadData->OutputHandler->newTmpWriter(destNameDuringWrite,
                                  readInitialStat->size,
                                  noCopy,
                                  std::move(attrData),
                                  std::move(aclData));
                } else {
                    writeHandle = threadData->OutputHandler->newWriter(destNameDuringWrite,
                                  (appendableFiles.fIsSet() &&
                                   regex_match(destination, appendableFiles)) ||
                                  std::any_of(append_attrs.cbegin(), append_attrs.cend(),
                    [this](decltype(append_attrs)::value_type attr) {
                        return ! threadData->InputHandler->getXattr(source, attr).empty();
                    }),
                    readInitialStat->size,
                    input->getBlockSize(),
                    state,
                    noCopy,
                    std::move(attrData),
                    std::move(aclData));
                }
                doBlockSizeSetup(*input, *writeHandle);
 
                input->setupSparseRegions(sparse);
 
                if (state & stateBitType::append) { // we want to append
                    auto startPosition = writeHandle->getSize();
                    // align start position to a disk block
                    startPosition = (startPosition / input->getBlockSize()) * input->getBlockSize();
                    writeHandle->seek(startPosition);
                    input->seek(startPosition);
                }
 
                if (perFileThreads
                        && readInitialStat->size / memoryBlockSize > 1 // no point in threading small files
                   ) {
                    doThreadedCopy(*input, writeHandle);
                } else {
                    doUnthreadedCopy(*input, writeHandle);
                }
 
                input->checkUnchangedness();
 
                checkAttributes(*input, check_source_attrs);
 
                if (!noCopy) {
                    if (syncWrittenFiles) {
                        writeHandle->sync();
                    }
                    if (!setAttributesAfterClose) {
                        writeHandle->doAttributePreservations(*readInitialStat);
                    }
                    attrset(*writeHandle, success_dest_attrs);
                }
                // writeHandle is deleted here when the unique_ptr goes out of scope
            } // scope for writeHandle
 
            // special workaround for dCache that sets mtime on close()
            if (setAttributesAfterClose && !noCopy) {
                threadData->OutputHandler->doAttributePreservations(destNameDuringWrite, *readInitialStat);
            }
            switch (backupMode) {
                case backupModeType::during: {
                    stateType dummy;
                    threadData->OutputHandler->remove(backupFile, dummy);
                    break;
                }
                case backupModeType::after:
                    threadData->OutputHandler->renameSimple(destNameDuringWrite, destination);
                default:
                    break;
            }
        } catch (const delayAdvisingError& e) {
            adviseDelay();
            errorMessage = e.what();
            attrset(*input, failure_source_attrs); // mark on input file
            throw; // continue in outer try/catch
        } catch (const std::exception& e) {
            errorMessage = e.what();
            attrset(*input, failure_source_attrs); // mark on input file
            throw; // continue in outer try/catch
        }
        attrdel(*input, process_source_attrs);
        attrdel(*input, failure_source_attrs); // we succeeded, remove old failure marks
        attrset(*input, success_source_attrs);
    } catch (const fatalException&) { // always throw, never return
        throw;
    } catch (const std::exception& e) { // no attribute massage for input, we don't have that
        errorMessage = e.what();
        state |= stateBitType::failed;
        if (continueOnError) {
            return;
        } else {
            throw;
        }
    }
    // check if earliest process time changed
    if (tWorkStart < filesInWorkIterator->second.getEarliestprocessTime()) {
        adviseDelay(); // try to avoid this next time
        errorMessage = "changed earliest process time, ";
        auto deltaT = filesInWorkIterator->second.getEarliestprocessTime() - tWorkStart;
        errorMessage += std::to_string(std::chrono::duration_cast<std::chrono::duration<double>>(deltaT).count());
        state |= stateBitType::failed;
        return;
    }
    state |= stateBitType::done;
}
 
 
void copyRequest::base::doBlockSizeSetup(ioHandle &input,
        ioHandle &output) {
    if (copyRequest::base::userReadBlockSize != 0) {
        input.setBlockSize(userReadBlockSize);
    }
    if (copyRequest::base::userWriteBlockSize != 0) {
        output.setBlockSize(userWriteBlockSize);
    }
 
    memoryBlockSize = input.getBlockSize() * output.getBlockSize();
    // sanitize block sizes, ensure they at least fit
    memoryBlockSize = std::min(memoryBlockSize, copyRequest::base::maxMemoryBlockSize.fGetValue());
    memoryBlockSize = std::max(memoryBlockSize, copyRequest::base::minMemoryBlockSize.fGetValue());
    input.setBlockSize(std::min(input.getBlockSize(), memoryBlockSize));
    output.setBlockSize(std::min(output.getBlockSize(), memoryBlockSize));
}
 
void copyRequest::base::waitForAllInstancesGone() {
    std::unique_lock<decltype(objectCountMutex)> lock(objectCountMutex);
    while (objectCount > 0) {
        objectCountCondVar.wait(lock);
    }
}
bool copyRequest::base::checkForInstances() {
    std::unique_lock<decltype(objectCountMutex)> lock(objectCountMutex);
    return (objectCount > 0);
}
 
void copyRequest::base::getSuffix(const std::string& path,
                                  std::string& suffix) {
    auto fileNameStart = path.find_last_of('/');
    if (fileNameStart == std::string::npos) {
        fileNameStart = 0;
    }
    auto const fileName = path.substr(fileNameStart);
    auto const suffixStart = fileName.find_last_of('.');
    if (suffixStart != decltype(fileName)::npos) {
        suffix = fileName.substr(suffixStart);
    }
}
 
void copyRequest::base::init(inputHandler::base* InputHandler) {
    threadData = nullptr;
    tEnqueue = clock_type::now();
    {
        std::unique_lock<decltype(objectCountMutex)> lock(objectCountMutex);
        ++objectCount;
    }
    getSuffix(source, suffix);
 
    for (const auto& item : std::vector<std::pair<const std::string&,
    const decltype(sourcePrefixRegex)&>>({
    {source, sourcePrefixRegex},
    {destination, destinationPrefixRegex}
})) {
        if (item.second.fIsSet()) {
            std::smatch match;
            if (std::regex_match(item.first, match, item.second)) {
                if (match.size() == 2) { // we have a sub match as required
                    prefix = match[1].str();
                }
            }
        }
    }
    // set this up before any return can happen, otherwise we can get empty lists..
    for (auto checksumCreator : checksumCreators) {
        checkSums.push_front(checksumCreator->create());
    }
 
    if (std::any_of(prohibitive_attrs.cbegin(), prohibitive_attrs.cend(),
    [this, InputHandler](decltype(prohibitive_attrs)::value_type attr) {
    return ! InputHandler->getXattr(source, attr).empty();
    })) {
        state.set(stateBitType::trucated);
        state.set(stateBitType::ignore);
        return;
    }
    if (readInitialStat == nullptr) {
        readInitialStat = InputHandler->getStat(source, false);
    }
    if (readInitialStat == nullptr) {
        if (!(state & stateBitType::fileToBeRenamed)) {
            state.set(stateBitType::vanished);
        }
        return;
    }
    if (readInitialStat->isLink()) {
        if (dereference) {
            readInitialStat = InputHandler->getStat(source, true);
            if (readInitialStat == nullptr) {
                throw std::runtime_error("can't dereference " + source);
            }
        } else if (state & stateBitType::fileToBeCopied) {
            state = stateBitType::linkToBeMade;
        }
    }
 
    if (state & stateBitType::fileToBeRenamed) {
        return; // no directory test needed if we move
    }
 
    if (readInitialStat->isDir()) {
        if (state & stateBitType::fileToBeRemoved) {
            state |= stateBitType::ignore;
        } else {
            state |= stateBitType::ignore;
            errMsg::emit(errMsg::level::notice, errMsg::location(),
                         source, "init",
                         "is a directory ", destination, " inotified at ", std::fixed, std::chrono::duration_cast<std::chrono::duration<double>>(tInotify.time_since_epoch()).count());
            //throw std::runtime_error(aSource + " is a directory, not expected here");
        }
    }
}
copyRequest::base::base(inputHandler::base* InputHandler,
                        const std::string& aSource,
                        const std::string& aDestination,
                        std::unique_ptr<const genericStat>& aStat,
                        const singleMap& aMapEntry,
                        bool remove,
                        clock_type::time_point timestamp) :
    readInitialStat(std::move(aStat)),
    state(remove ? stateBitType::fileToBeRemoved : stateBitType::fileToBeCopied),
    source(aSource),
    destination(aDestination),
    mapEntry(aMapEntry),
    tInotify(timestamp),
    retries(0) {
    init(InputHandler);
}
copyRequest::base::base(inputHandler::base* InputHandler,
                        const std::string& aSource,
                        const std::string& aDestination,
                        const singleMap& aMapEntry,
                        bool remove,
                        clock_type::time_point timestamp) :
    state(remove ? stateBitType::fileToBeRemoved : stateBitType::fileToBeCopied),
    source(aSource),
    destination(aDestination),
    mapEntry(aMapEntry),
    tInotify(timestamp),
    retries(0) {
    init(InputHandler);
}
copyRequest::base::base(inputHandler::base* InputHandler,
                        const std::string& aSource,
                        const std::string& aDestination,
                        const std::string& aMoveSource,
                        const std::string& aOrigSource,
                        const singleMap& aMapEntry,
                        clock_type::time_point timestamp) :
    state(stateBitType::fileToBeRenamed),
    source(aSource),
    destination(aDestination),
    moveSource(aMoveSource),
    origSource(aOrigSource),
    mapEntry(aMapEntry),
    tInotify(timestamp),
    retries(0) {
    init(InputHandler);
}
 
copyRequest::base::~base()  noexcept (false) {
    {
        std::unique_lock<decltype(objectCountMutex)> lock(objectCountMutex);
        --objectCount;
        if (objectCount == 0) {
            objectCountCondVar.notify_all();
        }
    }
 
    if (state & stateBitType::ignore) {
        return;
    }
 
    tTotalStat.addValue(tWorkDone - tInotify);
    tPipeStat.addValue(tEnqueue - tInotify);
    tWaitStat.addValue(tWorkStart - tEnqueue);
 
    if (state & stateBitType::done && state & stateBitType::fileToBeCopied) {
        auto dtCopy = tWorkDone - tWorkStart;
        auto tModification = readInitialStat->getMtime();
        auto d = tInotify - tModification;
        tTotal2Stat.addValue(tWorkDone - tModification);
        tInotifyStat.addValue(d);
        tCopyStat.addValue(dtCopy);
        if (readInitialStat) {
            bytesStat.addValue(readInitialStat->size);
        }
        auto dtCopyReal = std::chrono::duration_cast<std::chrono::duration<double>>(dtCopy).count();
 
        if (dtCopyReal > 0) {
            speedStat.addValue(readInitialStat->size / (1024. * 1024) / dtCopyReal);
        }
    }
}
 
 
void copyRequest::base::addExpectedChecksumResult(const std::string& checkSumType,
        const std::string& expectedValue) {
    for (auto& cksum : checkSums) {
        if (cksum->getName() == checkSumType) {
            cksum->setExpectedResult(expectedValue);
            return;
        }
    }
    auto checksumCreator = checksum::base::newFactory(checkSumType);
    checksumCreators.push_front(checksumCreator);
    auto cksum = checksumCreator->create();
    cksum->setExpectedResult(expectedValue);
    checkSums.push_front(cksum);
}
 
 
 
void copyRequest::base::printResults(std::ostream& hashStream,
                                     std::ostream& logStream) {
    timerInst(all);
    if (!(state & stateBitType::append)
            && (state & stateBitType::done)
            && !continueOnError) { 
        static auto escaper = escapism::newEscaper("C");
        for (const auto& sum : checkSums) {
            std::string name;
            escaper->escape(noCopy ? source : destination, name);
            if (checkSums.size() > 1) {
                hashStream << sum->getName() << ": ";
            }
            hashStream << sum->getResult() << "  " << name << "\n";
        }
    }
 
    bool needSpace = false;
 
    if (!logFields.empty() && !(state & stateBitType::done)) {
        logStream << logstream::level::warning;
    }
    std::string operation;
    if (state & stateBitType::fileToBeCopied) {
        operation = "copy";
    } else if (state & stateBitType::linkToBeMade) {
        operation = "link";
    } else if (state & stateBitType::fileToBeRemoved) {
        operation = "remove";
    } else if (state & stateBitType::fileToBeRenamed) {
        operation = "move";
    }
    if (resultOutput && (state & stateBitType::done)) {
        timerInst(sendMsg);
        #ifdef WithJsonCpp
        static std::atomic<Json::UInt64> messageCounter(0);
        Json::Value root;
        root["msgnr"] = messageCounter++;
        root["operation"] = operation;
        root["path"] = destination;
        root["source"] = source;
        if (readInitialStat != nullptr) {
            root["size"] = static_cast<Json::Int64>(readInitialStat->size);
        }
        for (const auto& sum : checkSums) {
            root[sum->getName()] = sum->getResult();
        }
        #ifdef WithMagic
        if (doMagic) {
            root["fileType"] = fileMagic;
        }
        #endif
        root["finishTime"] = std::chrono::duration_cast<std::chrono::duration<double>>(tWorkDone.time_since_epoch()).count();
        root["inotifyTime"] = std::chrono::duration_cast<std::chrono::duration<double>>(tInotify.time_since_epoch()).count();
        root["retries"] = retries;
        if (!prefixJsonName.empty()) {
            root[prefixJsonName] = prefix.empty() ? statPrefix : prefix;
        }
        for (auto& item : jsonExtraFields) {
            root[item.first] = item.second;
        }
        static Json::FastWriter jsonWriter;
        resultOutput->send(jsonWriter.write(root), prefix);
        #else
        resultOutput->send(destination, prefix);
        #endif
    }
    if (!(state & stateBitType::done) || (state & stateBitType::failed)) {
        timerInst(errmsgEmit);
        errMsg::emit(errMsg::level::info, errMsg::location(),
                     source, operation, errorMessage);
    }
    timerInst(toLogStream);
    for (const auto& field : logFields) {
        std::string value(field);
        static auto escaper = escapism::newEscaper("Url");
        if (!expandAttrValue(value)) { // not expanded, try special ones
            if (value == "%source") {
                value = "'";
 
                if ((state & stateBitType::fileToBeRenamed)
                        && !(state & stateBitType::fileToBeCopied)) {
                    value += moveSource;
                } else {
                    value += source;
                }
 
                value.push_back('\'');
            } else if (value == "%destination") {
                value = "'";
                value += destination;
                value.push_back('\'');
            } else if (value == "%urlsource") {
                value.clear();
 
                if ((state & stateBitType::fileToBeRenamed)
                        && !(state & stateBitType::fileToBeCopied)) {
                    escaper->escape(moveSource, value);
                } else {
                    escaper->escape(source, value);
                }
            } else if (value == "%urldestination") {
                value.clear();
                escaper->escape(destination, value);
            } else if (value == "%retries") {
                value = std::to_string(retries);
            } else if (value == "%error") {
                value = errorMessage;
 
                if (state & stateBitType::vanished
                        && !(state & stateBitType::fileToBeRemoved)) {
                    value += " vanished before action started";
                }
            } else if (value == "%op") {
                if (state & stateBitType::fileToBeCopied) {
                    value = "->";
                } else if (state & stateBitType::linkToBeMade) {
                    value = "ln";
                } else if (state & stateBitType::fileToBeRemoved) {
                    value = "rm";
                } else if (state & stateBitType::fileToBeRenamed) {
                    value = "mv";
                }
            }
        }
 
        if (needSpace) {
            logStream << " ";
        }
 
        logStream << value;
        needSpace  = true;
    }
 
    if (needSpace) {
        logStream << "\n";
    }
 
 
    if (verbose) {
        logStream << "'" << source << "' ";
 
        if (state & stateBitType::fileToBeCopied) {
            logStream << " -> ";
        }
 
        if (state & stateBitType::linkToBeMade) {
            logStream << " link ";
        }
 
        if (state & stateBitType::fileToBeRemoved) {
            logStream << " rm ";
        }
 
        if (state & stateBitType::fileToBeRenamed) {
            logStream << " mv ";
        }
 
        if (state & stateBitType::vanished) {
            logStream << " vanished before action started";
        }
 
        if (state & stateBitType::append) {
            logStream << " was appended to";
        }
 
        if (state & stateBitType::done) {
            logStream << " '" << destination << "' after " << retries << " attempts";
        }
 
        if (state & stateBitType::ignore) {
            logStream << " is ignored";
        }
 
        if (state & stateBitType::trucated) {
            logStream << " due to truncation";
        }
 
        if (state &  stateBitType::failed) {
            logStream << " failed due to " << errorMessage;
        }
 
        if (state &  stateBitType::inWork) {
            logStream << " inWork not cleared";
        }
 
        if (state & stateBitType::attributeMismatch) {
            logStream << " attribute mismatch: " << errorMessage;
        }
 
        logStream << "\n";
    }
}
 
std::function<void(std::ostream&)>& copyRequest::base::getStatPrinter() {
    return statPrinter;
}
void copyRequest::base::setStatPrinter(void(*f)(std::ostream&)) {
    statPrinter = f;
}
void copyRequest::base::printStats(std::ostream& stream) {
    stream << statPrefix << tTotal2Stat << "\n";
    stream << statPrefix << tTotalStat << "\n";
    stream << statPrefix << tInotifyStat << "\n";
    stream << statPrefix << tPipeStat << "\n";
    stream << statPrefix << tWaitStat << "\n";
    stream << statPrefix << tCopyStat << "\n";
    stream << statPrefix << tEnqueueStat << "\n";
    stream << statPrefix << bytesStat << "\n";
    stream << statPrefix << speedStat << "\n";
    stream << statPrefix << bytesStat.getN() << " files with " << bytesStat.sum << "B transferred\n";
 
    auto tElapsed = std::chrono::duration_cast<std::chrono::duration<double>>(clock_type::now() - tProgramStart).count();
    stream << statPrefix << "avg rate: " << bytesStat.sum / (tElapsed * 1024 * 1024) << " MiB/s in " << tElapsed << " s\n";
    for (unsigned int i = 0; i < processMultiplicities->size(); i++) {
        stream << statPrefix << i << " processes at once: "
               << (*processMultiplicities)[i] << " times\n";
    }
 
    std::multimap<clock_type::duration, std::string> waitMap;
    for (const auto& item : advisoryWaitMap) {
        waitMap.emplace(item.second, item.first);
    }
    for (const auto& item : waitMap) {
        stream << statPrefix << " wait time for " << item.second << " files: "
               << std::chrono::duration_cast<std::chrono::duration<double>>(item.first).count() << "s\n";
    }
    timer::anchor::print(stream, statPrefix);
};
 
void copyRequest::base::resetStats() {
    tTotal2Stat.reset();
    tTotalStat.reset();
    tInotifyStat.reset();
    tPipeStat.reset();
    tWaitStat.reset();
    tCopyStat.reset();
    tEnqueueStat.reset();
    bytesStat.reset();
    speedStat.reset();
    bytesStat.reset();
    timer::anchor::reset();
}
 
bool copyRequest::base::retry(std::unique_ptr<base>& request,
                              timedQueue& delayedRequests) {
    if ((request->state & stateBitType::failed) &&
            !(request->state & stateBitType::fileToBeRemoved) &&
            (request->retries < maxRetries)) {
        static ssize_t lastSize;
        ssize_t size;
        {
            std::ifstream statm("/proc/self/statm");
            statm >> size;
        }
        lastSize = size;
        request->state.clear(stateBitType::failed);
        auto waitTime = request->tWorkDone - request->tWorkStart + std::chrono::seconds(1);
        waitTime *= request->retries;
        errMsg::emit(errMsg::level::debug, errMsg::location(),
                     request->source, "re-sched",
                     request->errorMessage,
                     ", delaying by ", std::fixed, std::chrono::duration_cast<std::chrono::duration<double>>(waitTime).count(),
                     "s vmsize: ", size, "(", lastSize, ", ", size - lastSize, "), retries: ", request->retries);
        delayedRequests.enqueue(request, waitTime);
        if (request != nullptr) {
            throw std::runtime_error("requesy not null");
        }
        return true;
    }
    return false;
}
 
void copyRequest::base::processQueue(simpleQueue& queue,
                                     simpleQueue& resultQueue,
                                     timedQueue& delayedRequests) {
    try {
        perThreadData threadData;
 
        while (auto request = queue.dequeue()) {
            if (stopRequest::Requested(stopRequest::handlerIdType::processLoop)) {
                break;
            }
            while (request) {
                auto dt = request->filesInWorkIterator->second.getEarliestprocessTime()
                          - clock_type::now();
                if (dt > decltype(dt)::zero()) {
                    errMsg::emit(errMsg::level::debug, errMsg::location(),
                                 request->source, "delay",
                                 "scheduled too early by ", std::chrono::duration_cast<std::chrono::duration<double>>(dt).count(), "s");
                    auto newDelay = request->filesInWorkIterator->second.getEarliestprocessTime() - request->tInotify;
                    request->adviseDelay(newDelay);
                    delayedRequests.enqueue(request, dt);
                    if (request != nullptr) {
                        throw std::runtime_error("request not nullptr");
                    }
                    break;
                }
 
 
                throttle::action action(fileRateLimit);
                request->process(threadData);
                if (! retry(request, delayedRequests)) {
                    copyRequest::fileInWork::eraser eraser(*request);
                    resultQueue.enqueue(request);
                    if (request != nullptr) {
                        throw std::runtime_error("request not nullptr");
                    }
                    request = eraser.getNext();
                }
            }
        }
    } catch (const std::exception& e) {
        stopRequest::RequestStop(e.what());
        errMsg::emit(errMsg::level::debug, errMsg::location(),
                     "--", "process", "stop requested due to ", e.what());
    } catch (...) {
        stopRequest::RequestStop("unknown exception");
        errMsg::emit(errMsg::level::debug, errMsg::location(),
                     "--", "process", "stop requested due to unknown exception");
    }
}
 
namespace copyRequest {
    std::ostream& operator<< (std::ostream& stream,
                              const copyRequest::clock_type::time_point& t) {
        stream << std::chrono::duration_cast<std::chrono::duration<double>>(t.time_since_epoch()).count();
        return stream;
    }
 
    std::ostream& operator<<(std::ostream& out, const base& request) {
        out << std::fixed;
        out << " request " << static_cast<const void*>(&request);
        out << " for '" << request.source << "' to '" << request.destination << "'";
        out << " state " << std::hex << static_cast<unsigned>(request.state) << std::dec;
        out << " tInotify " << request.tInotify;
        out << " tEarliest " << request.filesInWorkIterator->second.getEarliestprocessTime();
        out << " now: " << clock_type::now() << "\n";
        return out;
    }
}