一、進程間通信機制
rt-thread操作系統的IPC(Inter-Process Communication,進程間同步與通信)包含有中斷鎖、調度器鎖、信號量、互斥鎖、事件、郵箱、消息隊列。其中前5個主要表現為線程間同步,郵箱與消息隊列表現為線程間通信。本文主要介紹它們的一些特性及使用場合。
1、中斷鎖
關閉中斷也叫中斷鎖,是禁止多任務訪問臨界區最簡單的一種方式,即使是在分時操作系統中也是如此。當中斷關閉的時候,就意味着當前任務不會被其他事件打斷(因為整個系統已經不再響應那些可以觸發線程重新調度的外部事件),也就是當前線程不會被搶占,除非這個任務主動放棄了處理器控制權。關閉中斷/恢復中斷API接口由BSP實現,根據平台的不同其實現方式也大不相同。比如在stm32平台中中斷鎖機制通過關閉中斷函數(rt_base_t rt_hw_interrupt_disable(void),這個函數用於關閉中斷並返回關閉中斷前的中斷狀態。)以及恢復中斷函數(void rt_hw_interrupt_enable(rt_base_t level),恢復調用rt_hw_interrupt_disable()函數前的中斷狀態)實現。
警告: 由於關閉中斷會導致整個系統不能響應外部中斷,所以在使用關閉中斷做為互斥訪問臨界區的手段時,首先必須需要保證關閉中斷的時間非常短,例如數條機器指令。
使用中斷鎖來操作系統的方法可以應用於任何場合,且其他幾類同步方式都是依賴於中斷鎖而實現的,可以說中斷鎖是最強大的和最高效的同步方法。只是使用中斷鎖最主要的問題在於,在中斷關閉期間系統將不再響應任何中斷,也就不能響應外部的事件。所以中斷鎖對系統的實時性影響非常巨大,當使用不當的時候會導致系統完全無實時性可言(可能導致系統完全偏離要求的時間需求);而使用得當,則會變成一種快速、高效的同步方式。例如,為了保證一行代碼(例如賦值)的互斥運行,最快速的方法是使用中斷鎖而不是信號量或互斥量。
2、調度器鎖
同中斷鎖一樣把調度器鎖住也能讓當前運行的任務不被換出,直到調度器解鎖。但和中斷鎖有一點不相同的是,對調度器上鎖,系統依然能響應外部中斷,中斷服務例程依然能進行相應的響應。所以在使用調度器上鎖的方式進行任務同步時,需要考慮好任務訪問的臨界資源是否會被中斷服務例程所修改,如果可能會被修改,那么將不適合采用此種方式進行同步。在rt-therad系統中通過上鎖函數(void rt_enter_critical(void),在系統鎖住調度器的期間,系統依然響應中斷,如果中斷喚醒了的更高優先級線程,調度器並不會立刻執行它,直到調用解鎖調度器函數才嘗試進行下一次調度。)以及解鎖函數(void rt_exit_critical(void),當系統退出臨界區的時候,系統會計算當前是否有更高優先級的線程就緒,如果有比當前線程更高優先級的線程就緒,將切換到這個高優先級線程中執行;如果無更高優先級線程就緒,將繼續執行當前任務。)實現調度鎖機制。
注意: rt_enter_critical/rt_exit_critical可以多次嵌套調用,但每調用一次rt_enter_critical就必須相對應地調用一次rt_exit_critical退出操作,嵌套的最大深度是65535。
調度器鎖能夠方便地使用於一些線程與線程間同步的場合,由於輕型,它不會對系統中斷響應造成負擔;但它的缺陷也很明顯,就是它不能被用於中斷與線程間的同步或通知,並且如果執行調度器鎖的時間過長,會對系統的實時性造成影響(因為使用了調度器鎖后,系統將不再具備優先級的關系,直到它脫離了調度器鎖的狀態)。
3、信號量
信號量是一種輕型的用於解決線程間同步問題的內核對象,線程可以獲取或釋放它,從而達到同步或互斥的目的。信號量就像一把鑰匙,把一段臨界區給鎖住,只允許有鑰匙的線程進行訪問:線程拿到了鑰匙,才允許它進入臨界區;而離開后把鑰匙傳遞給排隊在后面的等待線程,讓后續線程依次進入臨界區。
使用信號量會導致的另一個潛在問題是線程優先級翻轉。所謂優先級翻轉問題即當一個高優先級線程試圖通過信號量機制訪問共享資源時,如果該信號量已被一低優先級線程持有,而這個低優先級線程在運行過程中可能又被其它一些中等優先級的線程搶占,因此造成高優先級線程被許多具有較低優先級的線程阻塞,實時性難以得到保證。例如:有優先級為A、B和C的三個線程,優先級A> B > C。線程A,B處於掛起狀態,等待某一事件觸發,線程C正在運行,此時線程C開始使用某一共享資源M。在使用過程中,線程A等待的事件到來,線程A轉為就緒態,因為它比線程C優先級高,所以立即執行。但是當線程A要使用共享資源M時,由於其正在被線程C使用,因此線程A被掛起切換到線程C運行。如果此時線程B等待的事件到來,則線程B轉為就緒態。由於線程B的優先級比線程C高,因此線程B開始運行,直到其運行完畢,線程C才開始運行。只有當線程C釋放共享資源M后,線程A才得以執行。在這種情況下,優先級發生了翻轉,線程B先於線程A運行。這樣便不能保證高優先級線程的響應時間。
在RT-Thread操作系統中實現的是優先級繼承算法。優先級繼承是通過在線程A被阻塞的期間內,將線程C的優先級提升到線程A的優先級別,從而解決優先級翻轉引起的問題。這樣能夠防止C(間接地防止A)被B搶占。優先級繼承協議是指,提高某個占有某種資源的低優先級線程的優先級,使之與所有等待該資源的線程中優先級最高的那個線程的優先級相等,然后執行,而當這個低優先級線程釋放該資源時,優先級重新回到初始設定。因此,繼承優先級的線程避免了系統資源被任何中間優先級的線程搶占。
線程同步是信號量最簡單的一類應用。例如,兩個線程用來進行任務間的執行控制轉移,信號量的值初始化成具備0個信號量資源實例(信號量的值初始化為0),而等待線程先直接在這個信號量上進行等待。當信號線程完成它處理的工作時,釋放這個信號量,以把等待在這個信號量上的線程喚醒,讓它執行下一部分工作。這類場合也可以看成把信號量用於工作完成標志:信號線程完成它自己的工作,然后通知等待線程繼續下一部分工作。
鎖,單一的鎖常應用於多個線程間對同一臨界區的訪問。信號量在作為鎖來使用時,通常應將信號量資源實例初始化成1(信號量的值初始化為1),代表系統默認有一個資源可用。當線程需要訪問臨界資源時,它需要先獲得這個資源鎖。當這個線程成功獲得資源鎖時,其他打算訪問臨界區的線程將被掛起在該信號量上,這是因為其他線程在試圖獲取這個鎖時,這個鎖已經被鎖上(信號量值減1變為0)。當獲得信號量的線程處理完畢,退出臨界區時,它將會釋放信號量並把鎖解開,而掛起在鎖上的第一個等待線程將被喚醒從而獲得臨界區的訪問權。因為信號量的值始終在1和0之間變動,所以這類鎖也叫做二值信號量。
信號量也能夠方便的應用於中斷與線程間的同步,例如一個中斷觸發,中斷服務例程需要通知線程進行相應的數據處理。這個時候可以設置信號量的初始值是0,線程在試圖持有這個信號量時,由於信號量的初始值是0,線程直接在這個信號量上掛起直到信號量被釋放。 當中斷觸發時,先進行與硬件相關的動作,例如從硬件的I/O口中讀取相應的數據,並確認中斷以清除中斷源,然后釋放一個信號量來喚醒相應的線程以做后續的數據處理。警告: 中斷與線程間的互斥不能采用信號量(鎖)的方式,而應采用中斷鎖。
資源計數適合於線程間速度不匹配的場合,這個時候信號量可以做為前一線程工作完成的計數,而當調度到后一線程時,它可以以一種連續的方式一次處理數個事件。例如,生產者與消費者問題中,生產者可以對信號進行多次釋放,而后消費者被調度到時能夠一次處理多個資源。注意: 一般資源計數類型多是混合方式的線程間同步,因為對於單個的資源處理依然存在線程的多重訪問,這就需要對一個單獨的資源進行訪問、處理,並進行鎖方式的互斥操作。
4、互斥量
互斥量又叫相互排斥的信號量,是一種特殊的二值性信號量。它和信號量不同的是,它支持互斥量所有權、遞歸訪問以及防止優先級翻轉的特性。互斥量的狀態只有兩種,開鎖或閉鎖(兩種狀態值)。當有線程持有它時,互斥量處於閉鎖狀態,由這個線程獲得它的所有權。相反,當這個線程釋放它時,將對互斥量進行開鎖,失去它的所有權。當一個線程持有互斥量時,其他線程將不能夠對它進行開鎖或持有它,持有該互斥量的線程也能夠再次獲得這個鎖而不被掛起。這個特性與一般的二值信號量有很大的不同,在信號量中,因為已經不存在實例,線程遞歸持有會發生主動掛起(最終形成死鎖)。
警告: 在獲得互斥量后,請盡快釋放互斥量,並且在持有互斥量的過程中,不得另行更改持有互斥量線程的優先級。
互斥量的使用比較單一,因為它是信號量的一種,並且它是以鎖的形式存在。在初始化的時候,互斥量永遠都處於開鎖的狀態,而被線程持有的時候則立刻轉為閉鎖的狀態。互斥量更適合於:
• 線程多次持有(獲取)互斥量的情況下。這樣可以避免同一線程多次遞歸持有而造成死鎖的問題;
• 可能會由於多線程同步而造成優先級翻轉的情況;
另外需要切記的是互斥量不能在中斷服務例程中使用。信號量則可用於中斷與線程同步。
5、事件
事件主要用於線程間的同步,與信號量不同,它的特點是可以實現一對多,多對多的同步。即一個線程可等待多個事件的觸發:可以是其中任意一個事件喚醒線程進行事件處理的操作;也可以是幾個事件都到達后才喚醒線程進行后續的處理;同樣,事件也可以是多個線程同步多個事件,這種多個事件的集合可以用一個32位無符號整型變量來表示,變量的每一位代表一個事件,線程通過“邏輯與”或“邏輯或”與一個或多個事件建立關聯,形成一個事件集。事件的“邏輯或”也稱為是獨立型同步,指的是線程與任何事件之一發生同步;事件“邏輯與”也稱為是關聯型同步,指的是線程與若干事件都發生同步。
RT-Thread定義的事件有以下特點:
• 事件只與線程相關,事件間相互獨立:每個線程擁有32個事件標志,采用一個32 bit無符號整型數進行記錄,每一個bit代表一個事件。若干個事件構成一個事件集;
• 事件僅用於同步,不提供數據傳輸功能;
• 事件無排隊性,即多次向線程發送同一事件(如果線程還未來得及讀走),其效果等同於只發送一次。
在RT-Thread實現中,每個線程都擁有一個事件信息標記,它有三個屬性,分別是RT_EVENT_FLAG_AND(邏輯與),RT_EVENT_FLAG_OR(邏輯或)以及RT_EVENT_FLAG_CLEAR(清除標記)。當線程等待事件同步時,可以通過32個事件標志和這個事件信息標記來判斷當前接收的事件是否滿足同步條件。
事件可使用於多種場合,它能夠在一定程度上替代信號量,用於線程間同步。線程或中斷服務例程發送一個事件給事件對象,而后等待的線程被喚醒並對相應的事件進行處理。但是它與信號量不同的是,事件的發送操作在事件未清除前,是不可累計的,而信號量的釋放動作是累計的。 事件另外一個特性是,接收線程可等待多種事件,即多個事件對應一個線程或多個線程。同時按照線程等待的參數,可選擇是“邏輯或”觸發還是“邏輯與”觸發。這個特性也是信號量等所不具備的,信號量只能識別單一的釋放動作,而不能同時等待多種類型的釋放。各個事件類型可分別發送或一起發送給事件對象,而事件對象可以等待多個線程,它們僅對它們感興趣的事件進行關注。當有它們感興趣的事件發生時,線程就將被喚醒並進行后續的處理動作。
6、郵箱
郵箱服務是實時操作系統中一種典型的任務間通信方法,特點是開銷比較低,效率較高。郵箱中的每一封郵件只能容納固定的4字節內容(針對32位處理系統,指針的大小即為4個字節,所以一封郵件恰好能夠容納一個指針)。典型的郵箱也稱作交換消息,線程或中斷服務例程把一封4字節長度的郵件發送到郵箱中。而一個或多個線程可以從郵箱中接收這些郵件進行處理。
RT-Thread操作系統采用的郵箱通信機制有點類似於傳統意義上的管道,用於線程間通訊。非阻塞方式的郵件發送過程能夠安全的應用於中斷服務中,是線程,中斷服務,定時器向線程發送消息的有效手段。通常來說,郵件收取過程可能是阻塞的,這取決於郵箱中是否有郵件,以及收取郵件時設置的超時時間。當郵箱中不存在郵件且超時時間不為0時,郵件收取過程將變成阻塞方式。所以在這類情況下,只能由線程進行郵件的收取。
RT-Thread操作系統的郵箱中可存放固定條數的郵件,郵箱容量在創建/初始化郵箱時設定,每個郵件大小為4字節。當需要在線程間傳遞比較大的消息時,可以把指向一個緩沖區的指針作為郵件發送到郵箱中。在一個線程向郵箱發送郵件時,如果郵箱沒滿,將把郵件復制到郵箱中。如果郵箱已經滿了,發送線程可以設置超時時間,選擇是否等待掛起或直接返回-RT_EFULL。如果發送線程選擇掛起等待,那么當郵箱中的郵件被收取而空出空間來時,等待掛起的發送線程將被喚醒繼續發送的過程。在一個線程從郵箱中接收郵件時,如果郵箱是空的,接收線程可以選擇是否等待掛起直到收到新的郵件而喚醒,或設置超時時間。當設置的超時時間,郵箱依然未收到郵件時,這個選擇超時等待的線程將被喚醒並返回-RT_ETIMEOUT。如果郵箱中存在郵件,那么接收線程將復制郵箱中的4個字節郵件到接收線程中。
郵箱是一種簡單的線程間消息傳遞方式,在RT-Thread操作系統的實現中能夠一次傳遞4字節郵件,並且郵箱具備一定的存儲功能,能夠緩存一定數量的郵件數(郵件數由創建、初始化郵箱時指定的容量決定)。郵箱中一封郵件的最大長度是4字節,所以郵箱能夠用於不超過4字節的消息傳遞,當傳送的消息長度大於這個數目時就不能再采用郵箱的方式。 最重要的是,在32位系統上4字節的內容恰好適合放置一個指針,所以郵箱也適合那種僅傳遞指針的情況。
7、消息隊列
消息隊列是另一種常用的線程間通訊方式,它能夠接收來自線程或中斷服務例程中不固定長度的消息,並把消息緩存在自己的內存空間中。其他線程也能夠從消息隊列中讀取相應的消息,而當消息隊列是空的時候,可以掛起讀取線程。當有新的消息到達時,掛起的線程將被喚醒以接收並處理消息。消息隊列是一種異步的通信方式。通過消息隊列服務,線程或中斷服務例程可以將一條或多條消息放入消息隊列中。同樣,一個或多個線程可以從消息隊列中獲得消息。當有多個消息發送到消息隊列時,通常應將先進入消息隊列的消息先傳給線程,也就是說,線程先得到的是最先進入消息隊列的消息,即先進先出原則(FIFO)。
RT-Thread操作系統的消息隊列對象由多個元素組成,當消息隊列被創建時,它就被分配了消息隊列控制塊:消息隊列名稱、內存緩沖區、消息大小以及隊列長度等。同時每個消息隊列對象中包含着多個消息框,每個消息框可以存放一條消息;消息隊列中的第一個和最后一個消息框被分別稱為消息鏈表頭和消息鏈表尾,對應於消息隊列控制塊中的msg_queue_head和msg_queue_tail;有些消息框可能是空的,它們通過msg_queue_free形成一個空閑消息框鏈表。所有消息隊列中的消息框總數即是消息隊列的長度,這個長度可在消息隊列創建時指定。
消息隊列可以應用於發送不定長消息的場合,包括線程與線程間的消息交換,以及中斷服務例程中發送給線程的消息(中斷服務例程不可能接收消息)。消息隊列和郵箱的明顯不同是消息的長度並不限定在4個字節以內,另外消息隊列也包括了一個發送緊急消息的函數接口。但是當創建的是一個所有消息的最大長度是4字節的消息隊列時,消息隊列對象將蛻化成郵箱。
在一般的系統設計中會經常遇到要發送同步消息的問題,這個時候就可以根據當時的狀態選擇相應的實現:兩個線程間可以采用[消息隊列+信號量或郵箱]的形式實現。 發送線程通過消息發送的形式發送相應的消息給消息隊列,發送完畢后希望獲得接收線程的收到確認。郵箱做為確認標志,代表着接收線程能夠通知一些狀態值給發送線程;而信號量作為確認標志只能夠單一的通知發送線程,消息已經確認接收。
二、IPC控制塊:在include/rtdef.h中
/** * IPC flags and control command definitions */ #define RT_IPC_FLAG_FIFO 0x00 /**< FIFOed IPC. @ref IPC. */ #define RT_IPC_FLAG_PRIO 0x01 /**< PRIOed IPC. @ref IPC. */ #define RT_IPC_CMD_UNKNOWN 0x00 /**< unknown IPC command */ #define RT_IPC_CMD_RESET 0x01 /**< reset IPC object */ #define RT_WAITING_FOREVER -1 /**< Block forever until get resource. */ #define RT_WAITING_NO 0 /**< Non-block. */ /** * Base structure of IPC object */ struct rt_ipc_object { struct rt_object parent; /**< inherit from rt_object *///可知其派生自內核對象 rt_list_t suspend_thread; /**< threads pended on this resource *///線程掛起鏈表 };
三、IPC內聯函數:在src/ipc.c中
rt_inline rt_err_t rt_ipc_object_init(struct rt_ipc_object *ipc) { /* init ipc object */ rt_list_init(&(ipc->suspend_thread)); //初始化線程掛起鏈表 return RT_EOK; }
/** * This function will suspend a thread to a specified list. IPC object or some * double-queue object (mailbox etc.) contains this kind of list. * * @param list the IPC suspended thread list * @param thread the thread object to be suspended * @param flag the IPC object flag, * which shall be RT_IPC_FLAG_FIFO/RT_IPC_FLAG_PRIO. * * @return the operation status, RT_EOK on successful */ rt_inline rt_err_t rt_ipc_list_suspend(rt_list_t *list, struct rt_thread *thread, rt_uint8_t flag) { /* suspend thread */ rt_thread_suspend(thread);//掛起線程 switch (flag) { case RT_IPC_FLAG_FIFO: //FIFO方式 rt_list_insert_before(list, &(thread->tlist));//直接放入隊列末尾 break; case RT_IPC_FLAG_PRIO: //線程優先級方式 { struct rt_list_node *n; struct rt_thread *sthread; /* find a suitable position */ for (n = list->next; n != list; n = n->next)//遍歷信號量的掛起鏈表 { sthread = rt_list_entry(n, struct rt_thread, tlist); /* find out */ if (thread->current_priority < sthread->current_priority)//按優先級找到合適位置 { /* insert this thread before the sthread */ rt_list_insert_before(&(sthread->tlist), &(thread->tlist));//將線程加入到鏈表中 break; } } /* * not found a suitable position, * append to the end of suspend_thread list */ if (n == list) rt_list_insert_before(list, &(thread->tlist));//沒有找到合適位置,則放到末尾 } break; } return RT_EOK; }
調用rt_ipc_list_suspend將當前線程掛起,這個掛起是指將當前線程加入到信號量的掛起鏈表中,這里有一個flag參數,即sem->parent.parent.flag(在信號量初始化時設置),其值有兩種RT_IPC_FLAG_FIFO,RT_IPC_FLAG_FIFO,前者表示按FIFO的方式放入掛起鏈表,后者是根據線程本身的優先級等級來決定放入到掛起鏈表的位置,由於每次釋放一個信號量,只會從信號量掛起鏈表上喚醒第一個線程,因此,掛起線程鏈表上的位置就決定了當信號到達時掛起的線程的喚醒順序。
/** * This function will resume the first thread in the list of a IPC object: * - remove the thread from suspend queue of IPC object * - put the thread into system ready queue * * @param list the thread list * * @return the operation status, RT_EOK on successful */ rt_inline rt_err_t rt_ipc_list_resume(rt_list_t *list) { struct rt_thread *thread; /* get thread entry */ thread = rt_list_entry(list->next, struct rt_thread, tlist);//獲取線程 RT_DEBUG_LOG(RT_DEBUG_IPC, ("resume thread:%s\n", thread->name)); /* resume it */ rt_thread_resume(thread);//喚醒此線程 return RT_EOK; } 函數rt_ipc_list_resume只會喚醒信號量中第一個掛起的線程。正常喚醒掛起線程時(如獲取信號量,互斥量等)不會修改線程的error值,即error原持原值RT_EOK不變.
/** * This function will resume all suspended threads in a list, including * suspend list of IPC object and private list of mailbox etc. * * @param list of the threads to resume * * @return the operation status, RT_EOK on successful */ rt_inline rt_err_t rt_ipc_list_resume_all(rt_list_t *list) { struct rt_thread *thread; register rt_ubase_t temp; /* wakeup all suspend threads */ while (!rt_list_isempty(list)) //遍歷線程掛起鏈表 { /* disable interrupt */ temp = rt_hw_interrupt_disable();//關中斷 /* get next suspend thread */ thread = rt_list_entry(list->next, struct rt_thread, tlist);//獲得線程 /* set error code to RT_ERROR */ thread->error = -RT_ERROR; //設置線程的錯誤碼為-RT_ERROR /* * resume thread * In rt_thread_resume function, it will remove current thread from * suspend list */ rt_thread_resume(thread); //喚醒此線程,表明為異常喚醒 /* enable interrupt */ rt_hw_interrupt_enable(temp); //開中斷 } return RT_EOK; } 將掛起鏈表中的所有線程都喚醒。需要注意地是,喚醒的線程的error值將會被設置為-RT_ERROR,以此標志此線程是被異常喚醒,並不是正常獲取到ipc內核對象(如信號量,互斥量等)而被喚醒,這在take函數中將會以線程的error值來進行判斷