基于SylixOS的高效生產(chǎn)者消費者模型編程應(yīng)用設(shè)計

秦飛

摘 要：在SylixOS下，介紹了基于傳統(tǒng)消息隊列和高效協(xié)程模式兩種生產(chǎn)者-消費者模型的編程方法，比較兩種模式的編程方法在互斥和同步機制上的區(qū)別及其帶來的消耗。

關(guān)鍵詞：SylixOS;同步和互斥;協(xié)程

中圖分類號：TP311.1? ? ? ? ? ? 文獻標(biāo)識碼：A

引言：

傳統(tǒng)的生產(chǎn)者-消費者模型中，生產(chǎn)者和消費者之間的關(guān)系可以總結(jié)為如下幾點：

●生產(chǎn)者與生產(chǎn)者：互斥關(guān)系

●消費者與消費者：互斥關(guān)系

●生產(chǎn)者與消費者：互斥與同步關(guān)系

要處理好這些關(guān)系，就要用到系統(tǒng)中的互斥與同步機制，并涉及到線程的阻塞和喚醒操作，而這些機制往往會帶來一些必要的消耗，影響運行效率。

協(xié)程，又稱作協(xié)同程序，是比線程還小的可執(zhí)行代碼序。在SylixOS中，一個線程可以擁有多個協(xié)程，這些協(xié)程共享了線程除了棧以外的所有資源，包括優(yōu)先級、內(nèi)核對象等。線程內(nèi)部的協(xié)程不可被搶占，只能輪轉(zhuǎn)運行。所以，利用這個特點可以構(gòu)建出一個基于協(xié)程的高效的生產(chǎn)者-消費者模型。

1、何為生產(chǎn)者消費者模型

經(jīng)典生產(chǎn)者-消費者模型描述如下：兩個線程共享一個消息隊列，其中一個線程，即生產(chǎn)者，向該消息隊列中放入信息;另外一個線程，即消費者，從該消息隊列中取走信息，這里就涉及到了多任務(wù)操作系統(tǒng)中的同步和互斥機制。

在SylixOS中，同步即一個任務(wù)的執(zhí)行結(jié)果是另一個任務(wù)能夠執(zhí)行的前提條件，互斥即多任務(wù)同時使用到臨界資源時可能會引發(fā)的競爭。在生產(chǎn)者-消費者模型中，消費者任務(wù)需要等待生產(chǎn)任務(wù)產(chǎn)生數(shù)據(jù)后才能執(zhí)行，即同步;而生產(chǎn)任務(wù)和消費任務(wù)要串行地訪問消息隊列，即互斥。

2、生產(chǎn)者消費者模型的兩種編程方法

2.1 傳統(tǒng)消息隊列方式

該方式中，生產(chǎn)者-消費者模型中的公共緩沖區(qū)即對應(yīng)著消息隊列。編程步驟如下：

① 創(chuàng)建生產(chǎn)者線程;

② 創(chuàng)建消費者線程;

③ 創(chuàng)建消息隊列，獲取操作句柄;

④ 生產(chǎn)者線程中循環(huán)調(diào)用消息發(fā)送函數(shù)，向消息隊列中發(fā)送數(shù)據(jù);

⑤ 消費者線程中循環(huán)調(diào)用消息接收函數(shù)，從消息隊列中獲取數(shù)據(jù)，若消息隊列中無數(shù)據(jù)，則會阻塞。

2.2 高效率協(xié)程模式

SylixOS中的協(xié)程需要依賴所屬線程被調(diào)度時執(zhí)行，所以需要先創(chuàng)建一個初始線程，再去調(diào)用相應(yīng)的函數(shù)創(chuàng)建生產(chǎn)者和消費者協(xié)程，需要注意的是，每一個線程創(chuàng)建時都會默認創(chuàng)建一個初始協(xié)程，并且線程總是從該初始協(xié)程開始運行，所以初始線程中需要主動放棄處理器，以保證生產(chǎn)者和消費者協(xié)程能運行。編程步驟如下：

① 創(chuàng)建初始線程;

② 創(chuàng)建生產(chǎn)者協(xié)程;

③ 創(chuàng)建消費者協(xié)程;

④ 初始線程放棄處理器調(diào)用;

⑤ 生產(chǎn)者協(xié)程向公共緩沖區(qū)填充消息后，主動放棄處理器;

⑥ 消費者協(xié)程從公共緩沖區(qū)獲取消息后，主動放棄處理器;

⑦ 初始線程恢復(fù)生產(chǎn)者和消費者協(xié)程，跳轉(zhuǎn)至步驟⑤。

