3
\$\begingroup\$

Edit: An improved code based on feedback is available here.


As a kind of sequel to the previous question, here is an improved version( with clarified naming ). The Idea is the same: An integer 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.

This version has no busy-waiting, and it's more generic because of templates. Is there any possibility to optimize more? I would think that using atomic for the state is overkill, as mostly it is modified under a mutex, but I couldn't get the program to work without using atomic.

I have tried to re-implement this using variable template arguments, but I failed. I would think it is mainly because the argument for running is provided through pointers. Would using lambdas be a better solution here? ( Mainly regarding portability, but also in performance )


#include <iostream>
#include <functional>
#include <vector>
#include <thread>
#include <mutex>
#include <iomanip>
#include <cassert>
#include <numeric>
#include <atomic>
#include <condition_variable>
#include <chrono>

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

template<typename T>
class ThreadGroup{
public:
    ThreadGroup(int number_of_threads, function<void(T&, int, int)> function)
    :  target_buffer(nullptr)
    ,  worker_function(function)
    ,  threads()
    ,  threads_ready(0)
    ,  state(IDLE_VALUE)
    ,  state_mutex()
    ,  synchroniser()
    {
      for(int i = 0; i < number_of_threads; ++i)
        threads.push_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_VALUE);
      }
      while(0 < threads.size()){
        if(threads.back().joinable())
          threads.back().join();
        threads.pop_back();
      }
    }

    void start_and_block(T& buffer){
      { /* initialize, start.. */
        unique_lock<mutex> my_lock(state_mutex);
        target_buffer = &buffer;
        state.store(START_VALUE);
      }

      { /* wait until the work is done */
        unique_lock<mutex> my_lock(state_mutex);
        if(threads.size() > threads_ready)synchroniser.wait(my_lock,[=](){
          return (threads.size() <= threads_ready);
        });
      }
      { /* set appropriate state */
        unique_lock<mutex> my_lock(state_mutex);
        state.store(IDLE_VALUE);
      }
      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);
        if(0 < threads_ready)synchroniser.wait(my_lock,[=](){
          return (0 >= threads_ready); /* All threads are notified once the @threads_ready variable is zero again */
        });
      }
    }

private:
    static const int IDLE_VALUE = 0;
    static const int START_VALUE = 1;
    static const int END_VALUE = 2;

    T* target_buffer;
    function<void(T&, int, int)> worker_function; /* buffer, start, length */
    vector<thread> threads;
    int threads_ready;
    atomic<int> state;
    mutex state_mutex;
    condition_variable synchroniser;

    void worker(int thread_index){
      while(END_VALUE != state.load()){ /* Until the pool is stopped */
        while(START_VALUE == state.load()){ /* Wait until start signal is provided */
          worker_function(
            (*target_buffer),
            (thread_index * (target_buffer->size()/threads.size())),
            (target_buffer->size()/threads.size())
          );/* 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);
            if(START_VALUE == state.load())synchroniser.wait(my_lock,[=](){
              return (START_VALUE != 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 */
        } /*while(START_VALUE == state)*/
      } /*while(END_VALUE != state)*/
    }
};

int main(int argc, char** agrs){
  int result = 0;
  mutex cout_mutex;

  ThreadGroup<vector<double>> pool(5,[&](vector<double>& buffer, int start, int length){
    double sum = 0;
    for(int i = 0; i < length; ++i){
      sum += buffer[i];
    }
    lock_guard<mutex> my_lock(cout_mutex);
    std::cout << "Partial sum: " << std::setw(4) << sum << " \t\t    |" << "\r";
    result += sum;
    //std::this_thread::sleep_for(std::chrono::milliseconds(200)); //to test with some payload
  });

  result = 0;
  for(int i = 0; i< 10000; ++i){
    vector<double> test_buffer(500, rand()%10);
    result = 0;
    pool.start_and_block(test_buffer);
    {
      lock_guard<mutex> my_lock(cout_mutex);
      std::cout << "result["<< i << "]: " << std::setw(4) << result << "\t\t    " << std::endl;
    }
    assert(std::accumulate(test_buffer.begin(),test_buffer.end(), 0) == result);
  }
  std::cout << "All assertions passed!   "<< std::endl;
  return 0;
}


