2
\$\begingroup\$

I created a class which is some kind of thread pool. The difference is that it holds foreign threads. This means that the threads are created somewhere else and are then held by calling holdThread. Then another thread can call executeOne or executeAll to execute a specific function. At the end the threads can be released by calling release or when the object is destroyed.

Now I'm not entirely sure about the race-condition-safeness of my code. Also, Valgrind gives me a warning about nextFunction, but I think as the code is designed, it doesn't need to be protected. From my testings it worked fine as it is now.

#ifndef THREADCONTROL_HPP
#define THREADCONTROL_HPP

#include <functional>
#include <atomic>
#include <mutex>
#include <condition_variable>
#include <unistd.h>

class ThreadException
{
public:
    ThreadException(const std::string &str_message) : message(str_message) {}
    std::string message;
};

class ThreadControl
{

public:
    ThreadControl() :
        nextFunction(),
        running(true),
        totalThreads(0),
        busy(0),
        cv(),
        cm()
{}

~ThreadControl()
{
    release();
}

/**
  * Releases all threads
 **/
void release()
{
    running = false;
    nextFunction = std::function<void()>();
    cv.notify_all();
}

/**
  * Controls the current thread until release is called
 **/
void holdThread()
{
    //The busy counter needs to be set while waiting for
    //the condition variable in order to avoid dead locks
    //The very first time the busy counter doesn't need to be decreased
    bool needToDecrement = false;

    ++totalThreads; //Count total threads
    while(running)
    {
        //Wait for condition
        {
            std::unique_lock<std::mutex> lock(cm);
            if(needToDecrement)--busy;
            cv.wait(lock);
        }

        //Execute desired function
        if(nextFunction)
        {
            ++busy;
            try {
                nextFunction();
            } catch(...) {
                //Rethrow all exceptions but set counters before
                    --busy;
                    --totalThreads;
                    throw;
            }
            needToDecrement = true;
        }
    }
    --totalThreads;
}

//gcc < 4.8 has a bug with lambdas and variadic templates
#if defined(__GNUC__) && __GNUC__ >= 4 && __GNUC_MINOR__ > 7

/**
  * Executes the function by one thread
 **/
template<typename function_definition, typename ...args>
void executeOne(function_definition fn, args ...a)
{
    if(totalThreads == 0)throw ThreadException("You need to have at least one thread");

    //Flag to determine if thread was started
    std::atomic_bool started(false);

    //Set function to start
    nextFunction = [fn,&a...,&started]()
    {
        started = true; //Thread "started"
        fn(a...);       //Execute desired function
    };

    //Notify one thread
    cv.notify_one();

    //Wait until the thread sets started to treu
    while(!started)
    {
        //If no thread was available wait until one gets available
        if(totalThreads - busy == 0)usleep(1000);
    }
}

/**
  * Executes the given function by all available threads.
  * Could also be executed bo 0 threads. The amount of threads
  * is returned
 **/
template<typename function_definition, typename ...args>
int executeAll(function_definition fn, args ...a)
{
    //Determine currently available threads
    int availableThreads = totalThreads - busy;

    //Count the amount of threads started
    std::atomic_int started(0);

    //Set function to be called
    nextFunction = [fn,&a...,&started]()
    {
        ++started;  //Tell the main thread that I'm started
        fn(a...);   //Execute desired function
    };

    //Notify all threads
    cv.notify_all();

    //Wait until all threads are started
    while(started < availableThreads);

    return availableThreads;
}
#else
/**
  * Hack for gcc < 4.8
 **/
template<typename function_definition>
void executeOne(function_definition fn)
{
    if(totalThreads == 0)throw ThreadException("You need to have at least one thread");

    //Flag to determine if thread was started
    std::atomic_bool started(false);

    //Set function to start
    nextFunction = [fn,,&started]()
    {
        started = true; //Thread "started"
        fn();           //Execute desired function
    };

    //Notify one thread
    cv.notify_one();

    //Wait until the thread sets started to treu
    while(!started)
    {
        //If no thread was available wait until one gets available
        if(totalThreads - busy == 0)usleep(1000);
    }
}

/**
  * Hack for gcc
 **/
template<typename function_definition>
int executeAll(function_definition fn)
{
    //Determine currently available threads
    int availableThreads = totalThreads - busy;

    //Count the amount of threads started
    std::atomic_int started(0);

    //Set function to be called
    nextFunction = [fn,&a...,&started]()
    {
        ++started;  //Tell the main thread that I'm started
        fn(a...);   //Execute desired function
    };

    //Notify all threads
    cv.notify_all();

    //Wait until all threads are started
    while(started < availableThreads);

    return availableThreads;
}
#endif

/**
  * Waits until all threads have completed their jobs
 **/
void waitForAll() const
{
    while(busy > 0)usleep(1000);
}

/**
  * Returns the total amount of threads hold by this class
 **/
unsigned int getTotalThreads() const
{
    return totalThreads;
}

/**
  * Returns the amount of currently busy threads
 **/
unsigned int getBusyThreads() const
{
    return busy;
}

/**
  * Returns the amount of currently available threads
 **/
unsigned int getAvailableThreads() const
{
    return totalThreads - busy;
}

private:

/**
  * The desired function to be executed
 **/
std::function<void()> nextFunction;

/**
  * A flag whether the ThreadControl is currently active
 **/
std::atomic_bool running;

/**
  * The total amount of threads
 **/
std::atomic_int totalThreads;

/**
  * The amount of currently working threads
 **/
std::atomic_int busy;

/**
  * The condition variable which is used to notify the threads
 **/
std::condition_variable cv;

/**
  * The mutex which is used for the condition variable to
  * notify the threads
 **/
std::mutex cm;

}; //end class ThreadControl

#endif //THREADCONTROL_HPP

Here is how it could be used:

#include "threadcontrol.hpp"
#include <iostream>
#include <thread>
#include <vector>
#include <map>
#include <mutex>

using namespace std;

int main(int argc, char *args[])
{
ThreadControl c;
vector<thread> threads(5);

for(size_t i = 0; i < threads.size(); ++i)
    threads[i] = thread([&c]() { c.holdThread(); });

string text;
while(text != "exit")
{
    cout<<"Type something (exit will exit)"<<endl;
    cin>>text;

    c.executeAll([](const string &text)
    {
        cout<<"Workerthread says "<<text<<endl;
        sleep(1);
    }, text);

    c.waitForAll();
}

c.release();

for(size_t i = 0; i < threads.size(); ++i)
    threads[i].join();
}
\$\endgroup\$

Your Answer

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

Browse other questions tagged or ask your own question.