前言

在本篇文章當(dāng)中主要給大家介紹一下 OpenMP 當(dāng)中經(jīng)常使用到的鎖并且仔細(xì)分析它其中的內(nèi)部原理！在 OpenMP 當(dāng)中主要有兩種類型的鎖，一個(gè)是 omp_lock_t 另外一個(gè)是 omp_nest_lock_t，這兩個(gè)鎖的主要區(qū)別就是后者是一個(gè)可重入鎖，所謂可沖入鎖就是一旦一個(gè)線程已經(jīng)拿到這個(gè)鎖了，那么它下一次想要拿這個(gè)鎖的就是就不會(huì)阻塞，但是如果是 omp_lock_t 不管一個(gè)線程是否拿到了鎖，只要當(dāng)前鎖沒有釋放，不管哪一個(gè)線程都不能夠拿到這個(gè)鎖。在后問當(dāng)中將有仔細(xì)的例子來解釋這一點(diǎn)。本篇文章是基于 GNU OpenMP Runtime Library !

深入分析 omp_lock_t

這是 OpenMP 頭文件給我們提供的一個(gè)結(jié)構(gòu)體，我們來看一下它的定義：

typedef?struct
{
??unsigned?char?_x[4]?
????__attribute__((__aligned__(4)));
}?omp_lock_t;

事實(shí)上這個(gè)結(jié)構(gòu)體并沒有什么特別的就是占 4 個(gè)字節(jié)，我們甚至可以認(rèn)為他就是一個(gè) 4 字節(jié)的 int 的類型的變量，只不過使用方式有所差異。與這個(gè)結(jié)構(gòu)體相關(guān)的主要有以下幾個(gè)函數(shù)：

omp_init_lock，這個(gè)函數(shù)的主要功能是初始化 omp_lock_t 對象的，當(dāng)我們初始化之后，這個(gè)鎖就處于一個(gè)沒有上鎖的狀態(tài)，他的函數(shù)原型如下所示：

void?omp_init_lock(omp_lock_t?*lock);

omp_set_lock，在調(diào)用這個(gè)函數(shù)之前一定要先調(diào)用函數(shù) omp_init_lock 將 omp_lock_t 進(jìn)行初始化，直到這個(gè)鎖被釋放之前這個(gè)線程會(huì)被一直阻塞。如果這個(gè)鎖被當(dāng)前線程已經(jīng)獲取過了，那么將會(huì)造成一個(gè)死鎖，這就是上面提到了鎖不能夠重入的問題，而我們在后面將要分析的鎖 omp_nest_lock_t 是能夠進(jìn)行重入的，即使當(dāng)前線程已經(jīng)獲取到了這個(gè)鎖，也不會(huì)造成死鎖而是會(huì)重新獲得鎖。這個(gè)函數(shù)的函數(shù)原型如下所示：

void?omp_set_lock(omp_lock_t?*lock);

omp_test_lock，這個(gè)函數(shù)的主要作用也是用于獲取鎖，但是這個(gè)函數(shù)可能會(huì)失敗，如果失敗就會(huì)返回 false 成功就會(huì)返回 true，與函數(shù) omp_set_lock 不同的是，這個(gè)函數(shù)并不會(huì)導(dǎo)致線程被阻塞，如果獲取鎖成功他就會(huì)立即返回 true，如果失敗就會(huì)立即返回 false 。它的函數(shù)原型如下所示：

int?omp_test_lock(omp_lock_t?*lock);?

omp_unset_lock，這個(gè)函數(shù)和上面的函數(shù)對應(yīng)，這個(gè)函數(shù)的主要作用就是用于解鎖，在我們調(diào)用這個(gè)函數(shù)之前，必須要使用 omp_set_lock 或者 omp_test_lock 獲取鎖，它的函數(shù)原型如下：

void?omp_unset_lock(omp_lock_t?*lock);

omp_destroy_lock，這個(gè)方法主要是對鎖進(jìn)行回收處理，但是對于這個(gè)鎖來說是沒有用的，我們在后文分析他的具體的實(shí)現(xiàn)的時(shí)候會(huì)發(fā)現(xiàn)這是一個(gè)空函數(shù)。

我們現(xiàn)在使用一個(gè)例子來具體的體驗(yàn)一下上面的函數(shù)：

#include?<stdio.h>
#include?<omp.h>

