开源鸿蒙内核源码分析系列 | 消息队列 | 进程间如何异步传递大数据（转载）

开源鸿蒙内核源码分析系列 | 消息队列 | 进程间如何异步传递大数据（转载）

基本概念

队列又称消息队列，是一种常用于任务间通信的数据结构。队列接收来自任务或中断的不固定长度消息，并根据不同的接口确定传递的消息是否存放在队列空间中。

任务能够从队列里面读取消息，当队列中的消息为空时，挂起读取任务；当队列中有新消息时，挂起的读取任务被唤醒并处理新消息。任务也能够往队列里写入消息，当队列已经写满消息时，挂起写入任务；当队列中有空闲消息节点时，挂起的写入任务被唤醒并写入消息。如果将读队列和写队列的超时时间设置为0，则不会挂起任务，接口会直接返回，这就是非阻塞模式。

消息队列提供了异步处理机制，允许将一个消息放入队列，但不立即处理。同时队列还有缓冲消息的作用。

队列用于任务间通信，可以实现消息的异步处理。同时消息的发送方和接收方不需要彼此联系，两者间是解耦的。

队列特性

  • 消息以先进先出的方式排队，支持异步读写。
  • 读队列和写队列都支持超时机制。
  • 每读取一条消息，就会将该消息节点设置为空闲。
  • 发送消息类型由通信双方约定，可以允许不同长度（不超过队列的消息节点大小）的消息。
  • 一个任务能够从任意一个消息队列接收和发送消息。
  • 多个任务能够从同一个消息队列接收和发送消息。
  • 创建队列时所需的队列空间，默认支持接口内系统自行动态申请内存的方式，同时也支持将用户分配的队列空间作为接口入参传入的方式。

Nu descendant un escalier n° 2 by Marcel Duchamp

消息队列长什么样？


#ifndef LOSCFG_BASE_IPC_QUEUE_LIMIT
#define LOSCFG_BASE_IPC_QUEUE_LIMIT 1024 //队列个数
#endif
LITE_OS_SEC_BSS LosQueueCB *g_allQueue = NULL;//消息队列池
LITE_OS_SEC_BSS STATIC LOS_DL_LIST g_freeQueueList;//空闲队列链表，管分配的，需要队列从这里申请

typedef struct {
    UINT8 *queueHandle; /**< Pointer to a queue handle */  //指向队列句柄的指针
    UINT16 queueState; /**< Queue state */  //队列状态
    UINT16 queueLen; /**< Queue length */  //队列中消息总数的上限值，由创建时确定，不再改变
    UINT16 queueSize; /**< Node size */    //消息节点大小，由创建时确定，不再改变，即定义了每个消息长度的上限。
    UINT32 queueID; /**< queueID */      //队列ID
    UINT16 queueHead; /**< Node head */    //消息头节点位置（数组下标）
    UINT16 queueTail; /**< Node tail */    //消息尾节点位置（数组下标）
    UINT16 readWriteableCnt[OS_QUEUE_N_RW]; /**< Count of readable or writable resources， 0:readable， 1:writable */
                      //队列中可写或可读消息数，0表示可读，1表示可写
    LOS_DL_LIST readWriteList[OS_QUEUE_N_RW]; /**< the linked list to be read or written， 0:readlist， 1:writelist */
                      //挂的都是等待读/写消息的任务链表，0表示读消息的链表，1表示写消息的任务链表
    LOS_DL_LIST memList; /**< Pointer to the memory linked list */  //@note_why 这里尚未搞明白是啥意思 ，是共享内存吗？
} LosQueueCB;//读写队列分离

