LiteOS-M內(nèi)核隊(duì)列的關(guān)鍵數(shù)據(jù)結(jié)構(gòu)及關(guān)鍵算法

一、前言

隨著數(shù)字經(jīng)濟(jì)的發(fā)展，作為數(shù)字基礎(chǔ)設(shè)施根技術(shù)的操作系統(tǒng)成為數(shù)字變革的關(guān)鍵力量，OpenAtom OpenHarmony（以下簡稱“OpenHarmony”） 以泛智能終端數(shù)字為底座支撐著千行百業(yè)的產(chǎn)業(yè)生態(tài)。

構(gòu)建開源生態(tài)，需要讓開發(fā)者先用起來，本文希望通過分享 OpenHarmony 的 LiteOS-M 內(nèi)核對象隊(duì)列的算法詳解，讓大家對這一算法有更加清晰的認(rèn)識。 OpenHarmony 當(dāng)前分為以下幾種系統(tǒng)類型：輕量系統(tǒng) 、小型系統(tǒng)、標(biāo)準(zhǔn)系統(tǒng)。針對不同量級的系統(tǒng)，分別使用了不同形態(tài)的內(nèi)核。在輕量系統(tǒng)上，可以選擇 LiteOS-M；在小型系統(tǒng)和標(biāo)準(zhǔn)系統(tǒng)上，可以選用 LiteOS-A；在標(biāo)準(zhǔn)系統(tǒng)上，可以選用 Linux。 在輕小型系統(tǒng)中，OpenHarmony 所使用的內(nèi)核為 LiteOS，在標(biāo)準(zhǔn)系統(tǒng)中使用 Linux。LiteOS-M 在面向 loT 領(lǐng)域構(gòu)建了一款輕量級物聯(lián)網(wǎng)操作系統(tǒng)內(nèi)核，嵌入式從業(yè)者如果能更好地掌握內(nèi)核相關(guān)的知識，就能在未來做研發(fā)或者定制產(chǎn)品的時候獨(dú)當(dāng)一面。

二、關(guān)鍵數(shù)據(jù)結(jié)構(gòu)

首先關(guān)注隊(duì)列的關(guān)鍵數(shù)據(jù)結(jié)構(gòu) LosQueueCB，有了這個數(shù)據(jù)，才能理解隊(duì)列是如何工作的：

typedefstruct{    UINT8 *queue;      /**< Pointer to a queue handle */    UINT16 queueState; /**< Queue state */    UINT16 queueLen;   /**< Queue length */    UINT16 queueSize;  /**< Node size */    UINT16 queueID;    /**< queueID */    UINT16 queueHead;  /**< Node head */    UINT16 queueTail;  /**< Node tail */    UINT16 readWriteableCnt[OS_READWRITE_LEN]; /**< Count of readable or writable resources, 0:readable, 1:writable */    LOS_DL_LIST readWriteList[OS_READWRITE_LEN]; /**< Pointer to the linked list to be read or written,                                                      0:readlist, 1:writelist */    LOS_DL_LIST memList; /**< Pointer to the memory linked list */}LosQueueCB;

*queue：指向消息節(jié)點(diǎn)內(nèi)存區(qū)域，創(chuàng)建隊(duì)列時按照消息節(jié)點(diǎn)個數(shù)乘每個節(jié)點(diǎn)大小從動態(tài)內(nèi)存池中申請一片空間。

queueState：隊(duì)列狀態(tài)，表明隊(duì)列控制塊是否被使用，有 OS_QUEUE_INUSED和OS_QUEUE_UNUSED 兩種狀態(tài)。

queueLen：消息節(jié)點(diǎn)個數(shù)，表示該消息隊(duì)列最大可存儲多少個消息。

queueSize：每個消息節(jié)點(diǎn)大小，表示隊(duì)列每個消息可存儲信息的大小。

queueID：消息 ID，通過它來操作隊(duì)列。

消息節(jié)點(diǎn)按照循環(huán)隊(duì)列的方式訪問，隊(duì)列中的每個節(jié)點(diǎn)以數(shù)組下標(biāo)表示，下面的成員與消息節(jié)點(diǎn)循環(huán)隊(duì)列有關(guān)：

queueHead：循環(huán)隊(duì)列的頭部。

queueTail：循環(huán)隊(duì)列的尾部。

readWriteableCnt[OS_QUEUE_WRITE]：消息節(jié)點(diǎn)循環(huán)隊(duì)列中可寫的消息個數(shù)，為 0 表示循環(huán)隊(duì)列為滿，等于 queueLen 表示循環(huán)隊(duì)列為空。