int?main()
{
???omp_lock_t?lock;
???//?對鎖進(jìn)行初始化操作
???omp_init_lock(&lock);
???int?data?=?0;
#pragma?omp?parallel?num_threads(16)?shared(lock,?data)?default(none)
???{
??????//?進(jìn)行加鎖處理?同一個(gè)時(shí)刻只能夠有一個(gè)線程能夠獲取鎖
??????omp_set_lock(&lock);
??????data++;
??????//?解鎖處理?線程在出臨界區(qū)之前需要解鎖?好讓其他線程能夠進(jìn)入臨界區(qū)
??????omp_unset_lock(&lock);
???}
???omp_destroy_lock(&lock);
???printf("data?=?%d\n",?data);
???return?0;
}

在上面的函數(shù)我們定義了一個(gè) omp_lock_t 鎖，并且在并行域內(nèi)啟動(dòng)了 16 個(gè)線程去執(zhí)行 data ++ 的操作，因?yàn)槭嵌嗑€程環(huán)境，因此我們需要將上面的操作進(jìn)行加鎖處理。

omp_lock_t 源碼分析

omp_init_lock，對于這個(gè)函數(shù)來說最終在 OpenMP 動(dòng)態(tài)庫內(nèi)部會(huì)調(diào)用下面的函數(shù)：

typedef?int?gomp_mutex_t;
static?inline?void
gomp_mutex_init?(gomp_mutex_t?*mutex)
{
??*mutex?=?0;
}

從上面的函數(shù)我們可以知道這個(gè)函數(shù)的作用就是將我們定義的 4 個(gè)字節(jié)的鎖賦值為0，這就是鎖的初始化，其實(shí)很簡單。

omp_set_lock，這個(gè)函數(shù)最終會(huì)調(diào)用 OpenMP 內(nèi)部的一個(gè)函數(shù)，具體如下所示：

static?inline?void
gomp_mutex_lock?(gomp_mutex_t?*mutex)
{
??int?oldval?=?0;
??if?(!__atomic_compare_exchange_n?(mutex,?&oldval,?1,?false,
????????MEMMODEL_ACQUIRE,?MEMMODEL_RELAXED))
????gomp_mutex_lock_slow?(mutex,?oldval);
}

在上面的函數(shù)當(dāng)中線程首先會(huì)調(diào)用 __atomic_compare_exchange_n 將鎖的值由 0 變成 1，還記得我們在前面對鎖進(jìn)行初始化的時(shí)候?qū)㈡i的值變成0了嗎？

我們首先需要了解一下 __atomic_compare_exchange_n ，這個(gè)是  gcc 內(nèi)嵌的一個(gè)函數(shù)，在這里我們只關(guān)注前面三個(gè)參數(shù)，后面三個(gè)參數(shù)與內(nèi)存模型有關(guān)，這并不是我們本篇文章的重點(diǎn)，他的主要功能是查看 mutex 指向的地址的值等不等于 oldval ，如果等于則將這個(gè)值變成 1，這一整個(gè)操作能夠保證原子性，如成功將 mutex 指向的值變成 1 的話，那么這個(gè)函數(shù)就返回 true 否則返回 false 對應(yīng) C 語言的數(shù)據(jù)就是 1 和 0 。如果 oldval 的值不等于 mutex 所指向的值，那么這個(gè)函數(shù)就會(huì)將這個(gè)值寫入 oldval 。

