class for copy requests. More...

#include <copyRequest.h>

Inheritance diagram for copyRequest::base:

Collaboration diagram for copyRequest::base:

Classes
class	blockReadRequest

class	hashCalculator

class	pointerCompare

class	registerme
	class to register the keyword expansion member functions in the map More...

class	writeActor

Public Member Functions
const genericStat &	getInitialStat () const

const std::string &	getSource () const

clock_type::duration	getAdvisedDelay () const
	returns the advised delay More...

void	registerIterator (decltype(filesInWorkIterator) iter)

decltype(filesInWorkIterator)	getFilesInWorkIteraror () const

fileInWork::slotTypes	getSlotType () const

void	changeRequestType (stateBitType newType)

	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)
	constructor for copy,link and remove requests. More...

	base (inputHandler::base *InputHandler, const std::string &aSource, const std::string &aDestination, const singleMap &aMapEntry, bool remove, clock_type::time_point timestamp)
	constructor for copy,link and remove requests. More...

	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)
	constructor for rename requests only. More...

virtual	~base () noexcept(false)

virtual void	process (perThreadData &threadData)

virtual void	printResults (std::ostream &hashStream, std::ostream &logStream)

const std::string &	getSuffix () const

void	addExpectedChecksumResult (const std::string &checkSumType, const std::string &expectedValue)


void	kw_size (std::string &value) const

void	kw_mtime (std::string &value) const

void	kw_inow (std::string &value) const

void	kw_now (std::string &value) const

void	kw_version (std::string &value) const

void	kw_commit (std::string &value) const

void	kw_prefix (std::string &value) const


void	kw_reason (std::string &value) const

void	kw_state (std::string &value) const

void	kw_finishtime (std::string &value) const

Static Public Member Functions
static std::map< std::string, void(base::*)(std::string &) const > &	keywordMap ()

static bool	adviseDelay (clock_type::duration dt, const std::string &suffix)
	update the advised delay in the map. More...

static void	printStats (std::ostream &stream)

static void	resetStats ()

static bool	retry (std::unique_ptr< base > &request, timedQueue &delayedRequests)

static void	waitForAllInstancesGone ()

static bool	checkForInstances ()

static std::function< void(std::ostream &)> &	getStatPrinter ()
	we can't have static functions virtual, so we explicitly use a function to give us the proper print function, which can be changed in derived classes. More...

static void	setStatPrinter (void(*f)(std::ostream &))

static void	processQueue (simpleQueue &queue, simpleQueue &resultQueue, timedQueue &delayedRequests)

static void	getSuffix (const std::string &path, std::string &suffix)

Public Attributes
decltype(fileInWork::filesInWork.begin())	filesInWorkIterator = fileInWork::filesInWork.end()

Static Public Attributes
static options::single< bool >	setAttributesAfterClose

static options::single< bool >	perFileThreads

static options::single< size_t >	userWriteBlockSize

static options::single< size_t >	userReadBlockSize

static options::single< size_t >	maxMemoryBlockSize

static options::single< size_t >	minMemoryBlockSize

static statCollector::typed< copyRequest::clock_type::duration >	tEnqueueStat

static std::vector< std::atomic< unsigned int > > *	processMultiplicities

Protected Member Functions
void	adviseDelay (clock_type::duration dt=clock_type::duration::zero())

Protected Attributes
std::unique_ptr< const genericStat >	readInitialStat

std::list< checksum::base * >	checkSums

perThreadData *	threadData

stateType	state

std::string	errorMessage

const std::string	source

const std::string	destination

const std::string	moveSource

const std::string	origSource

const singleMap &	mapEntry

std::string	suffix

std::string	prefix

size_t	memoryBlockSize

clock_type::time_point	tInotify

clock_type::time_point	tEnqueue

clock_type::time_point	tWorkStart

clock_type::time_point	tWorkDone

unsigned	retries

Static Protected Attributes
static statCollector::typed< decltype(tWorkDone - tInotify) >	tTotalStat

static statCollector::typed< decltype(tWorkDone - tInotify) >	tTotal2Stat

static statCollector::typed< decltype(tWorkDone - tInotify) >	tInotifyStat

static statCollector::typed< decltype(tEnqueue - tInotify) >	tPipeStat

static statCollector::typed< decltype(tWorkStart - tEnqueue) >	tWaitStat

static statCollector::typed< decltype(tWorkDone - tWorkStart) >	tCopyStat

static statCollector::typed< decltype(readInitialStat->size) >	bytesStat

static statCollector::typed< double >	speedStat

static clock_type::time_point	tProgramStart

static std::atomic< unsigned int >	nConcurrentProcesses

static options::single< double >	maxDelayTime

static std::function< void(std::ostream &)>	statPrinter

Private Types
typedef waitQueues::simple< blockReadRequest >	blockReadRequestQueue

Private Member Functions
bool	expandAttrValue (std::string &value)
	expand the keywords in value, see Expansion of key words in attributes More...

void	attrset (ioHandle &writeHandle, const std::map< std::string, std::string > &attrs)

void	checkAttributes (ioHandle &, const std::map< std::string, std::string > &check_source_attr)

void	hash_worker (checksum::parallel *sum, blockQueue &blocksToHash, blockQueue &hashedBlocks, exceptionList &exceptions)

void	hasher (blockQueue &blocksToHash, blockQueue &hashedBlocks, exceptionList &exceptions)

void	readWorker (inputHandler::base::reader &input, blockReadRequestQueue &blockRequests, blockQueue &freeBlocks, blockQueue &readBlocks, exceptionList &exceptions)

