1. 程式人生 > >RT-thread 任務間同步及通訊

RT-thread 任務間同步及通訊

目錄

0. 任務間同步

1. 中斷(鎖)

1.2程式碼

1.3 注意事項

1.4 使用場合

2. 排程器鎖

2.1 程式碼

2.2 使用場合

3. 訊號量

3.1 程式碼操作

3.2 使用場合

4. 互斥量

4.1 示例:

4.2 使用場合:

5. 事件()

5.1 示例:

5.2 使用場合:

6. 郵箱(任務間通訊)

6.1 示例:

6.2 使用場合:

7. 訊息佇列(任務見通訊)

7.1 示例:

7.2 使用場合


0. 任務間同步

  1. 在多工實時系統中,一項工作的完成往往可以通過多個任務協調的方式共同來完成。例如一個任務是從感測器中接受資料並將資料寫到共享記憶體中,同時另一個任務週期的從共享記憶體中讀取資料併發送去顯示。
  2. 對共享記憶體的訪問不是排他性的,那麼各個執行緒間可能同時訪問它。這將引起資料一致性問題,例如,在顯示執行緒試圖顯示資料之前,感測器執行緒還未完成資料的寫入,那將包含不同的取樣的資料,造成顯示資料的迷惑。
  3. 將感測器資料寫入到共享記憶體的程式碼是接收執行緒的關鍵程式碼段;將感測器資料從共享內

存中讀出的程式碼是顯示執行緒的關鍵程式碼段;這兩段程式碼都會訪問共享記憶體。正常的操作序列

應該是在一個執行緒對共享記憶體塊操作完成後,才允許另一個執行緒去操作。對於操作/訪問同

一塊區域,稱之為臨界區。任務的同步方式有很多種,其核心思想都是:在訪問臨界區的時

候只允許一個(或一類)任務執行。

4. 同步就是幾個執行緒同時訪問共享資源,如何按順序一個一個執行的問題。

 

1. 中斷(鎖)

關閉中斷也叫中斷鎖,是禁止多工訪問臨界區最簡單的一種方式,即使是在分時操作

系統中也是如此。當中斷關閉的時候,就意味著當前任務不會被其他事件打斷(因為整個系

統已經不再響應那些可以觸發執行緒重新排程的外部事件),也就是當前執行緒不會被搶佔,除

非這個任務主動放棄了處理器控制權。關閉中斷/恢復中斷API介面由BSP實現,根據平臺的

不同其實現方式也大不相同。

1.2程式碼

while(1)

{

/* 關閉中斷*/

level = rt_hw_interrupt_disable();

cnt += no;

/* 恢復中斷*/

rt_hw_interrupt_enable(level);

rt_kprintf("thread[%d]'s counter is %d\n", no, cnt);

rt_thread_delay(no);

}

1.3 注意事項

由於關閉中斷會導致整個系統不能響應外部中斷,所以在使用關閉中斷做為互

斥訪問臨界區的手段時,首先必須需要保證關閉中斷的時間非常短,例如數條機器指

令。

1.4 使用場合

  1. 只是使用中斷鎖最主要的問題在於,在中斷關閉期間系統將不再響應任何中斷,也就不能響應外部的事件。所以中斷鎖對系統的實時性影響非常巨大,當使用不當的時候會導致系統完全無實時性可言(可能導致系統完全偏離要求的時間需求);而使用得當,則會變成一種快速、高效的同步方式。
  2. 使用中斷鎖來作業系統的方法可以應用於任何場合。且其他幾類同步方式都是依賴於中

斷鎖而實現的,可以說中斷鎖是最強大的和最高效的同步方法。

 

2. 排程器鎖

同中斷鎖一樣把排程器鎖住也能讓當前執行的任務不被換出,直到排程器解鎖。但和中

斷鎖有一點不相同的是,對排程器上鎖,系統依然能響應外部中斷,中斷服務例程依然能進

行相應的響應。所以在使用排程器上鎖的方式進行任務同步時,需要考慮好任務訪問的臨界

