C语言的原子操做

时间 2019-12-24

标签 c语言原子繁體版

原文原文链接

###gcc内建函数内建gcc在4.0.1版本后就经过其内建函数支持原子操做。在这以前编程真必需要经过参考各类cpu的指令操做手册，用其汇编指令编写原子操做。而gcc经过内建函数屏蔽了这些差别。gcc支持以下原子操做：编程

#if (GCC_VERSION >= 40100)
/* 内存访问栅 */
  #define barrier()             	(__sync_synchronize())
/* 原子获取 */
  #define AO_GET(ptr)       		({ __typeof__(*(ptr)) volatile *_val = (ptr); barrier(); (*_val); })
/*原子设置，若是原值和新值不同则设置*/
  #define AO_SET(ptr, value)        ((void)__sync_lock_test_and_set((ptr), (value)))
/* 原子交换，若是被设置，则返回旧值，不然返回设置值 */
  #define AO_SWAP(ptr, value)       ((__typeof__(*(ptr)))__sync_lock_test_and_set((ptr), (value)))
/* 原子比较交换，若是当前值等于旧值，则新值被设置，返回旧值，不然返回新值*/
  #define AO_CAS(ptr, comp, value)  ((__typeof__(*(ptr)))__sync_val_compare_and_swap((ptr), (comp), (value)))
/* 原子比较交换，若是当前值等于旧指，则新值被设置，返回真值，不然返回假 */
  #define AO_CASB(ptr, comp, value) (__sync_bool_compare_and_swap((ptr), (comp), (value)) != 0 ? true : false)
/* 原子清零 */
  #define AO_CLEAR(ptr)             ((void)__sync_lock_release((ptr)))
/* 经过值与旧值进行算术与位操做，返回新值 */
  #define AO_ADD_F(ptr, value)      ((__typeof__(*(ptr)))__sync_add_and_fetch((ptr), (value)))
  #define AO_SUB_F(ptr, value)      ((__typeof__(*(ptr)))__sync_sub_and_fetch((ptr), (value)))
  #define AO_OR_F(ptr, value)       ((__typeof__(*(ptr)))__sync_or_and_fetch((ptr), (value)))
  #define AO_AND_F(ptr, value)      ((__typeof__(*(ptr)))__sync_and_and_fetch((ptr), (value)))
  #define AO_XOR_F(ptr, value)      ((__typeof__(*(ptr)))__sync_xor_and_fetch((ptr), (value)))
/* 经过值与旧值进行算术与位操做，返回旧值 */
  #define AO_F_ADD(ptr, value)      ((__typeof__(*(ptr)))__sync_fetch_and_add((ptr), (value)))
  #define AO_F_SUB(ptr, value)      ((__typeof__(*(ptr)))__sync_fetch_and_sub((ptr), (value)))
  #define AO_F_OR(ptr, value)       ((__typeof__(*(ptr)))__sync_fetch_and_or((ptr), (value)))
  #define AO_F_AND(ptr, value)      ((__typeof__(*(ptr)))__sync_fetch_and_and((ptr), (value)))
  #define AO_F_XOR(ptr, value)      ((__typeof__(*(ptr)))__sync_fetch_and_xor((ptr), (value)))
#else
  #error "can not supported atomic operation by gcc(v4.0.0+) buildin function."
#endif	/* if (GCC_VERSION >= 40100) */
/* 忽略返回值，算术和位操做 */
#define AO_INC(ptr)                 ((void)AO_ADD_F((ptr), 1))
#define AO_DEC(ptr)                 ((void)AO_SUB_F((ptr), 1))
#define AO_ADD(ptr, val)            ((void)AO_ADD_F((ptr), (val)))
#define AO_SUB(ptr, val)            ((void)AO_SUB_F((ptr), (val)))
#define AO_OR(ptr, val)			 ((void)AO_OR_F((ptr), (val)))
#define AO_AND(ptr, val)			((void)AO_AND_F((ptr), (val)))
#define AO_XOR(ptr, val)			((void)AO_XOR_F((ptr), (val)))
/* 经过掩码，设置某个位为1，并返还新的值 */
#define AO_BIT_ON(ptr, mask)        AO_OR_F((ptr), (mask))
/* 经过掩码，设置某个位为0，并返还新的值 */
#define AO_BIT_OFF(ptr, mask)       AO_AND_F((ptr), ~(mask))
/* 经过掩码，交换某个位，1变0，0变1，并返还新的值 */
#define AO_BIT_XCHG(ptr, mask)      AO_XOR_F((ptr), (mask))

###普通汇编指令以加法指令操做实现 x = x + n为例，gcc编译出来的汇编形式上以下：多线程

...
movl 0xc(%ebp), %eax
addl $n, %eax
movl %eax, 0xc(%ebp)
...

