2
\$\begingroup\$

Problem statement: Consider a scenario where a vector of very small tasks, each encapsulated within a class 'Task' with a thread-safe method Task::process(), needs to be efficiently processed in batches. We need to design an API processTasks (std::vector taskVec)

  • It can be called multiple times.
  • It needs to process each task in vector by calling "call task.process()" A generic ThreadPool can be used to submit these tasks on each call of 'processTasks'when total runtime to process taskVec is large enough to justify the overhead of ThreadPool/parallelization. For small tasks, ThreadPool/parallelization overhead becomes significant. I have tried writing a customized threadpool below to optimize it for small tasks.

Lets look at why a generic threadpool is not ideal solution for us ?

  • A generic threadpool is optimized for throughput
  • need to support hetrogenious tasks
  • can't make any assumption about usage pattern

How to customize it for our use-case ?

Since our priorities and requirements are different so lets try to create a tailor made solution for us.

  • Optimize for Latency rather than throughput (Do you know why?)
  • Homogenious Tasks: Same function needs to be called on each element of vector. (SFMF or Single Function Multiple Data)
  • Usage Pattern: We get work in batches. So we get some work and then all threads will finish doing it and finish. Only then we start next batch of work.

Here is my Implementation for review.

template<typename TASK>
class HTaskVecThreadPool {
public:
    explicit HTaskVecThreadPool(size_t numThreads) : myThreadCount(numThreads), myIsRunning(numThreads, 0) {
        for (size_t tIndex = 0; tIndex < myThreadCount; tIndex++)
            myThreads.emplace_back(&HTaskVecThreadPool::thread_worker, this, tIndex);
    }
    ~HTaskVecThreadPool() { stopThreads(); }
    void submit(std::vector<TASK>& task) {
        assert(!myTaskVec && numRunningWorkers == 0);
        myTaskVec = &task;
        numRunningWorkers.store(myThreadCount);
        std::for_each(myIsRunning.begin(), myIsRunning.end(), [](auto& v) { v = true; });
    }
    void waitForAll() {
        while (numRunningWorkers.load()) {}
        myTaskVec = nullptr;
    }
private:
    void stopThreads() {
        myFinish = true;
        std::for_each(myThreads.begin(), myThreads.end(), std::mem_fn(&std::thread::join));
    }
    TASK& task(size_t index) { return (*myTaskVec)[index]; }
    void processGroup(size_t groupNum, size_t numGroups) {
        size_t numTasks = myTaskVec->size();
        size_t groupSize = numTasks / numGroups;
        size_t from = (groupNum - 1) * groupSize;
        size_t toe = (groupNum == numGroups) ? numTasks : groupNum * groupSize;
        for(size_t i = from; i < toe; i++) { task(i).process(); }
    }

    struct AlignedBool {
        alignas(std::hardware_destructive_interference_size) int value;
        AlignedBool(int v) : value(v) {};
    };

    std::vector<TASK>* myTaskVec = nullptr;
    const size_t myThreadCount;
    std::vector<std::thread> myThreads;
    std::vector<AlignedBool> myIsRunning;
    std::atomic<unsigned char> numRunningWorkers {0};
    std::atomic<bool> myFinish {false};

    void thread_worker(int threadIndex) {
        auto& hasWorkToDo = myIsRunning[threadIndex].value;
        while (true) {
            while (!hasWorkToDo) { if(myFinish.load()) return; }
            size_t groupNum = 1 + threadIndex;
            processGroup(groupNum, myThreadCount);
            hasWorkToDo = false;
            numRunningWorkers--;
        }
    }
};

Edit: I have updated above implementation by fixing some synchronisation issues.

template<typename TASK>
class HTaskVecThreadPool {
public:
    explicit HTaskVecThreadPool(size_t numThreads) : myThreadIdVsHasWork(numThreads, 0) {
        for (size_t tIndex = 0; tIndex < numThreads; tIndex++)
            myThreads.emplace_back(&HTaskVecThreadPool::thread_worker, this, tIndex);
    }
    ~HTaskVecThreadPool() { stopThreads(); }
    [[nodiscard]] bool submit(std::vector<TASK>& taskVec) {
        if(myNumThreadsWithWork.load(std::memory_order_acquire)) return false;  // Already busy, Can't accept new work
        myTaskVec = &taskVec;
        std::for_each(myThreadIdVsHasWork.begin(), myThreadIdVsHasWork.end(), [](auto& v) { v = true; });
        myNumThreadsWithWork.store(numThreads(), std::memory_order_release);
        return true;
    }
    void waitForAll() {
        while (myNumThreadsWithWork.load(std::memory_order_acquire)) {}
        myTaskVec = nullptr;
    }
private:
    void stopThreads() {
        myFinish = true;
        std::for_each(myThreads.begin(), myThreads.end(), std::mem_fn(&std::thread::join));
    }
    size_t numThreads() { return myThreads.size(); }
    TASK& task(size_t index) { return (*myTaskVec)[index]; }
    struct AlignedBool {
        alignas(std::hardware_destructive_interference_size) int value;
        AlignedBool(int v) : value(v) {};
    };