如果這個(gè)操作不成功那么就會(huì)調(diào)用 gomp_mutex_lock_slow 函數(shù)這個(gè)函數(shù)的主要作用就是如果使用不能夠使用原子指令獲取鎖的話，那么就需要進(jìn)入內(nèi)核態(tài)，將這個(gè)線程掛起。在這個(gè)函數(shù)的內(nèi)部還會(huì)測試是否能夠通過源自操作獲取鎖，因?yàn)榭赡茉谖覀冋{(diào)用 gomp_mutex_lock_slow 這個(gè)函數(shù)的時(shí)候可能有其他線程釋放鎖了。如果仍然不能夠成功的話，那么就會(huì)真正的將這個(gè)線程掛起不會(huì)浪費(fèi) CPU 資源，gomp_mutex_lock_slow 函數(shù)具體如下：

void
gomp_mutex_lock_slow?(gomp_mutex_t?*mutex,?int?oldval)
{
??/*?First?loop?spins?a?while.??*/
??//?先自旋?如果自旋一段時(shí)間還沒有獲取鎖?那就將線程刮掛起
??while?(oldval?==?1)
????{
??????if?(do_spin?(mutex,?1))
?{
???/*?Spin?timeout,?nothing?changed.??Set?waiting?flag.??*/
???oldval?=?__atomic_exchange_n?(mutex,?-1,?MEMMODEL_ACQUIRE);
????//?如果獲得???就返回
???if?(oldval?==?0)
?????return;
????//?如果沒有獲得???那么就將線程刮起
???futex_wait?(mutex,?-1);
????//?這里是當(dāng)掛起的線程被喚醒之后的操作?也有可能是?futex_wait?沒有成功
???break;
?}
??????else
?{
???/*?Something?changed.??If?now?unlocked,?we're?good?to?go.??*/
???oldval?=?0;
???if?(__atomic_compare_exchange_n?(mutex,?&oldval,?1,?false,
????????MEMMODEL_ACQUIRE,?MEMMODEL_RELAXED))
?????return;
?}
????}

??/*?Second?loop?waits?until?mutex?is?unlocked.??We?always?exit?this
?????loop?with?wait?flag?set,?so?next?unlock?will?awaken?a?thread.??*/
??while?((oldval?=?__atomic_exchange_n?(mutex,?-1,?MEMMODEL_ACQUIRE)))
????do_wait?(mutex,?-1);
}

在上面的函數(shù)當(dāng)中有三個(gè)依賴函數(shù)，他們的源代碼如下所示：

static?inline?void
futex_wait?(int?*addr,?int?val)
{
??//?在這里進(jìn)行系統(tǒng)調(diào)用，將線程掛起?
??int?err?=?syscall?(SYS_futex,?addr,?gomp_futex_wait,?val,?NULL);
??if?(__builtin_expect?(err?<?0?&&?errno?==?ENOSYS,?0))
????{
??????gomp_futex_wait?&=?~FUTEX_PRIVATE_FLAG;
??????gomp_futex_wake?&=?~FUTEX_PRIVATE_FLAG;
????//?在這里進(jìn)行系統(tǒng)調(diào)用，將線程掛起?
??????syscall?(SYS_futex,?addr,?gomp_futex_wait,?val,?NULL);
????}
}

static?inline?void?do_wait?(int?*addr,?int?val)
{
??if?(do_spin?(addr,?val))
????futex_wait?(addr,?val);
}

static?inline?int?do_spin?(int?*addr,?int?val)
{
??unsigned?long?long?i,?count?=?gomp_spin_count_var;

??if?(__builtin_expect?(__atomic_load_n?(&gomp_managed_threads,
?????????????????????????????????????????MEMMODEL_RELAXED)
????????????????????????>?gomp_available_cpus,?0))
????count?=?gomp_throttled_spin_count_var;
??for?(i?=?0;?i?<?count;?i++)
????if?(__builtin_expect?(__atomic_load_n?(addr,?MEMMODEL_RELAXED)?!=?val,?0))
??????return?0;
????else
??????cpu_relax?();
??return?1;
}

static?inline?void
cpu_relax?(void)
{
??__asm?volatile?(""?:?:?:?"memory");
}

如果大家對具體的內(nèi)部實(shí)現(xiàn)非常感興趣可以仔細(xì)研讀上面的代碼，如果從 0 開始解釋上面的代碼比較麻煩，這里就不做詳細(xì)的分析了，簡要做一下概括：

1.在鎖的設(shè)計(jì)當(dāng)中有一個(gè)非常重要的原則：一個(gè)線程最好不要進(jìn)入內(nèi)核態(tài)被掛起，如果能夠在用戶態(tài)最好在用戶態(tài)使用原子指令獲取鎖，這是因?yàn)檫M(jìn)入內(nèi)核態(tài)是一個(gè)非常耗時(shí)的事情相比起原子指令來說。

2.鎖（就是我們在前面討論的一個(gè) 4 個(gè)字節(jié)的 int 類型的值）有以下三個(gè)值：

• -1 表示現(xiàn)在有線程被掛起了。
• 0  表示現(xiàn)在是一個(gè)無鎖狀態(tài)，這個(gè)狀態(tài)就表示鎖的競爭比較激烈。
• 1 表示這個(gè)線程正在被一個(gè)線程用一個(gè)原子指令——比較并交換(CAS)獲得了，這個(gè)狀態(tài)表示現(xiàn)在鎖的競爭比較輕。

