ewmscp.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 <atomic>
#include <condition_variable>
#include <csignal>
#include <deque>
#include <forward_list>
#include <iomanip>
#include <iostream>
#include <mutex>
#include <thread>
#include <vector>
 
 
#include <dirent.h>
#include <grp.h>
#include <libgen.h>
#include <pwd.h>
#include <string.h>
#include <sys/resource.h>
#include <sys/time.h>
#include <unistd.h>
 
#include "timer.h"
#include "copyRequest.h"
#include "ewmscp.h"
#include "scoped.h"
#include "syslogstream.h"
#include "errMsgQueue.h"
#include "throwcall.h"
#ifdef WithKafka
#include "kafkaQueue.h"
#endif
#ifdef WithJsonCpp
#include <json/json.h>
#endif
#include "requestProvider.h"
#include "git-rev.h"
 
defineStatic(uid, '\0', "uid", "uid of created files and directories", -1);
defineStatic(gid, '\0', "gid", "gid of created files and directories", -1);
defineStatic(noCopy, '\0', "noCopy", "do not create a copy", false);
defineStatic(dereference, 'L', "dereference",
             "always follow symbolic links in SOURCE", false);
defineStatic(verbose, 'v', "verbose", "explain what is being done", false);
defineStatic(quiet, 'q', "quiet", "be as quiet as possible", false);
defineStatic(workDir, '\0', "workDir", "use DIRECTORY as work dir");
defineStatic(continueOnError, '\0', "continueOnError", "don't stop on errors in non-follow mode", false);
defineStatic(nThreads, '\0', "nThreads", "number of simultaneous copys", 1);
 
defineStatic(writeRateLimit, "writeRateLimit");
defineStatic(fileRateLimit, "fileRateLimit");
defineStatic(readRateLimit, "readRateLimit");
 
defineStatic(modeBits, '\0', "mode",
             "access mode for copied files", 0644);
 
defineStatic(statPrefix, '\0', "statPrefix",
             "prefix for statistics log lines", "");
 
defineStatic(requestProviderName, '\0', "requestProvider",
             "name of request provider",
             "cmdLine");
defineStatic(printStatAnyway, '\0', "printStatAnyway",
             "print stat values at the end anyway");
 
#ifdef WithJsonCpp
defineStatic(prefixJsonName, '\0', "prefixJsonName",
             "name of statPrefix in json output", "");
defineStatic(jsonExtraFields, '\0', "jsonExtraFields",
             "extra field to add to json messages");
#endif
 
 
defineStatic(outputHandlerName, '\0', "outputHandler",
             "name of outputHandler", "posixFile");
defineStatic(inputHandlerName, '\0', "inputHandler",
             "name of inputHandler", "posixFile");
 
 
 
defineStaticNoArg(preserve);
 
defineStaticNoArg(checksumCreators);
 
defineStatic(resultOutput, nullptr);
 
static options::single<std::string> pidFileName('\0', "pidFile",
        "name of PID file");
 
static void pidFileRemover() {
    if (! pidFileName.empty()) {
        throwcall::good0(unlink(pidFileName.c_str()), "can't unlink pid file ", pidFileName);
    }
}
 
 
static volatile std::atomic<bool> statPrintRequested(false);
static volatile std::atomic<bool> statResetRequested(false);
void sigUsrHandler(int sigNum) {
    statPrintRequested = true;
    if (sigNum == SIGUSR2) {
        statResetRequested = true;
    }
}
 
