5
\$\begingroup\$

I'm currently in the process of replacing an archaic multithreading solution using some of the newer C++ standard library features now that our software has been updated to use C++20.

Previously, most of our multithreading was implemented using thread managers that controlled the flow of the threads and allocated tasks to them. Typically this was done to chop up a very slow function into smaller "bites" of work that could be spread across available threads in order to improve performance.

I wanted to create a templated solution that removes the overhead of creating a unique thread manager class and that also reduces complexity for future multithreading implementations.

Here is the new solution I've come up with:

Threads.h

#include <chrono>
#include <future>
#include <vector>
using namespace std::chrono_literals;

template <class TPayload>
void DoTasksAsync(
    std::function<void(const TPayload&)> asyncTask,
    const std::vector<TPayload>& jobs,
    std::function<bool()> shouldContinue)
{
    int numWorkerThreads = std::thread::hardware_concurrency() - 1;
    size_t jobsLeft = jobs.size();
    std::vector<std::future<void>> futures;
    do
    {
        // create a time point 1 microsecond from now
        std::chrono::system_clock::time_point aLittleLater = std::chrono::system_clock::now() + 1us;

        // remove all futures that have finished in space of 1 microsecond
        std::erase_if(futures, [aLittleLater](std::future<void>& thr)
            {
                return thr.wait_until(aLittleLater) == std::future_status::ready;
            });

        // if there is space in the futures vector, begin another task in another thread
        // with the next payload
        while (futures.size() < numWorkerThreads && jobsLeft > 0)
        {
            TPayload payload = std::move(jobs[--jobsLeft]);
            futures.push_back(
                std::async(
                    std::launch::async,
                    [&asyncTask, payload = std::move(payload)]
                    {
                        asyncTask(payload);
                    })
            );
        }
    } while (futures.size() > 0 && shouldContinue());

    // ensure all threads have finished
    // -- important when shouldContinue() breaks the loop while threads are still executing
    for (std::future<void>& thr : futures)
    {
        if (thr.valid())
        {
            thr.wait();
        }
        else
        {
            throw std::future_error(std::future_errc::no_state);
        }
    }
}

Here is an example of how it might be used:

Sorter.cpp

#include "Threads.h"

#include <algorithm>
#include <iostream>
#include <vector>

enum SortingAlgorithm { Bubble, IntroSort };

template<class T>
class Sorter
{
public:
    void Sort(std::vector<T>& vector, SortingAlgorithm algorithm)
    {
        switch (algorithm)
        {
        case Bubble:
            DoBubble(vector);
            break;
        case IntroSort:
            DoIntroSort(vector);
            break;
        }
    }

private:
    void DoBubble(std::vector<T>& vector)
    {
        size_t size = vector.size();
        for (int i = 0; i < size - 1; i++)
        {
            for (int j = 1; j < size - 1; j++)
            {
                if (vector[j] < vector[j - 1])
                {
                    T temp = vector[j];
                    vector[j] = vector[j - 1];
                    vector[j - 1] = temp;
                }
            }
        }
    }

    void DoIntroSort(std::vector<T>& vector)
    {
        std::sort(vector.begin(), vector.end());
    }
};

struct SortVectorPayload
{
    Sorter<int>* Sorter;
    SortingAlgorithm Algorithm;
    std::vector<int>* Vector;
};

static void SortVectorAsync(const SortVectorPayload& payload)
{
    payload.Sorter->Sort(*payload.Vector, payload.Algorithm);
}