解读：

  • 和进程，线程，定时器一样，消息队列也由全局统一的消息队列池管理，池有多大？默认是1024。
  • 鸿蒙内核对消息的总个数有限制，queueLen消息总数的上限，在创建队列的时候需指定，不能更改。
  • 对每个消息的长度也有限制， queueSize 规定了消息的大小，也是在创建的时候指定。
  • 为啥要指定总个数和每个的总大小，是因为内核一次性会把队列的总内存(queueLen*queueSize)申请下来，确保不会出现后续使用过程中内存不够的问题出现，但同时也带来了内存的浪费，因为很可能大部分时间队列并没有跑满。
  • 队列的读取由queueHead，queueTail管理，Head表示队列中被占用的消息节点的起始位置。Tail表示被占用的消息节点的结束位置，也是空闲消息节点的起始位置。队列刚创建时，Head和Tail均指向队列起始位置。
  • 写队列时，根据readWriteableCnt[1]判断队列是否可以写入，不能对已满（readWriteableCnt[1]为0）队列进行写操作。写队列支持两种写入方式：向队列尾节点写入，也可以向队列头节点写入。尾节点写入时，根据Tail找到起始空闲消息节点作为数据写入对象，如果Tail已经指向队列尾部则采用回卷方式。头节点写入时，将Head的前一个节点作为数据写入对象，如果Head指向队列起始位置则采用回卷方式。
  • 读队列时，根据readWriteableCnt[0]判断队列是否有消息需要读取，对全部空闲（readWriteableCnt[0]为0）队列进行读操作会引起任务挂起。如果队列可以读取消息，则根据Head找到最先写入队列的消息节点进行读取。如果Head已经指向队列尾部则采用回卷方式。
  • 删除队列时，根据队列ID找到对应队列，把队列状态置为未使用，把队列控制块置为初始状态。如果是通过系统动态申请内存方式创建的队列，还会释放队列所占内存。
  • 留意readWriteList，这又是两个双向链表， 双向链表是内核最重要的结构体，牢牢的寄生在宿主结构体上。readWriteList上挂的是未来读/写消息队列的任务列表。

初始化队列


LITE_OS_SEC_TEXT_INIT UINT32 OsQueueInit(VOID)//消息队列模块初始化
{
    LosQueueCB *queueNode = NULL;
    UINT32 index;
    UINT32 size;

    size = LOSCFG_BASE_IPC_QUEUE_LIMIT * sizeof(LosQueueCB);//支持1024个IPC队列
    /* system resident memory， don't free */
    g_allQueue = (LosQueueCB *)LOS_MemAlloc(m_aucSysMem0， size);//常驻内存
    if (g_allQueue == NULL) {
        return LOS_ERRNO_QUEUE_NO_MEMORY;
    }
    (VOID)memset_s(g_allQueue， size， 0， size);//清0
    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]);//通过写节点挂到空闲队列链表上
    }//这里要注意是用 readWriteList 挂到 g_freeQueueList链上的，所以要通过 GET_QUEUE_LIST 来找到 LosQueueCB

    if (OsQueueDbgInitHook() != LOS_OK) {//调试队列使用的。
        return LOS_ERRNO_QUEUE_NO_MEMORY;
    }
    return LOS_OK;
}

解读：

  • 初始队列模块，对几个全局变量赋值，创建消息队列池，所有池都是常驻内存，关于池后续有专门的文章整理，到目前为止已经解除到了进程池，任务池，定时器池，队列池，==
  • 将LOSCFG_BASE_IPC_QUEUE_LIMIT个队列挂到空闲链表g_freeQueueList上，供后续分配和回收。熟悉内核全局资源管理的对这种方式应该不会再陌生。

创建队列


