2. 程式人生 > 其它 >佇列Queue：任務間的訊息讀寫，安排起來~

佇列Queue：任務間的訊息讀寫，安排起來~

阿新 • • 發佈：2021-07-26
摘要：本文通過分析鴻蒙輕核心佇列模組的原始碼，掌握佇列使用上的差異。

本文分享自華為雲社群《鴻蒙輕核心M核原始碼分析系列十三 訊息佇列Queue》，作者：zhushy 。

佇列（Queue）是一種常用於任務間通訊的資料結構。任務能夠從佇列裡面讀取訊息，當佇列中的訊息為空時，掛起讀取任務；當佇列中有新訊息時，掛起的讀取任務被喚醒並處理新訊息。任務也能夠往佇列裡寫入訊息，當佇列已經寫滿訊息時，掛起寫入任務；當佇列中有空閒訊息節點時，掛起的寫入任務被喚醒並寫入訊息。如果將讀佇列和寫佇列的超時時間設定為0，則不會掛起任務，介面會直接返回，這就是非阻塞模式。訊息佇列提供了非同步處理機制，允許將一個訊息放入佇列，但不立即處理。同時佇列還有緩衝訊息的作用。

本文通過分析鴻蒙輕核心佇列模組的原始碼，掌握佇列使用上的差異。本文中所涉及的原始碼，以OpenHarmony LiteOS-M核心為例，均可以在開源站點https://gitee.com/openharmony/kernel_liteos_m獲取。

1、佇列結構體定義和常用巨集定義

1.1 佇列結構體定義

在檔案kernel\include\los_queue.h中定義佇列控制塊結構體為LosQueueCB，結構體原始碼如下。佇列狀態.queueState取值OS_QUEUE_UNUSED、OS_QUEUE_INUSED，其他結構體成員見註釋部分。

typedef struct {
    UINT8 
 
*queue;      /**< 佇列記憶體空間的指標 */
    UINT16 queueState; /**< 佇列的使用狀態 */
    UINT16 queueLen;   /**< 佇列長度，即訊息數量 */
    UINT16 queueSize;  /**< 訊息節點大小 */
    UINT16 queueID;    /**< 佇列編號  */
    UINT16 queueHead;  /**< 訊息頭節點位置 */
    UINT16 queueTail;  /**< 訊息尾節點位置 */
    UINT16 readWriteableCnt[OS_READWRITE_LEN]; 
 
/**< 2維陣列，可讀、可寫的訊息數量, 0:可讀, 1:可寫 */
    LOS_DL_LIST readWriteList[OS_READWRITE_LEN]; /**< 2維雙向連結串列陣列，阻塞讀、寫任務的雙向連結串列, 0:讀連結串列, 1:寫連結串列 */
    LOS_DL_LIST memList; /**< 記憶體節點雙向連結串列 */
} LosQueueCB;

1.2 佇列常用巨集定義

系統支援建立多少佇列是根據開發板情況使用巨集LOSCFG_BASE_IPC_QUEUE_LIMIT定義的，每一個佇列queueID是queueID型別的，取值為[0,LOSCFG_BASE_IPC_QUEUE_LIMIT)，表示佇列池中各個佇列的編號。

⑴處的巨集從佇列池中獲取指定佇列編號QueueID對應的佇列控制塊。⑵處根據雙向連結串列節點readWriteList[OS_QUEUE_WRITE]獲取佇列控制塊記憶體地址。

⑴    #define GET_QUEUE_HANDLE(QueueID) (((LosQueueCB *)g_allQueue) + (QueueID))

⑵    #define GET_QUEUE_LIST(ptr) LOS_DL_LIST_ENTRY(ptr, LosQueueCB, readWriteList[OS_QUEUE_WRITE])

另外，佇列中還提供了比較重要的佇列讀取訊息操作相關的列舉和巨集。列舉QueueReadWrite區分佇列的讀和寫，列舉QueueHeadTail區分佇列的首和尾，列舉QueuePointOrNot區分讀寫訊息時是使用值還是指標。

