C開發 中原子性操作 , 除了快什麼都不剩下了，原子什麼都不

C開發 中原子性操作 , 除了快什麼都不剩下了，原子什麼都不

題外話

　　今天,聽歌曲聽到一首緬懷邁克爾·傑克遜的歌曲 如下:

 http://music.163.com/#/song?id=1696048  Breaking News

每次聽邁克爾 音樂,特別有戰斗力,特別興奮,學起技術來也特別帶感,推薦喜歡的人試試.

#include <stdio.h>

int man(int argc, char* argv[])
{
   printf("Hope you are better %s\n","Michael Jackson"); 
   return 0;  
}

 

前言

今天要說的是

1. 通過實際例子開頭 說明 互斥量, 原子操作, 原子互斥鎖 性能對比

2. 簡單 說一下Linux 同樣的性能對比

3. 構建一個跨平台的簡單原子操作框架

4. Window 和 Linux 都簡單測試一下

 

預備知識  

1.線程的使用 最好是 posix線程庫

2.簡單C基礎(說個 題外話,將C之父 那本書來回看個3遍,多寫幾遍,基本就是一個合格C程序員了)

 

參照資料

1. gcc 原子操作幫助文檔 https://gcc.gnu.org/onlinedocs/gcc-4.1.1/gcc/Atomic-Builtins.html

2. Window 提供的原子操作API https://msdn.microsoft.com/en-us/library/windows/desktop/ms686360(v=vs.85).aspx#interlocked_functions

 

正文

  首先我們直奔 任務1, 通過實際例子開頭 說明 互斥量, 原子操作, 原子互斥鎖 性能對比

1.0 pthread 預備

　　需要會使用pthread 線程庫,特別是在Window上, 在 Linux 默認就是posix 線程庫,只需要在gcc 後面 加上 -lpthread 例如如下

gcc -Wall -o atom.out atom.c sc_atom.h -lpthread

沒使用的可以 參照 我的其它博文,好像是一個介紹 printf 函數一個博文中簡單講解了,怎麼 在Window上使用 pthread的全過程.

這裡 再 簡單說一下,為什麼 要在 Window上折騰,是這樣的 畢竟 是從開機 -> 走上程序開發道路 , 那時候 上大一,第一次了解Window其實個操作系統.

以前心裡默認以為 window 就是 電腦,電腦就是window 二者是一個對映關系. 後來常在 Linux工作,開發學習. 還是覺得 Window有很多優點,是不錯的操作系統.

其實 再閒扯一點(個人比較菜 望見諒)

Window 挺難的,源碼 看得好惡心 工資低

Linux 簡單,源碼容易看,工資高一點

這 業界 喜好 就出來了.簡單回報高,誰不喜歡.

 

1.1 簡單看代碼

 

  這裡先看下面代碼 ,第一段 普通 pthread_mutex 互斥量

//測試 pthread 互斥量
void* test_one(void* arg);

#define _INT_CUTS (2000000)

//測試 的全局區變量,默認值為0
static int __cut;
//全局的鎖變量
static pthread_mutex_t __mx = PTHREAD_MUTEX_INITIALIZER;

void* 
test_one(void* arg)
{
    for (int i = 0; i < _INT_CUTS; ++i) {
        pthread_mutex_lock(&__mx);
        ++__cut;
        pthread_mutex_unlock(&__mx);
    }

    return NULL;
}

上面代碼 特別標准,一般多線程程序代碼,基本就是上面結構,扯一點 對於下面多線程函數

PTW32_DLLPORT int PTW32_CDECL pthread_mutex_destroy (pthread_mutex_t * mutex);

只有 在 通過

PTW32_DLLPORT int PTW32_CDECL pthread_mutex_init (pthread_mutex_t * mutex,
                                const pthread_mutexattr_t * attr);

初始化的互斥量,才需要調用,對於直接 通過 初始化值初始化的互斥量就不需要調用了.

 

現在介紹 一個 原子操作的版本,先看Windows 的

