Сообщение Re: thread pool и борьба с race condition от 19.02.2015 8:10
Изменено 20.02.2015 6:43 vdimas
Здравствуйте, RedApe, Вы писали:
Наверно, нужен подход "получил результат и забыл" (об исходном объекте-задаче). В этом случае, thread_pool::detach_task становится не нужен, пусть задача живет ровно столько, сколько она нужна потоку из пула, т.е. ровно столько, сколько производятся вычисления. В этом случае "задача" может быть произвольным функтором/лямбдой и тогда достаточно будет сосредоточиться только на сигналинге готовности и на разделяемом владении данными, оформив это дело в стиле С++:
Для полноценного соответствия С++ надо дописать еще метод last_value_promise::set_exception и допилить методы get, поможет в этом следующая ссылка:
http://en.cppreference.com/w/cpp/error/exception_ptr
Далее пишем еще один простой хелпер:
Затем берем произвольную реализацию пула потоков (из буста или например: http://habrahabr.ru/post/188234/)
Используем все вместе:
Примечание.
Обрати внимание на get и get_then_reset. Первый вариант не сбрасывает значение, поэтому последующий вызов get/get_then_reset не будет ожидать обновления значения, а просто вернет текущее (оно же последнее). Второй вариант сбрасывает всю систему в начальное состояние.
Дедлоков нет by design.
Наверно, нужен подход "получил результат и забыл" (об исходном объекте-задаче). В этом случае, thread_pool::detach_task становится не нужен, пусть задача живет ровно столько, сколько она нужна потоку из пула, т.е. ровно столько, сколько производятся вычисления. В этом случае "задача" может быть произвольным функтором/лямбдой и тогда достаточно будет сосредоточиться только на сигналинге готовности и на разделяемом владении данными, оформив это дело в стиле С++:
// пишу прямо в браузер
namespace details {
typedef some_mutex mutex;
typedef some_lock guard;
typedef some_condition condition;
template<typename T>
struct last_value : boost::noncopyable
{
optional<T> value_;
mutex mutex_;
condition condition_;
T get()
{
guard g(mutex);
while(!value_)
condition_.wait(g);
return value_.value();
}
T get_then_reset()
{
guard g(mutex);
while(!value_)
condition_.wait(g);
T tmp = std::move(value_.value());
value_ = std::nullopt;
return tmp;
}
void set_value(const T & value)
{
guard g(mutex_);
value_ = value;
condition_.notify_all();
}
void wait()
{
guard g(mutex_);
while(!value_)
condition_.wait(l);
}
future_status wait_until(const time_point & abs_time)
{
guard g(mutex_);
while(!value_) {
condition_.wait_until(g, abs_time);
if(now() > abs_time) break;
}
return
shared_state_->value_ ? future_status::ready
: future_status::timeout;
}
};
} // namespace details
template<typename T>
class last_value_promise
{
typedef shared_ptr<details::last_value<T>> shared_state_ptr;
public:
last_value_promise(shared_state_ptr shared_state)
: shared_state_(shared_state)
{}
void set_value(const T & value) {
shared_state_->set_value(value);
}
private:
shared_state_ptr shared_state_;
};
template<typename T>
class last_value_future
{
typedef shared_ptr<details::last_value<T>> shared_state_ptr;
public:
last_value_future()
: shared_state_(make_shared<details::last_value<T>>())
{}
last_value_promise<T> get_promise() {
return last_value_promise<T>(shared_state_);
}
T get() {
return shared_state_->get();
}
T get_then_reset() {
return shared_state_->get_then_reset();
}
void wait() {
return shared_state_->wait();
}
future_status wait_until(const time_point & abs_time) {
return shared_state_->wait_until(abs_time);
}
future_status wait_for(const duration & timeout) {
return shared_state_->wait_until(now() + timeout);
}
private:
shared_state_ptr shared_state_;
};Для полноценного соответствия С++ надо дописать еще метод last_value_promise::set_exception и допилить методы get, поможет в этом следующая ссылка:
http://en.cppreference.com/w/cpp/error/exception_ptr
Далее пишем еще один простой хелпер:
template<typename T, typename F>
function<void()> last_value_aggregator(last_value_promise<T> p, F/*&&*/ f)
{
return [] {
try {
p.set_value(f());
} catch(...) {
p.set_exception(current_exception());
}
};
}Затем берем произвольную реализацию пула потоков (из буста или например: http://habrahabr.ru/post/188234/)
Используем все вместе:
size_t task(size_t param)
{
this_thread::sleep_for(random_delay());
return param;
}
void printer(last_value_future<size_t> fut)
{
while(true) {
size_t last_value = fut.get_then_reset();
cout << " last value: " << last_value << endl;
}
}
void main()
{
thread_pool pool;
last_value_future<size_t> fut;
async(pool, printer, fut);
last_value_promise<size_t> p = fut.get_promise();
for(size_t i = 0; i < 42; i++)
async(pool, last_value_aggregator(p, task), i);
// press CTRL+C to exit...
}Примечание.
Обрати внимание на get и get_then_reset. Первый вариант не сбрасывает значение, поэтому последующий вызов get/get_then_reset не будет ожидать обновления значения, а просто вернет текущее (оно же последнее). Второй вариант сбрасывает всю систему в начальное состояние.
Дедлоков нет by design.
Здравствуйте, RedApe, Вы писали:
Наверно, нужен подход "получил результат и забыл" (об исходном объекте-задаче). В этом случае, thread_pool::detach_task становится не нужен, пусть задача живет ровно столько, сколько она нужна потоку из пула, т.е. ровно столько, сколько производятся вычисления. В этом случае "задача" может быть произвольным функтором/лямбдой и тогда достаточно будет сосредоточиться только на сигналинге готовности и на разделяемом владении данными, оформив это дело в стиле С++:
Для полноценного соответствия С++ надо дописать еще метод last_value_promise::set_exception и допилить методы get, поможет в этом следующая ссылка:
http://en.cppreference.com/w/cpp/error/exception_ptr
Далее пишем еще один простой хелпер:
Затем берем произвольную реализацию пула потоков (из буста или например: http://habrahabr.ru/post/188234/)
Используем все вместе:
Примечание.
Обрати внимание на get и get_then_reset. Первый вариант не сбрасывает значение, поэтому последующий вызов get/get_then_reset не будет ожидать обновления значения, а просто вернет текущее (оно же последнее). Второй вариант сбрасывает всю систему в начальное состояние.
Дедлоков нет by design.
Наверно, нужен подход "получил результат и забыл" (об исходном объекте-задаче). В этом случае, thread_pool::detach_task становится не нужен, пусть задача живет ровно столько, сколько она нужна потоку из пула, т.е. ровно столько, сколько производятся вычисления. В этом случае "задача" может быть произвольным функтором/лямбдой и тогда достаточно будет сосредоточиться только на сигналинге готовности и на разделяемом владении данными, оформив это дело в стиле С++:
// пишу прямо в браузер
namespace details {
typedef some_mutex mutex;
typedef some_lock guard;
typedef some_condition condition;
template<typename T>
struct last_value : boost::noncopyable
{
optional<T> value_;
mutex mutex_;
condition condition_;
T get()
{
guard g(mutex);
while(!value_)
condition_.wait(g);
return value_.value();
}
T get_then_reset()
{
guard g(mutex);
while(!value_)
condition_.wait(g);
T tmp = std::move(value_.value());
value_ = std::nullopt;
return tmp;
}
void set_value(const T & value)
{
guard g(mutex_);
value_ = value;
condition_.notify_all();
}
void wait()
{
guard g(mutex_);
while(!value_)
condition_.wait(g);
}
future_status wait_until(const time_point & abs_time)
{
guard g(mutex_);
while(!value_) {
condition_.wait_until(g, abs_time);
if(now() > abs_time) break;
}
return
shared_state_->value_ ? future_status::ready
: future_status::timeout;
}
};
} // namespace details
template<typename T>
class last_value_promise
{
typedef shared_ptr<details::last_value<T>> shared_state_ptr;
public:
last_value_promise(shared_state_ptr shared_state)
: shared_state_(shared_state)
{}
void set_value(const T & value) {
shared_state_->set_value(value);
}
private:
shared_state_ptr shared_state_;
};
template<typename T>
class last_value_future
{
typedef shared_ptr<details::last_value<T>> shared_state_ptr;
public:
last_value_future()
: shared_state_(make_shared<details::last_value<T>>())
{}
last_value_promise<T> get_promise() {
return last_value_promise<T>(shared_state_);
}
T get() {
return shared_state_->get();
}
T get_then_reset() {
return shared_state_->get_then_reset();
}
void wait() {
return shared_state_->wait();
}
future_status wait_until(const time_point & abs_time) {
return shared_state_->wait_until(abs_time);
}
future_status wait_for(const duration & timeout) {
return shared_state_->wait_until(now() + timeout);
}
private:
shared_state_ptr shared_state_;
};Для полноценного соответствия С++ надо дописать еще метод last_value_promise::set_exception и допилить методы get, поможет в этом следующая ссылка:
http://en.cppreference.com/w/cpp/error/exception_ptr
Далее пишем еще один простой хелпер:
template<typename T, typename F>
function<void()> last_value_aggregator(last_value_promise<T> p, F/*&&*/ f)
{
return [] {
try {
p.set_value(f());
} catch(...) {
p.set_exception(current_exception());
}
};
}Затем берем произвольную реализацию пула потоков (из буста или например: http://habrahabr.ru/post/188234/)
Используем все вместе:
size_t task(size_t param)
{
this_thread::sleep_for(random_delay());
return param;
}
void printer(last_value_future<size_t> fut)
{
while(true) {
size_t last_value = fut.get_then_reset();
cout << " last value: " << last_value << endl;
}
}
void main()
{
thread_pool pool;
last_value_future<size_t> fut;
async(pool, printer, fut);
last_value_promise<size_t> p = fut.get_promise();
for(size_t i = 0; i < 42; i++)
async(pool, last_value_aggregator(p, task), i);
// press CTRL+C to exit...
}Примечание.
Обрати внимание на get и get_then_reset. Первый вариант не сбрасывает значение, поэтому последующий вызов get/get_then_reset не будет ожидать обновления значения, а просто вернет текущее (оно же последнее). Второй вариант сбрасывает всю систему в начальное состояние.
Дедлоков нет by design.