int main()
{
    Sorter<int>* sorter = new Sorter<int>();
    std::vector<SortVectorPayload> payloads;

    // fill the list of payloads with some random stuff to do
    std::vector<std::vector<int>*> lists;
    for (int i = 0; i < 100; i++)
    {
        lists.push_back(new std::vector<int>(1000));
        std::generate(lists.back()->begin(), lists.back()->end(), []() { return std::rand(); });
        payloads.push_back(SortVectorPayload{ sorter, Bubble, lists.back() });
    }
    for (int i = 0; i < 100; i++)
    {
        lists.push_back(new std::vector<int>(1000));
        std::generate(lists.back()->begin(), lists.back()->end(), []() { return std::rand(); });
        payloads.push_back(SortVectorPayload{ sorter, IntroSort, lists.back() });
    }

    // sort the lists on available threads
    DoTasksAsync(std::function(SortVectorAsync), payloads, []() { return true; });

    for (std::vector<int>* list : lists)
    {
        delete list;
    }

    delete sorter;
}

So, here are my major concerns I'd like to get some help with:

  1. I'm using two std::move() calls inside DoTasksAsync to put the payloads in the right location. Is there anything inherently wrong with using both of these calls when perhaps one would suffice, or am I using too few and copying more data than I need to?

  2. The only way I could come up with for checking the status of all the futures that are currently running was to create a timepoint 1 microsecond in the future and calling wait_until() on each future. Is there a cleaner way of doing this that would also allow the main thread to continue checking the shouldContinue() function each iteration?

  3. I never call get() on any of my futures. Will these leave futures floating around that are still waiting for something to get their result even after they've left the scope of the function?

  4. I don't deal with return types at all and just use the payload to have each thread modify some shared data (with appropriate locks and such). What changes could I make to this function to enable return values from each thread's task?

  5. And finally, is what I'm doing here even a good idea at all or am I completely barking up the wrong tree?

\$\endgroup\$
0

2 Answers 2

8
\$\begingroup\$

Use worker threads that pick jobs themselves

There are a few issues with your code. First, you call std::async() once for every job. With most implementations of the standard library, it means it creates a new std::thread for every job. While you limit the amount of concurrent threads, creating and destroying a std::thread still has some cost. So it would be better to just create numWorkerThread std::threads, and have each thread pick multiple jobs from jobs themselves, until all jobs are finished.

Another issue is that in your implementation, you have a busy-loop waiting for futures to become ready. Sure, it waits up to a microsecond, but that's usually not enough for a CPU core to go into low-power idle mode, so effectively you are still burning power continuously just waiting for another core to finish a task. It's much better to use std::condition_variable to signal that something has finished. Or you could even avoid explicitly waiting entirely. Consider this implementation:

template <class TPayload>
void DoTasksAsync(
    std::function<void(const TPayload&)> asyncTask,
    const std::vector<TPayload>& jobs,
    std::function<bool()> shouldContinue)
{
    const std::size_t numWorkerThreads =
            std::min(jobs.size(), std::thread::hardware_concurrency());
    std::atomic_size_t next_job = 0;

    std::vector<std::jthread> threads;
    threads.reserve(numWorkerThreads);

    for (std::size_t i = 0; i != numWorkerThreads; ++i) {
        threads.emplace([&](){
            std::size_t job;
            while (shouldContinue() && (job = next_job++) < jobs.size()) {
                std::invoke(asyncTask, jobs[job]);
            }
        });
    }
}

Moving from a const container

You pass jobs by const reference, but later you try to std::move() items from jobs. Because jobs is const, it cannot actually do a real move, and will instead just perform a copy instead. You could pass jobs as a non-const reference, but even better is to not std::move() at all (as shown above), as no copy has to be made if asyncTask itself takes the payload parameter by const reference.

Handling the return value of asyncTask

This is definitely possible, you just need to ensure DoTasksAsync is also templated on the return value of asyncTask somehow, and you have to decide on some way to store the return values. You could for example use a std::vector for that:

template <class TResult, class TPayload>
auto DoTasksAsync(
    std::function<TResult(const TPayload&)> asyncTask,
    …)
{
    …
    std::vector<TResult> results(numWorkerThreads);
    …
        results[job] = std::invoke(asyncTask, jobs[job]);
    …
    return results;
}