void	reader (inputHandler::base::reader &input, blockQueue &freeBlocks, blockQueue &readBlocks, bool mayParallelize, exceptionList &exceptions)

void	writer (std::unique_ptr< outputHandler::base::writer > &writeHandle, blockQueue &blocksToWrite, blockQueue &writtenBlocks, exceptionList &exceptions)

void	writeWorker (std::unique_ptr< outputHandler::base::writer > &writeHandle, blockQueue &blocksToWrite, blockQueue &writtenBlocks, exceptionList &exceptions)

void	doThreadedCopy (inputHandler::base::reader &input, std::unique_ptr< outputHandler::base::writer > &writeHandle)

void	doUnthreadedCopy (inputHandler::base::reader &input, std::unique_ptr< outputHandler::base::writer > &writeHandle)

void	doBlockSizeSetup (ioHandle &input, ioHandle &output)

bool	makeSymLink (inputHandler::base InputHandler, outputHandler::base OutputHandler)

void	removeFileOrDirectory (inputHandler::base InputHandler, outputHandler::base OutputHandler)

std::string	getBackupSuffix ()

void	init (inputHandler::base *InputHandler)

virtual bool	compare (const base &rhs) const

Static Private Member Functions
static const std::map< std::string, backupModeType > &	getBackupModeNameMap ()

static void	attrdel (ioHandle &, const std::map< std::string, std::string > &attrs)

Static Private Attributes
static options::map< std::string >	success_source_attrs

static options::map< std::string >	failure_source_attrs

static options::map< std::string >	process_source_attrs

static options::map< std::string >	start_source_attrs

static options::map< std::string >	check_source_attrs

static options::map< std::string >	success_dest_attrs

static options::container< std::string >	prohibitive_attrs

static options::container< std::string >	append_attrs

static options::single< std::regex >	appendableFiles

static options::single< bool >	ignoreExisting

static options::single< bool >	changeAttrsOnIgnoreExisting

static options::single< unsigned >	maxRetries

static options::map< std::string >	linkBaseMap

static options::single< std::string >	sparse

static options::withAction< options::single< std::string > >	backupModeName

static backupModeType	backupMode

static options::single< std::string >	backupSuffix

static options::single< bool >	syncWrittenFiles

static options::single< unsigned >	nSumThreads

static options::single< unsigned >	nReadThreads

static options::single< bool >	forceParallelRead

static options::single< unsigned >	nWriteThreads

static options::container< std::string >	logFields

static options::single< std::regex >	sourcePrefixRegex

static options::single< std::regex >	destinationPrefixRegex

static std::map< std::string, clock_type::duration >	advisoryWaitMap

static std::mutex	advisoryWaitMapMutex

static std::mutex	objectCountMutex

static unsigned	objectCount

static std::condition_variable	objectCountCondVar

Friends
class	fileInWork

std::ostream &	operator<< (std::ostream &out, const base &request)

Detailed Description

class for copy requests.

Well, in fact not only copy but also rename, link and delete requests. Objects of this class are created by either from the command line or in The follow-mode. from a command stream, kept in queues and precessed as resources are available. The instances hold all data needed to perform the required operation and stores performance data and a result.

Definition at line 99 of file copyRequest.h.

Member Typedef Documentation

◆ blockReadRequestQueue

typedef waitQueues::simple<blockReadRequest> copyRequest::base::blockReadRequestQueue

private

Definition at line 223 of file copyRequest.h.

Constructor & Destructor Documentation

◆ base() [1/3]

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
	)

constructor for copy,link and remove requests.

Definition at line 1517 of file copyRequest.cpp.

                                                         :
     readInitialStat(std::move(aStat)),
     state(remove ? stateBitType::fileToBeRemoved : stateBitType::fileToBeCopied),
     source(aSource),
     destination(aDestination),
     mapEntry(aMapEntry),
     tInotify(timestamp),
     retries(0) {
     init(InputHandler);
 }

References copyRequest::fileToBeCopied, copyRequest::fileToBeRemoved, init(), and copyRequest::remove.

Here is the call graph for this function:

◆ base() [2/3]

copyRequest::base::base	(	inputHandler::base *	InputHandler,
		const std::string &	aSource,
		const std::string &	aDestination,
		const singleMap &	aMapEntry,
		bool	remove,
		clock_type::time_point	timestamp
	)

constructor for copy,link and remove requests.

Definition at line 1533 of file copyRequest.cpp.

                                                         :
     state(remove ? stateBitType::fileToBeRemoved : stateBitType::fileToBeCopied),
     source(aSource),
     destination(aDestination),
     mapEntry(aMapEntry),
     tInotify(timestamp),
     retries(0) {
     init(InputHandler);
 }

References copyRequest::fileToBeCopied, copyRequest::fileToBeRemoved, init(), and copyRequest::remove.

Here is the call graph for this function:

◆ base() [3/3]

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
	)

constructor for rename requests only.

Definition at line 1547 of file copyRequest.cpp.

                                                         :
     state(stateBitType::fileToBeRenamed),
     source(aSource),
     destination(aDestination),
     moveSource(aMoveSource),
     origSource(aOrigSource),
     mapEntry(aMapEntry),
     tInotify(timestamp),
     retries(0) {
     init(InputHandler);
 }

References copyRequest::fileToBeRenamed, and init().

Here is the call graph for this function:

◆ ~base()

copyRequest::base::~base ( )

virtualnoexcept

Definition at line 1565 of file copyRequest.cpp.

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

References copyRequest::done, copyRequest::fileToBeCopied, and copyRequest::ignore.

Member Function Documentation

◆ addExpectedChecksumResult()

void copyRequest::base::addExpectedChecksumResult	(	const std::string &	checkSumType,
		const std::string &	expectedValue
	)

