【转】QNX系列：进程间消息传递_qnx 任务之间通信

pid    tid name               prio STATE       Blocked       
     4   1 devc-pty            10r RECEIVE     1             

  
  

  
  
1
2
1
2

   pid tid name               prio STATE       Blocked      
     1   1 to/x86/sys/procnto   0f READY                    
     1   2 to/x86/sys/procnto  10r RECEIVE     1            
     1   3 to/x86/sys/procnto  10r NANOSLEEP                
     1   4 to/x86/sys/procnto  10r RUNNING                  
     1   5 to/x86/sys/procnto  15r RECEIVE     1            
 16426   1 esh                 10r REPLY       1            

  
  

  
  
1
2
3
4
5
6
7
1
2
3
4
5
6
7

#include <fcntl.h>
#include <unistd.h>
1
2

int
main (void)
{
int fd;

fd = open ("filename", O_WRONLY);
write (fd, "This is message passing\n", 24);
close (fd);

return (EXIT_SUCCESS);
1
2
3
4
5

}

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14

看到？标准C代码，没什么棘手的。

消息传递是由Neutrino C库完成的。您只需发出标准POSIX 1003.1或ANSI C函数调用，C库即可为您完成消息传递工作。

在上面的示例中，我们看到了三个函数被调用并且三个不同的消息被发送：

• open（）发送了“打开”消息
• write（）发送了一条“写”消息
• close（）发送了一个“关闭”消息

当我们查看资源管理器时（在“ 资源管理器”一章中），我们将更详细地讨论消息本身，但是现在您只需要知道发送了不同类型的消息这一事实即可。

让我们退一步，将其与示例在传统操作系统中的工作方式进行对比。

客户端代码将保持不变，而差异将被供应商提供的C库隐藏。在这样的系统上，open（）函数调用将调用内核函数，然后该内核函数将直接调用文件系统，该文件系统将执行一些代码，并返回文件描述符。在写（）和close（）方法的调用会做同样的事情。

所以？用这种方式做事有好处吗？继续阅读！

网络分布式消息传递

假设我们想更改上面的示例，以与网络上的其他节点通信。您可能认为我们必须调用特殊的函数调用才能“联网”。这是网络版本的代码：

#include <fcntl.h>
#include <unistd.h>
1
2

int
main (void)
{
int fd;
fd = open (“/net/wintermute/home/rk/filename”, O_WRONLY);
write (fd, “This is message passing\n”, 24);
close (fd);

return (EXIT_SUCCESS);
1

}

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13

如果您认为两个版本中的代码几乎相同，那您是对的。它是。

在传统的OS中，C库open（） 调用内核，内核查看文件名并说“哎呀，这是在另一个节点上”。然后内核调用网络文件系统（NFS）代码，该代码确定/net/wintermute/home/rk/filename实际位置。然后，NFS调用网络驱动程序并将消息发送到node上的内核wintermute，然后该消息重复我们在原始示例中描述的过程。请注意，在这种情况下，实际上涉及两个文件系统。一种是NFS客户端文件系统，另一种是远程文件系统。不幸的是，根据不兼容的情况，取决于远程文件系统和NFS的实现，某些操作可能无法按预期进行（例如，文件锁定）。

在Neutrino下，C库open（）创建 与发送到本地文件系统相同的消息，并将其发送到node上的文件系统wintermute。在本地和远程情况下，将使用完全相同的文件系统。

这是Neutrino的另一个基本特征：网络分发的操作基本上是“免费的”，因为已经通过消息传递完成了将客户端的功能需求与服务器提供的服务脱钩的工作。

在传统内核上有一个“双重标准”，其中本地服务以一种方式实现，而远程（网络）服务则以完全不同的方式实现。

对你意味着什么

消息传递是优雅的并且是网络分布的。所以呢？程序员，您能从中得到什么？

好吧，这意味着您的程序继承了这些特性-它们也可以通过网络分发，而工作量却比其他系统少得多。但是我发现最有用的好处是，它们使您能够以一种很好的模块化方式来测试软件。

您可能已经在大型项目中工作，其中许多人必须提供不同的软件。当然，这些人中有些人迟早要完成。

这些项目通常会在两个阶段出现问题：最初是在项目定义时，很难确定一个人的开发工作在哪里结束而另一个人开始了，然后在测试/集成时，不可能进行完整的系统集成测试。因为所有的东西都不可用。

通过消息传递，可以非常轻松地解耦项目的各个组件，从而实现非常简单的设计和相当简单的测试。如果您想以现有的范式来考虑这一点，则它与面向对象编程（OOP）中使用的概念非常相似。

归结为可以逐个进行测试。您可以设置一个简单的程序，将消息发送到服务器进程，并且由于该服务器进程的输入和输出已经（或应该被记录），因此可以确定该进程是否正常运行。哎呀，这些测试用例甚至可以自动化并放置在定期运行的回归套件中！

中微子的哲学

信息传递是Neutrino哲学的核心。了解消息传递的用途和含义将是有效使用操作系统的关键。在详细介绍之前，让我们先看一些理论。

多线程

尽管客户端/服务器模型易于理解，并且是最常用的模型，但是在主题上还有两个变体。第一个是使用多个线程（本节的主题），第二个是称为服务器/子服务器的模型，该模型 有时对常规设计有用，但在网络分布式设计中确实很有用。两者的结合可能非常强大，尤其是在SMP盒网络上！

正如我们在“ 进程和线程”一章中讨论的那样，Neutrino能够在同一进程中运行多个执行线程。当我们将其与消息传递结合在一起时，如何利用它来发挥优势？

答案很简单。我们可以启动一个线程池 （使用我们在“ 进程和线程”一章中讨论过的*thread_pool _ *（）*函数），每个线程都可以处理来自客户端的消息：

客户端访问服务器中的线程。

这样，当客户向我们发送消息时，只要工作完成，我们就真的不在乎哪个线程收到消息 。这具有许多优点。与仅使用一个线程为多个客户端提供服务相比，使用多个线程为多个客户端提供服务的能力是一个强大的概念。主要优点是内核可以在各个客户端之间对服务器执行多任务处理，而服务器本身不必执行多任务处理。

在单处理器计算机上，运行一堆线程意味着它们都在争夺CPU时间。

但是，在SMP盒上，我们可以有多个线程争用多个CPU，同时在多个CPU之间共享相同的数据区域。这意味着我们仅受该特定计算机上可用CPU数量的限制。

服务器/子服务器

现在让我们看一下服务器/子服务器模型，然后将其与多线程模型结合起来。

在此模型中，服务器仍然为客户端提供服务，但是由于这些请求可能需要很长时间才能完成，因此我们需要能够启动请求，并且仍然能够处理来自其他客户端的新请求。

如果我们尝试使用传统的单线程客户端/服务器模型执行此操作，则一旦收到并启动一个请求，除非定期停止正在执行的操作，否则我们将无法再接收到任何请求，请快速浏览一下看看是否还有其他待处理的请求，将它们放在工作队列中，然后继续进行，将我们的注意力分散到工作队列中的各种工作上。不太有效。您实际上是通过在多个作业之间进行“时间分割”来复制内核的工作！

想象一下，如果您这样做的话会是什么样子。您在办公桌旁，有人拿着一个满是工作的文件夹走到您身边。您开始处理它。当您忙于工作时，您会注意到其他人正站在您的隔间门口，而他们的工作同样具有较高的优先级（当然）！现在您的办公桌上有两堆工作。您要花一分钟时间在一个桩上，然后切换到另一个桩，依此类推，一直在看着门口，看看是否有人在忙着做更多的工作。

服务器/子服务器模型在这里更有意义。在此模型中，我们有一台服务器，该服务器创建其他几个进程（子服务器）。这些子服务器每个都向服务器发送一条消息，但是服务器直到收到客户端的请求才回复它们。然后，它将客户端的请求通过应执行的工作进行回复，从而将客户端的请求传递给子服务器之一。下图说明了这一点。注意箭头的方向-它们指示发送的方向！

服务器/子服务器模型。

如果您正在做这样的工作，那么首先要雇用一些额外的员工。这些员工都会来找您（就像子服务器将消息发送到服务器一样-因此请注意上图中的箭头），寻找工作要做。最初，您可能没有任何查询，因此您不会回复他们的查询。当有人带着满是工作文件夹的人走进您的办公室时，您对您的一名员工说：“这是您要做的一些工作。” 然后，该员工离开并开始工作。随着其他工作的到来，您会将它们委派给其他员工。

该模型的诀窍是它是由回复驱动的 —当您回复子服务器时，工作就开始了。标准的客户端/服务器模型是*发送驱动的，*因为工作是在您向服务器发送消息时开始的。

那么，为什么客户要进入您的办公室，而不是您雇用的雇员的办公室呢？您为什么要“仲裁”工作？答案很简单：您是负责执行特定任务的协调员。您需要确保完成工作。与您一起工作的客户认识您，但他们不知道您（可能是临时）员工的姓名或位置。

您可能会怀疑，您当然可以将多线程服务器与服务器/子服务器模型混合使用。主要技巧是确定“问题”的哪些部分最适合在网络上分发（通常是那些不会占用过多网络带宽的部分），以及哪些部分最适合在网络上分发SMP体系结构（通常是那些希望使用公共数据区域的部分）。

那么，为什么我们要使用一个？使用服务器/子服务器方法，我们可以将工作分配到网络上的多台计算机上。这实际上意味着我们仅受网络上可用计算机数量的限制（当然还有网络带宽）。将其与分布在网络上的一堆SMP盒上的多个线程结合在一起会产生“计算集群”，其中中央“仲裁员”将工作（通过服务器/子服务器模型）委派给网络上的SMP盒。

一些例子

现在，我们将考虑每种方法的一些示例。

发送驱动（客户端/服务器）

文件系统，串行端口，控制台和声卡均使用客户端/服务器模型。AC语言应用程序担当客户端的角色，并将请求发送到这些服务器。服务器执行指定的任何工作，然后给出答案。

实际上，其中一些传统的“客户端/服务器”服务器实际上可能是答复驱动的（服务器/子服务器）服务器！这是因为，即使服务器本身使用服务器/子服务器方法来完成工作，对于最终客户而言，它们还是作为标准服务器出现的。我的意思是，客户端仍然向其认为的“服务提供过程”发送一条消息。实际发生的是，“服务提供过程”只是将客户的工作委托给另一个过程（子服务器）。

回复驱动（服务器/子服务器）

一种比较流行的答复驱动程序是分布在网络上的分形图形程序。主程序将屏幕分为几个区域，例如64个区域。在启动时，将为主程序提供可以参与此活动的节点列表。主程序启动工作程序（子服务器），在每个节点上启动一个程序，然后等待工作程序发送到主程序。

然后，主节点反复选择（屏幕上的64个）“未填充”区域，并将分形计算工作通过答复将其委托给另一个节点上的工作程序。工作程序完成计算后，它将结果发送回主站，主站将结果显示在屏幕上。

因为工作程序已发送给主服务器，所以现在要由主服务器来再次进行更多工作。主机继续执行此操作，直到屏幕上的所有64个区域都已填满。

重要的微妙之处

因为主程序将工作委派给工作程序，所以主程序不能被任何一个程序阻塞！在传统的发送驱动方法中，您希望主服务器创建一个程序然后发送给它。不幸的是，在完成工作程序之前，主程序不会得到答复，这意味着主程序无法同时发送到另一个工作程序，这实际上消除了拥有多个工作程序节点的优势。

一位主人，多名工人。

