一切为了并行：MS Axum语言教程 <二>

代理编程

上面的斐波那契数列的例子只是展示了一个微不足道的构建数据流网的例子。这类网络适合那些“数据流入，数据流出”的场景，但它不能指定数据如何在网络中传播，也不允许不同类型的数据流入或者流出网络。

代理与通道给了我们创建复杂数据流网所需的一切。

通过通道沟通的两个代理是种弱关系：它不需要知道或关系另一个代理是如何实现的。它们之间的“契约”只由通道指定。借用OOP中的概念，通道扮演了接口的角色，而代理则是实现了该接口的类。


使用通道时，你向输入端口发送数据，从输出端口接受数据，这里，输入端口扮演了目标，而输出端口扮演了源头。而实现通道时，输入端口则是源头，输出端口则是目标。这有点难以理解，想想家里的水处理机，如果水是数据的话，考虑水的流向，你把水倒入入水口（水相对于你是流出），从出水口接水（水相对于你是流入），而处理机内部而言，水则是从进水口流入，从出水口流出。


请看例子，有个“Adder”通道接受输入两个数字并得出二者之和。通道的使用者发送数字到输入端口Num1和Num2，并从输出端口Sum接收结果。MainAgent代理是通道的使用者，而AdderAgent实现了该通道。

channel Adder { 
	input int Num1; 
	input int Num2; 
	output int Sum; 
} 

agent AdderAgent : channel Adder 
{ 
	public AdderAgent() 
    {
        int result = receive(PrimaryChannel::Num1) + receive(PrimaryChannel::Num2);
        PrimaryChannel::Sum <-- result;
    }
} 

agent MainAgent : channel Microsoft.Axum.Application { 
    public MainAgent() {
        var adder = AdderAgent.CreateInNewDomain();
        adder::Num1 <-- 10;
        adder::Num2 <-- 20;
        // do something useful ...
        var sum = receive(adder::Sum);
        Console.WriteLine(sum); 
        PrimaryChannel::ExitCode <-- 0; 
    } 
}

通道用关键字“channel”定义，端口则用关键字“input”和“output"定义。如果你仔细读过前面的文字，这些代码应该不难理解，不再赘述。

Schemas

不是所有的数据都能用通道来传送，原因有二：

1.在分布式场景中，代理可能存在两个不同进程中，甚至是两台机器上，传送的数据都必须被系列化。

2.Axum的孤立模式要求两个代理并发执行时不存取共享的可变状态。要满足这个条件，消息中的数据必须不是共享或可变的，否则代理不能并发执行。

为了满足这些要求，Axum引入了schema的概念，Schema是种类型，用于定义必需的或可选的字段，但不包含方法。这个概念与XML schema概念非常相似，或着说有点类似C#中的struct.

当消息中的schema的实例越过进程或者计算机的边界时，数据的有效载荷被深度系列化，这意味着你只能在schema中定义那些能用.NET-serialization系列化的字段。当消息在进程内传送的时候，则不需要深度系列化，但schema中的字段会被视为不可变的。Schema用关键字“schema”定义，随后跟着schema的名称：

schema Customer 
{ 
  required string Name; 
  optional string Address; 
}

“必需”与“可选”字段的区别允许不同schema之间，或者同形（same shape)的schema和类型之间的松耦合。同形意味着schema必须有相同的“必需”字段，但不要求必须有相同的“可选”字段。

当沟通各方未使用相同类型的schema的时候，这就很有用了。当系统的一个组件更新成新版本，schema发生了变化，但其仍可以被转换或强制成另一种近似的类型以便与其他组件继续通讯。比如这个C#类

class CustomerData { public string Name; }
Customer c = new Customer{Name = "Artur", Address = "One Microsoft Way"}; 
CustomerData cd = coerce<CustomerData>(c);

如果编译时，编译器知道这种强制转换会失败的话，Axum编译器将报错，反之，编译器将产生代码，但这些代码可能在运行时报错的。schema还可以将许多不同的数据类型包装成一个单一的包装类型以便在一个消息中一次发送。

请求/回答端口

回顾上面的例子,它只能接受一次请求，之后代理AdderAgent就关闭了。我们可以修改AdderAgent让它继续接受新的起请求直到满足某些条件才关闭。这是个好的开始，不过如果我们希望用户能够提交多个请求，并按发送顺序接受这些请求，这该怎么办？为此，我们需要某种方法来关键请求与AdderAgent产生的结果。一个方法是返回个收据给客户端,客户端稍后凭此来认领结果。在Axum种，这种收据被称为请求关联（request correlator），支持请求关联的端口则成为请求/回答端口（request-reply ports）.我们接下来修改上面的例子