//创建一个队列，根据用户传入队列长度和消息节点大小来开辟相应的内存空间以供该队列使用，参数queueID带走队列ID
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))) {// maxMsgSize上限 为啥要减去 sizeof(UINT32) ，因为前面存的是队列的大小
        return LOS_ERRNO_QUEUE_SIZE_TOO_BIG;
    }

    if ((len == 0) || (maxMsgSize == 0)) {
        return LOS_ERRNO_QUEUE_PARA_ISZERO;
    }

    msgSize = maxMsgSize + sizeof(UINT32);//总size = 消息体内容长度 + 消息大小(UINT32) 
    /*
     * Memory allocation is time-consuming， to shorten the time of disable interrupt，
     * move the memory allocation to here。
     *///内存分配非常耗时，为了缩短禁用中断的时间，将内存分配移到此处，用的时候分配队列内存
    queue = (UINT8 *)LOS_MemAlloc(m_aucSysMem1， (UINT32)len * msgSize);//从系统内存池中分配，由这里提供读写队列的内存
    if (queue == NULL) {//这里是一次把队列要用到的所有最大内存都申请下来了，能保证不会出现后续使用过程中内存不够的问题出现
        return LOS_ERRNO_QUEUE_CREATE_NO_MEMORY;//调用处有 OsSwtmrInit sys_mbox_new DoMqueueCreate ==
    }

    SCHEDULER_LOCK(intSave);
    if (LOS_ListEmpty(&g_freeQueueList)) {//没有空余的队列ID的处理，注意软时钟定时器是由 g_swtmrCBArray统一管理的，里面有正在使用和可分配空闲的队列
        SCHEDULER_UNLOCK(intSave);//g_freeQueueList是管理可用于分配的队列链表，申请消息队列的ID需要向它要
        OsQueueCheckHook();
        (VOID)LOS_MemFree(m_aucSysMem1， queue);//没有就要释放 queue申请的内存
        return LOS_ERRNO_QUEUE_CB_UNAVAILABLE;
    }

    unusedQueue = LOS_DL_LIST_FIRST(&g_freeQueueList);//找到一个没有被使用的队列
    LOS_ListDelete(unusedQueue);//将自己从g_freeQueueList中摘除， unusedQueue只是个 LOS_DL_LIST 结点。
    queueCB = GET_QUEUE_LIST(unusedQueue);//通过unusedQueue找到整个消息队列(LosQueueCB)
    queueCB->queueLen = len;  //队列中消息的总个数，注意这个一旦创建是不能变的。
    queueCB->queueSize = msgSize;//消息节点的大小，注意这个一旦创建也是不能变的。
    queueCB->queueHandle = queue;  //队列句柄，队列内容存储区。
    queueCB->queueState = OS_QUEUE_INUSED;  //队列状态使用中
    queueCB->readWriteableCnt[OS_QUEUE_READ] = 0;//可读资源计数，OS_QUEUE_READ(0):可读。
    queueCB->readWriteableCnt[OS_QUEUE_WRITE] = len;//可些资源计数 OS_QUEUE_WRITE(1):可写， 默认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);//

    OsQueueDbgUpdateHook(queueCB->queueID， OsCurrTaskGet()->taskEntry);//在创建或删除队列调试信息时更新任务条目
    SCHEDULER_UNLOCK(intSave);

    *queueID = queueCB->queueID;//带走队列ID
    return LOS_OK;
}

解读：

  • 创建和初始化一个LosQueueCB；
  • 动态分配内存来保存消息内容，LOS_MemAlloc(m_aucSysMem1， (UINT32)len * msgSize);
  • msgSize = maxMsgSize + sizeof(UINT32);头四个字节放消息的长度，但消息最大长度不能超过maxMsgSize；
  • readWriteableCnt记录读/写队列的数量，独立计算；
  • readWriteList挂的是等待读取队列的任务链表 将在OsTaskWait(&queueCB->readWriteList[readWrite]， timeout， TRUE);中将任务挂到链表上。

关键函数OsQueueOperate

队列的读写都要经过 OsQueueOperate。


/************************************************
队列操作。是读是写由operateType定
本函数是消息队列最重要的一个函数，可以分析出读取消息过程中
发生的细节，涉及任务的唤醒和阻塞，阻塞链表任务的相互提醒。
************************************************/
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 intSave;

    SCHEDULER_LOCK(intSave);
    queueCB = (LosQueueCB *)GET_QUEUE_HANDLE(queueID);//获取对应的队列控制块
    ret = OsQueueOperateParamCheck(queueCB， queueID， operateType， bufferSize);//参数检查
    if (ret != LOS_OK) {
        goto QUEUE_END;
    }

    if (queueCB->readWriteableCnt[readWrite] == 0) {//根据readWriteableCnt判断队列是否有消息读/写
        if (timeout == LOS_NO_WAIT) {//不等待直接退出
            ret = OS_QUEUE_IS_READ(operateType) ？ LOS_ERRNO_QUEUE_ISEMPTY : LOS_ERRNO_QUEUE_ISFULL;
            goto QUEUE_END;
        }

        if (!OsPreemptableInSched()) {//不支持抢占式调度
            ret = LOS_ERRNO_QUEUE_PEND_IN_LOCK;
            goto QUEUE_END;
        }
    //任务等待，这里很重要啊，将自己从就绪列表摘除，让出了CPU并发起了调度，并挂在readWriteList[readWrite]上，挂的都等待读/写消息的task
        ret = OsTaskWait(&queueCB->readWriteList[readWrite]， timeout， TRUE);//任务被唤醒后会回到这里执行，什么时候会被唤醒？当然是有消息的时候!
        if (ret == LOS_ERRNO_TSK_TIMEOUT) {//唤醒后如果超时了，返回读/写消息失败
            ret = LOS_ERRNO_QUEUE_TIMEOUT;
            goto QUEUE_END;//
        }
    } else {
        queueCB->readWriteableCnt[readWrite]--;//对应队列中计数器--，说明一条消息只能被读/写一次
    }

    OsQueueBufferOperate(queueCB， operateType， bufferAddr， bufferSize);//发起读或写队列操作

    if (!LOS_ListEmpty(&queueCB->readWriteList[!readWrite])) {//如果还有任务在排着队等待读/写入消息(当时不能读/写的原因有可能当时队列满了==)
        resumedTask = OS_TCB_FROM_PENDLIST(LOS_DL_LIST_FIRST(&queueCB->readWriteList[!readWrite]));//取出要读/写消息的任务
        OsTaskWake(resumedTask);//唤醒任务去读/写消息啊
        SCHEDULER_UNLOCK(intSave);
        LOS_MpSchedule(OS_MP_CPU_ALL);//让所有CPU发出调度申请，因为很可能那个要读/写消息的队列是由其他CPU执行
        LOS_Schedule();//申请调度
        return LOS_OK;
    } else {
        queueCB->readWriteableCnt[!readWrite]++;//对应队列读/写中计数器++
    }

