6
\$\begingroup\$

Topic: C++ Implementation of Lock-Free Queue and ThreadPool classes

In a project I'm currently working on, I need the implementation of ThreadPool along with a thread-safe queue. I've tried implementing the above structures and wonder about their safety and performance.

Reason for asking

I'm looking for your review to ensure it's bug-free and well-constructed. As far as I know, everything is working well, but I feel like there is much to improve, as I'm a beginner in this field, and probably I can miss something pretty easily.

Lock free queue

lock_free_queue.hpp:

#include <deque>
#include <mutex>
#include <condition_variable>

template<typename T>
class LockFreeQueue {
private:
    std::deque<T> queue;
    std::mutex mutex;
    std::condition_variable element_available;
    std::condition_variable space_available; // keep control of queue size
    std::atomic_uint active_producers; // to throw poison pill when needed
    size_t max_size{};

public:

    LockFreeQueue() : max_size(std::numeric_limits<size_t>::max()), active_producers(1) {}

    explicit LockFreeQueue(size_t max_size, unsigned active_producers = 1) :
    max_size(max_size), active_producers(active_producers) {}

    ~LockFreeQueue() = default;

    LockFreeQueue(const LockFreeQueue &) = delete;

    LockFreeQueue &operator=(const LockFreeQueue &) = delete;

    template<typename A>
    void push(A &&item) {
        {
            std::unique_lock<std::mutex> lock(mutex);
            space_available.wait(lock, [this] { return queue.size() < max_size; });
            queue.push_back(std::forward<A>(item));
        }
        element_available.notify_one();
    }

    T pop() {
        T item;
        {
            std::unique_lock<std::mutex> lock(mutex);
            while (queue.empty()) {
                if (active_producers.load(std::memory_order_acquire) == 0) {
                    return {}; // poison pill
                }
                element_available.wait(lock);
            }
            item = std::move(queue.front());
            queue.pop_front();
        }
        space_available.notify_one();
        return item;
    }

    bool try_pop(T &item) {
        std::unique_lock<std::mutex> lock(mutex);
        if (queue.empty()) {
            return false;
        }
        item = std::move(queue.front());
        queue.pop_front();
        return true;
    }

    void shutdown() {
        active_producers.fetch_sub(1, std::memory_order_release);
        element_available.notify_all();
    }

    [[maybe_unused]] bool empty() {
        std::unique_lock<std::mutex> lock(mutex);
        return queue.empty();
    }

    [[maybe_unused]] size_t size() {
        std::unique_lock<std::mutex> lock(mutex);
        return queue.size();
    }

    [[maybe_unused]] void clear() {
        std::unique_lock<std::mutex> lock(mutex);
        queue.clear();
    }
};

Thread pool

thread_pool.hpp:

#include "lock_free_queue.hpp"
#include <atomic>
#include <vector>
#include <thread>
#include <functional>
#include <future>
#include <cstddef>

using job_t = std::function<void()>;

class JoinThreads {
    std::vector<std::thread> &threads;
public:
    explicit JoinThreads(std::vector<std::thread> &to_join);

    ~JoinThreads();
};

class ThreadPool {
    std::atomic_bool done;
    LockFreeQueue<job_t> jobs;
    std::vector<std::thread> threads;
    JoinThreads joiner;
    std::condition_variable cv;
    std::mutex cv_m;
    size_t pending_jobs = 0;

    void worker_thread();

    void job_done();

public:
    explicit ThreadPool(size_t num_threads);

    ~ThreadPool();

    template<typename func_T>
    void submit(func_T &&f) {
        jobs.push(std::forward<func_T>(f));
        ++pending_jobs;
    }

    void wait();
};

thread_pool.cpp:

#include "thread_pool.hpp"
#include <iostream>

using job_t = std::function<void()>;

JoinThreads::JoinThreads(std::vector<std::thread> &to_join) : threads(to_join) {}

JoinThreads::~JoinThreads() {
    for (auto &thread: threads) {
        if (thread.joinable()) {
            thread.join();
        }
    }
}

