IPC进程间通信

1万 · 2022-9-19 16:28

IPC进程间通信
IPC在QNX Neutrino (微内核) 从一个嵌入式实时系统向一个全面的POSIX系统转变起着至关重要的作用
在QNX中，消息传递是IPC的主要形式，也提供了其他的形式，除非有特殊的说明，否则这些形式也都是基于本地消息传递而实现的。QNX Neutrino提供以下形式的IPC：

1万 · 2022-9-19 16:32

Channels and connections

IPC中消息传递是基于CS架构实现的,最简单的,制定对方进程号,要发送的一方,将消息加一个头,告诉内核把该消息发送给 pid12345就行了,其实QNX4的时候时这么做的. 但是

如果服务器有两个线程,分别进行不同的服务, 你或者可以把消息发送给 pid 12345 tid3 就可以, 但如果某一服务,由一组线程进行,就没办法了

为此, QNX6就抽象出了Channel这个概念, 一个Channel,就是一个服务的入口, 至于这个channel到底有多少线程为其服务, 由server负责 ChannelId = ChannelCreate(flags)

一个server可以有多个channels, 而客户端再向频道发送消息前, 需要先建立连接 Connection, 然后再将消息在connection上发送出去, 如果需要, 一个频道可以建立多个连接. ConnectionId = ConnectAttach(Node, Pid, Chid, Index, Flag)

在QNX Neutrino中，消息传递是面向通道（channel）和连接（connection）的，而不是直接从线程到线程的。接收消息的线程需要创建一个channel，发送消息的线程需要与该channel建立connection.
服务器使用MsgReceive()接收消息时需要使用channels，客户端则需要创建connections，以连接到服务器的通道上，连接建立好之后，客户端便可通过MsgSend()来发送消息了。如果进程中有很多线程都连接到一个通道上，为了提高效率，这些所有的连接都会映射到同一个内核对象中。在进程中，channels和connecttions会用一个小的整型标识符来标记。客户端connections会直接映射到文件描述符，在架构上这是一个关键点，可以消除另一层转换，不需要根据文件描述符来确定往哪里发消息，而是直接将消息发往文件描述符即可

与channel有关联的列表：

Receive，等待消息的LIFO线程队列；
Send，已发送消息但还未被接收的优先级FIFO线程队列；
Reply，已发送消息，并且已经被收到，但尚未回复的无序线程列表；
不管在上述哪个列表中，线程都是阻塞状态，多个线程和多个客户端可能等待在同一个channel上

1万 · 2022-9-19 16:37

Synchronous message passing

比较传统的IPC方式是基于主从式架构(CS), 并且是双向通信

发送(Send)，接收(Receive)和应答(Reply) SRR过程

MsgReceive中第四个参数的类型为 struct _msg_info info 用于处理可变的信息长度
MsgRead() 与 MsgWrite(rcvid, buffer, len, offset) 用于客户端与服务端分块处理较大的信息

1万 · 2022-9-19 16:42

客户端

当客户端建立connection到channel后,

1.client调用MsgSend(), 其状态将变为阻塞blocked状态:
如果server还没调用MsgReceive(), client线程状态则为SEND blocked
一旦server调用了MsgReceive(), client线程状态变为REPLY blocked, 同时将SendBuf里的东西复制到ReceiveBuf中
当server线程执行MsgReply(), client线程状态就变成了READY

2.如果client线程调用MsgSend()后，而server线程正阻塞在MsgReceive()上，则client线程状态直接跳过SEND blocked，直接变成REPLY blocked；

3.当server线程失败、退出、或者消失了，client线程状态变成READY，此时MsgSend()会返回一个错误值。

1万 · 2022-9-19 16:44

客户端
客户端在频道上进行接收,

server线程调用MsgReceive()时，当没有线程给它发送消息，它的状态为RECEIVE blocked，当有线程发送时变为READY；server线程调用MsgReceive()时，当已经有其他线程给它发送过消息，MsgReceive()会立马返回，而不会阻塞；server线程调用MsgReply()时，不会阻塞；应答时,可以调用MsgError()告诉发送方有错误发送,MsgError(rcvid, EINVAL), MsgReply(rcvid, EINVAL, 0, 0), MsgSend()返回-1，而errno被设为EINVAL

1万 · 2022-9-19 16:49

Message copying
QNX的消息服务，是直接将消息从一个线程的地址空间拷贝到另一个线程地址空间，不需要中间缓冲，因此消息传递的性能接近底层硬件的内存带宽。消息内容对内核来说没有特殊的意义，只对消息的发送和接收者才有意义.
当然，除了用线性的缓冲区进行消息传递以外，QNX也提供了定义良好的消息类型iov_t来"汇集"数据，以便能扩充或替代系统提供的服务。
消息在拷贝的时候，支持分块传输，也就是不要求连续的缓冲区，发送和接收线程可以指定IOV向量表，在这个表中去指定消息在内存中的位置。

分块传输也用在文件系统中，比如读数据的时候，将文件系统缓存中的数据分块读到用户提供的空间内，如下图：

对于简单的单块消息传递，就不需要通过IOV(input/output vector)的形式了，直接指向缓冲区即可。对于发送和接收的接口，多块发送和单块发送如下：

举个例子:
“header” 与 “databuf” 是不连续的两块数据,
虽然在客户端的Header同databuf是两块不相邻的内存，但传递到服务器端的ReceiveBuffer里，就是连续的了。也就是说在服务器端，要想得到原来databuf里的数据，只需要(ReceiveBuffer + sizeof(header))就可以了。（要注意数据结构对齐）

复制

// 户端:                "header" 与 "databuf" 是不连续的两块数据

SETIOV(&iov[0], &header, sizeof(header));

SETIOV(&iov[1], databuf, datalen);

