鸿蒙通用业务模型与IPC

代码位置

设备模型

业务概念

特性概念

抽象API

API寻址

why

线程间通信

消息队列

消息体

消息分发和处理

进程间通信

通信基础

注册和查找API

总结

鸿蒙通用业务模型与IPC

代码位置

目录描述

foundation/distributedschedule/services/samgr_lite鸿蒙通用业务模型实现代码

foundation/distributedschedule/services/samgr_lite/communication实现了广播业务(service)和发布订阅特性(feature)

foundation/distributedschedule/services/samgr_lite/samgr支持跨线程的远过程调用及对应的抽象业务模型

foundation/distributedschedule/services/samgr_lite/samgr/samgr_client跨进程远过程调用客户端进程的实现逻辑

foundation/distributedschedule/services/samgr_lite/samgr/samgr_server跨进程远过程调用服务器端进程的实现逻辑

foundation/distributedschedule/services/samgr_lite/samgr/samgr_endpoint跨进程远过程调用与通信紧密相关的逻辑

设备模型

鸿蒙系统支持2大种类的设备：

1. 只支持线程概念的设备，无进程概念。

2. 支持进程概念的设备，这样的设备中，设备可以运行多个进程，进程内还可以有多个线程。

不管上述哪种设备。都可以在鸿蒙系统中部署各种各样的业务(service)。在保持每个业务简洁的同时，多个业务协同还能完成复杂的功能。

业务概念

业务(service)是一个抽象的概念，代表一组紧密相关的功能或者操作，可以认为是一个软件模块，部署在线程中。多个业务模块可以部署在同一个线程中；重要的业务模块也可以独占一个线程。实现了下述结构(类似于Java的Interface, C++的Abstract class)的软件模块称为一个具体的业务模块。

struct Service {

/**

* @since 1.0

* @version 1.0 */

const char *(*GetName)(Service *service); //获取业务名称

/**

* @since 1.0

* @version 1.0

BOOL (*Initialize)(Service *service, Identity identity); //初始化业务，ID为业务在业务表中的序号

/**

* @since 1.0

* @version 1.0

BOOL (*MessageHandle)(Service *service, Request *request); //在业务线程中处理业务逻辑的回调函数

/**

* @since 1.0

* @version 1.0 */

TaskConfig (*GetTaskConfig)(Service *service); //获取业务的线程配置的函数，用于创建业务线程的参数

};

其中，业务的线程配置(独占or共享等)由GetTaskConfig给出，业务具体实现的功能由MessageHandle给出。不同业务的初始化方法定义不同的Initialize函数。

特性概念

站在使用者的角度，大多数软件功能有多个对外的软件接口，基于接口的概念，鸿蒙抽象出了特性这个概念。一个业务可以有多个接口(特性)。如下定义了接口的抽象表达形式。

