tpl_concurrent_hash_map.H

Go to the documentation of this file.

 
/*
                          Aleph_w
 
  Data structures & Algorithms
  https://github.com/lrleon/Aleph-w
 
  This file is part of Aleph-w library
 
  Copyright (c) 2002-2026 Leandro Rabindranath Leon
 
  Permission is hereby granted, free of charge, to any person obtaining a copy
  of this software and associated documentation files (the "Software"), to deal
  in the Software without restriction, including without limitation the rights
  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  copies of the Software, and to permit persons to whom the Software is
  furnished to do so, subject to the following conditions:
 
  The above copyright notice and this permission notice shall be included in all
  copies or substantial portions of the Software.
 
  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  SOFTWARE.
*/
 
#ifndef TPL_CONCURRENT_HASH_MAP_H
#define TPL_CONCURRENT_HASH_MAP_H
 
#include <array>
#include <memory>
#include <mutex>
#include <optional>
#include <shared_mutex>
#include <type_traits>
#include <utility>
 
#include <ah-errors.H>
#include <tpl_dynSetHash.H>
 
namespace Aleph
{
 
template <typename Key, typename T,
          class Cmp = Aleph::equal_to<Key>,
          size_t Shards = 64>
class ConcurrentHashMap
{
  static_assert(Shards > 0, "ConcurrentHashMap requires at least one shard");
 
  using Map = DynMapHashTable<Key, T, LhashTable, Cmp>;
 
public:
  using Hash_Fct_Ptr = typename Map::Hash_Fct_Ptr;
 
private:
  // Internal structure representing a single partition of the hash map.
  // Each Shard is protected by its own shared_mutex. alignas(64) pads each
  // Shard to its own cache line: without it, two Shard objects placed
  // next to each other by the allocator could share a cache line, and
  // then threads locking/unlocking *different* shards' mutexes under
  // write contention would still ping-pong that line between cores
  // (false sharing) -- defeating part of the point of sharding. 64 bytes
  // matches the cache-line size assumed elsewhere in Aleph (see
  // concurrency_utils.H's SpscQueue::head_/tail_).
  struct alignas(64) Shard
  {
    mutable std::shared_mutex mutex;
    Map map;
 
    Shard(Hash_Fct_Ptr hash_fct, const Cmp &cmp)
      : map(Primes::DefaultPrime, hash_fct, cmp)
    {}
  };
 
  std::array<std::unique_ptr<Shard>, Shards> shards_;
  Hash_Fct_Ptr hash_fct_;
 
  [[nodiscard]] Shard &shard_for(const Key &key) noexcept
  {
    return *shards_[hash_fct_(key) % Shards];
  }
 
  [[nodiscard]] const Shard &shard_for(const Key &key) const noexcept
  {
    return *shards_[hash_fct_(key) % Shards];
  }
 
public:
  explicit ConcurrentHashMap(Hash_Fct_Ptr hash_fct = dft_hash_ptr_fct<Key>,
                              const Cmp &cmp = Cmp())
    : hash_fct_(hash_fct)
  {
    for (auto &shard : shards_)
      shard = std::make_unique<Shard>(hash_fct, cmp);
  }
 
  ConcurrentHashMap(const ConcurrentHashMap &) = delete;
  ConcurrentHashMap &operator=(const ConcurrentHashMap &) = delete;
  ConcurrentHashMap(ConcurrentHashMap &&) = delete;
  ConcurrentHashMap &operator=(ConcurrentHashMap &&) = delete;
 
  bool insert(const Key &key, const T &value)
  {
    Shard &shard = shard_for(key);
    std::unique_lock lock(shard.mutex);
    return shard.map.insert(key, value) != nullptr;
  }
 
  bool insert(const Key &key, T &&value)
  {
    Shard &shard = shard_for(key);
    std::unique_lock lock(shard.mutex);
    return shard.map.insert(key, std::move(value)) != nullptr;
  }
 
