PGM 和 WCF 的可靠多播

• 下载源代码 - 147 KB

问题所在

IP 组播是指将单个消息发送到组播组的过程，该消息由交换机根据注册在该组中的用户数量进行复制。此技术允许用户将单个消息发送给多个接收者。最广泛使用的组播传输是 UDP。

UDP 组播提供了出色的性能和网络利用率；但是，它存在两个缺点

1. 最大消息大小限制为 64 KB。
2. UDP 数据包可能会被网络设备（路由器、交换机等）丢弃。

尽管可以通过不同的压缩算法部分克服第一个缺点，但第二个缺点可能会成为一个真正的问题。

PGM 来拯救

多年来，在学术和商业场景中开发和测试了许多可靠的组播协议；然而，PGM（实用通用组播）是最合适的，因此被微软的实现采纳，作为 MSMQ 3.0 的一部分，MSMQ 3.0 必须安装在 Windows XP 及更高版本的操作系统上才能运行此示例。

PGM 适用于

• 股票和新闻更新
• 数据会议
• 实时视频传输
• 批量数据传输

PGM 是一种传输层协议，它直接通过原始套接字与网络层进行接口。这意味着与大多数广泛使用的协议不同，PGM 不基于 TCP 或 UDP，而是实现自己的传输机制。

PGM 协议规范定义了五种数据包类型

• ODATA – 原始数据内容
• NAK – 否定确认
• NCF – NAK 确认
• RDATA – 数据重传（修复）
• SPM – 源路径消息

PGM 将数据包接收的责任转移到接收方，从而消除了可能泛滥发送方链路的 ACK 的需求。相反，当接收方检测到序列中缺少某个数据包时，会发送 NAK（否定确认）。

此外，PGM 对 ODATA 数据包、RDATA 数据包或两者都使用 FEC（前向纠错）。FEC 允许接收方即使在数据包部分丢失或乱码的情况下也能组装数据包。

PGM 是一种异步协议，这意味着当套接字被创建为发送方时，它只能发送数据，同样，接收方也只能接收数据。

PGM 通过下面的分层结构实现其效率。SPM 消息由 PGM 发送方发送以构建分层树。

利用这些技术，PGM 保证组内的接收方要么接收所有传输和修复的数据包，要么能够检测到（罕见的）无法恢复的数据包丢失。PGM 的网络利用率高于 91%，并且在带宽受限的网络和高速网络（>100 Mbps）上都能很好地扩展。

WCF 实现

自定义传输

为了在 WCF 框架内利用 PGM 的优势，需要自定义传输。编写自定义传输绑定元素并非易事，但得益于 WCF SDK 示例和 Roman Kiss 的杰出开创性工作，这项任务是可以实现的。

MEP

由于 PGM 本质上是单向协议，我们将使用 OneWay MEP，即实现 IInputChannel/IOutputChannel 接口，以及 Session 接口。

架构

PgmTransportElement 是入口点。它由 WCF 配置引擎创建。通过调用重载的 CreateBindingElement 方法，创建 PgmTransportBindingElement。

protected override BindingElement CreateBindingElement()
{
    PgmTransportBindingElement bindingElement = 
                 new PgmTransportBindingElement();
    this.ApplyConfiguration(bindingElement);
    DiagnosticsManager.Instance.Configure(Diagnostics);
    return bindingElement;
}

此类负责在发送方创建通道工厂

public override IChannelFactory<TChannel> 
       BuildChannelFactory<TChannel>(BindingContext context)
{
   if (typeof(TChannel) == typeof(IOutputChannel))
   {
        return (IChannelFactory<TChannel>)
               (object)new PgmChannelFactory(this, context);
   }
   else if (typeof(TChannel) == typeof(IOutputSessionChannel))
   {
        if (DataMode == DataMode.Stream)
        {
           throw new InvalidOperationException("Sessionful contracts" + 
             " cannot use Stream mode. Either configure " + 
             "dataMode to Message or change the contract to non sessionful");
        }

        return (IChannelFactory<TChannel>)(object)
                new PgmSessionChannelFactory(this, context);
   }
   else
   {
        throw new ArgumentException("Unsupported channel type " + 
                                    typeof(TChannel).Name);
   }
}

