Swiss Table：Google 的 SIMD 加速哈希表

// std::unordered_map 的内部结构（简化）
struct Node {
    Node* next;
    size_t hash_cache;   // 缓存的哈希值
    pair<Key, Value> kv; // 键值对
};

struct Bucket {
    Node* head; // 指向链表头
};

// 查找过程：
// 1. hash(key) -> bucket_index     （计算）
// 2. buckets[bucket_index].head    （一次内存访问）
// 3. node->next, node->next, ...   （每个节点一次内存访问）
// 4. 每个节点还要比较 key           （又一次内存访问）

// 控制字节的取值
#define CTRL_EMPTY    ((int8_t)0xFF)  // -1:   槽位为空
#define CTRL_DELETED  ((int8_t)0x80)  // -128: 槽位已删除（tombstone）
#define CTRL_SENTINEL ((int8_t)0xFE)  // -2:   哨兵（表尾标记）
// 0x00 ~ 0x7F: 槽位已占用，值为哈希值的高 7 位（H2）

// 假设哈希值为 64 位
static inline size_t H1(size_t hash) {
    return hash >> 7;  // 高 57 位：用于定位 group
}

static inline int8_t H2(size_t hash) {
    return hash & 0x7F;  // 低 7 位：存入控制字节
}

#include <immintrin.h>

typedef struct {
    __m128i match;  // SSE2 128-bit 寄存器
} GroupMatch;

// 在一个 group（16 个控制字节）中查找与 h2 匹配的位置
static inline uint32_t group_match(const int8_t* ctrl, int8_t h2) {
    // 1. 将 h2 广播到 128-bit 寄存器的所有 16 个字节
    __m128i needle = _mm_set1_epi8(h2);

    // 2. 加载 16 个控制字节到另一个 128-bit 寄存器
    __m128i haystack = _mm_loadu_si128((const __m128i*)ctrl);

    // 3. 逐字节比较：相等的位置设为 0xFF，不等的设为 0x00
    __m128i cmp = _mm_cmpeq_epi8(needle, haystack);

    // 4. 提取每个字节的最高位，压缩成 16-bit 掩码
    return (uint32_t)_mm_movemask_epi8(cmp);
}

// 查找空槽位（EMPTY 或 DELETED）
static inline uint32_t group_match_empty_or_deleted(const int8_t* ctrl) {
    // 特殊状态的最高位都是 1，已占用的最高位是 0
    // _mm_movemask_epi8 直接提取最高位
    __m128i v = _mm_loadu_si128((const __m128i*)ctrl);
    return (uint32_t)_mm_movemask_epi8(v);
}

// 查找空槽位（只匹配 EMPTY，不匹配 DELETED）
static inline uint32_t group_match_empty(const int8_t* ctrl) {
    // EMPTY = 0xFF = -1，DELETED = 0x80 = -128
    // 技巧：将每个字节视为有符号数，与 EMPTY(-1) 比较
    __m128i v = _mm_loadu_si128((const __m128i*)ctrl);
    __m128i special = _mm_set1_epi8((int8_t)CTRL_EMPTY);
    return (uint32_t)_mm_movemask_epi8(_mm_cmpeq_epi8(v, special));
}

// 查找 key 的过程（伪代码）
void* swiss_table_find(SwissTable* t, Key key) {
    size_t hash = hash_function(key);
    int8_t h2 = H2(hash);                 // 低 7 位
    size_t group_index = H1(hash) % t->num_groups;

    while (1) {
        int8_t* ctrl = t->ctrl + group_index * 16;

        // SIMD: 一条指令比较 16 个控制字节
        uint32_t match = group_match(ctrl, h2);

        // 遍历所有匹配的候选位置
        while (match != 0) {
            int pos = __builtin_ctz(match);  // 最低位的 1 在哪里
            size_t slot = group_index * 16 + pos;

            // 只有在这里才做完整的 key 比较
            if (key_equal(t->slots[slot].key, key)) {
                return &t->slots[slot].value;
            }
            match &= match - 1;  // 清除最低位的 1
        }

        // 如果这个 group 里有 EMPTY 槽位，说明 key 不存在
        uint32_t empty = group_match_empty(ctrl);
        if (empty != 0) {
            return NULL;  // 未找到
        }

        // 否则继续探测下一个 group
        group_index = (group_index + 1) % t->num_groups;  // 简化版
    }
}