能够看出，实现这条c语句，须要先将x所在内存0xc(%ebp)中的值装载到寄存器％eax中，而后用addl指令进行与一个当即数$n进行加操做，以后再寄存器中的结果装载回原内存中。若是在时序上又另外一个线程也操做该内存中的值，且在指令addl $n, %eax完成以后，时间片切换到了另外一个线程中，该线程进行了该内存的修改操做，并且还会在后续的操做中使用，这个时候发生又发生时间片切换，切回到原线程中，进行movl %eax, 0xc(%ebp)指令覆盖了前一个线程修改内容，若是在这时再切换到另外一个线程中，该线程就会使用到一个错误的值进行后续的操做。 ###gcc原子汇编指令仍然以加法指令操做实现 x = x + n为例，gcc编译出来的原子汇编形式上以下：函数

...
mov    $0x1,%eax
lock   xadd %eax,-0x4(%rbp)
mov    %eax,-0x4(%rbp)
...

gcc的原子操做是内建函数经过汇编实现的，统一命名以__sync_xxx()起头，原子操做作了什么事情呢？原子操做的原理都是经过汇编指令lock在各类xadd、cmpxchg或xchg指令前进行锁定操做内存的总线，并将上述的普通3条指令的操做合并为一条操做，由于内存与cpu都是经过总线进行数据交换，因此即便其它cpu核也同时（真正意义上的多线程，而不是单核上的时间片切换）要对该内存的存取，也要等待。（由于我不是低层开发人员，因此具体时序和动做我不是太了解，只能以应用层的锁动做理解这里的总线锁，若是你了解，请更正），而被锁总线的单核应该不会进行时间片切换，直到该指令完成。 ###优化带来语句倒置除了多线程操做同一个内存时会发生数据的一致性错误，由于编译器的优化问题也会形成数据一致性问题。若是你的原意要进行以下的操做：fetch

int a = 0;
int b = 0;
void A() {
    a = 1; 
    b = 2;
}
void B() {
    if (b > 0)
        printf("a :%d\n", a);
}

那么通过编译器的优化，A()中的两条复制语句可能被调换顺序，若是两个线程分别同时执行A()和B()，那么由于这个缘由，B()可能输出1，也可能输出0；解决方法是让a = 1必定在b = 2执行，那么在二者之间插入内存栅栏__sync_synchronize()能够保证前后次序。（由于我对这样的优化发生状况不是很明了，故这里不能详细的描述这样的优化对同线程产生的影响） ###volatile关键字与原子原子操做的内存，要保证其内容已定是存取最新的，而不是cache中的数据，因此要用volatile关键字代表，这样每次存取cpu直接存取内存，而非cache中的数据，咱们定义一个原子类型：优化

#ifndef AO_T
typedef volatile long AO_T;
#endif

##原子操做与普通C语句的等效操做

这里用上面定义的宏说明原子操做，等效的C语言非原子的操做为了保证一致性，咱们使用lock()和 unlock这个伪语句表示锁的加锁和解锁。固然原子操做要比应用层加锁快了太多太多。ui

内存栅栏使用

int a = 0;
barrier();
int b = 2;

保证a的复制在b的复制前执行atom

原子获取

int a = 5;
int b = AO_GET(&a); //b==5;

int a = 5;
lock();
int b = a; 
unlock();

保证读取a的值是内存中的值，而不是寄存器或cache中的值 ###原子设置线程

int a = 0;
AO_SET(&a, 10); //a==10;

int a = 0;
lock();
a = 10;
unlock();

###原子交换设计

int a = 10;
AO_SWAP(&a, 9);

int a = 10;
lock();
if (a != 9)
    a = 9;
unlock();

###原子比较交换code

int a = 10;
int b = AO_CAS(&a, 10, 9); //b==10, a==9;
int c = AO_CAS(&a, 9, 8); //c==8, a==10;

int a = 10;
int b = 0;
int c = 0;
lock();
if (a == 10) {
    b = a;
    a = 9;
} else {
    b = 10;
}
unlock();
lock();
if (a == 9) {
    b = a;
    a = 8;
} else {
    b = 9;
}
unlock();

AO_CASB()的逻辑与AO_CAS()一致，只是返还一个真假值判断是否发生了交换，就再也不赘诉了。 ###原子清零

int a = 10;
AO_CLEAR(&a); //a==0;

int a = 10;
lock();
a = 0;
unlock();

###先操做后使用的加减运算和逻辑运算

先加一个数，再使用和值 AO_xxx_F()中的F表示fetch提取的意思

int a = 1;
int b = AO_ADD_F(&a, 10);//a==11, b==11

int a = 1;
int b = 0;
lock();
a += 10;
b = a;
unlock();

其它的运算（减，或，与，异或）与加法操做逻辑同样，就再也不赘诉了 ###先使用后操做的加减运算与逻辑运算
使用原值，后加上一个数

int a = 1;
int b = AO_F_ADD(&a, 10);//a==11, b==1

int a = 1;
int b = 0;
lock();
b = a;
a += 10;
unlock();

##什么时候使用原子操做最合适

