condition_variable_impl.hpp

Go to the documentation of this file.

/*
 * Copyright (c) 2014-2015, Hewlett-Packard Development Company, LP.
 * This program is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License as published by the Free
 * Software Foundation; either version 2 of the License, or (at your option)
 * any later version.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
 * more details. You should have received a copy of the GNU General Public
 * License along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
 *
 * HP designates this particular file as subject to the "Classpath" exception
 * as provided by HP in the LICENSE.txt file that accompanied this code.
 */
#ifndef FOEDUS_THREAD_CONDITION_VARIABLE_IMPL_HPP_
#define FOEDUS_THREAD_CONDITION_VARIABLE_IMPL_HPP_
#include <stdint.h>

#include <atomic>
#include <chrono>
#include <condition_variable>
#include <mutex>

#include "foedus/assert_nd.hpp"
#include "foedus/assorted/assorted_func.hpp"

namespace foedus {
namespace thread {
class ConditionVariable final {
 public:
  ConditionVariable() : waiters_(0), notifiers_(0) {}
  ~ConditionVariable() {
    ASSERT_ND(waiters_ == 0);
    // we must wait until all notifiers exit notify_all.
    // this assumes no new notifiers are newly coming in this situation.
    while (notifiers_ > 0) {
      assorted::spinlock_yield();
    }
  }

  // not copyable, assignable.
  ConditionVariable(const ConditionVariable &other) = delete;
  ConditionVariable& operator=(const ConditionVariable &other) = delete;
  ConditionVariable(ConditionVariable &&other) = delete;
  ConditionVariable& operator=(ConditionVariable &&other) = delete;

  template<typename PREDICATE>
  void wait(PREDICATE predicate) {
    WaiterScope scope(this);
    condition_.wait(scope.lock_, predicate);
  }
  void wait() {
    WaiterScope scope(this);
    condition_.wait(scope.lock_);
  }

  template<class REP, class PERIOD, typename PREDICATE>
  bool wait_for(const std::chrono::duration<REP, PERIOD>& timeout, PREDICATE predicate) {
    WaiterScope scope(this);
    return condition_.wait_for(scope.lock_, timeout, predicate);
  }

  template<class REP, class PERIOD>
  bool wait_for(const std::chrono::duration<REP, PERIOD>& timeout) {
    WaiterScope scope(this);
    return condition_.wait_for(scope.lock_, timeout) == std::cv_status::no_timeout;
  }

  template< class CLOCK, class DURATION, typename PREDICATE>
  bool wait_until(const std::chrono::time_point<CLOCK, DURATION>& until, PREDICATE predicate) {
    WaiterScope scope(this);
    return condition_.wait_until(scope.lock_, until, predicate);
  }

  template< class CLOCK, class DURATION>
  bool wait_until(const std::chrono::time_point<CLOCK, DURATION>& until) {
    WaiterScope scope(this);
    return condition_.wait_until(scope.lock_, until) == std::cv_status::no_timeout;
  }

  template<typename SIGNAL_ACTION>
  void notify_all(SIGNAL_ACTION signal_action) {
    NotifierScope scope(this);
    {
      std::lock_guard<std::mutex> guard(mutex_);
      signal_action();  // conduct the action to update actual values *in* critical section
    }
    while (waiters_ > 0) {
      condition_.notify_one();
      assorted::spinlock_yield();
    }
  }
  void notify_all() {
    notify_all([]{});
  }

  template<typename SIGNAL_ACTION>
  void notify_one(SIGNAL_ACTION signal_action) {
    NotifierScope scope(this);
    {
      std::lock_guard<std::mutex> guard(mutex_);
      signal_action();
    }
    if (waiters_ > 0) {
      condition_.notify_one();
    }
  }

  void notify_one() {
    notify_one([]{});
  }

 private:
  std::condition_variable         condition_;
  std::mutex                      mutex_;

  std::atomic<uint32_t>           waiters_;
  std::atomic<uint32_t>           notifiers_;

  struct WaiterScope {
    explicit WaiterScope(ConditionVariable* enclosure)
      : enclosure_(enclosure), lock_(enclosure_->mutex_) {
      ASSERT_ND(lock_.owns_lock());
      ++enclosure_->waiters_;
    }
    ~WaiterScope() {
      ASSERT_ND(lock_.owns_lock());
      ASSERT_ND(enclosure_->waiters_ > 0);
      --enclosure_->waiters_;
    }
    ConditionVariable* const enclosure_;
    std::unique_lock<std::mutex> lock_;
  };
  struct NotifierScope {
    explicit NotifierScope(ConditionVariable* enclosure) : enclosure_(enclosure) {
      ++enclosure_->notifiers_;
    }
    ~NotifierScope() {
      ASSERT_ND(enclosure_->notifiers_ > 0);
      --enclosure_->notifiers_;
    }
    ConditionVariable* const enclosure_;
  };
};


}  // namespace thread
}  // namespace foedus
#endif  // FOEDUS_THREAD_CONDITION_VARIABLE_IMPL_HPP_