// 探测序列：group 0, group 0+1, group 0+1+2, group 0+1+2+3, ...
// 即增量为 0, 1, 3, 6, 10, 15, 21, ...（三角数）
size_t probe_seq(size_t start, size_t step, size_t num_groups) {
    // step 从 0 开始递增
    // 偏移量 = step * (step + 1) / 2
    return (start + step * (step + 1) / 2) % num_groups;
}

// Abseil 中的探测序列实现
class probe_seq {
public:
    probe_seq(size_t hash, size_t mask) {
        pos_ = hash & mask;    // 起始位置
        stride_ = 0;           // 初始步长
    }

    size_t offset() const { return pos_; }

    void next() {
        stride_ += kGroupWidth;     // kGroupWidth = 16
        pos_ = (pos_ + stride_) & mask_;  // 三角探测
    }

private:
    size_t pos_;
    size_t stride_;
    size_t mask_;
};

#include <immintrin.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include <stdbool.h>

// 控制字节常量
#define CTRL_EMPTY    ((int8_t)0xFF)
#define CTRL_DELETED  ((int8_t)0x80)
#define GROUP_WIDTH   16

// 哈希分离
static inline size_t H1(size_t hash)  { return hash >> 7; }
static inline int8_t H2(size_t hash)  { return (int8_t)(hash & 0x7F); }

// 键值对
typedef struct {
    uint64_t key;
    uint64_t value;
} Slot;

// Swiss Table 结构
typedef struct {
    int8_t*  ctrl;          // 控制字节数组（长度 = capacity + GROUP_WIDTH）
    Slot*    slots;         // 键值对数组（长度 = capacity）
    size_t   capacity;      // 总槽位数（必须是 GROUP_WIDTH 的倍数）
    size_t   size;          // 已存储元素数
    size_t   growth_left;   // 还能插入多少个元素
} SwissTable;

// ===== SIMD 匹配函数 =====

// 在 group 中查找与 h2 匹配的槽位，返回 16-bit 掩码
static inline uint32_t match_h2(const int8_t* ctrl, int8_t h2) {
    __m128i group = _mm_loadu_si128((const __m128i*)ctrl);
    __m128i target = _mm_set1_epi8(h2);
    __m128i cmp = _mm_cmpeq_epi8(group, target);
    return (uint32_t)_mm_movemask_epi8(cmp) & 0xFFFF;
}

// 查找空或已删除的槽位
static inline uint32_t match_empty_or_deleted(const int8_t* ctrl) {
    __m128i group = _mm_loadu_si128((const __m128i*)ctrl);
    return (uint32_t)_mm_movemask_epi8(group) & 0xFFFF;
}

// 查找空槽位
static inline uint32_t match_empty(const int8_t* ctrl) {
    __m128i group = _mm_loadu_si128((const __m128i*)ctrl);
    __m128i empty = _mm_set1_epi8(CTRL_EMPTY);
    __m128i cmp = _mm_cmpeq_epi8(group, empty);
    return (uint32_t)_mm_movemask_epi8(cmp) & 0xFFFF;
}

// ===== 探测序列 =====

typedef struct {
    size_t pos;
    size_t stride;
    size_t mask;  // capacity - 1
} ProbeSeq;

static inline ProbeSeq probe_seq_start(size_t h1, size_t mask) {
    ProbeSeq seq;
    seq.pos = h1 & mask;
    seq.stride = 0;
    seq.mask = mask;
    return seq;
}

static inline void probe_seq_next(ProbeSeq* seq) {
    seq->stride += GROUP_WIDTH;
    seq->pos = (seq->pos + seq->stride) & seq->mask;
}

// ===== 简单哈希函数（生产环境请使用 wyhash 等） =====

static inline size_t hash_u64(uint64_t key) {
    key ^= key >> 33;
    key *= 0xFF51AFD7ED558CCDULL;
    key ^= key >> 33;
    key *= 0xC4CEB9FE1A85EC53ULL;
    key ^= key >> 33;
    return key;
}