However, note that we now have to explicitly pass the type of the return value as a template parameter when calling DoTasksAsync(). Unfortunately, it cannot deduce this. The solution to this is not to use std::function for passing asyncTask, but rather use a template parameter for the function type, and then deducing the return type from that:

template <class TPayload, class TAsyncTask>
auto DoTasksAsync(
    TAsyncTask asyncTask,
    …)
{
    using TResult = std::invoke_result_t<TAsyncTask, decltype(jobs.first())>;
    …
    std::vector<TResult> results(numWorkerThreads);
    …
}

Make it more generic

Your function requires that the payloads are stored in a std::vector. However, what if you had them in a std::list instead? Or a std::deque, or any other type of container? The only thing you care about is that you can iterate over the range of payloads. So consider writing the function like so:

template <class TJobs, class TAsyncTask>
auto DoTasksAsync(
    TAsyncTask asyncTask,
    TJobs&& jobs,
    …)
{
    …
}

You might also consider that the caller wants the result in something different than a std::vector. You could instead have it provide an output iterator to store the results in. This brings me to:

You are effectively implementing a parallel std::transform()

You are applying asyncTask to all elements of jobs. That's basically a parallel std::transform(). Consider looking at the interface of std::transform() and std::ranges::transform().

Even better, std::transform() can actually run things in parallel, if you it pass an execution policy as the first parameter. So maybe you don't need DoTasksAsync() at all?

Consider using a dynamic thread-safe queue for jobs

You have to pass a vector of a fixed size to DoAsyncTasks(). However, if you look at your example code, then you are building up payloads, but only once that is fully built can you start processing the payloads. But ideally you could already have a thread start working on the first list right after you added that.

The usual solution to that is to use a thread-safe queue. The worker threads will pick items from that queue to process, and the main thread can push new jobs to that queue whenever it wants to. You can find many examples here on Code Review.

Avoid manual memory management

In your example code, you do a lot of manual memory management. This often leads to bugs, as indeed happens in your code: you forgot to delete sorter, so you have a memory leak. Either avoid using pointers altogether, use smart pointers and/or use better containers. For example, there is no need to allocate sorter on the heap:

CSorter sorter();

And to get a stable array of lists:

std::deque<std::vector<int>> lists;

This works since std::deque will never move its elements around in memory.

\$\endgroup\$
3
  • \$\begingroup\$ Sniped me. Will come back later and look for points you didn’t cover. \$\endgroup\$
    – Davislor
    Commented Mar 15 at 19:54
  • \$\begingroup\$ Sorry @Davislor! I hope I left something for you ;) \$\endgroup\$
    – G. Sliepen
    Commented Mar 15 at 19:57
  • \$\begingroup\$ This is absolutely brilliant. Thank you so much for taking the time to write such a well thought out answer to my question. I'm going to pore over all of your suggestions on Monday and take the time to learn a thing or five. I'm not entirely surprised to find out I was trying to reinvent the wheel with something like this, I just wish I'd found out about std::transform before starting this exercise! \$\endgroup\$
    – Swepps
    Commented Mar 15 at 23:03
6
\$\begingroup\$

Not a full review of the correctness of your multi-threading code, but some initial thoughts on the algorithm design.

Your Example Doesn’t Compile

I don’t think a review of it is the main point of your question, but it’s not clear what a CSorter is, and there are no comments.

Use a Barrier Wait

This is the classic solution to waiting for all threads in a pool to complete. In the C++ standard library, it’s std::barrier::wait.

With this implementation, the main thread could spawn the worker threads and return a manager object to the caller. A method on it could have the main thread wait on the barrier, and sync up with all the workers. But until then, the main thread could do as much work as it pleases.

Alternatively, if the threads detect when there are no jobs from them and quit on their own, you could just join the worker threads as they finish, or even return without blocking and let them run asynchronously.

Consider a FIFO Queue