\$\endgroup\$
5
  • \$\begingroup\$ not really, I just prefer to keep it lowest as possible, because of portability reasons. I have been using C++14, and I found no features in c++17 that would make me switch. But any relevant feature would make me switch to a higher standard. \$\endgroup\$ Commented Jun 15, 2021 at 8:36
  • \$\begingroup\$ I was able to include multiple arguments into the template by using std::tuple, by using c++14 standard. I'M not sure if I should post the code here , or to a new question ( as there were no suggestions as of yet... ); And even in C++11 there is the index based extract, instead of the type based one I'm using in the improved code... en.cppreference.com/w/cpp/utility/tuple/get \$\endgroup\$ Commented Jun 15, 2021 at 12:11
  • \$\begingroup\$ "have this implementation variable template arguments" I don't understand that. Have this implementation do what? (missing verb?) \$\endgroup\$
    – JDługosz
    Commented Jun 15, 2021 at 23:53
  • \$\begingroup\$ Right, that's my eNgLuSh, sorry. I'll fix the phrasing. \$\endgroup\$ Commented Jun 16, 2021 at 6:23
  • \$\begingroup\$ The way to implement using template arguments is to pass the function as a parameter, but its type is a template. Like any of the standard algorithms. Then you avoid the overhead of the std::function and enable inlining. I don't think inlining is at all useful here, and you need a std::function to start the thread anyway, so it's fine as you have it. \$\endgroup\$
    – JDługosz
    Commented Jun 16, 2021 at 13:58

2 Answers 2

5
\$\begingroup\$

Fix compiler warnings

Enable compiler warnings and try to fix all of them. There are several unused parameters. You can avoid the warnings by not giving the parameters a name. For example, you can omit the name start from the lambda in main():

[&](std::vector<double>& buffer, int, int length){...}

As for the parameters of main() itself, you can use the same trick, or use the other allowed form of main() that takes no parameters:

int main() {

The remaining warnings I get are about comparisons between signed and unsigned integers. Use std::size_t as the type of threads_ready.

Unconditionally call wait()

There is no need to check the condition before calling wait() on a condition variable if you are passing a predicate. The first thing wait() will do is to execute the predicate to see if it needs to wait at all.

Prefer default initializers over initializer lists

The constructor of ThreadGroup has a large initializer list. Some of them are redundant, and some of them could be replaced by default initializers. The only things that you should have to put in initializer lists normally are things that depend on the parameters passed to the constructor. So:

class ThreadGroup {
public:
    ThreadGroup(std::size_t number_of_threads, std::function<void(T&, int, int)> function)
    : worker_function(function)
    {
        for(std::size_t i = 0; i < number_of_threads; ++i)
            threads.emplace_back(&ThreadGroup::worker, this, i);
    }

    ...

private:
    T* target_buffer = nullptr;
    std::function<void(T&, int, int)> worker_function;
    std::vector<std::thread> threads;
    std::size_t threads_ready = 0;
    std::atomic<int> state = {IDLE_VALUE};
    std::mutex state_mutex;
    std::condition_variable synchroniser;
    ...
};

Note that we have to use aggregate initialization for std::atomic<int> here, otherwise the deleted copy constructor would be selected.

Don't busy-loop

You start worker threads in the constructor, but state is initialized to IDLE_VALUE. This causes worker() to go into a busy-loop until state changes. Use a condition variable so the worker thread can wait() for work to arrive.

Unconditionally join() threads

I don't see why you are checking if threads are joinable or not. Once they are added to threads, they are always in a joinable state. So I would change the destructor to:

~ThreadGroup() {
    {
        std::lock_guard<mutex> my_lock(state_mutex);
        state.store(END_VALUE);
        // signal a condition variable to ensure idle threads get woken up
    }

    for (auto &thread: threads)
        thread.join();
}

Avoid using namespace std or using std:: in headers

While it is usually safe to use using namespace std in a source file, and it's even better to just bring individual elements from std:: into the global namespace as you did, as soon as you would move the definition of class ThreadGroup to a header file, you should not do this anymore, as that will result in unexpected behavior for source files that don't want to use that but do #include your header file. It's not that much extra work to type std::, and as a bonus you avoid possible confusion when you have things like function<...> function, where the variable starts shadowing the type.

\$\endgroup\$
9
  • \$\begingroup\$ Thank you for the suggestions! I'll have my questions in the following comments.. :) \$\endgroup\$ Commented Jun 16, 2021 at 5:53
  • 1
    \$\begingroup\$ I don't use namespace std... I explicitly state each variable I'm using at the beginning.. \$\endgroup\$ Commented Jun 16, 2021 at 5:57
  • 1
    \$\begingroup\$ In my example, the pointer member is initialized explicitly to nullptr at the point where it is declared. Yes, the busy-loop can be solved by using the condition variable. You don't need to call threads.clear() in the destructor, since it will be destroyed automatically (member variables have their destructor called after the class's destructor function has run, in reverse order of how they were constructed). \$\endgroup\$
    – G. Sliepen
    Commented Jun 16, 2021 at 6:30
  • 1
    \$\begingroup\$ I like this practice, thank you for sharing it with me! Except the initializer of atomic is deleted; So ( at least with my skills ) I'm forced to include it into the initialization list. \$\endgroup\$ Commented Jun 16, 2021 at 6:38
  • 1
    \$\begingroup\$ You can get the atomic initialized properly, if you use braces, sorry about that. \$\endgroup\$
    – G. Sliepen
    Commented Jun 16, 2021 at 21:07
