考虑下面的代码,这些代码使用非阻塞语义来弹出堆栈:
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
,因此逻辑是:这是否仍然可以免于等待†取决于您是否可以找到在中间状态下有用的操作。
与使用哨兵相比,有很多文章可供阅读,以更有效的方式进行阅读-实际上,您正在用一个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隐式锁定了总线,并且所花费的时间与要同步的缓存数量成正比。因此,您可以拥有不会旋转并等待的代码,但是您不会拥有不会锁定的代码。