解决此问题的方法是启动工作程序，并 通过向其发送消息来询问主程序是否有任何工作要做。再一次，我们使用了图中箭头的方向来指示发送的方向。现在，工作程序正在等待主程序答复。当某事告诉主程序执行某些工作时，它会答复一个或多个工人，这使他们离开并开始工作。这使工人可以开展业务。主程序仍然可以响应新请求（在等待来自其中一位工作人员的答复之前，它不会被阻止）。

多线程服务器

从客户端的角度来看，多线程服务器与单线程服务器没有区别。实际上，服务器的设计者可以通过启动另一个线程来“打开”多线程。

无论如何，即使服务器仅服务于一个“客户端”，服务器仍可以在SMP配置中使用多个CPU。那是什么意思？让我们回顾一下分形图形示例。当子服务器收到服务器的“计算”请求时，绝对没有阻止子服务器启动多个CPU上的多个线程来满足一个请求的操作。实际上，为了使应用程序在具有一些SMP框和某些单CPU框的网络上更好地扩展，服务器和子服务器可以最初交换一条消息，从而子服务器告诉服务器它有多少个CPU –这样就可以知道有多少个CPU。请求它可以同时服务。然后，服务器将排队更多对SMP盒的请求，从而使SMP盒比单CPU盒做更多的工作。

使用消息传递

既然我们已经了解了消息传递所涉及的基本概念，并且了解到甚至C语言库之类的日常事物都在使用它，让我们来看一些细节。

建筑与结构

我们一直在谈论“客户端”和“服务器”。我还使用了三个关键短语：

• “客户端发送到服务器。”
• “服务器从客户端接收。”
• “服务器回复客户端。”

我专门使用了这些短语，因为它们紧密反映了Neutrino消息传递操作中使用的实际函数名称。

这是在Neutrino下可用的处理消息传递的功能的完整列表（按字母顺序）：

• ChannelCreate（）， ChannelDestroy（）
• ConnectAttach（）， ConnectDetach（）
• MsgDeliverEvent（）
• MsgError（）
• MsgRead（）， MsgReadv（）
• MsgReceive（）， MsgReceivePulse（）， MsgReceivev（）
• MsgReply（）， MsgReplyv（）
• MsgSend（）， MsgSendnc（）， MsgSendsv（）， MsgSendsvnc（）， MsgSendv（）， MsgSendvnc（）， MsgSendvs（）， MsgSendvsnc（）
• MsgWrite（）， MsgWritev（）

不要让这个清单让您不知所措！您可以仅使用列表中的一小部分调用来编写非常有用的客户端/服务器应用程序-当您习惯了这些主意时，您会发现某些其他功能在某些情况下可能非常有用。

我们将把讨论分为适用于客户端的功能和适用于服务器端的功能。

客户端

客户端希望向服务器发送请求，阻止直到服务器完成请求，然后在请求完成并且客户端被解除阻止后才能获得“答案”。

这意味着两件事：客户端需要能够建立与服务器的连接，然后通过消息传输数据-消息从客户端到服务器（“发送”消息）以及消息从服务器回到服务器。客户端（“答复”消息，服务器的答复）。

建立连接

对于客户端

因此，让我们依次看一下这些功能。我们需要做的第一件事是建立连接。我们使用功能ConnectAttach（）来执行此操作，该函数如下所示：

#include <sys/neutrino.h>
1

int ConnectAttach (int nd,
pid_t pid,
int chid,
unsigned index,
int flags);

• 1
• 2
• 3
• 4
• 5
• 6
• 7

为ConnectAttach（）提供了三个标识符：nd（它是节点描述符），pid（它是进程ID）和chid（它是通道ID）。这三个ID（通常称为“ ND / PID / CHID”）唯一标识客户端要连接的服务器。我们将忽略索引和标志（只需将它们设置为0）。

因此，假设我们要连接到节点上的进程ID 77（通道ID 1）。这是执行此操作的代码示例：

int coid;
1

coid = ConnectAttach (0, 77, 1, 0, 0);

• 1
• 2
• 3

如您所见，通过指定nd为零，我们告诉内核我们希望在节点上建立连接。

至此，我有了一个连接ID —一个小的整数，可以唯一地标识从客户端到特定服务器上特定通道的连接。

发送到服务器时，我可以随意使用此连接ID。完成后，可以通过以下方式销毁它：

ConnectDetach (coid);

 
 

 
 
1
1

因此，让我们看看我如何实际使用它。

传送讯息

使用*MsgSend *（）*函数系列的某些变体可以实现在客户端上传递消息。我们将看看最简单的成员MsgSend（）：

#include <sys/neutrino.h>
1

int MsgSend (int coid,
const void *smsg,
int sbytes,
void *rmsg,
int rbytes);

• 1
• 2
• 3
• 4
• 5
• 6
• 7

*MsgSend（）*的参数为：

• 目标服务器的连接ID（coid），
• 指向发送消息的指针（smsg），
• 发送消息的大小（sbytes），
• 指向回复消息的指针（rmsg），以及
• 回复消息的大小（rbytes）。

不光有发送消息，还有回复消息，这一点注意。

没有比这更简单的了！

让我们发送一条简单消息到进程ID 77，通道ID 1：

#include <sys/neutrino.h>
1

char *smsg = “This is the outgoing buffer”;
char rmsg [200];
int coid;

// establish a connection
coid = ConnectAttach (0, 77, 1, 0, 0);
if (coid == -1) {
fprintf (stderr, “Couldn’t ConnectAttach to 0/77/1!\n”);
perror (NULL);
exit (EXIT_FAILURE);
}

// send the message
if (MsgSend (coid,
smsg,
strlen (smsg) + 1,
rmsg,
sizeof (rmsg)) == -1) {
fprintf (stderr, “Error during MsgSend\n”);
perror (NULL);
exit (EXIT_FAILURE);
}

if (strlen (rmsg) > 0) {
printf (“Process ID 77 returns “%s”\n”, rmsg);
}

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28

假定进程ID 77是活动服务器，并且在其通道ID 1上期望该消息的特定格式。服务器接收到该消息后，将对其进行处理，并在某时回复结果。那时，**MsgSend（）将返回0，表示一切正常。如果服务器在答复中向我们发送了任何数据，我们将在最后一行代码中进行打印（假设我们返回的是NUL终止的ASCII数据）。

服务器

现在我们已经看到了客户端，让我们看一下服务器。客户端使用*ConnectAttach（）*创建与服务器的连接，然后使用 *MsgSend（）*进行所有消息传递。

建立频道

这意味着服务器必须创建一个通道-这是客户端在发出ConnectAttach（） 函数调用时所连接的东西。创建通道后，服务器通常会将其永久保留。

通过ChannelCreate（）函数创建通道，并通过ChannelDestroy（）函数销毁通道：

#include <sys/neutrino.h>
1

int ChannelCreate (unsigned flags);

int ChannelDestroy (int chid);

• 1
• 2
• 3
• 4
• 5

稍后，我们将返回flags参数（在下面的“ Channel flags ”部分）。现在，我们仅使用0。因此，要创建通道，服务器将发出：

int  chid;
1

chid = ChannelCreate (0);

• 1
• 2
• 3

所以我们有一个频道。此时，客户端可以（通过ConnectAttach（））连接到此通道并开始发送消息：

服务器通道和客户端连接之间的关系。

讯息处理

就消息传递方面而言，服务器分两个阶段处理消息传递。“接收”阶段和“答复”阶段：

客户端和服务器消息传递功能的关系。

我们将首先看一下这些函数的两个简单版本MsgReceive（）和MsgReply（），然后再看一些变体。

#include <sys/neutrino.h>
1

int MsgReceive (int chid,
void *rmsg,
int rbytes,
struct _msg_info *info);

int MsgReply (int rcvid,
int status,
const void *msg,
int nbytes);

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11

让我们看一下参数之间的关系：

消息数据流。

从图中可以看到，我们需要谈论四件事：

1. 客户端发出MsgSend（）并指定其发送缓冲区（smsg指针和sbytes 长度）。这将被传输到服务器的MsgReceive（）函数提供的缓冲区中，以rmsg 表示，长度为rbytes。客户端现在被阻止。
2. 服务器的MsgReceive（）函数取消阻止，并返回rcvid，服务器稍后将使用该rcvid进行回复。此时，数据可供服务器使用。
3. 服务器已完成的消息的处理，现在使用rcvid它从得到MsgReceive（）通过它传递给MsgReply（） 。注意，MsgReply（） 函数获得一个缓冲器（SMSG具有限定大小（）sbytes）作为数据的位置发射到所述客户端。现在，数据已由内核传输。
4. 最后，sts参数由内核传输，并显示为来自客户端*MsgSend（）*的返回值。客户端现在解除阻止。

您可能已经注意到，有两种尺寸，每缓冲传输（在客户端发送的情况下，有sbytes在客户端和rbytes已在服务器端，在服务器应答的情况下，有sbytes在服务器端和rbytes已在客户端上出现了两组大小，以便每个组件的程序员可以指定其缓冲区的大小。这样做是为了增加安全性。

在我们的示例中，*MsgSend（）*缓冲区的大小与消息字符串的长度相同。让我们看一下服务器，看看那里的大小是如何使用的。

服务器框架

这是服务器的整体结构：

#include <sys/neutrino.h>
1

…

