碼上快樂

相關內容    簡體    繁體

Erlang運行時源碼分析之——線程進度機制

本文轉載自   查看原文   2013-01-27 00:54   3920    Erlang運行時代碼分析/ Erlang/ ERTS

歡迎下載本文精美排版的的pdf版本： http://vdisk.weibo.com/s/oIICP

1 概述

線程進度跟蹤機制（thread progress）是Erts 5.9引入的一個重要的內部改進，如release notes中提到的:

The ERTS internal system block functionality has been replaced by new functionality for blocking the system. The old system block functionality had contention issues and complexity issues. The new functionality piggy-backs on thread progress tracking functionality needed by newly introduced lock-free synchronization in the runtime system. When the functionality for blocking the system isn't used, there is more or less no overhead at all. This since the functionality for tracking thread progress is there and needed anyway.

ERTS采用了全新的系統阻塞功能。之前的系統阻塞功能可能會產生爭用的問題，而且非常復雜。新的系統阻塞功能依賴於“線程進度跟蹤機制”，這也是運行時中新引入的無鎖同步依賴的機制。如果沒有使用系統阻塞的功能，那么幾乎不會有什么開銷，因為線程進度跟蹤的機制總是在那。

這是Erlang運行時提升在多核（甚至眾核）平台上性能的眾多改進之一。本文分析線程進度跟蹤機制的原理及其代碼，並分析使用了線程進度跟蹤機制的系統阻塞相關的api代碼。下一篇博文將分析新引入的“無鎖隊列（lock-free queue）”的原理和代碼，即上述引用中提到的無鎖同步機制。

本文基於R15B02版本進行分析。

2 線程進度跟蹤機制

無鎖算法可以通過線程進度值來判斷是否所有相關的線程都已經完成了執行的某個特定的進度點。Erts中的線程分為兩類：受管（managed）線程和非受管（unmanaged）線程。線程進度機制只跟蹤受管線程的進度值。在Erts中，目前只有三類線程是受管線程：

  • 調度器線程（erts/emulator/beam/erl_process.c/sched_thread_func()函數表示的線程）
  • 完成輔助工作的線程（aux線程，erts/emulator/beam/erl_process.c/aux_thread()函數表示的線程）
  • 系統消息分發線程（erts/emulator/beam/erl_trace.c/sys_msg_dispatcher_func()函數表示的線程）。

調度器線程的數目是可配置的，其他兩種線程都只有1個，因此受管線程的總數是調度器數目+2。受管線程都是“行為良好”的，可以保證以一定的頻度更新自己的進度值。

非受管線程目前包括Erlang虛擬機中的異步線程，也就是虛擬機+A參數配置的異步線程池（async thread pool）中的線程，這些線程由於不能保證一定頻度的更新進度，所以是不受管的線程。

每一個線程都有一個私有的進度值，還有一個全局的進度值。全局的進度值表示所有的線程都已經達到的進度值。下面是進度值的規則：

  • 在受管的線程集合中，有且只有一個線程是“領導（leader）”線程
  • 每一個受管線程都在固定的位置更新自己的進度值
  • 領導線程除了更新自己的進度值之外，還要更新全局進度值
  • 所有受管線程的進度值和全局進度值初始化為0
  • 線程運行到更新點的時候更新自己的進度值，但是這個進度值不超過全局進度值+1
  • 領導線程更新完了自己的進度值之后，要檢查所有受管線程的進度值是否都達到了全局進度值+1，如果達到了，則更新全局進度值+1
  • 如果沒有領導線程，那么第一個發現這個事實的受管線程要搶先爭當領導。當然如果有多個受管線程同時爭當領導，要通過原子操作保證只有一個領導產生
  • 如果有線程睡眠，那么這個線程要設置特殊的進度值，因此在睡覺的線程不會影響領導更新全局進度
  • 如果線程進度值達到最大值了，則繞回到0。由於上面的幾條進度值規則，就算有線程的進度值繞回了，也不會影響進度之間的比較。在Erts中，用無符號的64位整數表示進度值

從以上規則可以看出，全局進度值小於等於所有線程當前的進度值。如果有線程運行很快，那么快線程的進度值更新一次之后不會再更新，而是會等全局進度值更新了之后再更新，所以可以看出線程之間的進度值最大相差1。

