可重复使用地并行大数据结构和算法

1、9种可重复使用的并行数据结构和算法目录倒计数锁存(Countdown Latch)可重用旋转等待 (Spin Wait)屏障(Barrier)阻塞队列受限缓冲区(Bounded Buffer)Thin事件无锁定LIFO堆栈循环分块并行分拆总结本专栏并未涉及很多公共语言运行库(CLR)功能的机制问题，而是更多介绍了如何有效使用您手头所具有的工具。身为一名程序员，必须做出很多决策，而选择正确的数据结构 和算法无疑是最常见的，也是最重要的决策之一。错误的选择可能导致程序无法运行，而大多数情况下，则决定了性能的好坏。 鉴于并行编程通常旨在改进性能，并且要难于串行编程，因此所作的选择对您程序的成功就更为

2、重要。在本专栏中，我们将介绍九种可重复使用的数据结构和算法，这些结构和算法是许多并|行程序所常用的，您应该能够轻松将它们应用到自己的.NET软件中。专栏中每个示例随附的代码都是可用的， 但尚未经过完全定型、测试和优化。这里列举的模式虽然并不详尽，但却代表了一些较为常见的模式。如您所见，很多示例都是互为补充的。在开始前，我想还是先介绍一些相关内容。Microsoft ? .NET Framework提供了几个现有的并发基元。虽然我要为您讲解如何构建自己的基元，但实际上现有基元是足以应付大多数情况的。我只是想说某些可选的方案有时也是有参考价值的。此外，了解这些技巧如何应用于实际操作也有助于加深您对

3、并行编程的整体理解。在开始讲解前，我假定您对现有基元已经有了一个基本的了解。您也可以参阅MSDN 杂志2005年8月版的文章“关于多线程应用程序：每个开发人员都应了解的内容”，以全面了解其概念。、倒计数锁存(Co un tdow n Latch)Semaphore之所以成为并发编程中一种较为知名的数据结构，原因是多方面的，而并不只是因为它在计算机科学领域有着悠久的历史(可以追溯到19世纪60年代的操作系统设计)。Semaphore只是一种带有一个计数字段的数据结构，它只支持两种操作：放置和取走(通常分别称为 P和V )。一次放置操作会增加一个 semaphore计数，而一次取走操作会减少一个s

4、emaphore计数。当semaphore计数为零时，除非执行一项并发的放置操作使计 数变为非零值，否则任何后续的取走尝试都将阻塞(等待)。这两种操作均为不可再分 (atomic) 操作，并发时不会产生错误，能够确保并发的放置和取走操作有序地进行。Windows具有基础内核和对 semaphore 对象的 Win32支持(请参阅 CreateSemaphore 和相关 API)，并且在.NET Framework 中这些对象可以通过 System.Threadi ng.Se map hore类公开到上层。Mutex和Monitor 所支持的临界区，通常被认为是一种特殊的semaphore，其计

5、数会在0和1之间来回切换，换句话说，是一个二进制的semaphore o另外还有一种"反向semaphore ”也是非常有用。也就是说，有时您需要数据结构能够等待数据结构计数归零。Fork/join式并行模式在数据并行编程中是极为常见的，其中由单个“主”线程控制执行若干“辅助”线程并等待线程执行完毕。在这类情况下，使用反向 semaphore会很有帮助。大多数时候，您其实并不想唤醒线程来修改计数。因此在这种情况下，我们将结构称为倒计数“锁存”，用来表示计数的减少，同时还表明一旦设置为“ Sig naled ” 状态，锁存将保持“ sig naled ”(这是一个与锁存相关的属性)。遗

6、憾的是，Win dows和.NETFramework均不支持这种数据结构。但令人欣慰的是，构建这种数据闭锁并不难。要构建倒计数锁存，只需将其计数器初始值设为n,并让每项辅助任务在完成时不可再分地将n减掉一个计数，这可以通过为减量操作加上“锁”或调用Interlocked.Decrement来实现。接下来，线程可以不执行取走操作，而是减少计数并等待计数器归零；而当线程被唤醒时，它就可以得知已经有n个信号向锁存注册。在 while (count != 0) 循环中，让等待的线程阻塞通常是不错的选择(这种情况下，您稍后将不得不使用事件)，而不是使用旋转。public class Coun tdow

