佇列基本概念

佇列是一種常用于任務間通信的資料結構，佇列可以在任務與任務間、中斷和任務間傳遞訊息，實作了任務接收來自其他任務或中斷的不固定長度的訊息，任務能夠從佇列里面讀取訊息，當佇列中的訊息是空時，讀取訊息的任務將被阻塞，用戶還可以指定任務等待訊息的時間timeout，在這段時間中，如果佇列為空，該任務將保持阻塞狀態以等待佇列資料有效，當佇列中有新訊息時，被阻塞的任務會被喚醒并處理新訊息；當等待的時間超過了指定的阻塞時間，即使佇列中尚無有效資料，任務也會自動從阻塞態轉為就緒態，訊息佇列是一種異步的通信方式，

通過佇列服務，任務或中斷服務例程可以將一潭訓多條訊息放入佇列中，同樣，一個或多個任務可以從佇列中獲得訊息，當有多個訊息發送到佇列時，通常是將先進入佇列的訊息先傳給任務，也就是說，任務先得到的是最先進入佇列的訊息，即先進先出原則（FIFO），其實TencentOS tiny暫時不支持后進先出原則LIFO操作佇列，但是支持后進先出操作訊息佇列，

提示：TencentOS tiny 的佇列不等同于訊息佇列，雖然佇列的底層實作是依賴訊息佇列，但在TencentOS tiny中將它們分離開，這是兩個概念，畢竟操作是不一樣的，

佇列的阻塞機制

舉個簡單的例子來理解作業系統中的阻塞機制：

假設你某天去餐廳吃飯，但是餐廳沒菜了，那么你可能會有3個選擇，你扭頭就走，既然都沒菜了，肯定換一家餐廳啊是吧，或者你會選擇等一下，說不定老板去買菜了，一會就有菜了呢，就能吃飯，又或者，你覺得這家餐廳非常好吃，吃不到飯你就不走了，在這死等~

同樣的：假設有一個任務A對某個佇列進行讀操作的時候（出隊），發現它此時是沒有訊息的，那么此時任務A有3個選擇：第一個選擇，任務A扭頭就走，既然佇列沒有訊息，那我也不等了，干其它事情去，這樣子任務A不會進入阻塞態；第二個選擇，任務A還是在這里等等吧，可能過一會佇列就有訊息，此時任務A會進入阻塞狀態，在等待著訊息的到來，而任務A的等待時間就由我們自己指定，當阻塞的這段時間中任務A等到了佇列的訊息，那么任務A就會從阻塞態變成就緒態；假如等待超時了，佇列還沒訊息，那任務A就不等了，從阻塞態中喚醒；第三個選擇，任務A死等，不等到訊息就不走了，這樣子任務A就會進入阻塞態，直到完成讀取佇列的訊息，

佇列實作的資料結構

佇列控制塊

TencentOS tiny 通過佇列控制塊操作佇列，其資料型別為k_queue_t，佇列控制塊由多個元素組成，主要有 pend_obj_t 型別的pend_obj以及k_msg_queue_t 型別的msg_queue訊息串列，其實整個佇列的實作非常簡單，主要靠msg_queue中的queue_head成員變數（這其實是一個訊息串列（訊息鏈表）），所有的訊息都會被記錄在這個訊息串列中，當讀取訊息的時候，會從訊息串列讀取訊息，

繼承自內核物件的資料結構 在 \kernel\core\include\tos_pend.h 的 35 行

typedef struct pend_object_st {
    pend_type_t     type;
    k_list_t        list;
} pend_obj_t;

訊息串列的資料型別（訊息佇列控制塊），在 \kernel\core\include\tos_msg.h 檔案的第 13 行

typedef struct k_msg_queue_st {
#if TOS_CFG_OBJECT_VERIFY_EN > 0u
    knl_obj_t       knl_obj;
#endif

    k_list_t        queue_head;
} k_msg_queue_t;

佇列控制塊，在 \kernel\core\include\tos_queue.h 檔案的第 6 行

typedef struct k_queue_st {
    pend_obj_t      pend_obj;
    k_msg_queue_t msg_queue;
} k_queue_t;

佇列控制塊示意圖如下：

訊息控制塊