Definition at line 1601 of file copyRequest.cpp.

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

References checksumCreators, and checksum::base::newFactory().

Here is the call graph for this function:

◆ adviseDelay() [1/2]

bool copyRequest::base::adviseDelay	(	clock_type::duration	dt,
		const std::string &	suffix
	)

static

update the advised delay in the map.

Creates entries in the advisoryWaitMap or updates them. Is to be called when a copy request failed due to changes to the source file, the advised delay will be the time difference between the notofication by The inotify_watch command. and the current time.

Definition at line 1011 of file copyRequest.cpp.

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

References errMsg::emit(), and errMsg::notice.

Referenced by followRequestProvider::enqueueOrAppend().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ adviseDelay() [2/2]

void copyRequest::base::adviseDelay ( clock_type::duration dt = clock_type::duration::zero() )

protected

Definition at line 1043 of file copyRequest.cpp.

                                                        {
     if (dt == decltype(dt)::zero()) {
         dt = clock_type::now() - tInotify;
     }
  
     if (adviseDelay(dt, getSuffix())) {
         if (filesInWorkIterator == fileInWork::filesInWork.end()) {
             return;
         }
         filesInWorkIterator->second.setWaitTime(dt);
     }
 }

References copyRequest::fileInWork::filesInWork.

◆ attrdel()

void copyRequest::base::attrdel	(	ioHandle &	handle,
		const std::map< std::string, std::string > &	attrs
	)

staticprivate

Definition at line 481 of file copyRequest.cpp.

                                                                                          {
     for (const auto& attr : attrs) {
         handle.removeXattr(attr.first);
     }
 }

References ioHandle::removeXattr().

Here is the call graph for this function:

◆ attrset()

void copyRequest::base::attrset	(	ioHandle &	writeHandle,
		const std::map< std::string, std::string > &	attrs
	)

private

Definition at line 446 of file copyRequest.cpp.

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

References errMsg::debug, errMsg::emit(), ioHandle::getXattr(), and ioHandle::setXattr().

Here is the call graph for this function:

◆ changeRequestType()

void copyRequest::base::changeRequestType ( stateBitType newType )

Definition at line 1072 of file copyRequest.cpp.

                                                             {
     fileInWork::typeChanger changer(*this, newType);
 }

◆ checkAttributes()

void copyRequest::base::checkAttributes	(	ioHandle &	handle,
		const std::map< std::string, std::string > &	check_source_attr
	)

private

Definition at line 486 of file copyRequest.cpp.

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

References copyRequest::attributeMismatch, and ioHandle::getXattr().

Here is the call graph for this function:

◆ checkForInstances()

bool copyRequest::base::checkForInstances ( )

static

Definition at line 1427 of file copyRequest.cpp.

                                         {
     std::unique_lock<decltype(objectCountMutex)> lock(objectCountMutex);
     return (objectCount > 0);
 }

◆ compare()

virtual bool copyRequest::base::compare ( const base & rhs ) const

inlineprivatevirtual

Reimplemented in copyRequest::listingRequest.

Definition at line 374 of file copyRequest.h.

                                                     {
             return readInitialStat->size > rhs.readInitialStat->size;
         }

References readInitialStat.

Referenced by copyRequest::base::pointerCompare::operator()().

Here is the caller graph for this function:

◆ doBlockSizeSetup()

void copyRequest::base::doBlockSizeSetup	(	ioHandle &	input,
		ioHandle &	output
	)

private

Definition at line 1404 of file copyRequest.cpp.

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

References ioHandle::getBlockSize(), maxMemoryBlockSize, minMemoryBlockSize, ioHandle::setBlockSize(), userReadBlockSize, and userWriteBlockSize.

Here is the call graph for this function:

◆ doThreadedCopy()

void copyRequest::base::doThreadedCopy	(	inputHandler::base::reader &	input,
		std::unique_ptr< outputHandler::base::writer > &	writeHandle
	)

private

Definition at line 813 of file copyRequest.cpp.

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

References errMsg::emit(), copyRequest::exceptionList::empty(), copyRequest::exceptionList::front(), copyRequest::base::hashCalculator::joinable(), copyRequest::base::writeActor::joinable(), noCopy, ioHandle::parallelizable(), copyRequest::base::hashCalculator::parallelized(), copyRequest::exceptionList::size(), and errMsg::warning.

Here is the call graph for this function:

◆ doUnthreadedCopy()

void copyRequest::base::doUnthreadedCopy	(	inputHandler::base::reader &	input,
		std::unique_ptr< outputHandler::base::writer > &	writeHandle
	)

private

Definition at line 868 of file copyRequest.cpp.

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

References copyRequest::append, block::bufferAt(), errMsg::crit, errMsg::emit(), block::isHole(), noCopy, inputHandler::base::reader::readBlock(), block::size(), timerInst, and outputHandler::base::writer::writeBlock().

Here is the call graph for this function:

◆ expandAttrValue()

bool copyRequest::base::expandAttrValue ( std::string & value )

private

expand the keywords in value, see Expansion of key words in attributes

Returns: true if a keyword was found, false if not

Definition at line 365 of file copyRequest.cpp.

                                                       {
     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...
 }

References f().

Here is the call graph for this function:

◆ getAdvisedDelay()

copyRequest::clock_type::duration copyRequest::base::getAdvisedDelay ( ) const

returns the advised delay

retuns special duration value zero if no delay is advised.

Locking the map is not needed here because read access to containers where no elements are removed (which does not happen here) is thread safe.

