实现线程安全的缓存¶
来源：https://github.com/byrnexu/betterquant2
/*!
* \file LRUCache.cpp
* \project BetterQuant
*
* \author byrnexu
* \date 2023/02/09
*
* \brief
*/
#pragma once
#include <cassert>
#include <cstddef>
#include <list>
#include <sstream>
#include <stdexcept>
#include <unordered_map>
namespace std::ext {
template <typename key_t, typename value_t>
class lru_cache;
template <typename key_t, typename value_t>
using lru_cache_sptr = std::shared_ptr<lru_cache<key_t, value_t>>;
template <typename key_t, typename value_t>
class lru_cache {
public:
typedef typename std::pair<key_t, value_t> key_value_pair_t;
typedef typename std::list<key_value_pair_t>::iterator list_iterator_t;
explicit lru_cache(std::size_t capacity) : capacity_(capacity) {}
void push(const key_t& key, const value_t& value,
bool key_must_be_unique = true) {
assert(cache_items_map_.size() == cache_items_list_.size());
cache_items_list_.push_front(key_value_pair_t(key, value));
if (key_must_be_unique) {
auto it = cache_items_map_.find(key);
if (it != cache_items_map_.end()) {
cache_items_list_.erase(it->second);
cache_items_map_.erase(it);
}
}
cache_items_map_.emplace(key, cache_items_list_.begin());
if (cache_items_map_.size() > capacity_) {
auto last = cache_items_list_.end();
last--;
cache_items_map_.erase(last->first);
cache_items_list_.pop_back();
}
}
std::pair<bool, value_t> pop(const key_t& key) {
assert(cache_items_map_.size() == cache_items_list_.size());
value_t value{};
const auto it = cache_items_map_.find(key);
if (it == cache_items_map_.end()) {
return std::make_pair(false, value);
} else {
value = it->second->second;
cache_items_list_.erase(it->second);
cache_items_map_.erase(it);
return std::make_pair(true, value);
}
return std::make_pair(false, value);
}
std::pair<bool, value_t> get(const key_t& key) {
assert(cache_items_map_.size() == cache_items_list_.size());
value_t value{};
const auto it = cache_items_map_.find(key);
if (it == cache_items_map_.end()) {
return std::make_pair(false, value);
} else {
cache_items_list_.splice(cache_items_list_.begin(), cache_items_list_,
it->second);
return std::make_pair(true, it->second->second);
}
return std::make_pair(false, value);
}
inline bool exists(const key_t& key) const {
assert(cache_items_map_.size() == cache_items_list_.size());
return cache_items_map_.find(key) != cache_items_map_.end();
}
inline std::string to_string() const {
std::ostringstream oss;
for (const auto& rec : cache_items_list_) {
oss << rec.first << "=" << rec.second << ",";
}
std::string ret = oss.str();
if (ret.empty() == false) ret.pop_back();
return ret;
}
inline void clear() {
assert(cache_items_map_.size() == cache_items_list_.size());
cache_items_list_.clear();
cache_items_map_.clear();
}
inline std::size_t size() const { return cache_items_map_.size(); }
inline std::size_t capacity() const noexcept { return capacity_; }
private:
const std::size_t capacity_;
std::list<key_value_pair_t> cache_items_list_;
std::unordered_multimap<key_t, list_iterator_t> cache_items_map_;
};
} // namespace std::ext
#pragma once
#include <array>
#include <atomic>
#include <cstddef>
#include <cstdint>
#include <optional>
#include <thread>
#include <type_traits>
// 在x86/x64架构下，用于在自旋等待中降低CPU功耗并提示超线程
#if defined(__i386__) || defined(__x86_64__)
#include <immintrin.h>
#define LFC_CPU_RELAX() _mm_pause()
#else
#define LFC_CPU_RELAX() std::this_thread::yield()
#endif
namespace lfc {
/**
* @brief 编译时FNV-1a哈希函数
*/
constexpr uint64_t fnv1a_hash(const char* str, size_t len) noexcept {
uint64_t hash = 14695981039346656037ull;
for (size_t i = 0; i < len; ++i) {
hash ^= static_cast<uint64_t>(str[i]);
hash *= 1099511628211ull;
}
return hash;
}
/**
* @brief 默认的哈希器，适用于任何可平凡复制的类型
*/
template <typename Key>
struct DefaultHasher {
uint64_t operator()(const Key& key) const noexcept {
static_assert(std::is_trivially_copyable_v<Key>,
"DefaultHasher requires a trivially copyable key.");
return fnv1a_hash(reinterpret_cast<const char*>(&key), sizeof(Key));
}
};
/**
* @brief 一个基于线性探测和标签位域的、线程安全的无锁哈希表（缓存）
* @tparam Key 键的类型。必须是可平凡复制的 (trivially copyable)。
* @tparam Value 值的类型。必须是可平凡复制的 (trivially copyable)。
* @tparam TableSize 哈希表的大小。必须是2的幂。
* @tparam Hasher 用于计算键哈希值的函数对象。
*/
template <typename Key, typename Value, size_t TableSize = 4096,
typename Hasher = DefaultHasher<Key> >
class LockFreeCache {
static_assert((TableSize > 0) && ((TableSize & (TableSize - 1)) == 0),
"TableSize must be a power of 2");
static_assert(std::is_trivially_copyable_v<Key> &&
std::is_trivially_copyable_v<Value>,
"Key and Value must be trivially copyable for this lock-free "
"implementation.");
private:
// Tag位域定义: | 63: valid | 62: writing | 61..0: hash |
static constexpr uint64_t VALID_BIT = 1ULL << 63;
static constexpr uint64_t WRITING_BIT = 1ULL << 62;
static constexpr uint64_t HASH_MASK = WRITING_BIT - 1;
struct alignas(64) Entry {
std::atomic<uint64_t> tag;
Key key;
Value value;
};
std::array<Entry, TableSize> table;
Hasher hasher;
static size_t probe_next(size_t index) {
return (index + 1) & (TableSize - 1);
}
public:
LockFreeCache() {
for (auto& entry : table) {
entry.tag.store(0, std::memory_order_relaxed);
}
}
bool update(const Key& key, const Value& value) {
uint64_t hash_val = hasher(key);
size_t index = hash_val & (TableSize - 1);
uint64_t partial_hash = hash_val & HASH_MASK;
for (size_t i = 0; i < TableSize; ++i) {
Entry* entry = &table[index];
uint64_t current_tag = entry->tag.load(std::memory_order_relaxed);
while (true) {
if (current_tag == 0) { // Case 1: 空槽位
uint64_t new_tag = WRITING_BIT | partial_hash;
if (entry->tag.compare_exchange_strong(current_tag, new_tag,
std::memory_order_acq_rel)) {
entry->key = key;
entry->value = value;
entry->tag.store(VALID_BIT | partial_hash,
std::memory_order_release);
return true;
}
continue;
}
if (current_tag & WRITING_BIT) { // Case 2: 正在写入
LFC_CPU_RELAX();
current_tag = entry->tag.load(std::memory_order_relaxed);
continue;
}
if ((current_tag & HASH_MASK) == partial_hash) { // Case 3: 有效槽位
std::atomic_thread_fence(std::memory_order_acquire);
if (entry->key == key) {
uint64_t expected_tag = VALID_BIT | partial_hash;
uint64_t new_tag = WRITING_BIT | partial_hash;
if (entry->tag.compare_exchange_strong(expected_tag, new_tag,
std::memory_order_acq_rel)) {
entry->value = value;
entry->tag.store(VALID_BIT | partial_hash,
std::memory_order_release);
return true;
}
current_tag = expected_tag;
continue;
}
}
break;
}
index = probe_next(index);
}
return false; // 表已满
}
std::optional<Value> get(const Key& key) const {
uint64_t hash_val = hasher(key);
size_t index = hash_val & (TableSize - 1);
uint64_t partial_hash = hash_val & HASH_MASK;
for (size_t i = 0; i < TableSize; ++i) {
const Entry* entry = &table[index];
uint64_t current_tag = entry->tag.load(std::memory_order_acquire);
while (true) {
if (current_tag == 0) {
return std::nullopt;
}
if (current_tag & WRITING_BIT) {
LFC_CPU_RELAX();
current_tag = entry->tag.load(std::memory_order_acquire);
continue;
}
if ((current_tag & HASH_MASK) == partial_hash) {
if (entry->key == key) {
return entry->value;
}
}
break;
}
index = probe_next(index);
}
return std::nullopt;
}
};
} // namespace lfc
#include <atomic>
#include <chrono>
#include <cstdio>
#include <cstring>
#include <functional>
#include <iomanip>
#include <iostream>
#include <string>
#include <thread>
#include <vector>
#include "lockfree_cache.hpp"
// --- 定义测试用的 Key 和 Value ---
struct SymbolKey {
char name[8]{};
SymbolKey() = default;
constexpr SymbolKey(const char* n) {
for (size_t i = 0; i < 7 && n[i] != '\0'; ++i) {
name[i] = n[i];
}
}
bool operator==(const SymbolKey& other) const {
return std::memcmp(name, other.name, 8) == 0;
}
};
namespace lfc {
template <>
struct DefaultHasher<SymbolKey> {
uint64_t operator()(const SymbolKey& key) const noexcept {
return fnv1a_hash(key.name, strnlen(key.name, 8));
}
};
} // namespace lfc
struct MarketData {
double price;
uint64_t volume;
bool operator==(const MarketData& other) const {
return price == other.price && volume == other.volume;
}
};
// 将缓存容量从 16384 增加到 65536，以容纳所有 40000 个键。
lfc::LockFreeCache<SymbolKey, MarketData, 65536> market_cache;
void writer_thread(int thread_id, int num_keys, std::atomic<bool>& start_flag) {
while (!start_flag.load()) {
} // 等待开始信号
for (int i = 0; i < num_keys; ++i) {
std::string key_str = "K" + std::to_string(thread_id * num_keys + i);
market_cache.update(SymbolKey(key_str.c_str()),
{(double)i, (uint64_t)thread_id});
}
}
void reader_thread(int num_writers, int num_keys_per_writer,
std::atomic<bool>& start_flag,
std::atomic<int>& found_count) {
while (!start_flag.load()) {
} // 等待开始信号
int found = 0;
for (int i = 0; i < num_writers * num_keys_per_writer; ++i) {
std::string key_str = "K" + std::to_string(i);
if (market_cache.get(SymbolKey(key_str.c_str()))) {
found++;
}
}
found_count += found;
}
int main() {
const int NUM_WRITERS = 4;
const int NUM_READERS = 4;
const int KEYS_PER_WRITER = 10000;
std::vector<std::thread> threads;
std::atomic<bool> start_flag(false);
std::atomic<int> total_found(0);
std::cout << "准备并发测试..." << std::endl;
std::cout << NUM_WRITERS << "个写入线程, " << NUM_READERS << "个读取线程。"
<< std::endl;
std::cout << "每个写入线程将写入 " << KEYS_PER_WRITER << " 个键。"
<< std::endl;
for (int i = 0; i < NUM_WRITERS; ++i) {
threads.emplace_back(writer_thread, i, KEYS_PER_WRITER,
std::ref(start_flag));
}
for (int i = 0; i < NUM_READERS; ++i) {
threads.emplace_back(reader_thread, NUM_WRITERS, KEYS_PER_WRITER,
std::ref(start_flag), std::ref(total_found));
}
std::cout << "3秒后开始..." << std::endl;
std::this_thread::sleep_for(std::chrono::seconds(3));
auto test_start_time = std::chrono::high_resolution_clock::now();
start_flag.store(true);
for (auto& t : threads) {
t.join();
}
auto test_end_time = std::chrono::high_resolution_clock::now();
// 主线程做最终验证
int final_check_found = 0;
const int total_keys = NUM_WRITERS * KEYS_PER_WRITER;
for (int i = 0; i < total_keys; ++i) {
std::string key_str = "K" + std::to_string(i);
if (market_cache.get(SymbolKey(key_str.c_str()))) {
final_check_found++;
}
}
std::chrono::duration<double, std::milli> test_duration =
test_end_time - test_start_time;
std::cout << "--- 测试结果 ---" << std::endl;
std::cout << "所有线程执行完毕，耗时: " << std::fixed << std::setprecision(2)
<< test_duration.count() << " ms" << std::endl;
std::cout << "读取线程在并发期间共找到 " << total_found.load() << " 个键。"
<< std::endl;
std::cout << "主线程最终检查，在缓存中找到 " << final_check_found << " / "
<< total_keys << " 个键。" << std::endl;
if (final_check_found == total_keys) {
std::cout << "测试成功: 所有写入的键都被正确找到。" << std::endl;
} else {
std::cout << "测试失败: " << total_keys - final_check_found
<< " 个键丢失或未被正确写入。" << std::endl;
}
return 0;
}
zhixuan@gpu04:~/cacheimpl$ ./cache_test
准备并发测试...
4个写入线程, 4个读取线程。
每个写入线程将写入 10000 个键。
3秒后开始...
--- 测试结果 ---
所有线程执行完毕，耗时: 5.29 ms
读取线程在并发期间共找到 120427 个键。
主线程最终检查，在缓存中找到 40000 / 40000 个键。
测试成功: 所有写入的键都被正确找到。
#pragma once
#include <array>
#include <atomic>
#include <cstddef>
#include <cstdint>
#include <limits>
#include <optional>
#include <thread>
#include <type_traits>
#if defined(__i386__) || defined(__x86_64__)
#include <immintrin.h>
#define LFC_CPU_RELAX() _mm_pause()
#else
#define LFC_CPU_RELAX() std::this_thread::yield()
#endif
namespace lfc {
constexpr uint64_t fnv1a_hash(const char* str, size_t len) noexcept {
uint64_t hash = 14695981039346656037ull;
for (size_t i = 0; i < len; ++i) {
hash ^= static_cast<uint64_t>(str[i]);
hash *= 1099511628211ull;
}
return hash;
}
template <typename Key>
struct DefaultHasher {
uint64_t operator()(const Key& key) const noexcept {
static_assert(std::is_trivially_copyable_v<Key>,
"DefaultHasher requires a trivially copyable key.");
return fnv1a_hash(reinterpret_cast<const char*>(&key), sizeof(Key));
}
};
template <typename Key, typename Value, size_t TableSize = 4096,
typename Hasher = DefaultHasher<Key> >
class LockFreeLruCache {
static_assert((TableSize > 0) && ((TableSize & (TableSize - 1)) == 0),
"TableSize must be a power of 2");
static_assert(std::is_trivially_copyable_v<Key> &&
std::is_trivially_copyable_v<Value>,
"Key and Value must be trivially copyable for this lock-free "
"implementation.");
private:
static constexpr uint64_t VALID_BIT = 1ULL << 63;
static constexpr uint64_t WRITING_BIT = 1ULL << 62;
static constexpr uint64_t HASH_MASK = WRITING_BIT - 1;
struct alignas(64) Entry {
std::atomic<uint64_t> tag;
Key key;
Value value;
std::atomic<uint64_t> last_access_time;
};
std::array<Entry, TableSize> table_;
Hasher hasher_;
std::atomic<uint64_t> access_timer_{0};
static size_t probe_next(size_t index) {
return (index + 1) & (TableSize - 1);
}
public:
LockFreeLruCache() {
for (auto& entry : table_) {
entry.tag.store(0, std::memory_order_relaxed);
entry.last_access_time.store(0, std::memory_order_relaxed);
}
}
bool put(const Key& key, const Value& value) {
uint64_t hash_val = hasher_(key);
size_t index = hash_val & (TableSize - 1);
uint64_t partial_hash = hash_val & HASH_MASK;
uint64_t new_time = access_timer_.fetch_add(1, std::memory_order_relaxed);
for (size_t i = 0; i < TableSize; ++i) {
Entry* entry = &table_[index];
uint64_t current_tag = entry->tag.load(std::memory_order_relaxed);
while (true) {
if (current_tag == 0) {
uint64_t new_tag_val = WRITING_BIT | partial_hash;
if (entry->tag.compare_exchange_strong(current_tag, new_tag_val,
std::memory_order_acq_rel)) {
entry->key = key;
entry->value = value;
entry->last_access_time.store(new_time, std::memory_order_relaxed);
entry->tag.store(VALID_BIT | partial_hash,
std::memory_order_release);
return true;
}
continue;
}
if (current_tag & WRITING_BIT) {
LFC_CPU_RELAX();
current_tag = entry->tag.load(std::memory_order_relaxed);
continue;
}
if ((current_tag & HASH_MASK) == partial_hash) {
std::atomic_thread_fence(std::memory_order_acquire);
if (entry->key == key) {
uint64_t expected_tag = VALID_BIT | partial_hash;
uint64_t new_tag_val = WRITING_BIT | partial_hash;
if (entry->tag.compare_exchange_strong(expected_tag, new_tag_val,
std::memory_order_acq_rel)) {
entry->value = value;
entry->last_access_time.store(new_time,
std::memory_order_relaxed);
entry->tag.store(VALID_BIT | partial_hash,
std::memory_order_release);
return true;
}
current_tag = expected_tag;
continue;
}
}
break;
}
index = probe_next(index);
}
// Phase 2: LRU Eviction.
// **LIVELOCK FIX: The outer loop ensures we eventually succeed, even under
// high contention.**
while (true) {
size_t victim_index = 0;
uint64_t min_time = std::numeric_limits<uint64_t>::max();
for (size_t i = 0; i < TableSize; ++i) {
uint64_t current_tag = table_[i].tag.load(std::memory_order_acquire);
if ((current_tag & VALID_BIT) && !(current_tag & WRITING_BIT)) {
uint64_t access_time =
table_[i].last_access_time.load(std::memory_order_relaxed);
if (access_time < min_time) {
min_time = access_time;
victim_index = i;
}
}
}
Entry* victim_entry = &table_[victim_index];
uint64_t expected_tag = victim_entry->tag.load(std::memory_order_relaxed);
// **LIVELOCK FIX: We removed the check on last_access_time here.**
// We only check if the slot is still valid. If so, we proceed forcefully.
// If another thread writes or removes it, expected_tag will change and
// CAS will fail, causing a retry, which is correct.
if (!(expected_tag & VALID_BIT) || (expected_tag & WRITING_BIT)) {
continue; // The victim was already claimed or removed by another
// writer. Find a new one.
}
uint64_t new_tag = WRITING_BIT | partial_hash;
if (victim_entry->tag.compare_exchange_strong(
expected_tag, new_tag, std::memory_order_acq_rel)) {
// Success, we claimed the slot.
victim_entry->key = key;
victim_entry->value = value;
victim_entry->last_access_time.store(new_time,
std::memory_order_relaxed);
victim_entry->tag.store(VALID_BIT | partial_hash,
std::memory_order_release);
return true; // Exit the function
}
// CAS failed. This means another thread (writer or remover) changed the
// tag. The outer while(true) loop will cause us to retry the entire
// eviction process.
}
}
// get() and remove() methods are unchanged
std::optional<Value> get(const Key& key) {
uint64_t hash_val = hasher_(key);
size_t index = hash_val & (TableSize - 1);
uint64_t partial_hash = hash_val & HASH_MASK;
for (size_t i = 0; i < TableSize; ++i) {
Entry* entry = &table_[index];
uint64_t current_tag = entry->tag.load(std::memory_order_acquire);
while (true) {
if (current_tag == 0) {
return std::nullopt;
}
if (current_tag & WRITING_BIT) {
LFC_CPU_RELAX();
current_tag = entry->tag.load(std::memory_order_acquire);
continue;
}
if ((current_tag & HASH_MASK) == partial_hash) {
if (entry->key == key) {
uint64_t new_time =
access_timer_.fetch_add(1, std::memory_order_relaxed);
entry->last_access_time.store(new_time, std::memory_order_relaxed);
return entry->value;
}
}
break;
}
index = probe_next(index);
}
return std::nullopt;
}
bool remove(const Key& key) {
uint64_t hash_val = hasher_(key);
size_t index = hash_val & (TableSize - 1);
uint64_t partial_hash = hash_val & HASH_MASK;
for (size_t i = 0; i < TableSize; ++i) {
Entry* entry = &table_[index];
uint64_t current_tag = entry->tag.load(std::memory_order_acquire);
while (true) {
if (current_tag == 0) {
return false;
}
if (current_tag & WRITING_BIT) {
LFC_CPU_RELAX();
current_tag = entry->tag.load(std::memory_order_acquire);
continue;
}
if ((current_tag & HASH_MASK) == partial_hash) {
std::atomic_thread_fence(std::memory_order_acquire);
if (entry->key == key) {
uint64_t expected_tag = VALID_BIT | partial_hash;
if (entry->tag.compare_exchange_strong(expected_tag, 0,
std::memory_order_acq_rel)) {
return true;
}
current_tag = expected_tag;
continue;
}
}
break;
}
index = probe_next(index);
}
return false;
}
};
} // namespace lfc
#include <atomic>
#include <cassert>
#include <chrono>
#include <cstring>
#include <functional>
#include <iomanip>
#include <iostream>
#include <numeric>
#include <random>
#include <thread>
#include <vector>
#include "LockFreeLruCache.hpp"
#ifdef _WIN32
#include <windows.h>
void set_console_color(int color) {
SetConsoleTextAttribute(GetStdHandle(STD_OUTPUT_HANDLE), color);
}
#endif
enum class Color { GREEN = 10, YELLOW = 14, RED = 12, BLUE = 9, DEFAULT = 7 };
void print_status(const std::string& msg, Color color) {
#ifdef _WIN32
set_console_color(static_cast<int>(color));
#else
const char* code = "\033[39m";
if (color == Color::GREEN)
code = "\033[32m";
else if (color == Color::YELLOW)
code = "\033[33m";
else if (color == Color::RED)
code = "\033[31m";
else if (color == Color::BLUE)
code = "\033[34m";
std::cout << code;
#endif
std::cout << msg;
#ifdef _WIN32
set_console_color(static_cast<int>(Color::DEFAULT));
#else
std::cout << "\033[39m";
#endif
std::cout << std::endl;
}
struct TriviallyCopyableString {
char data[16];
TriviallyCopyableString() { data[0] = '\0'; }
TriviallyCopyableString(const char* str) {
strncpy(data, str, sizeof(data) - 1);
data[sizeof(data) - 1] = '\0';
}
bool operator==(const char* other) const { return strcmp(data, other) == 0; }
};
void test_single_thread_lru() {
print_status("\n--- 正在运行单线程LRU逻辑验证 ---", Color::YELLOW);
lfc::LockFreeLruCache<int, TriviallyCopyableString, 4> cache;
print_status("[1] 填充缓存 (4个项目)...", Color::DEFAULT);
cache.put(1, "one");
cache.put(2, "two");
cache.put(3, "three");
cache.put(4, "four");
assert(cache.get(1).value_or(TriviallyCopyableString()) == "one");
assert(cache.get(2).value_or(TriviallyCopyableString()) == "two");
assert(cache.get(3).value_or(TriviallyCopyableString()) == "three");
assert(cache.get(4).value_or(TriviallyCopyableString()) == "four");
print_status(" ...缓存已满，所有项目已验证。", Color::GREEN);
print_status("[2] 访问键 2，使其成为最近使用的项...", Color::DEFAULT);
cache.get(2);
print_status("[3] 插入键 5，应淘汰键 1...", Color::DEFAULT);
cache.put(5, "five");
assert(!cache.get(1).has_value());
assert(cache.get(2).has_value());
assert(cache.get(3).has_value());
assert(cache.get(4).has_value());
assert(cache.get(5).has_value());
print_status(" ...已验证键 1 被淘汰，键 5 已添加。", Color::GREEN);
print_status("[4] 移除键 3...", Color::DEFAULT);
assert(cache.remove(3));
assert(!cache.get(3).has_value());
print_status(" ...已验证键 3 已被移除。", Color::GREEN);
assert(!cache.get(999).has_value());
print_status("[5] 已验证获取不存在的键返回nullopt。", Color::GREEN);
print_status("--- 单线程LRU测试通过 ---", Color::GREEN);
}
void test_multi_thread_lru_contention() {
print_status("\n--- 正在运行多线程LRU竞争测试 ---", Color::YELLOW);
const size_t CACHE_SIZE = 128;
const int HOT_KEYS = 32;
const int NUM_THREADS = 8;
lfc::LockFreeLruCache<int, int, CACHE_SIZE> cache;
std::atomic<bool> stop_signal{false};
print_status("[1] 预填充缓存至满容量...", Color::DEFAULT);
for (size_t i = 0; i < CACHE_SIZE; ++i) {
cache.put(i, i);
}
std::vector<std::thread> threads;
print_status("[2] 启动线程以重复访问 '热' 键 (0-31)...", Color::DEFAULT);
for (int i = 0; i < NUM_THREADS; ++i) {
threads.emplace_back([&, i]() {
// **CORRECTED LOGIC**: Each thread now accesses *random* hot keys
std::mt19937 rng(
std::chrono::steady_clock::now().time_since_epoch().count() + i);
std::uniform_int_distribution<int> dist(0, HOT_KEYS - 1);
while (!stop_signal.load()) {
int key_to_access = dist(rng);
cache.get(key_to_access);
std::this_thread::yield();
}
});
}
print_status("[3] 启动线程以连续插入新键，强制进行淘汰...", Color::DEFAULT);
threads.emplace_back([&]() {
int current_key = CACHE_SIZE;
while (!stop_signal.load()) {
cache.put(current_key, current_key);
current_key++;
std::this_thread::yield();
}
});
std::this_thread::sleep_for(std::chrono::seconds(3));
stop_signal.store(true);
for (auto& t : threads) {
t.join();
}
print_status("[4] 线程已完成，正在验证结果...", Color::DEFAULT);
int hot_keys_found = 0;
for (int i = 0; i < HOT_KEYS; ++i) {
if (cache.get(i).has_value()) {
hot_keys_found++;
}
}
print_status(" ...在缓存中找到的热键数量: " + std::to_string(hot_keys_found) +
" / " + std::to_string(HOT_KEYS),
Color::BLUE);
assert(hot_keys_found > HOT_KEYS * 0.95);
print_status("--- 多线程LRU竞争测试通过 ---", Color::GREEN);
}
void test_stress_concurrent_read_write() {
print_status("\n--- 正在运行并发读写及压力测试 ---", Color::YELLOW);
const size_t CACHE_SIZE = 8192;
const int NUM_WRITERS = 4;
const int NUM_READERS =
std::thread::hardware_concurrency() - NUM_WRITERS > 0
? std::thread::hardware_concurrency() - NUM_WRITERS
: 4;
const int OPS_PER_THREAD = 200000;
const int KEY_RANGE = OPS_PER_THREAD * NUM_WRITERS;
lfc::LockFreeLruCache<int, int, CACHE_SIZE> cache;
std::vector<std::thread> threads;
std::atomic<long long> total_ops{0};
print_status("[1] 启动 " + std::to_string(NUM_WRITERS) + " 个写入线程和 " +
std::to_string(NUM_READERS) + " 个读取线程...",
Color::DEFAULT);
auto start_time = std::chrono::high_resolution_clock::now();
for (int i = 0; i < NUM_WRITERS; ++i) {
threads.emplace_back([&, i]() {
for (int j = 0; j < OPS_PER_THREAD; ++j) {
int key = i * OPS_PER_THREAD + j;
cache.put(key, key);
}
total_ops += OPS_PER_THREAD;
});
}
for (int i = 0; i < NUM_READERS; ++i) {
threads.emplace_back([&]() {
std::mt19937 rng(
std::chrono::steady_clock::now().time_since_epoch().count());
std::uniform_int_distribution<int> dist(0, KEY_RANGE - 1);
for (int j = 0; j < OPS_PER_THREAD; ++j) {
int key = dist(rng);
auto result = cache.get(key);
if (result.has_value()) {
assert(result.value() == key);
}
}
total_ops += OPS_PER_THREAD;
});
}
for (auto& t : threads) {
t.join();
}
auto end_time = std::chrono::high_resolution_clock::now();
std::chrono::duration<double> duration = end_time - start_time;
print_status("[2] 所有线程已完成，未发生崩溃。", Color::GREEN);
long long final_total_ops = total_ops.load();
double ops_per_second = final_total_ops / duration.count();
std::cout << std::fixed << std::setprecision(2);
print_status(" 总操作数: " + std::to_string(final_total_ops), Color::BLUE);
print_status(" 执行时间: " + std::to_string(duration.count()) + " 秒",
Color::BLUE);
print_status(
" 吞吐率: " + std::to_string(ops_per_second / 1e6) + " 百万次操作/秒",
Color::BLUE);
print_status("--- 并发读写及压力测试通过 ---", Color::GREEN);
}
int main() {
try {
test_single_thread_lru();
test_multi_thread_lru_contention();
test_stress_concurrent_read_write();
} catch (const std::exception& e) {
print_status("发生异常: " + std::string(e.what()), Color::RED);
return 1;
} catch (...) {
print_status("发生未知异常。", Color::RED);
return 1;
}
print_status("\n所有测试均已成功完成！", Color::GREEN);
return 0;
}
zhixuan @gpu04 : ~/ Cache$ g++ - std =
c++ 17 - o cache_test main.cpp - pthread -
Wall zhixuan @gpu04 : ~/ Cache$./
cache_test
-- -
正在运行单线程LRU逻辑验证-- -
[1] 填充缓存(4个项目)...... 缓存已满，所有项目已验证。
[2] 访问键 2，使其成为最近使用的项...
[3] 插入键 5，应淘汰键 1... ... 已验证键 1 被淘汰，键 5 已添加。
[4] 移除键 3... ... 已验证键 3 已被移除。
[5] 已验证获取不存在的键返回nullopt。-- -
单线程LRU测试通过-- -
-- -正在运行多线程LRU竞争测试-- -
[1] 预填充缓存至满容量...[2] 启动线程以重复访问 '热' 键(
0 - 31)...[3] 启动线程以连续插入新键，强制进行淘汰...
[4] 线程已完成，正在验证结果...... 在缓存中找到的热键数量 : 32 /
32 -- -多线程LRU竞争测试通过-- -
-- -正在运行并发读写及压力测试-- -
[1] 启动 4 个写入线程和 92 个读取线程...[2] 所有线程已完成，未发生崩溃。
总操作数 : 19200000 执行时间 : 77.114936 秒 吞吐率
: 0.248979 百万次操作 /
秒-- -
并发读写及压力测试通过-- -
所有测试均已成功完成！ zhixuan @gpu04 : ~/ Cache$
#pragma once
#include <array>
#include <atomic>
#include <cstddef>
#include <cstdint>
#include <optional>
#include <thread>
#include <type_traits>
// 在x86/x64架构下，用于在自旋等待中降低CPU功耗并提示超线程
#if defined(__i386__) || defined(__x86_64__)
#include <immintrin.h>
#define LFC_CPU_RELAX() _mm_pause()
#else
#define LFC_CPU_RELAX() std::this_thread::yield()
#endif
namespace lfc {
/**
* @brief 编译时FNV-1a哈希函数
* @param str 指向要哈希的字符数组的指针
* @param len 字符数组的长度
* @return 64位的哈希值
*/
constexpr uint64_t fnv1a_hash(const char* str, size_t len) noexcept {
uint64_t hash = 14695981039346656037ull;
for (size_t i = 0; i < len; ++i) {
hash ^= static_cast<uint64_t>(str[i]);
hash *= 1099511628211ull;
}
return hash;
}
/**
* @brief 默认的哈希器，适用于任何可平凡复制的类型
* @tparam Key 要哈希的键类型
*/
template <typename Key>
struct DefaultHasher {
uint64_t operator()(const Key& key) const noexcept {
static_assert(std::is_trivially_copyable_v<Key>,
"DefaultHasher requires a trivially copyable key.");
return fnv1a_hash(reinterpret_cast<const char*>(&key), sizeof(Key));
}
};
/**
* @brief
* 一个基于线性探测和标签位域的、线程安全的无锁哈希表（缓存），集成了Clock淘汰策略。
* @tparam Key 键的类型。必须是可平凡复制的 (trivially copyable)。
* @tparam Value 值的类型。必须是可平凡复制的 (trivially copyable)。
* @tparam TableSize 哈希表的大小。必须是2的幂。
* @tparam Hasher 用于计算键哈希值的函数对象。
*/
template <typename Key, typename Value, size_t TableSize = 4096,
typename Hasher = DefaultHasher<Key> >
class LockFreeCache {
static_assert((TableSize > 0) && ((TableSize & (TableSize - 1)) == 0),
"TableSize must be a power of 2");
static_assert(std::is_trivially_copyable_v<Key> &&
std::is_trivially_copyable_v<Value>,
"Key and Value must be trivially copyable for this lock-free "
"implementation.");
private:
// | 63: valid | 62: writing | 61: used | 60..0: hash |
static constexpr uint64_t VALID_BIT = 1ULL << 63;
static constexpr uint64_t WRITING_BIT = 1ULL << 62;
static constexpr uint64_t USED_BIT = 1ULL << 61;
static constexpr uint64_t HASH_MASK = USED_BIT - 1;
struct alignas(64) Entry {
std::atomic<uint64_t> tag;
Key key;
Value value;
};
std::array<Entry, TableSize> table;
Hasher hasher;
// 时钟指针，用于Clock淘汰算法
alignas(64) std::atomic<size_t> clock_hand_;
static size_t probe_next(size_t index) {
return (index + 1) & (TableSize - 1);
}
public:
LockFreeCache() : clock_hand_(0) {
for (auto& entry : table) {
entry.tag.store(0, std::memory_order_relaxed);
}
}
/**
* @brief
* 更新或插入一个键值对。如果缓存已满，则会尝试淘汰一个最近未使用的条目。
* @param key 要插入/更新的键
* @param value 要关联的值
* @return 成功则返回true，如果尝试淘汰后仍然失败则返回false。
*/
bool update(const Key& key, const Value& value) {
uint64_t hash_val = hasher(key);
size_t index = hash_val & (TableSize - 1);
uint64_t partial_hash = hash_val & HASH_MASK;
// 阶段1：尝试查找空位或更新现有键
for (size_t i = 0; i < TableSize; ++i) {
Entry* entry = &table[index];
uint64_t current_tag = entry->tag.load(std::memory_order_relaxed);
while (true) {
if (current_tag == 0) { // Case 1: 空槽位
uint64_t new_tag = WRITING_BIT | partial_hash;
if (entry->tag.compare_exchange_strong(current_tag, new_tag,
std::memory_order_acq_rel)) {
entry->key = key;
entry->value = value;
// 插入新数据时，将其标记为VALID和USED
entry->tag.store(VALID_BIT | USED_BIT | partial_hash,
std::memory_order_release);
return true;
}
continue;
}
if (current_tag & WRITING_BIT) { // Case 2: 正在写入
LFC_CPU_RELAX();
current_tag = entry->tag.load(std::memory_order_relaxed);
continue;
}
if ((current_tag & HASH_MASK) == partial_hash) { // Case 3: 有效槽位
std::atomic_thread_fence(std::memory_order_acquire);
if (entry->key == key) {
uint64_t expected_tag = current_tag; // 使用最新的tag值
uint64_t new_tag = WRITING_BIT | partial_hash;
if (entry->tag.compare_exchange_strong(expected_tag, new_tag,
std::memory_order_acq_rel)) {
entry->value = value;
// 更新数据后，标记为VALID和USED
entry->tag.store(VALID_BIT | USED_BIT | partial_hash,
std::memory_order_release);
return true;
}
current_tag = expected_tag;
continue;
}
}
// 哈希冲突或非空槽，探测下一个
break;
}
index = probe_next(index);
}
// 阶段2：如果上面失败了 (表可能满了)，则启动Clock淘汰机制
constexpr size_t EVICTION_CANDIDATES = 20; // 每次尝试淘汰的候选数量
for (size_t i = 0; i < EVICTION_CANDIDATES; ++i) {
size_t evict_index =
clock_hand_.fetch_add(1, std::memory_order_relaxed) & (TableSize - 1);
Entry* entry = &table[evict_index];
uint64_t current_tag = entry->tag.load(std::memory_order_relaxed);
// 只淘汰有效的、未被写入的条目
if ((current_tag & VALID_BIT) && !(current_tag & WRITING_BIT)) {
if (current_tag & USED_BIT) {
// 如果使用位为1, 给他第二次机会，清零使用位
uint64_t new_tag = current_tag & ~USED_BIT;
entry->tag.compare_exchange_strong(current_tag, new_tag,
std::memory_order_relaxed);
} else {
// 如果使用位为0, 尝试淘汰它
uint64_t new_tag =
WRITING_BIT | (current_tag & HASH_MASK); // 保持旧的哈希
if (entry->tag.compare_exchange_strong(current_tag, new_tag,
std::memory_order_acq_rel)) {
// 成功锁定，现在可以安全地替换内容
entry->key = key;
entry->value = value;
entry->tag.store(VALID_BIT | USED_BIT | partial_hash,
std::memory_order_release);
return true;
}
}
}
}
return false; // 尝试淘汰后仍然失败
}
/**
* @brief 查找一个键。此操作是线程安全的。
* @param key 要查找的键
* @return 如果找到，返回包含值的std::optional，否则返回std::nullopt。
*/
std::optional<Value> get(const Key& key) const {
uint64_t hash_val = hasher(key);
size_t index = hash_val & (TableSize - 1);
uint64_t partial_hash = hash_val & HASH_MASK;
for (size_t i = 0; i < TableSize; ++i) {
// NOTE: get是const方法，所以需要const_cast来修改tag。
// 这是一个在并发数据结构中常见的做法。
// fetch_or是原子性的，因此线程安全。
Entry* entry = const_cast<Entry*>(&table[index]);
uint64_t current_tag = entry->tag.load(std::memory_order_acquire);
while (true) {
if (current_tag == 0) {
return std::nullopt;
}
if (current_tag & WRITING_BIT) {
LFC_CPU_RELAX();
current_tag = entry->tag.load(std::memory_order_acquire);
continue;
}
if ((current_tag & HASH_MASK) == partial_hash) {
if (entry->key == key) {
// 命中！原子地设置USED_BIT
if (!(current_tag & USED_BIT)) {
entry->tag.fetch_or(USED_BIT, std::memory_order_relaxed);
}
return entry->value;
}
}
break;
}
index = probe_next(index);
}
return std::nullopt;
}
};
} // namespace lfc