除了上述的佇列控制塊外，還有訊息佇列控制塊，這是因為TencentOS tiny中實作佇列是依賴訊息佇列的，既然佇列可以傳遞資料（訊息），則必須存在一種可以存盤訊息的資料結構，我稱之為訊息控制塊，訊息控制塊中記錄了訊息的存盤地址msg_addr，以及訊息的大小msg_size，此外還存在一個list成員變數，可以將訊息掛載到佇列的訊息串列中，

訊息控制塊資料結構， 在 \kernel\core\include\tos_msg.h 檔案的第 7 行

typedef struct k_message_st {
    k_list_t        list;
    void           *msg_addr;
    size_t          msg_size;
} k_msg_t;

其實佇列的實作依賴于訊息佇列，他們的關系如下：

任務控制塊中的訊息成員變數

假設任務A在佇列中等待訊息，而中斷或其他任務往任務A等待的佇列寫入（發送）一個訊息，那么這個訊息不會被掛載到佇列的訊息串列中，而是會直接被記錄在任務A的任務控制塊中，表示任務A從佇列中等待到這個訊息，因此任務控制塊必須存在一些成員變數用于記錄訊息相關資訊（如訊息地址、訊息大小等）：

任務控制塊資料結構 在\kernel\core\include\tos_task.h檔案的第 90 行

typedef struct k_task_st {
···
#if TOS_CFG_MSG_EN > 0u
    void               *msg_addr;           /**< 保存接收到的訊息地址 */
    size_t              msg_size;			/**< 保存接收到的訊息大小 */
#endif
···
} k_task_t;

與訊息相關的宏定義

在tos_config.h檔案中，使能佇列組件的宏定義TOS_CFG_QUEUE_EN，使能訊息佇列組件宏定義TOS_CFG_MSG_EN，系統支持的訊息池中訊息個數宏定義TOS_CFG_MSG_POOL_SIZE，

#define TOS_CFG_QUEUE_EN				1u

#define TOS_CFG_MSG_EN					1u

#define TOS_CFG_MSG_POOL_SIZE           3u

訊息池

在TencentOS tiny中定義了一個陣列k_msg_pool[TOS_CFG_MSG_POOL_SIZE]作為訊息池，它的資料型別是訊息控制塊型別k_msg_t，因為在使用訊息佇列的時候存取訊息比較頻繁，而在系統初始化的時候就將這個大陣列的各個元素串初始化，并掛載到空閑訊息串列中k_msg_freelist，組成我們說的訊息池k_msg_pool，而池中的成員變數就是我們所說的訊息，

為什么使用池化的方式處理訊息呢，因為高效，復用率高，就像我們在池塘中去一勺水，在使用完畢再將其歸還到池塘，這種操作是非常高效的，并且在有限資源的嵌入式中能將資源重復有效地利用起來，

訊息池相關的定義 在\kernel\core\tos_global.c檔案 第 51 行

#if TOS_CFG_MSG_EN > 0u
TOS_LIST_DEFINE(k_msg_freelist);

k_msg_t             k_msg_pool[TOS_CFG_MSG_POOL_SIZE];

#endif

佇列創建

tos_queue_create()函式用于創建一個佇列，佇列就是一個資料結構，用于任務間的資料的傳遞，每創建一個新的佇列都需要為其分配RAM，在創建的時候我們需要自己定義一個佇列控制塊，其記憶體是由編譯器自動分配的，在創建的程序中實際上就是將佇列控制塊的內容進行初始化，將佇列控制塊的 pend_obj成員變數中的type 屬性標識為PEND_TYPE_QUEUE，表示這是一個佇列，然后呼叫訊息佇列中的API函式tos_msg_queue_create()將佇列的訊息成員變數msg_queue初始化，實際上就是初始化訊息串列，

創建佇列函式，在\kernel\core\tos_queue.c 第 5 行

__API__ k_err_t tos_queue_create(k_queue_t *queue)
{
    TOS_PTR_SANITY_CHECK(queue);

    pend_object_init(&queue->pend_obj, PEND_TYPE_QUEUE);
    tos_msg_queue_create(&queue->msg_queue);
    return K_ERR_NONE;
}

銷毀佇列