readWriteableCnt[OS_QUEUE_READ]：消息節(jié)點(diǎn)循環(huán)隊(duì)列中可讀的消息個數(shù)，為 0 表示循環(huán)隊(duì)列為空，等于 queueLen 表示消息隊(duì)列為滿。

readWriteList[OS_QUEUE_WRITE]：寫消息阻塞鏈表，鏈接因消息隊(duì)列滿而無法寫入時需要掛起的 TASK。

readWriteList[OS_QUEUE_READ]：讀消息阻塞鏈表，鏈接因消息隊(duì)列空而無法讀取時需要掛起的 TASK。

memList：申請內(nèi)存塊阻塞鏈表，鏈接因申請某一靜態(tài)內(nèi)存池中的內(nèi)存塊失敗而需要掛起的 TASK。

注意：在老的版本中，readWriteableCnt 和 readWriteList 拆分為 4 個變量，新版本用宏定義合并，OS_QUEUE_READ 標(biāo)識是讀操作，OS_QUEUE_WRITE 標(biāo)識為寫操作。從中可看到代碼的微妙之處，0 的含義和 queueLen 對于讀寫是統(tǒng)一的，內(nèi)核開發(fā)者不斷使用抽象手段來優(yōu)化內(nèi)核。

三、關(guān)鍵算法

隊(duì)列的算法和 FIFO、FILO 有關(guān)，今天先給大家介紹 FIFO 算法。

百度定義：FIFO（First Input First Output），即先進(jìn)先出隊(duì)列。例如，在超市購物之后我們會到收銀臺排隊(duì)結(jié)賬，看著前面的客戶一個個離開，這就是一種先進(jìn)先出機(jī)制，先排隊(duì)的客戶先行結(jié)賬離開。

那么 OpenHarmony 的隊(duì)列如何實(shí)現(xiàn)這個算法？

3.1 FIFO算法之入隊(duì)列

第一步：隊(duì)列初始化

由于 LOS_QueueCreate 函數(shù)太長，便只截取關(guān)鍵函數(shù) LOS_QueueCreate。

LITE_OS_SEC_TEXT_INITUINT32LOS_QueueCreate(CHAR*queueName,                                         UINT16 len,                                         UINT32 *queueID,                                         UINT32 flags,                                         UINT16 maxMsgSize){    LosQueueCB *queueCB = NULL;    UINT32 intSave;    LOS_DL_LIST *unusedQueue = NULL;    UINT8 *queue = NULL;    UINT16 msgSize;    ...    queue = (UINT8 *)LOS_MemAlloc(m_aucSysMem0, len * msgSize);    ...    queueCB->queueLen = len;    queueCB->queueSize = msgSize;    queueCB->queue = queue;    queueCB->queueState = OS_QUEUE_INUSED;    queueCB->readWriteableCnt[OS_QUEUE_READ] = 0;    queueCB->readWriteableCnt[OS_QUEUE_WRITE] = len;    queueCB->queueHead = 0;    queueCB->queueTail = 0;    LOS_ListInit(&queueCB->readWriteList[OS_QUEUE_READ]);    LOS_ListInit(&queueCB->readWriteList[OS_QUEUE_WRITE]);    LOS_ListInit(&queueCB->memList);    LOS_IntRestore(intSave);
    *queueID = queueCB->queueID;    OsHookCall(LOS_HOOK_TYPE_QUEUE_CREATE, queueCB);
    return LOS_OK;}

數(shù)據(jù)結(jié)構(gòu)是支撐算法的靈魂，內(nèi)核對象的隊(duì)列控制結(jié)構(gòu) LosQueueCB 通過 queue 指針來指向具體隊(duì)列的內(nèi)容，隊(duì)列分配了 queueLen 個消息，每個消息的大小為 queueSize，與此同時頭指針和尾指針不約而同初始化為 0。

第二步：第一個消息入隊(duì)列

生產(chǎn)者通過隊(duì)列來傳遞信息，這個生產(chǎn)者可以是形形色色的各個任務(wù)，產(chǎn)生一個隊(duì)列后，任務(wù)就迫不及待的需要放置消息，選擇 FIFO 還是 FILO？這一次我們選擇了 FIFO。

下圖是 FIFO 插入第一個數(shù)據(jù)后的內(nèi)存形態(tài)。

OpenHarmony 作為一個開源系統(tǒng)，在下面的代碼中很好地體現(xiàn)了這個操作：