資源是否會被中斷服務例程所修改,如果可能會被修改,那麼將不適合採用此種方式進行同

步。

2.1 程式碼

void rt_enter_critical(void); /* 進入臨界區*/

呼叫這個函式後,排程器將被上鎖。在系統鎖住排程器的期間,系統依然響應中斷,如

果中斷喚醒了的更高優先順序執行緒,排程器並不會立刻執行它,直到呼叫解鎖排程器函式才嘗

試進行下一次排程。

void rt_exit_critical(void); /* 退出臨界區*/

當系統退出臨界區的時候,系統會計算當前是否有更高優先順序的執行緒就緒,如果有比當

前執行緒更高優先順序的執行緒就緒,將切換到這個高優先順序執行緒中執行;如果無更高優先順序執行緒

就緒,將繼續執行當前任務。

2.2 使用場合

排程器鎖能夠方便地使用於一些執行緒與執行緒間同步的場合,由於輕型,它不會對系統中

斷響應造成負擔;但它的缺陷也很明顯,就是它不能被用於中斷與執行緒間的同步或通知,並

且如果執行排程器鎖的時間過長,會對系統的實時性造成影響(因為使用了排程器鎖後,系

統將不再具備優先順序的關係,直到它脫離了排程器鎖的狀態)。

 

3. 訊號量

訊號量是一種輕型的用於解決執行緒間同步問題的核心物件,執行緒可以獲取或釋放它,從

而達到同步或互斥的目的。訊號量就像一把鑰匙,把一段臨界區給鎖住,只允許有鑰匙的線

程進行訪問:執行緒拿到了鑰匙,才允許它進入臨界區;而離開後把鑰匙傳遞給排隊在後面的

等待執行緒,讓後續執行緒依次進入臨界區。

3.1 程式碼操作

 

/*

* 程式清單:動態訊號量

*

* 這個例子中將建立一個動態訊號量(初始值為0)及一個動態執行緒,在這個動態執行緒中

* 將試圖採用超時方式去持有訊號量,應該超時返回。然後這個執行緒釋放一次訊號量,

* 並在後面繼續採用永久等待方式去持有訊號量, 成功獲得訊號量後返回。

*/

 

/* 建立一個訊號量,初始值是0 */

rt_sem_create("sem", 0, RT_IPC_FLAG_FIFO);

 

/* 試圖持有一個訊號量,如果10個OS Tick依然沒拿到,則超時返回*/

rt_sem_take(sem, 10);

/* 釋放一次訊號量*/

rt_sem_release(sem);

 

每個訊號量物件都有一個訊號量值和一個執行緒等待佇列,訊號量的值對應了訊號量物件的例項數目、資源數目,假如訊號量值為5,則表示共有5個訊號量例項(資源)可以被使用,當訊號量例項數目為零時,再申請該訊號量的執行緒就會被掛起在該訊號量的等待佇列上,等待可用的訊號量例項(資源)。

訊號量建立:

rt_sem_t rt_sem_create (const char* name, rt_uint32_t value, rt_uint8_t flag);

當呼叫這個函式時,系統將先分配一個semaphore物件,並初始化這個物件,然後初始化IPC物件以及與semaphore相關的部分。在建立訊號量指定的引數中,訊號量標誌引數決定了當訊號量不可用時,多個執行緒等待的排隊方式。當選擇FIFO方式時,那麼等待執行緒佇列將按照先進先出的方式排隊,先進入的執行緒將先獲得等待的訊號量;當選擇PRIO(優先順序等待)方式時,等待執行緒佇列將按照優先順序進行排隊,優先順序高的等待執行緒將先獲得等待的訊號量。

 

3.2 使用場合

  • 訊號量是一種非常靈活的同步方式,可以運用在多種場合中。形成鎖,同步,資源計數