tos_queue_destroy()函式用于銷毀一個佇列，當佇列不在使用是可以將其銷毀，銷毀的本質其實是將佇列控制塊的內容進行清除，首先判斷一下佇列控制塊的型別是PEND_TYPE_QUEUE，這個函式只能銷毀佇列型別的控制塊，然后判斷是否有任務在等待佇列中的訊息，如果有則呼叫pend_wakeup_all()函式將這項任務喚醒，然后呼叫tos_msg_queue_flush()函式將佇列的訊息串列的訊息全部“清空”，“清空”的意思是將掛載到佇列上的訊息釋放回訊息池（如果佇列的訊息串列存在訊息，使用msgpool_free()函式釋放訊息），knl_object_deinit()函式是為了確保佇列已經被銷毀，此時佇列控制塊的pend_obj成員變數中的type 屬性標識為KNL_OBJ_TYPE_NONE，最后在銷毀佇列后進行一次任務調度，以切換任務（畢竟剛剛很可能喚醒了任務），

但是有一點要注意，因為佇列控制塊的RAM是由編譯器靜態分配的，所以即使是銷毀了佇列，這個記憶體也是沒辦法釋放的~

銷毀佇列函式，在\kernel\core\tos_queue.c 第 14 行

__API__ k_err_t tos_queue_destroy(k_queue_t *queue)
{
    TOS_CPU_CPSR_ALLOC();

    TOS_PTR_SANITY_CHECK(queue);

#if TOS_CFG_OBJECT_VERIFY_EN > 0u
    if (!pend_object_verify(&queue->pend_obj, PEND_TYPE_QUEUE)) {
        return K_ERR_OBJ_INVALID;
    }
#endif

    TOS_CPU_INT_DISABLE();

    if (!pend_is_nopending(&queue->pend_obj)) {
        pend_wakeup_all(&queue->pend_obj, PEND_STATE_DESTROY);
    }

    pend_object_deinit(&queue->pend_obj);
    tos_msg_queue_flush(&queue->msg_queue);

    TOS_CPU_INT_ENABLE();
    knl_sched();

    return K_ERR_NONE;
}

清空佇列

清空佇列實際上就是將訊息釋放回訊息池中，本質上還是呼叫tos_msg_queue_flush()函式，它是依賴于訊息佇列實作的，

清空佇列函式，在\kernel\core\tos_queue.c 第 41 行

__API__ k_err_t tos_queue_flush(k_queue_t *queue)
{
    TOS_CPU_CPSR_ALLOC();

    TOS_PTR_SANITY_CHECK(queue);

#if TOS_CFG_OBJECT_VERIFY_EN > 0u
    if (!pend_object_verify(&queue->pend_obj, PEND_TYPE_QUEUE)) {
        return K_ERR_OBJ_INVALID;
    }
#endif

    TOS_CPU_INT_DISABLE();
    tos_msg_queue_flush(&queue->msg_queue);
    TOS_CPU_INT_ENABLE();

    return K_ERR_NONE;
}

等待佇列（訊息）

當任務試圖從佇列中的獲取訊息時，用戶可以指定一個等待時間，當且僅當佇列存在訊息的時候，任務才能獲取到訊息，在等待的這段時間中，如果佇列為空，該任務將保持阻塞狀態以等待佇列訊息有效，當其他任務或中斷服務程式往其等待的佇列中寫入了資料，該任務將自動由阻塞態轉為就緒態，當任務等待發生超時，即使佇列中尚無有效訊息，任務也會自動從阻塞態轉為就緒態，
等待佇列的程序也是非常簡單的，先來看看引數吧（其中msg_addr與msg_size引數是用于保存函式回傳的內容，即輸出）：

等待佇列訊息的程序如下：

1. 首先檢測傳入的引數是否正確

2. 嘗試呼叫tos_msg_queue_get()函式獲取訊息，如果佇列存在訊息則會獲取成功（回傳K_ERR_NONE），否則獲取失敗，（關于該函式在下一章講解）

3. 當獲取成功則可以直接退出函式，而當獲取訊息失敗的時候，則可以根據指定的等待時間timeout進行阻塞，如果不等待（timeout =TOS_TIME_NOWAIT），則直接回傳錯誤代碼K_ERR_PEND_NOWAIT，

4. 如果調度器被鎖了knl_is_sched_locked()，則無法進行等待操作，回傳錯誤代碼K_ERR_PEND_SCHED_LOCKED，畢竟需要切換任務，調度器被鎖則無法切換任務，

5. 呼叫pend_task_block()函式將任務阻塞，該函式實際上就是將任務從就緒串列中移除k_rdyq.task_list_head[task_prio]，并且插入到等待串列中object->list，如果等待的時間不是永久等待TOS_TIME_FOREVER，還會將任務插入時間串列中k_tick_list，阻塞時間為timeout，然后進行一次任務調度knl_sched()，

