tbb::concurrent_hash_map并行find与遍历数据量不一致问题排查

我在两个线程中操作同一个tbb::concurrent_hash_map：一个线程执行find()读操作，另一个线程遍历map，期间没有任何插入或删除操作。奇怪的是，不执行find()时遍历正常，但执行find()后，用HashMap::iterator遍历得到的元素数量和size()结果不一致。使用的tbb版本是2021.5.0-7ubuntu2，测试示例及结果如下：

测试结果 #

writeTime=0.367443
------------finish write---------------

 hashMap size = 3174014
 trav_cnt=3480029

 error: trav_cnt=3480029 hashmap_size=3174014
-----readTime=2.19917

测试代码 #

#include<map>
#include<string>
#include<vector>
#include<fstream>
#include<iostream>
#include <omp.h>
#include <tbb/concurrent_hash_map.h>
#include <tbb/tbb.h>

template <typename KeyType, typename ValueType>
class ConcurrentHashMap {
  private:
    typedef tbb::concurrent_hash_map<KeyType, ValueType> HashMap;
    typedef typename HashMap::const_accessor HashMapConstAccessor;
    typedef typename HashMap::accessor HashMapAccessor;
    typedef typename HashMap::iterator HashMapIterator;
    typedef typename HashMap::value_type HashMapValuePair;

  public:
    ConcurrentHashMap (ValueType default_value) : DEFAULT_VALUE(default_value) {
    }

    size_t size() {
      return hashMap.size();
    }

    bool insert(const KeyType& key, const ValueType& value) {
      HashMapAccessor accessor;
      bool inserted = hashMap.insert(accessor, key);
      if (inserted) {
        accessor->second = value;
      }
      return inserted;
    }
    bool erase(const KeyType& key) {
      return hashMap.erase(key);
    }

    bool find(const KeyType& key, ValueType& value) {
      HashMapConstAccessor accessor;
      if (hashMap.find(accessor, key)) {
        value = accessor->second;
        return true;
      }
      value = DEFAULT_VALUE;
      return false;
    }

    void clear() {
      hashMap.clear();
    }

    const ValueType& operator[](const KeyType& key) const {
      HashMapConstAccessor accessor;
      hashMap.find(accessor, key);
      if (hashMap.find(accessor, key)) {
        return accessor->second;
      }
      // accessor->second = DEFAULT_VALUE;
      return DEFAULT_VALUE;
    }

    HashMapIterator begin() {
      return hashMap.begin();
    }

    HashMapIterator end() {
      return hashMap.end();
    }

  private:
    HashMap hashMap;
    ValueType DEFAULT_VALUE;
};

class A;

using HashMap = ConcurrentHashMap<int, A*>;
// using HashMap = ConcurrentUnorderedMap<int, A*>;

class A {
public:
  A(int _a, int _b): a(_a), b(_b) {

  }
  void sub () {

  }
  int a = 1;
  int b = 0;;
};

void test(int N, HashMap& hashMap) {
  int thread_num = 16;

  std::thread writer(
    [&] () {
      auto writeStartTime = std::chrono::high_resolution_clock::now();
      #pragma omp parallel for num_threads(thread_num)
      for (int i = 0; i < N; i++) {
        hashMap.insert(i, new A(1, i));
      }
      auto writeEndTime = std::chrono::high_resolution_clock::now();
      double writeTime = std::chrono::duration<double>(writeEndTime 
                                                      - writeStartTime).count();
      std::cout << "writeTime=" << writeTime << std::endl;
    }
  );

  writer.join();
}

int random_uniform_int(const int min = 0, const int max = 1) {
  unsigned seed = 2000;
  static thread_local std::mt19937 generator(seed);
  std::uniform_int_distribution<int> distribution(min, max);
  return distribution(generator);
}

int main () {
  // cmd: g++ test_con_hashmap.cpp -fopenmp -ltbb  && ./a.out
  int N = 3174014;
  std::nullptr_t NULLPOINTER = nullptr;
  HashMap hashMap(NULLPOINTER);

  test(N, hashMap);

  std::cout << "------------finish write---------------" << std::endl;

  size_t hashmap_size = hashMap.size();
  std::cout << "\n hashMap size = " << hashmap_size << std::endl;

{ 
  int thread_num = 32;
  std::thread reader(
    [&] () {
      std::this_thread::sleep_for(std::chrono::milliseconds(1));
      auto readStartTime = std::chrono::high_resolution_clock::now();

      for (int i = 0; i < 30; i++) {
        #pragma omp parallel for num_threads(thread_num)
        for (int i = 0; i < N; i++) {
          A* value;
          int randomKey = random_uniform_int(0, N - 1);
          if(hashMap.find(randomKey, value)){
            
          }
        }
      }
      auto readEndTime = std::chrono::high_resolution_clock::now();
      double readTime = std::chrono::duration<double>(readEndTime 
                                               - readStartTime).count();
      std::cout << "-----readTime=" << readTime << std::endl;
    }
  );

  size_t trav_cnt = 0;
  for(auto iterator1 = hashMap.begin(); 
      iterator1 != hashMap.end(); ++iterator1 ){
    trav_cnt++;
  }
  std::cout << " trav_cnt=" << trav_cnt << std::endl;

  if (trav_cnt != hashmap_size) {
    std::cout << "\n error: trav_cnt=" << trav_cnt
              << " hashmap_size=" << hashmap_size
              << std::endl;
  }


  reader.join();
}

  hashMap.clear();

  return 0;
}

问题原因 #

tbb::concurrent_hash_map的非const迭代器遍历不支持与并发find操作同时进行，核心原因如下：

• 非const迭代器遍历是实时扫描容器的桶结构，而find操作会持有const_accessor锁，这会导致容器内部桶的访问状态发生变化，遍历过程中可能重复扫描到同一个元素，最终计数大于实际size。
• size()返回的是容器维护的原子性计数，结果准确；但遍历依赖迭代器对桶的逐个扫描，并发锁竞争会破坏遍历的唯一性，导致计数偏差。

解决方法 #

1. 使用const迭代器遍历 #

将封装类的begin()和end()改为返回const迭代器，const迭代器在并发读场景下是安全的，不会出现重复计数问题：

template <typename KeyType, typename ValueType>
class ConcurrentHashMap {
  private:
    // 新增const迭代器类型定义
    typedef typename HashMap::const_iterator HashMapConstIterator;
    // ... 原有代码

  public:
    // 修改begin/end为const版本并返回const迭代器
    HashMapConstIterator begin() const {
      return hashMap.cbegin();
    }

    HashMapConstIterator end() const {
      return hashMap.cend();
    }
    // ... 原有代码
};

2. 使用TBB官方并行遍历接口 #

改用tbb::parallel_do这类为并发场景设计的遍历接口，保证遍历正确性：

size_t trav_cnt = 0;
tbb::parallel_do(hashMap.begin(), hashMap.end(), [&](const auto& pair) {
  __sync_fetch_and_add(&trav_cnt, 1);
});

3. 同步遍历与find操作 #

如果必须使用非const迭代器，可在遍历期间加全局锁，禁止其他线程执行find操作，但会损失并发性能。

关键注意点 #

• tbb::concurrent_hash_map的非const迭代器不是线程安全的，不能在并发访问（即使是读）场景下使用。
• const迭代器的遍历在并发读场景下兼容良好，可保证元素不被重复计数。

内容的提问来源于stack exchange，提问作者ystraw y