__sync_fetch_and_add

本文轉載自查看原文 2016-07-19 17:27 4699 c++

最近在公司離職的前輩寫的代碼哪里看到了__sync_fetch_and_add這個東東.比較好奇.找些資料學習學習

http://www.lxway.com/4091061956.htm

http://www.cnblogs.com/FrankTan/archive/2010/12/11/1903377.html

可使用的環境: gcc.version > 4.1.2

作用:提供多線程下變量的加減和邏輯運算的原子操作

正文如下:

最近編碼需要實現多線程環境下的計數器操作，統計相關事件的次數。下面是一些學習心得和體會。不敢妄稱原創，基本是學習筆記。遇到相關的引用，我會致謝。
    當然我們知道，count++這種操作不是原子的。一個自加操作，本質是分成三步的：
     1 從緩存取到寄存器
     2 在寄存器加1
     3 存入緩存。

mov eax,dword ptr [a]

add eax,1

mov dword ptr [a],eax

由於時序的因素，多個線程操作同一個全局變量，會出現問題。這也是並發編程的難點。在目前多核條件下，這種困境會越來越彰顯出來。
最簡單的處理辦法就是加鎖保護，這也是我最初的解決方案。看下面的代碼：

1       pthread_mutex_t count_lock = PTHREAD_MUTEX_INITIALIZER;
2 
3       pthread_mutex_lock(&count_lock);
4       global_int++;
5       pthread_mutex_unlock(&count_lock);
linux 變量 : pthread_mutex_t   
linux 函數 : pthread_mutex_lock; pthread_mutex_unlock

后來在網上查找資料，找到了__sync_fetch_and_add系列的命令

__sync_fetch_and_add系列一共有十二個函數，有加/減/與/或/異或/等函數的原子性操作函數,

__snyc_fetch_and_add : 先fetch然后自加,返回的是自加以前的值
__snyc_add_and_fetch : 先自加然后返回,返回的是自加以后的值 (參照 ++i 和 i++)


__snyc_fetch_and_add的一個簡單使用
1 int count = 4;
2 __sync_fetch_and_add(&count, 1); // __sync_fetch_and_add(&count, 1) == 4
3 cout<<count<<endl; //--->count=5

對於多線程對全局變量進行自加，我們就再也不用理線程鎖了。

下面這行代碼，和上面被pthread_mutex保護的那行代碼作用是一樣的，而且也是線程安全的。

__sync_fetch_and_add( &global_int, 1 );

下面是這群函數的全家福，大家看名字就知道是這些函數是干啥的了。

 1 //在用gcc編譯的時候要加上選項 -march=i686
 2 type __sync_fetch_and_add (type *ptr, type value, ...);
 3 type __sync_fetch_and_sub (type *ptr, type value, ...);
 4 type __sync_fetch_and_or (type *ptr, type value, ...);
 5 type __sync_fetch_and_and (type *ptr, type value, ...);
 6 type __sync_fetch_and_xor (type *ptr, type value, ...);
 7 type __sync_fetch_and_nand (type *ptr, type value, ...);
 8 type __sync_add_and_fetch (type *ptr, type value, ...);
 9 type __sync_sub_and_fetch (type *ptr, type value, ...);
10 type __sync_or_and_fetch (type *ptr, type value, ...);
11 type __sync_and_and_fetch (type *ptr, type value, ...);
12 type __sync_xor_and_fetch (type *ptr, type value, ...);
13 type __sync_nand_and_fetch (type *ptr, type value, ...);

__sync_fetch_and_add,速度是線程鎖的6～7倍

type可以是1,2,3或者8字節長度的int類型,即

 1 int8_t    
 2 uint8_t
 3 
 4 int16_t
 5 uint16_t
 6 
 7 int32_t
 8 uint32_t
 9 
10 int64_t
11 uint64_t

后面的可擴展參數(...)用來指出哪些變量需要memory barrier,因為目前gcc實現的是full barrier（類似於linux kernel 中的mb(),表示這個操作之前的所有內存操作不會被重排序到這個操作之后）,所以可以略掉這個參數。

