c++ - 根据intel博客实现concurrent_vector

Question

我正在尝试实现一个线程安全的无锁容器，类似于 std::vector，根据这个 https://software.intel.com/en-us/blogs/2008/07/24/tbbconcurrent_vector-secrets-of-memory-organization

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

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

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

他们在网上发布了一些代码，但是所有重要的功能都没有可用的源代码，它们在他们的线程构建 block 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
        }
};

Answer 1

我不会尝试将 segmentsLarge 和 segmentsSmall 合并。是的，这又浪费了一个指针。然后指针，我们称之为 segments 最初可以指向segmentsSmall。

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

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

我不确定如何通过 union 安全地完成这项工作。

这个想法看起来像这样(请注意，我使用的是英特尔库早于的 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
  }
};