线程安全，C++ 中的有序映射/散列？

Question

在 C++ 中实现 threadsafe ordered(note1) 映射/散列的最佳方式是什么？又名，一种快速查找数据结构（又名，不是队列），不同线程可以迭代，偶尔插入或删除元素而不干扰其他线程的活动？

• std::map 不是线程安全的，它的操作是非原子的 - 尽管只有擦除会使迭代器失效

• 将每个函数包装在整个地图类中并不能解决问题 - 您可以有松散的迭代器指向一个被另一个线程擦除的节点。它应该锁定并阻止删除，直到当前线程是唯一引用它的线程，或者使用 UNIX 文件系统风格的“悬空但在删除时仍然有效的引用”方法

• tbb::concurrent_hash_map 被设计为线程安全的，但它的迭代器在删除其键时仍然无效。即使 Value 是智能指针，也不会保留数据，因为迭代器中的引用会丢失。
• 可以通过遍历键而不是迭代器来使用 tbb:concurrent_hash_map（可以将键查找为 O(1) 而不是像 std::map 那样的 O(log N)），但由于它是一个散列，它缺少与顺序相关的函数，如上界、下界等。关键的是，如果一个键在另一个线程中被擦除而悬空，没有明显的方法可以告诉代码跳回到哈希中离该键最近的元素。李>
• std::unordered_map 如果您的密钥通过存储桶访问功能被删除，则可能会尝试通过陪审团找到最近的元素。但是 std::unordered_map 不被认为是线程安全的（尽管它可能被包装为线程安全的）。 tbb::concurrent_hash_map 是线程安全的（受上述约束），但它不允许足够的存储桶访问权限来执行此操作。
• std::map，通过键而不是迭代器访问，如果键被另一个线程删除，则可以在映射中找到下一个最近的元素。但每次查找都是 O(log N) 而不是 O(1)。
• 另外，如前所述，std::map 是非原子的。它必须被包裹起来。使用键而不是迭代器还需要包装整个类，因为必须将通常采用或返回迭代器的所有函数更改为采用或返回键。
• 我读过的一个站点上的某个人说 std::map 在线程环境中无论如何都不能很好地工作（与哈希相比），因为线程不能很好地与树重新平衡一起使用，尽管我真的不知道为什么将是或者它是否普遍适用/准确。

你认为最好的方法是什么？

** 注1：当我写“ordered”时，我的意思是仅从“具有可以迭代的可靠排序”的角度来看，而不是“迭代必须按照键的顺序进行”。在我写完之后，我意识到我的几个用例确实关心迭代的顺序（大多数不关心）。但无论哪种方式，我都可以通过在 Value 类型中使用链表来判断正确的排序。只是更慢的丑陋和潜在的问题/疏忽......

** 注2：新思想。太糟糕的地图没有对其迭代器类型进行模板化......改变迭代器 std::map 的构建有多难？我之前一直在玩弄让迭代器成为引用计数（如 std::shared_ptr）的想法，但我一直错误地想用迭代器本身内部的辅助数据结构来实现引用计数，结果总是太丑陋了/缓慢/不切实际。但现在我突然想到，可以将引用计数包含在映射的键值对的值中。也就是说，每个值都将包括 A) 一个引用计数器（默认值 = 0），每个交互器在到达它时递增（operator=、operator++、operator-- 等）并在它离开时递减；和 B) 擦除功能设置的擦除标志 (default=false)。每当迭代器将引用计数器减为零时，如果设置了擦除标志，那么它将然后实际擦除它。

在我看来，虽然它会影响性能（额外的增量/减量/检查），但每次您想要逐步浏览结构时都必须进行完整的地图查找，这肯定没什么。谁能想到一个实用的方法来实现这个？

Answer 1

由于某种原因，您错过了tbb::concurrent_unordered_map，它是具有线程安全迭代支持的哈希表。它基于拆分有序列表算法，其中除了哈希表之外，元素还以容器范围的列表结构连接，因此迭代是直截了当的。但它并不完全符合您的要求，因为它不支持并发擦除。

这是一个根本问题，在没有内存回收机制的情况下，很难在一个并发数据结构中同时融合快速遍历和安全擦除属性，您必须在这里选择：安全/一致性或速度.