struct Feature {

/**

* @since 1.0

* @version 1.0

const char *(*GetName)(Feature *feature); //返回特性的名称

/**

* @since 1.0

* @version 1.0

//初始化特性，特性从属于某个业务(parent)。在业务内，特性具有唯一的编号(identity)

void (*OnInitialize)(Feature *feature, Service *parent, Identity identity);

/**

* @since 1.0

* @version 1.0

//此特性停止服务

void (*OnStop)(Feature *feature, Identity identity);

/**

* @since 1.0

* @version 1.0

//定义本特性具体实现的功能，request为请求方(客户方)发过来的消息

BOOL (*OnMessage)(Feature *feature, Request *request);

};

综上所述，service(业务)和feature(特性)一起定义了鸿蒙的抽象业务模型。其中service可以包含0个或者多个feature。

抽象API

如前所述，业务的对外接口始终要通过API来体现。那么鸿蒙定义了抽象的API（使用Iunknown来描述）。其与service和feature的绑定关系如下。

struct FeatureImpl {

Feature *feature; //每个feature

IUnknown *iUnknown; //绑定一个抽象API

};

struct ServiceImpl {

Service *service; //每个业务向外提供

IUnknown *defaultApi; //默认的抽象API

TaskPool *taskPool;

Vector features; //以及若干feature的 FeatureImpl

int16 serviceId;

uint8 inited;

Operations ops;

};

因此每个service 至少对外暴露一个抽象API(通过defaultApi字段)。也可以暴露更多的抽象API（通过其下面的features列表）。

API寻址

通过指定service的名称就能查询到service,然后再根据feature的名称就能查询到其中的feature。从而获取到对应的抽象API，如果没有指定feature名称，则使用service的默认API。

//查询抽象API

static IUnknown *GetFeatureApi(const char *serviceName, const char *feature)

{

ServiceImpl *serviceImpl = GetService(serviceName); //根据业务名称查询

if (serviceImpl == NULL) {

//本进程没有这个业务，则需要向其他进程查询

return SAMGR_FindServiceApi(serviceName, feature);

}

//根据feature名称在service中继续查询

FeatureImpl *featureImpl = DEFAULT_GetFeature(serviceImpl, feature);

if (featureImpl == NULL && feature == NULL) {

//查询feature失败或者没指定feature,则使用业务的默认API

return serviceImpl->defaultApi;

}

return SAMGR_GetInterface(featureImpl); //成功获取绑定在feature上的API

}

这里出现了跨进程查询抽象API的逻辑，在后续章节再讲，本处读者可以略过。

why

为何设计这样复杂？

答：

1. 通过这种设计。可以将业务实现在某线程中或某进程中。其它软件部件就可以远程调用业务提供的服务。即，通过业务和特性的名称获取到API, 然后再调用此API得到结果。

2. 另一方面，业务实现模块部署在独立的线程中或进程中。(由消息驱动，即接收其它软件部件发来的消息，并向其它软件部件回送消息)。

3. 站在业务模块的角度，业务模块的入口单纯了，只需要守护好消息入口即可。软件更稳定和可靠，也方便维护和管理。

这种做法对多线程程序是利好的，即多线程程序不会操作相同的数据，避免了对互斥锁的需求。

这种以消息驱动的多线程模型的另外一个词汇是无锁编程。

4. 远程调用同时支持2种模型，跨线程的远程调用和跨进程的远程调用。

当将不同的业务隔离在不同的进程中以后，业务互相的干扰降得更低，系统的可靠性进一步升高。(比如进程重启时，进程内的业务都会中断，进程中业务越少，服务中断的概率就越低）。

线程间通信

上述章节已说明跨线程和跨进程的远过程调用。远过程调用的概念是相对于传统的过程调用(C语言中称为函数调用)而言的。

传统过程调用中，调用函数和被调用函数位于同一个线程上下文中。

而跨线程远过程调用。表示调用者在一个线程，被调用者在另外一个线程。调用者和被调用者还共享相同的内存地址空间。

跨进程远程调用。表示调用者在一个进程，被调用者在另外一个进程。调用者和被调用者处于不同的内存地址空间。

本小节描述鸿蒙系统跨线程远程调用的设计和实现。下一节描述鸿蒙系统跨进程远过程的调用的设计和实现。

消息队列

首先涉及的是消息队列，消息队列是线程间通信或进程间通信的标准方式。鸿蒙系统跨线程远过程调用采用的是消息队列方式。使用的是操作系统提供的标准的消息队列。

消息体

在消息队列中，为了实现跨线程的远程调用，传输的消息体如下

#pragma pack(1)

typedef struct Exchange Exchange;

struct Exchange {

Identity id; /**业务或者特性的编号，下面展开描述 */

Request request; //请求消息

Response response; //应答消息

short type; /**< 消息的类型，下面展开描述 */

Handler handler; /** 收到应答后的回调处理，或者其它形式的异步处理，后续展开描述 */

uint32 *sharedRef; /** 请求或者应答消息可以共用时，用来记录共享数目 */

};

#pragma pack()

由于是线程间通信，所以上述结构体中的指针信息在消息传递后，接收线程仍然可见。比如sharedRef字段。

struct Identity {

/** Service ID */

int16 serviceId; //本消息应该由哪个业务处理

/** Feature ID */

int16 featureId; //本消息应该由业务下的哪个特性处理

/** Message queue ID */

//一般在发送消息前填目标线程的消息队列ID

//在发送消息时，修改成发送者的线程的队列ID

//这样，对方回应应答消息的时候，就知道怎么送回来了

MQueueId queueId;

};

对于featureId, 如果指定成一个不存在的feature，则目标线程使用service默认的API来处理，否则使用feature对应的API来处理。

发送者发送的调用请求封装在Request结构中，同理，data指针在跨线程后仍然有效。所以发送线程发送的实际消息和接收线程看到的消息属于同一块内存。即data指针所描述的内存块。如下

struct Request {

/** Message ID */

int16 msgId; //消息ID，目标API内部可以区分处理

/** Data length */

int16 len; //实际消息长度

/** Data content */

void *data; //实际消息内容

/** Message value, which is defined by developers */

uint32 msgValue; //由开发人员自定义的字段，如果不考虑这个字段，就是一个标准的TLV消息结构。

//但有时候消息比较简单，只需要msgId和msgValue就可以了，这个字段主要是方便这种场景

};

回应者的应答消息一般封装在Response结构中，同样，应答消息的实际内容(data所描述的内存)也跨线程可见。如下

struct Response {

/** Data content */

void *data; //应答的内容

/** Data length */

int16 len; //应答消息的长度

};

有时候，当请求发送者不需要服务线程回送应答的时候，Response结构也可以用来传递额外的请求数据，这个只是一种小技巧，阅读代码的时候不要诧异。

消息类型字段详述如下(注释很重要)

