2

私は次の場所でロックフリーキューのサンプルコードを見ていました。

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

（ C ++での本番環境対応のロックフリーキューやハッシュ実装はありますかなど、多くのSOの質問でも参照してください）

コードには多くのタイプミスがありますが、これは単一のプロデューサー/コンシューマーで機能するはずです。以下に示すようにコードを更新して読み取りましたが、クラッシュします。誰かがなぜ提案がありますか？

特に、分割して最後に次のように宣言する必要があります。

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 に答える