3._atomic_exchange_n (mutex, -1, MEMMODEL_ACQUIRE); ，這個(gè)函數(shù)也是 gcc 內(nèi)嵌的一個(gè)函數(shù)，這個(gè)函數(shù)的主要作用就是將 mutex 的值變成 -1，然后將 mutex 指向的地址的原來的值返回。

4.__atomic_load_n (addr, MEMMODEL_RELAXED)，這個(gè)函數(shù)的作用主要作用是原子的加載 addr 指向的數(shù)據(jù)。

5.futex_wait 函數(shù)的功能是將線程掛起，將線程掛起的系統(tǒng)調(diào)用為 futex ，大家可以使用命令 man futex 去查看 futex 的手冊。

6.do_spin 函數(shù)的功能是進(jìn)行一定次數(shù)的原子操作（自旋），如果超過這個(gè)次數(shù)就表示現(xiàn)在這個(gè)鎖的競爭比較激烈為了更好的使用 CPU 的計(jì)算資源可以將這個(gè)線程掛起。如果在自旋（spin）的時(shí)候發(fā)現(xiàn)鎖的值等于 val 那么就返回 0 ，如果在進(jìn)行 count 次操作之后我們還沒有發(fā)現(xiàn)鎖的值變成 val  那么就返回 1 ，這就表示鎖的競爭比較激烈。

7.可能你會(huì)疑惑在函數(shù) gomp_mutex_lock_slow 的最后一部分為什么要用 while 循環(huán)，這是因?yàn)?do_wait 函數(shù)不一定會(huì)將線程掛起，這個(gè)和 futex 系統(tǒng)調(diào)用有關(guān)，感興趣的同學(xué)可以去看一下 futex 的文檔，就了解這么設(shè)計(jì)的原因了。

8.在上面的源代碼當(dāng)中有兩個(gè) OpenMP 內(nèi)部全局變量，gomp_throttled_spin_count_var 和 gomp_spin_count_var 用于表示自旋的次數(shù)，這個(gè)也是 OpenMP 自己進(jìn)行設(shè)計(jì)的這個(gè)值和環(huán)境變量 OMP_WAIT_POLICY  也有關(guān)系，具體的數(shù)值也是設(shè)計(jì)團(tuán)隊(duì)的經(jīng)驗(yàn)值，在這里就不介紹這一部分的源代碼了。

其實(shí)上面的加鎖過程是非常復(fù)雜的，大家可以自己自行去好好分析一下這其中的設(shè)計(jì)，其實(shí)是非常值得學(xué)習(xí)的，上面的加鎖代碼貫徹的宗旨就是：能不進(jìn)內(nèi)核態(tài)就別進(jìn)內(nèi)核態(tài)。

omp_unset_lock，這個(gè)函數(shù)的主要功能就是解鎖了，我們再來看一下他的源代碼設(shè)計(jì)。這個(gè)函數(shù)最終調(diào)用的 OpenMP 內(nèi)部的函數(shù)為 gomp_mutex_unlock ，其源代碼如下所示：

static?inline?void
gomp_mutex_unlock?(gomp_mutex_t?*mutex)
{
??int?wait?=?__atomic_exchange_n?(mutex,?0,?MEMMODEL_RELEASE);
??if?(__builtin_expect?(wait?<?0,?0))
????gomp_mutex_unlock_slow?(mutex);
}

在上面的函數(shù)當(dāng)中調(diào)用一個(gè)函數(shù) gomp_mutex_unlock_slow ，其源代碼如下：

void
gomp_mutex_unlock_slow?(gomp_mutex_t?*mutex)
{
??//?表示喚醒?1?個(gè)線程
??futex_wake?(mutex,?1);
}

static?inline?void
futex_wake?(int?*addr,?int?count)
{
??int?err?=?syscall?(SYS_futex,?addr,?gomp_futex_wake,?count);
??if?(__builtin_expect?(err?<?0?&&?errno?==?ENOSYS,?0))
????{
??????gomp_futex_wait?&=?~FUTEX_PRIVATE_FLAG;
??????gomp_futex_wake?&=?~FUTEX_PRIVATE_FLAG;
??????syscall?(SYS_futex,?addr,?gomp_futex_wake,?count);
????}
}

