4
\$\begingroup\$

I'm new to std::atomic, std::mutex, std::unique_lock, std::condition_variable and more or less c++11 std::thread, so I wrote this little Job_Queue class, where you can submit a void() function and it will be distributed among some threads. I wonder if there's room for improvement or if I made to beginner's mistakes. The idea is, that you can have multiple job_queues waiting, but not running at the same time.

job_queue.hpp

#ifndef SNIPPETS_JOB_QUEUE_HPP
#define SNIPPETS_JOB_QUEUE_HPP

#include <atomic>
#include <condition_variable>
#include <functional>
#include <iostream>
#include <mutex>
#include <queue>
#include <thread>
#include <vector>


class Job_Queue
{
public:
    static std::mutex job_mutex;
    void start();
    void stop();
    void terminate();
    Job_Queue();

    void add(const std::function<void()>& f);
    void add(std::function<void()>&& f);

private:
    static bool                         is_running;
    static unsigned int                 n_cores;

    std::atomic<bool>                   terminate_after_current_job;
    std::atomic<bool>                   stop_when_queue_is_empty;
    std::vector<std::thread>            threads;

    std::condition_variable             thread_cv;
    std::mutex                          queue_mutex;
    std::queue<std::function<void()>>   job_queue;

    void thread_loop();
};


#endif //SNIPPETS_JOB_QUEUE_HPP

job_queue.cpp

#include "job_queue.hpp"
void Job_Queue::start()
{
    if( is_running )
    {
        std::cerr << "Error: Another JobQueue is currently running. Please end it first.\n";
        return;
    }
    is_running = true;
    threads.clear();
    terminate_after_current_job = false;
    stop_when_queue_is_empty = false;
    threads.reserve(n_cores);
    for( unsigned int k = 0; k < n_cores; ++k )
    {
        threads.emplace_back(&Job_Queue::thread_loop, this);
    }
}

void Job_Queue::terminate()
{
    terminate_after_current_job = true;
    stop();
}

void Job_Queue::stop()
{
    stop_when_queue_is_empty = true;
    thread_cv.notify_all();
    for( unsigned int k = 0; k < n_cores; ++k )
    {
        thread_cv.notify_all();
        threads[k].join();
    }
    threads.clear();
    is_running = false;
}

void Job_Queue::add(const std::function<void()>& f)
{
    job_queue.push(f);
    thread_cv.notify_all();
}

void Job_Queue::add(std::function<void()>&& f)
{
    job_queue.emplace(f);
    thread_cv.notify_all();
}

Job_Queue::Job_Queue()
: queue_mutex(std::mutex())
{
    if ( Job_Queue::n_cores == 0)
    {
        n_cores = std::thread::hardware_concurrency();
        if (n_cores == 0)
        {
            std::cerr << "Error: Could not find number of available cores.\n";
        }
    }
}




void Job_Queue::thread_loop()
{
    {
        std::unique_lock<std::mutex> mu(Job_Queue::job_mutex);
    }
    bool job_q_empty = false;
    while( true )
    {
        {
            std::unique_lock<std::mutex> lock(queue_mutex);
            thread_cv.wait(lock, [this, &job_q_empty] { job_q_empty = job_queue.empty(); return !job_queue.empty() || terminate_after_current_job || stop_when_queue_is_empty; });
        }
        if( !job_q_empty )
        {

            std::function<void()> f;
            {
                std::unique_lock<std::mutex> lock(queue_mutex);
                f = job_queue.front();
                job_queue.pop();
                std::cout << "jobs left in Q: " << job_queue.size() << "\n";
            }
            f();
        }
        else
        {
            if( stop_when_queue_is_empty )
            {
                break;
            }
        }
        if( terminate_after_current_job )
        {
            break;
        }
    }
}

std::mutex Job_Queue::job_mutex = std::mutex();
bool Job_Queue::is_running = false;
unsigned int Job_Queue::n_cores = 0;

main.cpp

#include "job_queue.hpp"

void short_job(std::size_t i)
{
    std::this_thread::sleep_for(std::chrono::seconds(5));
    std::unique_lock<std::mutex> lock(Job_Queue::job_mutex);
    std::cout << "Finished S job: " << i << " on thread: " << std::this_thread::get_id() << "\n" << std::flush;
}

