代码之家 › 专栏 › 技术社区 › Fanatic23

无阻塞堆栈弹出

nonblocking multithreading c++

Fanatic23 · 技术社区 · 14 年前

考虑以下使用非阻塞语义弹出堆栈的代码:

T Stack<T>::pop( ) 
{ 
    while (1) { 
        if (top == NULL) 
           throw std::string(âCannot pop from empty stackâ);
        Node* result = top;
        if (top && __sync_bool_compare_and_swap(&top, result, top->next)) { 
            return result->data;
        }
    }
}

我担心的是,如果执行pop的线程在第二个if语句之前被调度,并且在返回其时间片时堆栈是空的,那么我的check in 2nd循环是否足以避免崩溃?当然,在最坏的情况下,在比较top和zero之后,线程可能会被取消调度。

任何意见都可以。我知道ABA问题也可能发生。

5 回复 | 直到 14 年前

Pete Kirkham 14 年前

首先,假设 top 是不稳定的,任何时候都可以被另一个线程更改,每个循环只取一次它的值,这样就不会从下面拉出地毯:

T Stack<T>::pop( )
{
    while ( 1 ) {
        Node* result = top;
        if ( result == NULL )
            throw std::string ( âCannot pop from empty stackâ );
        // you now know result isn't NULL here 
        if ( __sync_bool_compare_and_swap ( &top, result, result -> next ) ) {
            return result -> data;
        }
    }
}

这仍然不能解决 result 在获得 顶部 的值并取消对它的引用。

你想用安全哨兵代替 result -> next ,所以逻辑是:

如果top为空,则队列为空
如果top是sentinel,那么有其他东西正在从堆栈中拉出来,去做其他有用的事情。
如果两者都不是,则将sentinel放在顶部,从result中获取值,将result->next放在顶部,删除result。

这是否仍算作“无等待”取决于您能否在中间状态下找到有用的操作。

有很多论文要阅读,比使用哨兵更有效-实际上,你是在用一个CAS模拟一个两个单词的CAS,因为你需要检查 结果 以及 顶部 . 这些太复杂了,在这里无法再现。

未以任何方式测试:

bool Stack<T>::pop ( T&out )
{
    static const Node* const empty ( 0 );
    static const Node* const sentinel ( empty + 1 );

    while ( true ) {
        Node* result = top;

        if ( result == empty )
            throw std::string ( âCannot pop from empty stackâ );

        if ( result == sentinel )
            // something else is popping, return false and allow 
            // current thread to do some work before retrying
            return false;

        if ( __sync_bool_compare_and_swap ( &top, result, sentinel ) ) {
            // only one thread can CAS from a given result to something, 
            // so we are the only thread which has this value of result
            // hence we can dereference it and delete it/return it to a pool
            //
            // no other thread will change top if top == sentinel, so this
            // CAS can't fail
            if ( !__sync_bool_compare_and_swap ( &top, sentinel, result -> next ))
                throw std::string ( "nobody's perfect" );

            out = result -> data;
            delete result;
            return true;
        }
    }
}

由于一次只能在一个线程中检查或变异结果的指针,因此应该是安全的(我以前没有使用过这种确切的模式,通常在设计一些东西的几天后,我会想到一些奇怪的情况)。这是否比用pthread_mutex_trylock包装std::deque更好,值得衡量。

当然,不管是这个还是原始线程都不是非阻塞的——如果一个线程一直从堆栈中拉出,那么任何其他线程都将无限期地循环,等待CAS成功。不太可能,如果CAS确实失败,则通过返回false很容易删除,但是如果线程不应该等待,则必须确定希望线程做什么。如果在某个东西可以退出队列之前旋转是可以的,则不需要返回代码。

_{我主要在x86/x64上工作,那里没有无锁代码,因为CMPXCHG隐式地锁定总线,并且需要与要同步的缓存数成比例的时间。所以你可以有不旋转和等待的代码,但你不能有不锁定的代码。}

SingleNegationElimination 14 年前

不如稍微改变一下,这样就可以抓住堆栈顶部的内容,如果发现堆栈顶部是空的,则引发异常。

valdo 14 年前

正确的关注。

此外,您还忘记了内存重新排序和缓存的可能性。例如,您的线程可能仍然看到 top ,即使其他线程已设置为 NULL .

Clifford 14 年前

在Dobb博士的论文中,无锁数据结构是一个很流行的话题。仍然可以找到这些物品 here .

Timo 14 年前

我不明白这怎么可能。我看到的一个问题是当你取消引用 top 在里面 top->next ,该内存位置可能不再有效。其他一些线程可能已经弹出堆栈并删除或修改了项。