2
\$\begingroup\$

Following on from my program to identify files that have common file sizes, I have developed an application that takes the output from that process and then compares each of the files that are the same size in order to identify duplicates which are then written to a '.matched' output file in the form <FullPathFile>|<FullPathOtherFile>, with one line per match.

So, given this input file 'filesToProcess.txt'

29    c:\Program Files\Git\mingw64\libexec\git-core\mergetools\gvimdiff
29    c:\Program Files\Git\mingw64\libexec\git-core\mergetools\gvimdiff2
29    c:\Program Files\Git\mingw64\libexec\git-core\mergetools\gvimdiff3
113   c:\Program Files\Git\mingw64\lib\tcl8.6\tzdata\Etc\GMT-7
113   c:\Program Files\Git\mingw64\lib\tcl8.6\tzdata\Etc\GMT-8
113   c:\Program Files\Git\mingw64\lib\tcl8.6\tzdata\Etc\GMT-9

If the processing identified that 'gvimdiff' and 'gvimdiff2' were the same, the output file 'filesToProcess.txt.matched' would look like this:

c:\Program Files\Git\mingw64\libexec\git-core\mergetools\gvimdiff|c:\Program Files\Git\mingw64\libexec\git-core\mergetools\gvimdiff2

The process is being run across a network connection and there are a significant number of files so it takes a long time to complete. With this in mind it is designed to be restartable (a '.processed' file is used keep track of sizes that have been processed so that they can be ignored on future runs). Pressing a key will shut the application down once the current sizes have been processed.

There are two worker threads that perform the actual file comparisons. Since the operations are largely IO bound (network and disk) there isn't a significant improvement by this but it does allow the comparison code to run in parallel and it was largely done as an exercise in using std::thread.

I'm not overly keen on the Logger implementation I've written, but it was simple, functional and I'm not sure if there's a standardised approach that's been introduced by any of the C++ standards.

Feedback on any aspect of the code is welcome.

DuplicateFileChecker.cpp

#include "stdafx.h"

#include "ILogger.h"
#include "ConsoleLogger.h"
#include "WorkItem.h"
#include "WorkQueue.h"
#include "ProcessedSizeTracker.h"
#include "WorkerThread.h"

/*

** DuplicateFileChecker.  Expects a single filename which specifies the input file for processing.

Input file consists of single line records ordered 'size', format:

<size>\t<fileName>

Outputs two files, the most important of which is "<inputFile>.matched".  This file contains
a list of matched files in the format:

<firstFile>|<matchedFile>

Also outputs a file to help track progress in the event of a failure / early termination.  Each
row in the processed file "<inputFile>.processed" simply consists of a "size" to indicate files of
that size have been processed correctly.


*/

static void SaveMatches(std::ofstream out, WorkerThread *threads, unsigned threadCount);
static void BuildWorkQueue(std::istream &in,  ProcessedSizeTracker &processedTracker, unsigned &maxWorkItemSize, WorkQueue<WorkItem> &workQueue);

ILogger &Logger = ConsoleLogger{};

int main(int argc, char *argv[])
{
    if (argc != 2) {
        std::cout << "Usage:" << argv[0] << " <process file>";
        return EXIT_FAILURE;
    }

    std::string inputFileName = argv[1];

    ProcessedSizeTracker processedTracker{};
    processedTracker.ReadProcesedState(std::ifstream(inputFileName + ".processed"));

    unsigned maxMatches=0;
    WorkQueue<WorkItem> workQueue{};

    BuildWorkQueue(std::ifstream(inputFileName), processedTracker, maxMatches, workQueue);

    WorkerThread threads[]{ {workQueue, processedTracker, maxMatches},{ workQueue, processedTracker, maxMatches} };

    uintmax_t printThreshold = workQueue.GetItemsRemaining() + 1;

    while (auto itemsToWork = workQueue.GetItemsRemaining()) {
        if (itemsToWork < printThreshold) {
            std::cout << "Items Remaining: " << itemsToWork << std::endl;
            printThreshold = itemsToWork > 100 ?  itemsToWork - 100 : 0;
        }
        if (_kbhit()) {
            std::cout << "Shutting down..." << std::endl;
            workQueue.Flush();
            break;
        }
        else {
            std::this_thread::sleep_for(std::chrono::seconds(1));
        }
    }

    for (auto &thread : threads) {
        thread.WaitUntilComplete();
    }

    processedTracker.WriteProcessedState(std::ofstream(inputFileName + ".processed", std::ios_base::trunc));

    SaveMatches(std::ofstream(inputFileName + ".matched", std::ios_base::app), threads, sizeof(threads)/sizeof(WorkerThread));
}

