C++ Thread Server: Thread control & management

Question

So I created a thread server that creates threads and associates them with a handle so that you can keep specific threads for specific tasks e.g. run all graphics rendering on thread 0, run physics stepping on thread 1 and use thread 2 for downloading a file of the internet etc. Personally I've found it useful in a project of mine and I highly prefer this to a typical job system where you don't have any control over what thread the task is executed on.

I'd like to know what room for improvement there is, especially in optimization and whether or not you think the overhead is justified.

Each thread executes tasks in a task queue, so I made a Thread_Safe_Queue which is just a wrapper for std::queue but with a mutex before read/writes:

template<typename T>
struct Thread_Safe_Queue {
    Thread_Safe_Queue() = default;

    Thread_Safe_Queue(Thread_Safe_Queue<T>&& other) noexcept {
        std::lock_guard<std::mutex> lock(mutex);
        queue = std::move(other.queue);
    }
    Thread_Safe_Queue(const Thread_Safe_Queue<T>& other) {
        std::lock_guard<std::mutex> lock(mutex);
        queue = other.queue;
    }

    virtual ~Thread_Safe_Queue() { }

    size_t size() const {
        std::lock_guard<std::mutex> lock(mutex);
        return queue.size();
    }

    std::optional<T> pop() {
        std::lock_guard<std::mutex> lock(mutex);
        if (queue.empty()) {
            return {};
        }
        T tmp = queue.front();
        queue.pop();
        return tmp;
    }

    std::optional<T> front() {
        std::lock_guard<std::mutex> lock(mutex);
        if (queue.empty()) {
            return {};
        }
        return queue.front();
    }

    void push(const T &item) {
        std::lock_guard<std::mutex> lock(mutex);
        queue.push(item);
    }

    bool empty() const {
        std::lock_guard<std::mutex> lock(mutex);
        return queue.empty();
    }

    void clear() {
        std::lock_guard<std::mutex> lock(mutex);
        queue = std::queue<T>();
    }

    std::queue<T> queue;
    mutable std::mutex mutex;
};

Thread_Server.h:

#include "thread_safe_queue.h"

#include <thread>
#include <functional>

typedef unsigned int thread_id_t;
constexpr thread_id_t NULL_THREAD = (thread_id_t)0 - (thread_id_t)1;

typedef std::function<void()> Thread_Task;

struct Thread_Context {
    Thread_Safe_Queue<Thread_Task> task_queue;
    bool pause;
    bool kill = false;
    bool dead = false;
};

struct Thread_Server {
    Thread_Server();
    ~Thread_Server();
    thread_id_t make_thread(bool start = true);

    void pause_thread(thread_id_t tid);
    void start_thread(thread_id_t tid);

    void kill_thread(thread_id_t tid);

    void queue_task(thread_id_t tid, const Thread_Task& task);

    void wait_for_thread(thread_id_t tid);

    bool is_thread_busy(thread_id_t tid);

    std::vector<Thread_Context> _thread_contexts; 
};

thread_server.cpp:

#include "thread_server.h"

void work(thread_id_t tid, std::vector<Thread_Context>* pcontexts) {
    auto& contexts = *pcontexts;
    while (!contexts[tid].kill) {
        while (contexts[tid].pause);
        auto cmd = contexts[tid].task_queue.front();
        if (cmd.has_value()) {
            cmd.value()();
            contexts[tid].task_queue.pop();
        }
    }
    contexts[tid].dead = true;
}

Thread_Server::Thread_Server() {

}
Thread_Server::~Thread_Server() {
    for (int i = 0; i < _thread_contexts.size(); i++) {
        wait_for_thread(i);
        _thread_contexts[i].kill = true;
    }
}