佇列的操作型別使用3位元位的數字來表示，見巨集OS_QUEUE_OPERATE_TYPE的定義，其中高1位表示讀寫數值還是讀寫指標地址，中1位表示隊首還是隊尾，低1位表示讀取還是寫入。列舉和巨集的定義如下：

typedef enum {
    OS_QUEUE_READ,
    OS_QUEUE_WRITE
} QueueReadWrite;

typedef enum {
    OS_QUEUE_HEAD,
    OS_QUEUE_TAIL
} QueueHeadTail;

typedef enum {
    OS_QUEUE_NOT_POINT,
    OS_QUEUE_POINT
} QueuePointOrNot;

#define OS_QUEUE_OPERATE_TYPE(ReadOrWrite, HeadOrTail, PointOrNot)  \
                (((UINT32)(PointOrNot) << 2) | ((UINT32)(HeadOrTail) << 1) | (ReadOrWrite))
#define OS_QUEUE_READ_WRITE_GET(type) ((type) & (0x01))
#define OS_QUEUE_READ_HEAD     (OS_QUEUE_READ | (OS_QUEUE_HEAD << 1))
#define OS_QUEUE_READ_TAIL     (OS_QUEUE_READ | (OS_QUEUE_TAIL << 1))
#define OS_QUEUE_WRITE_HEAD    (OS_QUEUE_WRITE | (OS_QUEUE_HEAD << 1))
#define OS_QUEUE_WRITE_TAIL    (OS_QUEUE_WRITE | (OS_QUEUE_TAIL << 1))
#define OS_QUEUE_OPERATE_GET(type) ((type) & (0x03))
#define OS_QUEUE_IS_POINT(type)    ((type) & (0x04))
#define OS_QUEUE_IS_READ(type)     (OS_QUEUE_READ_WRITE_GET(type) == OS_QUEUE_READ)
#define OS_QUEUE_IS_WRITE(type)    (OS_QUEUE_READ_WRITE_GET(type) == OS_QUEUE_WRITE)
#define OS_READWRITE_LEN           2

2、佇列初始化

佇列在核心中預設開啟，使用者可以通過巨集LOSCFG_BASE_IPC_QUEUE進行關閉。開啟佇列的情況下，在系統啟動時，在kernel\src\los_init.c中呼叫OsQueueInit()進行佇列模組初始化。下面，我們分析下佇列初始化的程式碼。

⑴為佇列申請記憶體，如果申請失敗，則返回錯誤。⑵初始化雙向迴圈連結串列g_freeQueueList，維護未使用的佇列。⑶迴圈每一個佇列進行初始化，為每一個佇列節點指定索引queueID，並把佇列節點插入未使用佇列雙向連結串列g_freeQueueList。程式碼上可以看出，掛在未使用佇列雙向連結串列上的節點是每個佇列控制塊的寫阻塞任務連結串列節點.readWriteList[OS_QUEUE_WRITE]。

LITE_OS_SEC_TEXT_INIT UINT32 OsQueueInit(VOID)
{
    LosQueueCB *queueNode = NULL;
    UINT16 index;

    if (LOSCFG_BASE_IPC_QUEUE_LIMIT == 0) {
        return LOS_ERRNO_QUEUE_MAXNUM_ZERO;
    }

⑴  g_allQueue = (LosQueueCB *)LOS_MemAlloc(m_aucSysMem0, LOSCFG_BASE_IPC_QUEUE_LIMIT * sizeof(LosQueueCB));
    if (g_allQueue == NULL) {
        return LOS_ERRNO_QUEUE_NO_MEMORY;
    }

    (VOID)memset_s(g_allQueue, LOSCFG_BASE_IPC_QUEUE_LIMIT * sizeof(LosQueueCB),
                   0, LOSCFG_BASE_IPC_QUEUE_LIMIT * sizeof(LosQueueCB));

⑵  LOS_ListInit(&g_freeQueueList);
⑶  for (index = 0; index < LOSCFG_BASE_IPC_QUEUE_LIMIT; index++) {
        queueNode = ((LosQueueCB *)g_allQueue) + index;
        queueNode->queueID = index;
        LOS_ListTailInsert(&g_freeQueueList, &queueNode->readWriteList[OS_QUEUE_WRITE]);
    }

    return LOS_OK;
}

