13

我正在尝试根据此https://software.intel.com/en-us/blogs/2008/07/24/tbbconcurrent_vector-secrets-of-实现类似于 std::vector 的线程安全无锁容器记忆组织

据我了解，为了防止重新分配并使所有线程上的所有迭代器无效，而不是单个连续数组，它们添加了新的连续块。
他们添加的每个块的大小都是 2 的递增幂，因此他们可以使用 log(index) 来找到应该在 [index] 处的项目的正确段。

据我所知，他们有一个指向段的静态指针数组，所以他们可以快速访问它们，但是他们不知道用户想要多少段，所以他们做了一个小的初始段，如果段的数量超过当前计数，他们分配了一个巨大的并切换到使用那个。

问题是，添加新段不能以无锁线程安全的方式完成，或者至少我还没有弄清楚如何。我可以原子地增加当前大小，但仅此而已。
而且从小段指针数组切换到大段指针数组还涉及大量分配和内存副本，所以我无法理解他们是如何做到的。

他们在网上发布了一些代码，但是所有重要的功能都没有可用的源代码，它们在他们的线程构建块 DLL 中。这是一些演示该问题的代码：

template<typename T>
class concurrent_vector
{
    private:
        int size = 0;
        int lastSegmentIndex = 0;

        union
        {
            T* segmentsSmall[3];
            T** segmentsLarge;
        };

        void switch_to_large()
        {
            //Bunch of allocations, creates a T* segmentsLarge[32] basically and reassigns all old entries into it
        }

    public:
        concurrent_vector()
        {
            //The initial array is contiguous just for the sake of cache optimization
            T* initialContiguousBlock = new T[2 + 4 + 8]; //2^1 + 2^2 + 2^3
            segmentsSmall[0] = initialContiguousBlock;
            segmentsSmall[1] = initialContiguousBlock + 2;
            segmentsSmall[2] = initialContiguousBlock + 2 + 4;
        }

        void push_back(T& item)
        {
            if(size > 2 + 4 + 8)
            {
                switch_to_large(); //This is the problem part, there is no possible way to make this thread-safe without a mutex lock. I don't understand how Intel does it. It includes a bunch of allocations and memory copies.
            }

            InterlockedIncrement(&size); //Ok, so size is atomically increased

            //afterwards adds the item to the appropriate slot in the appropriate segment
        }
};
4

1 回答 1

3

我不会尝试建立segmentsLarge和segmentsSmall联合。是的，这又浪费了一个指针。然后指针，让我们调用它，它segments最初可以指向segmentsSmall。

另一方面，其他方法总是可以使用相同的指针，这使得它们更简单。

而从小到大的切换可以通过指针的一次比较交换来完成。

我不确定如何通过工会安全地实现这一点。

这个想法看起来像这样（请注意，我使用的是英特尔库早于的 C++11，因此他们很可能使用其原子内在函数来做到这一点）。这可能会遗漏很多细节，我相信英特尔人员已经考虑了更多，因此您可能必须对照所有其他方法的实现来检查这一点。

#include <atomic>
#include <array>
#include <cstddef>
#include <climits>

template<typename T>
class concurrent_vector
{
private:
  std::atomic<size_t> size;
  std::atomic<T**> segments;
  std::array<T*, 3> segmentsSmall;
  unsigned lastSegmentIndex = 0;

  void switch_to_large()
  {
    T** segmentsOld = segments;
    if( segmentsOld == segmentsSmall.data()) {
      // not yet switched
      T** segmentsLarge = new T*[sizeof(size_t) * CHAR_BIT];
      // note that we leave the original segment allocations alone and just copy the pointers
      std::copy(segmentsSmall.begin(), segmentsSmall.end(), segmentsLarge);
      for(unsigned i = segmentsSmall.size(); i < numSegments; ++i) {
        segmentsLarge[i] = nullptr;
      }
      // now both the old and the new segments array are valid
      if( segments.compare_exchange_strong(segmentsOld, segmentsLarge)) {
        // success!
        return;
      }  else {
        // already switched, just clean up
        delete[] segmentsLarge;
      }
    }
  }

public:
  concurrent_vector()  : size(0), segments(segmentsSmall.data())
  {
    //The initial array is contiguous just for the sake of cache optimization
    T* initialContiguousBlock = new T[2 + 4 + 8]; //2^1 + 2^2 + 2^3
    segmentsSmall[0] = initialContiguousBlock;
    segmentsSmall[1] = initialContiguousBlock + 2;
    segmentsSmall[2] = initialContiguousBlock + 2 + 4;
  }

  void push_back(T& item)
  {
    if(size > 2 + 4 + 8) {
      switch_to_large();
    }
    // here we may have to allocate more segments atomically
    ++size;

    //afterwards adds the item to the appropriate slot in the appropriate segment
  }
};
于 2017-10-19T13:45:57.123 回答