c++ - C++中具有原子的乐观锁策略

在阅读了c++ 0x的原子并结合了非锁定队列之后，我决定尝试使用它们。

想法是编写带有乐观锁定的单个生产者，多个消费者队列。不需要消耗消息。跳过是完全可以的，只要消费者阅读它会读取最后一个版本或知道它的读取是错误的。

在我想到的策略下面的代码中失败了。数据被损坏，因为数据是无序写入的。任何关于为什么这样做以及如何解决它的指针将不胜感激。

在Linux上编译:g++ -std = c++ 0x -o代码code.cpp -lpthread

谢谢，
丹尼斯

//
// This features 2 threads in which the first writes to a structure
// and the second tries to read from that with an optimistic
// locking strategy. The data is equal to the versioning so we can
// see if the data is corrupt or not.
//
// @since: 2011-10-28
// @author: Dennis Fleurbaaij <[email protected]>
//

#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <stdatomic.h>
#include <sched.h>
#include <assert.h>
#include <iostream>
#include <xmmintrin.h>

struct structure_t
{
    std::atomic<unsigned int> id;
    unsigned int data_a;
    unsigned int data_b;

    char _pad[ 64 - 12 ];
};

#define NUM_STRUCTURES 2
struct structure_t structures[NUM_STRUCTURES];

std::atomic<size_t> current_version_index;

volatile bool start = false;
volatile bool run = true;
size_t const iter_count = 10000000;

/**
 * Write thread
 */
void* writer(void*)
{
    while(!start)
        sched_yield();

    size_t i;
    for( i=0 ; i<iter_count ; ++i )
    {
        size_t index = current_version_index.load(std::memory_order_relaxed);
        size_t next_index = ( current_version_index + 1 ) & NUM_STRUCTURES-1;

        structures[next_index].data_a = i;
        structures[next_index].data_b = i;

        structures[next_index].id.store(i, std::memory_order_release);

        current_version_index.store(next_index);

        //std::cout << "Queued - id: " << i << ", index: " << next_index << std::endl;
        //sleep(1);
    }

    run=false;
}

/**
 * Read thread
 */
void* reader(void*)
{
    while(!start)
        sched_yield();

    unsigned int prev_id=0;

    size_t i;
    while(run)
    {
        size_t index = current_version_index.load(std::memory_order_relaxed);
        unsigned int id = structures[index].id.load(std::memory_order_acquire);

        if( id > prev_id )
        {
            unsigned int data_a = structures[index].data_a;
            unsigned int data_b = structures[index].data_b;

            // Re-read the data and check optimistic lock. This should be read after
            // the lines above and should not be optimized away.
            //
            // This is what fails after a while:
            // Error in data. Index: 0, id: 24097, id2: 24097, data_a: 24099, data_b: 24099
            unsigned int id2 = structures[index].id.load(std::memory_order_acquire);
            if( id2 > id )
            {
                continue;
            }

            if( id != id2 ||
                id != data_a ||
                id != data_b )
            {
                std::cerr << "Error in data. Index: " << index << ", id: " << id
                          << ", id2: " << id2 << ", data_a: " << data_a << ", data_b: " << data_b << std::endl;

                exit(EXIT_FAILURE);
            }

            //std::cout << "Read. Index: " << index << ", id: " << id
            //              << ", data_a: " << data_a << ", data_b: " << data_b << std::endl;

            prev_id = id;
        }

        _mm_pause();
    }
}

/**
 * Main
 */
int main (int argc, char *argv[])
{
    assert( sizeof(structure_t) == 64 );

    pthread_t write_thread, read_thread;
    pthread_create(&write_thread, NULL, writer, (void*)NULL);
    pthread_create(&read_thread, NULL, reader, (void*)NULL);

    sleep(1);

    start = 1;

    void *status;
    pthread_join(read_thread, &status);
    pthread_join(write_thread, &status);
}

最佳答案

也许，这是一个错误:

    structures[next_index].data_a = i;
    structures[next_index].data_b = i;

// **** The writer may be interrupted (preempted) here for a long time ***
// at the same time the reader reads new data but old id (number of reads doesn't matter)

    structures[next_index].id.store(i, std::memory_order_release); // too late!

(current_version_index可能是读者从之前的步骤中获取的，因此竞争条件实际上是可能的)

关于c++ - C++中具有原子的乐观锁策略，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/7926829/