void middle_job(std::size_t i)
{
    std::this_thread::sleep_for(std::chrono::seconds(10));
    std::unique_lock<std::mutex> lock(Job_Queue::job_mutex);
    std::cout << "Finished M job: " << i << " on thread: " << std::this_thread::get_id() << "\n" << std::flush;
}

void long_job(std::size_t i)
{
    std::this_thread::sleep_for(std::chrono::seconds(15));
    std::unique_lock<std::mutex> lock(Job_Queue::job_mutex);
    std::cout << "Finished L job: " << i << " on thread: " << std::this_thread::get_id() << "\n" << std::flush;
}

int main()
{
    Job_Queue job_queue;
    job_queue.start();
    for( std::size_t k = 0; k < 100; ++k )
    {
        if( k%3 == 0 )
        {
            job_queue.add(std::move(std::bind(short_job, k)));
        }
        else if( k%3 == 1 )
        {
            job_queue.add(std::move(std::bind(middle_job, k)));
        }
        else
        {
            job_queue.add(std::move(std::bind(long_job, k)));
        }
    }
    job_queue.stop();
    return 0;
}
\$\endgroup\$

1 Answer 1

4
\$\begingroup\$

Observations:

I must say I was never very fond of std::bind and since the advent of lambdas I don't really see much need for it.

        job_queue.add(std::move(std::bind(short_job, k)));
        job_queue.add(std::move(std::bind(middle_job, k)));
        job_queue.add(std::move(std::bind(long_job, k)));

I would write:

        job_queue.add([k](){short_job(k);});
        job_queue.add([k](){middle_job(k);});
        job_queue.add([k](){long_job(k);});

Don't see the need for a start/stop paradigm:

{
    Job_Queue job_queue;
    job_queue.start();
    // STUFF
    job_queue.stop();
}

That is what we have constructor/destructor for.

{
    Job_Queue job_queue;   // Started by constructor.
    // STUFF
}                          // Stopped by destructor

The advantage of this is that the user can't accidentally forget to call start/stop. It is done automatically for you.


Why do these user functions have accesses to the queues lock?

void short_job(std::size_t i)
{
    std::this_thread::sleep_for(std::chrono::seconds(5));
    std::unique_lock<std::mutex> lock(Job_Queue::job_mutex);     // WHY !!!!!! 
                                                                 // After reading the code
                                                                 // it is not used by the class
                                                                 // just the external functions
                                                                 // So it should not be a
                                                                 // member of the class it should
                                                                 // simply defined before these
                                                                 // functions.


    std::cout << "Finished S job: " << i << " on thread: " << std::this_thread::get_id() << "\n" << std::flush;
}

Code Review

You seem to have forgotten your namespace!!!


I would not have made the mutex static.

    static std::mutex job_mutex;

Make it a normal member of the class. Otherwise all your queues will be waiting on other queues to finish using the lock.


Why are these static.

static bool                         is_running;
static unsigned int                 n_cores;

I can't have more than one job queue?


As mentioned above I would not have an explicit start stop.

    void start();
    void stop();

This is likely to cause issues with incorrect usage.


Just have the move version of add.

    void add(const std::function<void()>& f);
    void add(std::function<void()>&& f);

Why can't I have multiple queues?

    if( is_running )
    {
        std::cerr << "Error: Another JobQueue is currently running. Please end it first.\n";

        // Does not actually stop you having multiple queues!
        // Just seems to print a warning message.
        // The code is still running
        return;
    }

void Job_Queue::add(std::function<void()>&& f)
{
    job_queue.emplace(f);
    thread_cv.notify_all();     // There is only one new job.
                                // You don't need to wake all the threads.
                                // Simply notify one thread.
}

Why are you using the copy constructor!

: queue_mutex(std::mutex())

This is the same as:

: queue_mutex()

Why are you doing this here?

{
    if ( Job_Queue::n_cores == 0)
    {
        n_cores = std::thread::hardware_concurrency();
    }
}

Why not simply do this when you declare c_cores?

    unsigned int Job_Queue::n_cores = std::thread::hardware_concurrency();

