考虑下面的代码,这些代码使用非阻塞语义来弹出堆栈:

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;
        }
    }
}

我担心的是,如果执行弹出的线程恰好在第二条if语句之前被调度了,并且在返回其时间片之前,堆栈的空了,那么我在第二个循环中的检查是否足以避免崩溃?当然,在最坏的情况下,仅在将top与零进行比较之后,该线程可能会被调度。

任何意见表示赞赏。我知道可能还会发生的ABA问题。

最佳答案

首先,假设top是 volatile 的,并且可以在任何时候由另一个线程更改,则每个循环仅获取一次它的值,这样就不会把地毯从下面拉出来:

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的值与取消引用之间的top被删除或修改的问题。

您想使用安全的哨兵而不是result -> next,因此逻辑是:
  • 如果top为null,则队列为空
  • 如果top是哨兵,那么还有其他事情正在酝酿中,请执行其他有用的事情。
  • 如果top都不是,则将哨兵放在顶部,从结果中获取值,将result-> next放在顶部,删除结果。

  • 这是否仍然可以免于等待†取决于您是否可以找到在中间状态下有用的操作。

    与使用哨兵相比,有很多文章可供阅读,以更有效的方式进行阅读-实际上,您正在用一个CAS模拟两个单词的CAS,因为您需要检查有关result的状态以及top的状态。这些太复杂了,无法在此处复制。

    未经任何方式测试:
    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隐式锁定了总线,并且所花费的时间与要同步的缓存数量成正比。因此,您可以拥有不会旋转并等待的代码,但是您不会拥有不会锁定的代码。

    09-10 09:22
    查看更多