6. 當程式能執行到pend_state2errno()時，則表示任務等待到訊息，又或者發生超時，那么就呼叫pend_state2errno()函式獲取一下任務的等待狀態，看一下是哪種情況導致任務恢復運行，

7. 如果是正常情況（等待獲取到訊息），則將訊息從任務控制塊的k_curr_task->msg_addr讀取出來，并且寫入msg_addr 中用于回傳，同樣的訊息的大小也是會通過msg_size 回傳，

獲取（等待）佇列訊息函式，在\kernel\core\tos_queue.c 第 60 行

__API__ k_err_t tos_queue_pend(k_queue_t *queue, void **msg_addr, size_t *msg_size, k_tick_t timeout)
{
    TOS_CPU_CPSR_ALLOC();
    k_err_t err;

    TOS_PTR_SANITY_CHECK(queue);
    TOS_PTR_SANITY_CHECK(msg_addr);
    TOS_PTR_SANITY_CHECK(msg_size);

#if TOS_CFG_OBJECT_VERIFY_EN > 0u
    if (!pend_object_verify(&queue->pend_obj, PEND_TYPE_QUEUE)) {
        return K_ERR_OBJ_INVALID;
    }
#endif

    TOS_CPU_INT_DISABLE();

    if (tos_msg_queue_get(&queue->msg_queue, msg_addr, msg_size) == K_ERR_NONE) {
        TOS_CPU_INT_ENABLE();
        return K_ERR_NONE;
    }

    if (timeout == TOS_TIME_NOWAIT) {
        *msg_addr = K_NULL;
        *msg_size = 0;
        TOS_CPU_INT_ENABLE();
        return K_ERR_PEND_NOWAIT;
    }

    if (knl_is_sched_locked()) {
        TOS_CPU_INT_ENABLE();
        return K_ERR_PEND_SCHED_LOCKED;
    }

    pend_task_block(k_curr_task, &queue->pend_obj, timeout);

    TOS_CPU_INT_ENABLE();
    knl_sched();

    err = pend_state2errno(k_curr_task->pend_state);

    if (err == K_ERR_NONE) {
        *msg_addr = k_curr_task->msg_addr;
        *msg_size = k_curr_task->msg_size;
        k_curr_task->msg_addr = K_NULL;
        k_curr_task->msg_size = 0;
    }

    return err;
}

將等待訊息的任務添加到對應等待串列函式，在\kernel\core\tos_pend.c檔案的 第 106 行

__KERNEL__ void pend_task_block(k_task_t *task, pend_obj_t *object, k_tick_t timeout)
{
    readyqueue_remove(task);
    pend_list_add(task, object);

    if (timeout != TOS_TIME_FOREVER) {
        tick_list_add(task, timeout);
    }
}

獲取任務等待狀態的函式，在\kernel\core\tos_pend.c檔案的 第 72 行

__KERNEL__ k_err_t pend_state2errno(pend_state_t state)
{
    if (state == PEND_STATE_POST) {
        return K_ERR_NONE;
    } else if (state == PEND_STATE_TIMEOUT) {
        return K_ERR_PEND_TIMEOUT;
    } else if (state == PEND_STATE_DESTROY) {
        return K_ERR_PEND_DESTROY;
    } else if (state == PEND_STATE_OWNER_DIE) {
        return K_ERR_PEND_OWNER_DIE;
    } else {
        return K_ERR_PEND_ABNORMAL;
    }
}

（訊息）寫入佇列

任務或者中斷服務程式都可以給訊息佇列發送訊息，當發送訊息時，TencentOS tiny會從訊息池中取出一個訊息，掛載到佇列的訊息串列末尾（FIFO發送方式），tos_queue_post()是喚醒一個等待佇列訊息任務，tos_queue_post_all()則會喚醒所有等待佇列訊息的任務，無論何種情況，都是呼叫queue_do_post將訊息寫入佇列中，
訊息的寫入佇列程序：

1. 首先檢測傳入的引數是否正確

2. 判斷一下是否有任務在等待訊息，如果有則根據opt引數決定喚醒一個任務或者所有等待任務，否則直接將訊息寫入佇列中，

