1
\$\begingroup\$

After the amazing feedback from these questions; I have prepared a third version of the original posted code.

The Idea is the same: An std::size_t variable threads_ready is increased to threads.size() until all threads are finished with the payload, and then back to 0 when all threads are ready to execute again.

  • I eliminated all busy waiting that I could find
  • made the class more generic by making the class use of variadic templates.
    • The Best solution would be to have a lambda capture the required context, but unfortunately I couldn't find a way to deter thread related parameters comfortably from a generic lambda. The best I could manage was to reduce the used parameters to "thread index", so each thread would have an idea about the relevant regions in the inputs.
  • I made a better use-case to better test the implementation
#include <iostream>
#include <functional>
#include <tuple>
#include <vector>
#include <thread>
#include <mutex>
#include <iomanip>
#include <numeric>
#include <atomic>
#include <condition_variable>

#include <algorithm>
#include <cassert>
#include <chrono>
#include <cmath>

using std::atomic;
using std::vector;
using std::function;
using std::tuple;
using std::thread;
using std::mutex;
using std::unique_lock;
using std::lock_guard;
using std::condition_variable;
using std::size_t;

template<typename First, typename ...T>
class ThreadGroup{
public:
    ThreadGroup(int number_of_threads, function<void(tuple<First, T...>&, int)> function)
    :  worker_function(function)
    ,  state(Idle)
    {
      for(int i = 0; i < number_of_threads; ++i)
        threads.emplace_back(thread(&ThreadGroup::worker, this, i));
    }

    ~ThreadGroup(){
      { /* Signal to the worker threads that the show is over */
        lock_guard<mutex> my_lock(state_mutex);
        state.store(End);
      }
      synchroniser.notify_all();
      for(thread& thread : threads) thread.join();
    }

    void start_and_block(tuple<First, T...>& buffer){
      { /* initialize, start.. */
        unique_lock<mutex> my_lock(state_mutex);
        target_buffers = &buffer;
        state.store(Start);
      }
      synchroniser.notify_all(); /* Whip the peons */

      { /* wait until the work is done */
        unique_lock<mutex> my_lock(state_mutex);
        synchroniser.wait(my_lock,[this](){
          return (threads.size() <= threads_ready);
        });
      }
      { /* set appropriate state */
        unique_lock<mutex> my_lock(state_mutex);
        state.store(Idle);
      }
      synchroniser.notify_all(); /* Notify worker threads that the main thread is finished */

      { /* wait until all threads are notified */
        unique_lock<mutex> my_lock(state_mutex);
        synchroniser.wait(my_lock,[this](){
          return (0 >= threads_ready); /* All threads are notified once the @threads_ready variable is zero again */
        });
      }
    }

private:
    enum state_t{Idle, Start, End};

    tuple<First, T...>* target_buffers = nullptr;
    function<void(tuple<First, T...>&, int)> worker_function; /* start, length */
    vector<thread> threads;
    size_t threads_ready = 0;
    atomic<state_t> state;
    mutex state_mutex;
    condition_variable synchroniser;

    void worker(int thread_index){
      while(End != state.load()){ /* Until the pool is stopped */
        { /* Wait until main thread triggers a task */
          unique_lock<mutex> my_lock(state_mutex);
          synchroniser.wait(my_lock,[this](){
            return (Idle != state.load());
          });
        }
        if(End != state.load()){
          worker_function((*target_buffers), thread_index);/* do the work */

          { /* signal that work is done! */
            unique_lock<mutex> my_lock(state_mutex);
            ++threads_ready; /* increase "done counter" */
          }
          synchroniser.notify_all(); /* Notify main thread that this thread  is finsished */

          { /* Wait until main thread is closing the iteration */
            unique_lock<mutex> my_lock(state_mutex);
            synchroniser.wait(my_lock,[this](){
              return (Start != state.load());
            });
          }

          { /* signal that this thread is notified! */
            unique_lock<mutex> my_lock(state_mutex);
            --threads_ready; /* decrease the "done counter" to do so */
          }
          synchroniser.notify_all(); /* Notify main thread that this thread  is finsished */
        } /* Avoid segfault at destruction */
      } /*while(END_VALUE != state)*/
    }
};