Definition at line 1059 of file copyRequest.cpp.

                                                                      {
     auto it = advisoryWaitMap.find(getSuffix());
     if (it == advisoryWaitMap.end()) {
         return clock_type::duration::zero();
     }
     return it->second;
 }

Referenced by followRequestProvider::enqueueOrAppend().

Here is the caller graph for this function:

◆ getBackupModeNameMap()

const std::map< std::string, copyRequest::backupModeType > & copyRequest::base::getBackupModeNameMap ( )

staticprivate

Definition at line 235 of file copyRequest.cpp.

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

References copyRequest::after, copyRequest::before, copyRequest::during, copyRequest::none, and copyRequest::remove.

◆ getBackupSuffix()

std::string copyRequest::base::getBackupSuffix ( )

private

Definition at line 1077 of file copyRequest.cpp.

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

◆ getFilesInWorkIteraror()

decltype(filesInWorkIterator) copyRequest::base::getFilesInWorkIteraror ( ) const

inline

Definition at line 345 of file copyRequest.h.

                                                                      {
             return filesInWorkIterator;
         }

References filesInWorkIterator.

◆ getInitialStat()

const genericStat& copyRequest::base::getInitialStat ( ) const

inline

Definition at line 273 of file copyRequest.h.

                                                   {
             return *readInitialStat;
         }

References readInitialStat.

Referenced by copyRequest::listingRequest::compare().

Here is the caller graph for this function:

◆ getSlotType()

fileInWork::slotTypes copyRequest::base::getSlotType ( ) const

inline

Definition at line 348 of file copyRequest.h.

                                                 {
             if (state & stateBitType::fileToBeCopied ||
                     state & stateBitType::linkToBeMade) {
                 return fileInWork::slotTypes::copy;
             }
             if (state & stateBitType::fileToBeRemoved) {
                 return fileInWork::slotTypes::unlink;
             }
             if (state & stateBitType::fileToBeRenamed) {
                 return fileInWork::slotTypes::move;
             }
             throw std::logic_error("no slot type detected");
         }

References copyRequest::fileInWork::copy, copyRequest::fileToBeCopied, copyRequest::fileToBeRemoved, copyRequest::fileToBeRenamed, copyRequest::linkToBeMade, copyRequest::fileInWork::move, state, and copyRequest::fileInWork::unlink.

Referenced by followRequestProvider::enqueueOrAppend(), copyRequest::fileInWork::inserter::enRegister(), and copyRequest::fileInWork::inserter::markForQueueing().

Here is the caller graph for this function:

◆ getSource()

const std::string& copyRequest::base::getSource ( ) const

inline

Definition at line 292 of file copyRequest.h.

                                            {
             return source;
         };

References source.

Referenced by copyRequest::listingRequest::compare().

Here is the caller graph for this function:

◆ getStatPrinter()

std::function< void(std::ostream &)> & copyRequest::base::getStatPrinter ( )

static

we can't have static functions virtual, so we explicitly use a function to give us the proper print function, which can be changed in derived classes.

Definition at line 1816 of file copyRequest.cpp.

                                                                 {
     return statPrinter;
 }

Referenced by printResults().

Here is the caller graph for this function:

◆ getSuffix() [1/2]

const std::string& copyRequest::base::getSuffix ( ) const

inline

Definition at line 421 of file copyRequest.h.

                                            {
             return (suffix);
         }

References suffix.

Referenced by followRequestProvider::enqueueOrAppend().

Here is the caller graph for this function:

◆ getSuffix() [2/2]

void copyRequest::base::getSuffix	(	const std::string &	path,
		std::string &	suffix
	)

static

Definition at line 1432 of file copyRequest.cpp.

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

◆ hash_worker()

void copyRequest::base::hash_worker	(	checksum::parallel *	sum,
		blockQueue &	blocksToHash,
		blockQueue &	hashedBlocks,
		exceptionList &	exceptions
	)

private

Definition at line 720 of file copyRequest.cpp.

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

References copyRequest::exceptionList::add(), waitQueues::simple< T >::dequeue(), waitQueues::simple< T >::enqueue(), timerInst, and checksum::parallel::update().

Referenced by copyRequest::base::hashCalculator::hashCalculator().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ hasher()

void copyRequest::base::hasher	(	blockQueue &	blocksToHash,
		blockQueue &	hashedBlocks,
		exceptionList &	exceptions
	)

private

Definition at line 738 of file copyRequest.cpp.

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

References copyRequest::exceptionList::add(), checkSums, waitQueues::simple< T >::dequeue(), waitQueues::simple< T >::enqueue(), and timerInst.

Referenced by copyRequest::base::hashCalculator::hashCalculator().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ init()

void copyRequest::base::init ( inputHandler::base * InputHandler )

private

Definition at line 1445 of file copyRequest.cpp.

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

References checksumCreators, dereference, errMsg::emit(), copyRequest::fileToBeCopied, copyRequest::fileToBeRemoved, copyRequest::fileToBeRenamed, pathHandler::getStat(), pathHandler::getXattr(), copyRequest::ignore, copyRequest::linkToBeMade, errMsg::notice, copyRequest::trucated, and copyRequest::vanished.

Referenced by base().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ keywordMap()

std::map< std::string, void(copyRequest::base::*)(std::string &) const > & copyRequest::base::keywordMap ( )

static

Definition at line 246 of file copyRequest.cpp.

                                                                                           {
     static std::map<std::string, void (copyRequest::base::*)(std::string&) const> map;
     return map;
 }

Referenced by copyRequest::base::registerme::registerme().

Here is the caller graph for this function:

◆ makeSymLink()

bool copyRequest::base::makeSymLink	(	inputHandler::base *	InputHandler,
		outputHandler::base *	OutputHandler
	)

private