//測試 原子操作
void* test_two(void* arg);

#define _INT_CUTS (2000000)

//測試 的全局區變量,默認值為0
static int __cut;

//測試 原子操作
void* 
test_two(void* arg)
{
    for (int i = 0; i < _INT_CUTS; ++i) {
        InterlockedExchangeAdd(&__cut, 1);
    }

    return NULL;
}

這裡函數 InterlockedExchangeAdd 是 Window.h 中提供的 具體的意思 如下

#define InterlockedExchangeAdd _InterlockedExchangeAdd

/*
 *  原子操作,綁定在一起 完成 第一個數 = 第一個數+被加數 ,並返回開始加的時候第一個數
 * Addend : 加數的地址
 * Value : 被加數
 */
LONG
__cdecl
InterlockedExchangeAdd (
    _Inout_ _Interlocked_operand_ LONG volatile *Addend,
    _In_ LONG Value
    );

可以理解為,下面一起執行的操作命令集

tmp = old ; old = old + value ; return tmp ;

 

最後介紹一個 利用原子操作 實現的互斥鎖

//測試 原子鎖操作
void* test_three(void* arg);

#define _INT_CUTS (2000000)

//測試 的全局區變量,默認值為0
static int __cut;

//全局 鎖
static int __lk;

//測試 原子鎖操作
void* 
test_three(void* arg)
{
    for (int i = 0; i < _INT_CUTS; ++i) {

        while (InterlockedExchange(&__lk, 1)) {
            Sleep(0);
        }

        ++__cut;

        InterlockedExchange(&__lk, 0);
    }

    return NULL;
}

這裡同樣簡單 解釋一下

#define InterlockedExchange _InterlockedExchange

/*
 * 這個函數作用是 交換 *Target 和 Value 值,並返回老的值
 *  
 * Target : 目標值得地址
 * Value : 待交換的值
 *
 * return : 交換之前的 *Target 值
 */
LONG
__cdecl
InterlockedExchange (
    _Inout_ _Interlocked_operand_ LONG volatile *Target,
    _In_ LONG Value
    );

上面函數 等價於 下面指令集一起執行.

tmp = *Target ; *Target = Value ; Value = tmp ; return tmp ;

對於

        while (InterlockedExchange(&__lk, 1)) {
            Sleep(0);
        }

相當加鎖 , 例如當 __lk 為 0 的時候,先進入的線程 設置為1,返回0它進入了,別的線程來了,返回的值可能是1,那麼 就執行

Sleep(0),等待.假如你要是寫成下面這樣

while (InterlockedExchange(&__lk, 1)) { }

就是 出名的 "忙等待 ",語言層等待函數, 測試結果是程序 基本上卡死. 這就 人一樣,需要 休息 一下,才更有精力和動力 去做其它的事. 老是 加班效果不好. 但 沒辦法 , 因為自己只是個打工仔,,,,,

後面使用

InterlockedExchange(&__lk, 0);

將這個變量設置為 0 ,那麼 另一個線程 執行 set 1時候,返回 0 while 退出,獲得線程資源,就這樣 循環起來了,是不是 恍然大悟!!!

 

上面三種實現 多線程競爭的訪問資源的方式 ,都比較安全, 我們具體的看一下 測試結果 采用 Window 10 x64 + VS 2015 + Release + x86 模式 測試 截圖如下:

是不是很驚訝, 原子操作快 我也不說了,居然 原子互斥鎖更快,但事實就是這樣, 這裡 可以更快一點 優化點在 Sleep(0), 這個函數 參數 以微秒記錄. 更加優化,不同操作

系統,這個最優值不一樣.這裡 我就設了一個 性能可以的值. 大家可以嘗試一下.

扯一點 在 Linux 中 有個函數 usleep 函數,在Window 上沒有,但是也可以實現, 參照資料 如下

http://stackoverflow.com/questions/5801813/c-usleep-is-obsolete-workarounds-for-windows-mingw

覺得 Linux 的API 和 Window API 都各有所長,各有所短,綜合而言還是 Linux API人性化一點,能夠優化的東西更多.

 

