一、前言

隨著數字經濟的發展，作為數字基礎設施根技術的操作系統成為數字變革的關鍵力量，OpenAtom OpenHarmony（以下簡稱“OpenHarmony”） 以泛智能終端數字為底座支撐著千行百業的產業生態。

構建開源生態，需要讓開發者先用起來，本文希望通過分享 OpenHarmony 的 LiteOS-M 內核對象隊列的算法詳解，讓大家對這一算法有更加清晰的認識。 OpenHarmony 當前分為以下幾種系統類型：輕量系統 、小型系統、標準系統。針對不同量級的系統，分別使用了不同形態的內核。在輕量系統上，可以選擇 LiteOS-M；在小型系統和標準系統上，可以選用 LiteOS-A；在標準系統上，可以選用 Linux。 在輕小型系統中，OpenHarmony 所使用的內核為 LiteOS，在標準系統中使用 Linux。LiteOS-M 在面向 loT 領域構建了一款輕量級物聯網操作系統內核，嵌入式從業者如果能更好地掌握內核相關的知識，就能在未來做研發或者定制產品的時候獨當一面。

二、關鍵數據結構

首先關注隊列的關鍵數據結構 LosQueueCB，有了這個數據，才能理解隊列是如何工作的：

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：指向消息節點內存區域，創建隊列時按照消息節點個數乘每個節點大小從動態內存池中申請一片空間。

queueState：隊列狀態，表明隊列控制塊是否被使用，有 OS_QUEUE_INUSED和OS_QUEUE_UNUSED 兩種狀態。

queueLen：消息節點個數，表示該消息隊列最大可存儲多少個消息。

queueSize：每個消息節點大小，表示隊列每個消息可存儲信息的大小。

queueID：消息 ID，通過它來操作隊列。

消息節點按照循環隊列的方式訪問，隊列中的每個節點以數組下標表示，下面的成員與消息節點循環隊列有關：

queueHead：循環隊列的頭部。

queueTail：循環隊列的尾部。

readWriteableCnt[OS_QUEUE_WRITE]：消息節點循環隊列中可寫的消息個數，為 0 表示循環隊列為滿，等于 queueLen 表示循環隊列為空。

readWriteableCnt[OS_QUEUE_READ]：消息節點循環隊列中可讀的消息個數，為 0 表示循環隊列為空，等于 queueLen 表示消息隊列為滿。

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

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

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

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

三、關鍵算法

隊列的算法和 FIFO、FILO 有關，今天先給大家介紹 FIFO 算法。

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

那么 OpenHarmony 的隊列如何實現這個算法？

3.1 FIFO算法之入隊列

第一步：隊列初始化

由于 LOS_QueueCreate 函數太長，便只截取關鍵函數 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;}

數據結構是支撐算法的靈魂，內核對象的隊列控制結構 LosQueueCB 通過 queue 指針來指向具體隊列的內容，隊列分配了 queueLen 個消息，每個消息的大小為 queueSize，與此同時頭指針和尾指針不約而同初始化為 0。

第二步：第一個消息入隊列

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

下圖是 FIFO 插入第一個數據后的內存形態。

OpenHarmony 作為一個開源系統，在下面的代碼中很好地體現了這個操作：

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 是隊列內存的核心操作函數，FIFO 算法本質是往隊列的尾處添加數據，代碼抽象為 OS_QUEUE_WRITE_TAIL 操作，請注意隊列是個循環隊列，插入數據后移動 tail 這個“尾巴”指針要尤為小心，在最后一個物理空間用完成后需要移到隊列頭部，這就是環形隊列的“循環大法”。

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

第三步：繼續生產數據

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

第四步：生產數據結束

生產者生產了四個消息后就結束了。

3.2 FIFO算法之出隊列

第一步：隊列第一個消息

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

在生產消息過程中我們已經提到 OsQueueBufferOperate 這個函數，我們回顧關鍵代碼：

/*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 就是我們的頭指針，它的移動也面臨著生產過程相同的問題，在最后一個物理空間用完成后需要移到隊列的頭部。OS_QUEUE_READ_HEAD 是出隊列的關鍵處理，解決了 queueHead 頭指針如何移動的問題。

第二步：繼續消費

第三步：消費完畢

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

四、總結

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

當然隊列算法也不遠遠如此，linux 標準內核有加權隊列等更復雜的算法。但是“他山之石，可以攻玉”，技術萬變不離其宗，掌握了 FIFO 的細節有助于工程師設計其它隊列算法，也能夠把更多更新的技術帶入到 OpenHarmony 社區，繁榮開源生態。