int main(int argc, char** agrs){
  const int number_of_threads = 5;
  vector<double> test_buffer;
  double expected;
  double result = 0;
  mutex cout_mutex;

  ThreadGroup<vector<double>&> pool(number_of_threads,[&](tuple<vector<double>&>& inputs, int thread_index){
    double sum = 0;
    vector<double>& used_buffer = std::get<vector<double>&>(inputs);
    size_t length = (used_buffer.size() / number_of_threads) + 1u;
    size_t start = length * thread_index;
    length = std::min(length, (used_buffer.size() - start));
    if(start < used_buffer.size()) /* More threads could be available, than needed */
      for(size_t i = 0; i < length; ++i) sum += used_buffer[start + i];
    //std::this_thread::sleep_for(std::chrono::milliseconds(200)); //to test with some payload
    { /* Print partial results and accumulate the full results */
      lock_guard<mutex> my_lock(cout_mutex);
      std::cout << "Partial sum[" << thread_index << "]: " << std::setw(4) << sum << " \t\t    \r";
      result += sum;
    }
  });

  for(int i = 0; i< 1000; ++i){
    test_buffer = vector<double>(rand()%500);
    std::for_each(test_buffer.begin(),test_buffer.end(),[](double& element){
      element = rand()%10;
    });
    expected = std::accumulate(test_buffer.begin(),test_buffer.end(), 0.0);
    result = 0;
    auto tpl = std::forward_as_tuple(test_buffer);
    pool.start_and_block(tpl);
    std::cout << "result["<< i << "]: " << std::setw(4) << result << "\t\t    \r";
    assert(expected == result);
  }
  std::cout << "All assertions passed!   "<< std::endl;
  return 0;
}

Is there anything else that could be optimized/improved with this implementation?

\$\endgroup\$
7
  • \$\begingroup\$ I don't see why the function needs all those parameters. It should only take the thread ID, and anything else it needs would be lambda captures when making that specific function. \$\endgroup\$
    – JDługosz
    Commented Jun 16, 2021 at 15:15
  • \$\begingroup\$ Hmm right, the lambda function can be provided as a pointer through the start_and_block function! \$\endgroup\$ Commented Jun 17, 2021 at 6:00
  • \$\begingroup\$ Isn't providing a lambda at runtime inefficient? Are lambdas used like that created at program start, or a new lambda ( of the same functionality ) is created at every call? \$\endgroup\$ Commented Jun 17, 2021 at 6:03
  • 1
    \$\begingroup\$ You shouldn't need variadic anything. Just declare the function to take the thread ID. What the specific function actually needs will be handled by the lambda capture for that specific use. \$\endgroup\$
    – JDługosz
    Commented Jun 18, 2021 at 14:18
  • 1
    \$\begingroup\$ You're right, this is something I could accept as an answer! \$\endgroup\$ Commented Jun 23, 2021 at 16:11

1 Answer 1

1
\$\begingroup\$

Lambdas are not inefficient. You need to pack up those values somehow, and the way lambda captures work are just constructor arguments. That is, it copies the desired values into storage locations, which is exactly the work needed when passing a parameter (storing it into a local variable in the called function). Your tuple packs up values into the tuple, which is morally the same as a structure with unnamed members; again, assuming no extra copies are being made, that is the same amount of work again.

The most straightforward solution is to declare your function to just take the thread ID, and pass in a lambda that captures whatever it needs for the actual function to be performed.

We hope that (in a release build anyway) that the constructor argument gets optimized out. But in the case where you are capturing something like vector or string, you want to use move semantics. To this end, the constructor argument should be a "sink" parameter. That is, declare it to take by-value, and use std::move in the member initialization. In any case, you can play around in Compiler Explorer to make sure that the extra copy is actually optimized away. In practice, copying a few bytes is nothing compared to the cost of starting a thread or synchronizing anything! As long as you are not deep-copying something expensive like vector it's not inefficient.

To clarify:
ThreadGroup x { 5, [=](int id){ 🏭 } }; will initialize the lambda's body (the captures) directly in the function parameter of the constructor.
Then, : worker_function{std::move(function)} in that constructor will initialize the class member. That extra copying is what we hope to optimize out: store the captures directly into the final resting place inside the lambda inside the std::function inside the ThreadGroup. Using move semantics ensures that even if copies are not entirely eliminated, it will not do expensive deep copies.

\$\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.