copyRequest.h

Go to the documentation of this file.

#ifndef __copyRequest_h__
#define __copyRequest_h__
 
/*
  ewmscp: copy program with extended functionality
    Copyright (C) 2018  Juergen Hannappel
 
    This program is free software: you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation, either version 3 of the License, or
    (at your option) any later version.
 
    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.
 
    You should have received a copy of the GNU General Public License
    along with this program.  If not, see <https://www.gnu.org/licenses/>.
*/
 
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <deque>
#include <iomanip>
#include <iostream>
#include <list>
#include <mutex>
#include <regex>
#include <sstream>
#include <system_error>
#include <thread>
#include <unistd.h>
#include <unordered_set>
#include <vector>
 
#ifdef WithMagic
#include <magic.h>
#endif
 
#include "block.h"
#include "copyRequestTypes.h"
#include "ewmscp.h"
#include "fileInWork.h"
#include "throttle.h"
#include <Options.h>
#include <OptionsRegex.h>
#include <statCollector.h>
#include <waitQueues.h>
 
typedef std::pair<std::string, std::string> singleMap;
 
namespace copyRequest {
    enum class backupModeType {
        none,
        before,
        during,
        after,
        remove
    };
 
    typedef waitQueues::simple<block> blockQueue;
 
    class perThreadData {
      public:
        inputHandler::base* InputHandler;
        outputHandler::base* OutputHandler;
        blockQueue freeBlocks;
        #ifdef WithMagic
        magic_t magicCookie;
        #endif
        perThreadData();
    };
 
 
    class exceptionList {
        mutable std::mutex listMutex;
        std::forward_list<std::exception_ptr> list;
    public:
        void add(std::exception_ptr e);
        bool empty() const;
        size_t size() const;
        std::exception_ptr front();
    };
 
 
    class base {
        friend class fileInWork;
      private:
        static const std::map<std::string, backupModeType>& getBackupModeNameMap();
 
 
        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;
      public:
        static options::single<bool> setAttributesAfterClose;
        static options::single<bool> perFileThreads;
      private:
        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;
        #ifdef WithMagic
        static options::single<bool> doMagic;
        #endif
 
      public:
        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;
        #ifdef WithMagic
        void kw_magic(std::string& value) const;
        #endif
 
        static std::map<std::string, void (base::*)(std::string&) const>& keywordMap();
        class registerme {
          public:
            registerme(const std::string& keyword,
                       void (base::*fkt)(std::string&) const) {
                keywordMap()[keyword] = fkt;
            }
        };
      private:
 
        bool expandAttrValue(std::string& value);
        void attrset(ioHandle& writeHandle,
                     const std::map<std::string, std::string>& attrs);
        static void attrdel(ioHandle&,
                            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);
        class hashCalculator {
            std::vector<std::thread> workers;
            checksum::parallel* parallelSum;
            blockQueue& outputQueue;
          public:
            hashCalculator(blockQueue& blocksToHash,
                           blockQueue& hashedBlocks,
                           base* request,
                           bool mayParallelize,
                           exceptionList& exceptions);
            ~hashCalculator() noexcept(false);
            bool joinable() const;
            bool parallelized() const;
        };
        #ifdef WithMagic
        void innerMagic(const block& b);
        void magician(blockQueue& inputQueue,
                      blockQueue& outputQueue,
                      exceptionList& exceptions);
        class magicCalculator {
            std::thread worker;
            blockQueue& outputQueue;
          public:
            magicCalculator(blockQueue& inputQueue,
                            blockQueue& aOutputQueue,
                            base* request,
                            exceptionList& exceptions);
            ~magicCalculator() noexcept(false);
            bool joinable() const;
        };
        #endif
        class blockReadRequest {
          public:
            size_t size;
            off_t offset;
            blockReadRequest(size_t aSize, off_t aOffset): size(aSize), offset(aOffset) {};
        };
        typedef waitQueues::simple<blockReadRequest> blockReadRequestQueue;
 
        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);
        class writeActor {
            std::vector<std::thread> workers;
            blockQueue& outputQueue;
          public:
            writeActor(std::unique_ptr<outputHandler::base::writer>& writeHandle,
                       blockQueue& blocksToWrite,
                       blockQueue& writtenBlocks,
                       bool mayParallelize,
                       base* request,
                       exceptionList& exceptions);
            ~writeActor() noexcept(false);
            bool joinable() const;
        };
 
        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();
      protected:
        std::unique_ptr<const genericStat> readInitialStat;
      public:
        const genericStat& getInitialStat() const {
            return *readInitialStat;
        }
      protected:
        std::list<checksum::base*> checkSums;
        #ifdef WithMagic
        std::string fileMagic;
        #endif
        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;
      public:
        const std::string& getSource() const {
            return source;
        };
        static options::single<size_t> userWriteBlockSize;
        static options::single<size_t> userReadBlockSize;
        static options::single<size_t> maxMemoryBlockSize;
        static options::single<size_t> minMemoryBlockSize;
      protected:
        size_t memoryBlockSize;
 
 
        clock_type::time_point tInotify;
        clock_type::time_point tEnqueue;
        clock_type::time_point tWorkStart;
        clock_type::time_point tWorkDone;
 
        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;
      public:
        static statCollector::typed < copyRequest::clock_type::duration > tEnqueueStat;
        static std::vector<std::atomic<unsigned int>>* processMultiplicities;
      protected:
        unsigned retries;
      public:
        static bool adviseDelay(clock_type::duration dt, const std::string& suffix);
      protected:
        void adviseDelay(clock_type::duration dt = clock_type::duration::zero());
        static options::single<double> maxDelayTime;
 