3、佇列常用操作

3.1 佇列建立

建立佇列函式是LOS_QueueCreate()，先看看該函式的引數：queueName是佇列名稱，實際上並沒有使用。len是佇列中訊息的數量，queueID是佇列編號，flags保留未使用。maxMsgSize是佇列中每條訊息的最大大小。

我們分析下建立佇列的程式碼。⑴處對引數進行校驗，佇列編碼不能為空，佇列訊息長度不能太大，佇列訊息數量和佇列訊息大小不能為0。⑵處計算訊息的實際最大大小msgSize，即maxMsgSize + sizeof(UINT32)訊息最大大小再加4個位元組，在訊息的最後4個位元組用來儲存訊息的實際長度。然後呼叫⑶處函式LOS_MemAlloc()為對佇列動態申請記憶體，如果記憶體申請失敗，則返回錯誤碼。

⑷處判斷g_freeQueueList是否為空，如果沒有可以使用的佇列，釋放前文申請的記憶體。⑸處如果g_freeQueueList不為空，則獲取第一個可用的佇列節點，接著從雙向連結串列g_freeQueueList中刪除，然後呼叫巨集GET_QUEUE_LIST獲取LosQueueCB *queueCB，初始化建立的佇列資訊，包含佇列的長度.queueLen、訊息大小.queueSize，佇列記憶體空間.queue，訊息狀態.queueState，可讀的數量.readWriteableCnt[OS_QUEUE_READ]為0，可寫的數量readWriteableCnt[OS_QUEUE_WRITE]為佇列訊息長度len，佇列頭位置.queueHead和尾位置.queueTail為0。

⑹初始化雙向連結串列.readWriteList[OS_QUEUE_READ]，阻塞在這個佇列上的讀訊息任務會掛在這個連結串列上。初始化雙向連結串列.readWriteList[OS_QUEUE_WRITE]，阻塞在這個佇列上的寫訊息任務會掛在這個連結串列上。初始化雙向連結串列.memList。⑺賦值給輸出引數*queueID，後續程式使用這個佇列編號對佇列進行其他操作。

LITE_OS_SEC_TEXT_INIT UINT32 LOS_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;

    (VOID)queueName;
    (VOID)flags;

⑴  if (queueID == NULL) {
        return LOS_ERRNO_QUEUE_CREAT_PTR_NULL;
    }

    if (maxMsgSize > (OS_NULL_SHORT - sizeof(UINT32))) {
        return LOS_ERRNO_QUEUE_SIZE_TOO_BIG;
    }

    if ((len == 0) || (maxMsgSize == 0)) {
        return LOS_ERRNO_QUEUE_PARA_ISZERO;
    }
⑵  msgSize = maxMsgSize + sizeof(UINT32);

    /* Memory allocation is time-consuming, to shorten the time of disable interrupt,
       move the memory allocation to here. */
⑶  queue = (UINT8 *)LOS_MemAlloc(m_aucSysMem0, len * msgSize);
    if (queue == NULL) {
        return LOS_ERRNO_QUEUE_CREATE_NO_MEMORY;
    }

    intSave = LOS_IntLock();
⑷  if (LOS_ListEmpty(&g_freeQueueList)) {
        LOS_IntRestore(intSave);
        (VOID)LOS_MemFree(m_aucSysMem0, queue);
        return LOS_ERRNO_QUEUE_CB_UNAVAILABLE;
    }

⑸  unusedQueue = LOS_DL_LIST_FIRST(&(g_freeQueueList));
    LOS_ListDelete(unusedQueue);
    queueCB = (GET_QUEUE_LIST(unusedQueue));
    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;
}

3.2 佇列刪除

我們可以使用函式LOS_QueueDelete(UINT32 queueID)來刪除佇列，下面通過分析原始碼看看如何刪除佇列的。

⑴處判斷佇列queueID是否超過LOSCFG_BASE_IPC_QUEUE_LIMIT，如果超過則返回錯誤碼。如果佇列編號沒有問題，獲取佇列控制塊LosQueueCB *queueCB。⑵處判斷要刪除的佇列處於未使用狀態，則跳轉到錯誤標籤QUEUE_END進行處理。⑶如果佇列的阻塞讀、阻塞寫任務列表不為空，或記憶體節點連結串列不為空，則不允許刪除，跳轉到錯誤標籤進行處理。⑷處檢驗佇列的可讀、可寫數量是否出錯。