◆ printResults()

void copyRequest::base::printResults	(	std::ostream &	hashStream,
		std::ostream &	logStream
	)

virtual

Bug:: to little structure, unreadable code

Bug:: was follow before, ugly condition

Reimplemented in copyRequest::listingRequest, and copyRequest::checksumTest.

Definition at line 1618 of file copyRequest.cpp.

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

References copyRequest::append, copyRequest::attributeMismatch, continueOnError, copyRequest::done, errMsg::emit(), copyRequest::failed, copyRequest::fileToBeCopied, copyRequest::fileToBeRemoved, copyRequest::fileToBeRenamed, copyRequest::ignore, errMsg::info, copyRequest::inWork, copyRequest::linkToBeMade, escapism::newEscaper(), noCopy, resultOutput, messageQueue::queue::send(), statPrefix, timerInst, copyRequest::trucated, copyRequest::vanished, verbose, and logstream::warning.

Here is the call graph for this function:

◆ printStats()

void copyRequest::base::printStats ( std::ostream & stream )

static

Definition at line 1822 of file copyRequest.cpp.

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

References timer::anchor::print(), and statPrefix.

Referenced by quietStatPrinter().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ process()

void copyRequest::base::process ( perThreadData & threadData )

virtual

Reimplemented in copyRequest::listingRequest.

Definition at line 1109 of file copyRequest.cpp.

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

References copyRequest::after, copyRequest::append, copyRequest::before, continueOnError, outputHandler::base::writer::doAttributePreservations(), copyRequest::done, copyRequest::during, errMsg::emit(), copyRequest::failed, outputHandler::base::fileChanged, copyRequest::fileToBeCopied, copyRequest::fileToBeRemoved, copyRequest::fileToBeRenamed, outputHandler::base::fileVanished, outputHandler::base::writer::getSize(), copyRequest::ignore, copyRequest::inWork, copyRequest::linkToBeMade, nConcurrentProcesses, copyRequest::noActionNeeded, noCopy, copyRequest::none, errMsg::notice, outputHandler::base::ok, preserve, copyRequest::remove, outputHandler::base::writer::seek(), outputHandler::base::writer::sync(), and copyRequest::vanished.

Here is the call graph for this function:

◆ processQueue()

void copyRequest::base::processQueue	(	simpleQueue &	queue,
		simpleQueue &	resultQueue,
		timedQueue &	delayedRequests
	)

static

Definition at line 1895 of file copyRequest.cpp.

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

References copyRequest::action, errMsg::debug, waitQueues::simple< T >::dequeue(), errMsg::emit(), waitQueues::simple< T >::enqueue(), waitQueues::timed< T, clock_type >::enqueue(), fileRateLimit, copyRequest::fileInWork::eraser::getNext(), stopRequest::processLoop, stopRequest::Requested(), and stopRequest::RequestStop().

Here is the call graph for this function:

◆ reader()

void copyRequest::base::reader	(	inputHandler::base::reader &	input,
		blockQueue &	freeBlocks,
		blockQueue &	readBlocks,
		bool	mayParallelize,
		exceptionList &	exceptions
	)

private

Definition at line 530 of file copyRequest.cpp.

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

References copyRequest::exceptionList::add(), waitQueues::simple< T >::dequeue(), waitQueues::simple< T >::emplace(), waitQueues::simple< T >::enqueue(), nThreads, ioHandle::parallelizable(), inputHandler::base::reader::readBlock(), readWorker(), waitQueues::simple< T >::signalDone(), waitQueues::simple< T >::size(), and timerInst.

Here is the call graph for this function:

◆ readWorker()

void copyRequest::base::readWorker	(	inputHandler::base::reader &	input,
		blockReadRequestQueue &	blockRequests,
		blockQueue &	freeBlocks,
		blockQueue &	readBlocks,
		exceptionList &	exceptions
	)

private

Definition at line 510 of file copyRequest.cpp.

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

References copyRequest::exceptionList::add(), waitQueues::simple< T >::dequeue(), waitQueues::simple< T >::enqueue(), inputHandler::base::reader::readBlockP(), and waitQueues::simple< T >::size().

Referenced by reader().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ registerIterator()

void copyRequest::base::registerIterator ( decltype(filesInWorkIterator) iter )

inline

Definition at line 342 of file copyRequest.h.

                                                                   {
             filesInWorkIterator = iter;
         }

References filesInWorkIterator.

Referenced by copyRequest::fileInWork::inserter::enRegister(), and copyRequest::fileInWork::inserter::markForQueueing().

Here is the caller graph for this function:

◆ removeFileOrDirectory()

void copyRequest::base::removeFileOrDirectory	(	inputHandler::base *	InputHandler,
		outputHandler::base *	OutputHandler
	)

private

Definition at line 914 of file copyRequest.cpp.

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

References copyRequest::done, copyRequest::failed, pathHandler::pathExists(), outputHandler::base::remove(), and copyRequest::vanished.

Here is the call graph for this function:

◆ resetStats()

void copyRequest::base::resetStats ( )

static

Definition at line 1852 of file copyRequest.cpp.

                                  {
     tTotal2Stat.reset();
     tTotalStat.reset();
     tInotifyStat.reset();
     tPipeStat.reset();
     tWaitStat.reset();
     tCopyStat.reset();
     tEnqueueStat.reset();
     bytesStat.reset();
     speedStat.reset();
     bytesStat.reset();
     timer::anchor::reset();
 }

References timer::anchor::reset().

Referenced by printResults().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ retry()

bool copyRequest::base::retry	(	std::unique_ptr< base > &	request,
		timedQueue &	delayedRequests
	)

static