// ===== 控制字节的镜像（cloned bytes） =====

// Swiss Table 在 ctrl 数组末尾额外复制了前 GROUP_WIDTH 个字节
// 这样探测序列在表尾 wrap around 时不需要特殊处理
static inline void set_ctrl(SwissTable* t, size_t index, int8_t value) {
    t->ctrl[index] = value;
    // 镜像：如果 index < GROUP_WIDTH，在末尾也写一份
    if (index < GROUP_WIDTH) {
        t->ctrl[t->capacity + index] = value;
    }
}

// ===== 创建和销毁 =====

SwissTable* swisstable_new(size_t min_capacity) {
    // 向上取到 GROUP_WIDTH 的倍数，且至少为 GROUP_WIDTH
    size_t capacity = GROUP_WIDTH;
    while (capacity < min_capacity) {
        capacity <<= 1;
    }

    SwissTable* t = (SwissTable*)malloc(sizeof(SwissTable));
    // ctrl 多分配 GROUP_WIDTH 字节用于镜像
    t->ctrl = (int8_t*)aligned_alloc(16, capacity + GROUP_WIDTH);
    t->slots = (Slot*)calloc(capacity, sizeof(Slot));
    t->capacity = capacity;
    t->size = 0;
    // 负载因子上限约 7/8 = 87.5%
    t->growth_left = capacity - capacity / 8;

    // 初始化所有控制字节为 EMPTY
    memset(t->ctrl, CTRL_EMPTY, capacity + GROUP_WIDTH);

    return t;
}

void swisstable_free(SwissTable* t) {
    if (t) {
        free(t->ctrl);
        free(t->slots);
        free(t);
    }
}

// ===== 查找 =====

Slot* swisstable_find(SwissTable* t, uint64_t key) {
    size_t hash = hash_u64(key);
    int8_t h2 = H2(hash);
    ProbeSeq seq = probe_seq_start(H1(hash), t->capacity - 1);

    while (1) {
        // SIMD 匹配：一条指令比较 16 个控制字节
        uint32_t match = match_h2(t->ctrl + seq.pos, h2);

        while (match != 0) {
            int bit = __builtin_ctz(match);
            size_t idx = (seq.pos + bit) & (t->capacity - 1);
            if (t->slots[idx].key == key) {
                return &t->slots[idx];
            }
            match &= match - 1;  // 清除最低位
        }

        // 如果 group 中有 EMPTY，说明 key 不存在
        if (match_empty(t->ctrl + seq.pos) != 0) {
            return NULL;
        }

        probe_seq_next(&seq);
    }
}

// ===== 内部：找到插入位置 =====

static size_t find_insert_slot(SwissTable* t, size_t hash) {
    ProbeSeq seq = probe_seq_start(H1(hash), t->capacity - 1);

    while (1) {
        uint32_t mask = match_empty_or_deleted(t->ctrl + seq.pos);
        if (mask != 0) {
            int bit = __builtin_ctz(mask);
            return (seq.pos + bit) & (t->capacity - 1);
        }
        probe_seq_next(&seq);
    }
}

// ===== 扩容 =====

static void swisstable_rehash(SwissTable* t) {
    size_t old_capacity = t->capacity;
    int8_t* old_ctrl = t->ctrl;
    Slot* old_slots = t->slots;

    size_t new_capacity = old_capacity * 2;

    t->ctrl = (int8_t*)aligned_alloc(16, new_capacity + GROUP_WIDTH);
    t->slots = (Slot*)calloc(new_capacity, sizeof(Slot));
    t->capacity = new_capacity;
    t->size = 0;
    t->growth_left = new_capacity - new_capacity / 8;

    memset(t->ctrl, CTRL_EMPTY, new_capacity + GROUP_WIDTH);

    // 重新插入所有元素
    for (size_t i = 0; i < old_capacity; i++) {
        if (old_ctrl[i] >= 0) {  // FULL: 最高位为 0
            size_t hash = hash_u64(old_slots[i].key);
            size_t idx = find_insert_slot(t, hash);
            set_ctrl(t, idx, H2(hash));
            t->slots[idx] = old_slots[i];
            t->size++;
            t->growth_left--;
        }
    }

    free(old_ctrl);
    free(old_slots);
}