⑸處使用指標UINT8 *queue儲存佇列的記憶體空間，⑹處把.queue置空，把.queueState設定為未使用OS_QUEUE_UNUSED，並把佇列節點插入未使用佇列雙向連結串列g_freeQueueList。接下來會需要呼叫⑺處函式LOS_MemFree()釋放佇列記憶體空間。

LITE_OS_SEC_TEXT_INIT UINT32 LOS_QueueDelete(UINT32 queueID)
{
    LosQueueCB *queueCB = NULL;
    UINT8 *queue = NULL;
    UINT32 intSave;
    UINT32 ret;

⑴  if (queueID >= LOSCFG_BASE_IPC_QUEUE_LIMIT) {
        return LOS_ERRNO_QUEUE_NOT_FOUND;
    }

    intSave = LOS_IntLock();
    queueCB = (LosQueueCB *)GET_QUEUE_HANDLE(queueID);
⑵  if (queueCB->queueState == OS_QUEUE_UNUSED) {
        ret = LOS_ERRNO_QUEUE_NOT_CREATE;
        goto QUEUE_END;
    }

⑶  if (!LOS_ListEmpty(&queueCB->readWriteList[OS_QUEUE_READ])) {
        ret = LOS_ERRNO_QUEUE_IN_TSKUSE;
        goto QUEUE_END;
    }

    if (!LOS_ListEmpty(&queueCB->readWriteList[OS_QUEUE_WRITE])) {
        ret = LOS_ERRNO_QUEUE_IN_TSKUSE;
        goto QUEUE_END;
    }

    if (!LOS_ListEmpty(&queueCB->memList)) {
        ret = LOS_ERRNO_QUEUE_IN_TSKUSE;
        goto QUEUE_END;
    }

⑷  if ((queueCB->readWriteableCnt[OS_QUEUE_WRITE] + queueCB->readWriteableCnt[OS_QUEUE_READ]) !=
        queueCB->queueLen) {
        ret = LOS_ERRNO_QUEUE_IN_TSKWRITE;
        goto QUEUE_END;
    }

⑸  queue = queueCB->queue;
⑹  queueCB->queue = (UINT8 *)NULL;
    queueCB->queueState = OS_QUEUE_UNUSED;
    LOS_ListAdd(&g_freeQueueList, &queueCB->readWriteList[OS_QUEUE_WRITE]);
    LOS_IntRestore(intSave);

    OsHookCall(LOS_HOOK_TYPE_QUEUE_DELETE, queueCB);

⑺  ret = LOS_MemFree(m_aucSysMem0, (VOID *)queue);
    return ret;

QUEUE_END:
    LOS_IntRestore(intSave);
    return ret;
}

下面就來看看佇列的讀寫，有2點需要注意：

  • 隊首、隊尾的讀寫

只支援隊首讀取，不能隊尾讀取，否則就不算隊列了。除了正常的隊尾寫訊息外，還提供插隊機制，支援從隊首寫入。

  • 佇列訊息資料內容

往佇列中寫入的訊息的型別有2種，即支援按地址寫入和按值寫入（帶拷貝）。按哪種型別寫入，就需要配對的按相應的型別去讀取。

佇列讀取介面的類別，歸納如下：

3.3 佇列讀取

我們知道有2個佇列讀取方法，按指標地址讀取的函式LOS_QueueRead()和按訊息數值讀取的函式LOS_QueueReadCopy()。我們先看下函式LOS_QueueRead()，該函式的引數有4個，佇列編號queueID，存放讀取到的訊息的緩衝區地址*bufferAddr，存放讀取到的訊息的緩衝區大小bufferSize，讀佇列訊息的等待超時時間timeOut。程式碼如下，我們分析下程式碼。