完整的測試代碼 如下

#include "pthread.h"
#include <stdio.h>
#include <stdlib.h>
#include <Windows.h>
#include <time.h>

//1.0 簡單的time幫助宏
#ifndef TIME_PRINT
#define TIME_PRINT(code) {\
    clock_t __st,__et;\
    __st=clock();\
    code\
    __et=clock();\
    printf("當前代碼塊運行時間是:%lf秒\n",(0.0+__et-__st)/CLOCKS_PER_SEC);\
}
#endif /*!TIME_PRINT*/

//測試 pthread 互斥量
void* test_one(void* arg);
//測試 原子操作
void* test_two(void* arg);
//測試 原子鎖操作
void* test_three(void* arg);

#define _INT_TIDS (50)
#define _INT_CUTS (2000000)

//測試 的全局區變量,默認值為0
static int __cut;
//全局的鎖變量
static pthread_mutex_t __mx = PTHREAD_MUTEX_INITIALIZER;
//全局 鎖
static int __lk;

static void __test_func(void* (*func)(void *))
{
    pthread_t tids[_INT_TIDS];
    int i;

    for (i = 0; i < _INT_TIDS; ++i) {
        pthread_create(tids + i, NULL, func, NULL);
    }

    //等待結束
    for (i = 0; i < _INT_TIDS; ++i)
        pthread_join(tids[i], NULL);
}

int main(int argc, char *argv[])
{
    printf("__cut = %d, __mx = %d, __lk = %d\n", __cut, (int)__mx, __lk);

    // 只為簡單測試,沒有做安全檢查,假定都會調用成功

    puts("\n線程互斥鎖數據如下:");
    __cut = 0;
    TIME_PRINT({
        __test_func(test_one);
    });
    printf("__cut = %d, __mx = %d, __lk = %d\n", __cut, (int)__mx, __lk);


    puts("\n原子操作數據如下:");
    __cut = 0;
    TIME_PRINT({
        __test_func(test_two);
    });
    printf("__cut = %d, __mx = %d, __lk = %d\n", __cut, (int)__mx, __lk);


    puts("\n原子鎖操作數據如下:");
    __cut = 0;
    TIME_PRINT({
        __test_func(test_three);
    });
    printf("__cut = %d, __mx = %d, __lk = %d\n", __cut, (int)__mx, __lk);

    system("pause");
    return 0;
}

void* 
test_one(void* arg)
{
    for (int i = 0; i < _INT_CUTS; ++i) {
        pthread_mutex_lock(&__mx);
        ++__cut;
        pthread_mutex_unlock(&__mx);
    }

    return NULL;
}

//測試 原子操作
void* 
test_two(void* arg)
{
    for (int i = 0; i < _INT_CUTS; ++i) {
        InterlockedExchangeAdd(&__cut, 1);
    }

    return NULL;
}

//測試 原子鎖操作
void* 
test_three(void* arg)
{
    for (int i = 0; i < _INT_CUTS; ++i) {

        while (InterlockedExchange(&__lk, 1)) {
            Sleep(0);
        }

        ++__cut;

        InterlockedExchange(&__lk, 0);
    }

    return NULL;
}

到這裡 , 第一目標就告一段落,再扯一點,提升編程最好 手段 就是 臨摹,如果人不聰明的話 多敲鍵盤.

 

2. 簡單 說一下Linux 同樣的性能對比

2.1 直接上代碼

測試 代碼 gcc_sync.c , 采用Ubuntu 15.10 操作系統,感覺挺好用的,比Centos 好用,好開發.大家試試

(這裡,寫的比較簡單隨意,我是先在Linux上測試的,測試之後再在Window上按照同樣的設計思路寫代碼,開發中遇到比較難得問題,都是先用C寫一遍穩固設計思路,再轉成其它語言.)

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <time.h>
#include <unistd.h>

static int __cut = 0;

void* test_func(void* arg);

#define _INT_THREAD (30)

