|
雖然 std::simd 還沒有合入 C++26（屬于 Parallelism TS 2），但是這個庫的封裝確實合我心意，所以本文先用來當作 SIMD 練手工具了。需要聲明一下我不會向量算法，就隨便寫寫。
查找最值#include <algorithm>
#include <array>
#include <print>
#include <ranges>
#include <random>
#include <chrono>
#include <tuple>
#include <cassert>
#include <experimental/simd>
#include <immintrin.h>
namespace stdx = std::experimental;
namespace stdv = std::views;
namespace stdr = std::ranges;
template <typename T, auto N>
constexpr auto simd_max_element(const std::array<T, N> &arr) {
// std::native_simd 類型會根據 -march 挑選具體實現（native ABI）
//
// 注意 std::simd 類型默認使用不同的 ABI（不會跟隨 -march 調整最佳實現）
// 雖然 std::simd 同樣是 std::native_simd 的別名
// 但是 std::simd_abi 的挑選準則是 compatible ABI
using simd_t = stdx::native_simd<T>;
// SIMD 向量的構造函數有很多。如果使用標量 T 初始化，那會使得向量內所有的標量都廣播為相同數值
simd_t max_value = std::numeric_limits<T>::min();
constexpr auto step = simd_t::size();
constexpr auto tile = N / step;
constexpr auto left = N % step;
for(auto &batch : arr | stdv::stride(step) | stdv::take(tile)) {
// 另一種構造函數，從地址載入連續多個標量的數值
// 對齊方式也可以按照向量對齊
simd_t temp(std::addressof(batch), stdx::element_aligned);
// select 操作：where 表達式會根據 std::simd_mask（由 operator<(...) 生成）處理賦值操作符
where(max_value < temp, max_value) = temp;
}
if constexpr (left) {
auto left_view = arr | stdv::drop(tile * step);
auto v = *stdr::max_element(left_view);
if(stdx::none_of(max_value > v)) {
return v;
}
}
// SIMD 類型不提供迭代器，并且 operator[] 實際返回了代理類型
// 因此想要 ranged-based `for` loop 是不討好的
//
// 除非你實在想要，可以嘗試：
// stdv:: iota(0) | take(size()) | transform([](index) -> T { []... })
//
// 但是 SIMD 庫已經提供了 std::reduce() 重載以及一些便利函數
// 比如這里，可以 std::hmax() 返回向量當中的最大值（horizontal max）
return stdx::hmax(max_value);
}
constexpr std::size_t MASS = 1e8+3;
std::array<int, MASS> massive;
// 算法無關的函數
auto initiate(auto &arr);
auto tick(auto f);
int main() {
auto check = initiate(massive);
auto [v, elapsed] = tick([&] { return simd_max_element(massive); });
// auto [v, elapsed] = tick([&] { return *stdr::max_element(massive); });
assert(v == check);
std::println("max value: {}, elapsed: {}", v, elapsed);
}
算法無關的代碼部分
// Generate random values and flush cachelines.
// Return a maximum value for assert().
auto initiate(auto &arr) {
auto verifier = *begin(arr);
std::default_random_engine generator;
std::uniform_int_distribution<int> distribution(0, arr.size());
for(auto &v : arr) {
v = distribution(generator);
verifier = std::max(v, verifier);
}
constexpr auto cacheline_elements = 64 / sizeof(arr[0]);
auto flush_view = stdv::iota(arr.data())
| stdv::stride(cacheline_elements)
| stdv::take(arr.size() / cacheline_elements);
for(auto addr : flush_view) {
_mm_clflush(addr);
}
return verifier;
}
auto tick(auto f) {
namespace stdc = std::chrono;
auto start = stdc::steady_clock::now();
auto v = f();
auto end = stdc::steady_clock::now();
auto elapsed = stdc::duration_cast<stdc::milliseconds>(end - start);
return std::tuple(v, elapsed);
}
使用 clang++-18 -O3 -march=skylake -std=c++2b 編譯：
- SIMD 版本耗時 20 ms。
- 標準庫版本耗時 80 ms。
NOTES:
- 我的機子（按摩殿·禪叁）不支持 AVX512，所以這里跑分實際使用的是 AVX2。
- GCC 相當聰明，標準庫實現和 SIMD 實現均為 20 ms。
雙調排序#include <algorithm>
#include <array>
#include <print>
#include <ranges>
#include <random>
#include <chrono>
#include <tuple>
#include <cassert>
#include <thread>
#include <vector>
#include <experimental/simd>
#include <immintrin.h>
namespace stdx = std::experimental;
namespace stdv = std::views;
namespace stdr = std::ranges;
namespace bitonic {
template <typename> struct use_simd {};
//////////////////////////////////////////////////////////// parallel
// A rough tuning option for amortizing thread overheads.
constexpr std::size_t default_scale_hint = 1 << 20;
// We don't have a good thread pool here, so let it be false.
constexpr bool enable_parallel_for = false;
constexpr bool enable_parallel(std::integral auto hint) {
return hint >= default_scale_hint;
}
// Note: parallel(...) and parallel(...) are not parallel!
void parallel(std::size_t scale_hint, std::invocable auto job,
std::invocable auto ...jobs) {
if(!enable_parallel(scale_hint)) return job(), (jobs(), ...);
std::array parallel_jobs {std::jthread{std::move(jobs)}...};
job();
}
void parallel_for(auto view, auto func) {
auto f = [func](auto maybe_subview) {
for(auto &&v : maybe_subview) {
func(std::forward<decltype(v)>(v));
}
};
if(!enable_parallel_for || !enable_parallel(std::size(view))) {
return f(view);
}
auto per_thread_view = view | stdv::chunk(default_scale_hint);
std::vector<std::jthread> parallel_jobs;
for(auto g : per_thread_view) {
parallel_jobs.emplace_back(f, g);
}
}
template <typename simd_t>
auto parallel_for(use_simd<simd_t>, auto view, auto func) {
constexpr auto step = simd_t::size();
const auto tile = std::size(view) / step;
const auto consumed = step * tile;
auto simdify = stdv::stride(step) | stdv::take(tile);
parallel_for(view | simdify, std::move(func));
return consumed;
}
//////////////////////////////////////////////////////////// parallel (end)
//////////////////////////////////////////////////////////// misc
constexpr struct {
std::less<> incr;
std::greater<> decr;
} direction;
template <typename dir>
concept directional =
std::is_same_v<dir, decltype(direction.incr)> ||
std::is_same_v<dir, decltype(direction.decr)>;
void merge(auto range, directional auto dir);
void sort(auto range, directional auto dir);
auto cut(auto range) {
auto pivot = std::begin(range) + std::size(range) / 2;
auto lo = stdr::subrange(std::begin(range), pivot);
auto hi = stdr::subrange(pivot, std::end(range));
return std::tuple(lo, hi);
}
template <template<typename...> typename Tuple, typename T>
void perform(Tuple<T&, T&> bitonic_pair, directional auto dir) {
auto &[lhs, rhs] = bitonic_pair;
if(!dir(lhs, rhs)) {
std::swap(lhs, rhs);
}
}
template <typename simd_t, template<typename...> typename Tuple, typename T>
void perform(use_simd<simd_t>, Tuple<T&, T&> bitonic_pair, directional auto dir) {
auto &[lhs, rhs] = bitonic_pair;
simd_t x(std::addressof(lhs), stdx::element_aligned);
simd_t y(std::addressof(rhs), stdx::element_aligned);
// SIMD 庫提供了非常方便的 minmax 操作
// 返回 std::pair<> 類型的 [min, max] 向量
if constexpr (dir(0, 1)) std::tie(x, y) = stdx::minmax(x, y);
else std::tie(y, x) = stdx::minmax(x, y);
x.copy_to(std::addressof(lhs), stdx::element_aligned);
y.copy_to(std::addressof(rhs), stdx::element_aligned);
}
//////////////////////////////////////////////////////////// misc (end)
//////////////////////////////////////////////////////////// core
void sort(auto range, directional auto dir) {
if(std::size(range) < 2) return;
auto [lo, hi] = cut(range);
parallel(std::size(range),
[=]{ sort(lo, direction.incr); },
[=]{ sort(hi, direction.decr); });
merge(range, dir);
}
void merge(auto range, directional auto dir) {
if(std::size(range) < 2) return;
using T = stdr::range_value_t<decltype(range)>;
using simd_t = stdx::native_simd<T>;
auto [lo, hi] = cut(range);
auto zip_view = stdv::zip(lo, hi);
// All the comparisons can be done in parallel.
auto consumed = parallel_for(use_simd<simd_t>(), zip_view,
[=](auto tup) { perform(use_simd<simd_t>(), tup, dir); });
stdr::for_each(zip_view | stdv::drop(consumed),
[=](auto tup) { perform(tup, dir); });
parallel(std::size(range),
[=]{ merge(lo, dir); },
[=]{ merge(hi, dir); });
}
} // namespace bitonic
void simd_sort(stdr::random_access_range auto &&range) {
if(std::size(range) < 2) return;
assert(std::has_single_bit(std::size(range)));
bitonic::sort(range | stdv::all, bitonic::direction.incr);
}
//////////////////////////////////////////////////////////// core (end)
constexpr std::size_t MASS = 1 << 26;
std::array<int, MASS> massive;
// 算法無關的函數
void initiate(auto &arr);
auto tick(auto f);
int main() {
initiate(massive);
// bitonic sort 要求容器大小為 2 的冪
static_assert(std::has_single_bit(std::size(massive)));
auto elapsed = tick([&]{ simd_sort(massive); });
// auto elapsed = tick([&] { stdr::sort(massive); });
assert(std::ranges::is_sorted(massive));
std::println("time: {}", elapsed);
}
算法無關的代碼部分
// Generate random values and flush cachelines.
void initiate(auto &arr) {
std::default_random_engine generator;
std::uniform_int_distribution<int> distribution(0, arr.size());
for(auto &v : arr) {
v = distribution(generator);
}
constexpr auto cacheline_elements = 64 / sizeof(arr[0]);
auto flush_view = stdv::iota(arr.data())
| stdv::stride(cacheline_elements)
| stdv::take(arr.size() / cacheline_elements);
for(auto addr : flush_view) {
_mm_clflush(addr);
}
}
auto tick(auto f) {
namespace stdc = std::chrono;
auto start = stdc::steady_clock::now();
f();
auto end = stdc::steady_clock::now();
auto elapsed = stdc::duration_cast<stdc::milliseconds>(end - start);
return elapsed;
}
使用 clang++-18 -O3 -march=skylake -std=c++2b 編譯：
NOTES:
- 性能增益實際來自于雙調排序的并行性（可以多線程排序），事實上不開啟 SIMD 還能更快（2600 ms）。你也可以在這里獲取非 SIMD 版本的并行雙調排序算法實現。
- GCC 的表現在這里很丟人：SIMD 版本耗時 3500 ms+。
單詞統計std::string_view frost = "Whose woods these are I think I know.\n"
"His house is in the village though; \n"
"He will not see me stopping here \n"
"To watch his woods fill up with snow.\n";
std::size_t std_nvda_word_count(std::string_view s) {
if(!std::size(s)) return 0;
auto isspace = [](auto c) { return c == ' ' || c == '\n'; };
auto transform = [=](auto l, auto r) { return isspace(l) && !isspace(r); };
return std::transform_reduce(std::begin(s), std::end(s)-1, std::begin(s)+1,
std::size_t(!isspace(s.front())),
std::plus<>(),
transform);
}
上面是 NV 給出來的標準庫非 SIMD 實現版本，看起來還挺 NV 的。后面會用來做對比。 1 = space, 0 = letter.
E = End of a word.
E E
11000011000001000... bit
00111100111110111 (~)
00011110011111011 <<1
^ ^ bit & (~bit)<<1
count = 2
個人想到的是這種位運算思路，應該是哪里見過的技巧。不過由于 std::simd_mask 不支持位移（srli/slli）操作，因此不能高效實現上面的算法。其實從作者的提案實現也可看出這種問題（緊急畫餅中）。既然有提案實現這種先行版，那就先強上了。 #include <algorithm>
#include <print>
#include <ranges>
#include <random>
#include <chrono>
#include <cassert>
#include <string>
#include <iostream>
#include <bitset>
// 提案作者給出的先行實現
#include <vir/simd.h>
#include <vir/simd_bit.h>
#include <vir/simd_bitset.h>
// 大部分用法和 TS 版 std::simd 差不多
// 我們這里只用標準庫未提供的 std::simd_mask to std::bitset 功能
namespace stdx = vir::stdx;
namespace stdv = std::views;
namespace stdr = std::ranges;
std::string_view frost = "Whose woods these are I think I know.\n"
"His house is in the village though; \n"
"He will not see me stopping here \n"
"To watch his woods fill up with snow.\n";
namespace detail {
auto isspace(auto /* char or simd_chars */ c) {
return c == ' ' || c == '\n';
}
std::size_t std_nvda_word_count(stdr::view auto s) {
if(!std::size(s)) return 0;
return std::transform_reduce(std::begin(s), std::prev(std::end(s)),
std::next(std::begin(s)),
std::size_t(!isspace(s.front())),
std::plus<>(),
[](auto l, auto r) {
return isspace(l) && !isspace(r);
});
}
std::size_t simd_word_count(stdr::view auto s) {
using simd_t = stdx::native_simd<char>;
constexpr auto step = simd_t::size();
const auto tile = std::size(s) / step;
auto simdify = stdv::stride(step) | stdv::take(tile);
std::size_t count = 0;
//////////////////////////////////////////////////
auto simd_view = s | simdify;
auto simd_transform = [&](auto &to_simd) {
simd_t temp (std::addressof(to_simd), stdx::element_aligned);
auto mask = isspace(temp);
// 勉強當作 mask 之間的直接位操作使用
auto bits = vir::to_bitset(mask);
return (bits & (~(bits) << 1)).count();
};
for(auto &&to_simd : simd_view) {
count += simd_transform(to_simd);
}
//////////////////////////////////////////////////
auto x = s | simdify | stdv::drop(1);
auto y = s | stdv::drop(step - 1) | simdify;
auto z = stdv::zip(x, y);
for(auto [curr, prev] : z) {
count += isspace(curr) && !isspace(prev);
}
//////////////////////////////////////////////////
if(auto batch = step * tile; std::size(s) > batch) {
auto left_view = s | stdv::drop(batch);
auto prev_view = s | stdv::drop(batch - 1);
count += std_nvda_word_count(left_view);
count += isspace(left_view[0]) && !isspace(prev_view[0]);
}
return count;
}
} // namespace detail
template <typename T>
concept string_class = std::is_convertible_v<T, std::string_view>;
std::size_t std_nvda_word_count(string_class auto &&str) {
return detail::std_nvda_word_count(str | stdv::all);
}
std::size_t simd_word_count(string_class auto &&str) {
return detail::simd_word_count(str | stdv::all);
}
auto tick(auto f) {
namespace stdc = std::chrono;
auto start = stdc::steady_clock::now();
auto v = f();
auto end = stdc::steady_clock::now();
auto elapsed = stdc::duration_cast<stdc::milliseconds>(end - start);
return std::tuple(v, elapsed);
}
int main() {
std::string text;
for(std::string line; std::getline(std::cin, line);) {
if(!text.empty()) text += ' ';
text += line;
}
auto [count, elapsed] = tick([&] { return simd_word_count(text); });
// auto [count, elapsed] = tick([&] { return std_nvda_word_count(text); });
std::println("count: {}, time: {}", count, elapsed);
}
帶權重的隨機文本生成器
#include <iostream>
#include <vector>
#include <tuple>
#include <algorithm>
#include <cassert>
#include <cstddef>
#include <random>
#include <limits>
#include <cmath>
#include <numeric>
#include <bitset>
#include <cstdint>
#include <utility>
#include <cassert>
#include <ranges>
// 很久以前寫的 alias method 算法，細節我也差不多忘了 orz
// 成員的命名取自維基百科的 alias method 算法描述
class Alias {
public:
Alias(std::vector<double> probabilities) {
precondition(probabilities);
std::tie(_probs, _alias) = make_alias_table(std::move(probabilities));
}
/// @brief Just simply make a random O(1) choice.
/// @return Index.
std::size_t make() {
std::uniform_real_distribution dis01 {0.0, 1.0};
double x = dis01(_gen);
std::size_t i = _alias.size() * x; // round down
double p = _alias.size() * x - i;
return p < _probs[i] ? i : _alias[i];
}
/// @deprecated
/// __Simple method__ to make a random O(1) choice.
/// Use make() instead, which is faster (~10%) but harder to write correctly.
std::size_t make(decltype(std::ignore)...) {
std::uniform_int_distribution dis0N {0ul, _alias.size() - 1};
std::uniform_real_distribution dis01 {0.0, 1.0};
std::size_t i = dis0N(_gen);
double p = dis01(_gen);
return p < _probs[i] ? i : _alias[i];
}
private:
// O(n) construction.
static
auto make_alias_table(std::vector<double> probabilities)
-> std::tuple< std::vector<double>, std::vector<std::size_t> >
{
const auto N = probabilities.size();
for(auto &p : probabilities) p *= N;
// K: alias table.
// U: {index, probability} table.
std::vector<std::size_t> K(N, N /*uninitialized*/);
enum u_type {OVERFULL, FULL, UNDERFULL, UTYPE_MAX};
std::vector<std::tuple<std::size_t, double>> U[UTYPE_MAX];
auto get_type = [&](std::size_t i, double p) {
u_type type = p > 1 ? OVERFULL : UNDERFULL;
if(is_one(p)) [[unlikely]] {
type = FULL;
// Optional, but make less buggy code.
// NOTE: FULL or OVERFULL->FULL actually has no alias index.
K[i] = i;
}
return type;
};
for(std::size_t i = 0; i < probabilities.size(); ++i) {
auto p = probabilities[i];
u_type type = get_type(i, p);
U[type].emplace_back(i, p);
}
while(!U[OVERFULL].empty() && !U[UNDERFULL].empty()) {
/// Calculate.
auto [over_i, over_p] = pop(U[OVERFULL]);
auto [under_i, under_p] = pop(U[UNDERFULL]);
over_p -= (1.0 - under_p);
K[under_i] = over_i;
/// Reinsert.
U[FULL].emplace_back(under_i, under_p);
u_type type = get_type(over_i, over_p);
U[type].emplace_back(over_i, over_p);
}
/// I hate floating points.
auto corner_case = [&](auto &ulist) {
while(!ulist.empty()) {
auto [i, p] = pop(ulist);
assert(is_one(p));
K[i] = i;
U[FULL].emplace_back(i, p);
}
};
corner_case(U[OVERFULL]);
corner_case(U[UNDERFULL]);
// Now they are all FULL.
std::vector<double> full_u(N);
for(auto &&[i, p] : U[FULL]) full_u[i] = p;
return std::make_tuple(full_u, K);
}
// 1. size() > 0
// 2. \sum probabilities == 1.0
static
void precondition(const std::vector<double> &probabilities) noexcept {
assert(!probabilities.empty());
[[maybe_unused]] double p_sum = 0;
for(auto p : probabilities) p_sum += p;
assert(is_one(p_sum));
}
static
bool is_one(double x) noexcept {
constexpr auto EPS = std::numeric_limits<double>::epsilon();
return fabs(x - 1) < EPS;
}
template <typename T> static
T pop(std::vector<T> &ulist) {
auto elem = std::move(ulist.back());
ulist.pop_back();
return elem;
}
private:
std::random_device _rd;
std::mt19937 _gen{_rd()};
std::vector<double> _probs;
std::vector<std::size_t> _alias;
};
int main() {
std::vector letters_length {10, 20, 30, 40};
std::vector letters_probability {.1, .2, .3, .4};
std::vector spaces_length { 1, 2, 4, 8};
std::vector spaces_probability {.1, .2, .3, .4};
std::vector space_categories_type {' ', '\n'};
std::vector space_categories_probability {0.8, 0.2};
constexpr std::size_t total_length = 1e9;
Alias letters_alias(letters_probability);
Alias spaces_alias(spaces_probability);
Alias spaces_categories_alias(space_categories_probability);
std::string text;
std::size_t answer = 0;
// For random text.
std::default_random_engine generator;
std::uniform_int_distribution<char> distribution('a', 'z');
for(std::size_t length = 0; length < total_length;) {
auto letters_index = letters_alias.make();
namespace stdv = std::views;
for(auto _ : stdv::iota(0, letters_length[letters_index])) {
text += distribution(generator);
}
answer++;
auto spaces_index = spaces_alias.make();
auto space_categories_index = spaces_categories_alias.make();
for(auto _ : stdv::iota(0, spaces_length[spaces_index])) {
text += space_categories_type[space_categories_index];
}
length += letters_length[letters_index]
+ spaces_length[spaces_index];
}
// 使用管道時 std::cerr 仍輸出到終端，用于對比答案
std::cerr << answer << std::endl;
std::cout << text << std::endl;
}
使用 g++-14 -O3 -march=skylake -std=c++2b 編譯，以生成器提供的 1e9 文本測試：
- SIMD 實現耗時 149 ms。
- 標準庫實現耗時 776 ms。
NOTE：Clang 在這里的表現特別丟人，SIMD 居然要 2000 ms+。
瑞士甜表#include <array>
#include <ranges>
#include <tuple>
#include <algorithm>
#include <utility>
#include <cassert>
#include <chrono>
#include <concepts>
#include <type_traits>
#include <random>
#include <print>
#include <vir/simd.h>
#include <vir/simd_bit.h>
#include <vir/simd_bitset.h>
namespace stdx = vir::stdx;;
namespace stdr = std::ranges;
namespace stdv = std::views;
using byte = signed char;
using xbyte = stdx::native_simd<byte>;
// Special flags MUST start with 0b1_______.
constexpr byte empty_flag = 0b10000000;
constexpr byte deleted_flag = 0b11000000;
constexpr auto xwidth = xbyte::size();
template <std::size_t N>
concept swiss_like = N > 0 && N % xwidth == 0;
template <typename K, typename V>
concept simplified_toy = std::is_trivially_copyable_v<K>
&& std::is_trivially_copyable_v<V>;
// An extreme toy implementation of Swiss table.
// This version simplifies almost everything but still utilize the SIMD techniques.
//
// {K, V, N} = Key, Value, Capacity.
template <typename K, typename V, std::size_t N>
requires swiss_like<N> && simplified_toy<K, V>
struct Sweet_table: protected std::tuple<std::hash<K>, std::equal_to<K>> {
using T = std::pair<K, V>;
using H = std::hash<K>;
using E = std::equal_to<K>;
using Base = std::tuple<H, E>;
Sweet_table() { for(auto &cb : ctrls) stdr::fill(cb, empty_flag); }
auto hash(const auto &key) {
return std::get<0>(static_cast<Base&>(*this))(key);
}
auto equal(const auto &key1, const auto &key2) {
return std::get<1>(static_cast<Base&>(*this))(key1, key2);
}
////////////////////////////////////////////////////////////
using g_ctrl_t = std::array<byte, xwidth>;
using g_slot_t = std::array<T, xwidth>;
inline static constexpr
std::size_t group_size() { return N / xwidth; }
struct Group {
auto match(byte h2) {
return vir::to_bitset(temp == h2);
}
auto empty() {
// std::simd 做不到 absl 庫的 _mm_movemask_epi8(_mm_sign_epi8(ctrl, ctrl)) 操作
// 提案實現沒有提供對應的 intrinsics 指令映射
return vir::to_bitset(temp == empty_flag);
}
auto empty_deleted() {
static_assert(empty_flag <= deleted_flag);
return vir::to_bitset(temp <= deleted_flag);
}
g_ctrl_t &ctrls;
g_slot_t &slots;
xbyte temp{ctrls.data(), stdx::vector_aligned};
};
auto group_view(auto offset) {
return stdv::iota(offset / xwidth)
| stdv::transform([this](auto i) { return i % group_size(); })
| stdv::transform([this](auto i) { return Group{ctrls[i], slots[i]}; })
| stdv::take(group_size());
}
////////////////////////////////////////////////////////////
struct seq {
int operator*() { return std::countr_zero(x); }
seq operator++(auto...) { return {x &= x-1}; }
seq begin() { return {x}; }
seq end() { return {0}; }
auto operator<=>(const seq &) const=default;
unsigned long long x;
};
template <auto X>
auto bitseq(const std::bitset<X> &bits) { return seq{bits.to_ullong()}; }
auto salt() { return std::bit_cast<std::ptrdiff_t>(ctrls.data()) >> 12; }
auto H1(std::size_t hashed) { return hashed ^ salt(); }
auto H2(std::size_t hashed) { return hashed & 0x7f; }
////////////////////////////////////////////////////////////
// Don't blame me, this name was taken from std::map.
bool insert_or_assign(auto &&key, auto &&...args) {
auto hashed = hash(key);
auto h1 = H1(hashed);
auto h2 = H2(hashed);
for(auto group : group_view(h1 % N)) {
for(auto index : bitseq(group.match(h2))) {
auto &&[old_key, old_value] = group.slots[index];
if(equal(old_key, key)) {
old_value = V{std::forward<decltype(args)>(args)...};
return true;
}
}
for(auto index : bitseq(group.empty_deleted())) {
group.ctrls[index] = h2;
group.slots[index] = T(K{std::forward<decltype(key)>(key)},
V{std::forward<decltype(args)>(args)...});
return false;
}
}
// This is a fixed-size table.
// Otherwise you need to resize().
std::unreachable();
}
std::optional<T> find(const auto &key) {
auto hashed = hash(key);
auto h1 = H1(hashed);
auto h2 = H2(hashed);
for(auto group : group_view(h1 % N)) {
for(auto index : bitseq(group.match(h2))) {
auto &&[old_key, _] = group.slots[index];
if(equal(old_key, key)) {
return group.slots.begin()[index];
}
}
if(group.empty().any()) break;
}
return {};
}
// erase() is simple, just find then mark them DELETED.
alignas(64) std::array<g_ctrl_t, group_size()> ctrls;
alignas(64) std::array<g_slot_t, group_size()> slots;
};
// 算法無關的函數
auto initiate(auto &first, auto &next);
auto tick(auto f, auto &map);
constexpr std::size_t MAXN = 1 << 25;
constexpr std::size_t WORKLOAD = MAXN / 2;
auto first_input_data = std::array<int, WORKLOAD>{};
auto next_input_data = std::array<std::tuple<int /* old */, int /* new */>,
std::max(8zu, WORKLOAD >> 12)>{};
auto sweet_table = Sweet_table<int, int, MAXN>{};
auto std_umap = std::unordered_map<int, int>(MAXN);
int main() {
auto check = initiate(first_input_data, next_input_data);
auto f = [](auto &map) {
std::int64_t x = 0;
for(auto i : first_input_data) {
map.insert_or_assign(i, i);
}
for(auto i : first_input_data) {
auto [k, v] = *map.find(i);
x += v;
}
for(auto [o, n] : next_input_data) {
map.insert_or_assign(o, n);
auto [k, v] = *map.find(o);
x += v;
}
return x;
};
auto [x, e] = tick(f, sweet_table);
// auto [x, e] = tick(f, std_umap);
assert(x == check);
std::println("{}", e);
}
算法無關的代碼部分
auto initiate(auto &first, auto &next) {
std::int64_t result = 0;
std::default_random_engine generator;
std::uniform_int_distribution<int> distribution(0, std::numeric_limits<int>::max() / 64);
for(auto &v : first) {
v = distribution(generator);
result += v;
}
for(auto &&[old_value, overwritten] : stdv::zip(first, next)) {
auto new_value = distribution(generator);
overwritten = std::tuple(old_value, new_value);
result += new_value;
}
auto flush_view = [=](auto &arr) {
constexpr auto cacheline_elements = 64 / sizeof(arr[0]);
return stdv::iota(arr.data())
| stdv::stride(cacheline_elements)
| stdv::take(arr.size() / cacheline_elements);
};
for(auto addr : flush_view(first)) _mm_clflush(addr);
for(auto addr : flush_view(next)) _mm_clflush(addr);
return result;
}
auto tick(auto f, auto &map) {
namespace stdc = std::chrono;
auto start = stdc::steady_clock::now();
auto v = f(map);
auto end = stdc::steady_clock::now();
auto elapsed = stdc::duration_cast<stdc::milliseconds>(end - start);
return std::tuple(v, elapsed);
}
之前看 V 站有人做了一個針對小數據的 Swiss table。我也挺感興趣的，順手搞一個山寨版（設計偷懶 + 功能殘缺，所以叫 Sweet table）。使用方面基本算是故技重施了。其實這里 std::simd 是非常低效的，它做不到一些 signbit 操作，并且 perf 發現 SIMD 和 std::bitset 操作相當的熱（不排除我寫得差勁），除了跑得比標準庫快以外就沒啥優勢了。
使用 g++-14 -O3 -march=skylake -std=c++2b 編譯：
std::unordered_map 實現耗時：3866 ms。
sweet_table + std::simd 實現耗時：1725 ms。
TBC
雖說本文只關注 std::simd 的使用，但是目前就連 API 都是處于缺失狀態，感覺也做不了什么事情。后續不定期更新。
|
