tdlight/tdutils/td/utils/ConcurrentHashTable.h

323 lines
8.7 KiB
C
Raw Normal View History

//
// Copyright Aliaksei Levin (levlam@telegram.org), Arseny Smirnov (arseny30@gmail.com) 2014-2019
//
// Distributed under the Boost Software License, Version 1.0. (See accompanying
// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
//
#pragma once
#include "td/utils/common.h"
#include "td/utils/HazardPointers.h"
#include "td/utils/logging.h"
#include "td/utils/port/thread_local.h"
#include <atomic>
#include <condition_variable>
#include <mutex>
namespace td {
// AtomicHashArray<KeyT, ValueT>
// Building block for other concurrent hash maps
//
// Support one operation:
// template <class F>
// bool with_value(KeyT key, bool should_create, F &&func);
//
// Finds slot for key, and call func(value)
// Creates slot if should_create is true.
// Returns true if func was called.
//
// Concurrent calls with the same key may result in concurrent calls to func(value)
// It is responsibility of the caller to handle such races.
//
// Key should already be random
// It is responsibility of the caller to provide unique random key.
// One may use injective hash function, or handle collisions in some other way.
template <class KeyT, class ValueT>
class AtomicHashArray {
public:
explicit AtomicHashArray(size_t n) : nodes_(n) {
}
struct Node {
std::atomic<KeyT> key{KeyT{}};
ValueT value{};
};
size_t size() const {
return nodes_.size();
}
Node &node_at(size_t i) {
return nodes_[i];
}
static KeyT empty_key() {
return KeyT{};
}
template <class F>
bool with_value(KeyT key, bool should_create, F &&f) {
DCHECK(key != empty_key());
size_t pos = static_cast<size_t>(key) % nodes_.size();
size_t n = td::min(td::max(static_cast<size_t>(300), nodes_.size() / 16 + 2), nodes_.size());
for (size_t i = 0; i < n; i++) {
pos++;
if (pos >= nodes_.size()) {
pos = 0;
}
auto &node = nodes_[pos];
while (true) {
auto node_key = node.key.load(std::memory_order_acquire);
if (node_key == empty_key()) {
if (!should_create) {
return false;
}
KeyT expected_key = empty_key();
if (node.key.compare_exchange_strong(expected_key, key, std::memory_order_relaxed,
std::memory_order_relaxed)) {
f(node.value);
return true;
}
} else if (node_key == key) {
f(node.value);
return true;
} else {
break;
}
}
}
return false;
}
private:
std::vector<Node> nodes_;
};
// Simple concurrent hash map with multiple limitations
template <class KeyT, class ValueT>
class ConcurrentHashMap {
using HashMap = AtomicHashArray<KeyT, std::atomic<ValueT>>;
static HazardPointers<HashMap> hp_;
public:
explicit ConcurrentHashMap(size_t n = 32) {
n = 1;
hash_map_.store(make_unique<HashMap>(n).release());
}
ConcurrentHashMap(const ConcurrentHashMap &) = delete;
ConcurrentHashMap &operator=(const ConcurrentHashMap &) = delete;
ConcurrentHashMap(ConcurrentHashMap &&) = delete;
ConcurrentHashMap &operator=(ConcurrentHashMap &&) = delete;
~ConcurrentHashMap() {
unique_ptr<HashMap>(hash_map_.load());
}
static std::string get_name() {
return "ConcurrrentHashMap";
}
static KeyT empty_key() {
return KeyT{};
}
static ValueT empty_value() {
return ValueT{};
}
static ValueT migrate_value() {
return (ValueT)(1); // c-style conversion because reinterpret_cast<int>(1) is CE in MSVC
}
ValueT insert(KeyT key, ValueT value) {
CHECK(key != empty_key());
CHECK(value != migrate_value());
typename HazardPointers<HashMap>::Holder holder(hp_, get_thread_id(), 0);
while (true) {
auto hash_map = holder.protect(hash_map_);
if (!hash_map) {
do_migrate(nullptr);
continue;
}
bool ok = false;
ValueT inserted_value;
hash_map->with_value(key, true, [&](auto &node_value) {
ValueT expected_value = this->empty_value();
if (node_value.compare_exchange_strong(expected_value, value, std::memory_order_release,
std::memory_order_acquire)) {
ok = true;
inserted_value = value;
} else {
if (expected_value == this->migrate_value()) {
ok = false;
} else {
ok = true;
inserted_value = expected_value;
}
}
});
if (ok) {
return inserted_value;
}
do_migrate(hash_map);
}
}
ValueT find(KeyT key, ValueT value) {
typename HazardPointers<HashMap>::Holder holder(hp_, get_thread_id(), 0);
while (true) {
auto hash_map = holder.protect(hash_map_);
if (!hash_map) {
do_migrate(nullptr);
continue;
}
bool has_value = hash_map->with_value(
key, false, [&](auto &node_value) { value = node_value.load(std::memory_order_acquire); });
if (!has_value || value != migrate_value()) {
return value;
}
do_migrate(hash_map);
}
}
template <class F>
void for_each(F &&f) {
auto hash_map = hash_map_.load();
CHECK(hash_map);
auto size = hash_map->size();
for (size_t i = 0; i < size; i++) {
auto &node = hash_map->node_at(i);
auto key = node.key.load(std::memory_order_relaxed);
auto value = node.value.load(std::memory_order_relaxed);
if (key != empty_key()) {
CHECK(value != migrate_value());
if (value != empty_value()) {
f(key, value);
}
}
}
}
private:
// use no padding intentionally
std::atomic<HashMap *> hash_map_{nullptr};
std::mutex migrate_mutex_;
std::condition_variable migrate_cv_;
int migrate_cnt_{0};
int migrate_generation_{0};
HashMap *migrate_from_hash_map_{nullptr};
HashMap *migrate_to_hash_map_{nullptr};
struct Task {
size_t begin;
size_t end;
bool empty() const {
return begin >= end;
}
size_t size() const {
if (empty()) {
return 0;
}
return end - begin;
}
};
struct TaskCreator {
size_t chunk_size;
size_t size;
std::atomic<size_t> pos{0};
Task create() {
auto i = pos++;
auto begin = i * chunk_size;
auto end = begin + chunk_size;
if (end > size) {
end = size;
}
return {begin, end};
}
};
TaskCreator task_creator;
void do_migrate(HashMap *ptr) {
//LOG(ERROR) << "In do_migrate: " << ptr;
std::unique_lock<std::mutex> lock(migrate_mutex_);
if (hash_map_.load() != ptr) {
return;
}
init_migrate();
CHECK(!ptr || migrate_from_hash_map_ == ptr);
migrate_cnt_++;
auto migrate_generation = migrate_generation_;
lock.unlock();
run_migrate();
lock.lock();
migrate_cnt_--;
if (migrate_cnt_ == 0) {
finish_migrate();
}
migrate_cv_.wait(lock, [&] { return migrate_generation_ != migrate_generation; });
}
void finish_migrate() {
//LOG(ERROR) << "In finish_migrate";
hash_map_.store(migrate_to_hash_map_);
hp_.retire(get_thread_id(), migrate_from_hash_map_);
migrate_from_hash_map_ = nullptr;
migrate_to_hash_map_ = nullptr;
migrate_generation_++;
migrate_cv_.notify_all();
}
void init_migrate() {
if (migrate_from_hash_map_ != nullptr) {
return;
}
//LOG(ERROR) << "In init_migrate";
CHECK(migrate_cnt_ == 0);
migrate_generation_++;
migrate_from_hash_map_ = hash_map_.exchange(nullptr);
auto new_size = migrate_from_hash_map_->size() * 2;
migrate_to_hash_map_ = make_unique<HashMap>(new_size).release();
task_creator.chunk_size = 100;
task_creator.size = migrate_from_hash_map_->size();
task_creator.pos = 0;
}
void run_migrate() {
//LOG(ERROR) << "In run_migrate";
size_t cnt = 0;
while (true) {
auto task = task_creator.create();
cnt += task.size();
if (task.empty()) {
break;
}
run_task(task);
}
//LOG(ERROR) << "In run_migrate " << cnt;
}
void run_task(Task task) {
for (auto i = task.begin; i < task.end; i++) {
auto &node = migrate_from_hash_map_->node_at(i);
auto old_value = node.value.exchange(migrate_value(), std::memory_order_acq_rel);
if (old_value == 0) {
continue;
}
auto node_key = node.key.load(std::memory_order_relaxed);
//LOG(ERROR) << node_key << " " << node_key;
auto ok = migrate_to_hash_map_->with_value(
node_key, true, [&](auto &node_value) { node_value.store(old_value, std::memory_order_relaxed); });
LOG_CHECK(ok) << "Migration overflow";
}
}
};
template <class KeyT, class ValueT>
td::HazardPointers<typename ConcurrentHashMap<KeyT, ValueT>::HashMap> ConcurrentHashMap<KeyT, ValueT>::hp_(64);
} // namespace td