恩.再找個帖子學習學習.http://blog.csdn.net/hzhsan/article/details/25124901

有一個概念叫過無鎖化編程, 知道linux支持的哪些操作是具有原子特性的是理解和設計無鎖化編程算法的基礎

除了上面提到的12個外還有4個可以實現互斥鎖的功能

//以下兩個函數提供原子的比較和交換, 如果*ptr = oldValue, 就將newValue寫入*ptr
//第一個函數在相等並寫入的情況下返回true
//第二個函數返回操作之前的值

bool __sync_bool_compare_and_swap(type* ptr, type oldValue, type newValue, ....);

type __sync_val_compare_and_swap(type* ptr, type oldValue, type newValue, ....);

//將*ptr設為value並返回*ptr操作之前的值
type __sync_lock_test_and_set(type *ptr, type value, ....);

//置*ptr為0
void __sync_lock_release(type* ptr, ....);

 1 __sync_synchronize(...)
 2 
 3 //作用 : 發出一個full barrier
 4 /*關於memory barrier,cpu會對我們的指令進行排序，一般說來會提高程序的效率，但有時候可能造成我們不希望得到的結果，舉一個例子，比如我們有一個硬件設備，它有4個寄存器，當你發出一個操作指令的時候，一個寄存器存的是你的操作指令（比如READ），兩個寄存器存的是參數（比如是地址和size），最后一個寄存器是控制寄存器，在所有的參數都設置好之后向其發出指令，設備開始讀取參數，執行命令，程序可能如下：*/
 5 write1(dev.register_size, size);
 6 write1(dev.register_addr, addr);
 7 write1(dev.register_cmd, Read);
 8 write1(dev.register_control, GO);
 9 /*如果最后一條write1被換到了前幾條語句之前，那么肯定不是我們所期望的，這時候我們可以在最后一條語句之前加入一個memory barrier,強制cpu執行完前面的寫入以后再執行最后一條：*/
10 write1(dev.register_size, size);
11 write1(dev.register_addr, addr);
12 write1(dev.register_cmd, Read);
13 __sync_synchronize();
14 write1(dev.register_control, GO);
15 
16 //memory barrier有幾種類型：
17 //acquire barrier : 不允許將barrier之后的內存讀取指令移到barrier之前（linux kernel中的wmb()）
18 //release barrier : 不允許將barrier之前的內存讀取指令移到barrier之后 (linux kernel中的rmb())
19 //full barrier    : 以上兩種barrier的合集(linux kernel中的mb())
20 
21 //好吧,說實話這個函數的說明基本沒看懂

最后從網上找一個代碼寫一寫:http://blog.csdn.net/hzhsan/article/details/25837189

測試場景：假設有一個應用：現在有一個全局變量，用來計數，再創建10個線程並發執行，每個線程中循環對這個全局變量進行++操作（i++)，循環加2000000次。

所以很容易知道，這必然會涉及到並發互斥操作。下面通過三種方式[傳統互斥量加鎖方式, no lock不加鎖的方式, 原子函數方式]來實現這種並發操作。並對比出其在效率上的不同之處。

這里先貼上代碼，共5個文件：2個用於做時間統計的文件：timer.h timer.cpp。這兩個文件是臨時封裝的，只用來計時，可以不必細看。

 1 //timer.h 用於計時
 2 
 3 #ifndef TIMER_H_
 4 #define TIMER_H_
 5 
 6 #include <sys/time.h>
 7 
 8 class Timer
 9 {
10     public:
11         Timer();
12         Timer(const Timer& t) = delete;
13         ~Timer();
14 
15         void start();
16         void stop();
17         void reset();
18 
19         double costTime();
20 
21     private: 
22         struct timeval t1;
23         struct timeval t2;
24         bool b1, b2;
25 };

//timer.cpp 
1 #include "timer.h"
 2 #include <iostream>
 3 
 4 using namespace std;
 5 
 6 Timer::Timer():b1(false), b2(false)
 7 {
 8 }
 9 Timer::~Timer()
