提振螺纹工人对象再利用线程完成后， [英] boost thread worker object re-use after thread has finished
问题描述
我有一个简单的线程对象，这需要一些执行（工人）的护理：
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__;
};
我用它的方式是它运行最多8个并发线程，并等待如果已完成得到通知，以产生一个循环中的下一个：
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.
原因很简单：这胎面对象包含推力:: device_vector＆LT;浮动＆GT; 这是昂贵的，以当对象被删除（重新）分配。
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.
所以，我找了一个机制，可以重新使用循环中创建的对象 - 事实上，我的手之前分配这些对象的8（或多达我的并发线程），然后用它们在一遍又一遍。
我很希望可以做的，是下面这样的：
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++;
}
我不知道如何以表明is_available（），除了其实现为一个类方法（工人阶级的）。
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.
推荐答案
执行一个线程池一个非常简单的方法是使用的boost :: ASIO 。
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;
}
例如输出：
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
...
这篇关于提振螺纹工人对象再利用线程完成后，的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