Definition at line 1866 of file copyRequest.cpp.

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

References enumAsBitmask< T >::clear(), errMsg::debug, errMsg::emit(), waitQueues::timed< T, clock_type >::enqueue(), errorMessage, copyRequest::failed, copyRequest::fileToBeRemoved, retries, source, state, tWorkDone, and tWorkStart.

Here is the call graph for this function:

◆ setStatPrinter()

void copyRequest::base::setStatPrinter ( void(*)(std::ostream &) f )

static

Definition at line 1819 of file copyRequest.cpp.

                                                           {
     statPrinter = f;
 }

References f().

Referenced by policyRunRequestProvider::processSource().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ waitForAllInstancesGone()

void copyRequest::base::waitForAllInstancesGone ( )

static

Definition at line 1421 of file copyRequest.cpp.

                                               {
     std::unique_lock<decltype(objectCountMutex)> lock(objectCountMutex);
     while (objectCount > 0) {
         objectCountCondVar.wait(lock);
     }
 }

◆ writer()

void copyRequest::base::writer	(	std::unique_ptr< outputHandler::base::writer > &	writeHandle,
		blockQueue &	blocksToWrite,
		blockQueue &	writtenBlocks,
		exceptionList &	exceptions
	)

private

Definition at line 592 of file copyRequest.cpp.

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

References copyRequest::exceptionList::add(), waitQueues::simple< T >::dequeue(), waitQueues::simple< T >::enqueue(), and outputHandler::base::writer::writeBlock().

Referenced by copyRequest::base::writeActor::writeActor().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ writeWorker()

void copyRequest::base::writeWorker	(	std::unique_ptr< outputHandler::base::writer > &	writeHandle,
		blockQueue &	blocksToWrite,
		blockQueue &	writtenBlocks,
		exceptionList &	exceptions
	)

private

Definition at line 610 of file copyRequest.cpp.

                                                                {
     try {
         while (auto b = blocksToWrite.dequeue()) {
             writeHandle->writeBlockP(*b);
             writtenBlocks.enqueue(b);
         }
     } catch (...) {
         exceptions.add(std::current_exception());
     }
 }

References copyRequest::exceptionList::add(), waitQueues::simple< T >::dequeue(), waitQueues::simple< T >::enqueue(), and outputHandler::base::writer::writeBlockP().

Referenced by copyRequest::base::writeActor::writeActor().

Here is the call graph for this function:

Here is the caller graph for this function:

Friends And Related Function Documentation

◆ fileInWork

friend class fileInWork

friend

Definition at line 100 of file copyRequest.h.

◆ operator<<

std::ostream& operator<<	(	std::ostream &	out,
		const base &	request
	)

friend

Definition at line 1952 of file copyRequest.cpp.

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

Member Data Documentation

◆ advisoryWaitMap

std::map<std::string, clock_type::duration> copyRequest::base::advisoryWaitMap

staticprivate

Definition at line 336 of file copyRequest.h.

◆ advisoryWaitMapMutex

std::mutex copyRequest::base::advisoryWaitMapMutex

staticprivate

Definition at line 337 of file copyRequest.h.

◆ append_attrs

options::container<std::string> copyRequest::base::append_attrs

staticprivate

Definition at line 112 of file copyRequest.h.

◆ appendableFiles

options::single<std::regex> copyRequest::base::appendableFiles

staticprivate

Definition at line 113 of file copyRequest.h.

◆ backupMode

backupModeType copyRequest::base::backupMode

staticprivate

Definition at line 120 of file copyRequest.h.

◆ backupModeName

options::withAction<options::single<std::string> > copyRequest::base::backupModeName

staticprivate

Definition at line 119 of file copyRequest.h.

◆ backupSuffix

options::single<std::string> copyRequest::base::backupSuffix

staticprivate

Definition at line 121 of file copyRequest.h.

◆ bytesStat

statCollector::typed< decltype(readInitialStat->size) > copyRequest::base::bytesStat

staticprotected

Definition at line 314 of file copyRequest.h.

◆ changeAttrsOnIgnoreExisting

options::single<bool> copyRequest::base::changeAttrsOnIgnoreExisting

staticprivate

Definition at line 115 of file copyRequest.h.

◆ check_source_attrs

options::map<std::string> copyRequest::base::check_source_attrs

staticprivate

Definition at line 109 of file copyRequest.h.

◆ checkSums

std::list<checksum::base*> copyRequest::base::checkSums

protected

Definition at line 277 of file copyRequest.h.

Referenced by copyRequest::base::hashCalculator::hashCalculator(), hasher(), and copyRequest::checksumTest::printResults().

◆ destination

const std::string copyRequest::base::destination

protected

Definition at line 285 of file copyRequest.h.

Referenced by copyRequest::operator<<().

◆ destinationPrefixRegex

options::single<std::regex> copyRequest::base::destinationPrefixRegex

staticprivate

Definition at line 133 of file copyRequest.h.

◆ errorMessage

std::string copyRequest::base::errorMessage

protected

Definition at line 283 of file copyRequest.h.

Referenced by retry().

◆ failure_source_attrs

options::map<std::string> copyRequest::base::failure_source_attrs

staticprivate

Definition at line 106 of file copyRequest.h.

◆ filesInWorkIterator

decltype(fileInWork::filesInWork.begin()) copyRequest::base::filesInWorkIterator = fileInWork::filesInWork.end()

Definition at line 341 of file copyRequest.h.

Referenced by getFilesInWorkIteraror(), copyRequest::operator<<(), and registerIterator().

◆ forceParallelRead

options::single<bool> copyRequest::base::forceParallelRead

staticprivate

Definition at line 129 of file copyRequest.h.

