4.1

1 files changed, 600 insertions, 0 deletions

diff --git a/lib/parallel_scan.cpp b/lib/parallel_scan.cpp
new file mode 100644
index 00000000..29d87ee7
--- /dev/null
+++ b/lib/parallel_scan.cpp
@@ -0,0 +1,600 @@
+// **************************************************************************
+// * This file is part of the FreeFileSync project. It is distributed under *
+// * GNU General Public License: http://www.gnu.org/licenses/gpl.html       *
+// * Copyright (C) 2008-2011 ZenJu (zhnmju123 AT gmx.de)                    *
+// **************************************************************************
+
+#include "parallel_scan.h"
+#include <boost/detail/atomic_count.hpp>
+#include "db_file.h"
+#include "lock_holder.h"
+#include <zen/file_traverser.h>
+#include <zen/file_error.h>
+#include <wx+/string_conv.h>
+#include <zen/thread.h> //includes <boost/thread.hpp>
+#include <zen/scope_guard.h>
+
+using namespace zen;
+
+
+#ifndef BOOST_HAS_THREADS
+#error just some paranoia check...
+#endif
+
+
+namespace
+{
+/*
+#ifdef FFS_WIN
+
+struct DiskInfo
+{
+    DiskInfo() :
+        driveType(DRIVE_UNKNOWN),
+        diskID(-1) {}
+
+    UINT driveType;
+    int diskID; // -1 if id could not be determined, this one is filled if driveType == DRIVE_FIXED or DRIVE_REMOVABLE;
+};
+
+inline
+bool operator<(const DiskInfo& lhs, const DiskInfo& rhs)
+{
+    if (lhs.driveType != rhs.driveType)
+        return lhs.driveType < rhs.driveType;
+
+    if (lhs.diskID < 0 || rhs.diskID < 0)
+        return false;
+    //consider "same", reason: one volume may be uniquely associated with one disk, while the other volume is associated to the same disk AND another one!
+    //volume <-> disk is 0..N:1..N
+
+    return lhs.diskID < rhs.diskID ;
+}
+
+
+DiskInfo retrieveDiskInfo(const Zstring& pathName)
+{
+    std::vector<wchar_t> volName(std::max(pathName.size(), static_cast<size_t>(10000)));
+
+    DiskInfo output;
+
+    //full pathName need not yet exist!
+    if (!::GetVolumePathName(pathName.c_str(),  //__in   LPCTSTR lpszFileName,
+                             &volName[0],       //__out  LPTSTR lpszVolumePathName,
+                             static_cast<DWORD>(volName.size()))) //__in   DWORD cchBufferLength
+        return output;
+
+    const Zstring rootPathName = &volName[0];
+
+    output.driveType = ::GetDriveType(rootPathName.c_str());
+
+    if (output.driveType == DRIVE_NO_ROOT_DIR) //these two should be the same error category
+        output.driveType = DRIVE_UNKNOWN;
+
+    if (output.driveType != DRIVE_FIXED && output.driveType != DRIVE_REMOVABLE)
+        return output; //no reason to get disk ID
+
+    //go and find disk id:
+
+    //format into form: "\\.\C:"
+    Zstring volnameFmt = rootPathName;
+    if (endsWith(volnameFmt, FILE_NAME_SEPARATOR))
+        volnameFmt.resize(volnameFmt.size() - 1);
+    volnameFmt = L"\\\\.\\" + volnameFmt;
+
+    HANDLE hVolume = ::CreateFile(volnameFmt.c_str(),
+                                  0,
+                                  FILE_SHARE_READ | FILE_SHARE_WRITE,
+                                  0,
+                                  OPEN_EXISTING,
+                                  0,
+                                  NULL);
+    if (hVolume == INVALID_HANDLE_VALUE)
+        return output;
+    ZEN_ON_BLOCK_EXIT(::CloseHandle(hVolume));
+
+    std::vector<char> buffer(sizeof(VOLUME_DISK_EXTENTS) + sizeof(DISK_EXTENT)); //reserve buffer for at most one disk! call below will then fail if volume spans multiple disks!
+
+    DWORD bytesReturned = 0;
+    if (!::DeviceIoControl(hVolume,                              // handle to device
+                           IOCTL_VOLUME_GET_VOLUME_DISK_EXTENTS, // dwIoControlCode
+                           NULL,                                 // lpInBuffer
+                           0,                                    // nInBufferSize
+                           &buffer[0],                           // output buffer
+                           static_cast<DWORD>(buffer.size()),    // size of output buffer
+                           &bytesReturned,                       // number of bytes returned
+                           NULL))                                // OVERLAPPED structure
+        return output;
+
+    const VOLUME_DISK_EXTENTS& volDisks = *reinterpret_cast<VOLUME_DISK_EXTENTS*>(&buffer[0]);
+
+    if (volDisks.NumberOfDiskExtents != 1)
+        return output;
+
+    output.diskID = volDisks.Extents[0].DiskNumber;
+
+    return output;
+}
+
+#elif defined FFS_LINUX
+#endif
+*/
+
+/*
+PERF NOTE
+
+--------------------------------------------
+|Testcase: Reading from two different disks|
+--------------------------------------------
+Windows 7:
+            1st(unbuffered) |2nd (OS buffered)
+			----------------------------------
+1 Thread:          57s      |        8s
+2 Threads:         39s      |        7s
+
+--------------------------------------------------
+|Testcase: Reading two directories from same disk|
+--------------------------------------------------
+Windows 7:                                        Windows XP:
+            1st(unbuffered) |2nd (OS buffered)                   1st(unbuffered) |2nd (OS buffered)
+			----------------------------------                   ----------------------------------
+1 Thread:          41s      |        13s             1 Thread:          45s      |        13s
+2 Threads:         42s      |        11s             2 Threads:         38s      |         8s
+
+=> Traversing does not take any advantage of file locality so that even multiple threads operating on the same disk impose no performance overhead!
+*/
+
+
+std::vector<std::set<DirectoryKey>> separateByDistinctDisk(const std::set<DirectoryKey>& dirkeys)
+{
+    //see perf note: use one thread per dirkey:
+    typedef std::map<int, std::set<DirectoryKey>> DiskKeyMapping;
+    DiskKeyMapping tmp;
+    int index = 0;
+    std::for_each(dirkeys.begin(), dirkeys.end(),
+    [&](const DirectoryKey& key) { tmp[++index].insert(key); });
+
+    /*
+    	//use one thread per physical disk:
+    	typedef std::map<DiskInfo, std::set<DirectoryKey>> DiskKeyMapping;
+        DiskKeyMapping tmp;
+        std::for_each(dirkeys.begin(), dirkeys.end(),
+        [&](const DirectoryKey& key) { tmp[retrieveDiskInfo(key.dirnameFull_)].insert(key); });
+    */
+    std::vector<std::set<DirectoryKey>> buckets;
+    std::transform(tmp.begin(), tmp.end(), std::back_inserter(buckets),
+    [&](const DiskKeyMapping::value_type& diskToKey) { return diskToKey.second; });
+    return buckets;
+}
+
+//------------------------------------------------------------------------------------------
+typedef Zbase<wchar_t, StorageRefCountThreadSafe> BasicWString; //thread safe string class for UI texts
+
+
+class AsyncCallback
+{
+public:
+    AsyncCallback() :
+        notifyingThreadID(-1),
+        textScanning(_("Scanning:")),
+        itemsScanned(0),
+        activeWorker(0) {}
+
+    FillBufferCallback::HandleError reportError(const std::wstring& msg) //blocking call: context of worker thread
+    {
+        boost::unique_lock<boost::mutex> dummy(lockErrorMsg);
+        while (!errorMsg.empty() || errorResponse.get())
+            conditionCanReportError.timed_wait(dummy, boost::posix_time::milliseconds(50)); //interruption point!
+
+        errorMsg = BasicWString(msg);
+
+        while (!errorResponse.get())
+            conditionGotResponse.timed_wait(dummy, boost::posix_time::milliseconds(50)); //interruption point!
+
+        FillBufferCallback::HandleError rv = *errorResponse;
+
+        errorMsg.clear();
+        errorResponse.reset();
+
+        //dummy.unlock();
+        conditionCanReportError.notify_one();
+
+        return rv;
+    }
+
+    void processErrors(FillBufferCallback& callback) //context of main thread, call repreatedly
+    {
+        boost::lock_guard<boost::mutex> dummy(lockErrorMsg);
+        if (!errorMsg.empty() && !errorResponse.get())
+        {
+            FillBufferCallback::HandleError rv = callback.reportError(cvrtString<std::wstring>(errorMsg)); //throw!
+            errorResponse.reset(new FillBufferCallback::HandleError(rv));
+
+            //dummy.unlock();
+            conditionGotResponse.notify_one();
+        }
+    }
+
+    void setNotifyingThread(int threadID) { notifyingThreadID = threadID; } //context of main thread
+
+    void reportCurrentFile(const Zstring& filename, int threadID) //context of worker thread
+    {
+        if (threadID != notifyingThreadID) return; //only one thread may report status
+
+        boost::lock_guard<boost::mutex> dummy(lockCurrentStatus);
+        currentFile = filename;
+        currentStatus.clear();
+    }
+
+    void reportCurrentStatus(const std::wstring& status, int threadID) //context of worker thread
+    {
+        if (threadID != notifyingThreadID) return; //only one thread may report status
+
+        boost::lock_guard<boost::mutex> dummy(lockCurrentStatus);
+        currentFile.clear();
+        currentStatus = BasicWString(status); //we cannot assume std::wstring to be thread safe (yet)!
+    }
+
+    std::wstring getCurrentStatus() //context of main thread, call repreatedly
+    {
+        std::wstring filename;
+        std::wstring statusMsg;
+        {
+            boost::lock_guard<boost::mutex> dummy(lockCurrentStatus);
+            if (!currentFile.empty())
+                filename = utf8CvrtTo<std::wstring>(currentFile);
+            else if (!currentStatus.empty())
+                statusMsg = cvrtString<std::wstring>(currentStatus);
+        }
+
+        if (!filename.empty())
+        {
+            std::wstring statusText = cvrtString<std::wstring>(textScanning);
+            const long activeCount = activeWorker;
+            if (activeCount >= 2)
+            {
+                statusText += L" " + _P("[1 Thread]", "[%x Threads]", activeCount);
+                replace(statusText, L"%x", toString<std::wstring>(activeCount));
+            }
+            statusText += std::wstring(L" \n") + L'\"' + filename + L'\"';
+            return statusText;
+        }
+        else
+            return statusMsg;
+    }
+
+    void incItemsScanned() { ++itemsScanned; }
+    long getItemsScanned() const { return itemsScanned; }
+
+    void incActiveWorker() { ++activeWorker; }
+    void decActiveWorker() { --activeWorker; }
+    long getActiveWorker() const { return activeWorker; }
+
+private:
+    //---- error handling ----
+    boost::mutex lockErrorMsg;
+    boost::condition_variable conditionCanReportError;
+    boost::condition_variable conditionGotResponse;
+    BasicWString errorMsg;
+    std::unique_ptr<FillBufferCallback::HandleError> errorResponse;
+
+    //---- status updates ----
+    volatile int notifyingThreadID; //theoretically racy, but there is nothing that could go wrong...
+    //CAVEAT: do NOT use boost::thread::id as long as this showstopper exists: https://svn.boost.org/trac/boost/ticket/5754
+    boost::mutex lockCurrentStatus; //use a different lock for current file: continue traversing while some thread may process an error
+    Zstring currentFile;        //only one of these two is filled at a time!
+    BasicWString currentStatus; //
+
+    const BasicWString textScanning; //this one is (currently) not shared and could be made a std::wstring, but we stay consistent and use thread-safe variables in this class only!
+
+    //---- status updates II (lock free) ----
+    boost::detail::atomic_count itemsScanned;
+    boost::detail::atomic_count activeWorker;
+};
+//-------------------------------------------------------------------------------------------------
+
+class DirCallback;
+
+struct TraverserShared
+{
+public:
+    TraverserShared(int threadID,
+                    SymLinkHandling handleSymlinks,
+                    const HardFilter::FilterRef& filter,
+                    std::set<Zstring>& failedReads,
+                    AsyncCallback& acb) :
+        handleSymlinks_(handleSymlinks),
+        filterInstance(filter),
+        failedReads_(failedReads),
+        acb_(acb),
+        threadID_(threadID) {}
+
+    typedef std::shared_ptr<DirCallback> CallbackPointer;
+    std::vector<CallbackPointer> callBackBox;  //collection of callback pointers to handle ownership
+
+    const SymLinkHandling handleSymlinks_;
+    const HardFilter::FilterRef filterInstance; //always bound!
+
+    std::set<Zstring>& failedReads_; //relative postfixed names of directories that could not be read (empty for root)
+
+    AsyncCallback& acb_;
+    int threadID_;
+};
+
+
+class DirCallback : public zen::TraverseCallback
+{
+public:
+    DirCallback(TraverserShared& config,
+                const Zstring& relNameParentPf, //postfixed with FILE_NAME_SEPARATOR!
+                DirContainer& output) :
+        cfg(config),
+        relNameParentPf_(relNameParentPf),
+        output_(output) {}
+
+    virtual void         onFile   (const Zchar* shortName, const Zstring& fullName, const FileInfo& details);
+    virtual void         onSymlink(const Zchar* shortName, const Zstring& fullName, const SymlinkInfo& details);
+    virtual ReturnValDir onDir    (const Zchar* shortName, const Zstring& fullName);
+    virtual HandleError  onError  (const std::wstring& errorText);
+
+private:
+    TraverserShared& cfg;
+    const Zstring relNameParentPf_;
+    DirContainer& output_;
+};
+
+
+void DirCallback::onFile(const Zchar* shortName, const Zstring& fullName, const FileInfo& details)
+{
+    boost::this_thread::interruption_point();
+
+    const Zstring fileNameShort = shortName;
+
+    //do not list the database file(s) sync.ffs_db, sync.x64.ffs_db, etc. or lock files
+    if (endsWith(fileNameShort, SYNC_DB_FILE_ENDING) ||
+        endsWith(fileNameShort, LOCK_FILE_ENDING))
+        return;
+
+    //update status information no matter whether object is excluded or not!
+    cfg.acb_.reportCurrentFile(fullName, cfg.threadID_);
+
+    //------------------------------------------------------------------------------------
+    //apply filter before processing (use relative name!)
+    if (!cfg.filterInstance->passFileFilter(relNameParentPf_ + fileNameShort))
+        return;
+
+    //    std::string fileId = details.fileSize >=  1024 * 1024U ?
+    //                                           util::retrieveFileID(fullName) :
+    //                                          std::string();
+    /*
+    Perf test Windows 7, SSD, 350k files, 50k dirs, files > 1MB: 7000
+    	regular:            6.9s
+    	ID per file:       43.9s
+    	ID per file > 1MB:  7.2s
+    	ID per dir:         8.4s
+
+    	Linux: retrieveFileID takes about 50% longer in VM! (avoidable because of redundant stat() call!)
+    */
+
+    output_.addSubFile(fileNameShort, FileDescriptor(details.lastWriteTimeRaw, details.fileSize));
+
+    cfg.acb_.incItemsScanned(); //add 1 element to the progress indicator
+}
+
+
+void DirCallback::onSymlink(const Zchar* shortName, const Zstring& fullName, const SymlinkInfo& details)
+{
+    boost::this_thread::interruption_point();
+
+    if (cfg.handleSymlinks_ == SYMLINK_IGNORE)
+        return;
+
+    //update status information no matter whether object is excluded or not!
+    cfg.acb_.reportCurrentFile(fullName, cfg.threadID_);
+
+    //------------------------------------------------------------------------------------
+    const Zstring& relName = relNameParentPf_ + shortName;
+
+    //apply filter before processing (use relative name!)
+    if (!cfg.filterInstance->passFileFilter(relName)) //always use file filter: Link type may not be "stable" on Linux!
+        return;
+
+    output_.addSubLink(shortName, LinkDescriptor(details.lastWriteTimeRaw, details.targetPath, details.dirLink ? LinkDescriptor::TYPE_DIR : LinkDescriptor::TYPE_FILE));
+
+    cfg.acb_.incItemsScanned(); //add 1 element to the progress indicator
+}
+
+
+TraverseCallback::ReturnValDir DirCallback::onDir(const Zchar* shortName, const Zstring& fullName)
+{
+    boost::this_thread::interruption_point();
+
+    //update status information no matter whether object is excluded or not!
+    cfg.acb_.reportCurrentFile(fullName, cfg.threadID_);
+
+    //------------------------------------------------------------------------------------
+    const Zstring& relName = relNameParentPf_ + shortName;
+
+    //apply filter before processing (use relative name!)
+    bool subObjMightMatch = true;
+    const bool passFilter = cfg.filterInstance->passDirFilter(relName, &subObjMightMatch);
+    if (!passFilter && !subObjMightMatch)
+        return Int2Type<ReturnValDir::TRAVERSING_DIR_IGNORE>(); //do NOT traverse subdirs
+    //else: attention! ensure directory filtering is applied later to exclude actually filtered directories
+
+    DirContainer& subDir = output_.addSubDir(shortName);
+    if (passFilter)
+        cfg.acb_.incItemsScanned(); //add 1 element to the progress indicator
+
+    TraverserShared::CallbackPointer subDirCallback = std::make_shared<DirCallback>(cfg, relName + FILE_NAME_SEPARATOR, subDir);
+    cfg.callBackBox.push_back(subDirCallback); //handle lifetime
+    return ReturnValDir(Int2Type<ReturnValDir::TRAVERSING_DIR_CONTINUE>(), *subDirCallback);
+}
+
+
+DirCallback::HandleError DirCallback::onError(const std::wstring& errorText)
+{
+    switch (cfg.acb_.reportError(errorText))
+    {
+        case FillBufferCallback::TRAV_ERROR_IGNORE:
+            cfg.failedReads_.insert(relNameParentPf_);
+            return TRAV_ERROR_IGNORE;
+
+        case FillBufferCallback::TRAV_ERROR_RETRY:
+            return TRAV_ERROR_RETRY;
+    }
+
+    assert(false);
+    return TRAV_ERROR_IGNORE;
+}
+
+
+#ifdef FFS_WIN
+class DstHackCallbackImpl : public DstHackCallback
+{
+public:
+    DstHackCallbackImpl(AsyncCallback& acb, int threadID) :
+        acb_(acb),
+        threadID_(threadID),
+        textApplyingDstHack(toZ(replaceCpy(_("Encoding extended time information: %x"), L"%x", L"\n\"%x\""))) {}
+
+private:
+    virtual void requestUiRefresh(const Zstring& filename) //applying DST hack imposes significant one-time performance drawback => callback to inform user
+    {
+        const Zstring statusText = replaceCpy(textApplyingDstHack, Zstr("%x"), filename);
+        acb_.reportCurrentStatus(utf8CvrtTo<std::wstring>(statusText), threadID_);
+    }
+
+    AsyncCallback& acb_;
+    int threadID_;
+    const Zstring textApplyingDstHack;
+};
+#endif
+//------------------------------------------------------------------------------------------
+
+
+class WorkerThread
+{
+public:
+    WorkerThread(int threadID,
+                 const std::shared_ptr<AsyncCallback>& acb,
+                 const std::vector<std::pair<DirectoryKey, DirectoryValue*>>& workload) :
+        threadID_(threadID),
+        acb_(acb),
+        workload_(workload) {}
+
+    void operator()() //thread entry
+    {
+        acb_->incActiveWorker();
+        ZEN_ON_BLOCK_EXIT(acb_->decActiveWorker(););
+
+        std::for_each(workload_.begin(), workload_.end(),
+                      [&](std::pair<DirectoryKey, DirectoryValue*>& item)
+        {
+            const Zstring& directoryName = item.first.dirnameFull_;
+            DirectoryValue& dirVal = *item.second;
+
+            acb_->reportCurrentFile(directoryName, threadID_); //just in case first directory access is blocking
+
+            TraverserShared travCfg(threadID_,
+                                    item.first.handleSymlinks_, //shared by all(!) instances of DirCallback while traversing a folder hierarchy
+                                    item.first.filter_,
+                                    dirVal.failedReads,
+                                    *acb_);
+
+            DirCallback traverser(travCfg,
+                                  Zstring(),
+                                  dirVal.dirCont);
+
+            bool followSymlinks = false;
+            switch (item.first.handleSymlinks_)
+            {
+                case SYMLINK_IGNORE:
+                    followSymlinks = false; //=> symlinks will be reported via onSymlink() where they are excluded
+                    break;
+                case SYMLINK_USE_DIRECTLY:
+                    followSymlinks = false;
+                    break;
+                case SYMLINK_FOLLOW_LINK:
+                    followSymlinks = true;
+                    break;
+            }
+
+            DstHackCallback* dstCallbackPtr = NULL;
+#ifdef FFS_WIN
+            DstHackCallbackImpl dstCallback(*acb_, threadID_);
+            dstCallbackPtr = &dstCallback;
+#endif
+
+            //get all files and folders from directoryPostfixed (and subdirectories)
+            traverseFolder(directoryName, followSymlinks, traverser, dstCallbackPtr); //exceptions may be thrown!
+        });
+    }
+
+private:
+    int threadID_;
+    std::shared_ptr<AsyncCallback> acb_;
+    std::vector<std::pair<DirectoryKey, DirectoryValue*>> workload_;
+};
+}
+
+
+void zen::fillBuffer(const std::set<DirectoryKey>& keysToRead, //in
+                     std::map<DirectoryKey, DirectoryValue>& buf, //out
+                     FillBufferCallback& callback,
+                     size_t updateInterval)
+{
+    buf.clear();
+
+    std::vector<std::set<DirectoryKey>> buckets = separateByDistinctDisk(keysToRead); //one bucket per physical device
+
+    std::vector<boost::thread> worker; //note: GCC doesn't allow to construct an array of empty threads since they would be initialized by const boost::thread&
+    worker.reserve(buckets.size());
+
+    zen::ScopeGuard guardWorker = zen::makeGuard([&]()
+    {
+        std::for_each(worker.begin(), worker.end(), std::mem_fun_ref(&boost::thread::interrupt)); //interrupt all at once, then join
+        std::for_each(worker.begin(), worker.end(), std::mem_fun_ref(&boost::thread::join));
+    });
+
+    std::shared_ptr<AsyncCallback> acb = std::make_shared<AsyncCallback>();
+
+    //init worker threads
+    for (auto iter = buckets.begin(); iter != buckets.end(); ++iter)
+    {
+        int threadID = iter - buckets.begin();
+        const std::set<DirectoryKey>& bucket = *iter;
+
+        std::vector<std::pair<DirectoryKey, DirectoryValue*>> workload;
+        std::for_each(bucket.begin(), bucket.end(),
+                      [&](const DirectoryKey& key)
+        {
+            auto rv = buf.insert(std::make_pair(key, DirectoryValue()));
+            assert(rv.second);
+            workload.push_back(std::make_pair(key, &rv.first->second));
+        });
+
+        worker.push_back(boost::thread(WorkerThread(threadID, acb, workload)));
+    }
+
+    //wait until done
+    for (auto iter = worker.begin(); iter != worker.end(); ++iter)
+    {
+        boost::thread& wt = *iter;
+        int threadID = iter - worker.begin();
+
+        acb->setNotifyingThread(threadID); //process info messages of first (active) thread only
+
+        do
+        {
+            //update status
+            callback.reportStatus(acb->getCurrentStatus(), acb->getItemsScanned()); //throw!
+
+            //process errors
+            acb->processErrors(callback);
+        }
+        while (!wt.timed_join(boost::posix_time::milliseconds(updateInterval)));
+    }
+
+    guardWorker.dismiss();
+}