3. 當沒有任務在等待訊息時，呼叫tos_msg_queue_put()函式將訊息寫入佇列，寫入佇列的方式遵循FIFO原則（TOS_OPT_MSG_PUT_FIFO），寫入成功回傳K_ERR_NONE，而如果訊息池已經沒有訊息了（訊息最大個數由TOS_CFG_MSG_POOL_SIZE宏定義決定），則寫入失敗，回傳K_ERR_QUEUE_FULL錯誤代碼，（關于該函式將在下一章講解）

4. 如果有任務在等待訊息，則呼叫queue_task_msg_recv()函式將訊息內容與大小寫入任務控制塊的msg_addr與msg_size成員變數中，此外還需要喚醒任務，就通過呼叫pend_task_wakeup()函式將對應的等待任務喚醒，核心處理思想就是通過TOS_LIST_FIRST_ENTRY獲取到等待在佇列上的任務，然后喚醒它，

5. 對于喚醒所有等待任務的處理其實也是一樣的，只不過是多了個回圈處理，把等待串列中的所有任務依次喚醒，僅此而已~

寫入佇列訊息函式，在\kernel\core\tos_queue.c 第 159 、164 行

__API__ k_err_t tos_queue_post(k_queue_t *queue, void *msg_addr, size_t msg_size)
{
    TOS_PTR_SANITY_CHECK(queue);
    TOS_PTR_SANITY_CHECK(msg_addr);

    return queue_do_post(queue, msg_addr, msg_size, OPT_POST_ONE);
}

__API__ k_err_t tos_queue_post_all(k_queue_t *queue, void *msg_addr, size_t msg_size)
{
    TOS_PTR_SANITY_CHECK(queue);
    TOS_PTR_SANITY_CHECK(msg_addr);

    return queue_do_post(queue, msg_addr, msg_size, OPT_POST_ALL);
}

寫入佇列訊息函式實際呼叫的函式，通過opt引數進行不一樣的處理，在\kernel\core\tos_queue.c 第 118 行

__STATIC__ k_err_t queue_do_post(k_queue_t *queue, void *msg_addr, size_t msg_size, opt_post_t opt)
{
    TOS_CPU_CPSR_ALLOC();
    k_list_t *curr, *next;

    TOS_PTR_SANITY_CHECK(queue);

#if TOS_CFG_OBJECT_VERIFY_EN > 0u
    if (!pend_object_verify(&queue->pend_obj, PEND_TYPE_QUEUE)) {
        return K_ERR_OBJ_INVALID;
    }
#endif

    TOS_CPU_INT_DISABLE();

    if (pend_is_nopending(&queue->pend_obj)) {
        if (tos_msg_queue_put(&queue->msg_queue, msg_addr, msg_size, TOS_OPT_MSG_PUT_FIFO) != K_ERR_NONE) {
            TOS_CPU_INT_ENABLE();
            return K_ERR_QUEUE_FULL;
        }
        TOS_CPU_INT_ENABLE();
        return K_ERR_NONE;
    }

    if (opt == OPT_POST_ONE) {
        queue_task_msg_recv(TOS_LIST_FIRST_ENTRY(&queue->pend_obj.list, k_task_t, pend_list),
                                msg_addr, msg_size);
    } else { // OPT_QUEUE_POST_ALL
        TOS_LIST_FOR_EACH_SAFE(curr, next, &queue->pend_obj.list) {
            queue_task_msg_recv(TOS_LIST_ENTRY(curr, k_task_t, pend_list),
                                msg_addr, msg_size);
        }
    }

    TOS_CPU_INT_ENABLE();
    knl_sched();

    return K_ERR_NONE;
}

喚醒等待的任務函式，在\kernel\core\tos_pend.c檔案 的第 87 行

喚醒等待任務的思想就是將任務從對應的等待串列移除，然后添加到就緒串列中，

__KERNEL__ void pend_task_wakeup(k_task_t *task, pend_state_t state)
{
    if (task_state_is_pending(task)) {
        // mark why we wakeup
        task->pend_state = state;
        pend_list_remove(task);
    }

    if (task_state_is_sleeping(task)) {
        tick_list_remove(task);
    }

    if (task_state_is_suspended(task)) {
        return;
    }

    readyqueue_add(task);
}

總結

代碼精悍短小，思想清晰，非常建議深入學習~

喜歡就關注我吧！

更多資料歡迎關注“物聯網IoT開發”公眾號！

標籤：嵌入式

上一篇：【TencentOS tiny】深度原始碼分析（2）——調度器

下一篇：【TencentOS tiny】深度原始碼分析（4）——訊息佇列