不同類型的受管線程在固定的位置更新線程進度。目前調度器線程在以下時間點會更新線程進度：

  • 前一個Erlang進程調度出之后，下一個Erlang進程調度執行之前
  • 要進入睡眠的時候
  • 喚醒的時候

輔助線程在完成一次輔助工作之后就更新一次線程進度。系統消息分發線程在每發送一條trace消息的時候就更新一次線程進度。

以上是線程進度跟蹤機制的基本原理，下面分析線程進度跟蹤機制模塊提供的api以及具體的代碼分析。本文只是介紹原理和分析代碼，而沒有介紹Erlang運行時如何通過這個機制實現無鎖的同步機制，敬請期待后續的博文：）

3 線程進度跟蹤機制模塊提供的api及實現

線程進度跟蹤機制相關的api和代碼分布在頭文件 erts/emulator/beam/erl_thr_process.h 和實現文件 erts/emulator/beam/erl_thr_process.c 中。這個模塊提供了以下api給整個Erlang運行時調用，了解了這些api調用的作用之后就可以理解實現的代碼了：

初始化類

  • void erts_thr_progress_pre_init(void):在erl_init.c/early_init() 中調用,做早期初始 化,創建線程進度數據相關的 tsd key,初始化全局進度值為 0。
  • void erts_thr_progress_init(int no_schedulers, int managed, int unmanaged):初始化線 程進度跟蹤機制使用的全局數據結構,詳見 3.1 小節對數據結構的描述。
  • void erts_thr_progress_register_managed_thread(ErtsSchedulerData *esdp, ErtsThrPrgrCallbacks *callbacks, int pref_wakeup):受管線程在系統中注冊,登記回調函數,創建線程 私有的進度數據,初始化數據結構。
  • void erts_thr_progress_register_unmanaged_thread(ErtsThrPrgrCallbacks * callbacks): 非受管的線程在系統中注冊,登記回調函數,創建線程私有的進度數據,初始化數據結構。

更新狀態類

  • int erts_thr_progress_update(ErtsSchedulerData *esdp):受管線程更新自己的進度值。返回結果表明當前線程是否是領導線程。如果是領導線程,還需要調用erts_thr_progress_leader_update。
  • int erts_thr_progress_leader_update(ErtsSchedulerData *esdp):領導線程更新自己的進度值和全局進度值。
  • void erts_thr_progress_active(ErtsSchedulerData *esdp, int on)、void erts_thr_progress_prepare_wait(ErtsSchedulerData *esdp)、void erts_thr_progress_finalize_wait(ErtsSchedulerData *esdp):如果受管線程需要等待 某個事件發生而需要進入睡眠狀態,那么在睡眠之前,需要調用erts_thr_progress_active 將線程狀態設置為非活躍,然后調用erts_thr_progress_prepare_wait 通知運行時進行線程睡眠之前的准備活動(例如設置線程的進度值為等待狀態)。線程被喚醒之后,要及時調用erts_thr_progress_finalize_wait 通知運行時進行線程喚醒之后的活動(例如恢復線程的進 度值)。然后調用erts_thr_progress_active 將線程狀態設置為活躍。
  • void erts_thr_progress_wakeup(ErtsSchedulerData *esdp, ErtsThrPrgrVal value):受管線 程和非受管線程都可以通過這個調用請求運行時在全局進度達到(或超越,不能保證准確地在達 到的時候)給定值value 的時候喚醒自己。調用之后線程就可以在線程事件上睡覺等待被運行時喚醒。運行時會在內部數據結構中注冊線程的請求,每次更新全局進度的時候如果發現到達了線程請求的進度值,則喚醒相應的線程。

系統阻塞類

  • void erts_thr_progress_block(void):調用的受管線程將其他受管線程阻塞。調用這個函數之后,其他受管線程在執行到下一次進度更新點的時候會發現這個線程的阻塞請求,從而進入阻塞狀態。因為有時間差的存在,所以調用的受管線程會等待其他受管線程都已經進入了阻塞狀態。 這個函數返回的時候可以保證其他受管線程都已經成功阻塞。調用的受管線程可以執行一些排他的操作。
  • void erts_thr_progress_unblock(void):受管線程在執行完排他的操作之后,調用這個函數解 除系統的阻塞,將其他被阻塞的受管線程喚醒。這是 Erts 5.9 引入的新的阻塞系統,替換了之前復雜且易產生爭用的阻塞系統。
  • int erts_thr_progress_is_blocking(void):判斷當前系統是否正在阻塞。顯然這個 api 沒多大作用。被阻塞的線程沒機會調用,阻塞別人的線程自己還不知道是不是在阻塞么。目前 Erlang 運行時中只有一些調試代碼使用了這個 api。

