ParallelSearch/ParallelSearch.cpp

#include "ParallelSearch.hpp"

#include <thread>

ParallelSearch::SearchJob::SearchJob(Iterator begin, Iterator end, const std::string& pattern) : begin_(begin), end_(end), pattern_(pattern)
{
    result_.reserve(INITIAL_RESERVE);
}

void ParallelSearch::SearchJob::execute()
{
    for (Iterator itr = begin_; itr != end_; itr++)
    {
        if (0 == itr->compare(0, pattern_.size(), pattern_))
        {
            // found a match, save the index
            result_.push_back(itr);
        }
    }
}

ParallelSearch::ParallelSearch(unsigned workerCount, Iterator begin, Iterator end, const std::string& pattern)
{
    std::size_t totalCount = end - begin;
    std::size_t bucketSize = totalCount / workerCount;
    std::size_t remains = totalCount % workerCount;

    std::size_t offset = 0;
    while (offset < totalCount)
    {
        // compute [first; last( iterators for sub-range
        std::size_t size = bucketSize + remains;
        Iterator first = begin + offset;
        Iterator last = begin + offset + size;

        // add appropriate search job
        jobList_.push_back(SearchJob(first, last, pattern));

        // increment offset
        offset += size;
        // clear remains (it has been added to first bucket)
        remains = 0;
    }
};

void ParallelSearch::run()
{
    // start thread for each job
    std::vector<std::thread> threads;
    for (SearchJob& job : jobList_)
    {
        threads.push_back(std::thread(&SearchJob::execute, &job));
    }

    // wait for all threads to join
    for (std::thread& th : threads)
    {
        th.join();
    }

    // collect the results
    for (SearchJob& job : jobList_)
    {
        result_.insert(result_.end(), job.get_result().begin(), job.get_result().end());
    }
}