10 {
11 }
12 void Timer::start()
13 {
14     gettimeofday(&t1, NULL);
15     b1 = true;
16     b2 = false;
17 }
18 void Timer::stop()
19 {
20     gettimeofday(&t2, NULL);
21     b2 = true;
22 }
23 void Timer::reset()
24 {
25     b1 = false;
26     b2 = false;
27 }
28 double Timer::costTime()
29 {
30     if (!b1)
31     {
32         cout<<"error, do not call function start()"<<endl;
33         cout<<"the right sequence : start() ..... stop() costTime()"<<endl;
34 
35         return 0;
36     }
37 
38     if (!b2)
39     {
40         cout<<"error, do not call function stop()"<<endl;
41         cout<<"the right sequence : start() ..... stop() costTime()"<<endl;
42         return 0;
43     }
44 
45     size_t sec = t2.tv_sec - t1.tv_sec; 
46     double usec = t2.tv_usec - t1.tv_usec;
47             
48     if (sec < 0)    
49     {
50         cout<<"error, call stop() before start()"<<endl;
51         cout<<"the right sequence : start() ..... stop() costTime()"<<endl;
52         return 0;
53     }
54     
55     if (usec < 0)
56     {
57         usec += 1000000;
58         --sec;
59         if (sec < 0)    
60         {
61             cout<<"error, call stop() before start()"<<endl;
62             cout<<"the right sequence : start() ..... stop() costTime()"<<endl;
63             return 0;
64         }
65     }
66     
67     return sec + usec * 1.0 / 1000000; 
68 }

1 //thread_function.h  -->多線程要調用的函數
2 #ifndef THREAD_FUNCTION_H_
3 #define THREAD_FUNCTION_H_
4 void* thread_lock_execFunc(void* arg);
5 void* thread_nolock_execFunc(void* arg);
6 void* thread_atom_execFunc(void* arg);
7 #endif

 1 //thread_function.cpp
 2 #include "thread_function.h"
 3 #include "lock.h"
 4 #include <pthread.h>
 5 #include <unistd.h>
 6 
 7 extern volatile int count;
 8 struct LOCK;
 9 
10 void* thread_lock_execFunc(void* arg)
11 {
12 
13 
14     for (int i = 0; i < 2000000; ++i)
15     {
16         pthread_mutex_lock(reinterpret_cast<pthread_mutex_t*>(arg));
17         ++count;
18         pthread_mutex_unlock(reinterpret_cast<pthread_mutex_t*>(arg));
19     }
20 
21     return NULL;
22 }
23 
24 void* thread_nolock_execFunc(void* arg)
25 {
26     LOCK* pLock = reinterpret_cast<LOCK*>(arg);
27     for (int i = 0; i < 2000000; ++i)
28     {
29         while(!(__sync_bool_compare_and_swap(&(pLock->mutex), pLock->use, 1)))
30         {
31             usleep(100000);
32         }
33         ++count;
34         __sync_bool_compare_and_swap(&(pLock->mutex), pLock->unUse, 0);
35     }
36     return NULL;
37 }
38 
39 void* thread_atom_execFunc(void* arg)
40 {
41     for (int i = 0; i < 2000000; ++i)
42     {
43         __sync_fetch_and_add(&count, 1);
44     }
45 
46     return NULL;
47 }

 1 //lock.h --->給mainnolock.cpp使用的類
 2 #ifndef LOCK_H_
 3 #define LOCK_H_
 4 struct LOCK
 5 {
 6     int mutex;
 7     int use;
 8     int unUse;
 9     LOCK() : mutex(0), use(0), unUse(1)
10     {
11     }
12 };
13 #endif

 1 //mainlock.cpp  使用mutex加鎖方式的多線程
 2 #include <iostream>
 3 #include <pthread.h>
 4 #include <iomanip>
 5 
 6 #include "timer.h"
 7 #include "thread_function.h"
 8 
 9 using namespace std;