You could put tasks (probably closures, such as std::regular_invokable objects) on a multiple-consumer FIFO queue, and the worker threads could dequeue their next tasks as they become free. This could be lock-free (like boost.lockfree.queue), or mutex-locked.

This would perform much better, as the manager would not need to poll at intervals to see if the threads have completed. Each thread takes a new task off the list and joins the others when no more tasks remain. This could even take more items and add them to the queue in real time (on which more later).

Alternatively, the task manager could partition the array of jobs into equal slices and pass each slice to a thread procedure that runs all the jobs in the slice, then joins at the barrier. This could produce unbalanced loads that leave some threads idle, but is much simpler, with trivial overhead, and wait-free. However, it requires a fixed list of jobs known in advance.

The Jobs Parameter

Currently, you take a const std::vector<Payload>&. Will any caller ever need the unmodified, original list of payloads after calling DoTasksAsync? In that situation, could you make a temporary copy?

If you actually move items from the input list, you are leaving whatever data structure it is in an unspecified-but-valid state, making it unusable afterward. It would then be better to pass the data structure as an rvalue reference. In this case, the object should implement the std::ranges::range concept.

Alternatively, you could take a start and end iterator as the input, like most of the C++ standard library. This would vastly increase the flexibility of the library, although you’d need to document any assumptions you make about when the complete list of jobs will be available for processing, whether it has random access, and so forth.

The Payload Type

Here, I have a different point of view from G. Sliepen. Any closure which doesn’t throw exceptions will work to return data, if you pass the output parameter as part of the closure. This would not go into a std::vector, without special precautions such as aligning each element to its own cache line, or you would get bugs from false sharing when multiple threads try to write to it. If you have jobs whose execution time will very a lot, or you need to dynamically add new tasks (such as a pool of worker threads that listens for new requests, and converts them into tasks for another pool of worker threads to handle), or you need to check for partial results asynchronously instead of blocking until all jobs have been completed, I suggest that you program each task to know where it should leave its result, and have the thread procedure itself return void.

But Take Advantage of Your Requirements

This interface, however, lets you presume that you have a complete vector of all the jobs in advance, and that you are allowed to block until all jobs are complete. (You don’t actually need it to be an input vector. Any array-like, contiguous input range will do.) In that case, you can partition the array into roughly-equal slices. You could then store the results in an output array, which you could slice the same way, so that no two slices overlap the same cache line. (This will require the output array be correctly aligned in memory.) Then, each thread could write to its slice of the output array safely. The manager thread itself could become a worker temporarily, join the others when its shift is done, and then return the filled output array.

If you’re reasonably sure that the execution times of the jobs in each slice will average out to about the same, this would be an extremely simple and efficient thread manager. It’s also, as ol’ G brought up, very similar to std::transform with a parallel execution policy. Or to an OpenMP loop.

You can get more flexible than that, but not if you’re already constraining the list of jobs to be in a std::vector.

Be Flexible About the Number of Threads

At present, this design always spawns as many threads as the system reports it has the hardware concurrency for. You might want to leave some cores for other purposes. You might want a function argument that defaults to this value, but which the caller can override:

unsigned nThreads = std::thread::hardware_concurrency() - 1
\$\endgroup\$
2
  • \$\begingroup\$ Another excellent answer, thank you! The example code was something I knocked up in 5 minutes to try to show a very simple use case - after spending some more time on here I can see that it's important to have compiling and usable code, even in the examples. I'll get it sorted. I think implementing a queue and a more generic container (probably using iterators) for the input would be the next best step for me. I am tempted by the idea of just slicing up the work evenly among available threads but in our real usage of this, an individual job can vary in length by orders of magnitude. \$\endgroup\$
    – Swepps
    Commented Mar 16 at 7:56
  • \$\begingroup\$ Swepps has improved the example code. While editing the code that's presented for review is not allowed, this seems to be okay. In which case your first observation can be edited/removed. \$\endgroup\$ Commented Mar 18 at 13:02

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.