在函數(shù) gomp_mutex_unlock 當(dāng)中首先調(diào)用原子操作 __atomic_exchange_n，將鎖的值變成 0 也就是無鎖狀態(tài)，這個(gè)其實(shí)是方便被喚醒的線程能夠不被阻塞（關(guān)于這一點(diǎn)大家可以好好去分分析 gomp_mutex_lock_slow 最后的 while 循環(huán)，就能夠理解其中的深意了），然后如果 mutex 原來的值（這個(gè)值會(huì)被賦值給 wait ）小于 0 ，我們在前面已經(jīng)談到過，這個(gè)值只能是 -1，這就表示之前有線程進(jìn)入內(nèi)核態(tài)被掛起了，因此這個(gè)線程需要喚醒之前被阻塞的線程，好讓他們能夠繼續(xù)執(zhí)行。喚醒之前線程的函數(shù)就是 gomp_mutex_unlock_slow，在這個(gè)函數(shù)內(nèi)部會(huì)調(diào)用 futex_wake 去真正的喚醒一個(gè)之前被鎖阻塞的線程。

omp_test_lock，這個(gè)函數(shù)主要是使用原子指令看是否能夠獲取鎖，而不嘗試進(jìn)入內(nèi)核，如果成功獲取鎖返回 1 ，否則返回 0 。這個(gè)函數(shù)在 OpenMP 內(nèi)部會(huì)最終調(diào)用下面的函數(shù)。

int
gomp_test_lock_30?(omp_lock_t?*lock)
{
??int?oldval?=?0;

??return?__atomic_compare_exchange_n?(lock,?&oldval,?1,?false,
??????????MEMMODEL_ACQUIRE,?MEMMODEL_RELAXED);
}

從上面源代碼來看這函數(shù)就是做了原子的比較并交換操作，如果成功就是獲取鎖并且返回值為 1 ，反之沒有獲取鎖那么就不成功返回值就是 0 。

總的說來上面的鎖的設(shè)計(jì)主要有一下的兩個(gè)方向：

• Fast path : 能夠在用戶態(tài)解決的事兒就別進(jìn)內(nèi)核態(tài)，只要能夠通過原子指令獲取鎖，那么就使用原子指令，因?yàn)檫M(jìn)入內(nèi)核態(tài)是一件非常耗時(shí)的事情。
• Slow path : 當(dāng)經(jīng)歷過一定數(shù)目的自旋操作之后發(fā)現(xiàn)還是不能夠獲得鎖，那么就能夠判斷此時(shí)鎖的競爭比較激烈，如果這個(gè)時(shí)候還不將線程掛起的話，那么這個(gè)線程好就會(huì)一直消耗 CPU ，因此這個(gè)時(shí)候我們應(yīng)該要進(jìn)入內(nèi)核態(tài)將線程掛起以節(jié)省 CPU 的計(jì)算資源。

雜談：

• 其實(shí)上面的鎖的設(shè)計(jì)是非公平的我們可以看到在 gomp_mutex_unlock 函數(shù)當(dāng)中，他是直接將 mutex 和 0 進(jìn)行交換，根據(jù)前面的分析現(xiàn)在的鎖處于一個(gè)沒有線程獲取的狀態(tài)，如果這個(gè)時(shí)候有其他線程進(jìn)來那么就可以直接通過原子操作獲取鎖了，而這個(gè)線程如果將之前被阻塞的線程喚醒，那么這個(gè)被喚醒的線程就會(huì)處于 gomp_mutex_lock_slow 最后的那個(gè)循環(huán)當(dāng)中，如果這個(gè)時(shí)候 mutex 的值不等于 0 （因?yàn)橛行聛淼木€程通過原子指令將 mutex 的值由 0 變成 1 了），那么這個(gè)線程將繼續(xù)阻塞，而且會(huì)將 mutex 的值設(shè)置成 -1。
• 上面的鎖設(shè)計(jì)加鎖和解鎖的交互情況是非常復(fù)雜的，因?yàn)樾枰_保加鎖和解鎖的操作不會(huì)造成死鎖，大家可以使用各種順序去想象一下代碼的執(zhí)行就能夠發(fā)現(xiàn)其中的巧妙之處了。
• 不要將獲取鎖和線程的喚醒關(guān)聯(lián)起來，線程被喚醒不一定獲得鎖，而且 futex 系統(tǒng)調(diào)用存在虛假喚醒的可能（關(guān)于這一點(diǎn)可以查看 futex 的手冊）。

深入分析 omp_nest_lock_t

