2

我正在查看无锁队列的示例代码：

http://drdobbs.com/high-performance-computing/210604448?pgno=2

（也在许多 SO 问题中参考，例如Is there a production ready lock-free queue or hash implementation in C++）

这看起来应该适用于单个生产者/消费者，尽管代码中有许多拼写错误。我已将代码更新为如下所示，但它让我崩溃了。任何人有建议为什么？

特别是，应该将divider 和last 声明为：

atomic<Node *> divider, last;    // shared

我在这台机器上没有支持 C++0x 的编译器，所以也许这就是我所需要的......

// Implementation from http://drdobbs.com/high-performance-computing/210604448
// Note that the code in that article (10/26/11) is broken.
// The attempted fixed version is below.

template <typename T>
class LockFreeQueue {
private:
    struct Node {
        Node( T val ) : value(val), next(0) { }
        T value;
        Node* next;
    };
    Node *first,      // for producer only
    *divider, *last;    // shared
public:
    LockFreeQueue()
    {
        first = divider = last = new Node(T()); // add dummy separator
    }
    ~LockFreeQueue()
    {
        while( first != 0 )    // release the list
        {
            Node* tmp = first;
            first = tmp->next;
            delete tmp;
        }
    }
    void Produce( const T& t )
    {
        last->next = new Node(t);    // add the new item
        last = last->next;      // publish it

        while (first != divider) // trim unused nodes
        {
            Node* tmp = first;
            first = first->next;
            delete tmp;
        }
    }
    bool Consume( T& result )
    {
        if (divider != last)         // if queue is nonempty
        {
            result = divider->next->value; // C: copy it back
            divider = divider->next;      // D: publish that we took it
            return true;                  // and report success
        }
        return false;                   // else report empty
    }
};

我编写了以下代码来测试这一点。Main（未显示）仅调用 TestQ()。

#include "LockFreeQueue.h"

const int numThreads = 1;
std::vector<LockFreeQueue<int> > q(numThreads);

void *Solver(void *whichID)
{
    int id = (long)whichID;
    printf("Thread %d initialized\n", id);
    int result = 0;
    do {
        if (q[id].Consume(result))
        {
            int y = 0;
            for (int x = 0; x < result; x++)
            { y++; }
            y = 0;
        }
    } while (result != -1);
    return 0;
}


void TestQ()
{
    std::vector<pthread_t> threads;
    for (int x = 0; x < numThreads; x++)
    {
        pthread_t thread;
        pthread_create(&thread, NULL, Solver, (void *)x);
        threads.push_back(thread);
    }
    for (int y = 0; y < 1000000; y++)
    {
        for (unsigned int x = 0; x < threads.size(); x++)
        {
            q[x].Produce(y);
        }
    }
    for (unsigned int x = 0; x < threads.size(); x++)
    {
        q[x].Produce(-1);
    }
    for (unsigned int x = 0; x < threads.size(); x++)
        pthread_join(threads[x], 0);    
}

更新：最终崩溃是由队列声明引起的：

std::vector<LockFreeQueue<int> > q(numThreads);

当我将其更改为简单数组时，它运行良好。（我实现了一个带锁的版本，它也崩溃了。）我看到析构函数在构造函数之后立即被调用，导致内存双重释放。但是，有谁知道为什么会立即使用 std::vector 调用析构函数？

4

3 回答 3

1

正如您所注意到的，您需要创建几个指针 std::atomic，并且您需要在循环中使用 compare_exchange_weak 以原子方式更新它们。否则，多个消费者可能会消费同一个节点，并且多个生产者可能会破坏列表。

于 2011-10-26T23:19:37.337 回答
1

这些写入（只是代码中的一个示例）按顺序发生是至关重要的：

last->next = new Node(t);    // add the new item
last = last->next;      // publish it

C++ 不能保证这一点——优化器可以按照它喜欢的方式重新排列事物，只要当前线程总是表现得好像程序完全按照你编写的方式运行。然后 CPU 缓存可以出现并进一步重新排序。

你需要记忆栅栏。使指针使用原子类型应该具有这种效果。

于 2011-10-27T01:23:27.470 回答
0

这可能完全不合时宜，但我不禁想知道您是否有某种与静态初始化相关的问题......为了笑，尝试声明q为指向无锁队列向量的指针并分配它在堆中main()。

于 2011-10-27T06:27:58.720 回答