ska::flat_hash_map 源码分析

背景
#

最近在调研各种 hashmap，发现 ska::flat_hash_map 性能优秀，于是来看看代码。最大的特点是，它使用了带 probe count 上限的 robin hood hashing。

相关概念
#

Distance_from_desired
#

对于采用了 open addressing 的 hash 实现，当插入发生冲突时，会以一定方式（如线性探测、平方探测等）探测下一个可以插入的 slot。因而实际插入的 slot 位置与理想的 slot 位置通常不相同，这段距离定义为 distance_from_desired。在没有冲突的理想情况下，所有 distance_from_desired 的值应该都为 0。 distance_from_desired 的一种更常见的说法叫做 probe sequence lengths（PSL）。

robin hood hashing
#

robin hood hashing 的核心思想是“劫富济贫”： distance_from_desired 小的 slot 被认为更“富有”，distance_from_desired 大的 slot 被认为更“贫穷”。具体来说，当插入一个新的元素时，如果当前位置元素的 distance_from_desired 小于待插入元素的 distance_from_desired，那么就将待插入元素放入当前位置，把当前位置的元素取出，继续寻找新的位置。

这样做使得所有元素的 distance_from_desired 分布更为平均，variance 更小。这样的分布对 cache 更友好（几乎全部元素的 distance_from_desired 都小于一个 cache line 的长度，因此在 find 的时候只需要 fetch 一次 cache line），从而拥有更好的性能。

一般的 robin hood hashing 在 find 时，会用一个全局最大的 distance_from_desired 作为没有找到该元素的终止条件。一种常见的改进是，不维护全局最大 distance_from_desired，而是在看到当前位置元素的 distance_from_desired 比要插入元素的 distance_from_desired 小时终止。

以插入 C 为例，探测与交换的具体过程如下：

 1
iterator find(const FindKey& key) {
  size_t index =
      hash_policy.index_for_hash(hash_object(key), num_slots_minus_one);
  EntryPointer it = entries + ptrdiff_t(index);
  for (int8_t distance = 0; it->distance_from_desired >= distance;
       ++distance, ++it) {
    if (compares_equal(key, it->value)) return {it};
  }
  return end();
}

带上限的 robin hood hashing
#

一般的 robin hood hashing 在 insert 时，会不断寻找（包括可能的 swap 过程），直到找到一个空的 slot 为止。该过程在 hash table 较满时可能接近线性的时间复杂度。 ska::flat_hash_map 对这一点的改进是，限制了 insert 时尝试的上限次数，作者给出的经验值为 log(N)，其中 N 为 slots 的个数。这样保证每个 slot 的最大 distance_from_desired 不会超过 log(N)。

关键实现
#

emplace
#

插入一个元素，分析见注释。其中 emplace 函数主要负责查找是否已经存在该元素，以及调整到合适的插入位置；emplace_new_key 函数执行真正的 emplace 操作。

详细代码

template <typename Key, typename... Args>
std::pair<iterator, bool> emplace(Key&& key, Args&&... args) {
  size_t index =
      hash_policy.index_for_hash(hash_object(key), num_slots_minus_one);
  EntryPointer current_entry = entries + ptrdiff_t(index);
  int8_t distance_from_desired = 0;
  // 插入前先查找是否存在
  // 只需要查找有限的距离
 9
  // trick在于，初始 current_entry->distance_from_desired为-1
  // 此时不会进入for loop,直接进行emplace_new_key。
  // 该for loop有两层意义: 1.在index位置不为空时找到合适的位置: 空的slot或者更富有的slot(也就是current_entry->distance_from_desired < distance_from_desired的slot)
  // 2.在该过程中找一下是否已经插入了该值
  for (; current_entry->distance_from_desired >= distance_from_desired;
       ++current_entry, ++distance_from_desired) {
    if (compares_equal(key, current_entry->value))
      return {{current_entry}, false};
  }
  return emplace_new_key(distance_from_desired, current_entry,
                         std::forward<Key>(key), std::forward<Args>(args)...);
}
22
23
24
25
template <typename Key, typename... Args>
SKA_NOINLINE(std::pair<iterator, bool>)
emplace_new_key(int8_t distance_from_desired, EntryPointer current_entry,
                Key&& key, Args&&... args) {
  using std::swap;
  // num_slots_minus_one初始值为0,表示第一次进行插入，需要先进行grow，很合理。
  // 如果得到max_load_factor，或者查找次数达到max_lookups，就进行rehash
  if (num_slots_minus_one == 0 || distance_from_desired == max_lookups ||
      num_elements + 1 >
          (num_slots_minus_one + 1) * static_cast<double>(_max_load_factor)) {
    grow();
    return emplace(std::forward<Key>(key), std::forward<Args>(args)...);
  } else if (current_entry->is_empty()) {
    current_entry->emplace(distance_from_desired, std::forward<Key>(key),
                           std::forward<Args>(args)...);
    ++num_elements;
    return {{current_entry}, true};
  }
44
  // 执行到这里，说明有更富有的slot。于是进行swap,转而为被换出的pair<key,value>找一个新的slot
  // to_insert是当前要插入的，由于swap的发生，可能并不是最初要插入的那一对值
  value_type to_insert(std::forward<Key>(key), std::forward<Args>(args)...);
  swap(distance_from_desired, current_entry->distance_from_desired);
  swap(to_insert, current_entry->value);
  iterator result = {current_entry};
  for (++distance_from_desired, ++current_entry;; ++current_entry) {
    if (current_entry->is_empty()) {
      // 如果被换过的slot后面某个slot是空的，就直接放置了
      current_entry->emplace(distance_from_desired, std::move(to_insert));
      ++num_elements;
      return {result, true};
    } else if (current_entry->distance_from_desired < distance_from_desired) {
      // 在找新slot的过程中，仍然进行劫富济贫的操作， 转而为被换出的pair<key,value>找一个新的slot
      swap(distance_from_desired, current_entry->distance_from_desired);
      swap(to_insert, current_entry->value);
      ++distance_from_desired;
    } else {
      // 如果没有空的slot,也没有更富有的slot,那就只能继续往前寻找了，直到达到上限
      ++distance_from_desired;
      if (distance_from_desired == max_lookups) {
        // 如果找了max_lookups个位置还没找到，就进行rehash
        swap(to_insert, result.current->value);
        grow();
        return emplace(std::move(to_insert));
      }
    }
  }
}