在某些限制和谨慎的情况下，您可以同时进行遍历和删除，如this blog 中所述。基本上，它说只要您可以交错（相互排除）遍历和擦除，tbb::concurrent_hash_map 可以与 find&insert 一起用于并发遍历。该博客建议使用双重检查模式进行额外优化。但可以简化为：

for(iterator = table.begin(); iterator != table.end(); iterator++ ) {
    accessor acc;
    // a key cannot be changed thus it is safe to read it without lock
    table.find( acc, iterator->first );   // now get the get the lock
    if( acc->second.market_for_deletion )
        table.erase( acc );               // erase only by accessor
}

它本质上类似于您应用于 concurrent_hash_map 案例的注释 2，因为最大开销不是来自查找（对于相邻元素，缓存未命中的可能性较小），而是来自与两个锁（内部存储桶的锁和元素的访问器）的同步。

但是如果这种遍历方式的速度太慢或者对你来说太hacky（依赖于实现细节），但你仍然迫切需要能够删除并发哈希表的元素，考虑使用RW - 像tbb::spin_rw_mutex 和tbb::concurrent_unordered_map 一样锁定。您需要找到一个可以不太频繁地获取读锁的最佳位置，以便在没有太多开销的情况下启用迭代、查找和插入，并且也不要太频繁地在写锁下进行擦除。可能需要额外的方案来标记和收集足够的元素，然后才能真正删除它们。例如。这是这样一个哈希表类的伪代码：

class concurrent_hash_table_with_erase_and_traverse {
    tbb::concurrent_unordered_map my_map;
    tbb::spin_rw_mutex            my_lock; // acquired as writer for cleanup only
    tbb::atomic<size_t>           my_trash_count; // indicates # of items for erase

public:
    void init_thread_for_concurrent_ops() { my_lock.lock_read(); }
    void release_thread()                 { my_lock.unlock(); } // assuming reader lock
    mapped_type read(key_type k) {
        // assert: under read lock (thread is initialized)
        if(my_trash_count > threshold) {  // time to remove items
            my_lock.unlock(); // release reader
            // waiting all the threads to enter this container
            // TODO: re-implement with try_lock and checking the condition 
            my_lock.lock();   // acquire writer

            if(my_trash_count > threshold) { // double-check
                my_trash_count = 0;
                for( auto it = my_map.begin(); it != my_map.end(); ) {
                    auto _it = it++;
                    if( _it->is_marked_for_erase )
                        my_map.unsafe_erase( _it );
                }
            }
            my_lock.unlock();    // release writer
            my_lock.lock_read(); // acquire reader
        }
        return my_map[k]; // note: access is not protected like in concurrent_hash_map
    }
    void safe_erase(key_type k) {
        // assert: under read lock
        my_map[k].is_marked_for_erase = true;
        my_trash_count++;
    }
};

Answer 2

好的，所以这需要很长时间......而且我决定用它制作一个 github 项目；）但我终于有了一个线程安全的映射类。完整的测试套件和许多可配置的选项。希望如果其他人需要这个，他们会利用它！

https://github.com/KarenRei/safe-map#

Answer 3

可能不是您真正想要的示例线程安全代码。

警告未经测试的代码

template<typename kType, typename dType>
class Locked {
  std::mutex mut;
  std::map<kType, dType> theMap; // change types as required
public:
  const dType get(const kType& key) const {
    std::lock_guard<std::mutex> g(mut);
    auto it = theMap.find(key);
    if (it != theMap.end())
      return *it;
    // throw or return buggy dType
    return dType(-1); // or whatever
  }
  void set(const kType& key, const dType& data) {
    std::lock_guard<std::mutex> g(mut);
    theMap[key] = data;
  }
  void delete(const kType& key) {
    std::lock_guard<std::mutex> g(mut);
    auto it = theMap.find(key);
    if (it != theMap.end()) {
      theMap.erase(it);
      return;
    }
    // throw?
  }
}

如果dType 是一个指针，我认为这不会起作用，除非它是一个 shared_ptr。

只能有一个。

它可以扩展为具有读取器计数器，因此仅设置/删除块，设置可以是 tbb 映射上的读取，因为它允许线程安全插入。

C++14 有std::shared_timed_mutex，这使得读取在性能上更容易一些。

C++17 有 std::shared_mutex 从中删除时间元素。

现在有许多不同的无锁、无等待等实现来解决性能问题。

根据实际负载，一些 spin_lock 而不是互斥锁可能会有所帮助，直到某一点。