static pthread_mutex_t __mx = PTHREAD_MUTEX_INITIALIZER;
static int __lk = 0;

int main(int argc,char* argv[])
{
        pthread_t tid[_INT_THREAD];
        int i;

        printf("__cut = %d\n",__cut);

        clock_t st = clock();
        for(i=0; i<20; ++i)
                pthread_create(tid + i, NULL, test_func, NULL);

        for(i=0; i<20; ++i)
                pthread_join(tid[i], NULL);
        clock_t et = clock();

        printf("__cut = %d\n",__cut);
        printf("經歷了 %lf 秒\n",(et-st)*1.0/CLOCKS_PER_SEC);

        return 0;
}

// 簡單的測試 自增N次
void*
test_func(void* arg)
{
        int i = 0;
        while(i++<2000000){
                //pthread_mutex_lock(&__mx);

                //__sync_fetch_and_add(&__cut,1);


                while(__sync_lock_test_and_set(&__lk,1)) {
                        usleep(0);
                }

                ++__cut;

                __sync_lock_release(&__lk);

                //pthread_mutex_unlock(&__mx);
        }
        return NULL;
}

編譯命令是

gcc -g -Wall -o gcc_sync.out gcc_sync.c -lpthread

測試結果 也 同樣 ,所用時間  pthread_mutex > 原子加 > 原子互斥量

對於 上面代碼 簡單說一下

__sync_fetch_and_add(&__cut,1);

意思是 為__cut增加 1,並返回 原先的__cut值.

更加詳細的關於 gcc 提供的原子操作,可以看 前言 中 參照資料的第一個.其實 gcc 提供的原子操作 "函數",真的 是魔法函數,是編譯器層的不是語言層的.

Windows 上的 InterlockedExchangeAdd 還是 語言層的,有固定的返回類型.

對於下面函數,詳細說一下

                while(__sync_lock_test_and_set(&__lk,1)) {
                        usleep(0);
                }

對於

/*
 * 原子操作,交換 *ptr 和 value ,並且返回交換後的 value值 
 *
 * type : 可以是1,2,4或8字節長度的int類型 , int8_t/uint64_t 都可以...
 * value : 目標 *ptr 交換的值
 * ... : 後面的可擴展參數(...)用來指出哪些變量需要memory barrier,因為目前gcc實現的是full barrier,這個意思表示 這個函數之前的內存變量操作指令不會出現在這個函數之後執行.
 * 
 * return : *ptr 之前的值就是 交換後的value
 */
type __sync_lock_test_and_set (type *ptr, type value, ...)

等同於

tmp = value ; value = *ptr ; *ptr = tmp ; return value ;

對於 usleep(0), 這個自己也在測試 目前 測了一下 感覺 usleep(2) 效果比較好, 和操作系統和 硬件和 代碼復雜度 關系大, 自己 目前就用 usleep(2); 這是個優化點.

最後一個是

void __sync_lock_release (type *ptr, ...)This builtin releases the lock acquired by __sync_lock_test_and_set. Normally this means writing the constant 0 to *ptr. 

將值設為0,和 __sync_lock_test_and_set 配套使用

等價於 下面原子操作

*ptr = 0

到這裡 大概 原子操作的概念就建立起來,至少知道 原子操作怎麼搞了.

 

3. 構建一個跨平台的簡單原子操作框架

 

這裡 同樣直接上代碼 ,再挨個解釋. 代碼總感覺有點難看, 但不知道怎麼改,下次再優化

文件名 sc_atom.h

 

#ifndef _H_SC_ATOM
#define _H_SC_ATOM

/*
 * 這段關於原子操作的宏 主要在 VS 和 GCC 中跑
 * 
 * 為什麼要用原子操作,因為它快,很快. 接近硬件層,怎麼使用會做具體的注釋
 */

#if defined(_MSC_VER)
//這裡主要 _WIN32 操作 ,對於WIN64 沒有做了,本質一樣,函數後面加上64. 這裡定位就是從簡單的跨平台來來    
#include <Windows.h>