void ThreadPool::worker_thread() {
    while (!done) {
        job_t job;
        if (jobs.try_pop(job)) {
            job();
            job_done();
        } else {
            std::this_thread::yield();
        }
    }
}

void ThreadPool::job_done() {
    {
        std::unique_lock<std::mutex> lock(cv_m);
        --pending_jobs;
    }
    cv.notify_all();
}

void ThreadPool::wait() {
    std::unique_lock<std::mutex> lock(cv_m);
    cv.wait(lock, [this] { return pending_jobs == 0; });
}

ThreadPool::ThreadPool(const size_t num_threads): done(false), joiner(threads) {
    try {
        for (size_t i = 0; i < num_threads; ++i) {
            threads.emplace_back(&ThreadPool::worker_thread, this);
        }
    } catch (...) {
        done = true;
        throw;
    }
}

ThreadPool::~ThreadPool() {
    jobs.shutdown();
    wait();
    done = true;
}

Usage example

#include "thread_pool.hpp"


ThreadPool pool(num_threads);

pool.submit([this] { some_method(); });
pool.submit([this] { another_method(); });

Important

I'm also particularly curious if I've correctly implemented the wait() mechanism.

Thank you! I will be grateful for any advice and comments.

\$\endgroup\$

2 Answers 2

6
\$\begingroup\$

Naming things

The name LockFreeQueue is a lie, it uses std::mutex to lock so it is not lock-free. However, this class does provide thread-safety, so a more appropriate name would be ThreadSafeQueue.

Don't mix atomics and mutexes

Avoid mixing atomic variables and mutexes. Doing so will create multiple synchronization scopes. Since you already have a mutex, I suggest that active_producers is only read from and written to while holding mutex, so there is no need to make it atomic. Just lock the mutex in shutdown().

There is an actual problem in your code. Consider a thread that's inside the while-loop in pop():

if (active_producers.load(std::memory_order_acquire) == 0) {
    return {}; // poison pill
}
// <-- thread is currently right here
element_available.wait(lock);

It has just checked active_producers and it was equal to 1. Now another thread calls shutdown(), which decrements active_producers and calls notify_all(). However, since the first thread hasn't started wait() yet, those notifications will have no effect. Now wait() is called, and it will block forever.

The correct way to handle this is to use wait() with a predicate, and to ensure everything this predicate checks is guarded by lock:

T pop() {
    T item;

    {
        std::unique_lock<std::mutex> lock(mutex);

        element_available.wait(lock, [&]{
            return !active_producers || !queue.empty();
        });

        if (queue.empty()) {
            // We can only reach this if there are no active producers left
            return {};
        }

        item = std::move(queue.front());
        queue.pop_front();
    }

    space_available.notify_one();
    return item;
}

void shutdown() {
    {
        std::unique_lock<std::mutex> lock(mutex);
        --active_producers;
    }

    element_available.notify_all();
}

About the use of attributes

I see you annotated some member functions with [[maybe_unused]]. That's quite unusual, and is unnecessary; compilers don't warn about class members functions that are not used.

What you can do is give empty(), size() and try_pop() the [[nodiscard]] attribute. That will ensure the compiler will warn if something calls those functions but doesn't look at their return value, which would almost certainly be an error.

Remove JoinThreads

I don't see the point of this class. Sure, it will join threads when it is destroyed, but you had to write a destructor to do that. Why not just move that destructor into ThreadPool itself? That would be less code to write, and is slightly more efficient because there is no need to store a reference to a vector of threads anymore.

Since C++20 you can use std::jthread: it's a single std::thread that joins when it is destroyed. You can create a vector of those, and that will automatically cause all threads to join when the vector is destroyed.

Don't busy-wait

Your ThreadPool::worker_thread() busy-waits if there are no jobs. I would just have it call pop() instead of try_pop(), and then check if the returned job_t contains a function:

void ThreadPool::worker_thread() {
    while (true) {
        job_t job = jobs.pop();
        if (!job)
            break;
        job();
    }
}

Rely on the queue to signal when work is done.

You also don't need to track pending_jobs and have a done variable at all; only when the queue is empty and there are no more producers will pop() return {}. That also means there is no need for ThreadPool::wait().

