文檔中心

TD開發平臺_基礎

03.任務運行系統

华东15选5 www.wjnoc.com 任務運行系統以任務為單位進行調度和運行,開發人員創建各種任務,當任務被觸發需要運行時,就加入到一個就緒隊列上,然后由若干個線程摘取就緒隊列上的任務,并執行該任務。這些執行線程的具體數量可以根據硬件核心的多少來確定,甚至可以在運行時動態添加和減少。
首先要解決的問題是時間的定義:

typedef Tint TTimeout;
#define TTIMEOUT_DIFF(a, b) ((Tint)((Tint)(a) - (Tint)(b)))

TTimeout定義為有符號整數,表示逝去的時間(毫秒為單位),可以理解為一個時間點,也可以理解為離0時刻的時間段。單個 TTimeout值因為回繞,無較大實際意義,但對任意兩個 TTimeout值,只要這兩個時間點在一輪計數的一半時間里 (231ms),那么這兩個值的差值可以正確表示它們之間的時間段。例如判定時刻 a是否在 b的前面,就是 TTIMEOUT_DIFF(a, b) <= 0,而不能直接用 (a < b)。
用下面的函數得到系統當前運行的時間(毫秒數),注意該函數永遠不會返回 0,時間回繞到 0時會返回 1。

TTimeout TTaskGetTimeout(void);

用戶可以創建三類任務:定時器任務,文件描述符任務和簡單任務。對于每一個任務都有如下幾種狀態變換:

所有任務在創建后都處于等待狀態,只有被觸發才進入就緒狀態等待運行。對于不同種類的任務,它們的觸發方式不同:定時器任務是每隔固定時間就自動觸發一次;文件描述符任務就是當某個文件句柄上有數據可讀或異常時自動觸發;簡單任務就是需要開發人員去主動觸發調用。若干個執行線程在空閑時會執行這些處于就緒狀態的任務(執行任務就是運行創建這些任務時設置的一個回調函數),任務執行完畢(即回調函數返回)后回到等待狀態。任務在任意狀態中都可以被銷毀,甚至在它的回調函數中都可以銷毀自己。
在整個任務系統的運轉中,如果開發人員安排有多個執行線程,那么怎么分配任務具體在哪個線程上運行是不可控的,即目前沒有任務和線程的綁定,有可能同一個任務上一次觸發和下一次觸發運行在兩個不同的線程上。任務是粗粒度的并發單位。但是任務系統有一個基本的保證:在一個任務被觸發直到運行完畢,這個任務是不能被再次觸發的,即任務的回調函數不會重入。
具體的任務類型:

1)定時器任務
typedef void (*TTaskTimer_Func) (TTaskTimer *ptimer, TTimeout timeout,void *arg);
TTaskTimer * TTaskAddTimer(TTimeout timeout, TTaskTimer_Func callback, void *arg);
void TTaskEnableTimer(TTaskTimer *timer, Tbool if_enable);
void TTaskResetTimer(TTaskTimer *timer);
void TTaskChangeTimeout(TTaskTimer *timer, TTimeout timeout);
TTaskTimer_Func是定時器任務的回調函數原型,TTaskAddTimer函數創建一個定時器任務,參數 timeout是自動觸發的時間間隔(毫秒為單位);參數 callback就是定時器任務的回調函數;參數 arg是開發人員設置的一個任意值,該值會在調用回調函數時,傳給回調函數的 arg參數。
注意一個定時器任務被創建后是永遠不會被觸發的,這時它的狀態是無效狀態(無效狀態是定時器任務特有的)。必須通過 TTaskEnableTimer函數使之有效后,才開始計時;同理以后也可以再使之進入無效狀態。當一個定時器任務被無效時,即使它已經處于就緒狀態,也將返回等待狀態,不會被運行;如果它已經在運行狀態,那么它的運行不受影響,回調函數仍然會執行完。
TTaskResetTimer函數用于重置定時器任務的計時,如一個定時器的間隔時間為 5秒,這時已經過去了 3秒,即 2秒之后該定時器會被觸發運行,如果這時調用TTaskResetTimer函數,那么將重新計時 5秒才會被觸發運行。對于已經在就緒或運行狀態的定時器,該函數不會取消,只是影響下一次的觸發時間。
TTaskChangeTimeout函數改變定時器的間隔時間,但是注意這個改變并不影響即將到來的觸發時間,只是影響下下次的觸發時間,如果要立即影響即將到來的觸發時間,只需要再調用 TTaskResetTimer即可。
下面是使用定時器的例子:
#include <TCore/TCore.h>
void timer1_cb(TTimer *ptimer, TTimeout timeout, void *arg)
{
    static int num=0;
    printf("timer1 : %d\n", ++num);
    return;
}
void timer2_cb(TTimer *ptimer, TTimeout timeout, void *arg)
{
    printf("timer2 come, exit\n");
    exit(0);
}
int main(int argc, char **argv)
{
    TTaskTimer *ptimer1, *ptimr2;
    ptimer1 = TTaskAddTimer(1000, timer1_cb, NULL);
    TTaskEnableTimer(ptimer1, TRUE);
    ptimer2 = TTaskAddTimer(5000, timer2_cb, NULL);
    TTaskEnableTimer(ptimer2, TRUE);
    while(1) TTaskLoopOnce(0);
    return 0;
}
例子中的 TTaskLoopOnce函數會等待就緒任務的產生,并執行任務。簡單的循環調
用該函數就是任務運行系統的事件循環。 TTaskLoopOnce函數在后面會進一步介紹。
2)文件描述符任務
當打開一個文件(普通文件或設備文件)、或創建一個管道或 socket,都會得到一個文件描述符,即文件句柄。文件描述符任務就是當這個文件上有指定事件發生時,自動觸發調用的一種任務。相關操作函數如下:
#define T_IO_READ 0x01
#define T_IO_PRI 0x02
#define T_IO_WRITE 0x04
#define T_IO_ERR 0x08
#define T_IO_HUP 0x10
typedef void (*TTaskFile_Func)(TTaskFile *pfd, Tint fd, Tint type, void*arg);
TTaskFile * TTaskAddFile(Tint fd, TTaskFile_Func callback, Tint type, void*arg);

首先定義文件描述符上發生的事件類型:
● T_IO_READ:該文件上有數據可讀;
● T_IO_PRI:該文件上有緊急的數據可讀(只用于 socket文件句柄);
● T_IO_WRITE:該文件可以寫入數據(不常用);
● T_IO_ERR:該文件異常,出錯;
● T_IO_HUP:該文件被斷開(常用于管道、socket和設備文件)

TTaskAddFile函數創建一個文件描述符任務,參數 fd是要監視的文件句柄,參數 type指明要監視的事件類型,可以同時監視多個事件類型( T_IO_READ|T_IO_PRI),參數 callback是任務的回調函數,參數 arg在執行任務時傳給回調函數的參數 arg?;氐骱橢械牟問?type表示實際發生的事件類型,注意:即使開發人員沒有要求監視 T_IO_ERR和 T_IO_HUP,它們也可能會發生,即 type可能等于 T_IO_ERR或 T_IO_HUP。
事件是電平觸發的,例如開發人員監視 T_IO_READ事件,假如開發人員在執行函數中沒有把文件中的數據讀完,那么在該任務執行完后,又會被立即觸發,直到數據被讀完。所以從性能上考慮,每次執行都應該把數據讀完。
下面是一個測試系統最大吞吐率的例子,通過 pipe系統調用創建兩個管道,然后創建兩個文件描述符任務來監視每個管道是否有數據可讀,在每個任務的執行函數中都把讀到的數據再發給另一個管道,再創建一個 1秒的定時器打印數據吞吐率。

