c - 如何防止通过原子比较和交换实现的并发lifo堆栈中的损坏

Question

以下是一个简化的C程序，该程序演示了我在使用intel cpu上内置的compare和swap的GNU来实现并发堆栈时遇到的问题。我花了一段时间来了解正在发生的事情，但是现在我知道这完全在原子比较和交换所提供的保证之内。

当节点从堆栈中弹出，修改然后放回堆栈时，修改后的值可能会成为堆栈的新头，从而破坏它。 test_get中的注释描述了导致这种情况的事件的顺序。

有什么方法可以可靠地多次使用同一节点和同一堆栈？这是一个夸张的示例，但是即使将未修改的节点返回给gHead也会泄漏其他节点。该数据结构的初衷是重复使用相同的节点。

typedef struct test_node {
    struct test_node *next;
    void *data;
} *test_node_p;

test_node_p gHead = NULL;
unsigned gThreadsDone = 0;

void test_put( test_node_p inMemory ) {
    test_node_p scratch;

    do {
        scratch = gHead;
        inMemory->next = scratch;
    } while ( !__sync_bool_compare_and_swap( &gHead , scratch , inMemory ) );
}

test_node_p test_get( void ) {
    test_node_p result;
    test_node_p scratch;

    do {
        result = gHead;
        if ( NULL == result ) break;
        //  other thread acquires result, modifies next
        scratch = result->next;     //  scratch is 0xDEFACED...
        //  other thread returns result to gHead
    } while ( !__sync_bool_compare_and_swap( &gHead , result , scratch ) );
    //  this thread corrupts gHead with 0xDEFACED... value

    if ( NULL == result ) {
        result = (test_node_p)malloc( sizeof(struct test_node) );
    }

    return result;
}

void *memory_entry( void *in ) {
    test_node_p node;
    int index , count = 1000;
    for ( index = 0 ; index < count ; ++index ) {
        node = test_get();
        *(uint64_t *)(node) |= 0xDEFACED000000000ULL;
        test_put( node );
    }

    __sync_add_and_fetch(&gThreadsDone,1);

    return NULL;
}

void main() {
    unsigned    index , count = 8;
    pthread_t   thread;

    gThreadsDone = 0;

    for ( index = 0 ; index < count ; ++index ) {
        pthread_create( &thread , NULL , memory_entry , NULL );
        pthread_detach( thread );
    }

    while ( __sync_add_and_fetch(&gThreadsDone,0) < count ) {}
}

我正在使用8个逻辑核心运行此测试。当我仅使用4个线程时，问题很少发生，但使用8个线程很容易重现。我有一台配备Intel Core i7的MacBook。

我对使用某些解决了该问题的库或框架不感兴趣，我想知道它是如何解决的。谢谢。

编辑：

这是两种不适用于我的情况的解决方案。

一些体系结构提供ll / sc指令对，它们对地址而不是值执行原子测试。对地址的任何写入（即使是相同的值）都将阻止成功。

一些架构提供了比指针大小更大的比较和交换。这样，单调计数器与指针配对，该指针每次使用时原子地递增，因此该值始终会变化。一些intel芯片支持此功能，但是没有GNU包装器。

这是问题的解决方法。两个线程A和B到达 test_get中的点，它们的 result值相同，但不是 NULL。然后，将发生以下顺序：


线程A通过比较和交换，并从 result返回 test_get
线程A修改 result的内容
线程B取消引用 result，获取放置在其中的任何线程A
线程A以 result结尾并调用 test_put
线程A通过比较并在 test_put中交换，将结果放回 gHead
线程B到达比较并在 test_get中交换并通过
线程B现在使用线程A中的值损坏了 gHead


这是具有三个线程的类似情况，不需要线程A来修改 result。


线程A通过比较和交换，并从 result返回 test_get
线程A不会修改 result的内容
线程B取消引用 result，从而在 scratch中获得有效值
线程C使用不相关的值调用 test_put并成功
线程A调用 test_put并成功将 result放回 gHead
线程B到达比较并在 test_get中交换并通过
线程B现在通过泄漏添加的任何线程C破坏了 gHead


在这两种情况下，问题都是线程A通过比较并交换两次，一次获取，然后再次放置，然后线程B到达比较并交换获取。线程B暂存的任何值都应该丢弃，但这不是因为gHead中的值看起来正确。

任何能够在没有实际阻止的情况下降低这种可能性的解决方案，只会使该错误更难被发现。

请注意，scratch变量只是原子指令开始之前放置result的解引用值的任何地方的名称。删除名称确实会删除取消引用和比较之间的时间片，而这可能会中断。

还要注意，原子意味着它整体上是成功还是失败。指针的任何对齐读取在所涉及的硬件上都是隐式原子的。就其他线程而言，没有可中断的点，其中只有一半的指针已被读取。

Answer 1

（我正在放弃以前的答案。）

问题是您没有自动读取gHead和gHead->next的机制，但是实现无锁堆栈需要这种机制。由于您打算无论如何都忙循环来处理比较和交换冲突，因此可以使用自旋锁的等效项：

void lock_get () {
    while (!_sync_bool_compare_and_swap(&gGetLock, 0, 1)) {}
}

void unlock_get () {
    unlock_get_success = _sync_bool_compare_and_swap(&gGetLock, 1, 0);
    assert(unlock_get_success);
}

现在， test_get()中的循环可以被 lock_get()和 unlock_get()包围。 test_get()的CAS循环只是一个与 test_put()竞争的线程。 Jens对CAS循环的实现似乎更干净。

lock_get();
result = __sync_val_compare_and_swap(&gHead, 0, 0);
while ((oldval = result)) {
   result = __sync_val_compare_and_swap(&gHead, result, result->next);
   if (oldval == result) break;
}
unlock_get();

这实现了目的，即仅一个线程应该从头部弹出。