C++11(14/17/20) Thread Management

Question

I'm implementing something of a thread janitor in modern C++. What I've got works, and while it's not organized in the way I'd like yet, I'd like some feedback on the fundamentals. Thank you in advance for any and all pointers. =)

All of the threads (including main() in order to signal program end) share data via this:

#ifndef SHAREDDATA_HPP
#define SHAREDDATA_HPP

#include <condition_variable>
#include <thread>

class SharedData {
    public:
        std::mutex              keypress_mutex;
        std::condition_variable keypress_cv;
        bool                    keypress_flag;

        std::mutex              thread_count_mutex;
        std::condition_variable thread_count_cv;
        unsigned int            thread_count;

        std::mutex              thread_kill_mutex;
        std::condition_variable thread_kill_cv;
        bool                    thread_kill_flag;

        SharedData():
            keypress_flag{false},
            thread_count{0},
            thread_kill_flag{false}
        { }
        ~SharedData() = default;
};

#endif // SHAREDDATA_HPP

Then we've got a little RAII guy to help keep track of thread count:

#ifndef THREADCOUNTER_HPP
#define THREADCOUNTER_HPP

#include <thread>

class SharedData;

class ThreadCounter {
    public:
        explicit ThreadCounter(SharedData *share);
        ~ThreadCounter();
    private:
        SharedData         *_share;
};

#endif // THREADCOUNTER_HPP


//--------------------------


#include "ThreadCounter.hpp"

#include "SharedData.hpp"

ThreadCounter::ThreadCounter(SharedData *share) :
    _share{share}
{
    std::lock_guard<std::mutex> _lock(_share->thread_count_mutex);
    share->thread_count++;
}

ThreadCounter::~ThreadCounter() {
    std::lock_guard<std::mutex> _lock(_share->thread_count_mutex);
    _share->thread_count--;

    if(_share->thread_count == 0) {
        _share->thread_count_cv.notify_one();
    }
}

Next, the ManagedThread class itself:

#ifndef MANAGEDTHREAD_HPP
#define MANAGEDTHREAD_HPP

#include <thread>

class SharedData;

class ManagedThread {
    public:
        void launch();

        explicit ManagedThread(SharedData *share);
        ~ManagedThread();
        ManagedThread() = delete;

        void operator()();

    private:
        SharedData *_share;
        std::thread _thread;
};

#endif // MANAGEDTHREAD_HPP


//--------------------------


#include "ManagedThread.hpp"

#include "SharedData.hpp"
#include "ThreadCounter.hpp"

#include <chrono>
#include <iostream>
#include <cassert>

using namespace std::chrono_literals;

void ManagedThread::launch() {
    if(!_thread.joinable()) {
        _thread = std::thread(std::ref(*this));
    }
    else {
        assert((false) && "Managed Thread not launchable!");
    }
}

ManagedThread::ManagedThread(SharedData *share) :
    _share{share},
    _thread{std::thread()}
{ }

ManagedThread::~ManagedThread() {
    if(_thread.joinable()) {
        _thread.join();
    }
}

void ManagedThread::operator()() {
    ThreadCounter tc(_share);
    while(true) {
        std::this_thread::sleep_for(250ms);
        std::cout << "Whee!" << std::endl;

        std::unique_lock<std::mutex> kill_lock(_share->thread_kill_mutex);
        if(_share->thread_kill_cv.wait_for(kill_lock, 100ms,
                [&](){ return _share->thread_kill_flag; })) {
            break;
        }
    }
}

And ThreadManager who does the cleanup coordination:

#ifndef THREADMANAGER_HPP
#define THREADMANAGER_HPP

class SharedData;

class ThreadManager {
    public:
        explicit ThreadManager(SharedData *share) : _share{share} { };
        ~ThreadManager() = default;

        void operator()();

    private:
        SharedData *_share;
};

#endif // THREADMANAGER_HPP


//--------------------------

#include "ThreadManager.hpp"

#include "SharedData.hpp"

#include <chrono>

using namespace std::chrono_literals;

void ThreadManager::operator()() {
    std::this_thread::sleep_for(25ms);

    std::unique_lock<std::mutex> keypress_lock(_share->keypress_mutex);
    _share->keypress_cv.wait(keypress_lock,
                            [&](){ return _share->keypress_flag; });
    keypress_lock.unlock();

    {
        std::lock_guard<std::mutex> kill_lock(_share->thread_kill_mutex);
        _share->thread_kill_flag = true;
    }
    _share->thread_kill_cv.notify_all();

    std::unique_lock<std::mutex> count_lock(_share->thread_count_mutex);
    _share->thread_count_cv.wait(count_lock,
                                [&](){ return _share->thread_count == 0; });
}

And, finally, a main.cpp to tie it all together.

#include "SharedData.hpp"
#include "ThreadManager.hpp"
#include "ManagedThread.hpp"

#include <iostream>
#include <future>