  void insert_or_assign(const Key &key, T value)
  {
    Shard &shard = shard_for(key);
    std::unique_lock lock(shard.mutex);
 
    // We do a double lookup here (search + insert) because DynMapHashTable
    // requires checking for existence first to avoid duplicating or throwing.
    auto *pair = shard.map.search(key);
    if (pair != nullptr)
      pair->second = std::move(value);
    else
      shard.map.insert(key, std::move(value));
  }
 
  bool erase(const Key &key)
  {
    Shard &shard = shard_for(key);
    std::unique_lock lock(shard.mutex);
 
    // remove_by_data() removes the already-located entry directly (via
    // pointer arithmetic back to its owning bucket), unlike
    // DynMapHashTable::remove(key), which re-searches internally and
    // move-constructs the removed value only to have it discarded here --
    // paying an extra lookup and a throwing-move risk for nothing.
    auto *pair = shard.map.search(key);
    if (pair == nullptr)
      return false;
 
    shard.map.remove_by_data(pair->second);
    return true;
  }
 
  [[nodiscard]] bool contains(const Key &key) const
  {
    const Shard &shard = shard_for(key);
    std::shared_lock lock(shard.mutex);
    return shard.map.contains(key);
  }
 
  [[nodiscard]] std::optional<T> find_copy(const Key &key) const
    requires std::is_copy_constructible_v<T>
  {
    const Shard &shard = shard_for(key);
    std::shared_lock lock(shard.mutex);
    const auto *pair = shard.map.search(key);
    if (pair == nullptr)
      return std::nullopt;
    return pair->second;
  }
 
  template <typename F>
  bool with_value(const Key &key, F &&f) const
  {
    using Result = std::invoke_result_t<F, const T &>;
    static_assert(not std::is_reference_v<Result>,
                  "ConcurrentHashMap::with_value callback must not return a reference");
    const Shard &shard = shard_for(key);
    std::shared_lock lock(shard.mutex);
    const auto *pair = shard.map.search(key);
    if (pair == nullptr)
      return false;
    std::invoke(std::forward<F>(f), std::as_const(pair->second));
    return true;
  }
 
  template <typename F>
  bool with_value_mut(const Key &key, F &&f)
  {
    using Result = std::invoke_result_t<F, T &>;
    static_assert(not std::is_reference_v<Result>,
                  "ConcurrentHashMap::with_value_mut callback must not return a reference");
    Shard &shard = shard_for(key);
    std::unique_lock lock(shard.mutex);
    auto *pair = shard.map.search(key);
    if (pair == nullptr)
      return false;
    std::invoke(std::forward<F>(f), pair->second);
    return true;
  }
 
  void clear()
  {
    for (auto &shard : shards_)
      {
        std::unique_lock lock(shard->mutex);
        shard->map.clear();
      }
  }
 
  [[nodiscard]] size_t size() const
  {
    size_t total = 0;
    for (const auto &shard : shards_)
      {
        std::shared_lock lock(shard->mutex);
        total += shard->map.size();
      }
    return total;
  }
 
  [[nodiscard]] bool is_empty() const
  {
    for (const auto &shard : shards_)
      {
        std::shared_lock lock(shard->mutex);
        if (not shard->map.is_empty())
          return false;
      }
    return true;
  }
 
  [[nodiscard]] Array<std::pair<Key, T>> snapshot() const
    requires std::is_copy_constructible_v<Key> and std::is_copy_constructible_v<T>
  {
    Array<std::pair<Key, T>> result;
    size_t reserved = 0;
    for (const auto &shard : shards_)
      {
        std::shared_lock lock(shard->mutex);
        // Grow the reservation by this (already-locked) shard's own O(1)
        // size *before* copying its entries in, so appends below never
        // trigger a reallocation -- without a separate extra pass over
        // every shard (and its lock) just to compute a grand total first.
        reserved += shard->map.size();
        result.reserve(reserved);
        for (const auto &entry : shard->map)
          result.append(entry);
      }
    return result;
  }
};
 
}  // namespace Aleph
 
#endif  // TPL_CONCURRENT_HASH_MAP_H