利用Protothread實(shí)現(xiàn)實(shí)時(shí)多任務(wù)系統(tǒng)

嵌入式程序框架一般類似于程序1所示結(jié)構(gòu)：系統(tǒng)中有3個(gè)任務(wù)——TaskA、TaskB、TaskC，均放置于主循環(huán)內(nèi)，在每一個(gè)循環(huán)周期內(nèi)都被執(zhí)行一次。在這種結(jié)構(gòu)中，能滿足系統(tǒng)實(shí)時(shí)性要求的條件是： （當(dāng)且僅當(dāng)）TaskA 、TaskB、TaskC三個(gè)任務(wù)的運(yùn)行時(shí)間之和要小于系統(tǒng)實(shí)時(shí)響應(yīng)的時(shí)間要求。在系統(tǒng)較為簡(jiǎn)單、任務(wù)運(yùn)行時(shí)間能滿足實(shí)時(shí)要求的情況下，可以采用這種最簡(jiǎn)單、最直接的順序執(zhí)行方式。但是更多的情形是，系統(tǒng)不僅要對(duì)一些事件做出實(shí)時(shí)響應(yīng)，并且還要承擔(dān)很多其他的非實(shí)時(shí)任務(wù)，并且這些非實(shí)時(shí)任務(wù)的運(yùn)行時(shí)間要遠(yuǎn)遠(yuǎn)超出了實(shí)時(shí)響應(yīng)時(shí)間的要求。傳統(tǒng)的這種程序結(jié)構(gòu)顯然不能滿足系統(tǒng)的實(shí)時(shí)性要求。通常的解決方案是，引入實(shí)時(shí)操作系統(tǒng)，由操作系統(tǒng)進(jìn)行任務(wù)的調(diào)度，優(yōu)先執(zhí)行實(shí)時(shí)任務(wù)，達(dá)到滿足系統(tǒng)實(shí)時(shí)性的要求。

程序1嵌入式程序框架
　　void main(void) {
　　　　Init();
　　　　while(1) {
　　　　　　TaskA();
　　　　　　TaskB();
　　　　　　TaskC();
　　　　}
　　}
　　void Interrupt_1(void) interrupt 1 {
　　　　…
　　}
　　void Interrupt_2(void) interrupt 2 {
　　　　…
　　}

一般來說，在嵌入式系統(tǒng)開發(fā)中引入實(shí)時(shí)操作系統(tǒng)有諸多優(yōu)點(diǎn)：
　　◆ 更好地支持多任務(wù)，實(shí)時(shí)性要求能夠得以保障；
　　◆ 程序開發(fā)更加容易，也更便于維護(hù)；
　　◆ 有利于提高系統(tǒng)的穩(wěn)定性和可靠性。但是，操作系統(tǒng)的引入也將帶來較多的系統(tǒng)開銷：
　　◆ 實(shí)時(shí)操作系統(tǒng)往往使用定時(shí)器中斷來切換任務(wù)，需要消耗不少的CPU處理時(shí)間；
　　◆ 實(shí)時(shí)操作系統(tǒng)在切換任務(wù)時(shí)需要保護(hù)當(dāng)前任務(wù)的執(zhí)行現(xiàn)場(chǎng)，這就需要為每個(gè)任務(wù)準(zhǔn)備足夠多的RAM空間來實(shí)現(xiàn)任務(wù)切換；
　　◆ 實(shí)時(shí)操作系統(tǒng)的本身也需要占用相當(dāng)數(shù)量的Flash空間和RAM空間。
　　如果這些系統(tǒng)開銷都在可承受的范圍內(nèi)，那么采用實(shí)時(shí)操作系統(tǒng)將是最佳的選擇。但是在很多應(yīng)用的場(chǎng)合，特別是系統(tǒng)的資源非常緊張的單片機(jī)應(yīng)用，實(shí)時(shí)操作系統(tǒng)帶來的系統(tǒng)開銷往往是不可接受的。而更換速度更快、RAM更大、Flash更多的CPU意味著成本的增加，且會(huì)降低產(chǎn)品的競(jìng)爭(zhēng)力。當(dāng)系統(tǒng)中的任務(wù)不須進(jìn)行非常復(fù)雜的優(yōu)先級(jí)調(diào)度，而且其任務(wù)也相對(duì)簡(jiǎn)單時(shí)，引入實(shí)時(shí)操作系統(tǒng)似有殺雞用牛刀之嫌。

1  Protothread的特點(diǎn)