在介紹可重入鎖（omp_nest_lock_t）之前，我們首先來介紹一個(gè)需求，看看之前的鎖能不能夠滿足這個(gè)需求。

#include?<stdio.h>
#include?<omp.h>

void?echo(int?n,?omp_nest_lock_t*?lock,?int?*?s)
{
???if?(n?>?5)
???{
??????omp_set_nest_lock(lock);
??????//?在這里進(jìn)行遞歸調(diào)用?因?yàn)樵谏弦恍写a已經(jīng)獲取鎖了?遞歸調(diào)用還需要獲取鎖
??????//?omp_lock_t?是不能滿足這個(gè)要求的?而?omp_nest_lock_t?能
??????echo(n?-?1,?lock,?s);
??????*s?+=?1;
??????omp_unset_nest_lock(lock);
???}
???else
???{
??????omp_set_nest_lock(lock);
??????*s?+=?n;
??????omp_unset_nest_lock(lock);
???}
}

int?main()
{
???int?n?=?100;
???int?s?=?0;
???omp_nest_lock_t?lock;
???omp_init_nest_lock(&lock);
???echo(n,?&lock,?&s);
???printf("s?=?%d\n",?s);
???omp_destroy_nest_lock(&lock);

???printf("%ld\n",?sizeof?(omp_nest_lock_t));
???return?0;
}

在上面的代碼當(dāng)中會(huì)調(diào)用函數(shù) echo，而在 echo 函數(shù)當(dāng)中會(huì)進(jìn)行遞歸調(diào)用，但是在遞歸調(diào)用之前線程已經(jīng)獲取鎖了，如果進(jìn)行遞歸調(diào)用的話，因?yàn)橹斑@個(gè)鎖已經(jīng)被獲取了，因此如果再獲取鎖的話就會(huì)產(chǎn)生死鎖，因?yàn)榫€程已經(jīng)被獲取了。

如果要解決上面的問題就需要使用的可重入鎖了，所謂可重入鎖就是當(dāng)一個(gè)線程獲取鎖之后，如果這個(gè)線程還想獲取鎖他仍然能夠獲取到鎖，而不會(huì)產(chǎn)生死鎖的現(xiàn)象。如果將上面的鎖改成可重入鎖 omp_nest_lock_t 那么程序就會(huì)正常執(zhí)行完成，而不會(huì)產(chǎn)生死鎖。

#include?<stdio.h>
#include?<omp.h>

void?echo(int?n,?omp_nest_lock_t*?lock,?int?*?s)
{
???if?(n?>?5)
???{
??????omp_set_nest_lock(lock);
??????echo(n?-?1,?lock,?s);
??????*s?+=?1;
??????omp_unset_nest_lock(lock);
???}
???else
???{
??????omp_set_nest_lock(lock);
??????*s?+=?n;
??????omp_unset_nest_lock(lock);
???}
}

int?main()
{
???int?n?=?100;
???int?s?=?0;
???omp_nest_lock_t?lock;
???omp_init_nest_lock(&lock);
???echo(n,?&lock,?&s);
???printf("s?=?%d\n",?s);
???omp_destroy_nest_lock(&lock);
???return?0;
}

上面的各個(gè)函數(shù)的使用方法和之前的 omp_lock_t 的使用方法是一樣的：

• 鎖的初始化 —— init 。
• 加鎖 —— set_lock。
• 解鎖 —— unset_lock 。
• 鎖的釋放 —— destroy 。

我們現(xiàn)在來分析一下 omp_nest_lock_t 的實(shí)現(xiàn)原理，首先需要了解的是 omp_nest_lock_t 這個(gè)結(jié)構(gòu)體一共占用 16 個(gè)字節(jié)，這 16個(gè)字節(jié)的字段如下所示：

typedef?struct?{?
??int?lock;?
??int?count;?
??void?*owner;?
}?omp_nest_lock_t;

上面的結(jié)構(gòu)體一共占 16 個(gè)字節(jié)現(xiàn)在我們來仔細(xì)分析以上面的三個(gè)字段的含義：