// ===== 插入 =====

bool swisstable_insert(SwissTable* t, uint64_t key, uint64_t value) {
    // 先查找是否已存在
    Slot* existing = swisstable_find(t, key);
    if (existing != NULL) {
        existing->value = value;  // 更新
        return false;
    }

    // 需要扩容？
    if (t->growth_left == 0) {
        swisstable_rehash(t);
    }

    size_t hash = hash_u64(key);
    size_t idx = find_insert_slot(t, hash);

    // 如果覆盖的是 EMPTY（而非 DELETED），减少 growth_left
    if (t->ctrl[idx] == CTRL_EMPTY) {
        t->growth_left--;
    }

    set_ctrl(t, idx, H2(hash));
    t->slots[idx].key = key;
    t->slots[idx].value = value;
    t->size++;

    return true;
}

// ===== 删除 =====

bool swisstable_erase(SwissTable* t, uint64_t key) {
    Slot* slot = swisstable_find(t, key);
    if (slot == NULL) {
        return false;
    }

    size_t idx = slot - t->slots;

    // 判断是否可以直接标记为 EMPTY（而非 DELETED）
    // 如果当前 group 中有 EMPTY 槽位，说明探测序列在此终止，
    // 可以安全地设为 EMPTY 而不影响其他探测
    uint32_t empty_mask = match_empty(t->ctrl + (idx & ~(GROUP_WIDTH - 1)));
    bool was_never_full = (empty_mask != 0);

    if (was_never_full) {
        set_ctrl(t, idx, CTRL_EMPTY);
        t->growth_left++;
    } else {
        set_ctrl(t, idx, CTRL_DELETED);
        // 注意：DELETED 不增加 growth_left
    }

    t->size--;
    return true;
}

int main(void) {
    SwissTable* table = swisstable_new(64);

    // 插入
    for (uint64_t i = 0; i < 1000; i++) {
        swisstable_insert(table, i, i * 100);
    }

    // 查找
    Slot* s = swisstable_find(table, 42);
    if (s != NULL) {
        printf("found: key=%lu, value=%lu\n", s->key, s->value);
    }

    // 删除
    swisstable_erase(table, 42);

    swisstable_free(table);
    return 0;
}

gcc -O2 -msse2 -o swiss_table swiss_table.c

// growth_left 的含义
// 初始值 = capacity * 7/8
// 插入到 EMPTY 位置 -> growth_left--
// 插入到 DELETED 位置 -> growth_left 不变
// 删除后设为 EMPTY -> growth_left++
// 删除后设为 DELETED -> growth_left 不变

// Abseil 源码中的删除逻辑（简化）
void erase_slot(size_t index) {
    size_t group_start = index & ~(kGroupWidth - 1);
    // 检查当前 group 或相邻 group 是否有 EMPTY
    auto empty_after = Group(ctrl_ + group_start).MatchEmpty();
    auto empty_before = Group(ctrl_ + ((group_start - kGroupWidth) & mask_)).MatchEmpty();

    bool was_never_full = empty_after || empty_before;

    if (was_never_full) {
        SetCtrl(index, kEmpty);
        growth_left_++;
    } else {
        SetCtrl(index, kDeleted);
    }
    size_--;
}

// 取模用位与替代
size_t index = hash & (capacity - 1);  // 等价于 hash % capacity

// 但这要求 capacity 是 2 的幂
// capacity = 16, 32, 64, 128, ...

void maybe_rehash(SwissTable* t) {
    if (t->growth_left > 0) return;

    // 如果实际元素数量 < capacity * 7/16，说明 tombstone 太多
    // 做就地 rehash（不扩容）
    if (t->size <= t->capacity * 7 / 16) {
        rehash_in_place(t);  // 清除 tombstone
    } else {
        rehash_grow(t);      // 容量翻倍
    }
}