10 
11 pthread_mutex_t mutex_lock;
12 volatile int count = 0;
13 
14 int main( int argc, char** argv)
15 {
16     pthread_mutex_init(&mutex_lock, NULL);
17 
18     Timer timer;
19     timer.start();
20     
21     /*test thread begin*/
22     pthread_t thread_ids[10];
23     
24     for (int i = 0; i < sizeof(thread_ids)/sizeof(pthread_t); ++i)
25     {
26         pthread_create(&thread_ids[i], NULL, thread_lock_execFunc, &mutex_lock);
27     }
28     
29     for (int i = 0; i < sizeof(thread_ids)/sizeof(pthread_t); ++i)
30     {
31         pthread_join(thread_ids[i], NULL);
32     }
33     /*test thread end*/
34 
35     timer.stop();
36     cout<<setiosflags(ios::fixed)<<setprecision(4)<<"lock cost["<<timer.costTime()<<"]second"<<endl;
37     return 0;
38 }

 1 //main_nolock.cpp 使用__sync_compare_and_swap的多線程
 2 #include <iostream>
 3 #include <pthread.h>
 4 #include <unistd.h>
 5 #include <iomanip>
 6 #include "timer.h"
 7 #include "thread_function.h"
 8 #include "lock.h"
 9 
10 using namespace std;
11 
12 volatile int count = 0;
13 
14 int main(int argc, char** argv)
15 {
16     LOCK lock;
17 
18     Timer timer;
19     timer.start();
20 
21     /*test thread begin*/
22     pthread_t thread_ids[10];
23     for (int i = 0; i < sizeof(thread_ids) / sizeof(pthread_t); ++i)
24     {
25         pthread_create(&thread_ids[i], NULL, thread_nolock_execFunc, &lock);
26     }
27 
28     for (int i = 0; i < sizeof(thread_ids) / sizeof(pthread_t); ++i)
29     {
30         pthread_join(thread_ids[i], NULL);
31     }
32     /*test thread end*/
33     
34     timer.stop();
35     cout<<setiosflags(ios::fixed)<<setprecision(4)<<"nolock cost["<<timer.costTime()<<"]\n";    
36     return 0;
37 }

 1 //main_atomic.cpp 使用__sync_fetch_and_add的多線程
 2 #include<iostream>
 3 #include<pthread.h>
 4 #include<unistd.h>
 5 #include<iomanip>
 6 #include "timer.h"
 7 #include "thread_function.h"
 8 
 9 using namespace std;
10 
11 volatile int count = 0;
12 
13 int main(int argc, char** argv)
14 {
15     Timer timer;
16     timer.start();
17 
18     /*pthread begin*/
19     pthread_t thread_ids[10];
20 
21     for (int i = 0; i < sizeof(thread_ids)/sizeof(pthread_t); ++i)
22     {
23         pthread_create(&thread_ids[i], NULL, thread_atom_execFunc, NULL);
24     }
25 
26     for (int i = 0; i < sizeof(thread_ids)/sizeof(pthread_t); ++i)
27     {
28         pthread_join(thread_ids[i], NULL);
29     }
30 
31     /*pthread end*/
32 
33     timer.stop();
34     cout<<setiosflags(ios::fixed)<<setprecision(4)<<"atomic cost["<<timer.costTime()<<"]\n";
35     return 0;
36 }

 1 //makefile
 2 
 3 CC = g++
 4 CFLAGS = -g -lpthread -std=c++11
 5 
 6 OBJS_LOCK = main_lock.o timer.o thread_function.o
 7 OBJS_UNLOCK = main_nolock.o timer.o thread_function.o
 8 OBJS_ATOMICLOCK = main_atomic.o timer.o thread_function.o
 9 
