OpenHarmony内核任务间IPC原理

近年来，国内开源实现跨越式发展，并成为企业提升创新能力、生产力、协作和透明度的关键。作为 OpenAtom OpenHarmony（以下简称“OpenHarmony”）开源项目共建单位之一，深开鸿以成为智能物联网操作系统领军者为战略目标，基于 OpenHarmony 聚焦智能物联网操作系统（KaihongOS）的技术研发与持续创新。

身为深开鸿 OS 内核开发师，我们常年深耕于 OpenHarmony 的内核开发，希望通过分享一些工作上的经验，帮助大家掌握开源知识。

OpenHarmony LiteOS-M 内核是面向 IoT 领域构建的轻量级物联网操作系统内核，具有小体积、低功耗、高性能的特点，其代码结构简单，实现了进程、线程、内存等管理机制，提供了常见任务间 IPC、软定时器等公共模块，大幅度降低了嵌入式设备开发的难度。目前 OpenHarmony 的事件提供一种任务间的 IPC，即一个或多个任务可以通过写一个或多个不同的事件来触发内核调度，让另一个等待读取事件的任务进入运行状态，从而实现任务间的同步。

对于嵌入式开发工作人员和技术爱好者来说，深入了解常见任务间 IPC，有助于学习和研发内核。本文将从数据结构和算法解析 OpenHarmony 的事件机制，带大家深入了解内核任务间 IPC 原理。

关键数据结构

在解读事件的源码之前，首先了解下事件的关键的数据结构 PEVENT_CB_S：

 

typedef struct tagEvent {    UINT32 uwEventID;           LOS_DL_LIST stEventList; /**< Event control block linked list */  } EVENT_CB_S, *PEVENT_CB_S;

 

uwEventID：即标记任务的事件类型，每个bit可以标识一个事件，最多支持 31 个事件（第 25bit 保留）。

stEventList：即事件控制块的双向循环链表，理解这个字段是理解事件的关键。在双向循环链表中唯一不变的节点就是头节点，而这里的 stEventList 就是头节点。当有任务等待事件但事件还没发生时，任务会被挂载到等待链表中；当事件发生时，系统唤醒等待事件的任务，此时任务就会被剔出链表。

事件初始化

下面是事件初始化源码：

 

LITE_OS_SEC_TEXT_INIT UINT32 LOS_EventInit(PEVENT_CB_S eventCB){    if (eventCB == NULL) {        return LOS_ERRNO_EVENT_PTR_NULL;    }    eventCB->uwEventID = 0;    LOS_ListInit(&eventCB->stEventList);    OsHookCall(LOS_HOOK_TYPE_EVENT_INIT, eventCB);    return LOS_OK;}

 

PEVENT_CB_S 相当于 EVENT_CB_S *， 因此 eventCB 是指针。

说明事件控制块由任务自己创建，内核事件模块只负责维护。任务定义自己的事件控制块变量，通过 LOS_EventInit 初始化，此时没有事件发生，事件链表为空。

用图来表达就是：

事件写操作

任务可以通过 LOS_EventWrite 来写触发一个或多个事件：

 

LITE_OS_SEC_TEXT UINT32 LOS_EventWrite(PEVENT_CB_S eventCB, UINT32 events){    ...    eventCB->uwEventID |= events;                    ---1    if (!LOS_ListEmpty(&eventCB->stEventList)) {     ---2        for (resumedTask = LOS_DL_LIST_ENTRY((&eventCB->stEventList)->pstNext, LosTaskCB, pendList);             &resumedTask->pendList != (&eventCB->stEventList);) { -------3            nextTask = LOS_DL_LIST_ENTRY(resumedTask->pendList.pstNext, LosTaskCB, pendList);
            if (((resumedTask->eventMode & LOS_WAITMODE_OR) && (resumedTask->eventMask & events) != 0) ||                ((resumedTask->eventMode & LOS_WAITMODE_AND) &&                 ((resumedTask->eventMask & eventCB->uwEventID) == resumedTask->eventMask))) {                exitFlag = 1;
                OsSchedTaskWake(resumedTask);       ---4            }            resumedTask = nextTask;        }
        if (exitFlag == 1) {            LOS_IntRestore(intSave);            LOS_Schedule();                        ---5            return LOS_OK;        }    }    ...}

 

1处，保存事件使用的或运算操作，因此一个或多个任务可以写一个或多个事件，写一次或多次，而且每次为不同的事件，多次写同一个事件相当于只写了一次；

2处，有事件发生了就该检查是否有任务在等待事件，事件链表不为空说明有任务在等待事件；

3处，遍历事件链表，唤醒符合条件的任务。LOS_DL_LIST_ENTRY((&eventCB->stEventList)->pstNext,LosTaskCB,pendList) 前面提到，头节点是空节点，第一次遍历从头节点的下一个节点开始，后续再依次找出 nextTask，直到回到头节点；

4处，针对事件读取模式，找到满足条件的任务并唤醒该任务；

5处，一旦匹配到等待事件的任务，则执行任务调度，被唤醒的任务得到执行。