QUEUE_END:
    SCHEDULER_UNLOCK(intSave);
    return ret;
}

解读：

  • queueID 指定操作消息队列池中哪个消息队列。
  • operateType 表示本次是是读/写消息。
  • bufferAddr，bufferSize表示如果读操作，用buf接走数据，如果写操作，将buf写入队列。
  • timeout只用于当队列中没有读/写内容时的等待。
  • 当读消息时发现队列中没有可读的消息，此时timeout决定是否将任务挂入等待读队列任务链表。
  • 当写消息时发现队列中没有空间用于写的消息，此时timeout决定是否将任务挂入等待写队列任务链表。
  • if (!LOS_ListEmpty(&queueCB->readWriteList[!readWrite]))最有意思的是这行代码。
  • 在一次读消息完成后会立即唤醒写队列任务链表的任务，因为读完了就有了剩余空间，等待写队列的任务往往是因为没有空间而进入等待状态。
  • 在一次写消息完成后会立即唤醒读队列任务链表的任务，因为写完了队列就有了新消息，等待读队列的任务往往是因为队列中没有消息而进入等待状态。

编程实例

创建一个队列，两个任务。任务1调用写队列接口发送消息，任务2通过读队列接口接收消息。

  • 通过LOS_TaskCreate创建任务1和任务2。
  • 通过LOS_QueueCreate创建一个消息队列。
  • 在任务1 send_Entry中发送消息。
  • 在任务2 recv_Entry中接收消息。
  • 通过LOS_QueueDelete删除队列。

#include "los_task.h"
#include "los_queue.h"
static UINT32 g_queue;
#define BUFFER_LEN 50
VOID send_Entry(VOID)
{
    UINT32 i = 0;
    UINT32 ret = 0;
    CHAR abuf[] = "test is message x";
    UINT32 len = sizeof(abuf);
    while (i < 5) {
        abuf[len -2] = '0' + i;
        i++;

        ret = LOS_QueueWriteCopy(g_queue， abuf， len， 0);
        if(ret != LOS_OK) {
            dprintf("send message failure， error: %x\n"， ret);
        }
        LOS_TaskDelay(5);
    }
}
VOID recv_Entry(VOID)
{
    UINT32 ret = 0;
    CHAR readBuf[BUFFER_LEN] = {0};
    UINT32 readLen = BUFFER_LEN;
    while (1) {
        ret = LOS_QueueReadCopy(g_queue， readBuf， &readLen， 0);
        if(ret != LOS_OK) {
            dprintf("recv message failure， error: %x\n"， ret);
            break;
        }
        dprintf("recv message: %s\n"， readBuf);
        LOS_TaskDelay(5);
    }
    while (LOS_OK != LOS_QueueDelete(g_queue)) {
        LOS_TaskDelay(1);
    }
    dprintf("delete the queue success!\n");
}
UINT32 Example_CreateTask(VOID)
{
    UINT32 ret = 0; 
    UINT32 task1， task2;
    TSK_INIT_PARAM_S initParam;
    initParam.pfnTaskEntry = (TSK_ENTRY_FUNC)send_Entry;
    initParam.usTaskPrio = 9;
    initParam.uwStackSize = LOS_TASK_MIN_STACK_SIZE;
    initParam.pcName = "sendQueue";
#ifdef LOSCFG_KERNEL_SMP
    initParam.usCpuAffiMask = CPUID_TO_AFFI_MASK(ArchCurrCpuid());
#endif
    initParam.uwResved = LOS_TASK_STATUS_DETACHED;
    LOS_TaskLock();
    ret = LOS_TaskCreate(&task1， &initParam);
    if(ret != LOS_OK) {
        dprintf("create task1 failed， error: %x\n"， ret);
        return ret;
    }
    initParam.pcName = "recvQueue";
    initParam.pfnTaskEntry = (TSK_ENTRY_FUNC)recv_Entry;
    ret = LOS_TaskCreate(&task2， &initParam);
    if(ret != LOS_OK) {
        dprintf("create task2 failed， error: %x\n"， ret);
        return ret;
    }
    ret = LOS_QueueCreate("queue"， 5， &g_queue， 0， BUFFER_LEN);
    if(ret != LOS_OK) {
        dprintf("create queue failure， error: %x\n"， ret);
    }
    dprintf("create the queue success!\n");
    LOS_TaskUnlock();
    return ret;
}