defineStatic(stopRequest::handlerId, stopRequest::handlerIdType::unhandled);
defineStaticNoArg(stopRequest::reason);
defineStatic(stopRequest::stopRequested, false);
void stopRequest::sigHandler(int) {
    stopRequested = true;
    //  static_assert(std::atomic_is_lock_free(&stopRequested), "atomic bool is not always lock free"); // assert that stopRequested is always lock-free
}
void stopRequest::instantiate(stopRequest::handlerIdType id) {
    if (handlerId != handlerIdType::unhandled) {
        throw std::logic_error("stop request handler cannot be instatiated twice");
    }
    if (id == handlerIdType::unhandled || id == handlerIdType::undefined) {
        throw std::logic_error("illegal handler id value");
    }
    std::signal(SIGTERM, sigHandler);
    handlerId = id;
}
bool stopRequest::Requested() {
    if (handlerId != handlerIdType::unhandled) {
        return stopRequested;
    }
    return false;
}
bool stopRequest::Requested(handlerIdType id) {
    return handlerId == id && stopRequested;
}
void stopRequest::RequestStop(const std::string& aReason) {
    if (handlerId != handlerIdType::unhandled) {
        static std::mutex instanceProtector;
        std::unique_lock<decltype(instanceProtector)> lock(instanceProtector);
        stopRequested = true;
        reason = aReason;
    } else {
        errMsg::emit(errMsg::level::crit, errMsg::location(),
                     "stopRequest", "requestStop", "no stop request instance");
    }
}
void stopRequest::ThrowUpReasonably() {
    if (handlerId != handlerIdType::unhandled) {
        if (stopRequested) {
            if (! reason.empty()) {
                throw std::runtime_error(reason);
            } else {
                throw std::runtime_error("SIGTERM received");
            }
        }
    }
}
 
void preservables::set(const std::string& opt)  {
    std::remove_reference<decltype(opt)>::type::size_type start = 0;
    auto stop = opt.find(',', start);
    while (true) {
        auto keyword = opt.substr(start, stop - start);
        if (keyword == "mode") {
            mode = true;
        } else if (keyword == "ownership") {
            ownership = true;
        } else if (keyword == "timestamps") {
            timestamps = true;
        } else if (keyword == "attr" || keyword == "xattr") {
            attrs = true;
        } else if (keyword == "acls") {
            acls = true;
        } else if (keyword == "all") {
            mode = true;
            ownership = true;
            timestamps = true;
            attrs = true;
            acls = true;
        } else {
            throw std::invalid_argument("bad --preserve argument " + keyword);
        }
        if (stop == std::remove_reference<decltype(opt)>::type::npos) {
            break;
        }
        start = stop + 1;
        stop = opt.find(',', start);
    }
}
 
 
 
 
 
void delayRequest(copyRequest::timedQueue& delayedRequests,
                  copyRequest::simpleQueue& requests) {
    while (auto request = delayedRequests.dequeue()) {
        requests.enqueue(request);
    }
}
 
void printResults(copyRequest::simpleQueue& results,
                  copyRequest::simpleQueue& requests,
                  std::ostream& hashStream,
                  std::ostream& logStream,
                  std::ostream& statStream) {
    try {
        bool timedOut = true;
        auto lastQueueSize = requests.size();
        auto lastQueueSizePrint = std::chrono::system_clock::now();
        
        while (timedOut == true) {
            while (auto request = results.dequeue(std::chrono::seconds(1), timedOut)) {
                request->printResults(hashStream, logStream);
            //      delete request;
            }
            
            if (timedOut) {
                logStream.flush();
            }
            
            auto now = std::chrono::system_clock::now();
            
            if (statPrintRequested) {
                statStream << statPrefix << "statistics values at " << std::fixed
                       << std::chrono::duration<double>(now.time_since_epoch()).count() << "\n";
                statStream << statPrefix << "outstanding requests: " << requests.size() << "\n";
                copyRequest::base::getStatPrinter()(statStream);
                statStream.flush();
                statPrintRequested = false;
            } else {
                auto queueSize = requests.size();
                
                if (lastQueueSize != queueSize &&
                    queueSize > 1 &&
                    now - lastQueueSizePrint > std::chrono::seconds(1)) {
                    statStream << statPrefix << "outstanding requests: " << requests.size() << "\n";
                    lastQueueSize = queueSize;
                    lastQueueSizePrint = now;
                }
                
            }
            if (statResetRequested) {
                copyRequest::base::resetStats();
                statResetRequested = false;
            }
        }
        
        copyRequest::base::getStatPrinter()(statStream);
    } catch (const std::exception& e) {
        errMsg::emit(errMsg::level::warning, errMsg::location(),
                     "printThread","caught exception ",e.what());
        stopRequest::RequestStop(e.what());
    } catch (...) {
        stopRequest::RequestStop("unknown exception");
    }
}
 
