有效处理并发 stl 容器扩展的方法答案

【问题标题】：Ways to handle concurrent stl-container expansions efficiently有效处理并发 stl 容器扩展的方法
【发布时间】：2013-12-04 14:01:09
【问题描述】：

我有一个可爱的小问题，我有一个 STL 容器（unordered_map 或向量），它可以被任意数量的线程读取和扩展，我希望尽可能高效地完成它（即减少锁定延迟尽可能）。

现在，最简单的部分是使用共享锁进行读取，使用排他锁进行扩展，所以我可能会这样做：

boost::shared_lock myLock;
std::unordered_map myMap;

...

myLock.lock_shared();
//try looking up key
myLock.unlock_shared();
if(!success) {
    myLock.lock();
    //retry looking up key
    if(!success) {
        myMap[key] = value;
    }
    myLock.unlock();
}

虽然我认为这在一般情况下相当有效，但如果myMap 决定需要重新分配其内部存储，则持有独占锁的时间可能会爆炸。持有锁的时间可能会达到几百微秒，并且在此期间它会阻塞任何试图读取容器的线程。

有人知道在重新分配存储时可用于避免阻塞所有读取线程的习语吗？

当然，我知道这些延迟不会累积来降低整体性能的数学公式，但是如果我能以某种方式限制可能的延迟，我会更开心。

【问题讨论】：

您可能想要一个在开发时考虑到并发访问的容器，因此它只会在扩展期间的关键点包含适当的锁定（例如，它可以分配新的存储空间并进行复制在锁之外，然后只需锁定换入）。我不相信 STL/Boost 的工作方式 - 因此您当前的粗粒度锁定方法。可能有其他库已经以这种方式工作，或者您可能必须推出自己的...
也许在读取线程中使用try_lock_shared() 会很好，这样它们就不会在尝试获取锁时被阻塞。当然，这是假设他们在无法获得锁时可以做其他事情。

标签： c++ multithreading stl

【解决方案1】：

我可以在这里看到两种可能的解决方案。但它们可能与您程序的其他逻辑不兼容。

1) 您可以使用vector::reserve()/unordered_map.reserve() 预先分配一次（或者比STL 更少）足够数量的容器元素。如果这适合您的程序，那么您将避免进一步的容器重新分配或减少它们的数量。

2) 您可以使用复制修改交换方法。它可能会大大增加内存分配和复制的数量，但也会大大减少线程阻塞的时间。

代码是这样的：

atomic<unordered_map*> pmyMap;

void ChangeMap(){
  unordered_map* pOldMap = pmyMap.load();
  unordered_map* pNewMap = 0;
  {
    if (pNewMap != 0) delete pNewMap;
    pNewMap = new std::unordered_map(*pOldMap); // copy
    (*pNewMap)[key] = value; // modify the copy
  }
  while (!pmyMap.compare_exchange_weak(pOldMap, pNewMap)) // swap
}

在这种情况下，您可以在没有锁的情况下从容器中读取。这里的主要问题是正确删除容器的旧副本。您不能简单地在 ChangeMap() 末尾删除它，因为其他线程可以同时从中读取（例如迭代它）。要正确删除旧容器，您必须以某种方式跟踪读取线程，并仅在所有读取完成后才删除容器对象。

【讨论】：

我已经考虑过这两种变体。 reserve() 的问题在于，在调用 reserve() 时我仍然需要进行排他锁，而且我没有容器大小的上限。如果容器很大，每次插入的复制修改交换都是昂贵的。尽管如此，我仍在绞尽脑汁寻找适合我需要的那些方法的变体......
@cmaster 如果您主要关心的是重新定位容器时的阻塞时间，那么您可能需要一个在增长时完全不重新定位内存的容器？像单链表或双链表（std::list，std::forward_list）？当然，这样的容器会增加搜索时间，但重定位的问题会消失。或者可能是 std::map/std::set？我可能错了，但据我所知，它们在增长时也不会重新定位内存。
@Alex_Antonov：正如我在对 MarkB 的评论中所说，（查找的）性能和最大延迟都是我关心的问题。虽然解决一个很容易，但只有一个并不能解决问题......

【解决方案2】：

英特尔的 TBB 库非常好。我使用了 concurrent_hash_map。

【讨论】：

感谢您的提示。不幸的是，我在文档中找不到任何说明concurrent_hash_map 在重新分配存储时是否阻塞所有读取线程的内容。你知道他们在内部是怎么做的吗？另一点是我需要处理不可变的对象，因此不需要锁定对象。然而，concurrent_hash_map 确实通过访问器锁定，这意味着我想避免开销。
根据我对实现的理解，他们使用 RW 信号量。我不确定为什么您认为不可变对象会改变并发要求，因为容器对象并不关心您在其中存储的内容。如果您确实拥有无限的容器大小，那么 RW 信号量是最有效的选择（读取是畅通的，前提是没有人在写入）。另外，您是否测量过代码的性能？锁定的开销很小……通常是应用程序导致锁争用成为瓶颈。
我知道 RW 锁是个好东西，当然，它们也是我目前正在使用的东西。另外，整体性能只是我关心的两个问题之一，另一个是不会长时间阻塞整个应用程序。在我需要这个的上下文中，阻塞整个应用程序一毫秒几乎是一个错误（上下文有点复杂，所以我不会详细说明）。正如我现在所看到的，我可以追求性能（vector/unordered_map）或保证低阻塞时间（链表）。我的问题是，如果我能同时获得...

【解决方案3】：

有两个明显的解决方案：reserve 开始之前有足够的空间，或者复制-修改-交换。

两者都有明显的缺点。如果你没有足够的reserve，你最终不得不偶尔阻止一切重新reserve。而且复制修改交换通常是矫枉过正的。

两者都这样做。

跟踪何时需要重新分配。当您不需要重新分配时，只需锁定并写入即可。

当您需要重新分配时，复制、重新分配、存储，然后锁定和换入。

现在事情变得棘手了。如果两个线程尝试存储，我们希望第二个线程阻塞。所以我们需要一个两层系统，如果发生复制交换，写入线程会阻塞，而读取线程只会在锁定和交换时阻塞。

boost::shared_lock myLock;
boost::lock lock2;
std::unordered_map myMap;

...

myLock.lock_shared();
//try looking up key
myLock.unlock_shared();
if(!success) {
  lock2.lock(); -- write-exclusive
  if (we have room to add another value) {
    myLock.lock(); -- block-readers
    myMap[key] = value;
    myLock.unlock(); // allow readers back in, time is one write with no realloc
  } else {
    auto myCopy = myMap; // ouch
    myCopy.reserve( 50% more memory, or 2x as much memory, or whatever );
    myCopy[key] = value;
    myLock.lock(); // -- block readers
    using std::swap;
    swap( myCopy, myMap );
    myLock.unlock(); // allow readers back in, time is 1 swap!         
  }
  lock2.unlock();
}

这样做的一个缺点是，如果我们正在执行 realloc，其他写入器可能需要等待很长时间才能写入。

如果你只有一个写线程，lock2 不需要存在。

【讨论】：