等關係,也能方便的用於執行緒與執行緒,中斷與執行緒的同步中。

  • 執行緒同步:執行緒同步是訊號量最簡單的一類應用。例如,兩個執行緒用來進行任務間的執行控制轉移,訊號量的值初始化成具備0個訊號量資源例項,而等待執行緒先直接在這個訊號量上進行等待。當訊號執行緒完成它處理的工作時,釋放這個訊號量,以把等待在這個訊號量上的執行緒喚醒,讓它執行下一部分工作。這類場合也可以看成把訊號量用於工作完成標誌:訊號執行緒完成它

自己的工作,然後通知等待執行緒繼續下一部分工作。

  • 鎖:鎖,單一的鎖常應用於多個執行緒間對同一臨界區的訪問。。訊號量在作為鎖來使用時,通常應將訊號量資源例項初始化成1,代表系統預設有一個資源可用。當執行緒需要訪問臨界資

源時,它需要先獲得這個資源鎖。當這個執行緒成功獲得資源鎖時,其他打算訪問臨界區的線

程將被掛起在該訊號量上,這是因為其他執行緒在試圖獲取這個鎖時,這個鎖已經被鎖上(信

號量值是0)。當獲得訊號量的執行緒處理完畢,退出臨界區時,它將會釋放訊號量並把鎖解

開,而掛起在鎖上的第一個等待執行緒將被喚醒從而獲得臨界區的訪問權。因為訊號量的值始終在

1和0之間變動,所以這類鎖也叫做二值訊號量,如圖鎖所示:

  • 中斷與執行緒的同步:訊號量也能夠方便的應用於中斷與執行緒間的同步,,例如一箇中斷觸發,中斷服務例程需要通知執行緒進行相應的資料處理。。這個時候可以設定訊號量的初始值是0,執行緒在試圖持有這個訊號量時,由於訊號量的初始值是0,執行緒直接在這個訊號量上掛起直到訊號量被釋放。當中斷觸發時,先進行與硬體相關的動作,例如從硬體的I/O口中讀取相應的資料,並確認中斷以清除中斷源,而後釋放一個訊號量來喚醒相應的執行緒以做後續的資料處理。例如finsh shell執行緒的處理方式,如圖finsh shell的中斷、執行緒間同步所示:
  • 警告: 中斷與執行緒間的互斥不能採用訊號量(鎖)的方式,而應採用中斷鎖。
  • 資源計數

4. 互斥量

互斥量又叫相互排斥的訊號量,是一種特殊的二值性訊號量。它和訊號量不同的是,它支援互斥量所有權、遞迴訪問以及防止優先順序翻轉的特性。互斥鎖和訊號量,一個有中間管理,一個沒有中間管理。互斥量的狀態只有兩種,開鎖或閉鎖(兩種狀態值)。當有執行緒持有它時,互斥量處於閉鎖狀態,由這個執行緒獲得它的所有權。相反,當這個執行緒釋放它時,將對互斥量進行開鎖,失去它的所有權。當一個執行緒持有互斥量時,其他執行緒將不能夠對它進行開鎖或持有它,持有該互斥量的執行緒也能夠再次獲得這個鎖而不被掛起。這個特性與一般的二值訊號量有很大的不同,在訊號量中,因為已經不存在例項,執行緒遞迴持有會發生主動掛起(最終形成死鎖)。

4.1 示例:

/*

* 程式清單:互斥量使用例程

* 這個例子將建立3個動態執行緒以檢查持有互斥量時,持有的執行緒優先順序是否

* 被調整到等待執行緒優先順序中的最高優先順序。

* 執行緒1,2,3的優先順序從高到低分別被建立,

* 執行緒3先持有互斥量,而後執行緒2試圖持有互斥量,此時執行緒3的優先順序應該

* 被提升為和執行緒2的優先順序相同。執行緒1用於檢查執行緒3的優先順序是否被提升

* 為與執行緒2的優先順序相同。

*/

/* 建立互斥鎖*/

mutex = rt_mutex_create("mutex", RT_IPC_FLAG_FIFO);

