数据并行（Data Parallelism）

数据并行（Data Parallelism）

数据并行，不同的数据输入以并行方式运行同一个函数，它把一个任务分解成不连续的单元，因此，可以在单独的线程上并行处理，保证这个任务可以在可用的处理之间进行分配。

通常是处理数据集合，这种方法利用了集合中的项目自然提供了任务分配。最简单的情况，是一个并行的映射函数，对集合的中每一项应用一个转换，结果形成一个新的集合。这种简单的情况通常是可以工作的，是因为集合中的每一项通常可按任意顺序、进行单独处理；用这种处理更复杂的情况，比如汇总列表中所有项，也是可能的；然而，对于一些更复杂的情况，以及必须按顺序处理的，就可能有问题了。

数据并行通常依赖于有并行处理能力的库和框架。虽然它们使用多线程或进程来处理并行，但是，并行并不要求用户创建或控制这些线程；相反，这是库或框架的任务。工作单元可以分布在不同物理机上，以形成计算网格；出于简化的目的，以及由于多核系统已经更为常见和强大，本章将只专注于在一台物理机的多个处理器之间分布工作的系统。微软正致力于在 .NET 框架中提供新的并行编程能力，将在 .NET 框架 4.0 版本中可用；已有其他一些库实现了 .NET 平台上的数据和任务的并行，但是，这一章将只关注.NET 4.0 中的内容。

有两种主要方法完成.NET 4.0 中数据的并行：使用mscorlib.dll 下的System.Threading.Parallel 类，或者使用 System.Core.dll下的 System.Linq.ParallelEnumerable 类。System.Threading.Parallel类完全可以在 F# 中使用，而System.Linq.ParallelEnumerable 类可能是 F# 程序员完成数据并行的首选方法，因为这个库更多地是用函数风格写的。

我们首选看一个示例，它演示如何使用System.Threading.Parallel 类中的并行 For，然后，讨论一下为什么我们并不想用它。假设我们想并行输出 0 到100 的整数，可以用下面的程序：

open System.Threading

Parallel.For(0, 100, (printfn"%i"))

在双核机器上运行时，前面的代码产生的下面的结果：

0

13

8

9

6

7

14

10

...

来自循环的数以不确定的顺序出现，这是因为以并行方式运行的，人每个函数运行的线程的计划任务是随机的。在 F# 中使用这种函数的问题在某种程度上是由于函数的副作用（side effect）。在 F# 中创建一个有副作用的函数很方便，但是，在并发处理时是不合适的，因为它会引入问题。即使在这个简单的示例中，我们也面对数字不能以原子方式打印的风险，而是混起来打印出来了。例如，你可能看到这样的结果：

...

6

174

10

...

这里，有两个数字构成了一个三位数，在集合中并不存在。

记住，System.Threading.Parallel 类在有些情况下仍是有用的。假设我们想要发送大量的电子邮件，且想同时发送，就可以用Parallel 类中的并行的ForEach 来并行化这个任务，因为发关电子邮件是输入输出，因此，有副作用：

open System.Threading

let emails = ["robert@strangelights.com"; "jon@doe.com";

          "jane@doe.com"; "venus@cats.com" ]

Parallel.ForEach(emails, (fun addr ->

  //code to create and send email goes here

  ()))

即使在这个简单的示例中，也需要保证能够调用任何函数，从多线程在Parallel.For 中调用。

System.Linq.ParallelEnumerable 类更方便 F# 程序的并行化，本质上，它是F# 序列（Seq）模块中函数的并行实现。因为在序列模块与ParallelEnumerable 类之间有大量的名称改变，常见的做法是创建 ParallelEnumerable的薄的包装，这样感觉就像在用序列模块，如下面的代码：

namespace Strangelights.Extensions

open System

open System.Linq

// Import a small number of functions fromParallelLinq

module PSeq =

  //helper function to convert an ordinary seq (IEnumerable) into aIParallelEnumerable

  letasParallel list: IParallelEnumerable<_> = ParallelQuery.AsParallel(list)

  //the parallel map function we going to test

  letmap f list = ParallelEnumerable.Select(asParallel list, new Func<_,_>(f))

  //other parallel functions you may consider using

  letreduce f list = ParallelEnumerable.Aggregate(asParallel list, new Func<_, _,_>(f))

  letfold f acc list = ParallelEnumerable.Aggregate(asParallel list, acc, newFunc<_, _, _>(f))

可以使用这段代码的完成版本替代序列模块中的函数调用，簮地换成 PSeq 包装模块的函数调用，在大多数情况下，程序会运行得更快；也有变慢的情况，例如，有些很短的列表，其中每一项要求的工作相对也很小，那么过个代码可能更慢。在微基准测试上可以看到并行映射函数的这种变化，通过比较并行映射与常规映射函数，可以改变输入列表的大小和输入列表中每一项的工作量。

注意

微基准测试是一项很有用的工具，有助于了解一段代码的性能，Vance Morrison 在 MSDN 上有一篇关于 .NET 平台上如何运行微基准测试的文章，http://msdn.microsoft.com/en-us/magazine/cc500596.aspx。

下面的示例演示了你可能的做法：

open System.Diagnostics

open Strangelights.Extensions

// the number of samples to collect

let samples = 5

// the number of times to repeat each testwithin a sample

let runs = 100

// this function provides the"work", by enumerating over a

// collection of a given size

let addSlowly x =

  Seq.fold(fun acc _ -> acc + 1) 0 (seq { 1 .. x })

// tests the sequentual map function byperforming a map on a

// a list with the given number of itemsand performing the given

// number of opertions for each item.

// the map is then iterated, to force it toperform the work.

let testMap items ops =

  Seq.map(fun _ -> addSlowly ops) (seq { 1 .. items })

  |>Seq.iter (fun _ -> ())

// test the parallel map function, works asabove

let testPMap items ops =

  PSeq.map(fun _ -> addSlowly ops) (seq { 1 .. items })

  |>Seq.iter (fun _ -> ())

// a test harness function, takes afunction and passes it the give

let harness f items ops =

  //run once to ensure everything is JITed

  fitems ops

  //collect a list of results

  letres =

    [for _ in 1 .. samples do

     let clock = new Stopwatch()

     clock.Start()

     for _ in 1 .. runs do

       f items ops

     clock.Stop()

     yield clock.ElapsedMilliseconds ]

  //calculate the average

  letavg = float (Seq.reduce (+) res) / (float samples)

  //output the results

  printf"Items %i, Ops %i," items ops

  Seq.iter(printf "%i,") res

  printfn"%f" avg

// the parameters to use

let itemsList = [ 10; 100; 200; 400; 800;1000 ]

let opsList = [ 1; 10; 100; 200; 400; 800;1000 ]

// test the sequential function

for items in itemsList do

  forops in opsList do

    harnesstestMap items ops

// test the parallel function

for items in itemsList do

  forops in opsList do

    harnesstestPMap items ops

在查看这个测试结果之前，了解一点微基准测试的知识还是值得的，因为它能更好地帮助我们对析理解。其中最重要的函数可能是harness，它是负责运行测试代码的任务。设置测试任务时，要记住两点：第一，要测量测试结果，每个测试要运行100 次，这是因为在某些小列表上的测试可能运行得异常地快，因此，如果只运行一次，很难测量出它们到底花了多少时间，重复运行就可能避免这个问题；第二，总是创建列表的五个结果，然后取这个列表的平均时间。这是因为在计算机上其他后台进程可能会影响某些测试，多运行几次取平均时间能够避免这个总是，还应该再进一步，计算出标准差（standard deviation），它能够突出那些特别长或特别短的测试结果。

另一个有趣的函数是testMap，它有一个任务，为映射函数提供需要做的工作。需要考虑两个不同的事情：每个输入列表的数目，和列表中每一项需要处理的数量，testMap 函数用两个参数来实现：items 和 ops。参数items 是映射函数必须处理的列表的数目，参数ops 是每一项必须执行的操作的数目。还有一点需要注意的，因为 Seq.map 和 PSeq.map 两个函数都延迟计算（lazy）的，因此，需要强制遍历结果列表。遍历列表将导致延迟序列的创建并计算；如果不这样做，会看到一个很小的、恒定时间的结果，这个是它创建一个能够用来生成列表对象所花费的时间，而不是生成列表本身所花费的时间。通过遍历强制生成列表，使用 Seq.iter 函数。

现在，我们已经可以看看结果本身了（见表 10-1）。

表 10-1 微基准测试序列的结果

解释原始数据是困难的，最好的办法是用这些数字绘制出图表。图 10-3、10-4 和10-5 是 Seq.map 和 PSeq.map 函数处理列表所花时间，以微秒表示，序列的长度是固定的（分别是 10 项、100 项和 1000 项），但是，在序列上的操作数目不同。

图 10-3 处理列表 10 次的时间，有不同的操作数目，以微秒表示

图 10-4 处理列表 100 次的时间，有不同的操作数目，以微秒表示

图 10-5 处理列表 1000 次的时间，有不同的操作数目，以微秒表示

这三个图形很好地说明了试验的结果。对于数量很少的序列（如图 10-3 中的 10项），可以看到，当每一项上执行的操作也少时，并行函数（PSeq.map）比串行函数（Seq.map）慢，随着每一项上执行的操作的增加，并行处理变得比串行的稍快，但差别并不明显；对于 100 项的序列（见图 10-4），可以看到类似的曲线，函数的并行版本比串行版本变快的发生要更提前了，并行函数的增量比串行更明显；最后，对于 1000 项的序列（见图 10-5），可以看到，由函数并行而增加的资源消耗已经完全被抵销了，并行函数线性地比顺序版本快一倍，是因为我们的测试运行在双核处理器上。从这里，我们可以得到结论，使用并行版本的映射函数是值得的，只要输入的列表足够大。

数据并行（Data Parallelism）,布布扣,bubuko.com

数据并行（Data Parallelism）

原文：http://blog.csdn.net/hadstj/article/details/25782797