同步push_back和std::thread

My code

void build(std::vector<RKD <DivisionSpace> >& roots, ...) {
  try {
    // using a local lock_guard to lock mtx guarantees unlocking on destruction / exception:
    std::lock_guard<std::mutex> lck (mtx);
    roots.push_back(RKD<DivisionSpace>(...));
  }
  catch (const std::bad_alloc&) {
    std::cout << "[exception caught when constructing tree]\n";
    return;
  }
}

现在，实际工作应该串行完成，而不是并行完成。

的构造函数RKD可以与其他构造函数并行运行RKD。然而，将物体推回原处std::Vector是一个临界区，对吗？

我要构建的对象的数量是已知的。实际上，它的范围将是 [2, 16]。理论上它可以是任何正数。

另外，我对它们插入容器的顺序不感兴趣。

所以我可以做类似的事情：

RKD tree = RKD(...);
mutex_lock(...);
roots.push_back(tree);

但这就意味着抄袭，不是吗？

我应该怎么做才能使我的代码并行？

我决定使用锁（而不仅仅是互斥锁），因为this answer.

Tomasz Lewowski 在他的评论中提出的建议非常简单，并且基于以下观察：push_back on a std::vector可能需要重新分配后备存储并复制（或者最好移动）元素。这构成了需要同步的关键部分。

对于接下来的示例，假设我们想要一个填充前 12 个素数的向量，但我们不关心它们的顺序。 （我刚刚在这里对数字进行了硬编码，但假设它们是通过一些昂贵的计算获得的，这对于并行计算是有意义的。）在以下场景中存在危险的竞争条件。

std::vector<int> numbers {};  // an empty vector

// thread A             // thread B             // thread C

numbers.push_back( 2);  numbers.push_back(11);  numbers.push_back(23);
numbers.push_back( 3);  numbers.push_back(13);  numbers.push_back(27);
numbers.push_back( 5);  numbers.push_back(17);  numbers.push_back(29);
numbers.push_back( 7);  numbers.push_back(19);  numbers.push_back(31);

另外还有一个问题push_back。如果两个线程同时调用它，它们都将尝试在同一索引处构造一个对象，这可能会带来灾难性的后果。所以问题并没有解决reserve(n)在分叉线程之前。

但是，由于您事先知道元素的数量，因此您可以简单地assign他们到一个特定的位置std::vector而不改变其大小。如果不改变大小，就没有临界区。因此，以下场景中不存在竞争。

std::vector<int> numbers(12);  // 12 elements initialized with 0

// thread A          // thread B          // thread C

numbers[ 0] =  2;    numbers[ 1] =  3;    numbers[ 2] =  5;
numbers[ 3] =  7;    numbers[ 4] = 11;    numbers[ 5] = 13;
numbers[ 6] = 17;    numbers[ 7] = 19;    numbers[ 8] = 23;
numbers[ 9] = 29;    numbers[10] = 31;    numbers[11] = 37;

当然，如果两个线程都尝试写入same指数，比赛将再次举行。幸运的是，在实践中防止这种情况并不困难。如果您的向量有 n 个元素并且您有 p 个线程，则线程 i 仅写入元素 [i n / p, (i + 1) n / p)。请注意，这比仅当 j mod p = i 写入索引 j 处的元素更好。 i>i 因为它会减少缓存失效。所以上面例子中的访问模式不是最优的，最好是这样的。

std::vector<int> numbers(12);  // 12 elements initialized with 0

// thread A          // thread B          // thread C

numbers[ 0] =  2;    numbers[ 4] = 11;    numbers[ 8] = 23;
numbers[ 1] =  3;    numbers[ 5] = 13;    numbers[ 9] = 29;
numbers[ 2] =  5;    numbers[ 6] = 17;    numbers[10] = 31;
numbers[ 3] =  7;    numbers[ 7] = 19;    numbers[11] = 37;

到目前为止，一切都很好。但如果你没有怎么办？std::vector<int> but a std::vector<Foo>? If Foo没有默认构造函数，那么

std::vector<Foo> numbers(10);

