无锁编程(四) - CAS与ABA问题

CAS

一般采用原子级的read-modify-write原语来实现Lock-Free算法，其中LL和SC是Lock-Free理论研究领域的理想原语，但实现这些原语需要CPU指令的支持，非常遗憾的是目前没有任何CPU直接实现了SC原语。根据此理论，业界在原子操作的基础上提出了著名的CAS（Compare-And-Swap）操作来实现Lock-Free算法，Intel实现了一条类似该操作的指令：cmpxchg8。

CAS原语负责将某处内存地址的值（1个字节）与一个期望值进行比较，如果相等，则将该内存地址处的值替换为新值，CAS 操作伪码描述如下：

Bool CAS(T* addr, T expected, T newValue) 
{ 
         if(*addr == expected ) 
         {
                   *addr=  newValue; 
                   returntrue; 
         }
         else
                   returnfalse; 
}

CAS实际操作

{

备份旧数据；

基于旧数据构造新数据；

}while(!CAS(内存地址，备份的旧数据，新数据))

就是指当两者进行比较时，如果相等，则证明共享数据没有被修改，替换成新值，然后继续往下运行；如果不相等，说明共享数据已经被修改，放弃已经所做的操作，然后重新执行刚才的操作。容易看出CAS操作是基于共享数据不会被修改的假设，采用了类似于数据库的commit-retry的模式。当同步冲突出现的机会很少时，这种假设能带来较大的性能提升。

CAS的Linux解法

cmpxchg先比较内存地址的值是否与传入的值相等，如果相等则执行xchg逻辑。

inline int CAS(unsigned long* mem, unsignedlong newval, unsigned long oldval)
{
         __typeof(*mem) ret;
         //这里测试的使用64位系统，如果是32位，这里使用cmpschgl
         __asm__volatile ("lock; cmpxchgq %2,%1"
                                                        :"=a"(ret), "=m"(*mem)
                                                        :"r"(newval), "m"(*mem), "0"(oldval));
         returnret==oldval;
}

CAS举例（简单应用AtomicInc）

#include <stdio.h>
#include <pthread.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/time.h>
#include <stdint.h>

int count = 0;

inline int CAS(unsigned long* mem, unsigned long oldval, unsigned long newval)
{
	__typeof (*mem) ret;
	// 这里测试的使用64位系统，如果是32位，这里使用cmpschgl
	__asm __volatile ("lock; cmpxchgq %2,%1"
						: "=a"(ret), "=m"(*mem)
						: "r"(newval), "m"(*mem), "0"(oldval));
	return ret==oldval;
}

void AtomicInc(int* addr)
{
	int oldval;
	int newval;
	do
	{
		oldval = *addr;
		newval = oldval+1;
	} while(!CAS((unsigned long*)addr, oldval, newval));
}

void *test_func(void *arg)
{
	int i=0;
	int confict = 0;
	for(i=0;i<2000000;++i)
	{
		AtomicInc(&count);
	}
	return NULL;
}

int main(int argc, const char *argv[])
{
	pthread_t id[20];
	int i = 0;

	uint64_t usetime;
	struct timeval start;
	struct timeval end;
	
	gettimeofday(&start,NULL);
	
	for(i=0;i<20;++i)
	{
		pthread_create(&id[i],NULL,test_func,NULL);
	}

	for(i=0;i<20;++i)
	{
		pthread_join(id[i],NULL);
	}
	
	gettimeofday(&end,NULL);

	usetime = (end.tv_sec-start.tv_sec)*1000000+(end.tv_usec-start.tv_usec);
	printf("count = %d, usetime = %lu usecsn", count, usetime);
	return 0;
}

CAS举例（复杂应用）

struct Node
{
	Node* next;
	int data;
}
Node* head = NULL;

void push(int t)
{
	Node* node = new Node(t);
	do
	{
		node->next = head;
	} while (!CAS(&head, node->next, node));
}

bool pop(int&t )
{
	Node* current = head;
	while(current)
	{
		if (CAS(&head, current, current->next)) // ABA问题
		{
			t = current->data;
			return true;
		}
		current = head;
	}
	return false;
}

ABA问题

一般的CAS在决定是否要修改某个变量时，会判断一下当前值跟旧值是否相等。如果相等，则认为变量未被其他线程修改，可以改。但是，“相等”并不真的意味着“未被修改”。另一个线程可能会把变量的值从A改成B，又从B改回成A。这就是ABA问题。很多情况下，ABA问题不会影响你的业务逻辑因此可以忽略。但有时不能忽略，这时要解决这个问题，一般的做法是给变量关联一个只能递增、不能递减的版本号。在compare时不但compare变量值，还要再compare一下版本号。 Java里的AtomicStampedReference类就是干这个的。