rehash
#

void rehash(size_t num_buckets) {
  num_buckets = std::max(
      num_buckets,
      static_cast<size_t>(
          std::ceil(num_elements / static_cast<double>(_max_load_factor))));
  if (num_buckets == 0) {
    reset_to_empty_state();
    return;
  }
  auto new_prime_index = hash_policy.next_size_over(num_buckets);
  if (num_buckets == bucket_count()) return;
  int8_t new_max_lookups = compute_max_lookups(num_buckets);
  // 额外分配了max_lookups个slots,避免了find时bound checking的开销
  EntryPointer new_buckets(
      AllocatorTraits::allocate(*this, num_buckets + new_max_lookups));
  EntryPointer special_end_item =
      new_buckets + static_cast<ptrdiff_t>(num_buckets + new_max_lookups - 1);
  for (EntryPointer it = new_buckets; it != special_end_item; ++it)
    it->distance_from_desired = -1;
  special_end_item->distance_from_desired = Entry::special_end_value;
  std::swap(entries, new_buckets);
  std::swap(num_slots_minus_one, num_buckets);
  --num_slots_minus_one;
  hash_policy.commit(new_prime_index);
  int8_t old_max_lookups = max_lookups;
  max_lookups = new_max_lookups;
  num_elements = 0;
  // new_buckets其实是旧的entries
  // num_buckets其实也是旧的值，因为已经被swap了
  for (EntryPointer
           it = new_buckets,
           end = it + static_cast<ptrdiff_t>(num_buckets + old_max_lookups);
       it != end; ++it) {
    if (it->has_value()) {
      emplace(std::move(it->value));
      it->destroy_value();
    }
  }
  deallocate_data(new_buckets, num_buckets, old_max_lookups);
}

一些其他 trick
#

通过多分配 log(N) 个 slot 消除 bound checking 的开销
#

代码见上面的 rehash 部分。由于每个元素的最大 distance_from_desired 不会超过 log(N)，因此可以保证查找时不需要做 bound checking，使得 find 部分的实现非常简洁。

使用素数个 slot 而不是 2 的整数次幂个 slot
#

2 的整数次幂个 slot 是一种很常见的实现。这种实现的主要好处是在将 hash 转换为 index 时，避免了代价高昂的取模操作，而是用代价很小的按位与（&）替代。

但是使用 2 的整数次幂个 slot 的缺点是，取模后得到的结果较少，比起使用素数个 slot 更容易发生冲突。 ska::flat_hash_map 的作者借鉴了 boost::multi_index 中的做法，将变量展开为 compile time const（对 compile time const 做取模运算要远远快于对变量做取模运算），从而减小了这部分开销的影响。

 1
 2struct prime_number_hash_policy {
static size_t mod0(size_t) {
  return 0llu;
}
static size_t mod2(size_t hash) {
  return hash % 2llu;
}
static size_t mod3(size_t hash) {
  return hash % 3llu;
}
static size_t mod5(size_t hash) {
  return hash % 5llu;
}
static size_t mod7(size_t hash) {
  return hash % 7llu;
}
static size_t mod11(size_t hash) {
  return hash % 11llu;
}
static size_t mod13(size_t hash) {
  return hash % 13llu;
}
static size_t mod17(size_t hash) {
  return hash % 17llu;
}
static size_t mod23(size_t hash) {
  return hash % 23llu;
}
static size_t mod29(size_t hash) {
  return hash % 29llu;
}
static size_t mod37(size_t hash) {
  return hash % 37llu;
}
static size_t mod47(size_t hash) {
  return hash % 47llu;
}
...

背景#

相关概念#

Distance_from_desired#

robin hood hashing#

带上限的 robin hood hashing#

关键实现#

emplace#

rehash#

一些其他 trick#

通过多分配 log(N) 个 slot 消除 bound checking 的开销#

使用素数个 slot 而不是 2 的整数次幂个 slot#

相关文章