其他狀態判斷類

  • int erts_thr_progress_is_managed_thread(void):判斷當前線程是否是受管線程。
  • ErtsThrPrgrVal erts_thr_progress_later(ErtsSchedulerData *):返回一個未來的進度值,當前還沒有受管線程達到這個進度值。實際上對於受管線程,返回的是受管線程當前進度值 +2,對於非受管線程,返回的是當前全局值 +2。根據進度值的規則,這樣可以保證返回的一定是一個未 來的進度值,盡管不一定是最小的。
  • ErtsThrPrgrVal erts_thr_progress_current(void):返回當前的全局進度值。
  • int erts_thr_progress_has_reached_this(ErtsThrPrgrVal this, ErtsThrPrgrVal val):判 斷進度值 val 是否已經超越了進度值 this。
  • int erts_thr_progress_equal(ErtsThrPrgrVal val1, ErtsThrPrgrVal val2):判斷進度值 val1 是否等於進度值 val2。
  • int erts_thr_progress_cmp(ErtsThrPrgrVal val1, ErtsThrPrgrVal val2):比較兩個進度值 val1 和 val2 的關系。如果相等,則返回 0,如果 val2 超過了 val1,則返回 1,反之返回 -1。 這個比較 api 考慮了進度值超過了最大值繞回的情況。
  • int erts_thr_progress_has_reached(ErtsThrPrgrVal val):判斷當前的全局進度值是否達到 了 val,同樣考慮了繞回的情況。

3.1 數據結構和初始化

3.1.1 運行時的公共管理數據結構

下面是運行時線程進度跟蹤機制使用的管理數據結構,這些數據是全局數據,圖 1 展示了這些數據結構之間的關系。

 1 typedef struct {
 2     erts_atomic32_t len;
 3     int id[1];
 4 } ErtsThrPrgrManagedWakeupData;
 5 
 6 typedef struct {
 7     erts_atomic32_t len;
 8     int high_sz;
 9     int low_sz;
10     erts_atomic32_t *high;
11     erts_atomic32_t *low;
12 } ErtsThrPrgrUnmanagedWakeupData;
13 
14 typedef struct {
15     erts_atomic32_t lflgs;
16     erts_atomic32_t block_count;
17     erts_atomic_t blocker_event;
18     erts_atomic32_t pref_wakeup_used;
19     erts_atomic32_t managed_count;
20     erts_atomic32_t managed_id;
21     erts_atomic32_t unmanaged_id;    
22 } ErtsThrPrgrMiscData;
23 
24 typedef struct {
25     ERTS_THR_PRGR_ATOMIC current;
26 } ErtsThrPrgrElement;
27 
28 typedef union {
29     ErtsThrPrgrElement data;
30     char align__[ERTS_ALC_CACHE_LINE_ALIGN_SIZE(sizeof(ErtsThrPrgrElement))];
31 } ErtsThrPrgrArray;
32 
33 typedef struct {
34     void *arg;
35     void (*wakeup)(void *);
36     void (*prepare_wait)(void *);
37     void (*wait)(void *);
38     void (*finalize_wait)(void *);
39 } ErtsThrPrgrCallbacks;
40 
41 typedef struct {
42     union {
43         ErtsThrPrgrMiscData data;
44         char align__[ERTS_ALC_CACHE_LINE_ALIGN_SIZE(
45                 sizeof(ErtsThrPrgrMiscData))];
46     } misc;
47     ErtsThrPrgrArray *thr;
48     struct {
49         int no;
50         ErtsThrPrgrCallbacks *callbacks;
51         ErtsThrPrgrManagedWakeupData *data[ERTS_THR_PRGR_WAKEUP_DATA_SIZE];
52     } managed;
53     struct {
54         int no;
55         ErtsThrPrgrCallbacks *callbacks;
56         ErtsThrPrgrUnmanagedWakeupData *data[ERTS_THR_PRGR_WAKEUP_DATA_SIZE];
57     } unmanaged;
58 } ErtsThrPrgrInternalData;
59 
60 static ErtsThrPrgrInternalData *intrnl;
61 ErtsThrPrgr erts_thr_prgr__;
62 erts_tsd_key_t erts_thr_prgr_data_key__;

 

 