◆ ignoreExisting

options::single<bool> copyRequest::base::ignoreExisting

staticprivate

Definition at line 114 of file copyRequest.h.

◆ linkBaseMap

options::map<std::string> copyRequest::base::linkBaseMap

staticprivate

Definition at line 117 of file copyRequest.h.

◆ logFields

options::container<std::string> copyRequest::base::logFields

staticprivate

Definition at line 131 of file copyRequest.h.

◆ mapEntry

const singleMap& copyRequest::base::mapEntry

protected

Definition at line 288 of file copyRequest.h.

◆ maxDelayTime

options::single<double> copyRequest::base::maxDelayTime

staticprotected

Definition at line 328 of file copyRequest.h.

◆ maxMemoryBlockSize

options::single<size_t> copyRequest::base::maxMemoryBlockSize

static

Definition at line 297 of file copyRequest.h.

Referenced by doBlockSizeSetup().

◆ maxRetries

options::single<unsigned> copyRequest::base::maxRetries

staticprivate

Definition at line 116 of file copyRequest.h.

◆ memoryBlockSize

size_t copyRequest::base::memoryBlockSize

protected

Definition at line 300 of file copyRequest.h.

Referenced by copyRequest::base::hashCalculator::hashCalculator(), and copyRequest::base::writeActor::writeActor().

◆ minMemoryBlockSize

options::single<size_t> copyRequest::base::minMemoryBlockSize

static

Definition at line 298 of file copyRequest.h.

Referenced by doBlockSizeSetup().

◆ moveSource

const std::string copyRequest::base::moveSource

protected

Definition at line 286 of file copyRequest.h.

◆ nConcurrentProcesses

std::atomic<unsigned int> copyRequest::base::nConcurrentProcesses

staticprotected

Definition at line 318 of file copyRequest.h.

Referenced by process(), and copyRequest::base::writeActor::writeActor().

◆ nReadThreads

options::single<unsigned> copyRequest::base::nReadThreads

staticprivate

Definition at line 128 of file copyRequest.h.

◆ nSumThreads

options::single<unsigned> copyRequest::base::nSumThreads

staticprivate

Definition at line 127 of file copyRequest.h.

Referenced by copyRequest::base::hashCalculator::hashCalculator().

◆ nWriteThreads

options::single<unsigned> copyRequest::base::nWriteThreads

staticprivate

Definition at line 130 of file copyRequest.h.

Referenced by copyRequest::base::writeActor::writeActor().

◆ objectCount

unsigned copyRequest::base::objectCount

staticprivate

Definition at line 371 of file copyRequest.h.

◆ objectCountCondVar

std::condition_variable copyRequest::base::objectCountCondVar

staticprivate

Definition at line 372 of file copyRequest.h.

◆ objectCountMutex

std::mutex copyRequest::base::objectCountMutex

staticprivate

Definition at line 370 of file copyRequest.h.

◆ origSource

const std::string copyRequest::base::origSource

protected

Definition at line 287 of file copyRequest.h.

◆ perFileThreads

options::single<bool> copyRequest::base::perFileThreads

static

Definition at line 125 of file copyRequest.h.

◆ prefix

std::string copyRequest::base::prefix

protected

Definition at line 290 of file copyRequest.h.

◆ process_source_attrs

options::map<std::string> copyRequest::base::process_source_attrs

staticprivate

Definition at line 107 of file copyRequest.h.

◆ processMultiplicities

std::vector<std::atomic<unsigned int> >* copyRequest::base::processMultiplicities

static

Definition at line 321 of file copyRequest.h.

◆ prohibitive_attrs

options::container<std::string> copyRequest::base::prohibitive_attrs

staticprivate

Definition at line 111 of file copyRequest.h.

◆ readInitialStat

std::unique_ptr<const genericStat> copyRequest::base::readInitialStat

protected

Definition at line 271 of file copyRequest.h.

Referenced by copyRequest::listingRequest::compare(), compare(), getInitialStat(), copyRequest::base::hashCalculator::hashCalculator(), copyRequest::listingRequest::printResults(), and copyRequest::base::writeActor::writeActor().

◆ retries

unsigned copyRequest::base::retries

protected

Definition at line 323 of file copyRequest.h.

Referenced by retry().

◆ setAttributesAfterClose

options::single<bool> copyRequest::base::setAttributesAfterClose

static

Definition at line 124 of file copyRequest.h.

Referenced by outputHandler::dcap::dcap().

◆ source

const std::string copyRequest::base::source

protected

Definition at line 284 of file copyRequest.h.

Referenced by copyRequest::listingRequest::compare(), getSource(), copyRequest::operator<<(), copyRequest::checksumTest::printResults(), copyRequest::listingRequest::printResults(), and retry().

◆ sourcePrefixRegex

options::single<std::regex> copyRequest::base::sourcePrefixRegex

staticprivate

Definition at line 132 of file copyRequest.h.

◆ sparse

options::single<std::string> copyRequest::base::sparse

staticprivate

Definition at line 118 of file copyRequest.h.

◆ speedStat

statCollector::typed<double> copyRequest::base::speedStat

staticprotected

Definition at line 315 of file copyRequest.h.

◆ start_source_attrs

options::map<std::string> copyRequest::base::start_source_attrs

staticprivate

Definition at line 108 of file copyRequest.h.

◆ state

stateType copyRequest::base::state

protected

Definition at line 282 of file copyRequest.h.

Referenced by getSlotType(), copyRequest::base::hashCalculator::hashCalculator(), copyRequest::operator<<(), retry(), and copyRequest::fileInWork::typeChanger::typeChanger().

◆ statPrinter