enum ExchangeType {

//这个消息用于退出消息处理循环。

//多数时候是线程自己想退出了，往自己的消息队列扔一个这种消息

//处理下一个消息的时候，线程就可以退出了

MSG_EXIT = -1,

//单向普通请求消息，不需要回应，希望对方收到消息后按业务或特性定义的方式处理

MSG_NON = 0,

//请求消息，在MSG_NON的基础上，还希望对方回送应答消息

MSG_CON = 1,

//本消息是一个应答消息

MSG_ACK = 2,

//暂时未看到使用

MSG_SYNC = 3,

//请求消息，不需要对方回应。且希望对方按自己定义的方式处理消息，在消息中指定处理函数

//即handler。实际上就是借用一下对方这个线程躯壳运行一下发送方的函数。

MSG_DIRECT = 4, //单向消息，并在目标线程处理回调函数

};

消息分发和处理

目标线程收到消息后，进行分发和具体处理。如下是处理消息的主循环

//业务所在的线程，本线程内可能有多个service（多service共享线程的情况）

//也可能只有一个service。即service独占线程的情况

static void *TaskEntry(void *argv)

{

ServiceImpl *serviceImpl = NULL;

THREAD_SetThreadLocal(argv); //argv是消息队里的ID，这里把它记录到线程本地变量，方便随时取用

while (TRUE) { //业务线程主要使用消息驱动循环模型

Exchange exchange; //消息的基本单元

//先从消息队列中读取一个消息

uint32 msgRcvRet = SAMGR_MsgRecv((MQueueId)argv, (uint8 *)&exchange, sizeof(Exchange));

if (msgRcvRet != EC_SUCCESS) {

continue; //没有取到，继续取消息

}

if (exchange.type == MSG_EXIT) { //要求退出当前线程

SAMGR_FreeMsg(&exchange); //则先释放消息

break; //然后退出

}

//根据消息中的业务ID获取到业务实例

serviceImpl = CorrectServiceImpl(&exchange, serviceImpl);

BeginWork(serviceImpl); //开始消息处理，记录运维统计

ProcResponse(&exchange); //先处理其它线程发来的应答消息

ProcDirectRequest(&exchange); //再执行由发送方指定的自定义过程

//最后针对业务中的具体功能进行处理，并根据需要回送响应

ProcRequest(&exchange, serviceImpl);

//注意上述3行代码是互斥的。因为消息类型是互斥的，不会都运行到。

//对于每种消息，其中1行函数调用生效，另外2行函数调用为空操作

EndWork(serviceImpl, &exchange); //结束本次消息处理，记录运维统计

SAMGR_FreeMsg(&exchange); //并释放消息

}

QUEUE_Destroy((MQueueId)argv); //线程退出前，先释放消息队列

return NULL;

}

上述过程为消息处理线程的整体逻辑。接下来看消息分发执行逻辑。首先是如何定位到service。

//定位需要处理消息的service

static ServiceImpl *CorrectServiceImpl(Exchange *exchange, ServiceImpl *serviceImpl)

{

if (exchange->type == MSG_ACK) {

// The ack message use the last service.

return serviceImpl; //对于收到应答消息，不关心是哪个业务发出的请求，因为处理函数已经封装在消息中,为Exchange中的handler字段

}

if (serviceImpl == NULL || serviceImpl->serviceId != exchange->id.serviceId) {

//请求消息，还没有业务认领

//或者业务ID发生了变化，则根据业务ID刷新业务实例

serviceImpl = SAMGR_GetServiceByID(exchange->id.serviceId);

}

if (serviceImpl == NULL) {

return NULL;

}

return serviceImpl; //返回业务实例

}

然后寻找需要处理此消息的feature

//请求消息分发处理

void DEFAULT_MessageHandle(ServiceImpl *serviceImpl, const Identity *identity, Request *msg)

{

if (serviceImpl->serviceId != identity->serviceId) {

return; //不是本业务的消息

}

if (identity->featureId < 0) {

//没有指定哪个feature来处理此消息

//则需要此业务的默认API提供服务

if (serviceImpl->service->MessageHandle != NULL) {

serviceImpl->service->MessageHandle(serviceImpl->service, msg);

}

return;

}

if (VECTOR_Size(&serviceImpl->features) <= identity->featureId) {

return; //请求的feature ID非法

}

//找到实际的feature

FeatureImpl *featureImpl = (FeatureImpl *)VECTOR_At(&(serviceImpl->features), identity->featureId);

if (featureImpl == NULL) {

return; //feature ID无对应的feature

}

//由此feature具体来处理这个消息

featureImpl->feature->OnMessage(featureImpl->feature, msg);

}

业务和特性的ID的生成逻辑如下

//注册业务，并记录业务ID(在业务集合中的序号)

//业务集合为一个动态数组--VECTOR

serviceImpl->serviceId = VECTOR_Add(&(samgr->services), serviceImpl);

//同理featureId也是业务中特性集合里feature所在的序号

int16 featureId = VECTOR_Add(&(serviceImpl->features), impl);

本节总结，通过如下数据信息能够在进程内定位鸿蒙业务模型中具体特性API的位置。定位过程是三步：先找到线程，再找到业务，最后找到特性。