//全部采用後置原子操作,先返回old的值 (前置等價 => tmp = v ; v = v + a ; return tmp)
#define SC_ATOM_ADD(v,a) \
    InterlockedAdd(&(v),(a))

//將a的值設置給v,返回設置之前的值
#define SC_ATOM_SET(v,a) \
    InterlockedExchange(&(v),(a))

// v == c ? swap(v,a) ; return true : return false.
#define SC_ATOM_COM(v,c,a) \
    ( c == InterlockedCompareExchange(&(v), (a), c))

//第一次使用 v最好是 0
#define SC_ATOM_LOCK(v) \
    while(SC_ATOM_SET(v,1)) { \
        Sleep(0); \
    }

#define SC_ATOM_UNLOCK(v) \
    SC_ATOM_SET(v,0)

#elif defined(__GNUC__)
#include <unistd.h>    

 //全部采用後置原子操作,先返回old的值 (前置等價 => a = i++)
#define SC_ATOM_ADD(v,a) \
    __sync_fetch_and_sub(&(v),(a))

 //將a的值設置給v,返回設置之前的值
#define SC_ATOM_SET(v,a) \
    __sync_lock_test_and_set(&(v),(a))

 // v == c ? swap(v,a) return true : return false.
#define SC_ATOM_CMP(v,c,a) \
    __sync_bool_compare_and_swap(&(v), (cmp), (val))

//等待的秒數,因環境而定 2是我自己測試的一個值
#define _INT_USLEEP (2) 
#define SC_ATOM_LOCK(v) \
    while(SC_ATOM_SET(v,1)) { \
        usleep(_INT_USLEEP); \
    }

#define SC_ATOM_UNLOCK(v) \
    __sync_lock_release(&(v))


#endif /* _MSC_VER || __GNU__*/

#endif /*!_H_SC_ATOM*/

先將 這節的上面沒出現過的api解析一下

// v == c ? swap(v,a) ; return true : return false.
#define SC_ATOM_COM(v,c,a) \
    ( c == InterlockedCompareExchange(&(v), (a), c))

首先是

#define InterlockedCompareExchange _InterlockedCompareExchange
/*
 *   原子操作,比較並交換,返回老的值 
 *  *Destination == Comperand 就交換 *Destination 和 ExChange,並返回交換後的
 * ExChange,如果不等 直接返回 *Destination
 */
LONG
CDECL_NON_WVMPURE
InterlockedCompareExchange (
    _Inout_ _Interlocked_operand_ LONG volatile * Destination,
    _In_ LONG ExChange,
    _In_ LONG Comperand
    );

這裡

( c == InterlockedCompareExchange(&(v), (a), c)) 等價於

tmp = v; v==c ? v =a : ; return c == tmp; 

繼續看看

#define SC_ATOM_CMP(v,c,a) \
    __sync_bool_compare_and_swap(&(v), (cmp), (val))

對於

bool __sync_bool_compare_and_swap (type *ptr, type oldval type newval, ...);
type __sync_val_compare_and_swap (type *ptr, type oldval type newval, ...);

These builtins perform an atomic compare and swap. That is, if the current value of *ptr is oldval, then write newval into *ptr. 
The “bool” version returns true if the comparison is successful and newval was written. The “val” version returns the contents of *ptr before the operation. 

第一條api意思就是 

tmp = *ptr ; *ptr == oldval ? *ptr = newval : ; return tmp == oldval;

第二個直接返回 tmp.

其實說的很簡單,當你看到這你需要 看更多的資料,寫 一點代碼,才能 理解 它的作用,並可以在 以後的代碼中使用它,輕巧的完成一些特定代碼了.

到這裡 不知道你是否和我一樣興奮,以後寫 代碼的速率又快樂一點.  扯一點,有一天去逛 一個 技術 站在雲端的人的博文,底下有人評價 是這樣的意思  "你也就是一個干了一輩子的技術狗".