static options::single<bool> printErrorLocation('\0', "printErrorLocation",
        "print error location (file,line,v fct) instead of prefix");
 
void printErrors(std::ostream& errStream, messageQueue::queue* errQueue) {
    try {
        while (auto msg = errMsg::message::getQueue().dequeue()) {
            timerInst(produceLogMsg);
            if (printErrorLocation) {
                errStream <<  msg->getLoc().getFile() << ":" << msg->getLoc().getLine()
                          << ": in " << msg->getLoc().getFunc()  << "(): ";
            } else {
                errStream << msg->getLogLevel() << statPrefix << ": ";
            }
            errStream << msg->getObject() << " " << msg->getAction() << " " << msg->getMessage() << "\n";
            if (errQueue) {
#ifdef WithJsonCpp
                Json::Value root;
                if (!prefixJsonName.empty()) {
                    root[prefixJsonName] = statPrefix;
                }
                for (auto& item : jsonExtraFields) {
                    root[item.first] = item.second;
                }
                root["level"] = logstream::namedLevel::getNameByLevel(msg->getLogLevel());
                root["path"] = msg->getObject();
                root["operation"] = msg->getAction();
                root["message"] = msg->getMessage();
                root["time"] = std::chrono::duration_cast<std::chrono::duration<double>>(msg->getTimeStamp().time_since_epoch()).count();
                static Json::FastWriter jsonWriter;
                errQueue->send(jsonWriter.write(root));
#else
                errQueue->send(msg->getMessage());
#endif
            }
        }
    } catch (const std::exception& e) {
        stopRequest::RequestStop(e.what());
    } catch (...) {
        stopRequest::RequestStop("unknown exception");
    }
}
 
class queuesAndThreads {
    template <class T> void stopQueue(T& queue, const std::string& name) {
        try {
            queue.signalDone();
        } catch (const std::logic_error& e) {
            if (queue.size() > 1) {
                errMsg::emit(errMsg::level::debug, errMsg::location(), name, "signalled", "done");
            }
        }
    }
    void joinThread(std::thread& t) {
        if (t.joinable()) {
            t.join();
        }
    }
  public:
    copyRequest::simpleQueue requests;
    copyRequest::timedQueue delayedRequests;
    copyRequest::simpleQueue results;
    std::thread printer;
    std::thread delayer;
    std::forward_list<std::thread> workers;
    queuesAndThreads() = default;
    ~queuesAndThreads() noexcept(true) {
        stopQueue(requests, "requests");
        stopQueue(delayedRequests, "delayedRequests");
        stopQueue(results, "results");
        joinThread(printer);
        joinThread(delayer);
        for (auto& worker : workers) {
            joinThread(worker);
        }
    }
};
 
 
 
