秦飛

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

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

中圖分類(lèi)號(hào)：TP311.1? ? ? ? ? ? 文獻(xiàn)標(biāo)識(shí)碼：A

引言：

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

2.3 兩種編程模式的比較

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

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

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

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

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

基于兩種方法設(shè)計(jì)應(yīng)用流程圖及其簡(jiǎn)單說(shuō)明

3.1 消息隊(duì)列方法

3.2 協(xié)程方法

4、代碼設(shè)計(jì)

4.1 消息隊(duì)列方法

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

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

#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é)程運(yùn)行? ? ? ? ? ? ?*/

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、測(cè)試驗(yàn)證

5.1 環(huán)境搭建

硬件環(huán)境：

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

● 內(nèi)存：4GB

● 硬盤(pán)：500GB

軟件環(huán)境

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

● SylixOS BSP版本：1.1.3

5.2 測(cè)試輸出

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

② 高效協(xié)程模式

5.3 測(cè)試分析

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

6、結(jié)語(yǔ)

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

參考文獻(xiàn)：

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

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

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