• lock，這個(gè)字段和上面談到的 omp_lock_t 是一樣的作用都是占用 4 個(gè)字節(jié)，主要是用于原子操作。
• count，在前面我們已經(jīng)談到了 omp_nest_lock_t 同一個(gè)線程在獲取鎖之后仍然能夠獲取鎖，因此這個(gè)字段的含義就是表示線程獲取了多少次鎖。
• owner，這個(gè)字段的含義就比較簡單了，我們需要記錄是哪個(gè)線程獲取的鎖，這個(gè)字段的意義就是執(zhí)行獲取到鎖的線程。
• 這里大家只需要稍微了解一下這幾個(gè)字段的含義，在后面分析源代碼的時(shí)候大家就能夠體會(huì)到這其中設(shè)計(jì)的精妙之處了。

omp_nest_lock_t 源碼分析

omp_init_nest_lock，這個(gè)函數(shù)的作用主要是進(jìn)行初始化操作，將 omp_nest_lock_t 中的數(shù)據(jù)中所有的比特全部變成 0 。在 OpenMP 內(nèi)部中最終會(huì)調(diào)用下面的函數(shù)：

void
gomp_init_nest_lock_30?(omp_nest_lock_t?*lock)
{
??//?字符?'\0'?對應(yīng)的數(shù)值就是?0?這個(gè)就是將?lock?指向的?16?個(gè)字節(jié)全部清零
??memset?(lock,?'\0',?sizeof?(*lock));
}

omp_set_nest_lock，這個(gè)函數(shù)的主要作用就是加鎖，在 OpenMP 內(nèi)部最終調(diào)用的函數(shù)如下所示：

void
gomp_set_nest_lock_30?(omp_nest_lock_t?*lock)
{
??//?首先獲取當(dāng)前線程的指針
??void?*me?=?gomp_icv?(true);
?//?如果鎖的所有者不是當(dāng)前線程，那么就調(diào)用函數(shù)?gomp_mutex_lock?去獲取鎖
??//?這里的?gomp_mutex_lock?函數(shù)和我們之前在?omp_lock_t?當(dāng)中所分析的函數(shù)
??//?是同一個(gè)函數(shù)
??if?(lock->owner?!=?me)
????{
??????gomp_mutex_lock?(&lock->lock);
?????//?當(dāng)獲取鎖成功之后將當(dāng)前線程的所有者設(shè)置成自己
??????lock->owner?=?me;
????}
?//?因?yàn)楂@取鎖了所以需要將當(dāng)前線程獲取鎖的次數(shù)加一
??lock->count++;
}

在上面的程序當(dāng)中主要的流程如下：

• 如果當(dāng)前鎖的所有者是自己，也就是說如果當(dāng)前線程之前已經(jīng)獲取到鎖了，那么久直接將 count 進(jìn)行加一操作。
• 如果當(dāng)線程還還沒有獲取到鎖，那么就使用 gomp_mutex_lock 去獲取鎖，如果當(dāng)前已經(jīng)有線程獲取到鎖了，那么線程就會(huì)被掛起。
• omp_unset_nest_lock

void
gomp_unset_nest_lock_30?(omp_nest_lock_t?*lock)
{
??if?(--lock->count?==?0)
????{
??????lock->owner?=?NULL;
??????gomp_mutex_unlock?(&lock->lock);
????}
}

在由了 omp_lock_t 的分析基礎(chǔ)之后上面的代碼也是比較容易分析的，首先會(huì)將 count 的值減去一，如果 count 的值變成 0，那么就可以進(jìn)行解鎖操作，將鎖的所有者變成 NULL ，然后使用 gomp_mutex_unlock 函數(shù)解鎖，喚醒之前被阻塞的線程。

omp_test_nest_lock

int
gomp_test_nest_lock_30?(omp_nest_lock_t?*lock)
{
??void?*me?=?gomp_icv?(true);
??int?oldval;

??if?(lock->owner?==?me)
????return?++lock->count;

??oldval?=?0;
??if?(__atomic_compare_exchange_n?(&lock->lock,?&oldval,?1,?false,
???????MEMMODEL_ACQUIRE,?MEMMODEL_RELAXED))
????{
??????lock->owner?=?me;
??????lock->count?=?1;
??????return?1;
????}

??return?0;
}

這個(gè)不進(jìn)入內(nèi)核態(tài)獲取鎖的代碼也比較容易，首先分析當(dāng)前鎖的擁有者是不是當(dāng)前線程，如果是那么就將 count 的值加一，否則就使用原子指令看看能不能獲取鎖，如果能夠獲取鎖就返回 1 ，否則就返回 0 。

源代碼函數(shù)名稱不同的原因揭秘