原子操做最合适用来管理状态，并且最好是程序发现状态不符合本身要求是，能够忽略这个错误，继续运行，或稍后在此尝试。好比咱们使用一个local static变量存储当前系统有多少个cpu核，以备给出一些策略，好比之后咱们要实现的自旋锁中的休眠。代码以下：

long GetCPUCores()
{
    static long g_CPUCores = 0;
    long gcpus = -1;
    /*原子获取，若是没有设置过，则继续，不然返回这个值*/
    if (likely((gcpus = AO_GET(&g_CPUCores)) != -1)) {
        return gcpus;
    }
    gcpus = sysconf(_SC_NPROCESSORS_CONF);
    if (unlikely(gcpus < 0)) {
        printf("Get number of CPU failure : %s", strerror(errno));
        abort();
    }
    /*原子设置*/
    AO_SET(&g_CPUCores, gcpus);
    return gcpus;
}

若是有多个线程同时调用，或单个线程屡次调用，咱们均可以保证g_CPUCores中数据的有效性，不会出现获取到一个大于0到假值致使后续的逻辑错误。并且这样的设计，还能够提升效率，若是获取的系统参数是一个像

#ifdef __APPLE__
	gtid = syscall(SYS_thread_selfid);
#else
	gtid = syscall(SYS_gettid);
#endif

的真正的系统调用，那么在结果固定的状况下，代价是昂贵的，由于程序必需要发起中断服务，切换到内核空间调用代码为SYS_thread_selfid 或SYS_gettid 的中段服务，从而获得线程ID（线程是一个轻量级的进程，只不过它的堆空间与其它线程共享，而不是进程那样是彼此独立的，我之后会在此细谈这个ID值的运用）。

##使用原子操做

改进上一篇文章中说起的结构魔数操做

上一节咱们说过，使用带魔数字段结构的函数经过判断、修改魔数作出相应的操做，试想若是两个线程同时操做魔数字段，确定会带来冲突，因此咱们将其对应的非原子操做，改成原子操做，代码以下：

/*
 * 魔数
 * 结构体中设置一个magic的成员变量，已检查结构体是否被正确初始化
 */
#if !defined(OBJMAGIC)
  #define OBJMAGIC (0xfedcba98)
#endif

/*原子的设置魔数*/
#undef REFOBJ
#define REFOBJ(obj)						     \
	({							     \
		int _old = 0;					     \
		bool _ret = false;				     \
		if (likely((obj))) {				     \
			_old = AO_SWAP(&(obj)->magic, OBJMAGIC); \
		}						     \
		_ret = (_old == OBJMAGIC ? false : true);	     \
		_ret;						     \
	})

/*原子的重置魔数*/
#undef UNREFOBJ
#define UNREFOBJ(obj)							\
	({								\
		bool _ret = false;					\
		if (likely((obj))) {					\
			_ret = AO_CASB(&(obj)->magic, OBJMAGIC, 0);	\
		}							\
		_ret;							\
	})

/*原子的验证魔数*/
#undef ISOBJ
#define ISOBJ(obj) ((obj) && AO_GET(&(obj)->magic) == OBJMAGIC)

/*断言魔数*/
#undef ASSERTOBJ
#define ASSERTOBJ(obj) (assert(ISOBJ((obj))))

其实这样的运用也不能100％的保证多线程下数据的一致性，好比两个线程Ａ和Ｂ，同时在操做一个结构体Ｔ： ###原子操做

初始化操做，initT():

void initT(T *t) {
    REFOBJ(t);
    //other initial operate
}

处理数据操做，dealT():

void dealT(T *t) {
    if(!ISOBJ(t)) return;
    //other deal operate
}

销毁数据，destroyT():

void destroy(T *t) {
    if(!UNREF(t)) return;
    //other destroy
}

###原子操做与时序

考虑下列时序：

A刚完成initT ()中的REFOBJ ()语句，将要真正的初始化T的其它的字段，这时切换到B；
B也调用initT ()中的REFOBJ ()语句，发现结构体的初始化标志已经设置了，则返回并切换到A；
A开始真正的初始化相关字段，在未处理完成时有切回了B；
B调用dealT()开始处理其它的字段了，结果固然是全处理的是脏数据。

再考虑下列时序：

A和B都使用T交错的操做了T一段时间，A和B都想销毁T持有的数据而调用destroy()；
假设A先进入destroy()，而后在UNREF ()调用以前切换到B；
B进入destroy(), 并成功的调用UNREF ()；
在后续的操做中，不论什么时候发生切换都不会形成数据重复销毁。

上面状况出现的根本缘由就是原子操做不原子，由于你企图使用这样的原子操做，进行非原子的多步骤的字段初始化操做，这是不会成功的。因此在你使用原子操做时，必定要考虑线程切换带来的时序问题和你的原子操做能不能使你的操做原子的进行。

##下一步咱们作什么

咱们将使用原子操做实现一个原子锁，并说明什么状况下应该使用原子锁，什么状况下不该该使用原子锁。敬请期待哦。