2.3 兩種編程模式的比較

通過上述兩種模式的步驟比較，可以得出以下結(jié)論：

① 傳統(tǒng)消息隊列模式中，需要創(chuàng)建多線程并進行線程間的切換，同時生產(chǎn)者和消費者線程間的通信采用消息隊列的形式，其發(fā)送和接收函數(shù)內(nèi)部實現(xiàn)涉及到進入和退出內(nèi)核態(tài)的模式切換，以保證生產(chǎn)者和消費者線程中的同步和互斥機制，這里面會有較大的時間消耗;

② 高效協(xié)程模式中，只有一個初始線程，內(nèi)部的協(xié)程并不會被調(diào)度器調(diào)度，所以不會有線程切換的消耗。另外，初始線程內(nèi)部的協(xié)程是以輪轉(zhuǎn)的方式運行，由用戶程序自行管理，所以也就不會有線程間通信帶來的消耗。

綜上，可以看出，協(xié)程模式比傳統(tǒng)消息隊列模式更高效。

3、模型編程流程設(shè)計

基于兩種方法設(shè)計應(yīng)用流程圖及其簡單說明

3.1 消息隊列方法

3.2 協(xié)程方法

4、代碼設(shè)計

4.1 消息隊列方法

基于消息隊列的設(shè)計代碼中，生產(chǎn)者線程獲取當(dāng)前時間t1并填入消息隊列中，消費者線程從消息隊列中獲取消息后再次獲取當(dāng)前時間t2，將t1和t2的差值作為該方法帶來的時間消耗。示例代碼如下。

#include

#include

#include

#include

#define MAX_MSG_COUNT? ?100

#define MAX_MSG_BYTES? ?sizeof（struct timeval）

LW_HANDLE? ?_G_hMsgQ = 0;

static PVOID __threadConsume （PVOID? pvArg）

{

struct timeval? end;

struct timeval? recv;

INT? ? iError? = PX_ERROR;

double t? ? ? ?= 0;

size_t stLen? ?= 0;

INT? ? iMsgCnt = 0;

while （1） {

lib_memset（&end， 0， sizeof（struct timeval））;

lib_memset（&recv， 0， sizeof（struct timeval））;

iError = Lw_MsgQueue_Receive（_G_hMsgQ，

（PVOID）&recv，

sizeof（struct timeval），

&stLen，

LW_OPTION_WAIT_INFINITE）;

if （ERROR_NONE ！= iError） {

printf（"msg recv error.＼n"）;

break;

}

lib_gettimeofday（&end， LW_NULL）;

t += end.tv_sec * 1e6 + end.tv_usec - （recv.tv_sec * 1e6 + recv.tv_usec）;

if （++iMsgCnt == 100） {

printf（"legancy msgQ： consumer get msg cost： %.02f us＼n"， t / iMsgCnt）;

iMsgCnt = 0;

t? ? ? ?= 0;

}

}

return? （LW_NULL）;

}

static PVOID __threadProduct （PVOID? pvArg）

{

struct timeval? start;

INT? ? ? ? ? ? ?iError;

while （1） {

lib_memset（&start， 0， sizeof（struct timeval））;

lib_gettimeofday（&start， LW_NULL）;

iError = Lw_MsgQueue_Send（_G_hMsgQ， （PVOID）&start， sizeof（struct timeval））;

if （ERROR_NONE ！= iError） {

printf（"msg send error.＼n"）;

break;

}

Lw_Time_Sleep（1）;

}

return? （LW_NULL）;

}

int main （int argc， char **argv）

{

LW_HANDLE? hProduct;

LW_HANDLE? hConsume;

_G_hMsgQ = Lw_MsgQueue_Create（"msg_cp"，

MAX_MSG_COUNT，

MAX_MSG_BYTES，

LW_OPTION_WAIT_FIFO | LW_OPTION_OBJECT_LOCAL，

LW_NULL）;

hProduct = Lw_Thread_Create（"t_product"， __threadProduct， LW_NULL， LW_NULL）;

if （LW_OBJECT_HANDLE_INVALID == hProduct） {

printf（"t_product create failed.＼n"）;

return? （-1）;

}

hConsume = Lw_Thread_Create（"t_consume"， __threadConsume， LW_NULL， LW_NULL）;

if （LW_OBJECT_HANDLE_INVALID == hConsume） {

printf（"t_consume create failed.＼n"）;

return? （-1）;

}

Lw_Thread_Join（hProduct， LW_NULL）;

Lw_Thread_Join（hConsume， LW_NULL）;

Lw_MsgQueue_Delete（&_G_hMsgQ）;

return? （0）;

}