⑴處校驗傳入引數，佇列編號不能超出限制，傳入的指標不能為空，緩衝大小不能為0。如果timeout不為零，不能在中斷中讀取佇列。⑵處操作型別表示隊首讀取訊息指標，然後呼叫函式OsQueueOperate()進一步操作佇列。

LITE_OS_SEC_TEXT UINT32 LOS_QueueRead(UINT32 queueID, VOID *bufferAddr, UINT32 bufferSize, UINT32 timeOut)
{
    UINT32 ret;
    UINT32 operateType;

⑴  ret = OsQueueReadParameterCheck(queueID, bufferAddr, &bufferSize, timeOut);
    if (ret != LOS_OK) {
        return ret;
    }

⑵  operateType = OS_QUEUE_OPERATE_TYPE(OS_QUEUE_READ, OS_QUEUE_HEAD, OS_QUEUE_POINT);

    OsHookCall(LOS_HOOK_TYPE_QUEUE_READ, (LosQueueCB *)GET_QUEUE_HANDLE(queueID));

    return OsQueueOperate(queueID, operateType, bufferAddr, &bufferSize, timeOut);
}

我們進一步分析下函式OsQueueOperate()，這是是比較通用的封裝，讀取，寫入都會呼叫這個函式，我們以讀取佇列為例分析這個函式。⑴處獲取佇列的操作型別，為讀取操作。⑵處先呼叫函式OsQueueOperateParamCheck()進行引數校驗，校驗佇列是使用中的佇列，並對讀寫訊息大小進行校驗。⑶處如果可讀數量為0，無法讀取時，如果是零等待則返回錯誤碼。如果當前鎖任務排程，跳出函式執行。否則，執行⑷把當前任務放入佇列的讀取訊息阻塞佇列，然後觸發任務排程，後續的程式碼暫時不再執行。如果可讀的數量不為0，可以繼續讀取時，執行⑹處程式碼把可讀數量減1，然後繼續執行⑺處程式碼讀取佇列。

等讀取佇列阻塞超時，或者佇列可以讀取後，繼續執行⑸處的程式碼。如果是發生超時，佇列還不能讀取，更改任務狀態，跳出函式執行。如果佇列可以讀取了，繼續執行⑺處程式碼讀取佇列。⑻處在成功讀取佇列後，如果有任務阻塞在寫入佇列，則獲取阻塞連結串列中的第一個任務resumedTask，然後呼叫喚醒函式OsSchedTaskWake()把待恢復的任務放入就緒佇列，觸發一次任務排程。如果無阻塞任務，則把可寫入的數量加1。

UINT32 OsQueueOperate(UINT32 queueID, UINT32 operateType, VOID *bufferAddr, UINT32 *bufferSize, UINT32 timeOut)
{
    LosQueueCB *queueCB = NULL;
    LosTaskCB *resumedTask = NULL;
    UINT32 ret;
⑴  UINT32 readWrite = OS_QUEUE_READ_WRITE_GET(operateType);
    UINT32 readWriteTmp = !readWrite;

    UINT32 intSave = LOS_IntLock();

    queueCB = (LosQueueCB *)GET_QUEUE_HANDLE(queueID);
⑵  ret = OsQueueOperateParamCheck(queueCB, operateType, bufferSize);
    if (ret != LOS_OK) {
        goto QUEUE_END;
    }

⑶  if (queueCB->readWriteableCnt[readWrite] == 0) {
        if (timeOut == LOS_NO_WAIT) {
            ret = OS_QUEUE_IS_READ(operateType) ? LOS_ERRNO_QUEUE_ISEMPTY : LOS_ERRNO_QUEUE_ISFULL;
            goto QUEUE_END;
        }

        if (g_losTaskLock) {
            ret = LOS_ERRNO_QUEUE_PEND_IN_LOCK;
            goto QUEUE_END;
        }

        LosTaskCB *runTsk = (LosTaskCB *)g_losTask.runTask;
⑷      OsSchedTaskWait(&queueCB->readWriteList[readWrite], timeOut);
        LOS_IntRestore(intSave);
        LOS_Schedule();

        intSave = LOS_IntLock();
⑸      if (runTsk->taskStatus & OS_TASK_STATUS_TIMEOUT) {
            runTsk->taskStatus &= ~OS_TASK_STATUS_TIMEOUT;
            ret = LOS_ERRNO_QUEUE_TIMEOUT;
            goto QUEUE_END;
        }
    } else {
⑹       queueCB->readWriteableCnt[readWrite]--;
    }

⑺   OsQueueBufferOperate(queueCB, operateType, bufferAddr, bufferSize);
⑻  if (!LOS_ListEmpty(&queueCB->readWriteList[readWriteTmp])) {
        resumedTask = OS_TCB_FROM_PENDLIST(LOS_DL_LIST_FIRST(&queueCB->readWriteList[readWriteTmp]));
        OsSchedTaskWake(resumedTask);
        LOS_IntRestore(intSave);
        LOS_Schedule();
        return LOS_OK;
    } else {
⑼      queueCB->readWriteableCnt[readWriteTmp]++;
    }

QUEUE_END:
    LOS_IntRestore(intSave);
    return ret;
}