圖 1: 線程進度跟蹤機制使用的共享數據結構（點擊小圖看大圖）

線程進度機制使用的數據結構

圖中的灰色 padding 部分是填充區域,用於將所在的數據結構填滿一條緩存線。從圖中可以看出, intrnl 是所有數據結構的根,指向 ErtsThrPrgrInternalData 結構體。在 ErtsThrPrgrInternalData 結構體中:

  • misc 的類型為 ErtsThrPrgrMiscData,顧名思義,就是一些不好分類的管理數據,后面會具體分析每一個字段的意義。
  • thr 是指向 ErtsThrPrgrArray 數組的指針。這個數組中的每一項實際上就是 ErtsThrPrgrElement 填充滿一條緩存線的內容,實際有用的數據是 64 位的原子變量 current。這個數組的長度等於受 管線程的數目,每一項表示一個受管線程當前的進度值。在代碼中,每一項值只能被其表示的那 個受管線程寫入,而其他線程只能讀取。
  • managed 字段管理了和受管線程相關的數據。
  • unmanaged 字段管理了非受管線程相關的數據。后面會具體分析這兩個數據結構。

thr 數組是一項重要的優化,在使用寫回策略(write-back policy)的處理器上,線程對自己進度的 更新甚至不需要寫入內存。例如,假設有 4 個受管線程,線程 1 是領導,而且每一個線程都運行在一 個處理器核心上,每一個處理器核心都有自己的私有緩存。所有線程都更新一次之后在每一個處理器核心的私有緩存中都有一條完整的緩存線中保存的有用數據只包括該線程當前的進度值,將這種更新頻繁 的數據單獨放在一條緩存線中可以避免偽共享。當領導線程要更新全局進度值的時候,需要讀取所有受 管線程的當前進度值。由於緩存一致性協議的作用,讀取之后在領導線程所在的處理器核心的私有緩存 中,會有每一個受管線程進度值所在的那條緩存線的副本。這時如果有一個普通受管線程要更新進度, 如果緩存使用了寫回策略(而不是寫穿策略,write-through policy),那么這個線程所在的處理器核心 會更新這個進度值所在的緩存線,並且緩存一致性協議通過緩存之間的高速網絡通知領導所在的私有緩 存這一個進度值所在的緩存線失效,除非這條緩存線被換出,否則不會將新的值寫入內存。當領導需要 訪問所有受管線程的進度值的時候,發現對應的緩存線失效,緩存一致性協議會通過緩存之間的高速網 絡從原本的緩存線中直接獲得最新的進度值。從此可以看出,在理想情況下,除了緩存預熱之外,整個 進度更新的過程都不需要讀寫內存,所有的數據訪問都通過緩存一致性協議在處理器內部的高速網絡上 完成了,因此在核數很多的多核處理器上這個進度更新的機制也能高效率地工作。

managed 字段包含了兩個數組:

  • ErtsThrPrgrCallbacks 數組 callbacks 包含受管線程數目個元素,每一個元素是對應受管線程 在注冊的時候登記的回調函數,每一個線程都可以登記自己私有的回調函數。稍后會分析回調函 數的作用。
  • 指向ErtsThrPrgrManagedWakeupData指針的數組data,一共有ERTS_THR_PRGR_WAKEUP_DATA_SIZE 個元素,用於登記和請求喚醒相關的數據。稍后會詳細分析這個數組的結構。

先說說回調函數 ErtsThrPrgrCallbacks 結構體。這個結構體就像一個閉包一樣,包含 4 個回調函 數和一個參數,這 4 個回調函數是:

  • wakeup:用於喚醒該線程。
  • prepare_wait:在睡該線程之前需要完成的准備工作。
  • wait:將該線程置入睡眠狀態。
  • finalize_wait:喚醒該線程之后需要完成的恢復工作。

這些回調函數都和睡覺有關。結構體中包含的參數就是在調用這些回調函數的時候傳遞進去的參數。為什么需要這些回調函數呢?肯定有人會發現這些回調函數名字有些眼熟,那么這些回調函數和 erts_thr_progress_active、erts_thr_progress_prepare_wait、erts_thr_progress_finalize_wait 以及 erts_thr_progress_wakeup 這幾個 api 函數的作用差別在哪呢?