One issue to worry about is what to do when creating worker threads fails. You then need to ensure the queue is shut down properly. You could do that in the constructor of ThreadPool by calling jobs.shutdown() inside the catch-block.

Remove empty() and size()

While most standard containers provide empty() and size() member functions, there is a problem with those functions in a thread-safe queue. While you take the lock inside those functions, the lock is released right before these functions return. That means that by the time the caller looks at the return value, that value might no longer be representative of the state of the queue anymore; another thread might have come and added or removed items. Since you cannot rely on the return value of these functions, it is better to remove them entirely.

\$\endgroup\$
1
  • \$\begingroup\$ Thank you a lot for your review! This is exactly what I've been looking for. All your comments actually make perfect sense, and there's a lot to learn from. \$\endgroup\$
    – andylvua
    Apr 1, 2023 at 12:31
-2
\$\begingroup\$

tl;dr: Writing down invariants helps the Gentle Reader better understand the technical ideas which code is trying to communicate. To the extent that code resembles a page of a text book or a file on GitHub, it is useful to cite your references. When supplied unit tests do not exercise a line of code, that will not instill confidence in the Reader that the line runs correctly.

Dear downvoter: Let me explain myself. Additional -1's won't cause me to better understand the technical details, since G. Sliepen was kind enough to show how my technical analysis was inaccurate -- my answer is incorrectly stating that there is a locking problem in push(), which I could correct, at the cost of papering over the communication failure. Nor are -1's likely to motivate deletion of the answer, as I have already considered that a few times and feel it would be a disservice to the community. I'm not complaining about what seem well-deserved downvotes. Is this train wreck embarrassing? Sure. But why are you even reading this (or any) review? To get better at communicating. I want you to be a better communicator. We learn from both positive and negative examples. The OP successfully communicated certain details to the machine, and tried to communicate technical ideas to an audience of collaborators. In at least one instance it didn't work, for reasons described below. It's worth understanding that particular failure to communicate, so as to generalize and to avoid it in future works. As described in SICP,

programs must be written for people to read, and only incidentally for machines to execute.

When you write, do it so people will successfully read. That is a challenging task, hard to anticipate the outcome. OTOH whether the machine can read your intent is just a Green / Red indicator displayed when tests run. I offer this recent example of keeping the Reader in mind when authoring code, orthogonal to some technical details raised.


This codebase does not appear to be ready for review yet, in the sense that it does not offer a robust argument for correctness.

There seems to be an implicit invariant that operations won't overflow the fixed-size queue. Other than that we see very few explicit checks.

The usage example is not self-evaluating and is geared toward happy path. It does not attempt to induce races, nor stress or verify this module in any way. For core parallel algorithms such as this one, the burden of proof is much higher than what is presented in this submission.


Let's assume the queue is nearly full. Then push looks like:

  • acquire mutex
    • wait for space
    • append an item
  • release mutex

and pop looks like

  • acquire mutex
    • remove an item
  • release mutex

I don't get it. In particular, when the queue is full, push has frozen out all other threads including instances of pop. And it waits for a queue item to drain. The only way that can happen is within the critical section of pop. But we're holding the mutex, so that critical section won't execute. How can this be good? [Please refer to the comments below, which explain that this analysis is just wrong.]


I am unwilling to believe that this code achieves its design goals. Recommend that it not be merged to main.

\$\endgroup\$
3
  • 3
    \$\begingroup\$ push() won't indefinitely block if the queue is full because wait() will unlock the mutex while waiting. \$\endgroup\$
    – G. Sliepen
    Apr 1, 2023 at 7:44
  • 1
    \$\begingroup\$ Thank you. I withdraw my objection . Looks like the code would benefit from greater use of comments. \$\endgroup\$
    – J_H
    Apr 1, 2023 at 14:14
  • 1
    \$\begingroup\$ One person can only downvote once, so you have three downvoters. I think the main reason for that is because your answer is incorrectly stating that there is a locking problem in push(). Instead of complaining about downvotes, your answer should just be fixed. \$\endgroup\$
    – G. Sliepen
    Apr 2, 2023 at 21:43

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.