我們再繼續看下函式OsQueueBufferOperate()是具體如何讀取佇列的。⑴處switch-case語句根據操作型別獲取操作位置。對於⑵頭部讀取的情況，先獲取讀取位置queuePosition。然後，如果當前頭節點位置.queueHead加1等於佇列訊息長度，頭節點位置.queueHead設定為0，否則加1。對於⑶頭部寫入的情況，如果當前頭節點位置.queueHead等於0，頭節點位置.queueHead設定為佇列訊息長度減1即queueCB->queueLen - 1，否則頭節點位置.queueHead減1即可。然後，獲取要寫入的位置queuePosition。對於⑷尾部寫入的情況，先獲取寫入位置queuePosition。然後，如果當前尾節點位置.queueTail加1等於佇列訊息長度，尾節點位置.queueTail設定為0，否則加1。

⑸處基於獲取的佇列讀取位置獲取佇列訊息節點queueNode。⑹處判斷操作型別如果是按指標讀寫訊息，直接讀取訊息節點的資料寫入指標對應的緩衝區*(UINT32 *)bufferAddr，或直接把指標對應的緩衝區*(UINT32 *)bufferAddr資料寫入訊息節點即可。我們接著看如何按數資料讀寫訊息，⑺處程式碼用於讀取資料訊息。每個訊息節點的後4個位元組儲存的是訊息的長度，首先獲取訊息的長度msgDataSize，然後把訊息內容讀取到bufferAddr。再看看⑻處如何寫入佇列訊息，首先把訊息內容寫入到queueNode，然後再把訊息長度的內容寫入到queueNode + queueCB->queueSize - sizeof(UINT32)，就是每個訊息節點的後4位元組。

static INLINE VOID OsQueueBufferOperate(LosQueueCB *queueCB, UINT32 operateType,
                                                                VOID *bufferAddr, UINT32 *bufferSize)
{
    UINT8 *queueNode = NULL;
    UINT32 msgDataSize;
    UINT16 queuePosion;
    errno_t rc;

    /* get the queue position */
⑴  switch (OS_QUEUE_OPERATE_GET(operateType)) {
        case OS_QUEUE_READ_HEAD:
⑵          queuePosion = 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);
            queuePosion = queueCB->queueHead;
            break;

        case OS_QUEUE_WRITE_TAIL:
⑷          queuePosion = queueCB->queueTail;
            ((queueCB->queueTail + 1) == queueCB->queueLen) ? (queueCB->queueTail = 0) : (queueCB->queueTail++);
            break;

        default:
            PRINT_ERR("invalid queue operate type!\n");
            return;
    }

⑸  queueNode = &(queueCB->queue[(queuePosion * (queueCB->queueSize))]);