/*

* 試圖持有互斥鎖,此時thread3持有,應把thread3的優先順序提升

* 到thread2相同的優先順序

*/

rt_mutex_take(mutex, RT_WAITING_FOREVER);

/* 釋放互斥鎖*/

rt_mutex_release(mutex);

4.2 使用場合:

互斥量的使用比較單一,因為它是訊號量的一種,並且它是以鎖的形式存在。在初始化的時候,互斥量永遠都處於開鎖的狀態,而被執行緒持有的時候則立刻轉為閉鎖的狀態。互斥量更適合於

 

 

5. 事件()

事件主要用於執行緒間的同步,與訊號量不同,它的特點是可以實現一對多,多對多的同步。即一個執行緒可等待多個事件的觸發:可以是其中任意一個事件喚醒執行緒進行事件處理的操作;也可以是幾個事件都到達後才喚醒執行緒進行後續的處理;同樣,事件也可以是多個執行緒同步多個事件,這種多個事件的集合可以用一個32位無符號整型變數來表示,變數的每一位代表一個事件,執行緒通過“邏輯與”或“邏輯或”與一個或多個事件建立關聯,形成一個事件集。事件的“邏輯或”也稱為是獨立型同步,指的是執行緒與任何事件之一發生同步;事件“邏輯與”也稱為是關聯型同步,指的是執行緒與若干事件都發生同步。

  • 事件只與執行緒相關,事件間相互獨立:每個執行緒擁有32個事件標誌,採用一個32 bit

無符號整型數進行記錄,每一個bit代表一個事件。若干個事件構成一個事件集;

  • 事件僅用於同步,不提供資料傳輸功能;
  • 事件無排隊性,即多次向執行緒傳送同一事件(如果執行緒還未來得及讀走),其效果等同

於只發送一次。

在RT-Thread實現中,每個執行緒都擁有一個事件資訊標記,它有三個屬性,分別是

RT_EVENT_FLAG_AND(邏輯與),RT_EVENT_FLAG_OR(邏輯或)以及T_EVENT_FLAG_CLEAR

(清除標記)。當執行緒等待事件同步時,可以通過32個事件標誌和這個事件資訊標記來判斷當前接收的事件是否滿足同步條件。

執行緒1的事件標誌中第2位和第29位被置位,如果事件資訊標記位設為邏輯與,則表示執行緒#1只有在事件1和事件29都發生以後才會被觸發喚醒,如果事件資訊標記位設為邏輯或,則事件1或事件29中的任意一個發生都會觸發喚醒執行緒#1。如果資訊標記同時設定了清除標記位,則當執行緒#1喚醒後將主動把事件1和事件29清為零,否則事件標誌將依然存在(即置1)。

5.1 示例:

/* 初始化事件物件*/

rt_event_init(&event, "event", RT_IPC_FLAG_FIFO);

/* 執行緒2入口函式 執行緒2持續地傳送事件#3 */

rt_event_send(&event, (1 << 3));

/* 執行緒3入口函式 執行緒3持續地傳送事件#5*/

rt_event_send(&event, (1 << 5));

 

/* 執行緒1入口函式 以邏輯與的方式接收事件*/

rt_event_recv(&event, ((1 << 3) | (1 << 5)),RT_EVENT_FLAG_AND | RT_EVENT_FLAG_CLEAR,RT_WAITING_FOREVER, &e)

/* 執行緒1入口函式 以邏輯或的方式接收事件*/

rt_event_recv(&event, ((1 << 3) | (1 << 5)),RT_EVENT_FLAG_OR | RT_EVENT_FLAG_CLEAR,RT_WAITING_FOREVER, &e)

5.2 使用場合:

事件可使用於多種場合,它能夠在一定程度上替代訊號量,用於執行緒間同步。一個執行緒或中斷服務例程傳送一個事件給事件物件,而後等待的執行緒被喚醒並對相應的事件進行處理。但是它與訊號量不同的是,事件的傳送操作在事件未清除前,是不可累計的,而訊號量的釋放動作是累計的。事件另外一個特性是,接收執行緒可等待多種事件,即多個事件對應一個執行緒或多個執行緒。同時按照執行緒等待的引數,可選擇是“邏輯或”觸發還是“邏輯與”觸發。這個特性也是訊號量等所不具備的,訊號量只能識別單一的釋放動作,而不能同時等待多種型別的釋放。如圖多事件接收所示:

各個事件型別可分別傳送或一起傳送給事件物件,而事件物件可以等待多個執行緒,它們僅對它們感興趣的事件進行關注。當有它們感興趣的事件發生時,執行緒就將被喚醒並進行後續的處理動作。

6. 郵箱(任務間通訊)

郵箱服務是實時作業系統中一種典型的任務間通訊方法,特點是開銷比較低,效率較高。郵箱中的每一封郵件只能容納固定的4位元組內容(針對32位處理系統,指標的大小即為4個位元組,所以一封郵件恰好能夠容納一個指標)。典型的郵箱也稱作交換訊息,如圖6-8所示,執行緒或中斷服務例程把一封4位元組長度的郵件傳送到郵箱中。而一個或多個執行緒可以從郵箱中接收這些郵件進行處理。

RT-Thread作業系統採用的郵箱通訊機制有點類似於傳統意義上的管道,用於執行緒間通訊。非阻塞方式的郵件傳送過程能夠安全的應用於中斷服務中,是執行緒,中斷服務,定時器向執行緒傳送訊息的有效手段。通常來說,郵件收取過程可能是阻塞的,這取決於郵箱中是否有郵件,以及收取郵件時設定的超時時間。當郵箱中不存在郵件且超時時間不為0時,郵件收取過程將變成阻塞方式。所以在這類情況下,只能由執行緒進行郵件的收取。

RT-Thread作業系統的郵箱中可存放固定條數的郵件,郵箱容量在建立/初始化郵箱時設定,每個郵件大小為4位元組。當需要線上程間傳遞比較大的訊息時,可以把指向一個緩衝區的指標作為郵件傳送到郵箱中。

在一個執行緒向郵箱傳送郵件時,如果郵箱沒滿,將把郵件複製到郵箱中。如果郵箱已經滿了,傳送執行緒可以設定超時時間,選擇是否等待掛起或直接返回-RT_EFULL。如果傳送執行緒選擇掛起等待,那麼當郵箱中的郵件被收取而空出空間來時,等待掛起的傳送執行緒將被喚醒繼續傳送的過程。

在一個執行緒從郵箱中接收郵件時,如果郵箱是空的,接收執行緒可以選擇是否等待掛起直到收到新的郵件而喚醒,或設定超時時間。當設定的超時時間,郵箱依然未收到郵件時,這個選擇超時等待的執行緒將被喚醒並返回-RT_ETIMEOUT。如果郵箱中存在郵件,那麼接收執行緒將複製郵箱中的4個位元組郵件到接收執行緒中。

6.1 示例:

/* 初始化一個mailbox */

rt_mb_init(&mb,

"mbt", /* 名稱是mbt */

&mb_pool[0], /* 郵箱用到的記憶體池是mb_pool */

sizeof(mb_pool)/4, /* 大小是mb_pool/4,因為每封郵件的大小是4位元組*/

RT_IPC_FLAG_FIFO); /* 採用FIFO方式進行執行緒等待*/

 

//可以在不同的執行緒中進行接受和傳送相關的資料。

/* 從郵箱中收取郵件*/

rt_mb_recv(&mb, (rt_uint32_t*)&str, RT_WAITING_FOREVER)

 

/* 傳送mb_str1地址到郵箱中*/mb_str1 要傳送的字串

rt_mb_send(&mb, (rt_uint32_t)&mb_str1[0]);

6.2 使用場合:

