基于SylixOS的高效生产者消费者模型编程应用设计

秦飞

摘 要：在SylixOS下，介紹了基于传统消息队列和高效协程模式两种生产者-消费者模型的编程方法，比较两种模式的编程方法在互斥和同步机制上的区别及其带来的消耗。

关键词：SylixOS;同步和互斥;协程

中图分类号：TP311.1            文献标识码：A

引言：

传统的生产者-消费者模型中，生产者和消费者之间的关系可以总结为如下几点：

●生产者与生产者：互斥关系

●消费者与消费者：互斥关系

●生产者与消费者：互斥与同步关系

要处理好这些关系，就要用到系统中的互斥与同步机制，并涉及到线程的阻塞和唤醒操作，而这些机制往往会带来一些必要的消耗，影响运行效率。

协程，又称作协同程序，是比线程还小的可执行代码序。在SylixOS中，一个线程可以拥有多个协程，这些协程共享了线程除了栈以外的所有资源，包括优先级、内核对象等。线程内部的协程不可被抢占，只能轮转运行。所以，利用这个特点可以构建出一个基于协程的高效的生产者-消费者模型。

1、何为生产者消费者模型

经典生产者-消费者模型描述如下：两个线程共享一个消息队列，其中一个线程，即生产者，向该消息队列中放入信息;另外一个线程，即消费者，从该消息队列中取走信息，这里就涉及到了多任务操作系统中的同步和互斥机制。

在SylixOS中，同步即一个任务的执行结果是另一个任务能够执行的前提条件，互斥即多任务同时使用到临界资源时可能会引发的竞争。在生产者-消费者模型中，消费者任务需要等待生产任务产生数据后才能执行，即同步;而生产任务和消费任务要串行地访问消息队列，即互斥。

2、生产者消费者模型的两种编程方法

2.1 传统消息队列方式

该方式中，生产者-消费者模型中的公共缓冲区即对应着消息队列。编程步骤如下：

① 创建生产者线程;

② 创建消费者线程;

③ 创建消息队列，获取操作句柄;

④ 生产者线程中循环调用消息发送函数，向消息队列中发送数据;

⑤ 消费者线程中循环调用消息接收函数，从消息队列中获取数据，若消息队列中无数据，则会阻塞。

2.2 高效率协程模式

SylixOS中的协程需要依赖所属线程被调度时执行，所以需要先创建一个初始线程，再去调用相应的函数创建生产者和消费者协程，需要注意的是，每一个线程创建时都会默认创建一个初始协程，并且线程总是从该初始协程开始运行，所以初始线程中需要主动放弃处理器，以保证生产者和消费者协程能运行。编程步骤如下：

① 创建初始线程;

② 创建生产者协程;

③ 创建消费者协程;

④ 初始线程放弃处理器调用;

⑤ 生产者协程向公共缓冲区填充消息后，主动放弃处理器;

⑥ 消费者协程从公共缓冲区获取消息后，主动放弃处理器;

⑦ 初始线程恢复生产者和消费者协程，跳转至步骤⑤。

2.3 两种编程模式的比较

通过上述两种模式的步骤比较，可以得出以下结论：

① 传统消息队列模式中，需要创建多线程并进行线程间的切换，同时生产者和消费者线程间的通信采用消息队列的形式，其发送和接收函数内部实现涉及到进入和退出内核态的模式切换，以保证生产者和消费者线程中的同步和互斥机制，这里面会有较大的时间消耗;

② 高效协程模式中，只有一个初始线程，内部的协程并不会被调度器调度，所以不会有线程切换的消耗。另外，初始线程内部的协程是以轮转的方式运行，由用户程序自行管理，所以也就不会有线程间通信带来的消耗。

综上，可以看出，协程模式比传统消息队列模式更高效。

3、模型编程流程设计

基于两种方法设计应用流程图及其简单说明

3.1 消息队列方法

3.2 协程方法

4、代码设计

4.1 消息队列方法

基于消息队列的设计代码中，生产者线程获取当前时间t1并填入消息队列中，消费者线程从消息队列中获取消息后再次获取当前时间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 協程方法

基于协程的设计代码中，生产者协程将数据放入公共缓冲区，并获取当前时间t1，然后放弃CPU，接着消费者协程中获取时间t2，并从公共缓冲区获取数据，将t1和t2的差值作为生产者和消费者协程切换的时间消耗。示例代码如下。

#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（）;                    /*  使其他協程运行             */

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 环境搭建

硬件环境：

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

● 内存：4GB

● 硬盘：500GB

软件环境

● SylixOS内核版本：1.9.9-9

● SylixOS BSP版本：1.1.3

5.2 测试输出

① 传统消息队列模式

② 高效协程模式

5.3 测试分析

根据代码运行结果，可以看出生产者-消费者模型中，采用协程模式比传统消息队列模式时间消耗要小得多。

6、结语

在嵌入式领域的资源受限，生产消费模型简单情况下，使用高效协程编程可以有效降低传统消息队列模式下同步和互斥机制、线程调度等带来的时间消耗，提高生产者和消费者的运行效率。SylixOS内核原生支持协程，而不是使用第三方库模拟，使得SylixOS内部的协程管理更加便捷高效，这样使用SylixOS协程构建的生产者-消费者模型也就更加高效。

参考文献：

[1] 陈莉君. Linux内核编程[M]. 北京：机械工业出版社，2004

[2] 刘俞. Linux多线程的互斥与同步控制及实践. 电脑知识与技术[J]. 2005，（18）

[3] 曹聪，范廉明. 操作系统原理与分析. 科学出版社[M]. 2003，09