struct Identity {

/** Service ID */

int16 serviceId; //实现特性API的业务位置

/** Feature ID */

int16 featureId; //实现特性API的特性位置

/** Message queue ID */

MQueueId queueId; //实现特性API的线程位置

};

跨线程远过程调用的其它逻辑不再赘述。

接下来我们需要讨论在鸿蒙中如何跨进程实现远过程调用，即跨进程如何定位抽象API的位置。

进程间通信

通信基础

消息队列

理论上讲，鸿蒙系统进程间通信也可以学习跨线程通信的方式，使用操作系统标准的消息队列机制，不过标准的消息队列不是足够高效。因为有2次消息拷贝代价。发送者将消息拷贝到消息队列，接收者将消息从消息队列拷贝出来。这个2次拷贝动作已经内置在消息队列实现中。

在跨线程间通信时，由于Request.data和Response.data并没有实际的拷贝动作，只传递Exchange结构，消息通信代价不大，所以使用消息拷贝没有啥问题。

在跨进程间通信时，上述data指针对应的数据必须拷贝，因为指针在跨进程后失效，所以代价增大。鸿蒙系统弃用了操作系统消息队列机制。改为采用共享内存机制，减少一次内存拷贝。只进行一次实际的内存拷贝。

鸿蒙IPC共享内存

首先交代一下几个内存地址的概念。

地址概念

地址概念描述

用户态虚拟地址进程空间看到的地址--与物理地址有映射关系

内核态虚拟地址内核空间看到的地址--与物理地址有映射关系

物理地址实际的物理内存地址

共享内存概念：

进程1的用户态虚拟地址1映射物理地址2，物理地址2又映射进程2用户态虚拟地址3。则称进程1和进程2存在共享内存关系，他们能访问到相同的物理内存。

由于物理内存按内存页划分。地址映射一般以页为单位。所以共享内存的基本单位是页。

由于映射关系的存在。连续的虚拟内存空间可以映射到不连续的物理内存页。

理论上。进程1和进程2可以共享很大的内存。

另外，内核虚拟地址与物理地址也存在映射关系(虽然映射算法与进程虚拟地址不同)。但也可以促成进程和内核共享内存的情况。

即可以达到这个效果(某块物理内存内核可以访问，进程1可以访问，进程2也可以访问)。

最容易想到的办法则是：如果2个进程之间需要通信，则都将自己映射到相同的物理地址页即可。但进程无法直接看到物理内存页，根本不知道哪些物理内存页当前空闲。

但内核能看到所有内存页。

所以：鸿蒙系统为了实现IPC通信，并减少通信的消耗，提升性能。就需要做到参与通信的进程以及内核共享同一块内存。大家读写这块内存实现通信。

如下代码展示了内存地址映射过程

//将内存区region映射到进程pcb

LITE_OS_SEC_TEXT STATIC INT32 DoIpcMmap(LosProcessCB *pcb, LosVmMapRegion *region)