static void SaveMatches(std::ofstream out, WorkerThread *threads, unsigned threadCount) {
    for (unsigned worker = 0; worker < threadCount; worker++) {
        for (auto &match : threads[worker].GetMatches()) {
            out << std::get<0>(match) << "|" << std::get<1>(match) << std::endl;
        }
    }
}

static void BuildWorkQueue(std::istream &in, ProcessedSizeTracker  &processedTracker, unsigned &maxWorkItemSize, WorkQueue<WorkItem> &workQueue) {
    std::string currentLine;
    size_t fileCount = 0;
    size_t totalFileSize = 0;
    unsigned workItems = 0;
    uintmax_t maxBuffer = 0;

    maxWorkItemSize = 0;

    std::shared_ptr<std::forward_list<std::string>> items = std::make_shared<std::forward_list<std::string>>();
    uintmax_t prevFileSize = 0;

    while (std::getline(in, currentLine)) {

        auto tabPosition = currentLine.find('\t');
        if (tabPosition == std::string::npos) {
            throw std::runtime_error("Invalid file format");
        }
        auto fileSize = stoul(currentLine.substr(0, tabPosition));
        std::string filePath = currentLine.substr(tabPosition + 1);

        if (processedTracker.HasSizeBeenProcessed(fileSize)) continue;

        if (prevFileSize != fileSize) {
            if (maxWorkItemSize < workItems) maxWorkItemSize = workItems;
            if (workItems * prevFileSize > maxBuffer) maxBuffer = workItems * prevFileSize;
            workItems = 0;

            prevFileSize = fileSize;
            items = std::make_shared<std::forward_list<std::string>>();
            workQueue.PushWorkItem(std::make_unique<WorkItem>(fileSize, items));
        }

        items->emplace_front(filePath);

        totalFileSize += fileSize;
        fileCount++;
        workItems++;
    }

    Logger.Informational() << "Files to process: " << fileCount << " Total Size: " << totalFileSize << " Biggest work item: " << maxWorkItemSize << " Largest required buffer: " << maxBuffer << std::endl;
}

WorkItem.h

#pragma once

class WorkItem {
public:
    WorkItem(size_t size, std::shared_ptr<std::forward_list<std::string>> files) :
        m_files(std::move(files)),
        m_size(size)
    {
    }

    const std::shared_ptr<std::forward_list<std::string>> GetFiles() const {
        return m_files;
    }

    auto GetFileSize() const { return m_size; }

private:
    const size_t m_size;
    const std::shared_ptr<std::forward_list<std::string>> m_files;
};

WorkQueue.h

#pragma once

template <class TWorkItem>
class WorkQueue {
public:
    void PushWorkItem(std::unique_ptr<TWorkItem> workItem) {
        std::lock_guard<std::mutex> guard(m_lock);

        m_workItems.push_front(std::move(workItem));
        m_size++;
    }

    std::unique_ptr<TWorkItem> GetNextWorkItem() {
        std::lock_guard<std::mutex> guard(m_lock);

        if (m_workItems.empty()) { return nullptr; }

        std::unique_ptr<TWorkItem> workItem = std::move(m_workItems.front());
        m_workItems.pop_front();
        m_size--;

        return workItem;
    }

    void Flush() {
        std::lock_guard<std::mutex> guard(m_lock);
        m_workItems.clear();
        m_size = 0;
    }

    unsigned GetItemsRemaining() const { return m_size; }