写事件实际操作如下图：

事件读操作

LiteOS 为用户提供了两个事件的函数：

● LOS_EventPoll()：根据任务传入的事件值、掩码及校验模式，返回满足条件的事件，任务可以主动检查事件是否发生而不必被挂起；

● LOS_EventRead()：读取事件，可以理解为阻塞式读，如果事件没有发生，可以指定等待时间，挂起当前任务。

下面是 LOS_EventPoll() 的实现：

 

LITE_OS_SEC_TEXT UINT32 LOS_EventPoll(UINT32 *eventID, UINT32 eventMask, UINT32 mode){    UINT32 ret = 0;    UINT32 intSave;
    if (eventID == NULL) {        return LOS_ERRNO_EVENT_PTR_NULL;    }    intSave = LOS_IntLock();    if (mode & LOS_WAITMODE_OR) {        if ((*eventID & eventMask) != 0) {      ---1            ret = *eventID & eventMask;        }    } else {        if ((eventMask != 0) && (eventMask == (*eventID & eventMask))) {   ---2            ret = *eventID & eventMask;        }    }    if (ret && (mode & LOS_WAITMODE_CLR)) {   ---3        *eventID = *eventID & ~(ret);    }    LOS_IntRestore(intSave);    return ret;}

 

1处，如果读取模式是LOS_WAITMODE_OR，只要有一个事件发生则读取成功，返回发生的那个事件；

2处，如果读取模式LOS_WAITMODE_AND，全部检查事件发生才算读取成功，并返回全部发生事件；

3处，事件读取成功后事件控制块中的事件标记怎么处理？这里通过LOS_WAITMODE_CLR来决定是否清除事件标记。

可以看出以上实现了两种事件的读取方式：一种是多个事件只要一个发生就算发生，另一种是全部事件发生才算发生。

下面是 LOS_EventRead()：

 

LITE_OS_SEC_TEXT UINT32 LOS_EventRead(PEVENT_CB_S eventCB, UINT32 eventMask, UINT32 mode, UINT32 timeOut){    ...    ret = LOS_EventPoll(&(eventCB->uwEventID), eventMask, mode);           ---1    OsHookCall(LOS_HOOK_TYPE_EVENT_READ, eventCB, eventMask, mode, timeOut);    if (ret == 0) {        if (timeOut == 0) {            LOS_IntRestore(intSave);            return ret;        }
        if (g_losTaskLock) {            LOS_IntRestore(intSave);            return LOS_ERRNO_EVENT_READ_IN_LOCK;        }        runTsk = g_losTask.runTask;        runTsk->eventMask = eventMask;        runTsk->eventMode = mode;        OsSchedTaskWait(&eventCB->stEventList, timeOut);                  ---2        LOS_IntRestore(intSave);        LOS_Schedule();                                                   ---3
        intSave = LOS_IntLock();        if (runTsk->taskStatus & OS_TASK_STATUS_TIMEOUT) {            runTsk->taskStatus &= ~OS_TASK_STATUS_TIMEOUT;            LOS_IntRestore(intSave);            return LOS_ERRNO_EVENT_READ_TIMEOUT;        }
        ret = LOS_EventPoll(&eventCB->uwEventID, eventMask, mode);       ---4    }     ...}

 

1处，主动查询想要的事件是否已经发生；

2处，如果事件没有发生，就把当前任务挂起到等待事件链表中；

3处，如果事件没有发生，当前读事件的任务被挂起，让出 CPU；

4处，事件发生时等待事件的任务被调度再次获得 CPU 恢复执行，读取事件。

事件读写整个过程串起来如下图所示：

事件销毁操作

做事有始有终，事件消费完成剩下的事情是清除事件和等待事件的任务链表。

 

LITE_OS_SEC_TEXT_MINOR UINT32 LOS_EventClear(PEVENT_CB_S eventCB, UINT32 eventMask){    ...    eventCB->uwEventID &= eventMask;    ...}

LITE_OS_SEC_TEXT_INIT UINT32 LOS_EventDestroy(PEVENT_CB_S eventCB){    ...    eventCB->stEventList.pstNext = (LOS_DL_LIST *)NULL;    eventCB->stEventList.pstPrev = (LOS_DL_LIST *)NULL;    ...}

 

在LOS_EventClear 中通过使 eventMask=0 来清空事件，在 LOS_EventDestroy 中清空事件链表指针。

小结

看了上面的描述，相信大家对 OpenHarmony LiteOS-M 内核事件的运作机制有了更加深刻的理解，开发者可以更好地使用事件的 API 来进行任务间的同步操作，也可以进一步尝试修改内核事件通知机制，做出一个更适合自己任务的IPC机制。

OpenHarmony 生态建设离不开每位开发者的参与，希望有更多的开发者分享自己开源项目的经验和成果，共同为 OpenHarmony 生态建设贡献一份力量。

原文标题：OpenHarmony——内核对象事件之源码详解

文章出处：【微信公众号：深开鸿】欢迎添加关注！文章转载请注明出处。　　

      审核编辑：彭静