int main() {
    SharedData share;
    ThreadManager tm(&share);
    ManagedThread t0(&share);

    t0.launch();

    std::future<void> cf = std::async(std::launch::async, std::move(tm));

    int input;
    std::cin >> input;
    {
        std::lock_guard<std::mutex> lock(share.keypress_mutex);
        share.keypress_flag = true;
    }
    share.keypress_cv.notify_one();
    cf.get();

    return 0;
}

What I know I don't really like is that ThreadManager doesn't own SharedData. I'd also like ThreadManager to generically manage any ManagedThread and perhaps overload operator() in an inherited thread of some sort so callable can do anything but still be "managed".

At this point, those are just musings, though - what I'd really like to know is how well or poorly the system has been implemented thus far. =)

Again, thanks for your feedback; I look forward to improving this setup and utilizing it in future.

Answer 1

Thread management in C++20

In C++20 (which you mention in the title of the question), there is std::jthread which basically does most of what you are doing in your classes: it handles joining a thread in its destructor, and there is a way to request a thread to stop what it is doing, which is also automatically called in the destructor. Your example could be rewritten like so in C++20:

#include <chrono>
#include <iostream>
#include <thread>

using namespace std::chrono_literals;

int main() {
    std::jthread whee([](std::stop_token stoken){
        while (!stoken.stop_requested()) {
            std::this_thread::sleep_for(250ms);
            std::cout << "Whee!\n";
        }
    });

    int input;
    std::cin >> input;

    // Unnecessary since the thread object is destroyed anyway:
    whee.request_stop();
}

If you have multiple std::jthreads and you want them all to stop at the same time, just let all the thread objects go out of scope simultaneously, or put them in a container that can be clear()ed. For example:

int main() {
    std::vector<std::jthread> threads;
    
    for (auto i: {1, 2, 3}) {
         threads.emplace_back([i](std::stop_token stoken){
            while (!stoken.stop_requested()) {
                std::this_thread::sleep_for(250ms);
                std::cout << "Whee " << i << "!\n";
            }
         });
    }

    int input;
    std::cin >> input;

    std::cout << "Stopping all threads...\n";
    threads.clear();
    std::cout << "Done.\n";
}

Basically, std::vector<std::jthread> replaces your whole thread manager.

Sharing state amonst threads

Your idea to have some shared state that indicates whether threads should stop has merit. You only need to update one variable to send a signal to all of them, whereas with the std::jthread solution I mentioned, each std::jthread object has its own stop token. As upkajdt mentioned in his answer though, you can simplify things a lot by just using a std::atomic<bool> as a flag. In his answer, he declared this as a global variable. In case you don't want or can't use a global variable, and you want some state shared amongst multiple threads, then indeed you want to manage this state in a thread-safe way. Most importantly, it shouldn't be destroyed before every thread using it is done with it. This is a large part of what your code tries to do. However, there is already functionality in the standard library that allows you to do this: std::shared_ptr. Your code could be rewritten like so:

#include <atomic>
#include <chrono>
#include <iostream>
#include <memory>
#include <thread>

using namespace std::chrono_literals;

int main() {
    auto kill_flag = std::make_shared<std::atomic<bool>>(false);

    std::thread whee([kill_flag](){
        while (!*kill_flag) {
            std::this_thread::sleep_for(250ms);
            std::cout << "Whee!\n";
        }
    });

    int input;
    std::cin >> input;

    *kill_flag = true;
    whee.join();
}

Here the kill_flag is passed by value to the thread. This uses the copy-constructor of std::shared_ptr, which will increase the reference count of the atomic bool, and once the thread stops it will call the destructor of its copy of kill_flag, which will in turn decrease the reference count. So this replaces your ThreadCounter.

Answer 2

Pass arguments by reference

You are passing SharedData by a pointer. It would have been much cleaner to pass it by reference:

class ThreadCounter {
public:
    explicit ThreadCounter(SharedData& share);
    ~ThreadCounter();
private:
    SharedData& _share;
};

Use atomic variables instead of lock.

Locking is costly and overly complicated when you could have simply used atomic variables:

struct SharedData
{
    std::atomic<bool> keypress_flag = false;
}

int main() {
    SharedData share;
    // start threads
    int input;
    std::cin >> input;
    share.keypress_flag = true;
    // join threads
}

Don't use std::endl

std::endl Flushes the output. You don't have to, simply use \n:

std::cout << "Whee!\n";

Final thoughts

I'm not really sure what you are trying to accomplish. It seems to me that you could have replaced all your code with something as simple as the following:

#include <atomic>
#include <iostream>
#include <thread>

std::atomic_bool running = true;

void func()
{
    using namespace std::chrono_literals;
    while (running)
    {
        std::this_thread::sleep_for(250ms);
        std::cout << "Whee!\n";
    }
}

int main()
{
    std::thread thread(func);
    int input;
    std::cin >> input;
    running = false;
    thread.join();
}