boost thread worker object re-use after thread has

I have a simple thread object which takes care of some execution (a worker):

In the simplest form, an object is created for every single thread:

class worker
{
public:

    worker  (
              boost::atomic<int> & threads,
              boost::mutex & mutex,
              boost::condition_variable & condition
            )
    : threads__(threads), mutex__(mutex), condition__(condition)
    {}  

    void run (
                 // some params
             )
    {   
        // ... do the threaded work here
        // finally, decrease number of running threads and notify
        boost::mutex::scoped_lock lock(mutex__);
        threads__--;
        condition__.notify_one();
    }   

private:
    boost::atomic<int> & threads__;
    boost::mutex & mutex__;
    boost::condition_variable & condition__;
};

The way I use it is within a loop which runs maximum 8 concurrent threads, and waits to be notified if one has finished, in order to spawn the next one:

boost::thread_group thread_group;
boost::mutex mutex;
boost::condition_variable condition;
boost::atomic<int> threads(0);

// Some loop which can be parallelised
for ( const auto & x : list )
{
    // wait if thread_count exceeds 8 threads
    boost::mutex::scoped_lock lock(mutex);
    while ( threads >= 8 ) 
         condition.wait( lock );

    // Create worker object
    worker _wrk_( threads, mutex, condition );

    boost::thread * thread = new boost::thread( &worker::run, &_wrk_, /* other params */ );
    thread_group.add_thread( thread );
    threads++;
}

This works for most of my scenarios, but now I have a thread object which I need to re-use.

The reason is simple: This tread object contains thrust::device_vector<float> which are expensive to (re)allocate when the object is removed.

Furthermore, those vectors can be re-used as most of their contents won't be changed.

Therefore, I am looking for a mechanism which could re-use the objects created within the loop - in fact, I will allocate 8 of those objects (or as many as my concurrent threads) before hand, and then use them over and over again. What I am hoping can be done, is something such as this:

boost::thread_group thread_group;
boost::mutex mutex;
boost::condition_variable condition;
boost::atomic<int> threads(0);

// our worker objects to be reused
std::vector<std::shared_ptr<worker>>workers(8,std::make_shared<worker>(threads,mutex,condition));

// Some loop which can be parallelised
for ( const auto & x : list )
{
    // wait if thread_count exceeds 8 threads
    boost::mutex::scoped_lock lock(mutex);
    while ( threads >= 8 ) 
         condition.wait( lock );

    // get next available thread object from the vector
    auto _wrk_ = std::find_if(workers.begin(), workers.end(), is_available() );

    // if we have less than 8 threads but no available thread object
    if ( _wrk_ == workers.end() ) throw std::runtime_error ("...");

    // Use the first available worker object for this thread
    boost::thread * thread = new boost::thread(&worker::run, &(*_wrk_));
    thread_group.add_thread( thread );
    threads++;
}

I don't know how to signal the is_available(), other than implementing it as a class method (of the worker class).

Second, this appears to me to be too complex for no reason, I'm sure there has to be some kind of other pattern I can use which is more simple and/or elegant.

a very simple way to implement a thread pool is to use boost::asio.

Complete example here, including two types of task (function and object) plus exception handling:

#include <iostream>
#include <vector>
#include <thread>
#include <string>
#include <chrono>
#include <random>
#include <condition_variable>
#include <boost/asio.hpp>

void emit(const char* txt, int index)
{
    static std::mutex m;
    std::lock_guard<std::mutex> guard { m };
    std::cout << txt << ' ' << index << std::endl;
}

struct worker_pool
{
    boost::asio::io_service _io_service;
    boost::asio::io_service::work _work { _io_service };

    std::vector<std::thread> _threads;

    std::condition_variable _cv;
    std::mutex _cvm;
    size_t _tasks = 0;


    void start()
    {
        for (int i = 0 ; i < 8 ; ++i) {
            _threads.emplace_back(std::bind(&worker_pool::thread_proc, this));
        }
    }

    void wait()
    {
        std::unique_lock<std::mutex> lock(_cvm);
        _cv.wait(lock, [this] { return _tasks == 0; });
    }

    void stop()
    {
        wait();
        _io_service.stop();
        for (auto& t : _threads) {
            if (t.joinable())
                t.join();
        }
        _threads.clear();

    }

    void thread_proc()
    {
        while (!_io_service.stopped())
        {
            try {
                _io_service.run();
            }
            catch(const std::exception& e)
            {
                emit(e.what(), -1);
            }
        }
    }

    void reduce() {

        std::unique_lock<std::mutex> lock(_cvm);
        if (--_tasks == 0) {
            lock.unlock();
            _cv.notify_all();
        }
    }

    template<class F>
    void submit(F&& f)
    {
        std::unique_lock<std::mutex> lock(_cvm);
        ++ _tasks;
        lock.unlock();
        _io_service.post([this, f = std::forward<F>(f)]
                         {
                             try {
                                 f();
                             }
                             catch(...)
                             {
                                 reduce();
                                 throw;
                             }
                             reduce();
                         });


    }
};



void do_some_work(int index, std::chrono::milliseconds delay)
{
    emit("starting work item ", index);
    std::this_thread::sleep_for(delay);
    emit("ending work item ", index);
}

struct some_other_work
{
    some_other_work(int index, std::chrono::milliseconds delay)
    : _index(index)
    , _delay(delay)
    {}

    void operator()() const {

        emit("starting some other work ", _index);

        if (!(_index % 7)) {
            emit("uh oh! ", _index);
            using namespace std::string_literals;
            throw std::runtime_error("uh oh thrown in "s + std::to_string(_index));
        }

        emit("ending some other work ", _index);
    }

    int _index;
    std::chrono::milliseconds _delay;
};

auto main() -> int
{
    worker_pool pool;
    pool.start();

    std::random_device rd;
    std::default_random_engine eng(rd());

    std::uniform_int_distribution<int> dist(50, 200);
    for (int i = 0 ; i < 1000 ; ++i) {
        std::chrono::milliseconds delay(dist(eng));
        pool.submit(std::bind(do_some_work, i, delay));
        pool.submit(some_other_work(i, delay));
    }

    pool.wait();
    pool.stop();

    return 0;
}

example output:

starting work item  0
starting some other work  0
starting work item  1
starting some other work  1
starting work item  2
starting some other work  2
starting work item  3
starting some other work  3
uh oh!  0
ending some other work  1
ending some other work  2
ending some other work  3
starting work item  4
uh oh thrown in 0 -1
starting some other work  4
starting work item  5
ending some other work  4
starting some other work  5
starting work item  6
ending some other work  5
starting some other work  6
ending some other work  6
starting work item  7
ending work item  0
starting some other work  7
uh oh!  7
uh oh thrown in 7 -1
starting work item  8
ending work item  1
starting some other work  8
ending some other work  8
starting work item  9
ending work item  5
starting some other work  9
ending some other work  9
starting work item  10
ending work item  7
starting some other work  10
ending some other work  10
starting work item  11
ending work item  4
starting some other work  11
ending some other work  11
starting work item  12
ending work item  3
starting some other work  12
ending some other work  12
starting work item  13
ending work item  10
ending work item  6
starting some other work  13
starting work item  14
ending some other work  13
starting some other work  14
uh oh!  14
uh oh thrown in 14 -1
...