Protothread是專為資源有限的系統(tǒng)設(shè)計(jì)的一種耗費(fèi)資源特別少并且不使用堆棧的線程模型，其特點(diǎn)是：
　　◆ 以純C語言實(shí)現(xiàn)，無硬件依賴性；
　　◆ 極少的資源需求，每個(gè)Protothread僅需要2個(gè)額外的字節(jié)；
　　◆ 可以用于有操作系統(tǒng)或無操作系統(tǒng)的場(chǎng)合；
　　◆ 支持阻塞操作且沒有棧的切換。

使用Protothread實(shí)現(xiàn)多任務(wù)的最主要的好處在于它的輕量級(jí)。每個(gè)Protothread不需要擁有自已的堆棧，所有的 Protothread共享同一個(gè)堆?？臻g，這一點(diǎn)對(duì)于RAM資源有限的系統(tǒng)尤為有利。相對(duì)于操作系統(tǒng)下的多任務(wù)而言，每個(gè)任務(wù)都有自已的堆?？臻g，這將消耗大量的RAM資源，而每個(gè)Protothread僅使用一個(gè)整型值保存當(dāng)前狀態(tài)。

2  Protothread的阻塞運(yùn)行機(jī)制

以下是一個(gè)典型的Protothread程序示例：

程序2Protothread程序示例
　　PT_THREAD(radio_wake_thread(struct pt *pt)) {
　　　　PT_BEGIN(pt);
　　　　while(1) {
　　　　　　radio_on();
　　　　　　timer_set(&timer, T_AWAKE);
　　　　　　PT_WAIT_UNTIL(pt, timer_expired(&timer));
　　　　　　timer_set(&timer, T_SLEEP);
　　　　　　if(!communication_complete()) {
　　　　　　　　PT_WAIT_UNTIL(pt, communication_complete()‖timer_expired(&timer));
　　　　　　}
　　　　　　if(!timer_expired(&timer)) {
　　　　　　　　radio_off();
　　　　　　　　PT_WAIT_UNTIL(pt, timer_expired(&timer));
　　　　　　}
　　　　}
　　　　PT_END(pt);
　　}

這是一個(gè)非常簡(jiǎn)單的無線通信的狀態(tài)切換程序①，展開Protothread的宏定義，便可以得到程序3所示的展開代碼：

程序3Protothread宏展開代碼
　　void radio_wake_thread(struct pt *pt) {
　　　　switch(pt﹥lc) {
　　　　　　case 0:
　　　　　　while(1) {
　　　　　　　　radio_on();
　　　　　　　　timer_set(&timer, T_AWAKE);
　　　　　　　　pt﹥lc = 8;
　　　　　　　　case 8:
　　　　　　　　　　if(!timer_expired(&timer)) {
　　　　　　　　　　　　return;
　　　　　　　　　　}
　　　　　　　　　　timer_set(&timer, T_SLEEP);
　　　　　　　　　　if(!communication_complete()) {
　　　　　　　　　　　　pt﹥lc = 13;
　　　　　　　　　　　　case 13:
　　　　　　　　　　　　if(!(communication_complete() ||timer_expired(&timer))) {
　　　　　　　　　　　　　　return;
　　　　　　　　　　　　}
　　　　　　　　　　}
　　　　　　　　　　if(!timer_expired(&timer)) {
　　　　　　　　　　　　radio_off();
　　　　　　　　　　　　pt﹥lc = 18;
　　　　　　　　　　　　case 18:
　　　　　　　　　　　　if(!timer_expired(&timer)) {
　　　　　　　　　　　　　　return;
　　　　　　　　　　　　}
　　　　　　　　　　}
　　　　　　　　}
　　　　　　}
　　　　}

當(dāng)Protothread程序運(yùn)行到PT_WAIT_UNTIL時(shí)，判斷其運(yùn)行條件是否滿足，若不滿足，則阻塞。通過比對(duì)程序2和程序3的程序代碼可以得知，Protothread的阻塞其實(shí)質(zhì)就是函數(shù)返回，只不過在返回前保存了當(dāng)前的阻塞位置，待下一次Protothread被調(diào)用時(shí)，直接跳到阻塞位置執(zhí)行，再次判斷運(yùn)行條件是否滿足，并執(zhí)行后續(xù)程序或繼續(xù)阻塞。

3  利用Protothread構(gòu)造實(shí)時(shí)多任務(wù)系統(tǒng)

與操作系統(tǒng)下的多任務(wù)不同，操作系統(tǒng)下的每個(gè)任務(wù)可在任意時(shí)刻被打斷并阻塞，Protothread僅能在程序員指定位置阻塞。用Protothread實(shí)現(xiàn)實(shí)時(shí)多任務(wù)，正是利用了Protothread在指定位置阻塞的特點(diǎn)，讓出執(zhí)行權(quán)限給更高優(yōu)先級(jí)的任務(wù)先運(yùn)行。

