1、可重复使用的并行数据结构与算法9 种可重复使用的并行数据结构和算法目录 倒计数锁存 (Countdown Latch) 可重用旋转等待 (Spin Wait) 屏障 (Barrier) 阻塞队列 受限缓冲区 (Bounded Buffer) Thin 事件 无锁定 LIFO 堆栈 循环分块 并行分拆 总结 本专栏并未涉及很多公共语言运行库 (CLR) 功能的机制问题，而是更多介绍了如何有效使用您手头所具有的工具。身为一名程序员，必须做出很多决策，而选择正确的数据结构和算法无疑是最常见的，也是最重要的决策之一。错误的选择可能导致程序无法运行，而大多数情况下，则决定了性能的好坏。鉴于并行编程通常旨
2、在改进性能，并且要难于串行编程，因此所作的选择对您程序的成功就更为重要。在本专栏中，我们将介绍九种可重复使用的数据结构和算法，这些结构和算法是许多并行程序所常用的，您应该能够轻松将它们应用到自己的 .NET 软件中。专栏中每个示例随附的代码都是可用的，但尚未经过完全定型、测试和优化。这里列举的模式虽然并不详尽，但却代表了一些较为常见的模式。如您所见，很多示例都是互为补充的。在开始前，我想还是先介绍一些相关内容。Microsoft .NET Framework 提供了几个现有的并发基元。虽然我要为您讲解如何构建自己的基元，但实际上现有基元是足以应付大多数情况的。我只是想说某些可选的方案有时也是有
3、参考价值的。此外，了解这些技巧如何应用于实际操作也有助于加深您对并行编程的整体理解。在开始讲解前，我假定您对现有基元已经有了一个基本的了解。您也可以参阅MSDN 杂志2005 年 8 月版的文章“关于多线程应用程序：每个开发人员都应了解的内容”，以全面了解其概念。一、倒计数锁存 (Countdown Latch)Semaphore 之所以成为并发编程中一种较为知名的数据结构，原因是多方面的，而并不只是因为它在计算机科学领域有着悠久的历史（可以追溯到 19 世纪 60 年代的操作系统设计）。Semaphore 只是一种带有一个计数字段的数据结构，它只支持两种操作：放置和取走（通常分别称为 P 和
4、 V）。一次放置操作会增加一个 semaphore 计数，而一次取走操作会减少一个 semaphore 计数。当 semaphore 计数为零时，除非执行一项并发的放置操作使计数变为非零值，否则任何后续的取走尝试都将阻塞（等待）。这两种操作均为不可再分 (atomic) 操作，并发时不会产生错误，能够确保并发的放置和取走操作有序地进行。Windows具有基础内核和对 semaphore 对象的 Win32支持（请参阅 CreateSemaphore 和相关 API），并且在 .NET Framework 中这些对象可以通过 System.Threading.Semaphore 类公开到上层。M
5、utex 和 Monitor 所支持的临界区，通常被认为是一种特殊的 semaphore，其计数会在 0 和 1 之间来回切换，换句话说，是一个二进制的 semaphore。另外还有一种“反向 semaphore”也是非常有用。也就是说，有时您需要数据结构能够等待数据结构计数归零。Fork/join 式并行模式在数据并行编程中是极为常见的，其中由单个“主”线程控制执行若干“辅助”线程并等待线程执行完毕。在这类情况下，使用反向 semaphore 会很有帮助。大多数时候，您其实并不想唤醒线程来修改计数。因此在这种情况下，我们将结构称为倒计数“锁存”，用来表示计数的减少，同时还表明一旦设置为“Si
6、gnaled”状态，锁存将保持“signaled”（这是一个与锁存相关的属性）。遗憾的是，Windows 和 .NET Framework 均不支持这种数据结构。但令人欣慰的是，构建这种数据闭锁并不难。要构建倒计数锁存，只需将其计数器初始值设为 n，并让每项辅助任务在完成时不可再分地将 n 减掉一个计数，这可以通过为减量操作加上“锁”或调用 Interlocked.Decrement 来实现。接下来，线程可以不执行取走操作，而是减少计数并等待计数器归零；而当线程被唤醒时，它就可以得知已经有 n 个信号向锁存注册。在 while (count != 0) 循环中，让等待的线程阻塞通常是不错的选择
7、（这种情况下，您稍后将不得不使用事件），而不是使用旋转。public class CountdownLatch private int m_remain; private EventWaitHandle m_event; public CountdownLatch(int count) m_remain = count; m_event = new ManualResetEvent(false); public void Signal() / The last thread to signal also sets the event. if (Interlocked.Decrement(ref
8、 m_remain) = 0) m_event.Set(); public void Wait() m_event.WaitOne(); 这看上去极为简单，但要正确运用还需要技巧。稍后我们将通过一些示例来讲解如何使用这种数据结构。请注意，此处所示基本实现还有很多可以改进地方，例如：在事件上调用 WaitOne 之前添加某种程度的旋转等待、缓慢分配事件而不是在构造器中进行分配（以防足够的旋转会避免出现阻塞，如本专栏稍后介绍的 ThinEvent 演示的那样）、添加重置功能以及提供 Dispose 方法（以便在不再需要内部事件对象时将对象关闭）。二、可重用旋转等待 (Spin Wait)虽然忙碌等
9、待 (busy waiting) 更容易实现阻塞，但在某些情况下，您也许的确想在退回到真正的等待状态前先旋转 (spin) 一段时间。我们很难理解为何这样做会有帮助，而大多数人之所以一开始就避免旋转等待，是因为旋转看上去像是在做无用功；如果上下文切换（每当线程等待内核事件时都会发生）需要几千个周期（在 Windows 上确实是这样），我们称之为 c，并且线程所等待的条件出现的时间少于 2c 周期时间（1c 用于等待自身，1c 用于唤醒），则旋转可以降低等待所造成的系统开销和滞后时间，从而提升算法的整体吞吐量和可伸缩性。如果您决定使用旋转等待，就必须谨慎行事。因为如果这样做，您可能需要注意很多问
10、题，比如：要确保在旋转循环内调用 Thread.SpinWait，以提高 Intel 超线程技术的计算机上硬件对其他硬件线程的可用性；偶尔使用参数 1 而非 0 来调用 Thread.Sleep，以避免优先级反向问题；通过轻微的回退 (back-off) 来引入随机选择，从而改善访问的局部性（假定调用方持续重读共享状态）并可能避免活锁；当然，在单 CPU 的计算机最好不要采用这种方法（因为在这种环境下旋转是非常浪费资源的）。SpinWait 类需要被定义为值类型，以便分配起来更加节省资源。现在，我们可以使用此算法来避免前述 CountdownLatch 算法中出现的阻塞。public stru
11、ct SpinWait private int m_count; private static readonly bool s_isSingleProc = (Environment.ProcessorCount = 1); private const int s_yieldFrequency = 4000; private const int s_yieldOneFrequency = 3*s_yieldFrequency; public int Spin() int oldCount = m_count; / On a single-CPU machine, we ensure our c
12、ounter is always / a multiple of s_yieldFrequency, so we yield every time. / Else, we just increment by one. m_count += (s_isSingleProc ? s_yieldFrequency : 1); / If not a multiple of s_yieldFrequency spin (w/ backoff). int countModFrequency = m_count % s_yieldFrequency; if (countModFrequency 0) Thr
13、ead.SpinWait(int)(1 + (countModFrequency * 0.05f); else Thread.Sleep(m_count = s_yieldOneFrequency ? 0 : 1); return oldCount; private void Yield() Thread.Sleep(m_count 0) if (s.Spin() = s_spinCount) m_event.WaitOne(); 不可否认，选择频率和旋转计数是不确定的。与 Win32 临界区旋转计数类似，我们应该根据测试和实验的结果来选择合理的数值，而且即使合理的数值在不同系统中也会发生变化
14、。例如，根据 Microsoft Media Center 和 Windows kernel 团队的经验，MSDN 文档建议临界区旋转计数为 4,000 ，但您的选择可以有所不同。理想的计数取决于多种因素，包括在给定时间等待事件的线程数和事件出现的频率等。大多数情况下，您会希望通过等待事件来消除显式让出时间，如锁存的示例中所示。您甚至可以选择动态调整计数：例如，从中等数量的旋转开始，每次旋转失败就增加计数。一旦计数达到预定的最大值，就完全停止旋转并立即发出 WaitOne。逻辑如下所示：您希望立即增加达到预定的最大周期数，但却无法超过最大周期数。如果您发现此最大值不足以阻止上下文切换，那么立即
15、执行上下文切换总的算来占用的资源更少。慢慢您就会希望旋转计数能够达到一个稳定的值。三、屏障 (Barrier)屏障，又称集合点，是一种并发性基元，它无需另一“主”线程控制即可实现各线程之间简单的互相协调。每个线程在到达屏障时都会不可再分地发出信号并等待。仅当所有 n 都到达屏障时，才允许所有线程继续。这种方法可用于协调算法 (cooperative algorithms)，该算法广泛应用于科学、数学和图形领域。很多计算中都适合使用屏障，实际上，甚至 CLR 的垃圾收集器都在使用它们。屏障只是将较大的计算分割为若干较小的协作阶段 (cooperative phase)，例如：const int
16、P = .;Barrier barrier = new Barrier(P);Data partitions = new DataP;/ Running on P separate threads in parallel:public void Body(int myIndex) FillMyPartition(partitionsmyIndex); barrier.Await(); ReadOtherPartition(partitionsP myIndex - 1); barrier.Await(); / .您会很快发现在这种情况下是可以使用倒计数锁存的。每个线程都可以在调用 Signal
17、 后立即调用 Wait，而不是调用 Await；在到达屏障后，所有线程都会被释放。但这里存在一个问题：前面所讲的锁存并不支持多次重复使用同一对象，而这却是所有屏障都支持的一个常用属性。实际上，上述示例就需要使用此属性。您可以通过单独的屏障对象来实现这一点，但这样做非常浪费资源；而由于所有线程每次只出现在一个阶段，因此根本无需多个屏障对象。要解决这个问题，您可以使用相同的基础倒计数锁存算法来处理减少计数器计数、发信号指示事件、等待等方面的问题，从而将其扩展为可重复使用。要实现这一点，您需要使用所谓的感应反向屏障 (sense reversing barrier)，该屏障需要在“偶数”和“奇数”阶
18、段之间交替。在交替阶段需要使用单独的事件。using System;using System.Threading;public class Barrier private volatile int m_count; private int m_originalCount; private EventWaitHandle m_oddEvent; private EventWaitHandle m_evenEvent; private volatile bool m_sense = false; / false=even, true=odd. public Barrier(int count) m
19、_count = count; m_originalCount = count; m_oddEvent = new ManualResetEvent(false); m_evenEvent = new ManualResetEvent(false); public void Await() bool sense = m_sense; / The last thread to signal also sets the event. if (m_count = 1 | Interlocked.Decrement(ref m_count) = 0) m_count = m_originalCount
20、; m_sense = !sense; / Reverse the sense. if (sense = true) / odd m_evenEvent.Reset(); m_oddEvent.Set(); else / even m_oddEvent.Reset(); m_evenEvent.Set(); else if (sense = true) m_oddEvent.WaitOne(); else m_evenEvent.WaitOne(); 为何需要两个事件，其原因很难讲清楚。一种方法是在 Await 中先执行 Set 随后立即执行 Reset，但这很危险，并会导致死锁。原因有二：第
21、一，另一线程的 m_count 可能已减少，但在依次快速调用 Set 和 Reset 时，计数的值还不足以达到可在事件上调用 WaitOne。第二，虽然等待的线程可能已经能够调用 WaitOne，但可报警等待（如 CLR 一贯使用的那些）可以中断等待，从而暂时从等待队列中删除等待的线程，以便运行异步过程调用 (APC)。等待的线程将始终看不到事件处于设置 (set) 状态。两种情况均会导致事件丢失，并可能导致死锁。针对奇数阶段和偶数阶段使用单独的事件即可避免这种情况。您可能想继续像 CountdownLatch 中那样将旋转添加到 Barrier。但如果您尝试这样做，可能会遇到一个问题：一般情
22、况下，旋转线程会开始旋转直到 m_count 归零。但通过上述实现，m_count 的值实际上永远不会为零，除非最后一个线程将其重置为 m_originalCount。这种想当然的方法将导致一个或多个线程进行旋转（永远地），而其他线程则会在下一阶段阻塞（也是永远地）。解决方法很简单。您旋转，等待感应发生变化public void Await() bool sense = m_sense; / The last thread to signal also sets the event. if (m_count = 1 | Interlocked.Decrement(ref m_count) =
23、0) m_count = m_originalCount; m_sense = !sense; / Reverse the sense. if (sense = true) / odd m_evenEvent.Set(); m_oddEvent.Reset(); else / even m_oddEvent.Set(); m_evenEvent.Reset(); else SpinWait s = new SpinWait(); while (sense = m_sense) if (s.Spin() = s_spinCount) if (sense = true) m_oddEvent.Wa
24、itOne(); else m_evenEvent.WaitOne(); 由于所有线程都必须离开前一阶段的 Await 才可以完成下一阶段，因此可以确定，所有线程要么会观察到感应发生变化，要么最终等待事件并被唤醒。阻塞队列在共享内存的体系结构中，两项或多项任务间唯一的同步点通常是一个位于中枢的共享集合的数据结构。通常，一项或多项任务会负责为其他一项或多项任务生成要执行的“工作”，我们称之为生产者/使用者关系 (producer/consumer)。这类数据结构的简单同步通常是非常简单的 使用 Monitor 或 ReaderWriterLock 即可解决，但当缓冲区为空时，任务间的协调就会变得
25、比较困难。要解决这个问题，通常需要使用阻塞队列。实际上，阻塞队列有几种稍微不同的变体，包括简单变体和复杂变体。在简单变量中，使用者仅在队列为空时才会阻塞，而在复杂变量中，每个生产者都正好“配有”一个使用者，也就是说，在使用者到达并开始处理排队项目之前，生产者会被阻塞，同理，在生产者交付项目前，所有使用者也会被阻塞。这时通常使用 FIFO（先进先出）排序，但我们并不总是有必要这么做。我们也可以对缓冲区进行限制，这一点稍后会为大家介绍。由于稍后将要介绍的受限缓冲区包含更为简单的“为空时阻塞”(blocking-when-empty) 行为，因此我们这里仅对配对变量做一了解。要实现这个目的，我们只需
26、封装一个简单的 Queue，上方保持同步。那么到底需要何种同步？每当线程对元素进行排队时，在返回前都会等待使用者取消元素排队。当线程取消元素排队时，如果发现缓冲区为空，则必须等待新元素的进入。当然，取消排队后，使用者必须通知生产者已取到其项目（请参见图 5）。Figure5阻塞队列 复制代码 class Cell internal T m_obj; internal Cell(T obj) m_obj = obj; public class BlockingQueue private QueueCell m_queue = new QueueCell(); public void Enqueu
27、e(T obj) Cell c = new Cell(obj); lock (m_queue) m_queue.Enqueue(c); Monitor.Pulse(m_queue); Monitor.Wait(m_queue); public T Dequeue() Cell c; lock (m_queue) while (m_queue.Count = 0) Monitor.Wait(m_queue); c = m_queue.Dequeue(); Monitor.Pulse(m_queue); return c.m_obj; 请注意，我们可以在 Enqueue 方法中依次调用 Pulse
28、 和 Wait，类似地，在 Dequeue 方法中依次调用 Wait 和 Pulse。只有在释放监视器之后，才会对内部事件进行设置，而由于监视器的这种实现方法，会导致某个线程调度 ping-pong。我们可能会想，也许可以使用 Win32 事件来构建一个更加优化的通知机制。但是，使用这类完善的 Win32 事件会带来巨大开销，尤其是使用它们时所进行的成本分配和内核跳转 (kernel transitions)，因此还是考虑其他选择吧。您可以像 CLR 的 ReaderWriterLock 那样将这些时间集中到池中，也可以像 ThinEvent 类型（稍后介绍）一样缓慢分配它们。这种实现方法也是
29、有弊端的，即要为每个新元素分配对象，备选方法可能也会将这些对象加入到池中，但同样会带来其他麻烦。受限缓冲区 (Bounded Buffer)某些类别的队列中会出现资源使用问题。如果生产者任务创建项目的速度快于使用者处理项目的速度，则系统的内存使用将不受限制。为了说明这个问题，我们假设一个系统，单个生产者每秒钟可将 50 个项目排入队列，而使用者每秒钟只能使用 10 个项目。首先，这样会打乱系统的平衡性，而且对于一对一的生产者 使用者配置无法进行很好的调整。只需一分钟，就会有 2,400 个项目堆积在缓冲区中。假设这些项目中每个项目使用 10KB 内存，那将意味着缓冲区本身就会占用 24MB 内
30、存。一小时后，这个数字将飙升至 1GB 以上。解决这一问题的一个方法是调整生产者线程与使用者线程的比例，在上述情况中，要将比例调整为一个生产者对应五个使用者。但是到达速度通常是不稳定的，这会导致系统周期性的不稳定，并带来一些明显的问题，简单的固定比例是无法解决这个问题的。服务器上的程序通常是长时间运行的，人们希望程序能够长期保持良好的运行状态，但如果内存使用不受限，就会造成不可避免的混乱，还可能导致必须定期回收服务器进程的情况。受限缓冲区允许您对生产者强制阻塞前缓冲区可能达到的大小进行限制。阻塞生产者可让使用者有机会“赶上”（通过允许其线程接收调度时间片），同时还能够解决内存使用量增加的问题。我们的方法还是仅封装 Queue，同时添加两个等待条件和两个事件通知条件：生产者在队列满时等待，直至队列变为非满，而使用者在队列空时等待，直至变为非空；生产者在生产出项目后通知等待的使用者，而使用者也会在取走项目后通知生产者，如图 6 中所示。Figure6 受限缓冲区 (Bounded Buffer) 复制代码 public class BoundedBuffer private Queue m_queue = new Queue(); private int m_consumer
