使用 MPI 和 C# 实现并行编程

• 下载源代码 - 101 KB

引言

纯 .NET MPI 是 MPI 的完全托管实现。面向对象的 API 功能强大，但易于用于并行编程。它基于 .NET Framework 3.0 开发，并使用 Windows Communication Foundation (WCF) 的许多功能移植到 3.5。

借助 WCF，您可以使用配置文件声明正在使用的进程数量，哪个进程是主进程，以及网络上的进程位置（IP 地址和端口）。

纯 .NET MPI 最棒的地方在于，您可以将所有并行程序作为单个进程中的线程启动，以便调试和测试问题。

背景

.NET 实现的消息传递接口 (标准 2.0)

• 基于 Windows Communication Foundation (WCF)。
• 用于调试和开发的单进程 F5 体验。
• 多线程、多进程或多机器执行，以及任何组合。
• MPI 处理器可以在单个进程内运行，也可以在任意数量的机器上运行任意数量的进程。
• 类型安全的 API。
• 支持 x86、x64 和 Itanium。
• 可扩展。支持现有的和自定义 WCF 绑定和通道。为通信对象添加了更高级和自定义的同步机制。
• 支持通信超时，用于死锁检测和更好的错误处理。
• 环境和 WCF 设置完全可配置。

使用代码

此示例使用多个进程绘制 MandelBrot 集。

附件包含三个不同的项目

1. 绘制 MandelBrot 集的接口。
2. MPI 程序。
3. 包含编组对象的库。

该接口可以在顺序模式和并行模式下工作。

在并行模式下工作时，该接口将启动启用了 MPI 的进程来计算 MandelBrot 集，并使用 .NET Remoting 从主进程收集结果，如下所示

TcpChannel channel = new TcpChannel(8090);
ChannelServices.RegisterChannel(channel, false);
RemotingConfiguration.RegisterWellKnownServiceType(typeof(Points), 
                      "MandelBrot", WellKnownObjectMode.SingleCall);
Cache.Attach(observer);

以下是主进程的实现

Console.WriteLine("Master Process: Working...");

int row = 0;
for (int i = 1; i < comm.Size; i++)
{
    comm.Send<int>(i, "ROW_NUMBER", row);
    row += processIncrement;
}

ArrayList list = new ArrayList();
int max = (int)Math.Floor((double)width * height / partition) + 1;
for (int i = 0; i <= max; i++)
{
    ArrayList partList = 
      comm.Receive<ArrayList>(Constants.AnySource, "RESULT");

    if (partList != null)
        list.AddRange(partList);
}

Points points;

TcpChannel chan = new TcpChannel();
ChannelServices.RegisterChannel(chan, false);

points = (Points)Activator.GetObject(typeof(Points), 
          "tcp://:8090/MandelBrot");

points.SetMessage(list);

Console.WriteLine("Finished Calculating: " + list.Count + " points.");

以下是从进程的实现

Console.WriteLine("Slave Process: " + comm.Rank + ".Working...");

int row = comm.Receive<int>(0, "ROW_NUMBER");

decimal scaleX = (maxNumber.Real - minNumber.Real) / width;
decimal scaleY = (maxNumber.Imaginary - minNumber.Imaginary) / height;

ComplexNumber c = new ComplexNumber();

ArrayList list = new ArrayList();

c.Real = minNumber.Real;
for (int x = 0; x < width; x++)
{
    c.Imaginary = minNumber.Imaginary + row * scaleY;
    for (int y = row; y < (row + processIncrement); y++)
    {
        int count = CalculatePixel(c);

        PointSet set = new PointSet();
        set.W = x;
        set.H = y;
        set.Pixel = count;

        list.Add(set);

        c.Imaginary += scaleY;

        if (list.Count == partition)
        {
            comm.Send<ArrayList>(0, "RESULT", list);
            list.Clear();
        }
    }

    c.Real += scaleX;
}

if (list.Count > 0)
{
    comm.Send<ArrayList>(0, "RESULT", list);
}

Console.WriteLine("Done");

应根据需要设置配置文件，以确定进程数量以及进程将启动的位置。