既然是回調函數,那么從設計的角度來說,如果 B 模塊向 A 模塊提供回調函數 func,說明 func 是 A 需要讓 B 執行的操作,但是 A 只知道需要進行這個操作,不知道這個操作的具體做法,所以具體的實現由 B 提供。那么在這里也是一樣,回調函數是由系統阻塞功能使用的。當有線程調用 api 函數 erts_thr_progress_block 要求運行時阻塞系統的時候,運行時要讓其他受管線程進入睡眠等待 的狀態,因而其他線程是被動進入睡眠狀態。這里提供回調函數接口的目的是為了讓線程被動睡眠前 后能完成必要的數據維護操作。而 erts_thr_progress_active、erts_thr_progress_prepare_wait、 erts_thr_progress_finalize_wait 以及 erts_thr_progress_wakeup 這幾個 api 函數是線程自己因為 種種原因需要主動進入睡眠狀態的時候,通過調用這些 api 函數通知運行時,讓運行時做好相關的數據 維護操作。

這些回調函數調用時傳入的參數一般都設置為對應線程的事件(thread-specific event,在代碼中常 縮寫為 tse),因為 Erlang 運行時中通常通過事件等待機制實現線程的睡眠。

下面再看請求喚醒數據。請求喚醒是線程進度跟蹤機制提供的一種功能,受管線程或非受管線程 調用 erts_thr_progress_wakeup 請求運行時在特定的進度時喚醒線程。受管線程的請求數據就保存在 ErtsThrPrgrManagedWakeupData 數據結構中。從圖 1 中可以看出,一個 ErtsThrPrgrManagedWakeup- Data 包含一個表示長度的 len 和一個 id 數組。len 表示后面這個 id 數組中有效元素的個數,從前往 后每一個有效元素保存一個登記了喚醒信息受管線程的 id。data 數組的每一項表示一個特定進度值的 喚醒信息。可是進度值的取值空間那么大(64 位無符號整數,也就是 ),那么這個喚醒信息的數組 應該多大?這個數組是 ERTS_THR_PRGR_WAKEUP_DATA_SIZE 這么大,這個常量目前在 Erlang 虛擬機中定義為 4,只有 。在線程請求喚醒的時候,運行時對給定的進度值進行掩碼運算,只取了最后 2 個 bit,然后把線程 id 放進對應 ErtsThrPrgrManagedWakeupData 的 id 數組最后一個有效元素之后,並且增加 len 的值。這樣,只要是指定進度值低位 2 個 bit 都相同的線程都會放在一起。每到一個新的 全局進度值的時候,運行時會對當前全局進度值進行同樣的掩碼運算,得到一個索引,然后把這個索引 對應一個 ErtsThrPrgrManagedWakeupData 中的有效 id 全部喚醒,通過調用這些線程的 wakeup 回調 函數。很明顯,每次喚醒的線程可能比實際需要喚醒的線程要多,但是線程喚醒之后可以重新檢查睡眠等待的條件是否滿足,如果不滿足,繼續睡眠。

unmanaged 字段也類似地包含了兩個數組:回調函數數組 callbacks 和指向 ErtsThrPrgrUnmanagedWakeupData 指針的喚醒數據數組 data。

callbacks 數組的元素個數等於非受管線程的數目。由於非受管線程不參與系統阻塞功能,所以 非受管線程在注冊的時候一般不需要登記 prepare_wait、wait 和 finalize_wait 函數,而只需要 wakeup,因為非受管線程還可以使用請求喚醒機制,而這個機制調用 wakeup 回調函數喚醒線程。

非受管線程的喚醒數據 ErtsThrPrgrUnmanagedWakeupData 稍復雜一些,從圖 1 中可以看出這個數 據結構包含兩個數組:low 和 high,這兩個數組中每一個值都是 32 位原子值。low_sz 和 high_sz 分別表示這兩個數組的大小。len 還是表示對應的進度值有多少個請求喚醒的線程。

