并行编程与 Mandelbrot 集

• 下载WaitableTaskLibrary.zip - 222.2 KB

引言

本文通过渲染Mandelbrot集的图像来演示并行处理库的使用。
通过渲染Mandelbrot集的图像来演示并行处理库的使用。

在C#中，有许多方法可以实现任务并行性。
大多数情况下，这些方法形成松散的编程模式，没有严格的实现，不能立即
重用。在开发这个库时，我希望能构建一个良好、可重用的
通用系统，并且效率合理。
我还通过构建一个Mandelbrot集探索器演示了它的使用，该探索器通过多线程快速创建了一些漂亮的碎形渲染图，尽管在优化算法方面投入的精力相对较少。

我还通过构建一个Mandelbrot集探索器演示了它的使用，该探索器通过多线程快速创建了一些漂亮的碎形渲染图，尽管在优化算法方面投入的精力相对较少。

 

背景

回顾我阅读或编写过的所有多线程代码，似乎遵循两种主要模式（这些可能是
公认的模式，但如果是，我不知道它们的名称）
我不知道它们的名称）

第一种是编写一个同步代码模型（即正常的
程序流），它在一个与UI线程分离的托管线程中执行。
UI线程。通常，这种方法用于运行时间非常长的串行操作，
以确保它们不会使应用程序无响应。

执行代码的线程通常是
专门为此操作手动启动的线程，但它也可能是
预初始化线程数组中的一个，这些线程从队列中获取工作。
队列。

一个例子是Tcp服务器。TcpListener类的阻塞方法
AcceptSocket将新接受的客户端传递给另一个线程，
因此通过AcceptClient方法监听连接的线程可以立即接受下一个客户端。
方法可以立即接受下一个客户端

一个简单的多线程套接字服务器实现（这假定TcpListener (Listener) 已初始化，并且有一个 [void SocketConsumer(object)] 方法处理入站套接字的请求）

new Thread(()=> {
 while (true) {
  var socket = Listener.AcceptSocket();
                new Thread(SocketConsumer).Start(socket);
 }
}).Start();

处理套接字的代码将在SocketConsumer方法中。
一个新线程需要几个周期才能启动，因此像上述方法一样即时创建新线程，
只应在操作足够长的情况下使用，
不介意创建线程会占用几毫秒（已经创建的线程正在等待
等待句柄或自动重置事件以进行更多工作是解决问题的一种方法，
NET ThreadPool类中使用了此方法）
在.NET ThreadPool类中使用了此方法）

另一个重要的模式是使用回调委托的“异步”代码模型。
我发现这种模式创建和调试起来要困难得多，因为程序流是断开的。
因为程序流是断开的。

您启动一个操作并使用 Begin... 定义一个回调。
method

例如

IAsyncResult rs = Listener.BeginAcceptSocket(callbackCode,
null);

然后，一旦操作完成，一个单独的方法（或匿名委托）将接收回调。
一旦操作完成，回调就会被接收

void callbackCode(IAsyncResult target) {
 
 Socket sock = Listener.EndAcceptSocket(target);
 SocketConsumer(sock);
 
}

IASyncResult接口有一个IsCompleted属性，
可以告知您方法是否已完成，这可以轮询（有代价），
但轮询周期a)浪费处理器周期，b)浪费方法完成与下一个轮询间隔之间的时间。
方法完成到下一个轮询间隔之间的时间。

.NET框架中的Begin…. 和 End…. 方法将工作提交到线程池，
然后在它们完成后执行回调方法委托（void (IAsyncResult)）。
回调方法中的代码可以引发一个事件，以表示方法已完成。
以表示方法已完成。

这些技术存在一些缺点。首先是，当使用Begin和End方法时，
您的代码最终会像意大利面条一样…您无法轻易地跟踪程序流程，并且
跟踪异常很痛苦。
跟踪异常很痛苦。

另一个缺点是，没有简单、一致的方法可以提交一系列任务，
让它们同时执行，然后暂停程序流，直到所有任务都完成。
直到它们全部完成。

一个例子可能是提交多个查询，以获取数据以输入到另一个进程中。
当数据库正在执行查询时，客户端机器几乎不做任何事情，
因此即使在单处理器机器上，并行运行查询也会带来很大的性能提升（假设
数据库服务器可以处理负载）——但是您可能需要所有数据都返回后才能
才能开始进程的下一个部分。能够等待
十几个操作完成，然后继续程序流程是经常需要的。
十几个操作完成，然后继续程序流程是经常需要的。
需要的需求。

开发

现有Begin.. End异步编程模型缺少关键要素，
即能够快速轻松地等待一个任务或一组任务完成，而无需定义事件或回调来弄乱代码。
而无需定义事件或回调来弄乱代码。
代码。

这使我走上了开发一种可以包装方法并使其“可等待”的道路——即“WaitableTask”
你可以将其包装在一个方法周围，使其“可等待”——即“WaitableTask”。
类。

“WaitableTask”将有一个“WaitComplete”方法，
该方法会暂停程序流，直到其工作负载完成。

这将通过在传递给委托的BeginInvoke方法的回调方法中设置ManualResetEvent来实现。
回调方法中设置一个ManualResetEvent来实现。

我决定使用手动重置事件（而不是
AutoResetEvent），因为我希望事件在设置后保持信号状态。
这样，如果操作在您开始等待之前已经完成，
WaitComplete方法将直接运行而不会停止。

WaitComplete方法将等待ManualResetEvent被设置，
然后返回控制权。 我的第一个设计
设计使用了没有委托调用定义的抽象类，以及许多抽象方法。
以及许多抽象方法。然后，可以为每个所需的委托签名从该类派生一个类。
每个所需的委托签名派生一个类。

不过，我从未满意这种抽象程度。
经过一番折腾，我意识到我可以将基类用于所有委托（Delegate类）——但主要问题是，
但主要问题是，这个类没有BeginInvoke方法（因为参数在这个类层次结构级别没有定义）。
因为参数在这个类层次结构级别没有定义）
的类层次结构）

然而，可以通过调用Delegate类Method属性返回的MethodInfo上的“Invoke”方法，
针对其Target属性调用“Invoke”，并提供一个对象数组作为方法参数，
并返回一个对象。
并返回一个对象。

无论委托的签名如何，此调用都有效（通常委托签名必须完全匹配才能绑定，
但在此情况下，绑定已通过被引用为抽象Delegate类的具体委托实现）。
但在此情况下，绑定已通过被引用为抽象Delegate类的具体委托实现）
通过抽象Delegate类引用）

下面的示例：InvokeMethod(object[]) 方法调用_method委托，
传入一个对象数组作为参数，并返回一个对象引用。

protected virtual object InvokeMethod(object[] parameters)
{
     // invoke the _method delegate directly and return the result.
     return _method.Method.Invoke(_method.Target, parameters);
}

这个仍然没有BeginInvoke方法。但是InvokeMethod的签名是具体的，所以我可以创建一个委托来调用它。

// concrete delegate matches the parameters of InvokeMethod
public delegate object Invoke(object[] parameters);

// assign the delegate:
Invoke _invokeDelegate = InvokeMethod;

// start invokation of the delegate and record the IAsyncResult.
_asyncToken = _invokeDelegate.BeginInvoke(_parameters, Callback, this);

所有这些很容易地组合起来形成了WaitableTask类。这个类既没有定义参数的泛型类型参数，也没有定义返回类型，但是Factory类中的Create方法能够定义参数类型参数和WaitableTask的具体委托，并且派生类能够提供返回类型参数。

/// <summary>
/// create a simple waitable with one argument and no return type.
/// </summary>
/// <typeparam name="TArg"></typeparam>
/// <param name="method"></param>
/// <param name="argument"></param>
/// <returns></returns>
public static WaitableTask Create<TArg>(Proc<TArg> method, TArg argument)
{
 return new WaitableTask(method, argument);
}

/// <summary>
/// create a waitable with a return type of TResult and 1 argument of type TArg1
/// </summary>
/// <typeparam name="TResult">
/// the return type for the operation
/// </typeparam>
/// <typeparam name="TArg1">
/// the argument type for the operation.
/// </typeparam>
/// <param name="method">
/// a delegate pointing to the method to run.
/// </param>
/// <param name="argument">the argument for the function</param>
/// <returns>an WaitableTask ready to run the operation</returns>
public static WaitableTask<TResult> Create<TArg1>(Func<TArg1, TResult> method, TArg1 argument)
{
    // create and return the delegate:
    return new WaitableTask<TResult>(method, argument);
}

我为最多三种参数类型创建了Create方法的重载，无论是否有返回类型。现在创建一个“WaitableTask”只需要一行代码。

除了WaitComplete方法和IsComplete属性，我还定义了一个“Completed”事件和一个“Errored”事件。

当使用BeginInvoke调用委托时，它被提交到ThreadPool。完成后，线程池执行回调委托以发出信号，操作结果通过原始委托的EndInvoke方法可用。任何异常也会抛出到EndInvoke方法。

WaitableTask类捕获由_invoke委托的EndInvoke方法抛出的任何异常，将它们内部存储在任务中，并引发“Errored”事件，将异常与EventArgs一起传递。

无论是否抛出任何异常，Completed事件仍然会在操作完成时引发。

通过使用类似的模式，可以对一组WaitableTasks进行分组，并作为一个整体进行等待。

WaitableTaskList<T>类中的AddTask(WaitableTask<T>)方法将指定的waitable-task添加到内部列表，并处理其“Completed”事件。

通过检查其他任务是否仍在运行，此事件处理程序能够确定最后一个任务是否刚刚完成处理，并可以引发自己的事件。

WaitableTaskList<T>类定义了两个事件：TaskComplete在每个任务完成时引发，而ListComplete仅在所有任务完成时触发。

通过向Factory类添加另一组Create<>方法，现在只需一行代码即可定义一个等待列表，该列表同时以多个参数运行相同的方法，并且可以等待完成。

public static WaitableTaskList Create<TArg1>(
Proc<TArg1> taskMethod, params Argument<TArg1>[] args)
{
    var taskList = new WaitableTaskList();
    foreach (var arg in args)
    {
        // add a task for each item in the args list.
        taskList.AddTask(Create<TArg1>(taskMethod, arg.Argument1));
    }
    // return the task-list:
    return taskList;
}

并行处理示例

为了演示WaitableTask库的使用，我需要解决一个并行编程问题。

我现有的使用此库的实现是数据库应用程序，它们使用它并行运行查询；但是我不想为了演示而随代码附带一个示例数据库。

我转向了“尴尬并行问题”列表寻求灵感……（“尴尬并行问题”是指几乎不需要努力即可将其划分为并行任务的问题——这通常是因为问题的每个元素不需要与其他元素交互）

我选择渲染Mandelbrot集，因为我认为它会制作一个很酷的程序，并且维基百科上有一个关于使用“逃逸时间”算法计算每个像素值的伪代码列表，我可以很容易地使用。

计算Mandelbrot集中像素的值是完全独立的——它不需要其他操作的结果，并且很容易实现“并行”。

我选择使用单独的任务/线程来渲染图像的每一“行”。这样我只需要传递一个描述正在渲染的行的整数。此外，这意味着每个任务的运行时间都足够长，足以证明其通过队列的合理性——如果我尝试定义一个任务来渲染每个像素，那么即使是小图像（320x240）也需要76,800个任务。CPU将花费更多的时间和内存来管理任务列表、将每个任务排队并处理其完成，而不是实际运行任务，任何性能提升都将丧失……事实上，渲染可能会花费更长的时间。

下一个问题是如何将逃逸时间算法计算出的整数值实际转换为颜色信息并将其设置在位图图像上。

这里有两个问题。

计算机图像中的每个像素都由红色、绿色和蓝色分量组成，这些分量与人视网膜中细胞的相应滤色器相匹配，但是计算每个像素值的逃逸时间算法的输出是一个单一的整数。

我需要一种将整数值转换为颜色信息的方法。我首先考虑生成一个公式/算法，将一个给定范围的整数转换为红色、绿色和蓝色值，但由于它不灵活、速度慢且编码痛苦而放弃了。

相反，我选择通过在一组指定颜色之间创建渐变来生成颜色信息“调色板”。分形算法输出的每个可能值都将映射到颜色数组中的不同元素。这提供了灵活性，因为可以使用不同的“调色板”与相同的算法，而且速度会很快。

此方法使用逃逸时间分形算法计算单个像素的颜色值。

/// <summary>
/// calculate the colour at the specified point in the mandelbrot-set
/// </summary>
/// <param name="x0">scaled X input (between -2.5 and 1)</param>
/// <param name="y0">scaled Y input (between -1 and 1)</param>
/// <param name="p">palette used to assign a colour to the iteration</param>
/// <returns>colour from the internal pallette</returns>
public static Color CalculateMandelbrotPoint(double x0, double y0, Palette p)
{
    double x = 0;
    double y = 0;
    int iteration = 0;
    while ((((x * x) + (y * y)) < (2 * 2)) && (iteration < p.Colours.Length))
    {
        double xtemp = (x * x) - (y * y) + x0;
        y = 2 * x * y + y0;
        x = xtemp;
        iteration++;
    }

    return p.Colours[iteration-1];
}

现在我只需要一种方法来将这些颜色信息设置到位图图像上。

System.Drawing.Bitmap对象提供了一个“SetPixel(x,y,color)”方法。但是，此方法不是线程安全的，并且出了名的慢。SetPixel方法上的线程冲突会导致奇怪的错误消息和主机应用程序中的GDI+错误。

这时我才想起，有一种略微可怕但快得多的方法可以在位图上设置像素值。

使用Bitmap对象的LockBits方法，您可以获取对包含实际位图数据的非托管内存区域的引用，并将其封送到托管字节数组中。然后您可以更改字节数组的值，将数据封送回去并释放位图。

_lockData = _bmp.LockBits(new Rectangle(0, 0, this._bmp.Width, this._bmp.Height), flags, this._bmp.PixelFormat);
_lockState = true;
_byteLen = Math.Abs(_lockData.Stride * _lockData.Height);
_imageBuffer = new byte[_byteLen];
IntPtr start = _lockData.Scan0;

// marshal the data into the array.
System.Runtime.InteropServices.Marshal.Copy(start, _imageBuffer, 0, _byteLen);

本质上，Bitmap.SetPixel就是这样做的，但对于您设置的每个像素，它都会锁定位图，封送出数据，修改正确的字节值，然后返回数据并解锁图像。您可以看到，如果两个线程调用相同的方法，就会出现问题：如果位图已被锁定，则无法再次锁定它，并且将数据封送回去可能会覆盖另一个毫秒前进入的调用设置的值。

通过本质上批处理这些操作，不仅图像数据的操作速度快得多，而且也变得线程安全。

该操作是线程安全的，因为每次为像素设置值时，您只是在使用绑定到调用线程的索引和值在公共数组中设置一个值。即使两个或更多线程同时运行相同的代码，每个线程的索引值也不同，并且没有冲突。

这里的困难在于计算要设置的值：返回的数据特定于源图像的像素格式。

例如，如果像素格式是RGB，每像素24位，那么每个像素由三个字节表示（一个红色，一个绿色，一个蓝色）——如果格式是每像素32位ARGB，那么每个像素有一个额外的字节表示其Alpha（透明度）值（255 = 不透明，0 = 透明）

除此之外，返回的数组是一维的，而图像有两维。本质上，图像按行展开成一个长的字节序列。您需要知道像素格式和图像宽度才能知道在哪里将序列分成每一行。

上表显示了3x3白色像素网格的索引、颜色类型和值。

图像的宽度是3，图像的高度是3，像素格式是24bppRGB，图像的“步幅”是9。

“步幅”是指图像的字节宽度，而不仅仅是像素。由于每个像素由三个字节值定义，因此3像素宽图像的步幅为3*3=9。查看表格，我们可以看到对于中间像素(1,1)，索引是12,13和14。

计算像素(x,y)的第一个索引位置的公式是：

Index = (stride * y) + (bytes-per-pixel * x)

例如，中间像素的第一个（蓝色）索引是：(9 * 1) + (3 * 1) = 12

或者对于右下角像素(2,2)，索引是：(9*2) + (3 *2) = 18+6 = 24。

下面是示例中Set Pixel方法的代码：它假定像素格式为24位每像素RGB，并且_imageBuffer数组已填充图像数据。它从Bitmap.LockBits方法返回的BitmapData中获取图像的步幅宽度。

protected void SetPixel24Bpp(int x, int y, Color color)
{
    if (_lockState)
    {
        // bytes-per-pixel is three:
        int bpp = 3;

        // first byte address:
        int read_from = (x * bpp) + (_lockData.Stride * y);

        // set the three byte values:
        this._imageBuffer[read_from + 0] = color.B;
        this._imageBuffer[read_from + 1] = color.G;
        this._imageBuffer[read_from + 2] = color.R;
    }
}

优化

通过不重新计算每个像素的索引值，可以节省一些额外的处理周期。由于我们是按行渲染，并且图像数据在数组中是按行排列的（这是因为旧的CRT显示器会通过水平扫描线显示图像），我们可以在扫描线开始时计算索引位置，然后为每个像素加三。

Using the Code

要创建可等待任务列表，首先需要创建Argument对象的数组或列表。每个Argument都将为列表创建一个新任务。

在分形渲染示例中，我们想要调用的方法（CalculateMandelbrotRow）接受一个整数参数，因此我们需要创建Argument<int>对象添加到列表中，并为图像中的每一行添加一个。

// create the argument list for the TaskFactory Create method.
var args = new List<Argument<int>>();

// setup an argument for each row of the output image: each row will be a queued task for the thread-pool.
for (int y = 0; y < _imageSize.Height; y++)
{
args.Add(new Argument<int>(y));
}

然后只需将方法和参数列表传递给TaskFactory类中的静态Create<>方法。

参数是一个可变参数数组，因此我们可以传入单个 Argument 对象（任意数量）或 Argument 对象数组。 这里我使用 List 类的 ToArray() 方法将列表转换为数组。

// create the task list: use the TaskFactory to shortcut and provide concrete types:
var renderTask = TaskFactory.Create<int>(this.CalculateMandelbrotRow, args.ToArray());

然后任务列表需要启动。

renderTask.BeginAll();

然后我们希望等待它完成。在这种情况下，我们希望每100毫秒调用一次Application.DoEvents，以保持调用UI线程在分形渲染时“响应”。

// wait until complete, call Application.DoEvents each 100 ms 
renderTask.WaitComplete(Application.DoEvents);

由示例生成的一些漂亮的分形