ewmscp.cpp File Reference

(v0.19-24-g0617ca1 with changes)

#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"
#include "requestProvider.h"
#include "git-rev.h"

Include dependency graph for ewmscp.cpp:

Go to the source code of this file.

Classes
class	queuesAndThreads
	class to make sure all threads are joined before thread object is deleted in case of an exception to do so, all queues must be signalled as done More...

Functions
decltype(uid)	uid ('\0', "uid", "uid of created files and directories", -1)
decltype(gid)	gid ('\0', "gid", "gid of created files and directories", -1)
decltype(noCopy)	noCopy ('\0', "noCopy", "do not create a copy", false)
decltype(dereference)	dereference ( 'L', "dereference", "always follow symbolic links in SOURCE", false)
decltype(verbose)	verbose ( 'v', "verbose", "explain what is being done", false)
decltype(quiet)	quiet ( 'q', "quiet", "be as quiet as possible", false)
decltype(workDir)	workDir ('\0', "workDir", "use DIRECTORY as work dir")
decltype(continueOnError)	continueOnError ('\0', "continueOnError", "don't stop on errors in non-follow mode", false)
decltype(nThreads)	nThreads ('\0', "nThreads", "number of simultaneous copys", 1)
decltype(writeRateLimit)	writeRateLimit ("writeRateLimit")
decltype(fileRateLimit)	fileRateLimit ("fileRateLimit")
decltype(readRateLimit)	readRateLimit ("readRateLimit")
decltype(modeBits)	modeBits ('\0', "mode", "access mode for copied files", 0644)
decltype(statPrefix)	statPrefix ('\0', "statPrefix", "prefix for statistics log lines", "")
decltype(requestProviderName)	requestProviderName ('\0', "requestProvider", "name of request provider", "cmdLine")
decltype(printStatAnyway)	printStatAnyway ('\0', "printStatAnyway", "print stat values at the end anyway")
decltype(outputHandlerName)	outputHandlerName ('\0', "outputHandler", "name of outputHandler", "posixFile")
decltype(inputHandlerName)	inputHandlerName ('\0', "inputHandler", "name of inputHandler", "posixFile")
decltype(resultOutput)	resultOutput (nullptr)
static void	pidFileRemover ()
static volatile std::atomic< bool >	statPrintRequested (false)
static volatile std::atomic< bool >	statResetRequested (false)
void	sigUsrHandler (int sigNum)
void	delayRequest (copyRequest::timedQueue &delayedRequests, copyRequest::simpleQueue &requests)
	runs as it's own thread to move delayed requests from the delay queue to the requests queue as soon as they become available. More...
void	printResults (copyRequest::simpleQueue &results, copyRequest::simpleQueue &requests, std::ostream &hashStream, std::ostream &logStream, std::ostream &statStream)
void	printErrors (std::ostream &errStream, messageQueue::queue *errQueue)

Variables
decltype(preserve)	preserve
	set of properties to preserve in the copy More...
decltype(checksumCreators)	checksumCreators
static options::single< std::string >	pidFileName ('\0', "pidFile", "name of PID file")
static options::single< bool >	printErrorLocation ('\0', "printErrorLocation", "print error location (file,line,v fct) instead of prefix")

Function Documentation

delayRequest()

void delayRequest	(	copyRequest::timedQueue &	delayedRequests,
		copyRequest::simpleQueue &	requests
	)

runs as it's own thread to move delayed requests from the delay queue to the requests queue as soon as they become available.

The delaying itself is hidden inside the delayedRequests queue, see waitQueues::timed

Definition at line 224 of file ewmscp.cpp.

                                                    {
    while (auto request = delayedRequests.dequeue()) {
        requests.enqueue(request);
    }
}

References waitQueues::timed< T, clock_type >::dequeue(), and waitQueues::simple< T >::enqueue().

Here is the call graph for this function:

pidFileRemover()

static void pidFileRemover ( )

static

Definition at line 120 of file ewmscp.cpp.

                             {
    if (! pidFileName.empty()) {
        throwcall::good0(unlink(pidFileName.c_str()), "can't unlink pid file ", pidFileName);
    }
}

References throwcall::good0(), and pidFileName.

Here is the call graph for this function:

printErrors()

void printErrors	(	std::ostream &	errStream,
		messageQueue::queue *	errQueue
	)

Definition at line 291 of file ewmscp.cpp.

                                                                     {
    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");
    }
}

References errStream(), logstream::namedLevel::getNameByLevel(), errMsg::message::getQueue(), printErrorLocation, stopRequest::RequestStop(), messageQueue::queue::send(), statPrefix(), and timerInst.

Here is the call graph for this function:

printResults()

void printResults	(	copyRequest::simpleQueue &	results,
		copyRequest::simpleQueue &	requests,
		std::ostream &	hashStream,
		std::ostream &	logStream,
		std::ostream &	statStream
	)

Definition at line 231 of file ewmscp.cpp.

                                          {
    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");
    }
}

References waitQueues::simple< T >::dequeue(), errMsg::emit(), copyRequest::base::getStatPrinter(), stopRequest::RequestStop(), copyRequest::base::resetStats(), waitQueues::simple< T >::size(), statPrefix(), statPrintRequested(), statResetRequested(), and errMsg::warning.

Here is the call graph for this function:

resultOutput()

decltype( resultOutput ) resultOutput

(

nullptr

)

sigUsrHandler()

void sigUsrHandler

(

int

sigNum

)

Definition at line 129 of file ewmscp.cpp.

                               {
    statPrintRequested = true;
    if (sigNum == SIGUSR2) {
        statResetRequested = true;
    }
}

References statPrintRequested(), and statResetRequested().

Here is the call graph for this function:

statPrintRequested()

static volatile std::atomic<bool> statPrintRequested ( false  )

static

Referenced by printResults(), and sigUsrHandler().

Here is the caller graph for this function:

statResetRequested()

static volatile std::atomic<bool> statResetRequested ( false  )

static

Referenced by printResults(), and sigUsrHandler().

Here is the caller graph for this function:

Variable Documentation

checksumCreators

decltype( checksumCreators ) checksumCreators

Definition at line 113 of file ewmscp.cpp.

Referenced by copyRequest::base::addExpectedChecksumResult(), and copyRequest::base::init().

pidFileName

options::single<std::string> pidFileName('\0', "pidFile", "name of PID file")

static

Referenced by pidFileRemover().

preserve

decltype( preserve ) preserve

set of properties to preserve in the copy

Definition at line 111 of file ewmscp.cpp.

Referenced by outputHandler::libssh::writerLibssh::doAttributePreservations(), outputHandler::posixFile::writerPosixFile::doAttributePreservations(), outputHandler::daosFs::writerDaosFs::doAttributePreservations(), outputHandler::dcap::doAttributePreservations(), outputHandler::posixFile::doAttributePreservations(), outputHandler::daosFs::doAttributePreservations(), outputHandler::dcap::ensureParentDirs(), and copyRequest::base::process().

printErrorLocation

options::single<bool> printErrorLocation('\0', "printErrorLocation", "print error location (file,line,v fct) instead of prefix")

static

Referenced by printErrors().