c++ - 节点预分配 vector 中的无锁树节点分配

Question

我目前正在尝试多线程创建树，其中包含树的类预先分配了一个std::vector。的 Node s 在一个特定大小的 block 中(概念上大小是任意的)。 Node 的附加 block 仅在必要时创建，这是因为树很快变得非常大，我想避免经常使用 new时间效率的运算符。

Node 的这些 vector s 定义为:std::vector< std::vector< Node > > nodes

head跟踪内部 vector 和 chunkCount 中的位置跟踪当前正在使用的外部 vector 。

vector 在构造函数中调整大小为:

nodes.resize( 1 );
nodes[chunkCount].resize( CHUNK_SIZE );

以及 Node 的简化版本会是:

typedef struct Node {
    int val;
    Node* subnodes[5];
} Node;

创建一个新的 Node如下所示:

void TreeClass::createNode( Node* node, short index, int val )
{
    omp_set_lock( &treeLock ); // treeLock belongs to TreeClass
    head++;
    if( head == CHUNK_SIZE ) {
        std::vector< Node > tempNodeVec( CHUNK_SIZE );
        nodes.push_back( tempNodeVec );
        chunkCount++;
        head = 0;
    }
    node->subnodes[index] = &( nodes[chunkCount][head] );
    omp_unset_lock( &treeLock );

    node->subnodes[index]->val = val;
}

这工作得很好。然而，我担心的是，在创建节点时，除了一个线程外，所有线程都被阻塞了，而且这种情况经常发生，所以很多时间都花在了阻塞或锁定/解锁上 treeLock ，因此我希望使此功能无锁，但到目前为止我的尝试都失败了。

改变 head和 chunkCount使用 #pragma omp atomic 不用锁就足够简单了(或通过使用 std::atomic< int > s)，但这是确保 if( ... ) 背后的逻辑语句仅在任何线程继续分配子地址之前执行一次，即确保它们使用正确/更新的 chunkCount和 head .

阅读无锁算法的一个想法是使用 std::atomic< Node* > subnodes[5]在Node并执行 CAS 操作等待正确更新 head和 chunkCnt但不知道什么是“正确的”，我怎么知道我在等什么？

另一个(天真的)想法是:

int myHead;
if( ++head == CHUNK_SIZE ) {
    std::vector< Node > tempNodeVec( CHUNK_SIZE );
    nodes.push_back( tempNodeVec );
    chunkCount++;
    myhead = head = 0;
} else {
    myhead = head;
    while( head > CHUNK_SIZE )
        myHead = ++head;
}
node->subnodes[index] = &( nodes[chunkCount][myHead] );

想法是只有一个线程进入 if( ... )直到设置 head为 0，其余的将卡在 else { ... } 中但我已经看到了这种方法的许多问题。

如有任何帮助，我们将不胜感激。

Answer 1

我建议您使用线程专用内存池。为此，您可以使用如下注释:

#pragma omp threadprivate(nodes)

这不仅比尝试保护对共享内存池的访问要简单得多，而且由于数据局部性，它还可能会提高性能。

注意:使用您的解决方案实现基于原子的无锁是不可能的，因为 nodes[chunkCount] - 每次分配都需要 - 必须始终受到 nodes.push_back 的保护.

全功能内存池更复杂，但作为一个小步骤，您可以尝试使用 std::deque。它提供了您需要的东西，而不会弄乱两个 vector - 在恒定时间内插入元素，同时不会使现有元素的指针无效。您的控制权较少，但这是一个好的开始。