int main(int argc, const char* argv[]) {
    options::parser parser(EWMSCP_VERSION " https://stash.desy.de/projects/LSDMA/repos/ewmscp/commits/" EWMSCP_COMMIT, "", {});
    options::positional<options::container<std::string>>sources(10, "source",
            "source files to be copied");
    options::positional<options::single<std::string>>destination(20, "dest",
            "destination file or directory");
    options::single<std::string>target('\0', "target-directory",
                                       "copy all source arguments into directory");
    options::single<bool>parents('\0', "parents",
                                 "use full source file name under DIRECTORY", false);
    options::single<std::string> user('\0', "user",
                                      "user of created files and directories", "");
    user.fForbid(&uid);
    uid.fForbid(&user);
    options::single<std::string> group('\0', "group",
                                       "group of created files and directories", "");
    group.fForbid(&gid);
    gid.fForbid(&group);
    options::single<std::string> preserveOpt('\0', "preserve",
            "preserve the given attribtes");
    options::single<bool> preserveAll('p', "preserveAll",
                                      "preserve mode,ownership,timestamps");
 
    options::single<std::string> hashListFile('\0', "hashList",
            "file to write hash results into");
    options::single<std::string> logFile('\0', "log",
                                         "file to write log messages into");
    options::single<std::string> statListFile('\0', "statList",
            "file to write statistics into");
    options::single<std::string> errorStream('\0', "errors",
            "file to write errors to");
 
    options::single<std::string> startStopMsg('\0', "startStopMsg",
            "message to show at start or stop");
 
    options::single<bool> allowCoreDumps('\0', "allowCoreDumps",
                                         "set core dump limit to inf, 0 else", false);
 
 
    requestProvider::addAllowedNamesToOption(requestProviderName);
 
    options::container<std::string> checksumCreatorNames('\0', "checksum",
            "name of a checksum");
    checksum::base::addAllowedNamesToOption(checksumCreatorNames);
 
    outputHandler::base::addAllowedNamesToOption(outputHandlerName);
    inputHandler::base::addAllowedNamesToOption(inputHandlerName);
 
    #ifdef WithKafka
    messageQueue::kafka kafkaOutQueue("kafkaOut");
    messageQueue::kafka kafkaErrQueue("kafkaErr");
    #endif
    parser.fParse(argc, argv);
 
 
    if (allowCoreDumps) {
        struct rlimit new_limit = {RLIM_INFINITY, RLIM_INFINITY};
        throwcall::good0(setrlimit(RLIMIT_CORE, &new_limit), "can't set coredump limit to infinity");
    }   else {
        struct rlimit new_limit = {0, 0};
        throwcall::good0(setrlimit(RLIMIT_CORE, &new_limit), "can't set coredump limit to 0");
    }
 
    if (!pidFileName.empty()) {
        std::ofstream pidFile(pidFileName);
        pidFile << getpid() << "\n";
        throwcall::good0(std::atexit(pidFileRemover), "can't register pidFileRemover");
    }
 
    messageQueue::queue* errorOutput = nullptr;
    #ifdef WithKafka
    kafkaOutQueue.init();
    if (kafkaOutQueue.isSenderConfigured()) {
        resultOutput = &kafkaOutQueue;
    }
    kafkaErrQueue.init();
    if (kafkaErrQueue.isSenderConfigured()) {
        errorOutput = &kafkaErrQueue;
    }
    #endif
 
    logstream::provider errStreamBase(errorStream, std::cerr);
    std::thread errPrinterThread(printErrors, std::ref(errStreamBase.getStream()),
                                 errorOutput);
 
    logstream::provider hashStreamBase(hashListFile);
    auto& hashStream = hashStreamBase.getStream();
    logstream::provider logStreamBase(logFile);
    auto& logStream = logStreamBase.getStream();
    logstream::provider statStreamBase(statListFile);
    auto&  statStream = statStreamBase.getStream();
 
 
    for (const auto& name : checksumCreatorNames) {
        checksumCreators.push_front(checksum::base::newFactory(name));
    }
 
    if (!statPrefix.empty()) {
        errMsg::emit(errMsg::level::info, errMsg::location(),
                     "ewmscp", "starting",
                     EWMSCP_VERSION, " pid: ", getpid());
    }
    if (!startStopMsg.empty()) {
        logStream << logstream::level::info << startStopMsg << " ewmscp " << EWMSCP_VERSION << " starting.\n";
    }
 
    try {
 
        if (group.fIsSet()) {
            auto entry = getgrnam(group.c_str());
 
            if (entry) {
                gid = entry->gr_gid;
            } else {
                throw std::runtime_error("groupname " +  group + " not found.");
            }
        }
 
        if (user.fIsSet()) {
            auto entry = getpwnam(user.c_str());
 
            if (entry) {
                uid = entry->pw_uid;
            } else {
                throw std::runtime_error("username " +  user + " not found.");
            }
        }
 
 
        if (nThreads == 0) {
            nThreads = sysconf(_SC_NPROCESSORS_ONLN);
            if (copyRequest::base::perFileThreads) {
                nThreads = std::max(nThreads / (checksumCreators.empty() ? 2 : 3), 1u);
            }
        }
        copyRequest::base::processMultiplicities = new  std::remove_pointer<decltype(copyRequest::base::processMultiplicities)>::type(nThreads + 1);
 
        if (preserveOpt.fIsSet()) {
            preserve.set(preserveOpt);
        }
 
        if (preserveAll) {
            preserve.mode = true;
            preserve.ownership = true;
            preserve.timestamps = true;
        }
 
 
        if (workDir.fIsSet()) {
            throwcall::good0(chdir(workDir.c_str()), "can't change work dir to ", workDir);
        }
 
 
 
        if (target.fIsSet() || noCopy) { // the last positional arg is also a source file name
            sources.push_back(destination);
            std::string& dstRef(destination);
            dstRef = target;
        }
 
        queuesAndThreads qt;
 
 
        auto reqProv = requestProvider::newProvider(requestProviderName,
                       qt.requests, qt.delayedRequests, parents,
                       inputHandler::base::newHandler(inputHandlerName),
                       outputHandler::base::newHandler(outputHandlerName)
                                                   );
 
        if ((nThreads > 1) || reqProv->isFollowMode())  {
            for (unsigned int i = 0; i < nThreads; i++) {
                qt.workers.emplace_front(std::thread(copyRequest::base::processQueue,
                                                     std::ref(qt.requests),
                                                     std::ref(qt.results),
                                                     std::ref(qt.delayedRequests)));
            }
 
            qt.printer = std::thread(printResults, std::ref(qt.results), std::ref(qt.requests),
                                     std::ref(hashStream), std::ref(logStream), std::ref(statStream));
            std::signal(SIGUSR1, sigUsrHandler);
            std::signal(SIGUSR2, sigUsrHandler);
            stopRequest::instantiate(reqProv->isFollowMode() ?
                                     stopRequest::handlerIdType::requestProvider :
                                     stopRequest::handlerIdType::processLoop);
        }
        if (reqProv->isFollowMode()) {
            continueOnError = true;
        }
        qt.delayer = std::thread(delayRequest,
                                 std::ref(qt.delayedRequests), std::ref(qt.requests));
        reqProv->prepareMappings(sources, destination);
        if (verbose) {
            reqProv->printMappings(logStream);
        }
        reqProv->processSources(sources);
        if (reqProv->isFollowMode()) {
            copyRequest::fileInWork::waitForAllInstancesGone();
        } else {
            if (qt.workers.empty()) {
                copyRequest::perThreadData threadData;
                while (copyRequest::base::checkForInstances()) {
                    auto request = qt.requests.dequeue();
                    request->process(threadData);
                    request->printResults(hashStream, logStream);
                    copyRequest::base::retry(request, qt.delayedRequests);
                }
                if (printStatAnyway) {
                    copyRequest::base::getStatPrinter()(statStream);
                }
            } else {
                copyRequest::base::waitForAllInstancesGone();
            }
        }
        stopRequest::ThrowUpReasonably();
    } catch (const std::exception& e) {
        errMsg::emit(errMsg::level::crit, errMsg::location(),
                     "ewmscp", "ended",
                     "badly: ", e.what(), ", pid: ", getpid());
        errMsg::message::getQueue().signalDone();
        errPrinterThread.join();
        logStream << logstream::level::crit << startStopMsg << " ended badly: " << e.what() << std::endl;
        return EXIT_FAILURE;
    }
    if (!statPrefix.empty()) {
        errMsg::emit(errMsg::level::info, errMsg::location(),
                     "ewmscp", "ended", "sucessfully, pid: ", getpid());
    }
    if (!startStopMsg.empty()) {
        logStream <<  logstream::level::info << startStopMsg << " ended sucessfully" << std::endl;
    }
    errMsg::message::getQueue().signalDone();
    errPrinterThread.join();
    return EXIT_SUCCESS;
}