    WorkQueue() = default;
    WorkQueue(const WorkQueue &) = delete;
    WorkQueue &operator=(const WorkQueue&) = delete;


private:
    std::forward_list<std::unique_ptr<TWorkItem>> m_workItems{};
    std::mutex m_lock;
    std::atomic_uint m_size { 0 };
};

ProcessedSizeTracker.h

#pragma once

class ProcessedSizeTracker
{
public:
    void ProcessedSize(size_t size);
    bool HasSizeBeenProcessed(size_t size);

    void WriteProcessedState(std::ostream &out);
    void ReadProcesedState(std::istream &in);


private:
    std::set<size_t> m_processed{};
    std::mutex m_lock;
};

ProcessedSizeChecker.cpp

#include "stdafx.h"
#include "ProcessedSizeTracker.h"


void ProcessedSizeTracker::ProcessedSize(size_t size) {
    std::lock_guard<std::mutex> guard(m_lock);
    m_processed.emplace(size);
}

void ProcessedSizeTracker::WriteProcessedState(std::ostream &out) {
    std::lock_guard<std::mutex> guard(m_lock);

    for (auto item : m_processed) {
        out << item << std::endl;
    }
}

void ProcessedSizeTracker::ReadProcesedState(std::istream &in) {
    std::lock_guard<std::mutex> guard(m_lock);

    std::string line;
    while (std::getline(in, line)) {
        m_processed.emplace(stoul(line));
    }
}

bool ProcessedSizeTracker::HasSizeBeenProcessed(size_t size)  {
    std::lock_guard<std::mutex> guard(m_lock);
    return m_processed.find(size) != m_processed.end();
}

WorkerThread.h

#pragma once

class WorkerThread {
    typedef std::forward_list<std::tuple<std::string, std::string>> MatchedFilesList;

    struct FileBuffer {
        char *buffer;
        std::string fileName;
    };

public:
    WorkerThread(const WorkerThread &) = delete;
    WorkerThread &operator=(const WorkerThread&) = delete;

    WorkerThread(WorkQueue<WorkItem> &queueToWork, ProcessedSizeTracker &processedTracker, unsigned maxGroupSize);
    ~WorkerThread();

    void WaitUntilComplete();
    const MatchedFilesList &GetMatches() const;

private:
    void WorkQueueUntilEmpty();

    ProcessedSizeTracker &m_processedTracker;
    WorkQueue<WorkItem> &m_workQueue;
    std::thread *m_workerThread;
    unsigned m_maxGroupSize;
    FileBuffer *m_fileBuffers = 0;
    MatchedFilesList  m_matchedFiles;
};

WorkerThread.cpp

#include "stdafx.h"

#include "ILogger.h"
#include "WorkItem.h"
#include "WorkQueue.h"
#include "ProcessedSizeTracker.h"
#include "WorkerThread.h"

WorkerThread::WorkerThread(WorkQueue<WorkItem> &queueToWork, ProcessedSizeTracker &processedTracker, unsigned maxGroupSize) : m_workQueue(queueToWork)
        , m_processedTracker(processedTracker)
        , m_maxGroupSize(maxGroupSize)  {
    m_fileBuffers = new FileBuffer[m_maxGroupSize];
    ::memset(m_fileBuffers, 0, sizeof(FileBuffer) * m_maxGroupSize);

    m_workerThread = new std::thread(&WorkerThread::WorkQueueUntilEmpty, this);
}

WorkerThread::~WorkerThread() {
    for (size_t bufferIndex = 0; bufferIndex < m_maxGroupSize; bufferIndex++) {
        if (m_fileBuffers[bufferIndex].buffer) {
            delete[] m_fileBuffers[bufferIndex].buffer;
            m_fileBuffers[bufferIndex].buffer = nullptr;
        }
    }
    delete[] m_fileBuffers;

    delete m_workerThread;
}

void WorkerThread::WaitUntilComplete() {
    m_workerThread->join();
}


const WorkerThread::MatchedFilesList &WorkerThread::GetMatches() const {
    return m_matchedFiles;
}

