无锁编程(二) - 原子操作
什么是原子操作
原子操作可以保证指令以原子的方式执行——执行过程不被打断，原子操作是多数无锁编程的基本前提。
原子操作分为以下几类
对1字节的读写
对2字节数（对齐到16位边界）读写
对4字节数（对齐到32位边界）读写
对8字节数（对齐到64位边界）读写
xchg
原子操作基本原理
在x86平台上，CPU提供了在指令执行期间对总线加锁的手段。CPU芯片上有一条引线#HLOCK pin，如果汇编语言的程序中在一条指令前面加上前缀"LOCK"，经过汇编以后的机器代码就使CPU在执行这条指令的时候把#HLOCK pin的电位拉低，持续到这条指令结束时放开，从而把总线锁住，这样同一总线上别的CPU就暂时不能通过总线访问内存了，保证了这条指令在多处理器环境中的原子性。
LOCK是一个指令的描述符，表示后续的指令在执行的时候，在内存总线上加锁。总线锁会导致其他几个核在一定时钟周期内无法访问内存。虽然总线锁会影响其他核的性能，但比起操作系统级别的锁，已经轻量太多了。
#lock是锁FSB(前端串行总线，front serial bus)，FSB是处理器和RAM之间的总线，锁住了它，就能阻止其他处理器或core从RAM获取数据。
内核提供atomic_*系列原子操作
声明和定义：
void atomic_set(atomic_t *v, int i);
atomic_t v = ATOMIC_INIT(0);读写操作：
int atomic_read(atomic_t *v);
void atomic_add(int i, atomic_t *v);
void atomic_sub(int i, atomic_t *v);加一减一：
void atomic_inc(atomic_t *v);
void atomic_dec(atomic_t *v);执行操作并且测试结果：执行操作之后，如果v是0，那么返回1，否则返回0
int atomic_inc_and_test(atomic_t *v);
int atomic_dec_and_test(atomic_t *v);
int atomic_sub_and_test(int i, atomic_t *v);
int atomic_add_negative(int i, atomic_t *v);
int atomic_add_return(int i, atomic_t *v);
int atomic_sub_return(int i, atomic_t *v);
int atomic_inc_return(atomic_t *v);
int atomic_dec_return(atomic_t *v);gcc内置__sync_*系列built-in函数
gcc内置的__sync_*函数提供了加减和逻辑运算的原子操作，__sync_fetch_and_add系列一共有十二个函数，有加/减/与/或/异或/等函数的原子性操作函数,__sync_fetch_and_add,顾名思义，先fetch，然后自加，返回的是自加以前的值。以count = 4为例，调用__sync_fetch_and_add(&count,1),之后，返回值是4，然后，count变成了5. 有__sync_fetch_and_add,自然也就有__sync_add_and_fetch，先自加，再返回。这两个的关系与i++和++i的关系是一样的。
type可以是1,2,4或8字节长度的int类型，即：
int8_t / uint8_t
int16_t / uint16_t
int32_t / uint32_t
int64_t / uint64_t
type __sync_fetch_and_add (type *ptr, typevalue);
type __sync_fetch_and_sub (type *ptr, type value);
type __sync_fetch_and_or (type *ptr, type value);
type __sync_fetch_and_and (type *ptr, type value);
type __sync_fetch_and_xor (type *ptr, type value);
type __sync_fetch_and_nand(type *ptr, type value);
type __sync_add_and_fetch (type *ptr, typevalue);
type __sync_sub_and_fetch (type *ptr, type value);
type __sync_or_and_fetch (type *ptr, type value);
type __sync_and_and_fetch (type *ptr, type value);
type __sync_xor_and_fetch (type *ptr, type value);
type __sync_nand_and_fetch (type *ptr, type value);代码讲解1：使用__sync_fetch_and_add操作全局变量
<strong>#include <stdio.h>
#include <pthread.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/time.h>
#include <stdint.h>
int count = 0;
void *test_func(void *arg)
{
int i=0;
for(i=0;i<2000000;++i)
{
__sync_fetch_and_add(&count,1);
}
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 usecs\n", count, usetime);
return 0;
}
</strong>代码讲解2：使用互斥锁mutex操作全局变量
<strong>#include <stdio.h>
#include <pthread.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/time.h>
#include <stdint.h>
int count = 0;
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
void *test_func(void *arg)
{
int i=0;
for(i=0;i<2000000;++i)
{
pthread_mutex_lock(&mutex);
++count;
pthread_mutex_unlock(&mutex);
}
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 usecs\n", count, usetime);
return 0;
}
</strong>结果说明：
[root@rocket lock-free]#./atom_add_gcc_buildin
count = 40000000, usetime = 756694 usecs
[root@rocket lock-free]# ./atom_add_mutex
count = 40000000, usetime = 3247131 usecs
可以看到，使用原子操作是使用互斥锁性能的5倍左右，随着冲突数量的增加，性能差距会进一步拉开。Alexander Sandler实测，原子操作性能大概是互斥锁的6-7倍左右。
有兴趣的同学请参考：
http://www.alexonlinux.com/multithreaded-simple-data-type-access-and-atomic-variables
xchg指令
xchg(ptr, new) 将ptr指向的值置为new，返回交换前的值。
cmpxchg(ptr, old, new) 比较当前值如果跟old相同，则将ptr指向的值置为new，否则不变，返回交换前的值。根据比较返回值是否和old一样来判断是否成功。
int fetch_and_add(int* i, int value, int* confict)
{
int old_value;
int new_value;
int v;
do
{
old_value = *i;
new_value = old_value + 1;
v = cmpxchg(i, old_value, new_value);
(*confict)++;
} while (old_value != v);
}版权声明：本文为博主原创文章，未经博主允许不得转载。