    std::vector<std::thread> myThreads;
    std::atomic<unsigned char> myNumThreadsWithWork {0};    // Used to check if all threads has finished their work and Pool is ready to accept next batch of work
    std::vector<AlignedBool> myThreadIdVsHasWork;           // Non-atomic read/write synchronized with 'myNumThreadsWithWork'
    std::vector<TASK>* myTaskVec = nullptr;                 // Non-atomic read/write synchronized with 'myNumThreadsWithWork'
    std::atomic<bool> myFinish {false};                     // Used to shut down all workers

    void thread_worker(int threadIndex) {
        auto &hasWorkToDo = myThreadIdVsHasWork[threadIndex].value;
        while (true) {
            if (!myNumThreadsWithWork.load(std::memory_order_acquire)) {   // Synchronizes with myNumThreadsWithWork.store() from 'submit'
                if (myFinish.load())
                    return;
                continue;
            }

            if (hasWorkToDo) {
                size_t groupNum = 1 + threadIndex;
                processGroup(groupNum, numThreads());
                hasWorkToDo = false;
                myNumThreadsWithWork.fetch_sub(1, std::memory_order_release);
            }

        }
    }
    void processGroup(size_t groupNum, size_t numGroups) {
        size_t numTasks = myTaskVec->size();
        if(numTasks < numGroups) {
            if(groupNum <= numTasks) { task(groupNum - 1).process(); }
            return;
        }
        size_t groupSize = numTasks / numGroups;
        size_t from = (groupNum - 1) * groupSize;
        size_t toe = (groupNum == numGroups) ? numTasks : groupNum * groupSize;
        for(size_t i = from; i < toe; i++) { task(i).process(); }
    }
};
\$\endgroup\$
2
  • \$\begingroup\$ This looks like homework? I’m pretty sure the assignment was to be able to add tasks to the pool after it was initialized. To minimize latency rather than throughput, you probably want a lock-free queue. (Or even wait-free, if you care about worst-case latency.) \$\endgroup\$
    – Davislor
    Dec 26, 2023 at 4:47
  • \$\begingroup\$ @Davislor, I want to avoid overhead of Queue itself as I already have a vector of tasks. As of now for my use-case with very small tasks, Overhead of ThreadPool isn't allowing me to parallelize. So I am trying to avoid every possible overhead. Also this is not homework but for a real project. \$\endgroup\$ Dec 27, 2023 at 7:57

1 Answer 1

1
\$\begingroup\$

Avoid busy loops

You have several busy loops in your code. Busy loops are bad because they keep a CPU core running at 100% all the time, thereby wasting energy and increasing temperature unnecessarily. While at first glance you might think it is faster to wake up a worker thread this way, consider that a continuously 100% used CPU core might prevent the CPU's boost frequency to be used, and on a laptop or mobile device it might even further reduce the clock speed to limit thermals and battery usage.

Also consider what happens if the main thread is doing something like this, which I think is the intended use case:

HTaskVecThreadPool threadPool(…);
std::vector<TASK> tasks = {…};
thread_pool.submit(tasks);
thread_pool.waitForAll();

Now the main thread is busy-looping as well, which means one less core available for the worker threads to run, if you don't want them to compete for CPU resources at least.

Safety

Your submit() is not very safe. First, it takes a reference to the list of tasks. If the caller accidentally modifies the tasks while the workers are still busy, that can cause problems. Furthermore, if you call submit() twice before calling waitForAll(), problems will happen. You have assert() calls in your code to prevent that from happening, but those only work in debug builds.

It's much better to combine the two functions into one. That way, it won't return until all the tasks have completed, so there is no chance for the caller to do anything wrong.

Alternatively, consider taking task by r-value reference, so your thread pool can take ownership of it by std::move()ing it into myTaskVec.

Tasks are rarely homogenious

Even if your tasks look very homogenious, it's rarely the case that they actually are executing homogenious. Caches, memory, thread scheduling and tons of other things will very likely ensure each task executes in a slightly different amount of time than the others. If you have lots of tasks, those differences add up, and then you are very likely ending up with a significant difference between the slowest and fastest worker, thereby wasting available CPU power.

Ideally, you use a hybrid approach: batch as many tasks together such that the overhead of processing a batch is insignificant compared to the time spent doing the actual tasks in that batch themselves. Then you put those batches (of which there might be more than the number of worker threads) in a queue, and let the workers pick up batches until the queue is empty.

The only time I think your implementation is really the most efficient is if you are running on a multi-core CPU in an embedded system where the operating system does not get in the way, and where the clock frequency is fixed.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.