⑹  if (OS_QUEUE_IS_POINT(operateType)) {
      if (OS_QUEUE_IS_READ(operateType)) {
            *(UINT32 *)bufferAddr = *(UINT32 *)(VOID *)queueNode;
        } else {
            *(UINT32 *)(VOID *)queueNode = *(UINT32 *)bufferAddr;  // change to pp when calling OsQueueOperate
        }
    } else {
⑺      if (OS_QUEUE_IS_READ(operateType)) {
            msgDataSize = *((UINT32 *)(UINTPTR)((queueNode + queueCB->queueSize) - sizeof(UINT32)));
            rc = memcpy_s((VOID *)bufferAddr, *bufferSize, (VOID *)queueNode, msgDataSize);
            if (rc != EOK) {
                PRINT_ERR("%s[%d] memcpy failed, error type = %u\n", __FUNCTION__, __LINE__, rc);
                return;
            }

            *bufferSize = msgDataSize;
        } else {
⑻          *((UINT32 *)(UINTPTR)((queueNode + queueCB->queueSize) - sizeof(UINT32))) = *bufferSize;
            rc = memcpy_s((VOID *)queueNode, queueCB->queueSize, (VOID *)bufferAddr, *bufferSize);
            if (rc != EOK) {
                PRINT_ERR("%s[%d] memcpy failed, error type = %u\n", __FUNCTION__, __LINE__, rc);
                return;
            }
        }
    }
}

3.4 佇列寫入

我們知道，有4個佇列寫入方法，2個隊尾寫入，2個隊首寫入，分別包含按指標地址寫入訊息和按數值寫入訊息。LOS_QueueWrite()會呼叫LOS_QueueWriteCopy()，LOS_QueueWriteHead()會呼叫LOS_QueueWriteHeadCopy()，然後指定不同的操作型別後，會進一步呼叫前文已經分析過的函式OsQueueOperate()。

小結

本文帶領大家一起剖析了鴻蒙輕核心的佇列模組的原始碼，包含佇列的結構體、佇列池初始化、佇列建立刪除、讀寫訊息等。感謝閱讀，如有任何問題、建議，都可以留言給我們：https://gitee.com/openharmony/kernel_liteos_m/issues。為了更容易找到鴻蒙輕核心程式碼倉，建議訪問https://gitee.com/openharmony/kernel_liteos_m，關注Watch、點贊Star、並Fork到自己賬戶下，謝謝。

點選關注，第一時間瞭解華為雲新鮮技術~

相關推薦

佇列Queue：任務間的訊息讀寫，安排起來~

摘要：本文通過分析鴻蒙輕核心佇列模組的原始碼，掌握佇列使用上的差異。

原始碼分析：升級版的讀寫鎖 StampedLock

簡介 StampedLock 是JDK1.8 開始提供的一種鎖， 是對之前介紹的讀寫鎖 ReentrantReadWriteLock 的功能增強。StampedLock 有三種模式：Writing（讀）、Reading（寫）、Optimistic Reading（樂觀度），StampedLock 的功

微信 8.0.9 安卓內測版開發者更新內容：小程式 NFC 讀寫/video 底層優化等

7 月 21 日訊息 近期，微信釋出了安卓版 8.0.9 內測版更新，此次版本更新包括支援設定個性來電鈴聲和訊息提示音、可以在訊息免打擾的群裡關注特定群成員、可以在電腦、平板等裝置同時登入同一個微訊號。

Python3+PCAN-USB基於PCAN-Basic二次開發實現CAN匯流排訊息讀寫

一、環境搭建 1.概述 本文主要是通過Python3與CAN匯流排工具PCAN-USB基於官方PCAN-Basic庫實現CAN匯流排訊息讀寫功能。

面試突擊45：為什麼要用讀寫鎖？它有什麼優點？

讀寫鎖（Readers-Writer Lock）顧名思義是一把鎖分為兩部分：讀鎖和寫鎖，其中讀鎖允許多個執行緒同時獲得，因為讀操作本身是執行緒安全的，而寫鎖則是互斥鎖，不允許多個執行緒同時獲得寫鎖，並且寫操作和讀操作也是

宜搭5月更新：跨應用資料讀寫能力升級，AI元件內測開放

​簡介：表單、許可權管理、AI元件等功能上新啦～ 本次，我們帶來了表單、許可權管理、資料管理、平臺管理許可權、元件等功能的升級。

【Java學習筆記（一百一十六）】之 二進位制資料讀寫， 隨機訪問檔案， ZIP文件

技術標籤：Java學習筆記# JavaEEjava 本文章由公號【開發小鴿】釋出！歡迎關注！！！