4.2 協(xié)程方法

基于協(xié)程的設(shè)計代碼中，生產(chǎn)者協(xié)程將數(shù)據(jù)放入公共緩沖區(qū)，并獲取當(dāng)前時間t1，然后放棄CPU，接著消費者協(xié)程中獲取時間t2，并從公共緩沖區(qū)獲取數(shù)據(jù)，將t1和t2的差值作為生產(chǎn)者和消費者協(xié)程切換的時間消耗。示例代碼如下。

#include

#include

static PVOID? pcCrcb0 = LW_NULL;

static PVOID? pcCrcb1 = LW_NULL;

static double t? ? ? ?= 0;

static INT? ? iCnt? ? = 0;

static struct timeval? start;

static struct timeval? end;

VOID? coroutine0 （PVOID? pvArg）

{

while （1） {

lib_memset（&start， 0， sizeof（struct timeval））;

lib_gettimeofday（&start， LW_NULL）;

Lw_Coroutine_Yield（）;

}

}

VOID? coroutine1 （PVOID? pvArg）

{

while （1） {

lib_memset（&end， 0， sizeof（struct timeval））;

lib_gettimeofday（&end， LW_NULL）;

t += end.tv_sec * 1e6 + end.tv_usec - （start.tv_sec * 1e6 + start.tv_usec）;

if （++iCnt == 100） {

printf（"coroutine： consumer get msg cost： %.02f us＼n"， t / iCnt）;

iCnt = 0;

t? ? = 0;

}

Lw_Coroutine_Yield（）;

}

}

PVOID? tTest （PVOID? pvArg）

{

pcCrcb0 = Lw_Coroutine_Create（coroutine0， 2 * 1024， LW_NULL）;

if （pcCrcb0 == LW_NULL） {

return? （LW_NULL）;

}

pcCrcb1 = Lw_Coroutine_Create（coroutine1， 2 * 1024， LW_NULL）;

if （pcCrcb1 == LW_NULL） {

return? （LW_NULL）;

}

Lw_Coroutine_Yield（）;? ? ? ? ? ? ? ? ? ? /*? 使其他協(xié)程運行? ? ? ? ? ? ?*/

while （1） {

Lw_Coroutine_Resume（pcCrcb0）;

Lw_Time_Sleep（1）;

}

return? （（PVOID）1）;

}

int main （int argc， char *argv[]）

{

LW_HANDLE? hId;

hId = Lw_Thread_Create（"t_test"， tTest， LW_NULL， LW_NULL）;

if （hId == LW_HANDLE_INVALID） {

return? （PX_ERROR）;

}

Lw_Thread_Join（hId， LW_NULL）;

return? （ERROR_NONE）;

}

5、測試驗證

5.1 環(huán)境搭建

硬件環(huán)境：

● CPU：Intel（R） Core（TM） i3-6100 CPU @ 3.70GHz

● 內(nèi)存：4GB

● 硬盤：500GB

軟件環(huán)境

● SylixOS內(nèi)核版本：1.9.9-9

● SylixOS BSP版本：1.1.3

5.2 測試輸出

① 傳統(tǒng)消息隊列模式

② 高效協(xié)程模式

5.3 測試分析

根據(jù)代碼運行結(jié)果，可以看出生產(chǎn)者-消費者模型中，采用協(xié)程模式比傳統(tǒng)消息隊列模式時間消耗要小得多。

6、結(jié)語

在嵌入式領(lǐng)域的資源受限，生產(chǎn)消費模型簡單情況下，使用高效協(xié)程編程可以有效降低傳統(tǒng)消息隊列模式下同步和互斥機制、線程調(diào)度等帶來的時間消耗，提高生產(chǎn)者和消費者的運行效率。SylixOS內(nèi)核原生支持協(xié)程，而不是使用第三方庫模擬，使得SylixOS內(nèi)部的協(xié)程管理更加便捷高效，這樣使用SylixOS協(xié)程構(gòu)建的生產(chǎn)者-消費者模型也就更加高效。

參考文獻：

[1] 陳莉君. Linux內(nèi)核編程[M]. 北京：機械工業(yè)出版社，2004

[2] 劉俞. Linux多線程的互斥與同步控制及實踐. 電腦知識與技術(shù)[J]. 2005，（18）

[3] 曹聰，范廉明. 操作系統(tǒng)原理與分析. 科學(xué)出版社[M]. 2003，09