        static std::function<void(std::ostream&)> statPrinter;
      public:
        static void printStats(std::ostream& stream);
        static void resetStats();
 
      private:
        static std::map<std::string, clock_type::duration> advisoryWaitMap;
        static std::mutex advisoryWaitMapMutex;
      public:
        clock_type::duration getAdvisedDelay() const; 
 
        decltype(fileInWork::filesInWork.begin()) filesInWorkIterator = fileInWork::filesInWork.end();
        void registerIterator(decltype(filesInWorkIterator) iter) {
            filesInWorkIterator = iter;
        }
        decltype(filesInWorkIterator) getFilesInWorkIteraror() const {
            return filesInWorkIterator;
        }
        fileInWork::slotTypes getSlotType() const {
            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");
        }
 
        void changeRequestType(stateBitType newType);
 
 
        static bool retry(std::unique_ptr<base>& request,
                          timedQueue& delayedRequests);
        static void waitForAllInstancesGone();
        static bool checkForInstances();
      private:
        static std::mutex objectCountMutex;
        static unsigned objectCount;
        static std::condition_variable objectCountCondVar;
        void init(inputHandler::base* InputHandler);
        virtual bool compare(const base& rhs) const {
            return readInitialStat->size > rhs.readInitialStat->size;
        }
      public:
 
        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); 
        base(inputHandler::base* InputHandler,
             const std::string& aSource,
             const std::string& aDestination,
             const singleMap& aMapEntry,
             bool remove,
             clock_type::time_point timestamp); 
        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); 
        virtual ~base() noexcept(false);
        virtual void process(perThreadData& threadData);
        virtual void printResults(std::ostream& hashStream,
                                  std::ostream& logStream);
        static std::function<void(std::ostream&)>& getStatPrinter();
        static void setStatPrinter(void(*f)(std::ostream&));
        class pointerCompare {
          public:
            bool operator () (const base* lhs,
                              const base* rhs) const {
                if (lhs == nullptr) {
                    return false;
                }
                if (rhs == nullptr) {
                    return true;
                }
                return lhs->compare(*rhs);
            }
        };
        static void processQueue(simpleQueue& queue,
                                 simpleQueue& resultQueue,
                                 timedQueue& delayedRequests);
        const std::string& getSuffix() const {
            return (suffix);
        }
        static void getSuffix(const std::string& path,
                              std::string& suffix);
        void addExpectedChecksumResult(const std::string& checkSumType,
                                       const std::string& expectedValue);
 
        friend std::ostream& operator<<(std::ostream& out, const base& request);
    };
}; // end of namespace copyRequest
#endif