MsgSendvs(ConnectionId, iov, 2, Replybf, ReplyLen);

// 服务器端:        "header"与"databuf"被连续地存在ReceiveBuffer里

ReceiveId = MsgReceive(ChannelId, ReceiveBuffer, ReceiveBufLength, &MsgInfo); 

header = (struct header *)ReceiveBuffer;

databuf = (char *)((char *)header + sizeof(*header));

1万 · 2022-9-19 16:55

Pulses
脉冲其实更像一个短消息，也是在“连接Connection”上发送的。脉冲最大的特点是它是异步的。发送方不必要等接收方应答，直接可以继续执行。
但是，这种异步性也给脉冲带来了限制。脉冲能携带的数据量有限，只有一个**8位的"code"域 (1byte)用来区分不同的脉冲，和一个32位的“value"域 (4字节)**来携带数据。脉冲最主要的用途就是用来进行“通知”(Notification)。不仅是用户程序，内核也会生成发送特殊的“系统脉冲”到用户程序，以通知某一特殊情况的发生。

MsgSendPulse() 只在一个进程中的通知，用与同一个进程中一个线程要通知另一个线程的情形, 其中 code 8bits; value 32bits
MsgDeliverEvent() 在跨进程的时候的通知
MsgReceivePulse() 用于频道上只有pulse的接收
MsgReceive() 用于频道上既接收message又接收pulse

脉冲的接收比较简单，如果你知道频道上不会有别的消息，只有脉冲的话，可以用MsgReceivePulse()来只接收脉冲；
如果频道既可以接收消息，也可以接收脉冲时，就直接用MsgReceive()，只要确保接收缓冲(ReveiveBuf)至少可以容下一个脉冲（sizeof struct _pulse)就可以了。
在后一种情况下，如果MsgReceive()返回的rcvid是0，就代表接收到了一个脉冲，反之，则收到了一个消息。所以，一个既接收脉冲，又接收消息的服务器，可以是这样的
struct _pulse 定义

复制

union {

    struct _pulse pulse;

    msg_header   header;

} msgs;

if ((rcvid = MsgReceive(chid, &msgs, sizeof(msgs), &info)) == -1) {

    perror("MsgReceive");

    continue;

}

if (rcvid == 0) {

// 此时为pulse信号

    process_pulse(&msgs, &info);

} else {

    process_message(&msgs, &info);

}

1万 · 2022-9-19 16:56

脉冲的发送，最直接的就是MsgSendPulse()。不过，这个函数通常只在一个进程中，用在一个线程要通知另一个线程的情形。
在跨进程的时候，通常不会用到这个函数，而是用到下面将要提到的 MsgDeliverEvent()。
与消息传递相比，消息传递永远是在进程间进行的。也就是说，不会有一个进程向内核发送数据的情形, 而脉冲就不一样，除了用户进程间可以发脉冲以外，内核也会向用户进程发送“系统脉冲”来通知某一事件的发生。

MsgDeliverEvent()

在现实情况中，客户端与服务器端并不是很容易区分开来的。有的服务器端为了处理客户端的请求，本身就需要向别的服务器发送消息；有的客户端需要从不同的服务器那里得到服务，而不能阻塞在某一特定的服务器上；还有的时候，两个进程间的数据是互相流动的

也许有人认为，两个进程互为通讯就可以了。每个进程都建立自己的频道，然后都与对方的频道建一个连接就好了；这样，需要的时候，就可以直接通过连接向对方发送消息了。就好象管道(pipe)或是socketpair一样。请注意，这种设计在QNX的消息传递中是应该避免的。因为很容易就造成死锁。一个常见的情形是这样的:
进程A：MsgSend() 到进程B
进程B：MsgReceive()接收到消息
进程B：处理消息，然后MsgSend()给进程A
因为进程A正在阻塞状态中，无法接收并处理B的请求；所以A会在STATE_REPLY里，而B则会因MsgSend()而进入STATE_SEND，两个进程就互为死锁住了。
当然，如果A和B都使用多线程，专门用一个线程来MsgReceive()，这个情形或许可以避免；但你要保证 MsgReceive()的线程不会去MsgSend()，否则一样会死锁。在程序简单的时候或许你还有控制，如果程序变得复杂，又或者你写的只是一个程序库，别人怎么来用你完全没有控制
在QNX中，正确的方法是这样的。

客户端：准备一个“通知事件”(Notification Event)，并把这个事件用MsgSend()发给服务器端，意思是：“如果xxx情况发生的话，请用这个事件通知我”。
服务器：收到这个消息后，记录下当时的rcvid，和传过来的事件，然后应答“好的，知道了”。
客户端：因为有了服务器的应答，客户端不再阻塞，可以去做别的事
…过了一段时间
服务器：在某个时刻，客户端所要求的“xxx情况”满足了，服务器调用 MsgDeliverEvent(rcvid, event);以通知客户端
客户端：收到通知，再用MsgSend()发关“xxx 情况的数据在哪里？”
服务器：用MsgReply()把数据返回给客户端

1万 · 2022-9-19 16:57

Robust implementations with Send/Receive/Reply
异步系统的一个重要问题是事件通知需要运行信号处理程序。异步IPC难以彻底对系统进行测试，此外也难以确保信号处理程序按预期的运行。基于Send/Receive/Reply构建的同步、非队列系统结构，可以让应用程序的架构更健壮
使用各种IPC机制时，避免死锁是一个难题，在QNX中只需要遵循两个原则，就可以构建无死锁系统：

永远不要两个线程相互发送消息；
将线程组织为层级结构，并只向上发送消息；

IPC进程间通信

相关下载

相关帖子

突出贡献奖章

沉静之湖泊

七世轮回

技术奇才奖章