mutex_set.hpp

Go to the documentation of this file.

#pragma once
 
/// @file userver/concurrent/mutex_set.hpp
/// @brief @copybrief concurrent::MutexSet
 
#include <chrono>
#include <functional>
#include <string>
#include <unordered_set>
#include <utility>
 
#include <userver/engine/condition_variable.hpp>
#include <userver/engine/mutex.hpp>
#include <userver/engine/task/cancel.hpp>
#include <userver/utils/cached_hash.hpp>
#include <userver/utils/fixed_array.hpp>
 
USERVER_NAMESPACE_BEGIN
 
namespace concurrent {
 
namespace impl {
 
template <typename T, typename Equal>
struct MutexDatum final {
  explicit MutexDatum(size_t way_size, const Equal& equal = Equal{})
      : set(way_size, {}, equal) {}
 
  ~MutexDatum() {
    UASSERT_MSG(set.empty(),
                "MutexDatum is destroyed while someone is holding the lock");
  }
 
  engine::Mutex mutex;
  engine::ConditionVariable cv;
  std::unordered_set<T, std::hash<T>, Equal> set;
};
 
}  // namespace impl
 
/// Mutex-like object associated with the key of a MutexSet. It provides the
/// same interface as engine::Mutex, you may use it with std::unique_lock,
/// std::lock_guard, etc.
/// @note different ItemMutex'es of the same MutexSet obtained with the same
///       argument to GetMutexForKey() share the same critical section. IOW, if
///       the first mutex is locked, the second one will block until the first
///       mutex is unlocked.
/// @note can be used only from coroutines.
template <typename Key, typename Equal>
class ItemMutex final {
 public:
  using HashAndKey = utils::CachedHash<Key>;
 
  using MutexDatum =
      impl::MutexDatum<HashAndKey, utils::CachedHashKeyEqual<Equal>>;
 
  ItemMutex(MutexDatum& md, HashAndKey&& key);
 
  void lock();
 
  void unlock();
 
  bool try_lock();
 
  template <typename Rep, typename Period>
  bool try_lock_for(std::chrono::duration<Rep, Period>);
 
  template <typename Clock, typename Duration>
  bool try_lock_until(std::chrono::time_point<Clock, Duration>);
 
 private:
  bool TryFinishLocking();
 
  MutexDatum& md_;
  const HashAndKey key_;
};
 
/// @ingroup userver_concurrency userver_containers
///
/// @brief A dynamic set of mutexes
///
/// It can be used for separate critical sections for
/// multiple keys when the key set is not known at compile time and may change
/// in runtime.
///
/// Example:
/// @snippet src/concurrent/mutex_set_test.cpp  Sample mutex set usage
template <typename Key = std::string, typename Hash = std::hash<Key>,
          typename Equal = std::equal_to<Key>>
class MutexSet final : Hash {
 public:
  explicit MutexSet(size_t ways = 1, size_t way_size = 1,
                    const Hash& hash = Hash{}, const Equal& equal = Equal{});
 
  /// Get the mutex-like object for a key. Coroutine-safe.
  /// @note the returned object holds a reference to MutexSet, so make sure
  ///       that MutexSet is alive while you're working with ItemMutex.
  ItemMutex<Key, Equal> GetMutexForKey(Key key);
 
 private:
  using MutexDatum = typename ItemMutex<Key, Equal>::MutexDatum;
  utils::FixedArray<MutexDatum> mutex_data_;
};
 
template <typename Key, typename Hash, typename Equal>
MutexSet<Key, Hash, Equal>::MutexSet(size_t ways, size_t way_size,
                                     const Hash& hash, const Equal& equal)
    : Hash(hash),
      mutex_data_(ways, way_size, utils::CachedHashKeyEqual<Equal>{equal}) {}
 
template <typename Key, typename Hash, typename Equal>
ItemMutex<Key, Equal> MutexSet<Key, Hash, Equal>::GetMutexForKey(Key key) {
  const auto hash_value = Hash::operator()(key);
  const auto size = mutex_data_.size();
 
  // Compilers optimize a % b and a / b to a single div operation
  const auto way = hash_value % size;
  const auto new_hash = hash_value / size;
 
  return ItemMutex<Key, Equal>(mutex_data_[way], {new_hash, std::move(key)});
}
 
template <typename Key, typename Equal>
ItemMutex<Key, Equal>::ItemMutex(MutexDatum& md, HashAndKey&& key)
    : md_(md), key_(std::move(key)) {}
 
template <typename Key, typename Equal>
void ItemMutex<Key, Equal>::lock() {
  engine::TaskCancellationBlocker blocker;
  std::unique_lock<engine::Mutex> lock(md_.mutex);
 
  [[maybe_unused]] auto is_locked =
      md_.cv.Wait(lock, [this] { return TryFinishLocking(); });
  UASSERT(is_locked);
}
 
template <typename Key, typename Equal>
void ItemMutex<Key, Equal>::unlock() {
  std::unique_lock lock(md_.mutex);
 
  // Forcing the destructor of the node to run outside of the critical section
  [[maybe_unused]] auto node = md_.set.extract(key_);
 
  // We must wakeup all the waiters, because otherwise we can wake up the wrong
  // one and loose the wakeup:
  //
  // Task #1 waits for mutex A, task #2 waits for mutex B
  // Task #3
  //  * unlocks mutex A
  //  * does NotifyOne
  //  * wakeups thread #2
  //  * mutex B is still locked, task #2 goes to sleep
  //  * we lost the wakeup :(
  md_.cv.NotifyAll();
 
  lock.unlock();
 
  // Destructor of node is run here
}
 
template <typename Key, typename Equal>
bool ItemMutex<Key, Equal>::try_lock() {
  std::unique_lock<engine::Mutex> lock(md_.mutex);
  return TryFinishLocking();
}
 
template <typename Key, typename Equal>
template <typename Rep, typename Period>
bool ItemMutex<Key, Equal>::try_lock_for(
    std::chrono::duration<Rep, Period> duration) {
  std::unique_lock<engine::Mutex> lock(md_.mutex);
  return md_.cv.WaitFor(lock, duration, [this] { return TryFinishLocking(); });
}
 
template <typename Key, typename Equal>
template <typename Clock, typename Duration>
bool ItemMutex<Key, Equal>::try_lock_until(
    std::chrono::time_point<Clock, Duration> time_point) {
  std::unique_lock<engine::Mutex> lock(md_.mutex);
  return md_.cv.WaitUntil(lock, time_point,
                          [this] { return TryFinishLocking(); });
}
 
template <typename Key, typename Equal>
bool ItemMutex<Key, Equal>::TryFinishLocking() {
  return md_.set.insert(key_).second;
}
 
}  // namespace concurrent
 
USERVER_NAMESPACE_END