2
\$\begingroup\$

Try something like enum state_t { Idle, Start, End }; and use the type state_t instead of a plain int for values of this type.

The first answer already covered the initializers for the constructor in some detail, and I had pointed this out in the answer to your previous post.

threads.push_back(thread(&ThreadGroup::worker, this, i));
You are constructing a temporary std::thread instance, copying that to the vector, and then destroying the temporary. This is a perfect use for emplace_back, where you give it the constructor arguments rather than a temporary constructed object. (Though I see that push_back has an rvalue form and thread has a move constructor, so the amount of work is reduced but not simply eliminated.)

while(0 < threads.size()){
This is more clearly stated as (while !threads.empty()).

unique_lock<mutex> my_lock(state_mutex);
I always found it annoying to have to specify the exact mutex type when creating a lock_guard or unique_lock, as usually the type name is a bit more verbose and specific. Now, you can use CTAD and just let the template automatically determine the argument type: unique_lock my_lock(state_mutex);

if(threads.size() > threads_ready)synchroniser.wait(my_lock,[=](){
          return (threads.size() <= threads_ready);

Are you aware that you are capturing this, not the individual members? I think so, since you are capturing by value and that makes sense for a pointer but not for the vector itself. It would be better (and safer) to list the captures rather than using a default. That is, simply [this]. In fact,

The implicit capture of *this when the capture default is = is deprecated. (since C++20)

\$\endgroup\$
4
  • \$\begingroup\$ Thank you very much! :) I'd ask some questions in the following comments, if it's okay .. :) \$\endgroup\$ Commented Jun 16, 2021 at 5:58
  • \$\begingroup\$ Well most comments I got to with a bit of googling, thank you! Most, except the ctags one; The way I understand it that would introduce an additional environment dependency here ( as it is a tool ) ? \$\endgroup\$ Commented Jun 16, 2021 at 6:22
  • \$\begingroup\$ I have improved upon the previous code based on your feedback. Thank you! \$\endgroup\$ Commented Jun 16, 2021 at 9:23
  • 1
    \$\begingroup\$ Sorry, that should have been CTAD. I updated my answer to include a useful link. lock_guard is the third example. \$\endgroup\$
    – JDługosz
    Commented Jun 16, 2021 at 13:53

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.