7、nLatch private int m_rema in;private Eve ntWaitHa ndle m_eve nt;public Coun tdow nLatch(i nt count) m_rema in = count;m_eve nt = new Manu alResetEve nt(false);public void Sig nal()/ The last thread to sig nal also sets the eve nt.if (In terlocked.Decreme nt(ref m_remai n) = 0)m_eve nt.Set();public v

8、oid Wait（） m_eve nt.Wait On e（）;这看上去极为简单，但要正确运用还需要技巧。稍后我们将通过一些示例来讲解如何使用这种数据结构。请注意，此处所示基本实现还有很多可以改进地方，例如：在事件上调用WaitOne之前添加某种程度的旋转等待、缓慢分配事件而不是在构造器中进行分配（以防足够的旋转会避免出现阻塞，如本专栏稍后介绍的Th in Eve nt 演示的那样）、添加重置功能以及提供Dispose方法（以便在不再需要内部事件对象时将对象关闭）。】、可重用旋转等待 （Spin Wait）虽然忙碌等待（busy wait in g）更容易实现阻塞，但在某些情况下，您也许的确想

9、在退回到真正的等待状态前先旋转（spin）段时间。我们很难理解为何这样做会有帮助，而大多数人之所以一开始就避免旋转等待，是因为旋转看上去像是在做无用功；如果上下文切换（每当线程等待内核事件时都会发生）需要几千个周期（在Windows上确实是这样），我们称之为c,并且线程所等待的条件出现的时间少于2c周期时间（1c用于等待自身，1c用于唤醒），则旋转可以降低等待所造成的系统开销和滞后时间，从而提升算法的整体吞吐量和可伸缩性。如果您决定使用旋转等待，就必须谨慎行事。因为如果这样做，您可能需要注意很多问题，比如：要确保在旋转循环内调用Thread.SpinWait |，以提高Intel超线程技术的计

10、算机上硬件对其他硬件线程的可用性；偶尔使用参数1而非0来调用Thread.Sleep ，以避免优先级反向问题；通过轻微的回退（back-off）来引入随机选择，从而改善访问的局部性（假定调用方持续重读共享状态）并可能避免活锁；当然，在单CPU的计算机最好不要采用这种方法（因为在这种环境下旋转是非常浪费资源的）。Spin Wait类需要被定义为值类型，以便分配起来更加节省资源。现在，我们可以使用此算 法来避免前述Cou ntdow nLatch 算法中出现的阻塞。public struct Sp in Wait private int m_co unt;private static readon

11、ly bool s_isSingleProc =（Environmen t.ProcessorC ount = 1）;private const int s_yieldFreque ncy = 4000;private const int s_yieldO neFreque ncy = 3*s_yieldFreque ncy;public int Spin() int oldCo unt = m_co unt;/ On a si ngle-CPU mach ine, we en sure our coun ter is always/ a multiple of s_yieldFrequenc