/*
* WorkQueueUntilEmpty - Thread function, terminates once _workQueue is empty
* 
* The work queue is assumed to be ordered by FileSize, with the largest
* files at the head of the queue.  The processing below takes advantage of
* this to allocate buffers as and when needed, so that the minimum memory
* can be allocated without having to worry about constantly freeing and
* allocating memory.
*
*/

void WorkerThread::WorkQueueUntilEmpty() {
    unsigned workItemsProcessed{ 0 };
    unsigned filesProcessed{ 0 };
    unsigned filesMatched{ 0 };
    unsigned workItemsFailed{ 0 };
    char *readBuffer{ nullptr };

    while (auto workItem = m_workQueue.GetNextWorkItem()) {
        workItemsProcessed++;
        try {
            unsigned buffersUsedThisWorkItem{ 0 };
            for (const auto &item : *(workItem->GetFiles())) {
                bool matchedExistingFile = false;
                std::ifstream currentFile(item, std::ios::binary);

                if (!readBuffer) {
                    readBuffer = new char[workItem->GetFileSize()];
                }

                currentFile.read(readBuffer, workItem->GetFileSize());

                if (!currentFile) {
                    throw "Failed to process file " + item;
                }

                for (unsigned i = 0; i < buffersUsedThisWorkItem; i++) {
                    if (::memcmp(readBuffer, m_fileBuffers[i].buffer, workItem->GetFileSize()) == 0) {
                        matchedExistingFile = true;
                        filesMatched++;
                        m_matchedFiles.emplace_front(std::tuple<std::string, std::string>{m_fileBuffers[i].fileName, item});
                        break;
                    }
                }

                if (!matchedExistingFile) {
                    if (m_fileBuffers[buffersUsedThisWorkItem].buffer) {
                        ::memcpy(m_fileBuffers[buffersUsedThisWorkItem].buffer, readBuffer, workItem->GetFileSize());
                    }
                    else {
                        m_fileBuffers[buffersUsedThisWorkItem].buffer = readBuffer;
                        readBuffer = nullptr;
                    }
                    m_fileBuffers[buffersUsedThisWorkItem].fileName = item;

                    buffersUsedThisWorkItem++;
                }

                filesProcessed++;
            }

            m_processedTracker.ProcessedSize(workItem->GetFileSize());
        }
        catch (const std::exception& ex) {
            Logger.Error() << "Exception: " << ex.what() << std::endl;
            workItemsFailed++;
        }
        catch (const std::string& message) {
            Logger.Error() << "Exception: " << message << std::endl;
            workItemsFailed++;
        }
        catch (...) {
            Logger.Error() << "Unknown Exception!" << std::endl;
            workItemsFailed++;
        }
    }
    Logger.Informational() << "Work Items Processed: " << workItemsProcessed - workItemsFailed << std::endl;
    Logger.Informational() << "Files Processed: " << filesProcessed << std::endl;
    Logger.Informational() << "Failed Processing: " << workItemsFailed << std::endl;
    Logger.Informational() << "Matches Found: " << filesMatched << std::endl;
}

ILogger.h

#pragma once

class ILogger {
public:
    virtual std::ostream & Informational()const = 0;
    virtual std::ostream & Error() const = 0;

    virtual ~ILogger() {};
};

extern ILogger &Logger;

ConsoleLogger.h

#pragma once

class ConsoleLogger : public ILogger {
public:
    virtual std::ostream & Informational()const {
        return std::cout;
    }

    virtual std::ostream & Error() const {
        return std::cerr;
    }
};

stdafx.h

#pragma once

#include <iostream>
#include <fstream>
#include <map>
#include <set>
#include <string>
#include <sstream>
#include <stdlib.h>
#include <stdexcept>
#include <forward_list>
#include <memory>
#include <thread>
#include <mutex>
#include <atomic>
#include <conio.h>
\$\endgroup\$
  • \$\begingroup\$ A good way to see if files are duplicates is to use md5sum (or md5). The hash of the file should be unique enough to determine if they are identical. \$\endgroup\$ – Martin York Jul 18 '16 at 15:21

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.