先看 low 數組的作用。在 Erlang 運行時中,異步線程就是非受管線程。Erlang 虛擬機允許通過 +A 參數指定異步線程的數目,目前這個參數允許取值范圍為 0 到 1024,默認值為 0,未來在眾核處理 器上 +A 參數的上限可能還會調整,也就是說,非受管線程數目可能很大。所以這里通過 bitmap 來表 示具體的線程。low 數組就是保存所有比特位的數組。一個線程對應一個比特位,low 數組中一個元素 可以表示 32 個線程。所以這個數組一共需要 unmanaged 個元素。那么 high 數組保存的又是什么呢?high 數組中的 bit 對應 low 數組中的元素就好像 low 數組中的 bit 對應非受管線程的 id,也 就是說,low 數組中的每一個元素在 high 數組中都有一個 bit 位對應。如果 low 數組中的某個元素中 有 bit 被設置了,那么在 high 數組中也要設置相應的位,實現了二級索引。目前 Erlang 虛擬機最多允 許 1024 個異步線程, ,所以目前 high 數組中只用到了一個元素。

下面再提一下 misc 字段的內容。ErtsThrPrgrMiscData 結構體中包含以下字段:

  • lflgs:標志位兼計數器。最高位是 ERTS_THR_PRGR_LFLG_BLOCK 標志,表示是否阻塞,接下來是 ERTS_THR_PRGR_LFLG_NO_LEADER 標志,表示是否還沒有領導。剩下的 30 位是活躍線程的計數器。
  • block_count:這個字段表示的意思並不是阻塞的計數器,而是系統阻塞功能中使用的一個計數 器,表示沒有阻塞的受管線程的數量。當這個計數器的值為 0 的時候,表示所有受管線程都阻塞 了。
  • blocker_event:阻塞者使用的事件。阻塞者在阻塞系統的時候,通過這個事件等待所有的受管線 程都阻塞了。
  • pref_wakeup_used:表示系統中是否已經設置了優先喚醒的受管線程。只能有一個線程是允許優 先喚醒的,這個線程在線程進度跟蹤機制中的 id 設置為 0。
  • managed_count:受管線程的數目。
  • managed_id 和unmanaged_id:在注冊線程分配 id 的時候使用。

3.1.2 線程私有數據結構

下面介紹用於線程進度跟蹤機制的線程私有數據結構。線程在注冊的時候創建這個數據結構,並 且以 erts_thr_prgr_data_key__ 作為鍵保存在自己的 TSD 中。這個數據簡稱 TPD(thread progress data)。

 1 typedef struct {
 2     int id;
 3     int is_managed;
 4     int is_blocking;
 5     int is_temporary;
 6     /* --- 以下字段是注冊的線程專用的 --- */
 7     ErtsThrPrgrVal wakeup_request[ERTS_THR_PRGR_WAKEUP_DATA_SIZE];
 8     /* --- 以下字段是受管線程專用的 --- */
 9     int leader;
10     int active;
11     struct {
12         ErtsThrPrgrVal local;
13         ErtsThrPrgrVal next;
14         ErtsThrPrgrVal current;
15     } previous;
16 } ErtsThrPrgrData;

 這個數據結構各個字段的意義如下:

  • id:這個線程在線程進度跟蹤機制下的編號。
  • is_managed:這個線程是否是受管線程。 
  • is_blocking:這個線程是否在阻塞系統。
  • is_temporary:表示這個 TPD 是否為臨時數據。在系統 crash dump 的時候會用到臨時 TPD,本文暫且不表。
  • wakeup_request:保存這個線程發出的喚醒請求。
  • leader:這個線程是否為領導線程。
  • active:這個線程是否處於活躍狀態。
  • previous:這個字段用於計算進度值。local 和 intrnl 中 thr 數組中保存的 current 值保持一致。next 和 current 用於領導線程計算下一個全局進度值。

了解了原理和數據結構之后,讀懂代碼就不是什么太困難的事情了。作為多線程的程序,最困難的 部分在於因為多個線程爭用而互相產生干擾的情況的處理。下面幾個小節主要分析一些代碼中比較麻煩的部分。

4 線程進度跟蹤機制代碼分析

先看進度更新的代碼。受管線程調用 erts_thr_progress_update 之后,這個函數調用 update 函數完成更新。如果受管線程是領導線程,則調用 leader_update 函數完成自己和全局的進度更新。在 update 函數中,如果通過檢查intrnl->misc.data.lflgs 標志發現系統正在阻塞,則應該讓 leader_update 阻塞線程。如果發現沒有領導線程存在,那么要嘗試通過一個原子操作去掉標志中的 ERTS_THR_PRGR_LFLG_NO_LEADER,如果操作成功,則說明自己成為了領導,而如果有並發線程也在搶奪 領導的話肯定會失敗。如果自己失敗了,則不做任何操作,說明有並發線程正在搶奪領導,而且肯定會 有線程搶奪成功。搶奪成功后,受管線程還應該調用 leader_update 函數進行全局更新。