并在接收方创建通道侦听器

public override IChannelListener<TChannel> 
       BuildChannelListener<TChannel>(BindingContext context)
{
     if (typeof(TChannel) == typeof(IInputChannel))
     {
        return (IChannelListener<TChannel>)
               (object)new PgmChannelListener(this, context);
     }
     else if (typeof(TChannel) == typeof(IInputSessionChannel))
     {
        if (DataMode == DataMode.Stream)
        {
           throw new InvalidOperationException("Sessionful contracts" + 
                 " cannot use Stream mode. Either configure dataMode " + 
                 "to Message or change the contract to non sessionful");
        }

        return (IChannelListener<TChannel>)
               (object)new PgmSessionChannelListener(this, context);
     }
     else
     {
        throw new ArgumentException("Unsupported channel type " + 
                                    typeof(TChannel).Name);
     }
}

此实现包含两种通道类型

• PgmInputDatagramChannel/PgmOutputDatagramChannel – 无会话的单向消息交换。
• PgmInputSessionChannel/PgmOutputSessionChannel – 有会话的单向消息交换。

每个输入通道都封装了 PgmReceiver 对象，每个输出通道都封装了 PgmSender 对象。这些对象反过来又直接与 PGM 套接字进行接口，并负责实际的数据传输。

InputChannel 及其后代的实现基于 UDP 示例中可用的 InputQueue 通用容器。这是生产者/消费者集合的高级实现，它将网络传入数据的处理与 WCF 通道堆栈的消息对象处理解耦。换句话说，数据读取的速度可能比处理的速度快；因此，它被存储在队列中并按顺序处理。

配置选项

通用选项

• maxMessageSize – 设置消息的最大大小。
• dataMode – PGM 套接字实现包含两种数据模式：Message 和 Stream。

在 Message 模式下，使用 SocketType.Rdm 值创建套接字，因此保留了消息边界。流套接字使用 SocketType.Stream 创建，类似于 TCP 等其他流协议，其中一次发送可能不会导致一次接收。因此，需要实现帧定界——我选择实现长度帧定界，这意味着每条消息都以一个包含消息长度的固定大小数组为前缀。这样，接收方就知道需要从套接字读取多少数据来组装一条有效消息。

发送方选项

• sendRate – 默认情况下，发送速率配置为 56Kbps，相对较慢，所以我将其设置为 5Mbps 的默认值，并且可以扩展到更大的值。
• lateJoin – 在传输中间加入会话的接收者可能会请求它丢失的数据。此值表示延迟加入者可以请求重传数据的百分比。
• fecMode – 如前所述，FEC 有助于恢复部分丢失或接收不正确的数据包。有四种选项
1. Disabled – FEC 已禁用。
2. OnDemand – FEC 应用于 RDATA 数据包。
3. ProActive – FEC 应用于 ODATA 数据包。
4. Both – 同时应用 OnDemand 和 ProActive。
• senderWindowAdvance – 如果接收方发现序列中缺少某个数据包，它会向发送方发送 NAK，发送方会发送 RDATA 来恢复丢失的数据包。此值表示发送窗口前进的百分比。
• setMessageBoundary – 不希望单次发送命令发送一个大缓冲区。因此，您可以告知套接字一个大包大小，然后套接字将以更符合网络的方式发送。
• multicstTTL – 设置每个数据包的生存时间，即消息在被丢弃之前可以路由通过的网络元素的数量。
• senderInterface – 在多宿主机器上，可以设置 PGM 传输所需的网络接口。

诊断

PGM 发送方和接收方都提供了一套丰富的实时统计数据。这些值通过将 PgmTransport 的诊断选项设置为 Enabled 来记录在专用文件中。此外，还可以通过配置设置时间间隔和目录。

已知问题

当从运行在 PGM 传输之上的 WCF 服务导入 WSDL 数据时，SvcUtil.exe 无法导入 PgmTransport 元素及其所有属性。我尝试实现取自 UDP 示例的 WSDL 和策略导出接口，但没有成功。因此，在运行 svcutil.exe 后，您需要手动复制丢失的 PgmTransport 元素。

PGM 支持流模式；但是，我的实现尚未完全支持。