结果验证：


create the queue success!
recv message: test is message 0
recv message: test is message 1
recv message: test is message 2
recv message: test is message 3
recv message: test is message 4
recv message failure， error: 200061d
delete the queue success!

总结

  • 消息队列解决任务间大数据的传递。
  • 以一种异步，解耦的方式实现任务通讯。
  • 全局由消息队列池统一管理。
  • 在创建消息队列时申请内存块存储消息内存。
  • 读/写操作统一管理，分开执行，A任务 读/写完消息后会立即唤醒等待写/读的B任务。

百文说内核 | 抓住主脉络

子曰：“诗三百，一言以蔽之，曰‘思无邪’。”——《论语》：为政篇。百文相当于摸出内核的肌肉和器官系统，让人开始丰满有立体感，因是直接从注释源码起步，在开源鸿蒙内核源码加注释过程中，每每有心得处就整理，慢慢形成了以下文章。内容立足源码，常以生活场景打比方尽可能多的将内核知识点置入某种场景，具有画面感，容易理解记忆。说别人能听得懂的话很重要! 百篇博客绝不是百度教条式的在说一堆诘屈聱牙的概念，那没什么意思。更希望让内核变得栩栩如生，倍感亲切.确实有难度，自不量力，但已经出发，回头已是不可能的了。
百万汉字注解内核目的是要看清楚其毛细血管，细胞结构，等于在拿放大镜看内核。内核并不神秘，带着问题去源码中找答案是很容易上瘾的，你会发现很多文章对一些问题的解读是错误的，或者说不深刻难以自圆其说，你会慢慢形成自己新的解读，而新的解读又会碰到新的问题，如此层层递进，滚滚向前，拿着放大镜根本不愿意放手。

与代码有bug需不断debug一样，文章和注解内容会存在不少错漏之处，请多包涵，但会反复修正，持续更新，v**.xx 代表文章序号和修改的次数，精雕细琢，言简意赅，力求打造精品内容。百篇博客系列思维导图结构如下：

根据上图的思维导图，我们未来将要和大家一一分享以上大部分关键技术点的博客文章。

百万汉字注解.精读内核源码

如果大家觉得看文章不过瘾，想直接撸代码的话，可以去下面四大码仓围观同步注释内核源码：

gitee仓：

https://gitee.com/weharmony/kernel_liteos_a_note

github仓 ：

https://github.com/kuangyufei/kernel_liteos_a_note

codechina仓：

https://codechina.csdn.net/kuangyufei/kernel_liteos_a_note

coding仓：

https://weharmony.coding.net/public/harmony/kernel_liteos_a_note/git/files

写在最后

我们最近正带着大家玩嗨OpenHarmony。如果你有用OpenHarmony开发的好玩的东东，或者有对OpenHarmony的深度技术剖析，想通过我们平台让更多的小伙伴知道和分享的，欢迎投稿，让我们一起嗨起来！有点子，有想法，有Demo，立刻联系我们：

合作邮箱：zzliang@atomsource.org