在 leader_update 函數中,如果發現調用者並不是領導,說明要求阻塞,進入阻塞狀態。 leader_update 首先要像 update 那樣對自己的進度進行更新。然后檢查所有線程的當前進度值是 否已經達到了下一個應該達到的進度值。如果都達到了,說明產生了進度,更新全局進度值,並且檢查線程登記的請求喚醒信息,有的話則喚醒相應的線程。

 

5 系統阻塞機制的實現

系統阻塞通過是由 thr_progress_block 函數實現的。這個函數一開始有一個判斷:

 

1 if (tpd->is_blocking++)
2     return (erts_aint32_t) 0;

 

 

 把標志當計數器用。這是為了處理嵌套的情況。如果一個線程調用了一次阻塞之后,不知道自己已經阻塞了系統,然后又調用一次就可以直接返回。接下來一個 while 循環:

 

1 while (1) {
2     lflgs = erts_atomic32_read_bor_nob(&intrnl->misc.data.lflgs,
3                                        ERTS_THR_PRGR_LFLG_BLOCK);
4     if (lflgs & ERTS_THR_PRGR_LFLG_BLOCK)
5         block_thread(tpd);
6     else
7         break;
8 }

 

這是為了防止有多個線程同時調用系統阻塞的情況。阻塞系統的那個線程負責設置 ERTS_THR_PRGR_LFLG_BLOCK 標志位。如果設置失敗,則說明已經被別人搶先設置了,那么我自己只好先阻塞。喚醒之后,再去嘗試 設置標志位阻塞系統。如果設置成功了,則說明我成功獲得了阻塞系統的權力,退出這個 while 循環 繼續后面的操作。

block_count_dec 函數和 block_count_inc 函數負責 intrnl->misc.data.block_count 計數器的遞減和遞增。block_count_dec 發現計數器歸零的時候就知道大家都完成阻塞了,所以發送事件給阻塞者通知阻塞者喚醒。

 

 

6 請求喚醒機制的實現

線程請求喚醒的時候調用 erts_thr_progress_wakeup 函數,這個函數根據調用者是受管線程還是 非受管線程分別調用 request_wakeup_managed 和 request_wakeup_unmanaged 函數。 request_wakeup_managed 函數有一個地方需要處理爭用情況:

1 if (tpd->previous.local == value) {
2     value++;
3     if (value == ERTS_THR_PRGR_VAL_WAITING)
4         value = 0;
5 
6     wix = ERTS_THR_PRGR_WAKEUP_IX(value);
7     if (tpd->wakeup_request[wix] == value)
8         return; /* Already got a request registered */
9 }

如果發現請求喚醒的進度值剛好等於線程當前的進度值,那么要擅自將請求喚醒的進度值向前步進 1。 這是為了防止在注冊進度值的時候全局進度已經達到這個值了,以免失去被喚醒的機會。將請求喚醒的進度值增加 1 是安全的,因為全局進度步進到下一個值的時候最多等於這個線程的當前值,而在寫入請 求的這段時間,這個線程不可能執行更新進度的代碼,所以全局進度肯定不會更新到這個線程當前值的 下一個值。所以寫完請求值之后這個線程一定會被喚醒。

喚醒機制另一個麻煩一點的地方就是非受管線程的多級喚醒數據,但是了解了原理之后代碼也很容易看懂了。

 

 

 

 

 

 


免責聲明！

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



猜您在找 Erlang運行時的錯誤 Erlang運行時中的無鎖隊列及其在異步線程中的應用 Erlang運行時中的原子操作 Android9.0動態運行時權限源碼分析及封裝改造<二>-----權限檢測requestPermissions源碼分析 Java運行時環境---ClassLoader類加載機制 [轉]詳解Runtime運行時機制 runtime 運行時機制 + 應用場景 C++並發之運行時決定線程的數量 Java異常與運行時異常，以及與線程的關系 一個線程運行時發生異常會怎樣？
 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM