Job_Queue (Thread_Pool) Program

Question

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;
}

Answer 1

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