下面舉例說明如何利用Protothread構(gòu)造實(shí)時(shí)多任務(wù)系統(tǒng)。

系統(tǒng)要求：
　　TaskA實(shí)時(shí)任務(wù)，30 ms內(nèi)響應(yīng)，運(yùn)行時(shí)間<20 ms；
　　TaskB實(shí)時(shí)任務(wù)，200 ms內(nèi)響應(yīng)，運(yùn)行時(shí)間<40 ms；
　　TaskC非實(shí)時(shí)任務(wù)，響應(yīng)時(shí)間無要求，運(yùn)行時(shí)間>30 ms。

設(shè)計(jì)思路：

將TaskB和TaskC分成若干步，每步運(yùn)行時(shí)間不超過10 ms（這個(gè)時(shí)間可視系統(tǒng)需求而定，例如TaskA若為40 ms內(nèi)響應(yīng)，則每步可擴(kuò)至20 ms）。任務(wù)以3個(gè)Protothread的方式運(yùn)行。首先執(zhí)行TaskA，在TaskA執(zhí)行完成1次后，釋放執(zhí)行權(quán)限，讓TaskB和TaskC執(zhí)行。 TaskB或TaskC在每執(zhí)行1步之前檢查運(yùn)行時(shí)間，一旦發(fā)現(xiàn)30 ms內(nèi)不夠執(zhí)行1步時(shí)，阻塞運(yùn)行，讓出執(zhí)行權(quán)限給TaskA。同樣，TaskB和TaskC的調(diào)度關(guān)系也類似，先運(yùn)行TaskB，完成時(shí)釋放執(zhí)行權(quán)限，讓 TaskC執(zhí)行；TaskC在每執(zhí)行1步之前檢查運(yùn)行時(shí)間，若發(fā)現(xiàn)200 ms內(nèi)不夠執(zhí)行1步時(shí)，阻塞運(yùn)行，讓出執(zhí)行權(quán)限重新交給TaskB。