thread_id_t Thread_Server::make_thread(bool start) {
    thread_id_t tid = NULL_THREAD;
    for (thread_id_t i = 0; i < _thread_contexts.size(); i++) {
        if (_thread_contexts[i].dead) {
            _thread_contexts[i].dead = false;
            _thread_contexts[i].kill = false;
            _thread_contexts[i].pause = !start;
            _thread_contexts[i].task_queue.clear();
            tid = i;
            break;
        }
    }
    if (tid == NULL_THREAD) {
        tid = (thread_id_t)_thread_contexts.size();
        Thread_Context ctx;
        ctx.pause = !start;
        _thread_contexts.push_back(ctx);
    }

    std::thread(work, tid, &_thread_contexts).detach();
    return tid;
}

void Thread_Server::pause_thread(thread_id_t tid) {
    _thread_contexts[tid].pause = true;
}

void Thread_Server::start_thread(thread_id_t tid) {
    _thread_contexts[tid].pause = false;
}

void Thread_Server::kill_thread(thread_id_t tid) {
    _thread_contexts[tid].kill = true;
}

void Thread_Server::queue_task(thread_id_t tid, const Thread_Task& task) {
    auto& ctx = _thread_contexts[tid];
    ctx.task_queue.push(task);
}

void Thread_Server::wait_for_thread(thread_id_t tid) {
    auto& ctx = _thread_contexts[tid];
    while (ctx.task_queue.size() > 0);
}

bool Thread_Server::is_thread_busy(thread_id_t tid) {
    return _thread_contexts[tid].task_queue.size() > 0;
}

As you can see there are also functions to pause/start/kill and wait for a specific thread. This can be used to sync threads and wait for return values etc.

As a use case example, you could have one thread for rendering a loading symbol until another thread is finished e.g. downloading a file from the internet and then using the downloaded file:

void update() {

    if (want_to_download_something) {
        thread_server.queue_task(download_thread, [url]() {
            download_result = download(url);
        });
    }

    if (thread_server.is_thread_busy(download_thread)) {
        render_loading_icon("Downloading file");
    } else if (download_result) {
        do_something(download_result);
    }
}

And here's a quick test to make sure it's working correctly:

#include <iostream>

#include "thread_server.h"

int main()
{
    Thread_Server thread_server;
    auto t1 = thread_server.make_thread();
    auto t2 = thread_server.make_thread();
    auto t3 = thread_server.make_thread();

    for (int i = 0; i < 10; i++) {
        thread_server.queue_task(t1, []() {
            std::cout << "\nHello from thread 1 (std::this_thread::get_id(): " << std::this_thread::get_id() << " )\n";
        });

        thread_server.queue_task(t2, []() {
            std::cout << "\nHello from thread 2 (std::this_thread::get_id(): " << std::this_thread::get_id() << " )\n";
        });

        thread_server.queue_task(t3, []() {
            std::cout << "\nHello from thread 3 (std::this_thread::get_id(): " << std::this_thread::get_id() << " )\n";
        });
    }

    std::cin.get();
}

Edit: I'm fairly new to multithreading so if you have anything to say about the safety of my code I'll be more than glad to hear it.

Answer 1

Overview

So I created a thread server that creates threads and associates them with a handle so that you can keep specific threads for specific tasks e.g. run all graphics rendering on thread 0, run physics stepping on thread 1 and use thread 2 for downloading a file of the internet etc.

I don't think that is a good idea in general. Though it is an easy way to think about things and may make it a good learning experience.

In general thread 2 will sit around doing nothing. Thread 0 will either be exceedingly busy or doing nothing and thread 1 will probably be clobbered with all the work.

I don't know about the physics stuff. Have no idea how that would work. But the other two (Graphics/ Internet) are both event driven activities. They would be best served with an event system Unless you plan on writing this yourself (non trivial) then use somebody else.

Now saying that. Both of these may be their own separate event loops with a thread each. But what usually happens is that you have a master thread the constantly runs the event loop then when an action happens the master thread creates a creates a job (work item) which is handed to a work queue. You then have a bunch of threads in the work queue that grab jobs as they appear in the queue and simply execute them.

Personally I've found it useful in a project of mine and I highly prefer this to a typical job system where you don't have any control over what thread the task is executed on.