10 INC = timer.h thread_function.h lock.h
11 
12 lock : $(OBJS_LOCK) $(INC) 
13     $(CC) -o mainlock $(OBJS_LOCK) $(CFLAGS)
14     rm *.o
15 
16 nolock : $(OBJS_UNLOCK) $(INC)
17     $(CC) -o mainnolock $(OBJS_UNLOCK) $(CFLAGS)
18     rm *.o
19 
20 atomiclock : $(OBJS_ATOMICLOCK) $(INC)
21     $(CC) -o mainatomic $(OBJS_ATOMICLOCK) $(CFLAGS)
22 
23 main_lock.o : main_lock.cpp 
24     $(CC) -c main_lock.cpp $(CFLAGS)
25 
26 main_nolock.o : main_nolock.cpp 
27     $(CC) -c main_nolock.cpp $(CFLAGS)
28 
29 main_atomic.o : main_atomic.cpp
30     $(CC) -c  main_atomic.cpp $(CFLAGS)
31 
32 timer.o : timer.cpp 
33     $(CC) -c timer.cpp $(CFLAGS)
34 
35 thread_function.o : thread_function.cpp
36     $(CC) -c thread_function.cpp $(CFLAGS)
37 
38 clean:
39     rm *.o

執行makefile

make lock

make nolock

make atomiclock

然后生成3個可執行文件

運行這3個可執行文件:

另外:針對main_nolock.cpp而言,作者提到了一個現象

在thread_function.cpp中, 隨着一下代碼的改變,運行時間會有變化

while (!(__sync_bool_compare_and_swap (&mutex,lock, 1) ));

while (!(__sync_bool_compare_and_swap (&mutex,lock, 1) )) usleep(1);

while (!(__sync_bool_compare_and_swap (&mutex,lock, 1) ))usleep(10);

while (!(__sync_bool_compare_and_swap (&mutex,lock, 1) ))usleep(100);

while (!(__sync_bool_compare_and_swap (&mutex,lock, 1) ))usleep(1000);

while (!(__sync_bool_compare_and_swap (&mutex,lock, 1) ))usleep(10000);

while (!(__sync_bool_compare_and_swap (&mutex,lock, 1) ))usleep(100000);

執行時間的關系是 : T(;)<T(1)<T(10)<T(100)<T(1000)<T(10000)>T(100000)

通過編程測試及測試得出結論：
1、如果是想用全局變量來做統計操作。而又不得不考慮多線程間的互斥訪問的話，最好使用編譯器支持的原子操作函數。再滿足互斥訪問的前提下，編程最簡單，效率最高。

2、lock-free，無鎖編程方式確實能夠比傳統加鎖方式效率高。所以在高並發程序中采用無鎖編程的方式可以進一步提高程序效率。但是得對無鎖方式有足夠熟悉的了解，不然效率反而會更低而且容易出錯。(比如在某些情況下main_nolock比main_lock的效率還要低)

在學習一個無鎖化編程的分析帖子 http://blog.csdn.net/hzhsan/article/details/25141421

Lock-free 算法通常比基於鎖的算法要好：

從其定義來看，它們是 wait-free 的，可以確保線程永遠不會阻塞。
狀態轉變是原子性的，以至於在任何點失敗都不會惡化數據結構。
因為線程永遠不會阻塞，所以當同步的細粒度是單一原子寫或比較交換時，它們通常可以帶來更高的吞吐量。
在某些情況下，lock-free 算法會有更少的同步寫操作（比如 Interlocked 操作），因此純粹從性能來看，它可能更便宜。

但是 lock-freedom 並不是萬能葯。下面是一些很明顯的不利因素：

樂觀的並發使用會對 hot data structures 導致 livelock。
代碼需要大量困難的測試。通常其正確性取決於對目標機器內存模型的正確解釋。
基於眾多原因，lock-free 代碼很難編寫和維護。

無鎖編程與分布式編程那個更適合多核CPU？

無鎖編程主要是使用原子操作替代鎖來實現對共享資源的訪問保護，舉個例子，要對某個整數變量進行加1操作的話，用鎖保護操作的代碼如下：

int a = 0;

Lock();

a+= 1;

Unlock();

如果對上述代碼反編譯可以發現 a+=1;被翻譯成了以下三條匯編指令：

mov eax,dword ptr [a]

add eax,1

mov dword ptr [a],eax

如果在單核系統中,由於在上述三條指令的任何一條執行完后都可能發生任務切換，比如執行完第1條指令后就發生了任務切換，這時如果有其他任務來對a進行操作的話，當任務切換回來后，將繼續對a進行操作，很可能出現不可預測的結果，因此上述三條指令必須使用鎖來保護，以使這段時間內其他任務無法對a進行操作。