将无效。即使它有一个，创建许多昂贵的默认构造对象只是为了很快重新分配它们而不检索值也是令人无法容忍的。

当然，大多数设计良好的类应该有一个非常便宜的默认构造函数。例如，一个std::string默认构造为不需要内存分配的空字符串。一个好的实现会将默认构造字符串的成本降低到仅

std::memset(this, 0, sizeof(std::string));

如果编译器足够聪明，可以发现我们正在分配和初始化整个std::vector<std::string>(n)，它也许能够进一步优化为一次调用

std::calloc(n, sizeof(std::string));

所以如果有任何机会你可以Foo廉价地默认构造和分配，你就完成了。但是，如果这变得很困难，您可以通过将其移至堆来避免该问题。智能指针可以廉价地默认构造，所以

std::vector<std::unique_ptr<Foo>> foos(n);

最终将减少到

std::calloc(n, sizeof(std::unique_ptr<Foo>));

无需你做任何事Foo。当然，这种便利是以每个元素的动态内存分配为代价的。

std::vector<std::unique_ptr<Foo>> foos(n);

// thread A                    // thread B                           // thread C

foos[0].reset(new Foo {...});  foos[n / 3 + 0].reset(new Foo {...});  foos[2 * n / 3 + 0].reset(new Foo {...});
foos[1].reset(new Foo {...});  foos[n / 3 + 1].reset(new Foo {...});  foos[2 * n / 3 + 1].reset(new Foo {...});
foos[2].reset(new Foo {...});  foos[n / 3 + 2].reset(new Foo {...});  foos[2 * n / 3 + 2].reset(new Foo {...});
...                            ...                                    ...

这可能没有您想象的那么糟糕，因为虽然动态内存分配不是免费的，sizeof a std::unique_ptr非常小，所以如果sizeof(Foo)很大，你会得到一个更紧凑的向量，迭代速度更快。当然，这完全取决于您打算如何使用您的数据。

如果您事先不知道元素的确切数量或者担心会弄乱索引，还有另一种方法可以做到这一点：让每个线程填充自己的向量并在最后合并它们。继续素数的例子，我们得到了这个。

std::vector<int> numbersA {};  // private store for thread A
std::vector<int> numbersB {};  // private store for thread B
std::vector<int> numbersC {};  // private store for thread C

// thread A              // thread B              // thread C

numbersA.push_back( 2);  numbersB.push_back(11);  numbersC.push_back(23);
numbersA.push_back( 3);  numbersB.push_back(13);  numbersC.push_back(27);
numbersA.push_back( 5);  numbersB.push_back(17);  numbersC.push_back(29);
numbersA.push_back( 7);  numbersB.push_back(21);  numbersC.push_back(31);

// Back on the main thread after A, B and C are joined:

std::vector<int> numbers(
    numbersA.size() + numbersB.size() + numbersC.size());
auto pos = numbers.begin();
pos = std::move(numbersA.begin(), numbersA.end(), pos);
pos = std::move(numbersB.begin(), numbersB.end(), pos);
pos = std::move(numbersC.begin(), numbersC.end(), pos);
assert(pos == numbers.end());

// Now dispose of numbersA, numbersB and numbersC as soon as possible
// in order to release their no longer needed memory.

(The std::move上面代码中使用的是算法库中的一个.)

这种方法具有最理想的内存访问模式，因为numbersA, numbersB and numbersC正在写入完全独立分配的内存。当然，我们还要做一些额外的事情顺序的连接中间结果的工作。请注意，效率在很大程度上取决于这样一个事实：与查找/创建元素的成本相比，移动分配元素的成本可以忽略不计。至少如上面所写，代码还假设您的类型有一个廉价的默认构造函数。当然，如果您的类型不是这种情况，您可以再次使用智能指针。

我希望这能为您提供足够的想法来优化您的问题。

如果您以前从未使用过智能指针，请看一下“C++ 中的 RAII 和智能指针”并查看标准库的动态内存管理库。上面显示的技术当然也适用于std::vector<Foo *>但在现代 C++ 中我们不再使用这样的拥有资源的原始指针。