thread that executes function calls from a main thread c++11

Question

I want to implement a thread that can accept function pointers from a main thread and execute them serially. My idea was to use a struct that keeps the function pointer and its object and keep pushing it to a queue. This can be encapsulated in a class. The task thread can then pop from the queue and process it. I also need to synchronize it(so it doesnt block the main thread?), so I was thinking of using a semaphore. Although I have a decent idea of the structure of the program, I am having trouble coding this up, especially the threading and semaphore sync in C++11. It'd be great if someone can suggest an outline by which I can go about implementing this.

EDIT: The duplicate question answers the question about creating a thread pool. It looks like multiple threads are being created to do some work. I only need one thread that can queue function pointers and process them in the order they are received.

Answer 1

Check this code snippet, I have implemented without using a class though. See if it helps a bit. Conditional variable could be avoided here, but I want the reader thread to poll only when there is a signal from the writer so that CPU cycles in the reader won't be wasted.

#include <iostream>
#include <functional>
#include <mutex>
#include <thread>
#include <queue>
#include <chrono>
#include <condition_variable>

using namespace std;

typedef function<void(void)> task_t;

queue<task_t> tasks;
mutex mu;
condition_variable cv;

bool stop = false;

void writer()
{
    while(!stop)
    {
        {
            unique_lock<mutex> lock(mu);
            task_t task = [](){ this_thread::sleep_for(chrono::milliseconds(100ms));   };
            tasks.push(task);
            cv.notify_one();
        }

        this_thread::sleep_for(chrono::milliseconds(500ms)); // writes every 500ms
    }
}

void reader()
{
    while(!stop)
    {
        unique_lock<mutex> lock(mu);
        cv.wait(lock,[]() { return !stop;});  
        while( !tasks.empty() )
        {

            auto task = tasks.front();            
            tasks.pop();
            lock.unlock();
            task();
            lock.lock();
        }

    }
}

int main()
{
    thread writer_thread([]() { writer();}  );
    thread reader_thread([]() { reader();}  );

    this_thread::sleep_for(chrono::seconds(3s)); // main other task

    stop = true;


    writer_thread.join();
    reader_thread.join();
}

Answer 2

Your problem has 2 parts. Storing the list of jobs and manipulating the jobs list in a threadsafe way.

For the first part, look into std::function, std::bind, and std::ref.

For the second part, this is similar to the producer/consumer problem. You can implement a semaphore using std::mutexand std::condition_variable.

There's a hint/outline. Now my full answer...

Step 1)

Store your function pointers in a queue of std::function.

std::queue<std::function<void()>>

Each element in the queue is a function that takes no arguments and returns void.

For functions that take arguments, use std::bind to bind the arguments.

void testfunc(int n);
...
int mynum = 5;
std::function<void()> f = std::bind(testfunction, mynum);

When f is invoked, i.e. f(), 5 will be passed as argument 1 to testfunc. std::bind copies mynum by value immediately.

You probably will want to be able to pass variables by reference as well. This is useful for getting results back from functions as well as passing in shared synchronization devices like semaphores and conditions. Use std::ref, the reference wrapper.

void testfunc2(int& n);  // function takes n by ref
...
int a = 5;
std::function<void()> f = std::bind(testfunction, std::ref(a));

std::function and std::bind can work with any callables--functions, functors, or lambdas--which is pretty neat!

Step 2)

A worker thread dequeues while the queue is non-empty. Your code should look similar to the producer/consumer problem.

class AsyncWorker
{
    ...

public:
    // called by main thread
    AddJob(std::function<void()> f)
    {
        {
            std::lock_guard<std::mutex> lock(m_mutex);
            m_queue.push(std::move(f));
            ++m_numJobs;
        }
        m_condition.notify_one();  // It's good style to call notify_one when not holding the lock. 
    }

private:
    worker_main()
    {
        while(!m_exitCondition)
            doJob();
    }

    void doJob()
    {
        std::function<void()> f;
        {
            std::unique_lock<std::mutex> lock(m_mutex);
            while (m_numJobs == 0)
                m_condition.wait(lock);

            if (m_exitCondition)
                return;

            f = std::move(m_queue.front());
            m_queue.pop();
            --m_numJobs;
        }
        f();
    }

    ...

Note 1: The synchronization code...with m_mutex, m_condition, and m_numJobs...is essentially what you have to use to implement a semaphore in C++'11. What I did here is more efficient than using a separate semaphore class because only 1 lock is locked. (A semaphore would have its own lock and you would still have to lock the shared queue).

Note 2: You can easily add additional worker threads.

Note 3: m_exitCondition in my example is an std::atomic<bool>

Actually setting up the AddJob function in a polymorphic way gets into C++'11 variadic templates and perfect forwarding...

class AsyncWorker
{
    ...

public:
    // called by main thread
    template <typename FUNCTOR, typename... ARGS>
    AddJob(FUNCTOR&& functor, ARGS&&... args)
    {
        std::function<void()> f(std::bind(std::forward<FUNCTOR>(functor), std::forward<ARGS&&>(args)...));
        {
            std::lock_guard<std::mutex> lock(m_mutex);
            m_queue.push(std::move(f));
            ++m_numJobs;
        }
        m_condition.notify_one();  // It's good style to call notify_one when not holding the lock. 
    }

I think it may work if you just used pass-by-value instead of using the forwarding references, but I haven't tested this, while I know the perfect forwarding works great. Avoiding perfect forwarding may make the concept slightly less confusing but the code won't be much different...