需要注意的是，在多核系統中，因為多個CPU核在物理上是並行的，可能發生同時寫的現象；所以必須保證一個CPU核在對共享內存進行寫操作時，其他CPU核不能寫這塊內存。因此在多核系統中和單核有區別，即使只有一條指令，也需要要加鎖保護。

如果使用原子操作來實現上述加1操作的話，例如使用VC里的InterlockedIncrement來操作的話，那么對a的加1操作需要以下語句

InterlockedIncrement (&a);

這條語句最終的實際加1操作會被翻譯成以下一條帶lock前綴的匯編指令：

lock xadd dword ptr [ecx],eax

使用原子操作時，在進行實際的寫操作時，使用了lock指令，這樣就可以阻止其他任務寫這塊內存，避免出現數據競爭現象。原子操作速度比鎖快，一般要快一倍以上。

使用lock前綴的指令實際上在系統中是使用了內存柵障（memory barrier），當原子操作在進行時，其他任務都不能對內存操作，會影響其他任務的執行。因此這種原子操作實際上屬於一種激烈競爭的鎖，不過由於它的操作時間很快，因此可以看成是一種極細粒度鎖。

在無鎖(Lock－free)編程環境中，主要使用的原子操作為 CAS（Compare and Swap）操作，在VC里對應的操作為InterlockedCompareExchange或者InterlockedCompareExchangeAcquire；如果是64位的操作，需要使用InterlockedCompareExchange64或者InterlockedCompareExchangeAcquire64。使用這種原子操作替代鎖的最大的一個好處是它是非阻塞的。

	比較項目	無鎖編程	分布式編程
1	加速比性能	取決於競爭方式，除非也采用分布式競爭，否則不如分布式鎖競爭的性能	加速比和CPU核數成正比關系，接近於單核多任務時的性能
2	實現的功能	有限	不受限制
3	程序員掌握難易程度	難度太高，過於復雜，普通程序員無法掌握，目前世界上只有少數幾個人掌握。	和單核時代的數據結構算法難度差不多，普通程序員可以掌握
4	現有軟件的移植	使用無鎖算法后，以往的算法需要廢棄掉，無法復用	可以繼承已有的算法，在已有程序基礎上重構即可。

從上表的四個方面的綜合比較可以看出， 無鎖編程的實用價值是遠遠不如分布式編程的，因此分布式編程比無鎖編程更適合多核CPU系統。

可在分布計算機系統的幾台計算機上同時協調執行的程序設計方法,分布式程序設計的主要特征是分布和通信。采用分布式程序設計方法設計程序時，一個程序由若干個可獨立執行的程序模塊組成。這些程序模塊分布於一個分布式計算機系統的幾台計算機上同時執行。分布在各台計算機上的程序模塊是相互關聯的，它們在執行中需要交換數據，即通信。只有通過通信，各程序模塊才能協調地完成一個共同的計算任務。采用分布式程序設計方法解決計算問題時，必須提供用以進行分布式程序設計的語言和設計相應的分布式算法。分布式程序設計語言與常用的各種程序設計語言的主要區別，在於它具有程序分布和通信的功能。因此，分布式程序設計語言，往往可以由一種程序設計語言增加分布和通信的功能而構成。分布式算法和適用於多處理器系統的並行算法，都具有並行執行的特點，但它們是有區別的。設計分布式算法時，必須保證實現算法的各程序模塊間不會有公共變量，它們只能通過通信來交換數據。此外，設計分布式算法時，往往需要考慮堅定性，即當系統中幾台計算機失效時，算法仍是有效的。

免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。

猜您在找 linux無鎖化編程--__sync_fetch_and_add系列原子操作函數 [充電]多線程無鎖編程--原子計數操作：__sync_fetch_and_add等12個操作 Add Google Maven repository and sync project usage: git remote add [ ] -f, --fetch fetch the remote branches --tags import all tags and associated objects when fetching fetch fetch的基本使用 Git fetch fetch的用法 Fetch API fetch詳解