{

UINT32 i;

INT32 ret = 0;

PADDR_T pa;

//为了安全可靠，针对此内存区只设置用户进程的读权限，对此内存的写操作由内核来完成

UINT32 uflags = VM_MAP_REGION_FLAG_PERM_READ | VM_MAP_REGION_FLAG_PERM_USER;

LosVmPage *vmPage = NULL;

//pcb进程用于IPC通信的用户空间虚拟地址

VADDR_T uva = (VADDR_T)(UINTPTR)pcb->ipcInfo.pool.uvaddr;

//pcb进程用于IPC通信的内核空间虚拟地址

VADDR_T kva = (VADDR_T)(UINTPTR)pcb->ipcInfo.pool.kvaddr;

//内存区域的添加删除需要保护起来

(VOID)LOS_MuxAcquire(&pcb->vmSpace->regionMux);

//遍历每一个虚拟内存页

for (i = 0; i < (region->range.size >> PAGE_SHIFT); i++) {

//从内核虚拟地址转换对应的物理地址

pa = LOS_PaddrQuery((VOID *)(UINTPTR)(kva + (i << PAGE_SHIFT)));

if (pa == 0) {

//获取物理地址失败

PRINT_ERR("%s, %d\n", __FUNCTION__, __LINE__);

ret = -EINVAL;

break;

}

//根据物理地址获取本物理地址的内存页描述

vmPage = LOS_VmPageGet(pa);

if (vmPage == NULL) {

PRINT_ERR("%s, %d\n", __FUNCTION__, __LINE__);

ret = -EINVAL;

break;

}

//将用户空间虚拟地址同物理地址映射起来，每次只映射一个页面

STATUS_T err = LOS_ArchMmuMap(&pcb->vmSpace->archMmu, uva + (i << PAGE_SHIFT), pa, 1, uflags);

if (err < 0) {

ret = err;

PRINT_ERR("%s, %d\n", __FUNCTION__, __LINE__);

break;

}

LOS_AtomicInc(&vmPage->refCounts); //此物理内存页的引用次数增加

}

//循环正常结束后，所有用户进程虚拟地址空间都映射到内核虚拟地址空间了。

/* if any failure happened, rollback */

if (i != (region->range.size >> PAGE_SHIFT)) { //for循环中途break的情况

//如果某些内存页的映射过程存在错误，则撤销之前的映射操作

while (i--) {

pa = LOS_PaddrQuery((VOID *)(UINTPTR)(kva + (i << PAGE_SHIFT)));

vmPage = LOS_VmPageGet(pa);

//撤销映射

(VOID)LOS_ArchMmuUnmap(&pcb->vmSpace->archMmu, uva + (i << PAGE_SHIFT), 1);

LOS_PhysPageFree(vmPage); //并释放物理内存页(先减去引用计数)

}

(VOID)LOS_MuxRelease(&pcb->vmSpace->regionMux);

return ret;

}

上述代码有一个细节，虽然这块内存是共享的，但只有内核才有改写权限，这样就防止了各进程对此内存空间的乱写出现，同时也方便各进程从共享内存中读取消息。

关于代码中涉及mmap的知识点请查阅相关资料。

如何做到只拷贝一次的呢？下面展开描述

//内核中IPC消息处理逻辑，含发送和接收

LITE_OS_SEC_TEXT STATIC UINT32 LiteIpcMsgHandle(IpcContent *con)

{

UINT32 ret = LOS_OK;

IpcContent localContent;

IpcContent *content = &localContent;

IpcMsg localMsg;

IpcMsg *msg = &localMsg;

IpcListNode *nodeNeedFree = NULL;

//从用户空间拷贝ipc意图，还不是实际消息

if (copy_from_user((void *)content, (const void *)con, sizeof(IpcContent)) != LOS_OK) {

PRINT_ERR("%s, %d\n", __FUNCTION__, __LINE__);

return -EINVAL;

}

//省略部分代码

if ((content->flag & SEND) == SEND) {

//需要发送消息

// 省略部分代码

//拷贝发送的消息控制块

if (copy_from_user((void *)msg, (const void *)content->outMsg, sizeof(IpcMsg)) != LOS_OK) {

PRINT_ERR("%s, %d\n", __FUNCTION__, __LINE__);

ret = -EINVAL;

goto BUFFER_FREE;

}

content->outMsg = msg; //记录输出的消息控制块

//省略部分代码

ret = LiteIpcWrite(content); //发送消息

//省略部分代码

}

//省略部分代码

if (ret != LOS_OK) {

return ret; //如果发送出错，则发送消息过程结束

}

//然后继续处理消息接收。

//如果成功发送后，再接收消息，说明是同步调用

//如果只有发送没有接收，则是异步发送

//如果只有接收没有发送，则是异步接收

if ((content->flag & RECV) == RECV) {

ret = LiteIpcRead(content); //接收消息，可能阻塞或超时等待

//省略部分代码

//收到的消息指针拷贝到用户空间

UINT32 offset = LOS_OFF_SET_OF(IpcContent, inMsg);

//inMsg字段以及相关的指针都转换成用户空间地址了，通过地址映射

//可以让内存在用户空间和内核空间共享内存块

//接收消息的过程并没有向用户空间拷贝实际数据，只是告诉用户空间，消息的起始地址在哪里

//这样避免了消息的实际拷贝，提升了性能。

//这里只是把inMsg字段的值拷贝到用户空间

ret = copy_to_user((char*)con + offset, (char*)content + offset, sizeof(IpcMsg *));

//省略部分代码

}

return ret;

}

从上述代码来看。消息的接收过程只是将消息的起始地址映射成用户空间后，通知了进程，进程直接取用，没有实际的消息拷贝过程。而消息的发送过程我们再细看。

//发送IPC消息的详细过程

LITE_OS_SEC_TEXT STATIC UINT32 LiteIpcWrite(IpcContent *content)

{

UINT32 ret, intSave;

UINT32 dstTid;

IpcMsg *msg = content->outMsg; //此时消息控制块已经拷贝到内核空间

ret = CheckPara(content, &dstTid); //检查消息，并获取目标线程ID

if (ret != LOS_OK) {

return ret;

}

//计算并申请内核消息缓冲区，这段内存区目标进程和内核共享

UINT32 bufSz = sizeof(IpcListNode) + msg->dataSz + msg->spObjNum * sizeof(UINT32);

IpcListNode *buf = (IpcListNode *)LiteIpcNodeAlloc(OS_TCB_FROM_TID(dstTid)->processID, bufSz);

if (buf == NULL) {

PRINT_ERR("%s, %d\n", __FUNCTION__, __LINE__);

return -ENOMEM;

}

//先将消息控制块拷贝到目标进程

ret = CopyDataFromUser(buf, bufSz, (const IpcMsg *)msg);

if (ret != LOS_OK) {

PRINT_ERR("%s, %d\n", __FUNCTION__, __LINE__);

goto ERROR_COPY;

}

//再将其他数据拷贝到目标进程

ret = HandleSpecialObjects(dstTid, buf, FALSE);

//后续就是放入目标进程对应的自定义消息队列，然后唤醒其收消息

}

最耗时的数据拷贝操作发生在这里，如下

//拷贝实际的消息数据

LITE_OS_SEC_TEXT STATIC UINT32 HandlePtr(UINT32 processID, SpecialObj *obj, BOOL isRollback)

{

//省略部分代码

if (LOS_IsUserAddress((vaddr_t)(UINTPTR)(obj->content.ptr.buff)) == FALSE) {

PRINT_ERR("Bad ptr address\n"); //这个时候buffer应该还是用户空间虚拟地址

return -EINVAL;

}

//在目标进程申请同样尺寸的内核空间内存

//当然这个内存目标进程空间也能看到

buf = LiteIpcNodeAlloc(processID, obj->content.ptr.buffSz);

//然后将数据拷贝到目标进程内核空间，等同于就是拷贝到目标进程空间

ret = copy_from_user(buf, obj->content.ptr.buff, obj->content.ptr.buffSz);

if (ret != LOS_OK) {

LiteIpcNodeFree(processID, buf);

return ret;

}

//将地址切换成目标进程用户空间地址，后续目标进程就可以直接使用这段内存了

//目标进程的read操作不用再次拷贝

obj->content.ptr.buff = (VOID *)GetIpcUserAddr(processID, (INTPTR)buf);

//将此buffer放入待释放队列

//这个设计非常棒，由于进程对共享内存没有写权限，所以，目标进程需要释放的时候，需要委托内核来释放，进程向内核发送一个释放指令即可。这里先将此段内存做一个记录，表明此刻起，所有权交给进程

//进程后续让我释放，我就释放，在这段时间之内，我(内核)不再操作它。

EnableIpcNodeFreeByUser(processID, (VOID *)buf);

return LOS_OK;

}

综上。A进程向B进程通过liteipc发送消息时，实际的消息数据只有1次拷贝(HandlePtr函数中)，从进程A（发送方）非共享内存区拷贝进了共享内存区（接收方），且只有这一次拷贝。进程B直接使用共享内存区的数据，没有从内核再拷贝到用户空间的过程，因为他们是共享的。

进程B使用完后，将新消息内存释放掉。所以，鸿蒙liteipc的进程间通信机制是非常高效的。

通信地址问题

进程间通信还有一个比较基础的问题，即通信地址的问题。

虽然说是进程间通信，消息实际是由线程来接收的，操作系统调度运行的单位是线程。

所以每个消息是发给一个具体的线程的。在一个多线程的进程中。如果消息发给了其中一个线程，我们说消息发给了这个进程。

所以，我们的实际通信地址应该包含一个线程ID。

在鸿蒙系统中，为了支持跨进程远过程调用，提出了EndPoint概念：即通信末端点。

通过EndPoint结构来描述通信节点，每个EndPoint绑定一个通信线程，如下：

struct Endpoint {

const char *name; //通信端点名称

//每个进程一个通信的context,其实就是共享内存的那套机制

IpcContext *context;

Vector routers;

ThreadId boss; //本进程通信主线程，用于接收其它进程发来的消息

uint32 deadId;

int running;

SvcIdentity identity; //EndPoint的编号，内部有handle字段，一般和线程id相同

RegisterEndpoint registerEP; //向系统注册通信端点的方法

TokenBucket bucket;

};

这里重点描述一下函数指针registerEP。为什么会存在多个注册EndPoint的版本。这是因为，需要参与IPC的进程都有1个EndPoint。需要有1个地方把这些EndPoint管理起来。这个信息记录的过程称为注册。怎么管理呢，而EndPoint本身就分布在多个进程，注册过程又依赖于IPC。类似于鸡生蛋，蛋生鸡问题了。

解决方法其实很简单。先生成一个知名EndPoint，其绑定在一个知名的地址上(人为设定)，这个地址全系统（因为人为设定了）都知道。然后其它EndPoint主动和它通信，注册自己的地址。所以这里分为了2个逻辑。知名EndPoint自己的地址由人为设定。其它EndPoint的地址主动向知名EndPoint报告登记。

具体代码如下：

//如果registry为NULL

//则是其它EndPoint向知名EndPoint注册自己的地址

endpoint->registerEP = (registry == NULL) ? RegisterRemoteEndpoint : registry;

//自己向自己注册，其实理解成人为设定EndPoint地址更好一些

//这里指定的函数就会替换上面的参数registry

g_server.samgr = SAMGR_CreateEndpoint("samgr", RegisterSamgrEndpoint);

我们看一下知名EndPoint的地址设定过程

//人为设定知名EndPoint的地址

static int RegisterSamgrEndpoint(const IpcContext* context, SvcIdentity* identity)

{

//先通知内核我是知名EndPoint

int ret = SetSaManager(context, MAX_SA_SIZE);

if (ret != LITEIPC_OK) {

HILOG_FATAL(HILOG_MODULE_SAMGR, "Set sa manager<%d> failed!", ret);

// Set sa manager failed! We need restart to recover

exit(-ret);

}

identity->handle = SAMGR_HANDLE; //这个就是特殊地址，硬编码了

identity->token = SAMGR_TOKEN; //对应的SA Router编号，其作用后面解释

identity->cookie = SAMGR_COOKIE; //暂时没有看到使用

return EC_SUCCESS;

}

//在内核中表明我是知名EndPoint所在的线程

LITE_OS_SEC_TEXT STATIC UINT32 SetCms(UINTPTR maxMsgSize)

{

if (maxMsgSize < sizeof(IpcMsg)) {

return -EINVAL;

}

(VOID)LOS_MuxLock(&g_serviceHandleMapMux, LOS_WAIT_FOREVER);

#if (USE_TASKID_AS_HANDLE == YES)

if (g_cmsTask.status == HANDLE_NOT_USED) {

g_cmsTask.status = HANDLE_REGISTED; //知名EndPoint生效，即管理线程生效

g_cmsTask.taskID = LOS_CurTaskIDGet(); //记录其对应的线程ID

g_cmsTask.maxMsgSize = maxMsgSize;

(VOID)LOS_MuxUnlock(&g_serviceHandleMapMux);

return LOS_OK;

}

#else

//省略部分代码

#endif

(VOID)LOS_MuxUnlock(&g_serviceHandleMapMux);

return -EEXIST;

}

然后再看普通EndPoint的注册过程

//向知名EndPoint注册我这个EndPoint的信息

//identity是我的身份标识，context为通信上下文

static int RegisterRemoteEndpoint(const IpcContext *context, SvcIdentity *identity)

{

IpcIo req;

uint8 data[MAX_DATA_LEN];

IpcIoInit(&req, data, MAX_DATA_LEN, 0);

IpcIoPushUint32(&req, RES_ENDPOINT); //endpoint信息类别

IpcIoPushUint32(&req, OP_POST); //推送信息

IpcIoPushUint32(&req, identity->handle); //我的通信地址(线程号)

uint8 retry = 0;

while (retry < MAX_RETRY_TIMES) { //重试的原因是需要知名EndPoint就绪，才能成功注册

++retry;

IpcIo reply;

void *replyBuf = NULL;

SvcIdentity samgr = {SAMGR_HANDLE, SAMGR_TOKEN, SAMGR_COOKIE}; //知名EndPoint地址

int err = Transact(context, samgr, INVALID_INDEX, &req, &reply, LITEIPC_FLAG_DEFAULT, (uintptr_t *)&replyBuf); //发消息过去

if (err == LITEIPC_OK) {

identity->handle = IpcIoPopUint32(&reply); // 获取知名EndPoint返回的通信ID

if (replyBuf != NULL) {

FreeBuffer(context, replyBuf);

}

if (identity->handle == (uint32)INVALID_INDEX) {

continue; //知名EndPoint还未就绪

}

return EC_SUCCESS;

}

sleep(RETRY_INTERVAL);

}

return EC_FAILURE;

}

在知名EndPoint所在进程内，如何注册和存储各EndPoint的通信ID, 涉及的数据结构是PidHandle位图，这里就不再详述了。

注册和查找API

接下来，就是我们最关心的：如何知道哪个进程提供了我们需要的服务和特性。

现在开始分析跨进程查找API的逻辑，如下：

//非samgr_server进程中调用

IUnknown *__attribute__((weak)) SAMGR_FindServiceApi(const char *service, const char *feature)

{

//如果EndPoint未创建，得先创建EndPoint和注册EndPoint。

InitializeRegistry();

SaName key = {service, feature}; //业务特性二元组

// the proxy already exits.

int index = VECTOR_FindByKey(&g_remoteRegister.clients, &key); //然后查询客户端代理

if (index != INVALID_INDEX) {

return VECTOR_At(&g_remoteRegister.clients, index); //如果客户端代理存在，则代理API即代表远端API

}

//否则向samgr_server查询业务特性的具体位置

SvcIdentity identity = QueryRemoteIdentity(service, feature);

if (identity.handle == INVALID_INDEX) {

return NULL; //查询失败，不存在这个业务

}

MUTEX_Lock(g_remoteRegister.mtx);

//既然系统中已存在这个业务，那么本地再查询一次，

//因为处理过程中已经有一段时间流逝了，可能这段时间客户端代理建立了

index = VECTOR_FindByKey(&g_remoteRegister.clients, &key);

if (index != INVALID_INDEX) {

MUTEX_Unlock(g_remoteRegister.mtx);

return VECTOR_At(&g_remoteRegister.clients, index); //再次查询客户端代理，成功则返回

}

//不存在则创建客户端代理

IUnknown *proxy = SAMGR_CreateIProxy(g_remoteRegister.endpoint->context, service, feature, identity);

VECTOR_Add(&g_remoteRegister.clients, proxy); //并在本地注册

MUTEX_Unlock(g_remoteRegister.mtx);

HILOG_INFO(HILOG_MODULE_SAMGR, "Create remote sa proxy[%p]<%s, %s> id<%u,%u>!",

proxy, service, feature, identity.handle, identity.token);

return proxy; //返回客户端代理接口

}

代码中的客户端代理逻辑可以先绕过，暂时不理解也没有关系，以后章节再描述。继续看ID查询接口

static SvcIdentity QueryRemoteIdentity(const char *service, const char *feature)

{

IpcIo req;

uint8 data[MIN_DATA_LEN];

//构造请求消息，获取service , feature对应的资源

IpcIoInit(&req, data, MIN_DATA_LEN, 0);

IpcIoPushUint32(&req, RES_FEATURE); //查询feature的所在位置

IpcIoPushUint32(&req, OP_GET); //查询请求

IpcIoPushString(&req, service); //存入service字符串

IpcIoPushBool(&req, feature == NULL); //存入是否有feature标记

if (feature != NULL) {

IpcIoPushString(&req, feature); //存入feature字符串

}

IpcIo reply;

void *replyBuf = NULL;

SvcIdentity samgr = {SAMGR_HANDLE, SAMGR_TOKEN, SAMGR_COOKIE}; //目标为samgr_server进程

//发起查询请求并等待响应

int ret = Transact(g_remoteRegister.endpoint->context, samgr, INVALID_INDEX, &req, &reply,

LITEIPC_FLAG_DEFAULT, (uintptr_t *)&replyBuf);

ret = (ret != LITEIPC_OK) ? EC_FAILURE : IpcIoPopInt32(&reply); //查询返回值

SvcIdentity target = {INVALID_INDEX, INVALID_INDEX, INVALID_INDEX};

if (ret == EC_SUCCESS) { //如果成功

SvcIdentity *svc = IpcIoPopSvc(&reply); //解析返回值

if (svc != NULL) {

target = *svc; //记录返回值

}

if (ret == EC_PERMISSION) {

HILOG_INFO(HILOG_MODULE_SAMGR, "Cannot Access<%s, %s> No Permission!", service, feature);

}

if (replyBuf != NULL) {

FreeBuffer(g_remoteRegister.endpoint->context, replyBuf); //释放应答消息

}

return target; //这里就得到了(service, feature)跨进程场景下的通信地址。

}

这个通信地址是到知名EndPoint进程(samgr_server进程)去查询的。所以，所有进程提供的(service+feature)都应该在(samgr_server进程)注册自己的信息。

这样，我们才可能查询得到。samgr_server进程对这些业务的管理存储在SAStore结构的(ListNode *root)集合里面，大家自行查阅代码。

我们进一步来看看这个通信地址的格式

typedef struct {

//消息的接收位置编号，一般是接收消息线程的线程ID

uint32_t handle;

//线程内的子通道，即具体应该处理这个消息的更细化的模块

uint32_t token;

//暂时未看到使用

uint32_t cookie;

} SvcIdentity;

我们将上述通信地址格式与线程间的消息通信的地址格式Identity做一个比较。发现都含有接收消息的线程寻址的信息。但Identity内部含service id和feature id。而本处的SvcIdentity只含有一个token。这个是为什么呢？

答案是：进程间远过程调用是线程间远过程调用的补充形式。在进程间IPC场景中，原有的线程间场景仍然可用(一个进程可以部署多个线程)。

所以。进程间通信的实质是：在原有的线程间通信的基础上，新增一个线程，用于和其它进程通信，从其它进程来的消息，先收到这个新增线程，然后通过线程间通信机制再转发到原来的实际service处理线程中处理。为了实现这个消息转发，鸿蒙系统新定义了router机制。上述地址信息的token字段在这个场景下就是代表router的编号。如下

typedef struct Router {

SaName saName; //service , feature 二元组，字符串形式

Identity identity; //业务特性进程内的寻址信息

IServerProxy *proxy;

PolicyTrans *policy;

uint32 policyNum;

} Router;

通过token字段，就知道用哪个router来处理消息，通过router, 就查询到了identity。

总结

鸿蒙系统提供了基于消息驱动的业务特性(抽象API)实现.实现了高度模块化

跨进程RPC是跨线程RPC的补充形式

守护进程维护知名EndPoint，各业务进程的EndPoint地址，以及各业务进程的业务特性地址。

业务进程创建普通EndPoint,实现各种业务特性，并将地址信息通告给守护进程。

客户端进程去往守护进程查询业务特性的位置

客户端进程通过业务位置信息访问业务进程提供的服务。

————————————————

原文链接：https://blog.csdn.net/weixin_40870382/article/details/111318010

鸿蒙通用业务模型与IPC

相关推荐