// absl/container/internal/raw_hash_set.h（简化）
template <class Policy, class Hash, class Eq, class Alloc>
class raw_hash_set {
    ctrl_t* ctrl_;        // 控制字节数组
    void* slots_;         // 键值对数组（类型擦除）
    size_t size_;         // 已存储元素数
    size_t capacity_;     // 总容量
    size_t growth_left_;  // 剩余增长空间
    // ...
};

// 控制字节组——封装了平台相关的 SIMD 操作
struct GroupSse2Impl {
    __m128i ctrl;

    explicit GroupSse2Impl(const ctrl_t* pos) {
        ctrl = _mm_loadu_si128(reinterpret_cast<const __m128i*>(pos));
    }

    // 返回与 h 匹配的位掩码
    BitMask<uint32_t, kGroupWidth> Match(h2_t h) const {
        auto match = _mm_set1_epi8(static_cast<char>(h));
        return BitMask<uint32_t, kGroupWidth>(
            static_cast<uint32_t>(
                _mm_movemask_epi8(_mm_cmpeq_epi8(match, ctrl))));
    }

    // 返回 EMPTY 位置的掩码
    BitMask<uint32_t, kGroupWidth> MatchEmpty() const {
        // 这里用了一个技巧：
        // 只有 EMPTY(0xFF) 对 _mm_sign_epi8 是特殊的
        return BitMask<uint32_t, kGroupWidth>(
            static_cast<uint32_t>(
                _mm_movemask_epi8(
                    _mm_cmpeq_epi8(
                        _mm_set1_epi8(static_cast<char>(kEmpty)),
                        ctrl))));
    }

    // 返回 EMPTY 或 DELETED 位置的掩码
    BitMask<uint32_t, kGroupWidth> MatchEmptyOrDeleted() const {
        // 所有特殊状态的最高位都是 1
        // _mm_movemask_epi8 提取最高位
        auto special = _mm_set1_epi8(static_cast<char>(kSentinel));
        return BitMask<uint32_t, kGroupWidth>(
            static_cast<uint32_t>(
                _mm_movemask_epi8(
                    _mm_cmpgt_epi8_fixed(special, ctrl))));
    }
};

template <class T, int SignificantBits>
class BitMask {
public:
    explicit BitMask(T mask) : mask_(mask) {}

    // 返回最低匹配位的索引
    uint32_t LowestBitSet() const {
        return static_cast<uint32_t>(__builtin_ctz(mask_));
    }

    // 清除最低匹配位
    BitMask& RemoveLowestBit() {
        mask_ &= mask_ - 1;
        return *this;
    }

    explicit operator bool() const { return mask_ != 0; }

    // 支持 range-for
    class iterator {
        // ...
    };

private:
    T mask_;
};

template <class K>
iterator find(const K& key) {
    auto hash = hash_ref()(key);
    auto seq = probe(ctrl_, hash, capacity_);

    while (true) {
        Group g{ctrl_ + seq.offset()};
        for (uint32_t i : g.Match(H2(hash))) {
            // 注意 i 需要 wrap around
            size_t idx = seq.offset(i);
            if (ABSL_PREDICT_TRUE(PolicyTraits::apply(
                    EqualElement<K>{key, eq_ref()},
                    PolicyTraits::element(slots_ + idx)))) {
                return iterator_at(idx);
            }
        }
        if (ABSL_PREDICT_TRUE(g.MatchEmpty())) {
            return end();
        }
        seq.next();
        // 安全检查：不应该探测超过 capacity 个 group
        assert(seq.index() <= capacity_ && "full table!");
    }
}

struct GroupPortableImpl {
    uint64_t ctrl;  // 将 8 个字节打包到一个 uint64_t

    explicit GroupPortableImpl(const ctrl_t* pos) {
        // 只处理 8 个字节而非 16 个
        memcpy(&ctrl, pos, sizeof(ctrl));
    }

    BitMask<uint64_t, kGroupWidth> Match(h2_t h) const {
        // 用位运算模拟 SIMD 比较
        auto x = ctrl ^ (kMsbs * h);
        // 如果某字节为 0，则 (x - kLsbs) & ~x & kMsbs 的对应位为 1
        return BitMask<uint64_t, kGroupWidth>(
            (x - kLsbs) & ~x & kMsbs);
    }
};