Yes it will be easier. But it sounds like you have to much global state. You should be wrapping state in work items not having global state that can be messed up by multiple threads.

I'd like to know what room for improvement there is, especially in optimization and whether or not you think the overhead is justified.

Lets take a look :-)

Each thread executes tasks in a task queue, so I made a Thread_Safe_Queue which is just a wrapper for std::queue but with a mutex before read/writes:

Code review.

Don't you want to lock the other queues here?

    Thread_Safe_Queue(Thread_Safe_Queue<T>&& other) noexcept {
        std::lock_guard<std::mutex> lock(mutex);
        queue = std::move(other.queue);
    }

Its not really thread safe if you lock the destination (which is not fully formed so can not have been handed to another thread) but the source is still being mutated possibly another thread.

---

Do you really want to be able to copy the queues?

    Thread_Safe_Queue(const Thread_Safe_Queue<T>& other) {
        std::lock_guard<std::mutex> lock(mutex);
        queue = other.queue;
    }

You should still lock the source!

---

So you wrapped the queue so that you could add lock_guards on each method. Fine. But a bit wasteful. A call to empty() will tell you if the queue is empty at that point but a subsequent pop can not guarantee that it is still empty as you released the lock between the call to empty and the call to pop.

    std::optional<T> pop() {
    std::optional<T> front() {
    void push(const T &item) {
    bool empty() const {
    void clear() {

I would write a queue that performs at a higher level. How about a blocking queue. Want to pop an item. If there is no item to pop the thread is blocked until there is one. Or will wait a minimum amount of time for the object to appear.

---

Interesting:

constexpr thread_id_t NULL_THREAD = (thread_id_t)0 - (thread_id_t)1;

Is this a complex way of writing:

constexpr thread_id_t NULL_THREAD = static_cast<thread_id_t>(-1);

Two things.

• Avoid all uppercase identifiers. Technically these are reserved for macros.
• Prefer to use C++ casts rather than C casts.

---

Why are you passing by pointer?

void work(thread_id_t tid, std::vector<Thread_Context>* pcontexts) {

Are you passing ownership (then use std::unique_ptr<>). Can the passed object be nullptr (does not look like it you don't check it for null).

Pass by reference rather than pointer. Then explicitly means you are not passing ownership and the called function should not delete the pointer. Otherwise there is confusion on if the work() function should or should not delete the pointer.

---

This look like a bug

        while (contexts[tid].pause);  // Loop forever !

This is equivalent to:

        while (contexts[tid].pause)
        {}

You hit the loop. The body changes no state so the loop can not be exited.

---

void work(thread_id_t tid, std::vector<Thread_Context>* pcontexts) {

Had to read forward to understand this.
Sorry this is broken.

You try and get around the fact that a vector may reallocate its space by passing an index to the work item in the vector. The problem here is that there is no access restriction once the thread is creating and a new thread (created with make_thread()) can cause the pcontexts to be resized at anytime. Access to a vector is not thread safe so if the vector is in the middle of being re-sized then accesses to its members via operator[] is not guaranteed to be valid.

---

Assigning a thread to a single work item is not very productive. A thread is a relatively heavyweight object so you don't want to be creating them willy nilly when new work items are created.

void work(thread_id_t tid, std::vector<Thread_Context>* pcontexts) {
    auto& contexts = *pcontexts;
    while (!contexts[tid].kill) {
        while (contexts[tid].pause);
        auto cmd = contexts[tid].task_queue.front();
        if (cmd.has_value()) {
            cmd.value()();
            contexts[tid].task_queue.pop();
        }
    }
    contexts[tid].dead = true;
}

You should create a bunch of work threads then let them pick up work items from the queue. When they have finished hold them with a condition variable until there is work available.

---

Expectations

void actionToUploadFile()
{
    workEventLoop.addItem([url]{
        guiEvenLoop.addAlert('Downloading');
        download(url);
        guiEvenLoop.delAlert();
        workEvenLoop.addItem(do_something);
    });
}