江蘇市場監察局調查：直播間網購圍巾，號稱含 96% 羊毛，實際一根羊毛都沒有

3月28日訊息據央視新聞，近日，江蘇省市場監管局抽查超 10 家電商直播平臺發現，有的平臺上售賣的品牌以次充好、缺斤短兩，三無產品層出不窮。

騰訊迴應互聯互通：平臺間會更加開放，致力於保護使用者和中小企業

11 月 10 日訊息，騰訊控股今日釋出了第三季度財報，營收 1423.68 億元人民幣，市場預估 1454.1 億元人民幣；同比增長 13%；第三季度淨利潤 395 億元人民幣，市場預估 326 億元人民幣。騰訊隨後還舉行 2021 年三季度

R資料分析：變數間的非線性關係，多項式，樣條迴歸和可加模型

之前的文章中都是給大家寫的變數間線性關係的做法，包括迴歸和廣義線性迴歸，變數間的非線性關係其實是很常見的，今天給大家寫寫如何擬合論文中常見的非線性關係。包括多項式迴歸Polynomial regression和樣條迴歸Sp

今天除錯W25Q16 SPI讀寫，發現一隻讀寫不成功的問題

今天除錯W25Q16 SPI讀寫，發現一隻讀寫不成功的問題 W25Q16在我之前的一個板子上已經用過，因為MCU與W25Q16連線的電路相同，只是IO引腳序號變動，於是就將之前的驅動程式，直接copy過來用，只是重新更改了IO引腳對應

im即時通訊開發：群聊訊息如此複雜，如何保證不丟不重

群聊已經成為主流IM軟體的基本功能，不管是QQ群、還是微信群，一個群友在群內發了一條訊息，那麼對於IM伺服器來說需要保證：

AOP實現mysql的主從資料庫：讀寫分離

1.問題 首先，為什麼會碰到這樣的問題？ 昨天寫的一個業務上線了，但是在dev環境和test環境都能跑，但是到了線上環境發生資料不能插入的問題。

JUC：讀寫鎖ReentrantReadWriteLock

前言 之前提到的ReentrantLock是排他鎖，在同一時刻只允許一個執行緒進行訪問，而讀寫鎖在同一時刻可以允許多個讀執行緒訪問，但是在寫執行緒訪問時，所有的讀執行緒和其他寫執行緒均被阻塞。讀寫鎖維護了一對鎖，一

Python高階程式設計之訊息佇列(Queue)與程序池(Pool)例項詳解

本文例項講述了Python高階程式設計之訊息佇列(Queue)與程序池(Pool)。分享給大家供大家參考，具體如下：

Python 程序操作之程序間通過佇列共享資料，佇列Queue簡單示例

本文例項講述了Python 程序操作之程序間通過佇列共享資料，佇列Queue。分享給大家供大家參考，具體如下：

js 事件迴圈訊息佇列和微任務巨集任務

事件迴圈與訊息佇列 因為js是單執行緒指令碼語言，一般情況下程式碼是同步執行。也就是說js執行程式碼是一行一行向下執行的，前面沒有執行完成是不會執行後面的程式碼的。

訊息佇列面試連環問：如何保證訊息不丟失？處理重複訊息？訊息有序性？訊息堆積處理？

大家好，我是 yes。 最近我一直紮在訊息佇列實現細節之中無法自拔，已經寫了 3 篇Kafka原始碼分析，還剩很多沒肝完。之前還存著RocketMQ原始碼分析還沒整理。今兒暫時先跳出來盤一盤大方向上的訊息佇列有哪些核心注意

第十九節：SQLServer通過釋出訂閱實現主從同步(讀寫分離)詳解

一.前言 1. 背景 　　大部分場景中，DB操作80%是讀，20%是寫，對於時效性要求不高的資料，為了減少磁碟讀和寫的競爭，引入讀寫分離的概念，即在資料庫上進行主從配置，一個主，多個從，實現主從同步，從而業務上實現

全棧測試 一 | py3基礎 三 ：file物件測試資料的讀寫與操作

檔案的開啟方式 開啟方式 詳細釋義 r 以只讀方式開啟檔案。檔案的指標會放在檔案的開頭。這是預設模式。