Информация об изменениях

Сообщение Re: thread pool и борьба с race condition от 19.02.2015 8:10

Изменено 20.02.2015 6:44 vdimas

Старый текст
Новый текст
Разница

Здравствуйте, RedApe, Вы писали:

Наверно, нужен подход "получил результат и забыл" (об исходном объекте-задаче). В этом случае, 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.

Re: thread pool и борьба с race condition

// пишу прямо в браузер

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

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());
    }
  };
}

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