using System; 
using System.Concurrency; 
using Microsoft.Axum; 

schema Pair { 
   required int Num1; 
   required int Num2; 
} 

channel Adder 
{ 
    input Pair Nums : int; // input request-reply port
} 

agent AdderAgent : channel Adder { 
  
  public AdderAgent() { 
    while(true) { 
        var result = receive(PrimaryChannel::Nums); 
        result <-- result.RequestValue.Num1 + result.RequestValue.Num2; 
     } 
  } 
}

agent MainAgent : channel Microsoft.Axum.Application { 
   
  public MainAgent() { 
      var adder = AdderAgent.CreateInNewDomain(); 
      var correlator1 = adder::Nums <-- new Pair{ Num1=10, Num2=20 };
      var correlator2 = adder::Nums <-- new Pair{ Num1=30, Num2=40 }; // do something useful here ... 
      var sum1 = receive(correlator1); 
      Console.WriteLine(sum1); 
      var sum2 = receive(correlator2); Console.WriteLine(sum2); 
      PrimaryChannel::ExitCode <-- 0; 
   } 
}

正如你所看到的，向Adder提交请求同时产生了请求关联(correlator1 and correlator2),以用来取得结果.(很自然的，用receive表达式).

在例子中，因为我们提交了多个请求，当这些请求被处理的时候，我们有更多时间完成其他事。

协议
回顾下上上个例子，如果Adder使用不正确，AdderAgent和MainAgent会彼此等待对方的消息，导致死锁。如果用户在向Num2端口发送数据前，忘记给Num1发送数据，这时会发生什么？MainAgent继续企图读取结果，但同时AdderAgent能继续等待来自端口Num1的消息。

现在我们遇上了一个经典的死锁：MainAgent等待来自AdderAgent的消息，而AdderAgent又在等待来自MainAgent的消息。谁也没法继续执行。

直觉上我们知道第一条消息应该假定发送给Num1，跟着的第二条消息则应该是给Num2的。把我们的直觉形式化下来，就是通道该如何按照通道的协议工作了。协议是有限状态机，带有状态和通道设计者定义的状态间的变换。协议开始通常带有一个特殊的状态“开始。当心的消息到达通道的任何端口，协议将变换成任一协议定义的有效状态或触发一个违背协议异常，如果无任何有效状态存在。

我们将展示如何给通道Adder写一个协议。

channel Adder { 
  input int Num1; 
  input int Num2; 
  output int Sum; 
  Start: { Num1 -> GotNum1; } 
  GotNum1: { Num2 -> GotNum2; } 
  GotNum2: { Sum -> End; }
 }

开始协议带有状态“Start”，当端口Num1收到一条消息时，状态变换成GotNum1，如果其他端口在这时得到消息将导致违反协议异常。接下来，状态变成GotNum1，下个被使用的端口应该是Num1，它将触发状态变迁成GotNum2.最后端口Sum上的消息将触发变迁成内建的状态"End"。这时，协议被关闭了，任何再试图向端口发送消息将触发异常。

你可以定义更详细描述的变迁来使协议更丰富，比如可以“当消息到达端口X或Y，变迁到状态Y”或者“当消息的值大于10的时候，变迁到状态X"等等。为简单起见，我们不做详述，你可参考Axum的规格书。

我们试着改变上上个例子，用上面带协议的通道并注释掉这行

adder::Num1 <-- 10;

协议会把本来难以检测的死锁变成直接的违背协议异常。你会得到一条运行时的错误：

“Invalid use of 'Num2' at this point in the conversation, i.e. in state 'Adder.Start'”

很遗憾，不过每一个死锁都可以用严格的协议来捕获。如果用户向Num1发送了数据，但忘记向Num2发送数据而直接试图从Sum得到结果。和以前一样，AdderAgent和MainAgent也会无法继续执行，但是，这不会被认为是违背协议因为你仍然可以向Num2发送消息来解开死锁。


版权所有，如有转载，请站内联系

评论

2 楼 cuiguanglei 2009-08-09  

我对代理这块了解不是很深，改天得详细作下试验才成啊。哈哈

1 楼 cuiguanglei 2009-08-09  

不错，好文章。为什么不讨论JAVA方面的技术类。