在上面的源代碼分析當(dāng)中我們可以看到我們真正分析的代碼并不是在 omp.h 的頭文件當(dāng)中定義的，這是因?yàn)樵?OpenMP 內(nèi)部做了很多的重命名處理：

#?define?gomp_init_lock_30?omp_init_lock
#?define?gomp_destroy_lock_30?omp_destroy_lock
#?define?gomp_set_lock_30?omp_set_lock
#?define?gomp_unset_lock_30?omp_unset_lock
#?define?gomp_test_lock_30?omp_test_lock
#?define?gomp_init_nest_lock_30?omp_init_nest_lock
#?define?gomp_destroy_nest_lock_30?omp_destroy_nest_lock
#?define?gomp_set_nest_lock_30?omp_set_nest_lock
#?define?gomp_unset_nest_lock_30?omp_unset_nest_lock
#?define?gomp_test_nest_lock_30?omp_test_nest_lock

在 OponMP 當(dāng)中一個(gè)跟鎖非常重要的文件就是 lock.c，現(xiàn)在查看一下他的源代碼，你的疑惑就能夠揭開了：

#include?<string.h>
#include?"libgomp.h"

/*?The?internal?gomp_mutex_t?and?the?external?non-recursive?omp_lock_t
???have?the?same?form.??Re-use?it.??*/

void
gomp_init_lock_30?(omp_lock_t?*lock)
{
??gomp_mutex_init?(lock);
}

void
gomp_destroy_lock_30?(omp_lock_t?*lock)
{
??gomp_mutex_destroy?(lock);
}

void
gomp_set_lock_30?(omp_lock_t?*lock)
{
??gomp_mutex_lock?(lock);
}

void
gomp_unset_lock_30?(omp_lock_t?*lock)
{
??gomp_mutex_unlock?(lock);
}

int
gomp_test_lock_30?(omp_lock_t?*lock)
{
??int?oldval?=?0;

??return?__atomic_compare_exchange_n?(lock,?&oldval,?1,?false,
??????????MEMMODEL_ACQUIRE,?MEMMODEL_RELAXED);
}

void
gomp_init_nest_lock_30?(omp_nest_lock_t?*lock)
{
??memset?(lock,?'\0',?sizeof?(*lock));
}

void
gomp_destroy_nest_lock_30?(omp_nest_lock_t?*lock)
{
}

void
gomp_set_nest_lock_30?(omp_nest_lock_t?*lock)
{
??void?*me?=?gomp_icv?(true);

??if?(lock->owner?!=?me)
????{
??????gomp_mutex_lock?(&lock->lock);
??????lock->owner?=?me;
????}

??lock->count++;
}

void
gomp_unset_nest_lock_30?(omp_nest_lock_t?*lock)
{
??if?(--lock->count?==?0)
????{
??????lock->owner?=?NULL;
??????gomp_mutex_unlock?(&lock->lock);
????}
}

int
gomp_test_nest_lock_30?(omp_nest_lock_t?*lock)
{
??void?*me?=?gomp_icv?(true);
??int?oldval;

??if?(lock->owner?==?me)
????return?++lock->count;

??oldval?=?0;
??if?(__atomic_compare_exchange_n?(&lock->lock,?&oldval,?1,?false,
???????MEMMODEL_ACQUIRE,?MEMMODEL_RELAXED))
????{
??????lock->owner?=?me;
??????lock->count?=?1;
??????return?1;
????}

??return?0;
}

總結(jié)

在本篇文章當(dāng)中主要給大家分析了 OpenMP 當(dāng)中兩種主要的鎖的實(shí)現(xiàn)，分別是 omp_lock_t 和 omp_nest_lock_t，一種是簡單的鎖實(shí)現(xiàn)，另外一種是可重入鎖的實(shí)現(xiàn)。其實(shí) critical 子句在 OpenMP 內(nèi)部的也是利用上面的鎖實(shí)現(xiàn)的。整個(gè)鎖的實(shí)現(xiàn)還是非常復(fù)雜的，里面有很多耐人尋味的細(xì)節(jié)，這些代碼真的很值得一讀，看看能操刀 OpenMP Runtime Library 這些編程大師的作品，真的很有收獲。

以上就是深入剖析OpenMP鎖的原理與實(shí)現(xiàn)的詳細(xì)內(nèi)容，更多關(guān)于OpenMP鎖的資料請關(guān)注腳本之家其它相關(guān)文章！

最新評論