Is this an actual issue?

        if (n_cores == 0)
        {
            std::cerr << "Error: Could not find number of available cores.\n";
        }

Will this ever return 0?
Can it return 0? Will a machine not always have at least one core? If it had zero cores can you run the code?


This is pointless:

    {
        std::unique_lock<std::mutex> mu(Job_Queue::job_mutex);
    }

You are declaring a variable outside the lambda. Passing it in by reference. Then just using it like a local variable in the lambda.

    bool job_q_empty = false;

Why? Just declare job_q_empty boolean inside the lambda!

Your current lambda looks like this:

bool job_q_empty = false;
[this, &job_q_empty]()
{
    job_q_empty = job_queue.empty();
    return !job_queue.empty() || 
            terminate_after_current_job || 
            stop_when_queue_is_empty;
}

You could have simply done this:

[this]()
{
    bool job_q_empty = job_queue.empty();      // Notice the bool here.
    return !job_queue.empty() || 
            terminate_after_current_job || 
            stop_when_queue_is_empty;
}

You are locking/unlocking/re-locking/unlocking the queue. The problem is that the check used in the test() includes job_queue.empty() and you release the lock and thus allow other threads to be released from the queue before this thread has extracted its job from the queue.

THIS IS A SERIOUS BUG

        {
            std::unique_lock<std::mutex> lock(queue_mutex);
            thread_cv.wait(lock, test());
        }
        if( !job_q_empty )
        {
            std::function<void()> f;
            {
                std::unique_lock<std::mutex> lock(queue_mutex);
                f = job_queue.front();
                job_queue.pop();
                std::cout << "jobs left in Q: " << job_queue.size() << "\n";
            }
            f();
        }

I would do this:

void Job_Queue::thread_loop()
{
   if (!terminated)
   {
        std::function<void()> nextJob = getNextJob();
        // Either get a job.
        // Or a do nothing empty job. Simply run it either way.
        nextJob();
   }
}


std::function<void()> Job_Queue::getNextJob()
{
    std::unique_lock<std::mutex> lock(queue_mutex);
    thread_cv.wait(lock, [&job_queue](){return !job_queue.empty() || terminated;});

    std::function<void()> result = [](){}; // default empty job.
    if (!terminated) {
        // If its not terminated then we know there is a job
        // in the job_queue to get the front one.
        result = job_queue.front();
        job_queue.pop();
    }
    return result;
}

You can check out my attempt here:

A simple Thread Pool

\$\endgroup\$
5
  • \$\begingroup\$ Thanks for pointing out some obvious flaws! I will definitely adapt! To address some of your (very welcomed) critic: 1) You can have multiple Queues, but not have multple Queues running. The reason is, if one Queue uses all threads, then multiple queues would need to share processor power, and ultimately everything slows down. 2) hardware_concurrency() can return 0. 3) Thanks for pointing out the bug, that is indeed quite severe. 4) The variables are static so that through all instances it is known whether it's running \$\endgroup\$ Jun 1, 2020 at 10:00
  • \$\begingroup\$ You could want to create a queue, but not run it yet, until resources are available. That's why there's the start function. I will add the destructor so it stops automatically. \$\endgroup\$ Jun 1, 2020 at 10:37
  • \$\begingroup\$ The reason is, if one Queue uses all threads, then multiple queues would need to share processor power Are confusing multiple threads with multiple cores. You can have many more threads than cores (not a good idea but I should be able to do it). \$\endgroup\$ Jun 1, 2020 at 15:19
  • \$\begingroup\$ then multiple queues would need to share processor power, and ultimately everything slows down You are making the assumption that all queues are being used at full capacity all the time. That is rarely the case. Threads can be blocked on operations then you have wasted parallelism. As the generic queue builder you can't know what an application is doing. I as the specific application developer will have more context and may be able to utilize two queues effectively. \$\endgroup\$ Jun 1, 2020 at 15:22
  • \$\begingroup\$ Thinking on this more. Sure you can design this to have one and only one thread queue (that is a design decision). But if that is your use case you should be creating a singleton object. Note: Use a factory method to construct your singleton so that you can replace it with an alternative in debug mode. \$\endgroup\$ Jun 2, 2020 at 0:06

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.