#include <unistd.h>
#include <string.h>
#include <errno.h>
#include <sys/types.h>
#include <TCore/TCore.h>
#define TESTN (16*1024)
static int fd1[2], fd2[2];
static char fd1_buf[TESTN], fd2_buf[TESTN];
static int fd1_n, fd2_n;
void fd_cb(TTaskFile *pfd, Tint fd, Tint type, void *arg)
{
    int n;
    if(fd == fd1[0]) {
        n = read(fd, fd1_buf, TESTN);
        write(fd2[1], fd1_buf, n);
        fd1_n += n;
    } else {
        n = read(fd, fd2_buf, TESTN);
        write(fd1[1], fd2_buf, n);
        fd2_n += n;
    }
    return;
}
void timer_cb(TTaskTimer *ptimer, TTimeout timeout, void *arg)
{
    printf("fd1 read %d, fd2 read %d\n" fd1_n, fd2_n);
    fd1_n = 0;
    fd2_n = 0;
    return;
}
int main(int argc, char **argv)
{
    pipe(fd1);
    pipe(fd2);
    TTaskRegisterFd(fd1[0], fd_cb, T_IO_READ, 0);
    TTaskRegisterFd(fd2[0], fd_cb, T_IO_READ, 0);
    TTaskEnableTimer(TTaskAddTimer(1000, timer_cb, (void *)0),TRUE);
    write(fd1[1], fd1_buf, TESTN); /*啟動傳輸*/
    while(1) TTaskLoopOnce(0); /*執行事件循環*/
    return 0;
}
在CPU 為P4(1.7G),內存為DDR2(1G),系統為ubuntu7.04 的環境下運行,打印的 fd1_n和 fd2_n每次都在 320M左右。
3)簡單任務
簡單任務就是由開發人員主動觸發執行的任務,相關函數如下:
typedef void (*TTaskOnce_Func)(TTaskOnce *ponce, Tint n, void *arg);
TTaskOnce * TTaskAddOnce(TTaskOnce_Func callback, void *arg);
void TTaskTriggerOnce(TTaskOnce *once, Tint n);
void TTaskDestroyOnce(TTaskOnce *once);
TaskAddOnce創建一個簡單任務,參數 callback是任務的執行函數,參數 arg在執行任務時傳給執行函數的參數 arg。簡單任務創建后永遠不會自動觸發。
TTaskTriggerOnce用于主動觸發簡單任務,該函數有一個參數 n,這個 n會傳給執行函數的參數 n。當連續調用兩次 TTaskTriggerOnce觸發同一個簡單任務時(觸發函數的參數 n分別為 n1和 n2),那么該簡單任務可能被執行 1次,也可能被執行 2次。如果是執行兩次,那么執行函數的參數 n分別為 n1和 n2;如果是執行一次,那么執行函數的參數 n為(n1+n2)。也就是觸發函數總是在累計 n,執行函數總是消耗完當前的 n,只要 n大于 0就會觸發任務執行。注意在 TTaskTriggerOnce中并不會實際執行任務,僅僅只是觸發。
在其他基于事件響應的運行系統中,如 libevent,glib等,都提供了定時器任務和文件描述符任務,并沒有提供簡單任務,但是這種由開發人員觸發的任務和其他任務配合起來使用更加靈活,也可以用在多任務的同步協作上。定時器任務和文件描述符任務就像操作系統里的中斷處理函數,它們被自動觸發調用,但是為了快速響應中斷,它們讀取到外設的數據后并不立即處理,而是觸發另一個核心任務處理。
4)事件循環
在應用程序創建任務和做好其他初始化工作之后,必須進入一個循環處理過程。在這個過程中,應用程序等待各種事件來觸發預先設定好的任務運行,在沒有事件發生時,開發人員進程是休眠的,不占用 CPU。任務被觸發后,也是在事件循環過程中執行的。
void TTaskLoopOnce(TTimeout check_awake_timepoint);
例如:
while(1) TTaskLoopOnce(0);
TTaskLoopOnce函數是本系統的核心函數,如果已經有任務被觸發,該函數會執行完所有的任務,然后就返回;如果目前沒有任務被觸發,該函數會等待直到有一個任務被觸發,然后就返回。所以循環調用該函數就是本系統的事件循環。
本系統支持多個線程來并發執行任務,具體實現就是創建多個線程同時循環調用 TTaskLoopOnce函數,該函數內部會自動分配就緒任務到多個線程上執行,并且協調多個線程的同步和等待。所以 TTaskLoopOnce是支持重入的。
TTaskLoopOnce的參數 check_awake_timepoint一般為 0,具體用法和多任務的同步和互斥實現有關。
{ganrao}