12、y ' , so we yield every time./ Else, we just increment by one.m_co unt += (s_isS in gleProc ? s_yieldFreque ncy : 1);/If not a multiple of s_yieldFreque ncy' spin (w/ backoff).int coun tModFreque ncy = m_co unt % s_yieldFreque ncy;if (co un tModFreque ncy > 0)Thread.SpinWait(int)(1 + (cou

13、ntModFrequency * 0.05f);elseThread.Sleep(m_co unt <= s_yield On eFreque ncy ? 0 : 1);retur n oldCo unt;private void Yield()Thread.Sleep(m_co unt < s_yield On eFreque ncy ? 0 : 1);private const int s_sp inCount = 4000;public void Wait()Spi nWait s = new Spi nWait();while (m_remai n > 0)if (s

14、.Sp in() >= s_sp inCount) m_eve nt.Wait On e();不可否认，选择频率和旋转计数是不确定的。与Win32临界区旋转计数类似，我们应该根据测试和实验的结果来选择合理的数值，而且即使合理的数值在不同系统中也会发生变化。例如，根据 Microsoft Media Ce nter 和 Win dows kernel团队的经验， MSDN文档建议临界区旋转计数为 4,000 ，但您的选择可以有所不同。理想的计数取决于多种因素，包 括在给定时间等待事件的线程数和事件出现的频率等。大多数情况下，您会希望通过等待事件来消除显式让出时间，如锁存的示例中所示。您甚至

15、可以选择动态调整计数：例如，从中等数量的旋转开始，每次旋转失败就增加计数。一旦计数达到预定的最大值，就完全停止旋转并立即发出Wait One。逻辑如下所示：您希望立即增加达到预定的最大周期数，但却无法超过最大周期数。如果您发现此最大值不足以阻止上下文切换，那么立即执行上下文切换总的算来占用的资源更少。慢慢您就会希望旋转计数能够达到一个稳定的值。三、屏障(Barrier)屏障，又称集合点，是一种并发性基元，它无需另一“主”线程控制即可实现各线程之 间简单的互相协调。每个线程在到达屏障时都会不可再分地发出信号并等待。仅当所有n都到达屏障时，才允许所有线程继续。这种方法可用于协调算法(coopera

16、tive algorithms)，该算法广泛应用于科学、数学和图形领域。很多计算中都适合使用屏障，实际上，甚至CLR的垃圾收集器都在使用它们。屏障只是将较大的计算分割为若干较小的协作阶段(cooperative phase)，例如：const int P =；Barrier barrier = new Barrier(P); Data partiti ons = new DataP;/ Running on P separate threads in parallel:public void Body(i nt myln dex) FillMyPartitio n( partitio nsmy

17、I ndex);barrier.Await();ReadOtherPartition(partitionsP- myIndex - 1);barrier.Await();/ .您会很快发现在这种情况下是可以使用倒计数锁存的。每个线程都可以在调用Sig nal后立即调用 Wait，而不是调用Await ;在到达屏障后，所有线程都会被释放。但这里存在 一个问题：前面所讲的锁存并不支持多次重复使用同一对象，而这却是所有屏障都支持的一个常用属性。实际上，上述示例就需要使用此属性。您可以通过单独的屏障对象来实现这一 点，但这样做非常浪费资源；而由于所有线程每次只出现在一个阶段，因此根本无需多个屏障对象。

18、要解决这个问题，您可以使用相同的基础倒计数锁存算法来处理减少计数器计数、发信号指示事件、等待等方面的问题，从而将其扩展为可重复使用。要实现这一点，您需要使用所谓的感应反向屏障(se nse reversi ng barrier)，该屏障需要在“偶数”和"奇数”阶段之间 交替。在交替阶段需要使用单独的事件。using System;using System.Thread ing;public class Barrier private volatile int m_count;private int m_originalCount;private Even tWaitHa ndle m

19、_oddEve nt;private Even tWaitHa ndle m_evenEvent;private volatile bool m_se nse = false; / false=eve n, true=odd.public Barrier(i nt count) m_co unt = count;m_origi nalCo unt = count;m_oddEve nt = new Manu alResetEve nt(false);m_eve nEvent = new Manu alResetEve nt(false);public void Await() bool sen

20、se = m_se nse;/ The last thread to sig nal also sets the eve nt.if (m_co unt = 1 | In terlocked.Decreme nt(ref m_co unt) = 0) m_co unt = m_orig in alCo unt;m_se nse = !se nse; / Reverse the sen se.if (se nse = true) / oddm_eve nEven t.Reset();m_oddEve nt.Set(); else / eve nm_oddEve nt.Reset();m_eve

21、nEven t.Set(); else if (se nse =true) m_oddEve nt.Wait On e();elsem_eve nEven t.Wait On e();为何需要两个事件，其原因很难讲清楚。一种方法是在Await中先执行Set随后立即执行Reset，但这很危险，并会导致死锁。原因有二：第一，另一线程的m_count可能已减少，但在依次快速调用Set和Reset时，计数的值还不足以达到可在事件上调用WaitO ne。第二，虽然等待的线程可能已经能够调用Wait One，但可报警等待（如 CLR 贯使用的那些）可以中断等待，从而暂时从等待队列中删除等待的线程，以便运行

22、异步过程调用（APC）。等待的线程将始终看不到事件处于设置（set）状态。两种情况均会导致事件丢失，并可能导致死锁。针对奇数阶段和偶数阶段使用单独的事件即可避免这种情况。您可能想继续像Cou ntdow nLatch 中那样将旋转添加到 Barrier 。但如果您尝试这样做，可 能会遇到一个问题：一般情况下，旋转线程会开始旋转直到m_count归零。但通过上述实现，m_count的值实际上永远不会为零，除非最后一个线程将其重置为m_originalCount 。这种想当然的方法将导致一个或多个线程进行旋转（永远地），而其他线程则会在下一阶段阻塞（也是永远地）。解决方法很简单。您旋转，等待感应发

23、生变化public void Await() bool sense =:m_se nse;/ The last thread to sig nal also sets the event.if (m_co unt =1 | In terlocked.Decreme nt(ref m_co unt) = 0) m_co unt =:m_origi nalCo unt;m_se nse :=!se nse; / Reverse the sen se.if (se nse =：=true) / oddm_even Eve nt.Set(); m_oddEve nt.Reset(); else / e

24、ve n m_oddEve nt.Set(); m_even Eve nt.Reset(); else Spi nWait s = new Spi nWait();while (se nse = m_se nse) if (s.Sp in() >= s_sp inCount) if (se nse = true) m_oddEve nt.Wait On e();elsem_eve nEven t.Wait On e();由于所有线程都必须离开前一阶段的Await才可以完成下一阶段，因此可以确定，所有线程要么会观察到感应发生变化，要么最终等待事件并被唤醒。阻塞队列 在共享内存的体系结构中，

25、两项或多项任务间唯一的同步点通常是一个位于中枢的共享集合的数据结构。通常，一项或多项任务会负责为其他一项或多项任务生成要执行的“工作”， 我们称之为生产者/使用者关系(producer/consumer)。这类数据结构的简单同步通常是非 常简单的一使用Monitor 或ReaderWriterLock即可解决，但当缓冲区为空时，任务间的协调就会变得比较困难。要解决这个问题，通常需要使用阻塞队列。实际上，阻塞队列有几种稍微不同的变体，包括简单变体和复杂变体。在简单变量中，使用者仅在队列为空时才会阻塞，而在复杂变量中，每个生产者都正好“配有”一个使用者，也就是说，在使用者到达并开始处理排队项目之前

26、，生产者会被阻塞，同理， 在生产者交付项目前，所有使用者也会被阻塞。这时通常使用FIFO (先进先出)排序，但我们并不总是有必要这么做。我们也可以对缓冲区进行限制，这一点稍后会为大家介绍。由于稍后将要介绍的受限缓冲区包含更为简单的“为空时阻塞” (blocki ng-whe n-empty)行为，因此我们这里仅对配对变量做一了解。要实现这个目的，我们只需封装一个简单的Queue<T>,上方保持同步。那么到底需要何种同步？每当线程对元素进行排队时，在返回前都会等待使用者取消元素排队。当线程取消元素排队时，如果发现缓冲区为空，则必须等待新元素的进入。当然，取消排队后，使用者必须通知生产

27、者已取到其项目(请参见图5)。Figure 5 阻塞队列复制代码class Cell<T> in ternal T m_obj;internal Cell(T obj) m_obj = obj; public class Block in gQueue<T> private Queue<Cell<T>> m_queue = new Queue<Cell<T>>();public void Enq ueue(T obj) Cell<T> c = new Cell<T>(obj);lock (m_que

28、ue) m_queue.E nq ueue(c);Mon itor.Pulse(m_queue);Mon itor.Wait(m_queue);public T Dequeue() Cell<T> c;lock (m_queue) while (m_queue.Co unt = 0)Mon itor.Wait(m_queue);c = m_queue.Dequeue();Mon itor.Pulse(m_queue);return c.m_obj;请注意，我们可以在Enqueue方法中依次调用 Pulse 和Wait ,类似地，在Dequeue方法 中依次调用 Wait和Pulse

29、 。只有在释放监视器之后，才会对内部事件进行设置，而由于监视器的这种实现方法，会导致某个线程调度ping-pong。我们可能会想，也许可以使用 Win32事件来构建一个更加优化的通知机制。但是，使用这类完善的Win32事件会带来巨大开销，尤其是使用它们时所进行的成本分配和内核跳转(kernel tran siti on s)，因此还是考虑其他选择吧。您可以像 CLR的ReaderWriterLock那样将这些时间集中到池中，也可以像ThinEvent类型(稍后介绍)一样缓慢分配它们。这种实现方法也是有弊端的，即要为每个新元素分配对象，备选方法可能也会将这些对象加入到池中，但同样会带来其他麻烦。

30、受限缓冲区(Bounded Buffer)某些类别的队列中会出现资源使用问题。如果生产者任务创建项目的速度快于使用者处理项目的速度，则系统的内存使用将不受限制。为了说明这个问题，我们假设一个系统，单个生产者每秒钟可将 50个项目排入队列，而使 用者每秒钟只能使用 10个项目。首先，这样会打乱系统的平衡性， 而且对于一对一的生产 者一使用者配置无法进行很好的调整。只需一分钟，就会有2,400个项目堆积在缓冲区中。假设这些项目中每个项目使用10KB内存，那将意味着缓冲区本身就会占用24MB内存。一小时后，这个数字将飙升至1GB以上。解决这一问题的一个方法是调整生产者线程与使用者线程的比例，在上述情

31、况中，要将比例调整为一个生产者对应五个使用者。但是到达速度通常是不稳定的，这会导致系统周期性的不稳定，并带来一些明显的问题，简单的固定比例是无法解决这个问题的。服务器上的程序通常是长时间运行的，人们希望程序能够长期保持良好的运行状态，但如果内存使用不受限，就会造成不可避免的混乱，还可能导致必须定期回收服务器进程的情况。受限缓冲区允许您对生产者强制阻塞前缓冲区可能达到的大小进行限制。阻塞生产者可让使用者有机会“赶上”(通过允许其线程接收调度时间片)，同时还能够解决内存使用量增加 的问题。我们的方法还是仅封装Queue<T>，同时添加两个等待条件和两个事件通知条件：生产者在队列满时等待

32、，直至队列变为非满， 而使用者在队列空时等待，直至变为非空；生产者在生产出项目后通知等待的使用者，而使用者也会在取走项目后通知生产者，如图6中所示。Figure 6 受限缓冲区(Bounded Buffer)复制代码public class Boun dedBuffer<T> private Queue<T> m_queue = new Queue<T>();private int m_con sumersWaiti ng;private int m_producersWaiti ng;private const int s_maxBufferSize =

33、128;public void Enq ueue(T obj) lock (m_queue) while (m_queue.Co unt = (s_maxBufferSize - 1) m_producersWaiti ng+;Mon itor.Wait(m_queue);m_producersWaiti ng-;m_queue.E nq ueue(obj);if (m_c on sumersWaiti ng > 0)Mo nitor.PulseAII(m_queue);public T Dequeue。T e;lock (m_queue) while (m_queue.Co unt =

34、 0) m_con sumersWaiti ng+;Mon itor.Wait(m_queue);m_con sumersWaiti ng-;e = m_queue.Dequeue();if (m_producersWaiti ng > 0)Mo nitor.PulseAII(m_queue);return e;这里我们再一次使用了一种比较天真的方法。但是我们确实优化了对 PulseAll 的调用(因为它们耗费的资源很多)，方法是使用m_consumersWaiting 和m_producersWaiting这两个计数器并仅就计数器值是否为零发出信号。除此以外，我们还可以再做进一步的改进

35、。 例如，共享与此类似的单个事件可能会唤醒过多线程：如果使用者将队列大小降为0,并且生产者和使用者同时处于等待状态，显然您只希望唤醒生产者(至少开始时要这么做)。此实现将以先进先出的规则处理所有等待者，这意味着您可能需要在任一生产者运行前唤醒使用者，这样做仅仅是为了让使用者发现队列为空，然后再次等待。令人欣慰的是，最终出现生 产者和使用者同时等待的情况是很少见的，但其出现也是有规律的，而且受限区域较小。Thin事件与Monitor.Wait、Pulse和PulseAll 相比，Win32事件有一个很大的优势：它们是“粘滞的”(sticky)。也就是说，一旦有事件收到信号，任何后续等待都将立即取

36、消阻止，即使 线程在信号发出前尚未开始等待。如果没有这个功能，您就需要经常编写一些代码将所有等 待和信号通知严格地限制在临界区域内，由于Windows计划程序总是会提升已唤醒线程的优先级，因此这些代码的效率很低；如果不采取这种方法，您就必须使用某种技巧型很强、 容易出现竞态条件的代码。作为替代方法，您可以使用“thin事件”，这是一种可重用数据结构，可在阻塞前短暂地旋转，仅在必要时才缓慢分配Win32事件，否则允许您执行类似手动重置事件的行为。换句话说，它可以对那些复杂的容易导致竞态条件的代码进行封装，这样您就不必在整个基本代码内散播它。此示例依赖于 Vance Morrison的文章中描述的

37、一些内存模型保证，使用的 时候要多加小心（请参见图7 ）。Figure 7 Th in 事件复制代码public struct ThinEvent private int m_state; / 0 means un set, 1 means set.private Even tWaitHa ndle m_even tObj;private const int s_sp inCount = 4000;public void Set() m_state = 1;Thread.MemoryBarrier(); / required.if (m_eve ntObj != null)m_eve ntOb

38、j.Set();public void Reset() m_state = 0;if (m_eventObj != null)m_even tObj.Reset();public void Wait() Spi nWait s = new Spi nWait();while (m_state = 0) if (s.Sp in() >= s_sp inCount) if (m_eventObj = null) Manu alResetEve nt n ewEve nt =new Manu alResetEve nt(m_state = 1);if (In terlocked.Compare

39、Excha nge<Eve ntWaitHa ndle>(ref m_eve ntObj, n ewEve nt, n ull) = n ull) /If some one set the flag before see ing the new/ eve nt obj, we must en sure it' s bee n set.if (m_state = 1)m_eve ntObj.Set(); else / Lost the race w/ ano ther thread. Just use/ its event.n ewEve nt.Close();m_eve n

40、tObj.WaitO ne();这基本上反映了 m_state 变量中的事件状态，其中值为0表示未设置，值为1表示已设置。现在，等待一个已设置事件耗费资源是很低的；如果m_state 在Wait例程的入口处值为1，我们会立即返回，无需任何内核跳转。但如果线程在事件设置完毕之前进入等待状态，处理上就需要很多技巧。要等待的首个线程必须分配一个新的事件对象，并对其进行比较后交换至m_eventObj字段中；如果 CAS失败，则意味着其他等待者对事件进行了初始 化，所以我们只可重复使用它；否则就必须重新检查自上次看到m_state 后其值是否发生更改。不然的话，m_state的值也许会为1，那么m_e

41、ventObj就无法收到信号通知，这将 导致在调用Wait One时出现死锁。调用Set的线程必须首先设置m_state，随后如果发现值为非空的m_eventObj，就会调用其上的 Set。这里需要两个内存屏障：需要注意的是切 不可提前移动m_state 的第二次读取，我们会使用In terlocked.CompareExcha nge设置m_eventObj来保证这点；在写入 m_eventObj之前，不可移动Set中的对 m_eventObj的 读取(这是一种在某些In tel和AMD处理器以及CLR 2.0内存模型上出现的奇怪的合法转换，并未显式调用Thread.MemoryBarrie

42、r )。并行重置事件通常是不安全的，因此调用方需要进行额外的同步化。现在您可以轻松在其他地方使用它，例如在上述的Cou ntdow nLatch 示例和队列示例中，而且通常这样做可以大幅度提升性能，尤其是当您可以巧妙地运用旋转时。我们上面介绍了一个技巧性很强的实现方式。请注意，您可以使用单个标志和监视器来实现自动和手动重置类型，这远比本示例简单(但效率方面有时会不及本例)。无锁定LIFO堆栈使用锁定构建线程安全的集合是相当直接明了的，即使限制和阻塞方面会有些复杂(如上面看到的)。但是，当所有协调均通过简单的后进先出(LIFO)堆栈数据结构实现时，使用锁定的开销会比实际所需的高。线程的临界区(即

43、保持锁定的时间)有开始点和结束点，其持续时间可能会跨越许多指令的有效期。保持锁定可阻止其他线程同时进行读取和写入操作。这样做可以实现序列化，这当然是我们所想要的结果，但严格地讲，这种序列化要比我们所 需的序列化强很多。 我们的目的是要在堆栈上推入和弹出元素，而这两项操作只需通过常规读取和单一的比较一交换写入即可实现。我们可以利用这点来构建伸缩性更强的无锁定堆 栈，从而不会强制线程进行没必要的等待。我们的算法工作方式如下。使用有链接的列表代表堆栈，其标头代表堆栈的顶端，并存储在 m_head字段中。在将一个新项推入堆栈时，要构造一个新节点，其值等于您要推入堆栈的 值，然后从本地读取 m_head

44、字段，并将其存储至新节点的m_next字段，随后执行不可再分的In terlocked.CompareExcha nge来替换堆栈当前的头部。如果顶端自首次读取后在序列中的任何点发生更改，CompareExcha nge都会失败，并且线程必须返回循环并再次尝试整个序列。弹出同样是比较直截了当的。读取m_head并尝试将其与 m_next引用的本地副本交换；如果失败，您只需一直尝试，如图8中所示。Win32提供了一种名为 SList的类比数据结构，其构建方法采用了类似的算法。Figure 8无锁定堆栈复制代码public class LockFreeStack<T> private

45、volatile StackNode<T> m_head;public void Push(T item) StackNode<T> node = new StackNode<T>(item);StackNode<T> head;do head = m_head;no de.m_ next = head; while (m_head != head | In terlocked.CompareExcha nge(ref m_head, no de, head) != head);public T Pop() StackNode<T>

46、head;Spi nWait s = new Spi nWait();while (true) StackNode<T> next;do head = m_head;if (head = n ull) goto emptySp in;n ext = head.m_ next; while (m_head != head | In terlocked.CompareExcha nge(ref m_head, n ext, head) != head);break;emptySpi n:s.Spi n();retur n head.m_value;class StackNode<

47、T> in ternal T m_value;internal StackNode<T> m_n ext;internal StackNode(T val) m_value = val; 请注意，这是理想的并发控制的一种形式：无需阻止其他线程存取数据，只需抱着会在争用中“胜出”的信念继续即可。如果事实证明无法 “胜出”，您将会遇到一些变化不定的问题， 例如活锁。这种设计方案还表明您无法确保能够实现FIFO调度。根据概率统计，系统中所有线程都将向前推进。而实际上从整体来看，系统进度总是向前推进的，因为其中一个线程的失败总是意味着至少有一个其他线程是推进的(这是调用该“无锁定”的

48、一个条件)。有 些情况下，当 CompareExchange无法避免m_head大量争用内存时，使用指数回退算法会 很有用。同时，我们还对堆栈变空的情况采取了相当直截了当的方法。我们只是一直旋转，等待有新项目推入。将Pop重新写入处于非等待状态的TryPop是很简单易懂的，但是要利用事件 进行等待则会有一些复杂。这两个功能都是十分重要的，所以留给那些喜欢动手的读者作为 练习之用。我们为每个Push都分配了对象，这让我们无需担心所谓的ABA问题。在内部重复使用已从列表中弹出的节点就会导致ABA问题。开发人员有时会尝试将节点集中到池中以减少对象分配的数目，但这样做是有问题的：结果会是，即使对m_h

49、ead执行了大量干扰性写入操作，一个不可再分割的操作也可以实现，但实际上却是不正确的。（例如：线程1读取节点A，然后由线程2将节点A删除并置入池中；线程2将节点B作为新的头推入，然后 节点A从池中返回到线程 2并被推入；随后，线程 1会成功执行CompareExchange，但 现在作为头的A已经与先前所读取的不同了。）尝试以本机C/C+编写此算法时也会出现 类似问题；因为一旦地址被释放，内存分配器就会立即重复使用这些地址，节点会被弹出并释放，然后该节点的地址会被用于新的节点分配，结果导致与上面相同的问题。接下来我们不再讨论ABA，有关它的详细介绍我们可以从其他的资源获得。最后，可以使用类似的

50、无锁定技术编写一个FIFO队列。它的优势在于并行推入和弹出的线程之间并不一定发生冲突，而在上述LockFreeStack 中，推入者和弹出者始终会争用同一m_head字段。然而，这种算法相当复杂，如果您好奇的话，我推荐您阅读 Maged M. Michael 和 Michael L. Scott 在 1996 年撰写的文章"Simple, Fast, and PracticalNon-Blockingand Block ing Con curre nt Queue Algorithms”。循环分块循环分块是指对循环的输入范围或数据进行分区，并将每个分区分配给单独的线程以实现并发。这是

51、某些编程模型 （例如OpenMP实现并行性的一项基本技术 （请参阅Kang Su Gatlin 的MSDN杂志文章），通常称为并行 Forall 循环，是从高性能的 FORTRAN术语中得到 启发而创造的。无论如何，范围都只是一组索引：复制代码for (int i = 0; i < c; i+) . 或者是一组数据:复制代码foreach （T e in list） . 我们都可以设计分区技术以提供分块的循环。您可能想应用多种特定于数据结构的分区技术，这些技术确实很多，本专栏无法一一列举。因此这里我们只关注一种常用技术，可用于将数组中各种非连续范围的元素分配到每个分区。我们只需计算跨距的

52、值（它大约等于元素数目除以分区数目的值），并用它来计算连续 范围（参见图9 ）。当然尽管其他方法是有效、有用并在某些情况下有必需的，但当输入内 容为值类型数组时，此方法可以实现较好的空间局部性。Figure 9 循环分块复制代码public static void ForAII(i nt from, int to, Action<int> a, int p) ForAII< int>(n ull, from, to, n ull, a, p);public static void ForAII<T>(IList<T> data, Acti on&

53、lt;T> a, int p) ForAll<T>(data, 0, data.Count, a, null, p);private static void ForAII<T>(IList<T> data, int from, int to,Action<T> a0, Action<int> a1, int p) int size = from - to;int stride = (size + p - 1) / p;Coun tdow nLatch latch = new Coun tdow nLatch(p);for (in

54、t i = 0; i < p; i+) in t idx = i;ThreadPool.QueueUserWorkltem(delegate int end = Math.Mi n(size, stride * (idx + 1);for (int j = stride * idx; j < en d; j+) if (data != null) a0(dataj);elsea1(j);latch.Sig nal(););Iatch.Wait();我们这里提供了两种公共版的ForAII : 种接受一定范围内的数字，另一种接受IList<T>(类似于C#中的Foreach

55、循环)。两者都转发到同一帮助程序重载，重载可调用从给定索引的列表传递元素这一操作，或者传递索引本身。您应使用第一个重载（通常在此加入一个普通For循环）。例如，下面的代码复制代码for （int i = 0; i < 10; i+） S; 将变为：复制代码Parallel.ForAII(0, 10, delegated nt i) S; ,Environmen t.ProcessorCou nt);现在可以使用第二个重载(通常在此加入一个 C# foreach 循环)，这样,复制代码List<T> list =.;foreach (T e in list) S; 将变为：复制

56、代码Parallel.ForAll（list, delegate（T e） S; ,Environmen t.ProcessorCou nt）;您需要谨慎以确保S中的语句不会写入共享内存；否则您需要向并行版本添加合适的同步操作。当然可以编写相应版本来支持IEnumerable<T>、以其他方式对迭代空间进行分区等（为了节省空间，本专栏省略了这些内容）。在本示例中，在n项子任务的持续时间内调用线程被“浪费”了。更好的方法是使用调用线程来运行其中一项任务，然后在其完成时加入其他任务。没有必要通过扩展 ForAII 方法来执行此操作。并行分拆有一类操作可使用分拆（又称为折叠或聚合）来执行，其中会以某种方式将许多值进行组合 以便生成单一输出。 一般分拆的工作方式