c++ - 线程完成后 boost 线程工作对象重用

Question

我有一个简单的线程对象负责一些执行(一个 worker ):

在最简单的形式中，为每个线程创建一个对象:

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

我使用它的方式是在一个最多运行 8 个并发线程的循环中，并等待一个线程完成后得到通知，以便产生下一个线程:

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

这适用于我的大部分情况，但现在我有一个需要重新使用的线程对象。

原因很简单:这个胎面对象包含thrust::device_vector<float>当对象被移除时(重新)分配是昂贵的。

此外，这些 vector 可以重复使用，因为它们的大部分内容不会改变。

因此，我正在寻找一种可以重用在循环中创建的对象的机制 - 事实上，我会事先分配其中 8 个对象(或与我的并发线程一样多)，然后再使用它们又一遍。我希望可以做的是这样的事情:

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

我不知道如何向 is_available() 发出信号，除了将其实现为( worker 类的)类方法。

其次，在我看来这无缘无故地太复杂了，我确信必须有某种其他模式我可以使用，它更简单和/或优雅。

Answer 1

实现线程池的一个非常简单的方法是使用boost::asio。

完整的例子在这里，包括两种类型的任务(函数和对象)加上异常处理:

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

示例输出:

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
...