staticINLINEVOIDOsQueueBufferOperate(LosQueueCB*queueCB,UINT32operateType,                                                            VOID *bufferAddr, UINT32 *bufferSize){    UINT8 *queueNode = NULL;    UINT32 msgDataSize;    UINT16 queuePosition;    errno_t rc;
    /* get the queue position */    switch (OS_QUEUE_OPERATE_GET(operateType)) {        case OS_QUEUE_READ_HEAD:            queuePosition = queueCB->queueHead;            ((queueCB->queueHead + 1) == queueCB->queueLen) ? (queueCB->queueHead = 0) : (queueCB->queueHead++);            break;
        case OS_QUEUE_WRITE_HEAD:            (queueCB->queueHead == 0) ? (queueCB->queueHead = (queueCB->queueLen - 1)) : (--queueCB->queueHead);            queuePosition = queueCB->queueHead;            break;
        case OS_QUEUE_WRITE_TAIL:            queuePosition = queueCB->queueTail;            ((queueCB->queueTail + 1) == queueCB->queueLen) ? (queueCB->queueTail = 0) : (queueCB->queueTail++);            break;    ...}

OsQueueBufferOperate 是隊(duì)列內(nèi)存的核心操作函數(shù)，F(xiàn)IFO 算法本質(zhì)是往隊(duì)列的尾處添加數(shù)據(jù)，代碼抽象為 OS_QUEUE_WRITE_TAIL 操作，請注意隊(duì)列是個循環(huán)隊(duì)列，插入數(shù)據(jù)后移動 tail 這個“尾巴”指針要尤為小心，在最后一個物理空間用完成后需要移到隊(duì)列頭部，這就是環(huán)形隊(duì)列的“循環(huán)大法”。

如何判斷最后一個物理空間已經(jīng)用完？(queueCB->queueTail + 1) == queueCB->queueLen)C 語言語句很好地解釋了這個疑問。queueLen 是隊(duì)列物理空間的邊界值，如果下一個消息已經(jīng)指到這個邊界值，那么內(nèi)核必須讓它回到原位，即 queueCB->queueTail = 0，不然可能會出現(xiàn)“內(nèi)存越界”的問題，可能會造成機(jī)毀物亡。因?yàn)?OpenHarmony 應(yīng)用在各個領(lǐng)域，如果是自動化駕駛領(lǐng)域那么造成的后果非常嚴(yán)重。

第三步：繼續(xù)生產(chǎn)數(shù)據(jù)

接下來，再來一些圖片示例：

第四步：生產(chǎn)數(shù)據(jù)結(jié)束

生產(chǎn)者生產(chǎn)了四個消息后就結(jié)束了。

3.2 FIFO算法之出隊(duì)列

第一步：隊(duì)列第一個消息

如上圖所示我們回顧下入隊(duì)列的步驟，知道了每個消息的入隊(duì)順序，于是第一個消息被消費(fèi)后:

在生產(chǎn)消息過程中我們已經(jīng)提到 OsQueueBufferOperate 這個函數(shù)，我們回顧關(guān)鍵代碼：

/*getthequeueposition*/switch (OS_QUEUE_OPERATE_GET(operateType)) {    case OS_QUEUE_READ_HEAD:        queuePosition = queueCB->queueHead;        ((queueCB->queueHead + 1) == queueCB->queueLen) ? (queueCB->queueHead = 0) : (queueCB->queueHead++);        break;

queueHead 就是我們的頭指針，它的移動也面臨著生產(chǎn)過程相同的問題，在最后一個物理空間用完成后需要移到隊(duì)列的頭部。OS_QUEUE_READ_HEAD 是出隊(duì)列的關(guān)鍵處理，解決了 queueHead 頭指針如何移動的問題。

第二步：繼續(xù)消費(fèi)

第三步：消費(fèi)完畢

最后一個消息也消失了，head指針和tail指針均移動到下圖的位置，此時隊(duì)列為空。

四、總結(jié)

本文主要介紹了 OpenHarmony 內(nèi)核對象隊(duì)列的算法之 FIFO，在后續(xù)的篇章中將給大家介紹內(nèi)核對象隊(duì)列另外一種算法——FILO。希望通過這篇文章，可以讓開發(fā)者們對于目前 OpenHarmony LiteOS-M 內(nèi)核隊(duì)列算法有了更全面的概念。

當(dāng)然隊(duì)列算法也不遠(yuǎn)遠(yuǎn)如此，linux 標(biāo)準(zhǔn)內(nèi)核有加權(quán)隊(duì)列等更復(fù)雜的算法。但是“他山之石，可以攻玉”，技術(shù)萬變不離其宗，掌握了 FIFO 的細(xì)節(jié)有助于工程師設(shè)計其它隊(duì)列算法，也能夠把更多更新的技術(shù)帶入到 OpenHarmony 社區(qū)，繁榮開源生態(tài)。