思來思去,發現 能當一只 "狗" 也挺好, 至少 "狗的忠誠,狗的勇敢,狗對待朋友的熱情" 太難得了.

最後總結,要想走得更遠,還是少生氣,少罵人的好,多行動,多給世界一點 熱. 

 

4. Window 和 Linux 都簡單測試一下

這個就輕松了,首先 是 Window 上測試案例 如下

 

#include <stdio.h>
#include <stdlib.h>
#include "sc_atom.h"
#include "pthread.h"

//1.0 簡單的time幫助宏
#ifndef TIME_PRINT
#define TIME_PRINT(code) {\
    clock_t __st,__et;\
    __st=clock();\
    code\
    __et=clock();\
    printf("當前代碼塊運行時間是:%lf秒\n",(0.0+__et-__st)/CLOCKS_PER_SEC);\
}
#endif /*!TIME_PRINT*/

//測試 的全局區變量,默認值為0
static int __cut;
//全局 鎖
static int __lk;

//測試 原子鎖操作
void* test_three(void* arg);

#define _INT_TIDS (50)
#define _INT_CUTS (2000000)

int main(int argc, char* argv[])
{
    pthread_t tids[_INT_TIDS];
    int i;

    puts("\n原子鎖操作數據如下:");
    __cut = 0;
    TIME_PRINT({
        for (i = 0; i < _INT_TIDS; ++i) {
            pthread_create(tids + i, NULL, test_three,NULL);
        }

    //等待結束
    for (i = 0; i < _INT_TIDS; ++i)
        pthread_join(tids[i],NULL);
    });
    printf("__cut = %d, __lk = %d\n", __cut, __lk);

    system("pause");
    return 0;
}

//測試 原子鎖操作
void*
test_three(void* arg)
{
    for (int i = 0; i < _INT_CUTS; ++i) {
        SC_ATOM_LOCK(__lk);
        ++__cut;
        SC_ATOM_UNLOCK(__lk);
    }

    return NULL;
}

運行的沒有問題,效果理想 運行圖如下:

速度還可以.采用 Window 10 x64 + VS2015 + Release + x86

 

下面是Linux 上 測試 案例

環境是 Ubuntu 15.10 x64 + gcc 5.2.1

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include "sc_atom.h"

//1.0 簡單的time幫助宏
#ifndef TIME_PRINT
#define TIME_PRINT(code) {\
    clock_t __st,__et;\
    __st=clock();\
    code\
    __et=clock();\
    printf("當前代碼塊運行時間是:%lf秒\n",(0.0+__et-__st)/CLOCKS_PER_SEC);\
}
#endif /*!TIME_PRINT*/

//測試 的全局區變量,默認值為0
static int __cut;
//全局 鎖
static int __lk;

//測試 原子鎖操作
void* test_three(void* arg);

#define _INT_TIDS (50)
#define _INT_CUTS (2000000)

int main(int argc, char* argv[])
{
    pthread_t tids[_INT_TIDS];
    int i;

    puts("\n原子鎖操作數據如下:");
    __cut = 0;
    TIME_PRINT({
        for (i = 0; i < _INT_TIDS; ++i) {
            pthread_create(tids + i, NULL, test_three,NULL);
        }

    //等待結束
    for (i = 0; i < _INT_TIDS; ++i)
        pthread_join(tids[i],NULL);
    });
    printf("__cut = %d, __lk = %d\n", __cut, __lk);

    return 0;
}

//測試 原子鎖操作
void*
test_three(void* arg)
{
    for (int i = 0; i < _INT_CUTS; ++i) {
        SC_ATOM_LOCK(__lk);
        ++__cut;
        SC_ATOM_UNLOCK(__lk);
    }

    return NULL;
}

測試效果截圖如下

到這裡 我們關於C中原子操作就告一段落,歡迎交流,互相提高. 有錯誤是肯定,指正之後馬上改!

 

後記

一天過得好快,下次繼續分享,一些關於C開發中一些技巧. 有好博文的同行多發廣告,不加班就去貴空間中拜訪交流學習的.