void
server (void)
{
int rcvid; // indicates who we should reply to
int chid; // the channel ID
char message [512]; // big enough for our purposes

<span class="token comment">// create a channel</span>
chid <span class="token operator">=</span> ChannelCreate <span class="token punctuation">(</span><span class="token number">0</span><span class="token punctuation">)</span><span class="token punctuation">;</span>

<span class="token comment">// this is typical of a server:  it runs forever</span>
<span class="token keyword">while</span> <span class="token punctuation">(</span><span class="token number">1</span><span class="token punctuation">)</span> <span class="token punctuation">{<!-- --></span>

    <span class="token comment">// get the message, and print it</span>
    rcvid <span class="token operator">=</span> MsgReceive <span class="token punctuation">(</span>chid<span class="token punctuation">,</span> message<span class="token punctuation">,</span> <span class="token keyword">sizeof</span> <span class="token punctuation">(</span>message<span class="token punctuation">)</span><span class="token punctuation">,</span>
                        <span class="token constant">NULL</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
    printf <span class="token punctuation">(</span><span class="token string">"Got a message, rcvid is %X\n"</span><span class="token punctuation">,</span> rcvid<span class="token punctuation">)</span><span class="token punctuation">;</span>
    printf <span class="token punctuation">(</span><span class="token string">"Message was \"%s\".\n"</span><span class="token punctuation">,</span> message<span class="token punctuation">)</span><span class="token punctuation">;</span>

    <span class="token comment">// now, prepare the reply.  We reuse "message"</span>
    strcpy <span class="token punctuation">(</span>message<span class="token punctuation">,</span> <span class="token string">"This is the reply"</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
    MsgReply <span class="token punctuation">(</span>rcvid<span class="token punctuation">,</span> EOK<span class="token punctuation">,</span> message<span class="token punctuation">,</span> <span class="token keyword">sizeof</span> <span class="token punctuation">(</span>message<span class="token punctuation">)</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
<span class="token punctuation">}</span>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

}

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28

如您所见，MsgReceive（）告诉内核它可以处理最多sizeof (message)（或512字节）的消息。我们的示例客户端（上述）仅发送了28个字节（字符串的长度）。下图说明：

传输的数据少于预期。

内核传输两种大小指定的最小值。在我们的例子中，内核将传输28个字节。服务器将被解除阻止并打印出客户端的消息。（512字节缓冲区中的）其余484字节将不受影响。

我们再次使用*MsgReply（）*遇到相同的情况。该 MsgReply（）函数说，它要转移512个字节，但我们的客户的MsgSend（）函数指定最多200个字节可以转移。因此内核再次传输最小值。在这种情况下，客户端可以接受的200个字节限制了传输大小。（这里一个有趣的方面是，一旦服务器传输了数据，如果客户端没有收到所有数据，如我们的示例所示，就无法取回数据-它永远消失了。）

发送层次结构

在消息传递环境中可能不明显的一件事是需要遵循严格的发送层次结构。这意味着两个线程绝不应该相互发送消息。相反，它们的组织方式应使每个线程都占据一个“级别”；所有发送都从一个级别到更高级别，从不到相同或更低级别。两个线程互相发送消息的问题在于，最终您将遇到死锁问题-两个线程都在等待对方答复各自的消息。由于线程被阻塞，它们将永远没有机会运行和执行回复，因此您最终将获得两个（或更多！）挂起的线程。

将级别分配给线程的方法是将最外部的客户端置于最高级别，然后从那里开始工作。例如，如果您有一个依赖于某些数据库服务器的图形用户界面，而数据库服务器又依赖于文件系统，而文件系统又依赖于块文件系统驱动程序，那么您将拥有不同的自然层次结构流程。发送将从最外部的客户端（图形用户界面）流向较低的服务器；答复将朝相反的方向流动。

尽管这在大多数情况下当然可行，但是您会 遇到需要“中断”发送层次结构的情况。永远不会仅通过违反发送层次结构并“针对流”发送消息来完成此操作，而不能使用MsgDeliverEvent（）函数，稍后我们将进行介绍。

接收ID，频道和其他参数

在上面的示例中，我们没有讨论各种参数，因此我们可以只关注消息传递。现在让我们看一下。

有关渠道的更多信息

在上面的服务器示例中，我们看到服务器仅创建一个通道。它当然可以创建更多，但通常服务器不这样做。（具有两个通道的服务器最明显的例子是透明分布式处理（TDP，也称为Qnet）本机网络管理器-绝对是一件“奇怪的”软件！）

事实证明，在现实世界中确实并不需要太多创建多个渠道。通道的主要目的是为服务器提供一个明确定义的位置来“监听”消息，并为客户端提供一个明确定义的位置（通过连接）发送消息。大约在服务器中唯一具有多个通道的时间是服务器是否希望根据消息到达的通道提供不同的服务或不同类别的服务。例如，第二个通道可用作丢弃唤醒脉冲的地方-这确保了将它们视为与到达第一个通道的消息不同的服务“类别”。

在上一段中，我曾说过，您可以在服务器中运行一个线程池，准备接受来自客户端的消息，并且哪个线程收到请求并不重要。这是通道抽象的另一方面。在QNX家族操作系统的早期版本（尤其是QNX 4）下，客户端会将消息定向到由节点ID和进程ID标识的服务器上。由于QNX 4是单线程的，因此这意味着不会对将消息发送给“谁”感到困惑。但是，一旦在图中引入了线程，就必须决定如何处理线程（实际上是“服务提供者”）。由于线程是短暂的，因此让客户端连接到特定的节点ID，进程ID，ID。另外，如果该特定线程很忙怎么办？我们必须提供一些方法，以允许客户端选择“已定义的服务提供线程池中的不忙线程”。

嗯，这就是渠道。它是“服务提供线程池”的“地址”。这里的含义是，一堆线程可以在特定通道上发出MsgReceive（）函数调用并阻塞，一次只有一个线程获取一条消息。

谁发的消息？

服务器通常需要知道是谁发送了消息。有许多的原因：

• 会计
• 访问控制
• 上下文关联
• 服务等级
• 等等

让客户端在发送的每条消息中都提供此信息会很麻烦（并且有安全漏洞）。因此，每当*MsgReceive（）*函数因为收到消息而解除阻止时，内核就会填充一个结构。此结构的类型为struct _msg_info，包含以下内容：

struct _msg_info
{
    int     nd;
    int     srcnd;
    pid_t   pid;
    int32_t chid;
    int32_t scoid;
    int32_t coid;
    int32_t msglen;
    int32_t tid;
    int16_t priority;
    int16_t flags;
    int32_t srcmsglen;
    int32_t dstmsglen;
};

 
 

 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

您将其作为最后一个参数传递给*MsgReceive（）*函数。如果传递NULL，则什么也不会发生。（信息可以稍后通过MsgInfo（）调用来检索，因此不会永远消失！）

让我们看一下这些字段：

• nd，srcnd，pid和tid

  客户端的节点描述符，进程ID和线程ID。（请注意nd是发送节点的接收节点的节点描述符；srcnd是接收节点的发送节点的节点描述符。这有一个很好的理由:-)，我们将在下面的“ 关于ND的一些说明 ”中看到）。

• 优先

  发送线程的优先级。

• CHID，COID

  消息发送到的通道ID，以及使用的连接ID。

• 螺旋状

  服务器连接ID。这是内核使用的内部标识符，用于将消息从服务器路由回客户端。除了有趣的事实，它是一个唯一代表客户端的小整数，您不需要了解它。

• 标志

  包含各种标志位_NTO_MI_ENDIAN_BIG，_NTO_MI_ENDIAN_DIFF，_NTO_MI_NET_CRED_DIRTY和_NTO_MI_UNBLOCK_REQ。_NTO_MI_ENDIAN_BIG和_NTO_MI_ENDIAN_DIFF告诉您发送计算机的字节序（如果消息是通过具有不同字节序的计算机通过网络发送的），则在内部使用_NTO_MI_NET_CRED_DIRTY；我们将在下面的“使用_NTO_MI_UNBLOCK_REQ”部分中查看_NTO_MI_UNBLOCK_REQ。

• msglen

  接收的字节数。

• srcmsglen

  客户端发送的源消息的长度（以字节为单位）。这可能大于msglen中的值，例如接收的数据少于发送的数据时的情况。请注意，只有在ChannelCreate（）的flags参数中为接收消息的通道设置了_NTO_CHF_SENDER_LEN时，此成员才有效。

• dstmsglen

  客户端的应答缓冲区的长度，以字节为单位。仅当在*ChannelCreate（）*的参数中为接收消息的通道设置了_NTO_CHF_REPLY_LEN标志时，此字段才有效。

接收ID（又称客户端Cookie）

在上面的代码示例中，请注意我们如何：

rcvid = MsgReceive (…);
…
MsgReply (rcvid, …);

 
 

 
 
1
2
3
1
2
3

这是关键的代码段，因为它说明了从客户端接收消息与然后（稍后再）回复该特定客户端之间的绑定。接收ID是一个整数，充当“魔术cookie”，如果以后要与客户端进行交互，则需要保留该ID。如果丢失了该怎么办？没了。直到您（服务器）死掉，或者如果客户端在消息传递调用上有超时，客户端才不会解除对MsgSend（）的阻止（即使那样也很棘手；请参见Neutrino库参考中的TimerTimeout（）函数，并 在“内核超时”下的“ 时钟，定时器和每时每刻获取”一章中讨论了它的用法。

这将我们带到*MsgReply（）*函数。

回复客户

MsgReply（）接受接收ID，状态，消息指针和消息大小。我们刚刚完成了接收ID的讨论；它标识应将回复消息发送给谁。状态变量指示应该传递给客户端的 MsgSend（）函数的返回状态。最后，消息指针和大小指示应发送的可选回复消息的位置和大小。

该*MsgReply（）*函数可能看起来是非常简单的（它是），但它的应用程序需要进行一些检查。

不回复客户

绝对没有必要先 通过*MsgReceive（）*接受来自其他客户端的新消息，然后回复客户端！这可以用于许多不同的情况。

在典型的设备驱动程序中，客户端可能会发出长时间无法处理的请求。例如，客户可能要求模数转换器（ADC）设备驱动程序“出去并收集价值45秒的样本”。同时，ADC驱动程序不应只关闭45秒！其他客户端可能希望服务请求（例如，可能有多个模拟通道，或者可能应立即提供状态信息，等等）。

在架构上，ADC驱动程序将简单地将它从*MsgReceive（）*获得的接收ID排队，开始45秒的累积过程，然后处理其他请求。当45秒结束且样本已累积时，ADC驱动程序可以找到与请求关联的接收ID，然后回复客户端。

对于答复驱动的服务器/子服务器模型（其中一些“客户机”是子服务器），您还希望推迟对客户机的答复。由于子服务器正在寻找工作，因此您只需记下其接收ID并将其存储起来即可。当实际工作到达时，您才会答复子服务器，从而表明它应该做一些工作。

无数据回复或出现错误

当您最终回复客户时，无需传输任何数据。这在两种情况下使用。

如果答复的唯一目的是解除对客户的阻止，则可以选择不包含任何数据的答复。假设客户只想在某个特定事件发生之前就被阻止，但是它不需要知道哪个事件。在这种情况下，*MsgReply（）*函数不需要任何数据。接收ID足够：

MsgReply (rcvid, EOK, NULL, 0);

 
 

 
 
1
1

这将解除对客户端的阻止（但不返回任何数据），并返回EOK“成功”指示。

作为对此的略微修改，您可能希望将错误状态返回给客户端。在这种情况下，您不能使用MsgReply（）来做到这一点，而必须使用MsgError（）：

MsgError (rcvid, EROFS);

 
 

 
 
1
1

在上面的示例中，服务器检测到客户端正在尝试写入只读文件系统，并且不返回任何实际数据，而只是将EROFS 的错误号返回给客户端。

另外，（稍后我们将看一下调用），您可能已经（通过MsgWrite（））传输了数据，并且没有其他数据可以传输。

为什么要打两个电话？他们有些不同。虽然这两个MsgError（）和MsgReply（）将解除客户端，MsgError（）将不会转让任何额外的数据，会导致客户端的MsgSend（）函数返回-1，并会导致客户端有 错误号设置为任何传递作为*MsgError（）*的第二个参数。

另一方面，MsgReply（） 可以传输数据（如第三个和第四个参数所示），并将导致客户端的MsgSend（）函数返回作为第二个参数传递给MsgReply（）的内容。MsgReply（）有没有在客户端的效果错误号。

通常，如果仅返回通过/失败指示（不包含任何数据），则应使用MsgError（），而如果返回数据，则应使用MsgReply（）。传统上，当你这样做的回报数据，第二个参数*MsgReply（）*将返回一个正整数，指示的字节数。

查找服务器的ND / PID / CHID

您已经注意到，在ConnectAttach（）函数中，我们需要一个节点描述符（ND），一个进程ID（PID）和一个通道ID（CHID）才能附加到服务器。到目前为止，我们还没有讨论客户端如何找到此ND / PID / CHID信息。

如果一个流程创建了另一个流程，则很容易-流程创建调用将返回新创建的流程的流程ID。创建进程可以在命令行上将其自己的PID和CHID传递给新创建的进程，或者新创建的进程可以发出getppid（）函数调用以获取其父级的PID并采用“众所周知的” CHID。

如果我们有两个完美的陌生人怎么办？例如，如果第三方创建了一个服务器，而您编写的应用程序想与该服务器通信，便会是这种情况。真正的问题是，“服务器如何 宣传其位置？”

有很多方法可以做到这一点。我们将按照编程“优雅”的升序查看其中四个：

1. 打开一个众所周知的文件名，并在其中存储ND / PID / CHID。这是UNIX风格的服务器采用的传统方法，它们在其中打开文件（例如/etc/httpd.pid），在其中以ASCII字符串形式写入其进程ID，并期望客户端将打开文件并获取进程ID。
2. 使用全局变量来通告ND / PID / CHID信息。这通常用在需要向自己发送消息的多线程服务器中，从本质上来说，它是一种非常有限的情况。
3. 使用名称定位功能（name_attach（）和name_detach（），然后在客户端使用name_open（）和name_close（）函数）。
4. 接管一部分路径名空间，然后成为资源管理器。在“ 资源管理器”一章中介绍资源管理器时，我们将对此进行讨论。

第一种方法非常简单，但是会遭受“路径名污染”的困扰，/etc 目录中包含各种*.pid文件。由于文件是永久性的（这意味着它们在创建过程终止并在计算机重新启动后仍然存在），因此没有明显的清除这些文件的方法，除非可能有个“垃圾收割者”任务，以查看这些事情是否仍然有效。

还有另一个相关的问题。由于创建文件的过程可以在不删除文件的情况下消失，因此，在尝试向其发送消息之前，无法知道该过程是否仍然存在。更糟糕的是，文件中指定的ND / PID / CHID可能过时，可能会被另一个程序重用！您发送到该程序的消息最多将被拒绝，最坏的情况可能会导致损坏。这样就行了。

第二种方法，我们使用全局变量来通告ND / PID / CHID值，这不是一般的解决方案，因为它依赖于客户端能够访问全局变量。而且由于这需要共享内存，因此它肯定无法在网络上工作！通常在小型测试用例程序或非常特殊的情况下都使用此方法，但总是在多线程程序的上下文中使用。实际上，所有发生的事情是程序中的一个线程是客户端，而另一个线程是服务器。服务器线程创建通道，然后将通道ID放到全局变量中（该进程中所有线程的节点ID和进程ID都相同，因此不需要播发它们。）然后，客户端线程选择全局通道ID并执行ConnectAttach（） 对此。

第三种方法，我们使用*name_attach（）和name_detach（）*函数，适用于简单的客户端/服务器情况。

服务器成为资源管理器的最后一种方法绝对是最干净的，并且是推荐的通用解决方案。的机制“如何”将成为明显的资源管理一章，但现在，所有你需要知道的是，服务器注册一个特定的路径作为其“权力的领域，”和客户端进行简单*的open（）*的该路径名。

优先事项呢？

如果低优先级进程和高优先级进程同时向服务器发送消息怎么办？

当我们在本章稍后讨论优先级反转时，我们将回到该问题引入的其他一些细微差别。

读写数据

到目前为止，您已经了解了基本的消息传递原语。如前所述，这些都是您所需要的。但是，还有一些额外的功能可以使生活更加轻松。

让我们考虑一个使用客户端和服务器的示例，其中可能需要其他功能。

客户端发出*MsgSend（）将一些数据传输到服务器。客户端发出MsgSend（）之后，*它将阻止；现在正在等待服务器回复。

一个有趣的事情发生在服务器端。服务器已调用MsgReceive（）来从客户端接收消息。根据您为邮件选择的设计，服务器可能会或可能不会知道客户端的邮件大小。到底为什么服务器不知道消息有多大？考虑我们一直在使用的文件系统示例。假设客户端这样做：

write (fd, buf, 16);

 
 

 
 
1
1

如果服务器执行MsgReceive（） 并指定例如1024字节的缓冲区大小，则这将按预期工作。由于我们的客户只发送了一条很小的消息（28个字节），所以我们没有问题。

但是，如果客户端发送的内容大于1024字节（例如1兆字节）怎么办？

write (fd, buf, 1000000);

 
 

 
 
1
1

服务器将如何优雅地处理此问题？我们可以随意说，不允许客户端写多于n个字节。然后，在用于write（）的客户端C库代码中，我们可以查看此要求，并将写入请求分成几个每个n个字节的请求。好尴尬

此示例的另一个问题是“ n应该是多少？”

您可以看到这种方法有很多缺点：

• 使用有限大小的消息传输的所有函数都必须在C库中进行修改，以便该函数打包请求。这本身可能是相当多的工作。而且，它可能会对多线程函数产生意外的副作用-如果发送了来自一个线程的消息的第一部分，然后客户端中的另一个线程抢占了当前线程并发送了自己的消息，该怎么办。那将原始线程留在哪里？
• 现在，所有服务器必须准备好处理可能到达的最大消息大小。这意味着所有服务器都必须具有较大的数据区域，否则库将必须将较大的请求分解为许多较小的请求，从而影响速度。

幸运的是，此问题的解决方法非常简单，这也给我们带来了一些好处。

MsgRead（）和MsgWrite（）这两个函数在这里特别有用。要记住的重要事实是客户端被阻止。这意味着在服务器尝试检查数据结构时，客户端不会去更改数据结构。

该*MsgRead（）*函数如下所示：

#include <sys/neutrino.h>
1

int MsgRead (int rcvid,
void *msg,
int nbytes,
int offset);

• 1
• 2
• 3
• 4
• 5
• 6

MsgRead（）允许您的服务器从被阻止的客户端的地址空间中读取数据，从客户端指定的“发送”缓冲区的开头开始的偏移字节，到msg为nbytes指定的缓冲区中。服务器不会阻止，客户端也不会阻止。*MsgRead（）*返回实际读取的字节数；如果有错误，则返回-1。

因此，让我们考虑一下在write（） 示例中如何使用它。C库的*write（）函数构造一个消息，并将其标头发送给文件系统服务器fs-qnx4。服务器通过MsgReceive（）*接收一小部分消息，查看该消息 ，然后决定将其余消息放置在何处。该fs-qnx4服务器可以决定把数据最好的地方就是到它已经分配一些缓存缓冲区。

我们来看一个例子：

该fs-qnx4消息例如，示出连续的数据图。

因此，客户端决定将4 KB发送到文件系统。（请注意，C库如何在数据前添加一个微小的标头，以便文件系统可以分辨出它实际上是一种什么样的请求。当我们查看多部分消息时，我们会回到这个问题上，在更多内容中当我们查看资源管理器时，将看到详细信息。）文件系统仅读取足够的数据（标头）来确定它是哪种消息：

// part of the headers, fictionalized for example purposes
struct _io_write {
    uint16_t    type;
    uint16_t    combine_len;
    int32_t     nbytes;
    uint32_t    xtype;
};
1
2
3
4
5
6
7

typedef union {
uint16_t type;
struct _io_read io_read;
struct _io_write io_write;
…
} header_t;

header_t header; // declare the header

rcvid = MsgReceive (chid, &header, sizeof (header), NULL);

switch (header.type) {
…
case _IO_WRITE:
number_of_bytes = header.io_write.nbytes;
…

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24

此时，fs-qnx4知道客户端的地址空间中有4 KB（因为消息在结构的nbytes成员中告知它），并且应该将其传输到缓存缓冲区。该fs-qnx4服务器可以执行以下命令：

MsgRead (rcvid, cache_buffer [index].data,
         cache_buffer [index].size, sizeof (header.io_write));

 
 

 
 
1
2
1
2

请注意，消息传输已指定偏移量，sizeof (header.io_write)以跳过客户端C库添加的写头。我们在这里假设cache_buffer [index].size实际上是4096（或更多）字节。

类似地，为了将数据写入客户端的地址空间，我们有：

#include <sys/neutrino.h>
1

int MsgWrite (int rcvid,
const void *msg,
int nbytes,
int offset);

• 1
• 2
• 3
• 4
• 5
• 6

MsgWrite（）允许您的服务器将数据写入客户端的地址空间，从客户端指定的“接收”缓冲区的开头开始偏移字节。在服务器空间有限但客户端希望从服务器获取大量信息的情况下，此功能最有用。

例如，使用数据获取驱动程序，客户端可以指定4兆字节的数据区域，并告诉驱动程序获取4兆字节的数据。如果有人要求进行大量数据传输，驾驶员确实不需要像这样大的区域。

驱动程序可能有128 KB的区域用于DMA数据传输，然后使用MsgWrite（）将消息零碎地传递到客户端的地址空间中（当然，每次将偏移量增加128 KB）。然后，在写入最后一条数据后，驱动程序将*MsgReply（）*发送给客户端。

使用* MsgWrite（）*传输几个块。

请注意，MsgWrite（）允许您在各个位置编写数据组件，然后只需使用MsgReply（）唤醒客户端即可：

MsgReply (rcvid, EOK, NULL, 0);

 
 

 
 
1
1

或在客户端缓冲区的开头写入标头后唤醒客户端：

MsgReply (rcvid, EOK, &header, sizeof (header));

 
 

 
 
1
1

对于写入未知数量的数据，这是一个相当巧妙的技巧，您可以在其中知道仅在完成写入后才写入多少数据。如果您正在使用这种在数据传输之后写入标头的方法，则必须记住要在客户端数据区的开头留有标头的空间！

多部分消息

到目前为止，我们仅显示了从客户端地址空间中的一个缓冲区到服务器地址空间中的另一缓冲区中发生的消息传输。（在答复期间，服务器空间中的一个缓冲区将进入客户端空间中的另一缓冲区。）

尽管此方法对于大多数应用程序来说已经足够好，但它可能导致效率低下。回想一下，我们的write（） C库代码采用了传递给它的缓冲区，并在其前面粘贴了一个小标头。使用到目前为止所学的知识，您可以期望C库将实现 类似于以下内容的write（）（这不是真正的源代码）：

ssize_t write (int fd, const void *buf, size_t nbytes)
{
    char        *newbuf;
    io_write_t  *wptr;
    int         nwritten;
newbuf = malloc (nbytes + sizeof (io_write_t));

// fill in the write_header at the beginning
wptr = (io_write_t *) newbuf;
wptr -> type = _IO_WRITE;
wptr -> nbytes = nbytes;

// store the actual data from the client
memcpy (newbuf + sizeof (io_write_t), buf, nbytes);

// send the message to the server
nwritten = MsgSend (fd,
                    newbuf, 
                    nbytes + sizeof (io_write_t), 
                    newbuf, 
                    sizeof (io_write_t));
free (newbuf);
return (nwritten);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
1
2
3
4
5

}

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25

看看发生了什么事？一些不好的事情：

• 现在*write（）必须能够malloc（）*一个足以容纳客户端数据（可能相当大）和标头的缓冲区。标题的大小不是问题-在这种情况下，它是12个字节。
• 我们必须复制数据两次：一次是通过memcpy（），然后是在消息传输期间再次。
• 我们必须建立一个指向该io_write_t类型的指针，并将其指向缓冲区的开头，而不是本地访问它（这是一个小麻烦）。

由于内核仍将复制数据，因此，如果我们能告诉它一部分数据（标头）位于某个地址，而另一部分（数据本身）位于某处，那就太好了。否则， 无需我们手动组装缓冲区并复制数据。

碰巧的是，Neutrino实现了一种机制，可以让我们做到这一点！该机制称为IOV，代表“输入/输出向量”。

让我们先看一些代码，然后我们讨论发生的情况：

#include <sys/neutrino.h>
1

ssize_t write (int fd, const void *buf, size_t nbytes)
{
io_write_t whdr;
iov_t iov [2];

// set up the IOV to point to both parts:
SETIOV (iov + 0, &whdr, sizeof (whdr));
SETIOV (iov + 1, buf, nbytes);

// fill in the io_write_t at the beginning
whdr.type = _IO_WRITE;
whdr.nbytes = nbytes;

// send the message to the server
return (MsgSendv (coid, iov, 2, iov, 1));
1
2
3
4
5
6
7
8
9
10

}

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18

首先，请注意没有malloc（）和memcpy（）。接下来，请注意该iov_t 类型的用法。这是一个包含地址和长度对的结构，我们分配了其中的两个（名为iov）。

该iov_t类型定义是由自动包含 <sys/neutrino.h>，并且被定义为：

typedef struct iovec
{
    void    *iov_base;
    size_t   iov_len;
} iov_t;

 
 

 
 
1
2
3
4
5
1
2
3
4
5

给定这种结构，我们用写头（对于第一部分）和来自客户端的数据（在第二部分）填充地址和长度对。有一个名为SETIOV（）的便利宏，可以为我们执行分配。正式定义为：

#include <sys/neutrino.h>
1

#define SETIOV(_iov, _addr, _len)
((_iov)->iov_base = (void *)(_addr),
(_iov)->iov_len = (_len))

• 1
• 2
• 3
• 4
• 5

*SETIOV（）*接受一个iov_t，并将地址和长度数据填充到IOV中。

还要注意，由于我们正在创建一个指向标头的IOV，因此可以在不使用malloc（）的情况下在堆栈上分配标头。这可能是一个祝福和诅咒—当标头很小时，这是一种祝福，因为避免了动态内存分配的麻烦，但是当标头很大时，这可能是一个诅咒，因为它会消耗相当多的堆栈空间。通常，标头很小。

无论如何，重要的工作都是由*MsgSendv（）完成的，该参数接受的参数 与上一个示例中使用的MsgSend（）*函数几乎相同：

#include <sys/neutrino.h>
1

int MsgSendv (int coid,
const iov_t *siov,
int sparts,
const iov_t *riov,
int rparts);

• 1
• 2
• 3
• 4
• 5
• 6
• 7

让我们检查一下参数：

• id

  与*MsgSend（）*一样，我们要发送到的连接ID 。

• sparts和rparts

  iov_t参数指定的发送和接收部件的数量。在我们的示例中，我们将sparts设置为2表示正在发送两部分的消息，将rparts设置为1表示我们正在接收一部分的答复。

• siov和riov

  该iov_t阵列表明我们要发送的地址和长度对。在上面的示例中，我们将siov的2部分设置为指向标题和客户端数据，将riov的1部分设置为仅指向标题。

这是内核如何查看数据的方式：

内核如何看待多部分消息。

内核只是将数据从客户端空间中的IOV的每个部分无缝地复制到服务器空间中（并返回以进行回复）。实际上，内核正在执行聚集-分散操作。

请记住以下几点：

• 部件数“限制”为512 KB；但是，我们的示例2是典型的。
• 内核只是将一个IOV中指定的数据从一个地址空间复制到另一个地址空间。
• 源和目标IOV不必相同。

为什么最后一点如此重要？为了回答这个问题，让我们看一看。在客户端，假设我们发布了：

write (fd, buf, 12000);

 
 

 
 
1
1

生成了两部分的IOV：

• 标头（12个字节）
• 数据（12000字节）

在服务器端（例如，它是文件系统fs-qnx4），我们有许多4 KB的缓存块，我们希望将消息直接有效地接收到缓存块中。理想情况下，我们想编写一些如下代码：

// set up the IOV structure to receive into:
SETIOV (iov + 0, &header, sizeof (header.io_write));
SETIOV (iov + 1, &cache_buffer [37], 4096);
SETIOV (iov + 2, &cache_buffer [16], 4096);
SETIOV (iov + 3, &cache_buffer [22], 4096);
rcvid = MsgReceivev (chid, iov, 4, NULL);

 
 

 
 
1
2
3
4
5
6
1
2
3
4
5
6

该代码几乎完成了您的期望：建立了一个由4部分组成的IOV结构，将结构的第一部分设置为指向标头，随后的三部分设置为指向缓存块37、16和22 （这些数字表示恰好在特定时间可用的缓存块。）这是一个图形表示：

将连续数据转换为单独的缓冲区。

然后，调用MsgReceivev（）函数，表明我们将从指定的频道（chid 参数）接收一条消息，并提供了一个由4部分组成的IOV结构。这也显示了IOV结构本身。

（除了其IOV功能之外，MsgReceivev（）的运行方式与MsgReceive（）相同。）

糟糕！引入*MsgReceive（）*函数时，我们犯了与以前相同的错误。在实际收到消息之前，我们如何知道我们正在接收的消息类型以及与之关联的数据量？

我们可以像以前一样解决此问题：

rcvid = MsgReceive (chid, &header, sizeof (header), NULL);
switch (header.message_type) {
…
case    _IO_WRITE:
    number_of_bytes = header.io_write.nbytes;
    // allocate / find cache buffer entries
    // fill 3-part IOV with cache buffers
    MsgReadv (rcvid, iov, 3, sizeof (header.io_write));

 
 

 
 
1
2
3
4
5
6
7
8
1
2
3
4
5
6
7
8

这将执行初始MsgReceive（）（请注意，我们没有为此使用IOV表单-确实没有必要使用单部分消息来完成此操作），找出它是哪种消息，然后继续阅读数据从客户端的地址空间（从offset开始sizeof (header.io_write)）进入由三部分IOV指定的缓存缓冲区。

注意，我们从使用4部分IOV（在第一个示例中）切换为3部分IOV。这是因为在第一个示例中，四部分IOV的第一部分是标头，我们使用*MsgReceive（）*直接读取了标头，而四部分IOV 的后三部分与三部分IOV相同-它们指定了我们希望数据去往何处。

您可以想象我们将如何执行对读取请求的回复：

1. 查找与请求的数据相对应的缓存条目。
2. 用这些条目填充IOV结构。
3. 使用MsgWritev（）（或MsgReplyv（））将数据传输到客户端。

请注意，如果数据不是从缓存块（或其他数据结构）的开头开始，这不是问题。只需使第一个IOV偏移以指向数据确实开始的位置，然后修改大小即可。

那其他版本呢？

除MsgSend *（） 系列以外的所有消息传递函数都具有相同的通用形式：如果函数v的末尾带有“ ”，则它将带有IOV和多个部分；否则，它需要一个指针和一个长度。

该*MsgSend *（）*家族在源和目的地的消息缓冲区，与内核调用本身的两个变化结合方面四大变化。

看下表：

“线性”是指void *传递单个类型的缓冲区及其长度。记住这一点的简单方法是，“ v”代表“向量”，并且与适当的参数位于相同的位置-第一或第二，分别表示“发送”或“接收”。

嗯……看起来MsgSendsv（）和MsgSendsvnc（）函数是相同的，不是吗？好吧，是的，就它们的参数而言，它们确实是。区别在于它们是否是取消点。“ nc”版本不是取消点，而非“ ”版本是取消点nc。（有关一般取消点和可取消性的更多信息，请查阅Neutrino库参考，位于pthread_cancel（）下。）

实作

您可能已经怀疑MsgRead（），MsgReceive（），MsgSend（）和MsgWrite（）函数的所有变体关系密切。（唯一的例外是MsgReceivePulse（） -我们将在稍后对此进行介绍。）

您应该使用哪个？好吧，这有点哲学上的争论。我个人的喜好是混合搭配。

如果我仅发送或接收一部分消息，为什么还要烦恼设置IOV的复杂性？不管您是自行设置还是让内核/库进行设置，设置它们的微小CPU开销基本上是相同的。单部分消息方法使内核不必进行地址空间操作，并且速度更快。

您应该使用IOV功能吗？绝对！每当您发现自己处理多部分消息时都可以使用它们。当您仅使用几行代码就可以使用多部分邮件传输时，切勿复制数据。这样可以通过最大程度地减少在系统中复制数据的次数来保持系统的尖叫声。传递指针比将数据复制到新缓冲区要快得多。

脉搏

到目前为止，我们讨论的所有消息传递都会阻止客户端。客户端一调用*MsgSend（）*就午睡了。客户端一直处于睡眠状态，直到服务器开始响应为止。

但是，在某些情况下，邮件的发件人无法阻止。我们将在“ 中断和时钟，计时器和时常发生”一 章中查看一些示例，但现在我们应该了解该概念。

实现无阻塞发送的机制称为脉冲。脉搏是一个很小的信息，它表明：

• 可以携带40位有效负载（8位代码和32位数据）
• 对发件人没有阻塞
• 可以像其他任何消息一样被接收
• 如果没有阻塞等待接收者，则将其排队。

接收脉冲信息

接收脉冲非常简单：微小的，定义明确的消息会呈现给MsgReceive（），就像线程已经发送了正常消息一样。唯一的区别是您不能 对此消息使用MsgReply（） -毕竟，脉冲的整个概念是异步的。在本节中，我们将介绍另一个函数MsgReceivePulse（），该函数对处理脉冲很有用。

关于脉冲的唯一“有趣”的事情是，从*MsgReceive（）*函数返回的接收ID 为零。这表明您是一个脉冲，而不是来自客户端的常规消息。您通常会在服务器中看到如下所示的代码：

#include <sys/neutrino.h>
rcvid = MsgReceive (chid, …);
if (rcvid == 0) {   // it's a pulse
    // determine the type of pulse

    // handle it
} else {            // it's a regular message
    // determine the type of message

    // handle it
}
1
2
3
4
5
6
7
8
9
10
1

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12

脉搏是什么？

好的，因此您收到的接收ID为零的消息。它实际上是什么样的？在<sys/neutrino.h>头文件中，这是_pulse结构的定义：

struct _pulse {
    _uint16         type;
    _uint16         subtype;
    _int8           code;
    _uint8          zero [3];
    union sigval    value;
    _int32          scoid;
};

 
 

 
 
1
2
3
4
5
6
7
8
1
2
3
4
5
6
7
8

两个类型和子类型的成员是零（一个进一步的指示，这是一个脉冲）。将代码和值成员设置为确定的脉冲发送者。通常，该代码将指示为什么发送脉冲。该值将是与脉冲关联的32位数据值。这两个字段是内容“ 40位”的来源；其他字段不是用户可调整的。

内核保留负值的代码，留下127个值供程序员认为合适使用。

该值成员实际上是一个联盟：

union sigval {
    int     sival_int;
    void    *sival_ptr;
};

 
 

 
 
1
2
3
4
1
2
3
4

因此（在上面的服务器示例中进行扩展），您通常会看到类似以下的代码：

#include <sys/neutrino.h>
rcvid = MsgReceive (chid, …

if (rcvid == 0) {   // it's a pulse

    // determine the type of pulse
    switch (msg.pulse.code) {

    case    MY_PULSE_TIMER:
        // One of your timers went off, do something 
        // about it...

        break;

    case    MY_PULSE_HWINT:
        // A hardware interrupt service routine sent 
        // you a pulse.  There's a value in the "value" 
        // member that you need to examine:

        val = msg.pulse.value.sival_int;

        // Do something about it...

        break;

    case    _PULSE_CODE_UNBLOCK:
        // A pulse from the kernel, indicating a client 
        // unblock was received, do something about it...

        break;

    // etc...

} else {            // it's a regular message

    // determine the type of message
    // handle it

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
1

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40

当然，此代码假定您已将msg 结构设置为包含一个struct _pulse pulse;成员，并且已定义清单常量MY_PULSE_TIMER和MY_PULSE_HWINT。脉冲代码_PULSE_CODE_UNBLOCK是上述那些负数内核脉冲之一。您可以找到它们的完整列表<sys/neutrino.h>以及对值字段的简短描述。

所述MsgReceivePulse（） 函数

所述MsgReceive（）和MsgReceivev（） 函数将接收或者“常规”消息或脉冲。在某些情况下，您只想接收脉冲。最好的例子是在服务器上，您已经收到客户端的请求以执行某项操作，但还不能完成请求（也许您必须进行长时间的硬件操作）。在这种设计中，通常会设置硬件（或计时器，或其他任何东西）以在发生重大事件时向您发送脉冲。

如果您使用经典的“在无限循环中等待消息”设计来编写服务器，则可能会遇到以下情况：一个客户端向您发送请求，然后在等待脉冲进入（发出信号）完成请求），另一个客户端向您发送另一个请求。通常，这正是您想要的-毕竟，您希望能够同时为多个客户端提供服务。但是，这可能是不可接受的充分原因-为客户端提供服务可能会占用大量资源，以至于您想限制客户端的数量。

在这种情况下，您现在需要能够“选择性地”仅接收脉冲，而不是常规消息。这是MsgReceivePulse（）起作用的地方：

#include <sys/neutrino.h>
1

int MsgReceivePulse (int chid,
void *rmsg,
int rbytes,
struct _msg_info *info);

• 1
• 2
• 3
• 4
• 5
• 6

如您所见，您使用与MsgReceive（）相同的参数；通道ID，缓冲区（及其大小）以及info参数。（我们在 上面的“ 谁发送了消息？ ”中讨论了info参数。）请注意，在脉冲情况下，不使用info参数；您可能会问为什么它出现在参数列表中。简单的答案：在实现中更容易做到这一点。只需传递一个NULL！

该MsgReceivePulse（）函数将接受什么，但脉冲。因此，如果您有一个通过MsgReceivePulse（）阻止了多个线程的通道（并且没有通过*MsgReceive（）阻止了该线程的线程），并且客户端尝试向您的服务器发送消息，则该客户端将保持SEND-直到线程发出MsgReceive（）调用为止。同时，将通过MsgReceivePulse（）*函数传输脉冲。

如果将MsgReceivePulse（）和MsgReceive（）混合使用，您唯一可以保证的是 MsgReceivePulse（）仅获得脉冲。该MsgReceive（）可以得到脉冲或短信！这是因为，通常，对于希望排除常规消息传递到服务器的情况，保留使用*MsgReceivePulse（）*函数。

这确实带来了一些混乱。由于MsgReceive（）函数可以接收两个消息和脉搏，但MsgReceivePulse（）函数只能接收一个脉冲，你如何处理与服务器，使得使用这两个功能？通常，这里的答案是您将拥有执行MsgReceive（）的线程池。该线程池（一个或多个线程；数量取决于您准备同时服务的客户端数量）负责处理客户端调用（服务请求）。由于您正在尝试控制“服务提供线程”的数量，并且由于其中一些线程可能需要阻塞，等待脉冲到达（例如，来自某些硬件或另一个线程），因此您将通常使用MsgReceivePulse（）阻塞服务提供线程。这样可以确保在等待脉冲时客户端请求不会“潜入”（因为MsgReceivePulse（）将仅接收脉冲）。

所述MsgDeliverEvent（） 函数

如上文“发送层次结构”中所述，在某些情况下，您需要中断发送的自然流程。

如果您有一个客户端向服务器发送了消息，则这种情况可能会发生，结果可能暂时无法使用，并且该客户端不想阻止。当然，您也可以使用线程来部分解决此问题，方法是让客户端简单地在阻塞服务器调用上“用完”一个线程，但是这可能不适用于大型系统（在这种情况下，您将使用大量线程等待许多不同的服务器）。假设您不想使用线程，而是希望服务器立即回复客户端，“我很快就会处理您的请求。” 此时，由于服务器已答复，因此客户端现在可以自由继续处理。一旦服务器完成了客户端提供的任务，服务器现在需要某种方式告诉客户端：“嘿，醒了，我完成了。” 明显，正如我们在上面的“发送层次结构”讨论中所看到的，您不能让服务器向客户端发送消息，因为如果客户端在同一时刻向服务器发送消息，这可能会导致死锁。因此，服务器如何“发送”消息给客户端而不会违反发送层次结构？

它实际上是一个多步骤操作。运作方式如下：

1. 客户创建一个struct sigevent结构，并填写它。
2. 客户端向服务器发送一条消息，有效地表明：“为我执行此特定任务，立即回复，顺便说一下，这struct sigevent是您应该在工作完成时通知我的。”
3. 服务器接收到该消息（包括struct sigevent），将struct sigevent和接收到的ID 存储起来，并立即回复客户端。
4. 客户端和服务器现在都在运行。
5. 服务器完成工作后，服务器使用MsgDeliverEvent（）通知客户端工作现已完成。

我们将在“如何填写时钟 ”一章struct sigevent中的“ 时钟，计时器和经常发动”一章中详细介绍struct sigevent。现在，只需将其struct sigevent视为“黑匣子”，其中就包含了服务器用来通知客户端的事件。

由于服务器存储了struct sigevent来自客户端的ID和接收ID，因此服务器现在可以调用MsgDeliverEvent（）**将客户端选择的事件传递给客户端：

int
MsgDeliverEvent (int rcvid,
                 const struct sigevent *event);

 
 

 
 
1
2
3
1
2
3

请注意，MsgDeliverEvent（）函数采用两个参数，即接收ID（在rcvid中）和在event中传递的事件。服务器不会 以任何方式修改或检查事件 ！这一点很重要，因为它允许服务器传递客户端选择的任何类型的事件，而无需服务器方面的任何特定处理。（但是，服务器可以 使用MsgVerifyEvent（）函数来验证事件是否有效。）

该rcvid是接收ID，服务器从客户端得到。请注意，这确实是一种特殊情况。通常，在服务器回复客户端后，接收ID不再具有任何含义（原因是客户端已解除阻止，服务器无法再次解除阻止，或者无法 从客户端读取数据或向客户端写入数据，等等。）。但是在这种情况下，接收ID仅包含足够的信息供内核使用，以决定事件应该传递给哪个客户端。当服务器调用*MsgDeliverEvent（）*函数时，服务器不会阻塞-这是服务器的非阻塞调用。客户端将事件（由内核）传递给它，然后可以执行任何适当的操作。

频道标志

当我们引入服务器时（在“ 服务器 ”中），我们提到ChannelCreate（）函数带有一个flags 参数，而我们将其保留为零。

现在该解释标志了。我们将仅检查一些可能的标志值：

• _NTO_CHF_FIXED_PRIORITY

  接收线程不会根据发送者的优先级更改优先级。（我们在下面的“ 优先级继承 ”部分中详细讨论了优先级问题）。通常（即，如果您不指定此标志），则接收线程的优先级更改为发送方的优先级。

• _NTO_CHF_UNBLOCK

  每当客户端线程尝试解除阻止时，内核都会发送一个脉冲。服务器必须回复客户端以便允许客户端解除阻止。我们将在下面讨论这一问题，因为它对客户端和服务器都有一些非常有趣的后果。

• _NTO_CHF_THREAD_DEATH

  每当在此通道上阻塞的线程死亡时，内核都会发送一个脉冲。这对于希望始终保持 可用于服务请求的固定“线程池”的服务器很有用。

• _NTO_CHF_DISCONNECT

  只要来自单个客户端的所有连接都与服务器断开连接，内核都会发出脉冲。

• _NTO_CHF_SENDER_LEN

  内核将客户端的消息大小作为提供给服务器的信息的一部分（结构的srcmsglen成员struct _msg_info）传递。

• _NTO_CHF_REPLY_LEN

  内核将客户端的回复消息缓冲区大小作为提供给服务器的信息的一部分（结构的dstmsglen成员 struct _msg_info）传递。

• _NTO_CHF_COID_DISCONNECT

  每当此进程拥有的任何连接由于另一端的通道消失而终止时，内核都会发出脉冲。

_NTO_CHF_UNBLOCK

让我们看一下_NTO_CHF_UNBLOCK标志；对于客户端和服务器，它都有一些有趣的折痕。

通常（当服务器未指定_NTO_CHF_UNBLOCK标志时），当客户端希望从MsgSend（）（以及相关的MsgSendv（），MsgSendvs（）等功能系列中）取消阻止时，客户端只会取消阻止 。客户端可能希望由于接收到信号或内核超时而解除阻塞（请参阅Neutrino库参考中的TimerTimeout（）函数，以及“ 时钟，计时器和时常获得踢动”一章）。不幸的是，服务器不知道客户端已解除阻止，并且不再等待答复。请注意，除非在非常特殊的情况下需要服务器与其所有客户端之间的配合，否则无法在禁用此标志的情况下编写可靠的服务器。

假设您有一台具有多个线程的服务器，所有线程都被该服务器的MsgReceive（）函数阻塞。客户端向服务器发送一条消息，服务器的线程之一接收到该消息。此时，客户端被阻止，服务器中的线程正在积极处理请求。现在，在服务器线程有机会回复客户端之前，客户端会解除对MsgSend（）的阻止（假设它是由于信号引起的）。

请记住，服务器线程仍在代表客户端处理请求。但是，由于客户端现在已解除阻塞（客户端的MsgSend（）会随EINTR返回），因此客户端可以自由地向服务器发送另一个请求。由于Neutrino服务器的体系结构，另一个线程将从客户端接收另一条消息，并且接收ID完全相同！服务器无法区分这两个请求！当第一个线程完成并回复客户端时，它实际上是在响应客户端发送的 第二条消息，而不是第一条消息（因为该线程实际上认为自己在做）。因此，服务器的第一个线程答复客户端的第二个消息。

这已经够糟糕了；但让我们更进一步。现在，服务器的第二个线程完成了请求，并尝试回复客户端。但是由于服务器的第一个线程已经回复了客户端，因此客户端现在已解除阻塞，服务器的第二个线程从其答复中得到了错误。

此问题仅限于多线程服务器，因为在单线程服务器中，服务器线程仍将忙于处理客户端的第一个请求。这意味着即使客户端现在已解除阻止并再次发送到服务器，客户端现在也将进入SEND阻止状态（而不是REPLY阻止状态），从而允许服务器完成处理并回复客户端（这将导致错误，因为客户端不再被REPLY阻止），然后服务器将从客户端接收第二条消息。真正的问题在于服务器正在代表客户端执行无用的处理（客户端的第一个请求）。该处理无用，因为客户端不再等待该工作的结果。

解决方案（在多线程服务器的情况下）是让服务器为其ChannelCreate（）调用指定_NTO_CHF_UNBLOCK标志。这对内核说：“当客户端尝试解除对我的阻止时（通过向我发送脉冲）告诉我，但不要让客户端解除阻止！我会亲自阻止客户。”

要牢记的关键是，这个服务器标志更改客户端通过行为不是允许客户机疏通，直到 服务器的说，这是好这样做。

在单线程服务器中，发生以下情况：

这并没有帮助客户疏通时，它应该有，但它并 确保服务器没有得到混淆。在这种示例中，服务器很可能会简单地忽略它从内核获得的脉冲。可以这样做-这里所做的假设是，让客户端阻塞直到服务器准备好数据是安全的。

如果希望服务器根据内核发送的脉冲进行操作，则有两种方法可以执行此操作：

• 在服务器中创建另一个侦听消息的线程（特别是侦听来自内核的脉冲）。第二个线程将负责取消第一个线程中正在进行的操作。两个线程之一将回复客户端。
• 不要在线程本身中完成客户的工作，而要把工作排队。这通常在服务器将客户端的工作存储在队列中并且服务器是事件驱动的应用程序中完成。通常，到达服务器的消息之一表明客户端的工作现在已经完成，服务器应该回复。在这种情况下，当内核脉冲到达时，服务器将代表客户端取消正在执行的工作并进行答复。

选择哪种方法取决于服务器执行的工作类型。在第一种情况下，服务器正在代表客户端积极地执行工作，因此您真的没有选择–您必须有第二个线程来侦听来自内核的取消阻塞脉冲（否则您可以定期在线程内进行轮询，以查看是否有脉冲到达，但通常不建议进行轮询）。

在第二种情况下，服务器还有其他工作要做–也许已经命令了一块硬件来“收集数据”。在这种情况下，无论如何，服务器的线程都会在*MsgReceive（）*函数上被阻塞，等待硬件指示命令已完成。

无论哪种情况，服务器都必须回复客户端，否则客户端将保持阻塞状态。

同步问题

即使您如上所述使用_NTO_CHF_UNBLOCK标志，仍然有另一个同步问题要处理。假设您在MsgReceive（）函数上阻塞了多个服务器线程，等待消息或脉冲，并且客户端向您发送消息。一个线程关闭，开始客户的工作。发生这种情况时，客户端希望解除阻塞，因此内核会生成解除阻塞脉冲。服务器中的另一个线程接收到此脉冲。此时，存在竞争条件–第一个线程可能正好在 准备回复客户。如果第二个线程（得到了脉冲）做了答复，那么客户端就有可能解除阻塞并向服务器发送另一条消息，而服务器的第一个线程现在有机会运行并回复客户端的第二个请求。第一个请求的数据：

多线程服务器中的混乱。

或者，如果获得脉冲的线程正要答复客户端，而第一个线程做了答复，那么您将遇到相同的情况-第一个线程解除对客户端的阻塞，该客户端发送另一个请求，而第二个线程（现在获得了响应）现在解除阻止客户的第二个请求。

情况是您有两个并行的执行流程（一个是由消息引起的，另一个是由脉冲引起的）。通常，我们会立即意识到这是需要互斥的情况。不幸的是，这会引起问题-互斥锁必须在*MsgReceive（）之后立即获取，并在MsgReply（）*之前释放。虽然这确实有效，但它破坏了解锁脉冲的全部目的！（服务器将获得消息并忽略取消阻止脉冲，直到它回复给客户端为止，否则服务器将获得取消阻止脉冲并取消客户端的第二个操作。）

看起来很有希望（但最终注定要失败）的解决方案是拥有细粒度的互斥体。我的意思是一个互斥锁仅在控制流的一小部分被锁定和解锁（您应该使用互斥锁的方式，而不是如上所述阻止整个处理部分）。您可以设置“我们回复了吗？” 标记在服务器中，并且在收到消息时将清除此标志，并在答复消息时将其设置。在您回复邮件之前，您需要检查该标志。如果该标志指示该邮件已被回复，则可以跳过回复。互斥锁将在标志的检查和设置周围被锁定和解锁。

不幸的是，这将无法正常工作，因为我们并不总是处理两个并行的执行流程-客户端在处理过程中不会总是受到信号的干扰（导致阻塞脉冲）。这是它打破的情况：

• 客户端向服务器发送消息；客户端现在被阻止，服务器正在运行。
• 由于服务器收到了来自客户端的请求，因此该标志被重置为0，表明我们仍然需要回复客户端。
• 服务器通常会回复客户端（因为该标志被设置为0），并且将该标志设置为1，表示如果出现了疏通脉冲，则应将其忽略。
• *（问题从这里开始。）*客户端向服务器发送第二条消息，并且在发送后几乎立即被信号击中。内核将解除阻塞脉冲发送到服务器。
• 接收消息的服务器线程将要获取互斥体以检查该标志，但并没有完全到达那里（它被抢占了）。
• 现在，另一个服务器线程获取了该脉冲，并且由于该标志自上次以来仍设置为1，因此将忽略该脉冲。
• 现在，服务器的第一个线程获取互斥锁并清除该标志。
• 此时，取消阻止事件已丢失。

如果您优化标志以指示更多状态（例如收到的脉冲，收到的脉冲，收到的消息，回复的消息），则仍然会遇到同步竞争状态，因为无法在两者之间创建原子绑定标志以及接收和回复功能调用。（从根本上讲，问题就出在这里-在MsgReceive（）之后和调整标志之前，以及在*MsgReply（）*之前调整标志之后的小计时窗口 。）解决此问题的唯一方法是让内核为您跟踪标志。

使用_NTO_MI_UNBLOCK_REQ

幸运的是，内核在消息信息结构中（作为struct _msg_info您最后一个参数传递给MsgReceive（），或者您可以稍后在给定接收ID的情况下，通过调用MsgInfo（ ））。

此标志称为_NTO_MI_UNBLOCK_REQ，如果客户端希望解除阻止（例如，在接收到信号之后），则设置该标志。

这意味着在多线程服务器中，通常会有一个“工作程序”线程正在执行客户端的工作，而另一个线程将要接收取消阻止消息（或其他消息；我们只关注针对现在）。当您从客户端收到取消阻止消息时，您将为自己设置一个标志，让程序知道线程希望取消阻止。

有两种情况需要考虑：

• “工人”线程被阻止；要么
• “工人”线程正在运行。

如果工作线程被阻止，则需要使获取取消阻止消息的线程唤醒它。例如，如果它正在等待资源，则可能被阻止。当工作线程唤醒时，它应检查_NTO_MI_UNBLOCK_REQ标志，如果设置了该标志，则以中止状态答复。如果未设置该标志，则线程在唤醒后可以执行其正常处理。

或者，如果工作线程正在运行，则应定期检查解锁线程可能已设置的“自我标志”，如果设置了标志，则应以中止状态回复客户端。请注意，这只是一种优化：在未优化的情况下，工作线程将不断在接收ID上调用“ MsgInfo”，并检查_NTO_MI_UNBLOCK_REQ位本身。

通过网络传递消息

为了使事情更清楚，尽管这是Neutrino灵活性的关键部分，但我避免谈论您如何使用通过网络传递消息的方式！

到目前为止，您学到的所有内容都适用于通过网络传递的消息。

在本章的前面，我向您展示了一个示例：

#include <fcntl.h>
#include <unistd.h>
1
2

int
main (void)
{
int fd;

fd = open ("/net/wintermute/home/rk/filename", O_WRONLY);
write (fd, "This is message passing\n", 24);
close (fd);

return (EXIT_SUCCESS);
1
2
3
4
5

}

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14

当时，我说这是“使用通过网络传递消息”的示例。客户端创建到ND / PID / CHID的连接（恰好在另一个节点上），服务器在其通道上执行MsgReceive（）。在这种情况下，客户端和服务器与本地单节点情况相同。您可以在这里停止阅读-确实没有关于通过网络传递消息的任何“棘手”信息。但是对于那些对如何操作感到好奇的读者，请继续阅读！

现在，我们已经了解了本地消息传递的一些细节，我们可以更深入地讨论通过网络传递消息的工作方式。尽管此讨论可能看起来很复杂，但实际上可以分为两个阶段：名称解析，一旦解决，就简单地传递消息。

以下图表说明了我们将要讨论的步骤：

通过网络传递的消息。请注意，Qnet分为两个部分。

在该图中，我们的节点称为magenta，并且如示例所示，目标节点称为wintermute。

让我们分析客户端程序使用Qnet通过网络访问服务器时发生的交互：

1. 告知客户端的*open（）*函数打开恰好/net位于其前面的文件名。（该名称/net是Qnet表示的默认名称。）此客户端不知道谁负责该特定的路径名，因此它连接到流程管理器（步骤1），以找出谁真正拥有该资源。无论我们是否通过网络传递消息，此操作都会自动完成。由于本机Neutrino网络管理器Qnet“拥有”所有以开头的路径名/net，因此流程管理器将信息返回给客户端，告诉客户端询问该路径名的Qnet。
2. 客户端现在将消息发送到Qnet的资源管理器线程，希望Qnet将能够处理该请求。但是，此节点上的Qnet不负责提供客户端所需的最终服务，因此它告诉客户端它实际上应该与node上的流程管理器联系wintermute。（完成此操作的方式是通过“重定向”响应，它向客户端提供应与其联系的服务器的ND / PID / CHID。）该重定向响应也由客户端库自动处理。
3. 客户端现在连接到上的流程管理器wintermute。这涉及通过Qnet的网络处理程序线程发送节点外消息。客户端节点上的Qnet进程获取消息并将其通过介质传输到远程Qnet，然后将其传递到上的进程管理器wintermute。那里的流程管理器解析路径的其余部分（在我们的示例中，这将是“/home/rk/filename”部分）并发送回重定向消息。该重定向消息遵循反向路径（从服务器的Qnet通过介质到客户端节点上的Qnet，最后回到客户端）。现在，此重定向消息包含客户端首先要联系的服务器的位置，即将用于服务于客户端请求的服务器的ND / PID / CHID。（在我们的示例中，服务器是一个文件系统。）
4. 客户端现在将请求发送到该服务器。除了直接与服务器联系而不是通过流程管理器联系之外，此处遵循的路径与上述步骤3中遵循的路径相同。

一旦建立了步骤1到步骤3，步骤4就成为所有将来通信的模型。在上面的客户端示例中，open（），read（）和close（） 消息均采用路径号4。请注意，客户端的open（） 首先触发了此事件序列的发生，但实际发生的是开放消息流如所述（通过路径号4）流动。

我们将返回“ 资源管理器”一章中用于open（），read（）和close（）（及其他）的消息。

网络邮件传递差异

因此，一旦建立连接，所有进一步的消息传递将使用上图中的步骤4进行。这可能会导致您错误地认为，通过网络传递的消息与在本地情况下传递的消息相同。不幸的是，这不是真的。区别如下：

• 更长的延迟
• 无论节点是否处于活动状态，ConnectAttach（）都会返回成功-真正的错误指示出现在第一次消息传递中
• *MsgDeliverEvent（）*不能保证可靠
• MsgReply（），MsgRead（），*MsgWrite（）*现在阻止了调用，而在本地情况下则不会
• *MsgReceive（）可能不会接收到客户端发送的所有数据。服务器可能需要调用MsgRead（）*来获取其余信息。

更长的延迟

由于消息传递现在是通过某种介质完成的，而不是直接由内核控制的内存到内存的副本进行的，因此可以预期，传输消息所花费的时间会大大增加（100 MB以太网与100 MHz 64-位宽的DRAM将慢一个数量级或两个）。另外，最重要的是协议开销（最小）和有损网络上的重试。

对ConnectAttach（）的影响

调用ConnectAttach（）时，将指定ND，PID和CHID。在Neutrino中发生的所有事情就是内核将连接ID返回到上图所示的Qnet“网络处理程序”线程。由于未发送任何消息，因此不会通知您刚刚连接到的节点是否仍处于活动状态。在正常使用中，这不是问题，因为大多数客户端将不会执行自己的ConnectAttach（）；相反，他们将使用库调用open（）的服务，该服务先执行ConnectAttach（），然后执行几乎立即发出“打开”消息。这具有几乎立即指示远程节点是否处于活动状态的作用。

对MsgDeliverEvent（）的影响

当服务器在本地调用MsgDeliverEvent（）时，由内核负责将事件传递到目标线程。通过网络，服务器仍然调用MsgDeliverEvent（），但是内核将事件的“代理”传递给Qnet，由Qnet负责将代理传递给另一个（客户端）Qnet，然后由后者传递给客户的实际事件。由于*MsgDeliverEvent（）函数的调用是非阻塞的，因此事情可能会在服务器端解决。这意味着，一旦服务器调用了MsgDeliverEvent（），它就会运行。转身说：“我讨厌告诉你，但是你知道我说的MsgDeliverEvent（）*成功了吗？好吧，不是！”

对MsgReply（），MsgRead（）和MsgWrite（）的影响

为了防止我刚才提到的MsgDeliverEvent（）问题与MsgReply（），MsgRead（）和MsgWrite（）一起发生，当在网络上使用时，这些函数被转换为阻塞调用。在本地，他们只需传输数据并立即取消阻止。在网络上，我们必须（对于MsgReply（））确保已将数据传递到客户端，或者（对于其他两个，则）确保通过网络实际向客户端或从客户端传输数据。

对MsgReceive（）的影响

最后，MsgReceive（）也受到影响（在网络情况下）。当服务器的*MsgReceive（）*解除阻止时，Qnet可能不会通过网络传输所有客户端的数据。这样做是出于性能原因。

struct _msg_info作为最后一个参数传递给MsgReceive（）的中有两个标志（我们在上面的“ 谁发送了消息？ ” 中详细了解了此结构）：

• msglen

  指示*MsgReceive（）*实际传输了多少数据（Qnet喜欢传输8 KB）。

• srcmsglen

  指示客户端要传输多少数据（由客户端确定）。

因此，如果客户端希望通过网络传输1兆字节的数据，则服务器的MsgReceive（）将解除阻止，并将msglen设置为8192（表示缓冲区中有8192字节可用），而srcmsglen 将设置为1048576（表示客户端尝试发送1兆字节）。

然后，服务器使用*MsgRead（）*从客户端的地址空间中获取其余数据。

关于ND的一些注意事项

关于消息传递，我们还没有谈到的另一个“有趣”的事情是“节点描述符”或简称为“ ND”的整个业务。

回想一下，我们/net/wintermute在示例中使用了符号节点名称。在QNX 4（Neutrono之前的OS的早期版本）下，本机联网基于节点ID的概念，该ID是网络上唯一的小整数。因此，我们将讨论“节点61”或“节点1”，这反映在函数调用中。

在Neutrino下，所有节点在内部都由32位数量引用，但这并不是网络唯一的！我的意思是，它wintermute可能被认为spud是节点描述符编号“ 7”，同时spud 也可能被认为magenta是节点描述符编号“ 7”。让我扩大一下范围，以便为您提供更好的画面。该表显示，可能由三个节点使用一些样品节点描述符， wintermute，spud，和foobar：

请注意，每个节点自身的节点描述符如何为零。另请注意 wintermute，的节点描述符的spud值为“ 7”，与foobar的节点描述符一样spud。但是wintermute“的节点描述符为foobar” 4”，而 spud“的节点描述符为foobar” 6”。就像我说的那样，尽管它们在每个节点上都是唯一的，但它们在网络上并不是唯一的。您可以有效地将它们视为文件描述符-如果两个进程访问相同的文件，则它们可能具有相同的文件描述符，但是它们可能没有相同的文件描述符。这仅取决于谁在何时打开哪个文件。

幸运的是，由于多种原因，您不必担心节点描述符：

1. 通常，您将要进行的大多数节点外消息传递都是通过更高级别的函数调用（如上例所示的open（））。
2. 节点描述符不被缓存-如果您得到一个，则应该立即使用它，然后再忽略它。
3. 有一些库调用可将路径名（如/net/magenta）转换为节点描述符。

要使用节点描述符，您将要包含该文件，<sys/netmgr.h>因为它包含一堆*netmgr _ *（）*函数。

您可以使用函数netmgr_strtond（）将字符串转换为节点描述符。一旦有了该节点描述符，就可以在ConnectAttach（）函数调用中立即使用它。特别是，您永远不应将其缓存在数据结构中！原因是，与该特定节点的所有连接都断开连接后，本机网络管理器可能决定重用它。因此，如果您获得的节点描述符为“ 7” /net/magenta，并连接到该节点，发送了一条消息，然后断开连接，则本机网络管理器可能会为另一个节点再次返回节点描述符“ 7”。 。

由于节点描述符在每个网络中不是唯一的，因此出现的问题是：“如何在网络中传递这些信息？” 显然，magenta关于节点描述符“ 7”的观点与根本不同wintermute。这里有两种解决方案：

• 不要绕过节点描述符；请使用符号名称（例如/net/wintermute）代替。
• 使用netmgr_remote_nd（）函数。

第一个是好的通用解决方案。第二种解决方案使用起来相当简单：

int
netmgr_remote_nd (int remote_nd, int local_nd);

 
 

 
 
1
2
1
2

该函数有两个参数：remote_nd是目标计算机的节点描述符，而local_nd是节点描述符（从本地计算机的角度来看）要转换为远程计算机的角度。结果是从远程计算机的角度来看有效的节点描述符。

例如，假设wintermute是我们的本地计算机。我们有一个在本地计算机上有效的节点描述符“ 7”，它指向magenta。我们想找出的是节点描述符magenta用来与我们交谈的内容：

int     remote_nd;
int     magenta_nd;
1
2

magenta_nd = netmgr_strtond (“/net/magenta”, NULL);
printf (“Magenta’s ND is %d\n”, magenta_nd);
remote_nd = netmgr_remote_nd (magenta_nd, ND_LOCAL_NODE);
printf (“From magenta’s point of view, we’re ND %d\n”,
remote_nd);

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8

这可能会打印类似于以下内容的内容：

Magenta's ND is 7
From magenta's point of view, we're ND 4

 
 

 
 
1
2
1
2

这表示在上 magenta，节点描述符“ 4”是指我们的节点。（请注意，使用特殊常量ND_LOCAL_NODE（实际上为零）来表示“此节点”。）

现在，回想一下，我们说过（在“谁发送了消息？”中），其中struct _msg_info包含两个节点描述符：

struct _msg_info
{
    int     nd;
    int     srcnd;
    …
};

 
 

 
 
1
2
3
4
5
6
1
2
3
4
5
6

我们在说明中针对这两个字段指出：

• nd是发送节点的接收节点的节点描述符
• srcnd是接收节点的发送节点的节点描述符

因此，对于上面的示例，当向我们发送消息（）时，wintermute本地节点和magenta远程节点在哪里，我们期望：magenta``wintermute

• nd将包含7
• srcnd将包含4。

优先继承

实时操作系统中有趣的问题之一是被称为优先级反转的现象。

优先级转换本身表现为，例如，一个低优先级线程消耗了所有可用的CPU时间，即使优先级较高的线程已准备好运行。

现在您可能在想：“等等！您说过，较高优先级的线程将始终抢占较低优先级的线程！怎么会这样？”

的确如此—较高优先级的线程将始终抢占较低优先级的线程。但是可能会发生一些有趣的事情。让我们看一下这样一个场景：我们有三个线程（在三个不同的进程中，为简单起见），“ L”是我们的低优先级线程，“ H”是我们的高优先级线程，“ S”是服务器。此图显示了三个线程及其优先级：

三个线程处于不同的优先级。

当前，H正在运行。S，一个更高优先级的服务器线程，目前无事可做，因此它正在等待消息，并在MsgReceive（）中被阻止。L想要运行，但优先级低于运行的H。一切都如您所愿，对吗？

现在，H决定要休眠100毫秒-也许它需要等待一些慢速的硬件。此时，L正在运行。

这就是事情变得有趣的地方。

作为其正常操作的一部分，L向服务器线程S发送一条消息，使S进入READY状态，并且（因为它是READY优先级最高的线程）开始运行。不幸的是，L发送给S的消息是“将pi计算为5000个小数位。”

显然，这需要100毫秒以上的时间。因此，当H的100毫秒上升并且H变为READY时，您猜怎么着？它不会运行，因为S是READY且具有更高的优先级！

发生的事情是，低优先级线程通过甚至更高优先级的线程来利用CPU，从而阻止了高优先级的线程运行。这是优先级倒置。

要解决此问题，我们需要讨论优先级继承。一个简单的解决方法是让服务器S 继承客户端线程的优先级：

线程阻塞。

在这种情况下，当H的100毫秒睡眠完成时，它将进入READY状态，并且由于它是优先级最高的READY线程而运行。

还不错，但是还有一个“陷阱”。

假设H现在决定也要执行计算。它想计算第5,034个素数，因此它向S和块发送一条消息。

但是，S仍在计算pi，优先级为5！在我们的示例系统中，有很多其他优先级高于5的线程正在使用CPU，这有效地确保了S 没有太多时间来计算pi。

这是优先级倒置的另一种形式。在这种情况下，优先级较低的线程阻止了优先级较高的线程访问资源。与此形成对比的是第一种优先级倒置形式，优先级较低的线程有效地消耗了 CPU —在这种情况下，它只是防止较高优先级的线程获取CPU —而不消耗任何CPU本身。

幸运的是，这里的解决方案也相当简单。提高服务器的优先级，使其成为所有阻止的客户端中最高的：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-J7nSkqJI-1595729252682)(…/…/…/…/…/…/…/…/pic/image-20200724164527815.png)]

提高服务器的优先级。

这样，通过让L的作业以高于L的优先级运行，我们受到了较小的打击，但是我们确实确保H在CPU上获得公平的破解。

那么诀窍是什么？

没有把戏！Neutrino会自动为您执行此操作。（如果不需要，可以关闭优先级继承；请参见ChannelCreate（）函数的文档中的_NTO_CHF_FIXED_PRIORITY标志。）

但是，这里存在一个较小的设计问题。您如何将优先级恢复为更改前的优先级？

您的服务器一直在运行，为来自客户端的请求提供服务，并在取消阻止MsgReceive（） 调用时自动调整其优先级。但是，何时应该将其优先级调整*回**MsgReceive（）*调用更改其优先级之前的优先级呢？

有两种情况需要考虑：

• 服务器在适当地为客户端提供服务之后，将执行一些其他处理。这应该以服务器的优先级而不是客户端的优先级完成。
• 服务器立即执行另一个*MsgReceive（）*来处理下一个客户端请求。

在第一种情况下，当服务器不再为该客户端工作时，以该客户端的优先级运行该服务器是不正确的！解决方案非常简单。使用pthread_setschedparam（）函数（在“ 进程和线程”一章中讨论）将优先级恢复为应有的优先级。

那另一种情况呢？答案很简单：谁在乎？

想一想。如果服务器在优先级为29时与优先级为2时成为接收阻止状态，那会有什么区别？事实是，它被接收阻止了！它没有占用任何CPU时间，因此其优先级无关紧要。一旦*MsgReceive（）*函数解除对服务器的阻止，服务器就会继承（新）客户端的优先级，并且一切都会按预期进行。

摘要

消息传递是一个非常强大的概念，并且是Neutrino（实际上是过去的所有QNX操作系统）所基于的主要功能之一。

通过消息传递，客户端和服务器交换消息（同一进程中的线程到线程，同一节点上的不同进程中的线程到线程，或网络中不同节点上的不同进程中的线程到线程）。客户端发送一条消息并阻止，直到服务器接收到该消息，对其进行处理并回复该客户端为止。

消息传递的主要优点是：

• 消息的内容不会根据目的地的位置（本地与网络）而改变。
• 一条消息为客户端和服务器提供了一个“干净”的解耦点。
• 隐式同步和序列化有助于简化应用程序的设计。

[！[上一个]]（prev.gif） [！[内容]]（contents.gif） [！[索引]]（keyword_index.gif） [！[[下一个]]（next.gif）](