源程序（Task0TimeCounter、Task1TimeCounter為計(jì)數(shù)器，每毫秒加1）：
　　#include "ptsem.h"
　　#define TASKA_MAX_RUN_TIME20
　　#define TASKA_CYCLE_TIME30
　　#define TASKB_CYCLE_TIME200
　　#define TASK_STEP_TIME10
　　#define TASK0_VALID_TIME TASKA_CYCLE_TIME?TASK_STEP_TIME
　　#define TASK1_VALID_TIME TASKB_CYCLE_TIME?TASK_STEP_TIME?TASKA_MAX_RUN_TIME *2
　　/*按照PT_WAIT_UNTIL 的宏定義擴(kuò)展一個(gè)新宏：當(dāng)程序進(jìn)入阻塞時(shí)發(fā)送一信號(hào)，告知高優(yōu)先級(jí)任務(wù)獲得執(zhí)行權(quán)限*/
　　#define LC_STEP_SET(s,n) s = __LINE__ + n; case __LINE__ + n:
　　#define PT_SEM_WAIT_UNTIL(pt, s, condition, n)\
　　　　do {
　　　　　　LC_STEP_SET((pt)﹥lc,n);
　　　　　　if(!(condition)) {if((s)﹥count==0)
　　　　　　　　PT_SEM_SIGNAL(pt,s);
　　　　　　　　return PT_WAITING;
　　　　　　}
　　　　} while(0)
　　struct pt TaskAPt;
　　struct pt TaskBPt;
　　struct pt TaskCPt;
　　struct pt_sem SemRunTaskA;
　　struct pt_sem SemRunTaskB;
　　/*若30 ms內(nèi)已經(jīng)不夠時(shí)間執(zhí)行1步，則讓出TaskA的執(zhí)行權(quán)限*/
　　#define TASKB_STEP(pt) \
　　　　PR_SEM_WAIT_UNTIL(pt, & SemRunTaskA,TaskOTimeCounter<=TASKO_VALID_TIME,0)
　　/*若200 ms內(nèi)已經(jīng)不夠時(shí)間執(zhí)行1步，則讓出TaskB的執(zhí)行權(quán)限*/
　　/*若30 ms內(nèi)已經(jīng)不夠時(shí)間執(zhí)行1步，則讓出TaskA的執(zhí)行權(quán)限*/
　　#define TASKC_STEP(pt) \
　　　　PT_SEM_WAIT_UNTIL(pt, &SemRunTaskB,Task1TimeCounter<=TASK1_VALID_TIME,0);\
　　　　PT_SEM_WAIT_UNTIL(pt, &SemRunTaskA,Task0TimeCounter<=TASK0_VALID_TIME,1)
　　int ProtothreadTaskA(struct pt *pt) {
　　　　PT_BEGIN(pt);
　　　　PT_SEM_WAIT(pt, &SemRunTaskA);/*等待其他任務(wù)讓出執(zhí)行權(quán)限*/
　　　　ResetTask0TimeCounter;/*對(duì)時(shí)間計(jì)數(shù)器置0*/
　　　　TaskA();/*TaskA任務(wù)*/
　　　　PT_END(pt);
　　}
　　int ProtothreadTaskB(struct pt *pt) {
　　　　PT_BEGIN(pt);
　　　　PT_SEM_WAIT(pt, &SemRunTaskB);/*等待TaskC讓出執(zhí)行權(quán)限*/
　　　　ResetTask1TimeCounter;/*對(duì)時(shí)間計(jì)數(shù)器置0*/
　　　　TASKB_STEP(pt);/*如果不夠1步執(zhí)行，則阻塞，讓出執(zhí)行權(quán)限*/
　　　　TaskB_1();/*TaskB任務(wù)的第1步*/
　　　　TASKB_STEP(pt);
　　　　TaskB_2();/*TaskB任務(wù)的第2步*/
　　　　TASKB_STEP(pt);
　　　　TaskB_3();/*…*/
　　　　PT_END(pt);
　　}
　　int ProtothreadTaskC(struct pt *pt) {
　　　　PT_BEGIN(pt);
　　　　TASKC_STEP(pt);/*如果不夠1步執(zhí)行，則阻塞，讓出執(zhí)行權(quán)限*/
　　　　TaskC_1();/*TaskB任務(wù)的第1步*/
　　　　TASKC_STEP(pt);
　　　　TaskC_2();/*TaskB任務(wù)的第2步*/
　　　　TASKC_STEP(pt);
　　　　TaskC_3();/*…*/
　　　　TASKC_STEP(pt);
　　　　TaskC_4();
　　　　PT_END(pt);
　　}
　　void main(void) {/*系統(tǒng)初始化*/?
　　　　PT_INIT(&TaskAPt);
　　　　PT_INIT(&TaskBPt);
　　　　PT_INIT(&TaskCPt);
　　　　PT_SEM_INIT(&SemRunTaskA,1);
　　　　PT_SEM_INIT(&SemRunTaskB,1);/*運(yùn)行任務(wù)*/
　　　　while(1) {
　　　　　　ProtothreadTaskA(&TaskAPt);
　　　　　　ProtothreadTaskB(&TaskBPt);
　　　　　　ProtothreadTaskC(&TaskCPt);
　　　　}
　　} 　

模擬運(yùn)行結(jié)果如表1所列。運(yùn)行結(jié)果顯示，3個(gè)任務(wù)的運(yùn)行情況完全滿足系統(tǒng)的設(shè)計(jì)要求。從資源需求來看，完成此例的系統(tǒng)設(shè)計(jì)，共需要12個(gè)字節(jié)的RAM空間。筆者進(jìn)一步對(duì)Protothread定義文件做了少許修改和優(yōu)化，最終僅耗費(fèi)6個(gè)字節(jié)。

結(jié)語

本文旨在解決資源緊張型應(yīng)用的、多任務(wù)環(huán)境下的實(shí)時(shí)性問題。 通過借助Protothread的阻塞運(yùn)行機(jī)制， 成功實(shí)現(xiàn)了低開銷的實(shí)時(shí)多任務(wù)系統(tǒng)。
表1  模擬運(yùn)行結(jié)果運(yùn)行

參考文獻(xiàn)
[1]  Adam Dunkels, Oliver Schmidt, Thiemo Voigt. Using Protothreads for Sensor Node Programming[C]. REALWSN'05 Workshop on RealWorld Wireless Sensor Networks, Stockholm, Sweden, June 2005
[2]  Adam Dunkels, Oliver Schmidt, Thiemo Voigt, et al. Protothreads: Simplifying EventDriven Programming of MemoryConstrained Embedded Systems[C]. In Proceedings of the Fourth ACM Conference on Embedded Networked Sensor Systems (SenSys 2006), Boulder, Colorado, USA, November 2006.
[3]  Labrosse Jean J. MicroC/OSII The Real Time Kernel Second Edition[M]. CMP Books, CMP Media.
[4]  冉全. 單片機(jī)中基于多線程機(jī)制的實(shí)時(shí)多任務(wù)研究[J] .微型機(jī)與應(yīng)用，2003(8): 39-40.