std::function<void(std::ostream&)> copyRequest::base::statPrinter

staticprotected

Definition at line 330 of file copyRequest.h.

Referenced by copyRequest::checksumTest::prepareStatPrint(), and copyRequest::listingRequest::prepareStatPrint().

◆ success_dest_attrs

options::map<std::string> copyRequest::base::success_dest_attrs

staticprivate

Definition at line 110 of file copyRequest.h.

◆ success_source_attrs

options::map<std::string> copyRequest::base::success_source_attrs

staticprivate

Definition at line 105 of file copyRequest.h.

◆ suffix

std::string copyRequest::base::suffix

protected

Definition at line 289 of file copyRequest.h.

Referenced by getSuffix().

◆ syncWrittenFiles

options::single<bool> copyRequest::base::syncWrittenFiles

staticprivate

Definition at line 122 of file copyRequest.h.

◆ tCopyStat

statCollector::typed< decltype(tWorkDone - tWorkStart) > copyRequest::base::tCopyStat

staticprotected

Definition at line 313 of file copyRequest.h.

◆ tEnqueue

clock_type::time_point copyRequest::base::tEnqueue

protected

Definition at line 304 of file copyRequest.h.

◆ tEnqueueStat

statCollector::typed< copyRequest::clock_type::duration > copyRequest::base::tEnqueueStat

static

Definition at line 320 of file copyRequest.h.

Referenced by followInotifyWatchRequestProvider::followStream().

◆ threadData

perThreadData* copyRequest::base::threadData

protected

Definition at line 281 of file copyRequest.h.

◆ tInotify

clock_type::time_point copyRequest::base::tInotify

protected

Definition at line 303 of file copyRequest.h.

Referenced by copyRequest::operator<<().

◆ tInotifyStat

statCollector::typed< decltype(tWorkDone - tInotify) > copyRequest::base::tInotifyStat

staticprotected

Definition at line 310 of file copyRequest.h.

◆ tPipeStat

statCollector::typed< decltype(tEnqueue - tInotify) > copyRequest::base::tPipeStat

staticprotected

Definition at line 311 of file copyRequest.h.

◆ tProgramStart

clock_type::time_point copyRequest::base::tProgramStart

staticprotected

Definition at line 316 of file copyRequest.h.

◆ tTotal2Stat

statCollector::typed< decltype(tWorkDone - tInotify) > copyRequest::base::tTotal2Stat

staticprotected

Definition at line 309 of file copyRequest.h.

◆ tTotalStat

statCollector::typed< decltype(tWorkDone - tInotify) > copyRequest::base::tTotalStat

staticprotected

Definition at line 308 of file copyRequest.h.

◆ tWaitStat

statCollector::typed< decltype(tWorkStart - tEnqueue) > copyRequest::base::tWaitStat

staticprotected

Definition at line 312 of file copyRequest.h.

◆ tWorkDone

clock_type::time_point copyRequest::base::tWorkDone

protected

Definition at line 306 of file copyRequest.h.

Referenced by retry().

◆ tWorkStart

clock_type::time_point copyRequest::base::tWorkStart

protected

Definition at line 305 of file copyRequest.h.

Referenced by retry().

◆ userReadBlockSize

options::single<size_t> copyRequest::base::userReadBlockSize

static

Definition at line 296 of file copyRequest.h.

Referenced by doBlockSizeSetup().

◆ userWriteBlockSize

options::single<size_t> copyRequest::base::userWriteBlockSize

static

Definition at line 294 of file copyRequest.h.

Referenced by doBlockSizeSetup().

The documentation for this class was generated from the following files:

Classes

Public Member Functions

Static Public Member Functions

Public Attributes

Static Public Attributes

Protected Member Functions

Protected Attributes

Static Protected Attributes

Private Types

Private Member Functions

Static Private Member Functions

Static Private Attributes

Friends

Detailed Description

Member Typedef Documentation

◆ blockReadRequestQueue

Constructor & Destructor Documentation

◆ base() [1/3]

◆ base() [2/3]

◆ base() [3/3]

◆ ~base()

Member Function Documentation

◆ addExpectedChecksumResult()

◆ adviseDelay() [1/2]

◆ adviseDelay() [2/2]

◆ attrdel()

◆ attrset()

◆ changeRequestType()

◆ checkAttributes()

◆ checkForInstances()

◆ compare()

◆ doBlockSizeSetup()

◆ doThreadedCopy()

◆ doUnthreadedCopy()

◆ expandAttrValue()

◆ getAdvisedDelay()

◆ getBackupModeNameMap()

◆ getBackupSuffix()

◆ getFilesInWorkIteraror()

◆ getInitialStat()

◆ getSlotType()

◆ getSource()

◆ getStatPrinter()

◆ getSuffix() [1/2]

◆ getSuffix() [2/2]

◆ hash_worker()

◆ hasher()

◆ init()

◆ keywordMap()

◆ makeSymLink()

◆ printResults()

◆ printStats()

◆ process()

◆ processQueue()

◆ reader()

◆ readWorker()

◆ registerIterator()

◆ removeFileOrDirectory()

◆ resetStats()

◆ retry()

◆ setStatPrinter()

◆ waitForAllInstancesGone()

◆ writer()

◆ writeWorker()

Friends And Related Function Documentation

◆ fileInWork

◆ operator<<

Member Data Documentation

◆ advisoryWaitMap

◆ advisoryWaitMapMutex

◆ append_attrs

◆ appendableFiles

◆ backupMode

◆ backupModeName

◆ backupSuffix

◆ bytesStat

◆ changeAttrsOnIgnoreExisting

◆ check_source_attrs

◆ checkSums

◆ destination