郵箱是一種簡單的執行緒間訊息傳遞方式,在RT-Thread作業系統的實現中能夠一次傳遞4位元組郵件,並且郵箱具備一定的儲存功能,能夠快取一定數量的郵件數(郵件數由建立、初始化郵箱時指定的容量決定)。郵箱中一封郵件的最大長度是4位元組,所以郵箱能夠用於不超過4位元組的訊息傳遞,當傳送的訊息長度大於這個數目時就不能再採用郵箱的方式。最重要的是,在32位系統上4位元組的內容恰好適合放置一個指標,所以郵箱也適合那種僅傳遞指標的情況,例如:

 

7. 訊息佇列(任務見通訊)

訊息佇列是另一種常用的執行緒間通訊方式,它能夠接收來自執行緒或中斷服務例程中不固

定長度的訊息,並把訊息快取在自己的記憶體空間中。其他執行緒也能夠從訊息佇列中讀取相應

的訊息,而當訊息佇列是空的時候,可以掛起讀取執行緒。當有新的訊息到達時,掛起的執行緒

將被喚醒以接收並處理訊息。訊息佇列是一種非同步的通訊方式。

通過訊息佇列服務,執行緒或中斷服務例程可以將一條或多條訊息放入訊息佇列中。同樣,一個或多個執行緒可以從訊息佇列中獲得訊息。當有多個訊息傳送到訊息佇列時,通常應將先進入訊息佇列的訊息先傳給執行緒,也就是說,執行緒先得到的是最先進入訊息佇列的訊息,即先進先出原則(FIFO)。

RT-Thread作業系統的訊息佇列物件由多個元素組成,當訊息佇列被建立時,它就被分配了訊息佇列控制塊:訊息佇列名稱、記憶體緩衝區、訊息大小以及佇列長度等。同時每個訊息佇列物件中包含著多個訊息框,每個訊息框可以存放一條訊息;訊息佇列中的第一個和最後一個訊息框被分別稱為訊息連結串列頭和訊息連結串列尾,對應於訊息佇列控制塊中的msg_queue_head和msg_queue_tail;有些訊息框可能是空的,它們通過msg_queue_free形成一個空閒訊息框連結串列。所有訊息佇列中的訊息框總數即是訊息佇列的長度,這個長度可在訊息佇列建立時指定。

7.1 示例:

/*

* 程式清單:訊息佇列例程*

* 這個程式會建立3個動態執行緒:

* 一個執行緒會從訊息佇列中收取訊息;

* 一個執行緒會定時給訊息佇列傳送訊息;

* 一個執行緒會定時給訊息佇列傳送緊急訊息。

*/

 

/* 初始化訊息佇列*/

rt_mq_init(&mq, "mqt",

&msg_pool[0], /* 記憶體池指向msg_pool */

128 - sizeof(void*), /* 每個訊息的大小是128 - void* */

sizeof(msg_pool), /* 記憶體池的大小是msg_pool的大小*/

RT_IPC_FLAG_FIFO); /* 如果有多個執行緒等待,按照FIFO的方法分配訊息*/

 

/* 從訊息佇列中接收訊息*/

rt_mq_recv(&mq, &buf[0], sizeof(buf), RT_WAITING_FOREVER)

 

/* 傳送訊息到訊息佇列中*/

result = rt_mq_send(&mq, &buf[0], sizeof(buf));

 

/* 傳送緊急訊息到訊息佇列中*/

rt_mq_urgent(&mq, &buf[0], sizeof(buf));

7.2 使用場合

訊息佇列可以應用於傳送不定長訊息的場合,包括執行緒與執行緒間的訊息交換,以及中斷

服務例程中傳送給執行緒的訊息(中斷服務例程不可能接收訊息)。

訊息佇列和郵箱的明顯不同是訊息的長度並不限定在4個位元組以內,另外訊息佇列也包括了一個傳送緊急訊息的函式介面。但是當建立的是一個所有訊息的最大長度是4位元組的訊息佇列時,訊息佇列物件將蛻化成郵箱。這個不限定長度的訊息,也及時的反應到了程式碼編寫的場合上,同樣是類似郵箱的程式碼: