【推荐阅读】并行性能：优化多核计算机的托管代码

java2000.net

浏览: 649962 次
性别:
来自: 天津

最近访客更多访客>>

kingkwj

Higher

sing9123

Mr_冰凉

博主相关

博客

微博

相册

留言

关于我

文章分类

社区版块

存档分类

虚拟机多线程工作算法 J#

本文讨论：

任务并行库
Parallel.For 与 ThreadPool 之比较
静态工作分配
Future

本文使用了以下技术:

Parallel FX 库

TPL 简介
结构化并行性
过度公开并行性
射线跟踪器示例
动态工作分配
聚合
派生-联结并行性
任务和 Future
可复制任务
任务管理器

由于单处理器速度无法再大幅提高，多处理器计逐渐成为标准工具。因此改善性能的关键是在多处理器上并行运行程序。遗憾的是，要编写真正充分利用那些多处理器的算法仍然相当困难。事实上，大部分的应用程序使用的只是单核，因此在多核计算机上运行时也看不出速度有任何改进。我们需要以新的方式来编写程序。

TPL 简介

任务并行库 (TPL) 的设计是为了能更简单地编写可自动使用多处理器的托管代码。使用该库，您可以非常方便地用现有序列代码表达潜在并行性，这样序列代码中公开的并行任务将会在所有可用的处理器上同时运行。通常这会大大提高速度。

TPL 是在 Microsoft^® 研究院、Microsoft 公共语言运行库团队 (CLR) 和并行计算平台团队的协作努力下创建的。TPL 是 Parallel FX 库中一个主要的组件，是对 Microsoft .NET Framework 的下一代并发支持。尽管版本尚未达到 1.0，但 MSDN^® 仍会在 07 年秋季发首次布 Parallel FX 的社区技术预览版 (CTP)。有关详细信息，请浏览 http://blogs.msdn.com/somasegar。TPL 对语言扩展没有任何要求，能与 .NET Framework 3.5 以及更高版本配合使用。

完全支持 Visual Studio^® 2008，且所有的并行性都通过普通的方法调用表达。例如，假设您有以下求数组元素平方的 for 循环：

for (int i = 0; i < 100; i++) {
a[i] = a[i]*a[i];
}

因为迭代是相互独立的（也就是说，后续迭代不会读取之前迭代进行的状态更新），因此可用 TPL 表达潜在的并行性，方法是调用并行的 for 方法，如下所示：

Parallel.For(0, 100, delegate(int i) {
a[i] = a[i]*a[i];
});

请注意，Parallel.For 只是一个带有三个参数的普通静态方法，其中最后一个参数是一个委托表达式。此委托捕获了之前示例中完全相同的循环主体，这使得在程序中尝试引入并发操作变得非常简单。

库中包含用于动态工作分配的复杂算法，并可自动适应工作负荷和特定的计算机。同时，库的原语仅仅表达潜在的并行性，但无法对它提供保证。例如，在单处理器计算机上，并行 for 循环是顺序执行的，与严格序列代码的性能紧密匹配。然而，在双核计算机上，根据工作负荷和配置，库使用两个工作线程来并行执行该循环。这意味着您可马上在代码中引入并行性，您的应用程序将会在条件许可时自动使用多处理器。同时，代码在旧式的单处理器计算机上仍然能够很好地运行。

遗憾的是，库并不能帮助正确实现使用共享内存的并行代码的同步。仍需程序员来负责确保某些代码能够安全地并行运行。其他机制（例如锁）对于保护共享内存的并发修改仍然十分必要。TPL 确实提供了一些有助于同步的抽象（接下来我们将向您展示）。

结构化并行性

并行程序最重要的抽象之一就是并行循环。例如，考虑以下的简单矩阵乘法例程：

void SeqMatrixMult(int size, double[,] m1, double[,] m2, double[,] result)
{
  for (int i = 0; i < size; i++) {
    for (int j = 0; j < size; j++) {
      result[i, j] = 0;
      for (int k = 0; k < size; k++) {
        result[i, j] += m1[i, k] * m2[k, j];
      }
    }
  }
}

在本示例中，外部迭代是各自独立的，并且有可能并行完成。通过 TPL 来发现此潜在的并行性非常简单。首先，我们在编译期间引用 System.Concurrency.dll 程序集。接着使用 using 语句，就能把库导入到代码中：

using System.Concurrency;

一旦命名空间可用，我们即可将矩阵乘法的外部 for 循环替换为调用静态 Parallel.For 方法：

void ParMatrixMult(int size, double[,] m1, double[,] m2, double[,] result)
{
  Parallel.For( 0, size, delegate(int i) {
    for (int j = 0; j < size; j++) {
      result[i, j] = 0;
      for (int k = 0; k < size; k++) {
        result[i, j] += m1[i, k] * m2[k, j];
      }
    }
  });
}

Parallel.For 的构造是一个带有三个参数的普通静态方法。前两个参数指定了迭代的限制（在 0 和具体大小值之间）。最后一个参数是为每个迭代调用的委托功能。此委托将迭代索引当作它的第一个参数，然后执行与之前示例中完全相同的循环主体。因为委托自动捕获循环主体的自由变量（比如 result 和 m1），所以不需要对最初的循环主体进行任何更改。有关委托表达式的详细信息，请参阅 msdn.microsoft.com/msdnmag/issues/06/00/C20。

最后，如果在任何迭代中引发异常，所有的迭代都会被取消，且第一个引发的异常将在调用线程中被重新引发，以确保异常得到正确传播并不会丢失。

如果没有 TPL，要在此循环中表达潜在的并行性就会困难得多。即使在 NET ThreadPool 类的帮助下，我们仍然必须考虑同步和工作划分的成本。下面的代码显示了使用线程池达成并行的矩阵乘法例程。

void ThreadpoolMatrixMult(int size, double[,] m1, double[,] m2,
    double[,] result)
{
  int N = size;
  int P = 2 * Environment.ProcessorCount; // assume twice the procs for
                                          // good work distribution
  int Chunk = N / P;                      // size of a work chunk
  AutoResetEvent signal = new AutoResetEvent(false);
  int counter = P;                        // use a counter to reduce
                                          // kernel transitions
  for (int c = 0; c < P; c++) {           // for each chunk
    ThreadPool.QueueUserWorkItem(delegate(Object o)
    {
      int lc = (int)o;
      for (int i = lc * Chunk;           // iterate through a work chunk
           i < (lc + 1 == P ? N : (lc + 1) * Chunk); // respect upper
                                                     // bound
           i++) {
        // original inner loop body
        for (int j = 0; j < size; j++) {
          result[i, j] = 0;
          for (int k = 0; k < size; k++) {
            result[i, j] += m1[i, k] * m2[k, j];
          }
        }
      }
      if (Interlocked.Decrement(ref counter) == 0) { // use efficient
                                                     // interlocked
                                                     // instructions
        signal.Set();  // and kernel transition only when done
      }
    }, c);
  }
  signal.WaitOne();
}

代码清单1

这个示例已相当先进，它将线程池用作工作项目，并使用计数器和一个等待句柄来最小化内核切换的次数。另外，它根据处理器的数目将循环静态划分成多个块，创建量相当于必需数量的两倍，从而更好地适应动态工作负荷。然而，与 Parallel.For 不同，代码清单1中所示的方法没有在循环主体中传播异常，因此无法取消。

显然，此代码比起 Parallel.For 方法要难编写得多，也更容易出错。而且，尽管经过手工调整并使用几乎完美的工作划分，一般来说线程池方法还是比 Parallel.For 方法要逊色。图 2 显示了一些有趣的测试。结果代表了当对具有 750x750 个元素的矩阵乘法的外部循环实现并行时获得的相对加速度，1 代表普通 for 循环的运行时间。这些测试是在一台四套接字、3GB 内存并运行 Windows Vista^® Ultimate 的双核计算机上进行的。请注意，在单核计算机上 Parallel.For 版本的实际执行过程与for 正循环相同。

图 2 Parallel.For 与 ThreadPool 性能比较

过度公开并行性

也许您已经注意到了，通过并行化第二个 for 循环，我们可以公开更多的并行性，如下所示：

Parallel.For( 0, size, delegate(int i) {
  Parallel.For( 0, size, delegate(int j) {
    result[i, j] = 0;
    for (int k = 0; k < size; k++) {
      result[i, j] += m1[i, k] * m2[k, j];
    }
  });
});

即使可以嵌套并行循环，此方法的性能一般会差一些，原因有两个。首先，在此特定的示例中，外部循环已经公开了太多的并行可能性，因为通常比起矩阵的大小，我们拥有的内核要少得多。其次，每个委托表达式都分配了一些内存来存放自由变量。这就是在我们最初的示例中只有一个分配，它的成本通过迭代得到分摊的原因。遗憾的是，在新的代码中，内部 Parallel.For 会在外部循环的每个迭代中执行堆分配。分配在 CLR 中非常有效，但与每个迭代中完成的工作量相比，仍是一个非常显著的成本。

请注意，由于那些迭代不是独立的，因此您不能并行内部循环。特别是，因为每个迭代都添加至了 result[i,j] 位置，所以存在争用现象。如果您并行此循环，两个迭代会同时将当前值读入寄存器，对它执行加法，然后写回结果，结果就是丢失一次加法！并行内部循环的唯一方法就是用锁来正确地保护加法。当然，我们并不建议真正这么做：即使您暂时忽略额外的分配，因为每个并发迭代都争用同一个锁，因此性能会遭受严重影响。稍后我们会在讨论聚合操作时回到这个话题。有关争用和锁的更多信息，请参阅 msdn.microsoft.com/msdnmag/issues/05/08/Concurrency/default.aspx.

射线跟踪器示例

射线跟踪器是一种简单但功能强大的方法，可生成图像逼真的呈现。但是该技术要求进行大量的计算。对于我们的库来说，射线跟踪器实际具备非常优秀的应用条件，因为每个射线都能并行计算。我们选择了一个现有的射线跟踪器，由 Luke Hoban 编写（请参阅 blogs.msdn.com/lukeh/archive/2007/04/03/a-ray-tracer-in-c-3-0.aspx），并对其进行了修改，以便用 TPL 并行运行。射线跟踪器会生成如图 3 中所示的图像，并且在 Parallel FX CTP 中用作示例。原来的射线跟踪器的核心循环会遍历最终图像的所有像素：

void Render(Scene scene, Color[,] rgb)
{
  for (int y = 0; y < screenHeight; y++)
  {
    for (int x = 0; x < screenWidth; x++) {
      rgb[x,y] = TraceRay(new Ray(scene,x,y));
    }
  }
}

因为每条射线能被独立跟踪，我们只需要改变一行原始代码即可实现并行：

void Render(Scene scene, Color[,] rgb)
{
  Parallel.For(0, screenHeight, delegate(int y)
  {
    for (int x = 0; x < screenWidth; x++) {
      rgb[x,y] = TraceRay(new Ray(scene,x,y));
    }
  });
}

在一台八核计算机上，原来的代码每秒能生成 1.7 帧 350 x 350 像素的图像。比较一下，在同一八处理器计算机上运行的并行版本每秒能生成 12 帧。这在八处理器的计算机上是 7 倍的加速，对于如此小的更改而言是相当好的结果。每秒 12 帧的速度足够制作一个在地板上跳动的球的流畅动画。而且因为这是一个非常简单的射线跟踪器，您可以进一步优化它，以获得更为流畅的动画。

动态工作分配

在使用线程池手动并行化循环时，开发人员通常最终做的就是静态划分工作。例如，在射线跟踪器中，图像通常会被平均分成多个部分，每个部分由单个线程进行处理。一般来说，因为实际工作负荷的分配也许不平均，所以这并不是一个好主意。例如，由于反射的原因，图像下部的计算时间要长两倍，那么处理图像上部的线程大部分时间都是在等待下部线程完成。即使工作是平均分配的，由于页面错误或系统上并发运行的其他进程，这样的问题也仍会发生。

为了很好地扩展到多个处理器，TPL 使用工作窃取技术将工作项目动态地适应和分配到工作线程上。库有一个任务管理器，默认情况下它会对每个处理器使用一个工作线程。这确保了 OS 执行的线程切换次数最少。每个工作线程有其各自等待完成的本地任务队列。每个工作线程通常只是把新任务推入队列中，并在任务完成时弹出工作。当其本地队列为空时，工作线程自己会寻找工作，尝试从其他工作线程的队列中“窃取”工作。

这里的优势在于工作线程之间几乎没有同步，因为工作队列是经过分配的，而且大多数操作对工作线程来说都是局部的，这对可伸缩性而言至关重要。此外，工作窃取具有已证实的良好缓存区域和工作分配属性。例如，如果工作负荷不平均，某个工作线程可能需要很长时间来完成某个特定任务，但是其他工作线程现在将从它的队列中窃取工作，保证所有处理器都处于忙碌状态。动态工作分配在典型的应用程序中至关重要，原因是很难预期某个任务需要多长时间才能完成。对于桌面系统（其中多个不同进程共享多个处理器，而且我们无法预期工作线程将得到的时间片）而言，情况尤其如此。

图 4 演示了正在使用四个线程的动态工作分配。它显示的射线跟踪器图像与图 3 中的相同，但这次每个工作线程使用不同的颜色来呈现其像素。您能看到，库在工作线程之间平均地分配工作，以动态适应工作负荷。

除了执行动态工作分配，库还能动态调整工作线程数量（如果工作线程被阻塞）。一些阻塞操作的例子是文件读取、等待按键以及检索用户名（因为这需要访问域上的网络）。如果某个任务在原因不明的情况下阻塞，性能会随着并发级别的降低而下降（但是程序仍然正常运行）。为了改善性能，库会自动跟踪以检测工作线程是否被阻塞，并在需要时插入额外的工作线程来保持并发级别。一旦操作解除阻塞，一些工作线程可能会注销，以便减少线程切换的成本。

聚合

一个 for 循环通常用于遍历某个域并将值聚合为单个结果。以下面将小于 100 的质数累加的迭代为例：

int sum = 0;
for(int i = 0; i < 100; i++) {
if (isPrime(i)) sum += i;
}

遗憾的是，我们无法按原样并行此循环，因为并行此循环会导致数据争用。每次迭代都在没有锁保护的情况下修改共享的 sum 变量. 如果两个并发迭代同时增加 sum，两者都有可能读取寄存器中的同一值、对其执行加法并写回各自的结果，结果就是丢失一次加法。正确的版本会使用锁来保护加法，如下所示：

int sum = 0;
Parallel.For(0, 100, delegate(int i) {
  if (isPrime(i)) {
    lock (this) { sum += i; }
  }
});

然而，因为所有的并行迭代都同时争用同一锁和同一内存区域 (sum)，程序现在又将遭遇性能问题。如果每个工作线程能够维持一个线程的局部 sum，而且仅在循环结束时将其增加到全局 sum，情况会得到改善。此模式由 Paral- lel.Aggregate 操作实现，这样我们将该示例重新编写为：

int sum = Parallel.Aggregate(0, 100, // iteration domain
              0,                     // initial value
              delegate(int i) { return (isPrime(i) ? i : 0) }, // apply
                                                      // on each element
              delegate(int x, int y) { return x+y; }  // combine results
          );

聚合操作有五个参数。前两个参数指定了迭代的域，该域也可以是枚举器。下一个参数是结果的初始值。接下来的两个参数是委托函数。第一个函数应用于每个元素，另一个用于组合元素结果。

库自动使用线程局部变量来计算线程局部结果（无任何锁定），仅在组合最终线程局部结果的时候使用一个锁。请记住，如果聚合是并行完成的，元素就有可能以不同于顺序聚合的次序进行组合。因此，组合委托函数必须是关联的，而且初始值必须是单位元素。

派生-联结并行性

另一种常见的并行模式是派生-联结并行性。作为示例，请考虑下面的顺序快速排序实现：

static void SeqQuickSort<T>(T[] domain, int lo, int hi) where T : IComparable<T>
{
  if (hi - lo <= Threshold) InsertionSort(domain, lo, hi);

  int pivot = Partition(domain, lo, hi);
  SeqQuickSort(domain, lo, pivot - 1);
  SeqQuickSort(domain, pivot + 1, hi);
}

该算法对元素类型 T 是通用的，只需要 T 实例能够比较即可。在一定的阈值下，算法回到插入排序，在元素数量较少时执行效果更佳。否则，我们将输入数组分成两部分，对各部分分别快速排序。因为每个排序都作用于数组的不同部分，因此两个排序可以并行执行。通过 Parallel.Do 方法可以方便地表达这种情形：

static void ParQuickSort<T>(T[] domain, int lo, int hi) where T : IComparable<T>
{
  if (hi - lo <= Threshold) InsertionSort(domain, lo, hi);

  int pivot = Partition(domain, lo, hi);
  Parallel.Do(
    delegate { ParQuickSort(domain, lo, pivot - 1); },
    delegate { ParQuickSort(domain, pivot + 1, hi); }
  );
}

Parallel.Do 方法是一个静态方法，把两个或更多委托当作参数并潜在地并行执行它们。因为快速排序是递归的，而且每次调用引入了许多并行任务，因此大多数并行性都得以公开。再次重申，因为库不能确保并行执行，大部分任务实际上是顺序执行的，这对于确保良好的性能是非常关键的。

任务和 Future

之前的示例都演示了结构化并行性，其中并行代码的范围是由词汇范围决定的。但并不是所有的并行算法都能以此方式来表达。庆幸的是，库对于普通并行任务也提供了支持：

class Task
{
  Task( Action action );

  void Wait();
  void Cancel();
  bool IsCompleted { get; }
  ...
}

在创建任务时，对其提供一个有可能并行执行的关联操作。该操作将在任务创建时至第一次调用 Wait 方法之间执行。关联操作可在另一个线程上并行执行，但是可以保证该操作不会在线程间迁移。这个保证非常有用，因为程序员能使用类似于 Windows 关键部分的线程仿射抽象，而不必担心（比方说）执行 LeaveCriticalSection 的线程与 EnterCriticalSection 有所不同。如果任务已经完成，Wait 会立即返回。

在关联操作中引发的任何异常都会存储在任务中，任何时候调用 Wait 时会再次引发。与之类似，Parallel.For 和 Parallel.Do 函数也累积引发的所有异常，并在所有任务完成时重新引发。这确保异常不会丢失，并能正确传播给相关者。

最后，通过调用 Cancel 可以取消任务以及在其关联的操作中创建的所有任务（子任务）。取消并不是强占式的，运行中的任务必须通过回调库配合退出工作。例如，这可以通过创建新任务或调用 Wait 方法来完成。如果父任务已经取消，这些库调用会引发一个（同步）OperationCanceled 异常来停止操作。

您可以将任务看成一个经过改进的线程池，其中工作项目返回一个可取消或等待的句柄，而且异常可得到传播。还有一种任务的变体，称为 future，其中关联的操作会计算某个结果：