C++并发编程实战：如何为多线程性能设计数据结构？

在8.1节中我们看到了在线程间划分工作的一些方法,在8.2节中我们看到了影响代码性能的一些因素。当设计多线程性能的数据结构的时候如何使用这些信息呢?这是在第6章和第7章中处理的很困难的问题,是关于设计可以安全并行读取的数据结构。正如你在8.2节中看到的一样,即使没有别的线程共享此数据,单个线程使用的数据布局也会对它产生影响。

当为多线程性能设计你的数据结构时需要考虑的关键问题是竞争、假共享以及数据接近。这三个方面都会对性能产生很大影响,并且通常你可以通过改变数据布局或者改变分配给某线程的数据元素来提高性能。首先,我们来看一个简单的例子,在线程间划分数组元素。

8.3.1 为复杂操作划分数组元素

假设你正在做一些复杂的数学计算,你需要将两个大矩阵想乘。为了实现矩阵相乘,你将第一个矩阵的第一行每个元素与第二个矩阵的第一列相对应的每个元素相乘,并将结果相加得到结果矩阵左上角第一个元素。然后你继续将第二行与第一列相乘得到结果矩阵第一列的第二个元素,以此类推。正如图8.3所示,突出显示的部分表明了第一个矩阵的第二行与第二个矩阵的第三列配对,得到结果矩阵的第三列第二行的值。

图8.3 矩阵相乘

为了值得使用多线程来优化该乘法运算,现在我们假设这些都有几千行和几千列的大矩阵。通常,非稀疏矩阵在内存中是用一个大数组表示的,第一行的所有元素后面是第二行的所有元素,以此类推。为了实现矩阵相乘,现在就有三个大数组了。为了获得更优的性能,你就需要注意数据存取部分,特别是第三个数组。

有很多在线程间划分工作的方法。假设你有比处理器更多的行例,那么你就可以让每个线程计算结果矩阵中某些列的值,或者让每个线程计算结果矩阵中某些行的值,或者甚至让每个线程计算结果矩阵中规则矩形子集的值。

回顾8.2.3节和8.2.4节,你就会发现读取数组中的相邻元素比到处读取数组中的值要好,因为这样减少了缓存使用以及假共享。如果你使每个线程处理一些列,那么就需要读取第一个矩阵中的所有元素以及第二个矩阵中相对应的列中元素,但是你只会得到列元素的值。假设矩阵是用行顺序存储的,这就意味着你从第一行中读取N个元素,从第二行中读取N个元素,以此类推(N的值是你处理的列的数目)。别的线程会读取每一行中别的元素,这就很清楚你应该读取相邻的的列,因此每行的N个元素就是相邻的,并且最小化了假共享。当然,如果这N个元素使用的空间与缓存线的数量相等的话,就不会有假共享,因为每个线程都会工作在独立的缓存线上。

另一方面,如果每个线程处理一些行元素,那么就需要读取第二个矩阵中的所有元素,以及第一个矩阵中相关的行元素,但是它只会得到行元素。因为矩阵是用行顺序存储的,因此你现在读取从N行开始的所有元素。如果你选择相邻的行,那么就意味着此线程是现在唯一对这N行写入的线程;它拥有内存中连续的块并且不会被别的线程访问。这就比让每个线程处理一些列元素更好,因为唯一可能产生假共享的地方就是一块的最后一些元素与下一个块的开始一些元素。但是值得花时间确认目标结构。

第三种选择一划分为矩形块如何呢?这可以被看做是先划分为列,然后划分为行。它与根据列元素划分一样存在假共享问题。如果你可以选择块的列数目来避免这种问题,那么从读这方面来说,划分为矩形块有这样的优点:你不需要读取任何一个完整的源矩阵。你只需要读取相关的目标矩阵的行与列的值。从具体方面来看,考虑两个1000行和1000列的矩阵相乘。就有一百万个元素。如果你有100个处理器,那么每个线程可以处理10行元素。尽管如此,为了计算这10000个元素,需要读取第二个矩阵的所有元素(一百万个元素)加上第一个矩阵相关行的10000个元素,总计1010000个元素;另一方面,如果每个线程处理100行100列的矩阵块(总计10000个元素) ,那么它们需要读取第一个矩阵的100行元素( 100 x 1000=100000元素)和第二个矩阵的100列元素(另一个100000个元素)。这就只有200000元素,将读取的元素数量降低到五分之一。如果你读取更少的元素,那么发生缓存未命中和更好性能的潜力的机会就更少了。

因此将结果矩阵划分为小的方块或者类似方块的矩阵比每个线程完全处理好几行更好。当然,你可以调整运行时每个块的大小,取决于矩阵的大小以及处理器的数量。如果性能很重要,基于目标结构分析各种选择是很重要的。

你也有可能不进行矩阵乘法,那么它是否适用呢?当你在线程间划分大块数据的时候,同样的原则也适用于这种情况。仔细观察数据读取方式,并且识别影响性能的潜在原因。在你遇到的问题也可能有相似的环境,就是只要改变工作划分方式可以提高性能而不需要改变基本算法。

好了,我们已经看到数组读取方式是如何影响性能的。其他数据结构类型呢?

8.3.2其他数据结构中的数据访问方式

从根本上说,当试图优化别的数据结构的数据访问模式时也是适用的。

1、在线程间改变数据分配,使得相邻的数据被同一个线程适用。

2、最小化任何给定线程需要的数据。

3、确保独立的线程访问的数据相隔足够远来避免假共享。

当然,运用到别的数据结构上是不容易的。例如,二叉树本来就很难用任何方式来再分,有用还是没用,取决于树是如何平衡的以及你需要将它划分为多少个部分。同样,树的本质意味着结点是动态分配的,并且最后在堆上不同地方。

现在,使数据最后在堆上不同地方本身不是一个特别的问题,但是这意味着处理器需要在缓存中保持更多东西。实际上这可以很有利。如果多个线程需要遍历树,那么它们都需要读取树的结点,但是如果树的结点至包含指向该结点持有数据的指针,那么当需要的时候,处理器就必须从内存中载入数据。如果线程正在修改需要的数据,这就可以避免结点数据与提供树结构的数据间的假共享带来的性能损失。

使用互斥元保护数据的时候也有同样的问题。假设你有一个简单的类,它包含一些数据项和一个互斥元来保护多线程读取。如果互斥元和数据项在内存中离得很近,对于使用此互斥元的线程来说就很好;它需要的数据已经在处理器缓存中了,因为为了修改互斥元已经将它载入了。但是它也有一个缺点:当第一个线程持有豆斥元的时候,如果别的线程试图锁住互斥元,它们就需要读取内存。互斥元的锁通常作为一个在互斥元内的存储单元上试图获取互斥元的读一修改一写原子操作来实现的,如果互斥元已经被锁的话,就接着调用操作系统内核。这个读一修改一写操作可能导致拥有互斥元的线程持有的缓存中的数据变得无效。只要使用互斥元,这就不是问题。尽管如此,如果互斥元和线程使用的数据共享同一个缓冲线,那么拥有此互斥元的线程的性能就会因为另一个线程试图锁住该互斥元而受到影响。

测试这种假共享是否是一个问题的方法就是在数据元素间增加可以被不同的线程并发读取的大块填充数据例如,你可以使用:

来测试互斥元竞争问题或者使用：

来测试数组数据是否假共享。如果这样做提高了性能,就可以得知假共享确实是一个问题,并且你可以保留填充数据或者通过重新安排数据读取的方式来消除假共享。

当然,当设计并发性的时候,不仅需要考虑数据读取模式,因此让我们来看看别的需要考虑的方面。

8.4 为并发设计时的额外考虑

本章我们看了一些在线程间划分工作的方法,影响性能的因素,以及这些因素是如何影响你选择哪种数据读取模式和数据结构的。但是,设计并发代码需要考虑更多。你需要考虑的事情例如异常安全以及可扩展性。如果当系统中处理核心增加时性能(无论是从减少执行时间还是从增加吞吐量方面来说)也增加的话,那么代码就是可扩展的。从理论上说,性能增加是线性的。因此一个有100个处理器的系统的性能比只有一个处理器的系统好100倍。

即使代码不是可扩展的,它也可以工作。例如,单线程应用不是可扩展的,异常安全是与正确性有关的。如果你的代码不是异常安全的,就可能会以破碎的不变量或者竞争条件结束,或者你的应用可能因为一个操作抛出异常而突然终止。考虑到这些,我们将首先考虑异常安全。

8.4.1 并行算法中的异常安全

异常安全是好的C++代码的一个基本方面,使用并发性的代码也不例外。实际上,并行算法通常比普通线性算法需要你考虑更多关于异常方面的问题。如果线性算法中的操作抛出异常,该算法只要考虑确保它能够处理好以避免资源泄漏和破碎的不变量。它可以允许扩大异常给调用者来处理。相比之下,在并行算法中,很多操作在不同的线程上运行。在这种情况下,就不允许扩大异常了,因为它在错误的调用栈中。如果一个函数大量产生以异常结束的新线程,那么该应用就会被终止。

作为一个具体的例子,我们来回顾清单2.8中的 parallel_accumulate函数,清单8.2中会做一些修改

清单8.2 sta::accumulate的并行版本(来自清单2.8)

现在我们检查并且确定抛出异常的位置:总的说来,任何调用函数的地方或者在用户定义的类型上执行操作的地方都可能抛出异常。

首先,你调用distance 2,它在用户定义的迭代器类型上执行操作。因为你还没有进行任何工作,并且这是在调用线程上,所以这是没问题的。下一步,你分配了results选代器3和threads迭代器4。同样,这是在调用线程上,并且你没有做任何工作或者生产任何线程,因此这是没问题的。当然,如果threads构造函数抛出异常,那么就必须清楚为results分配的内存,析构函数将为你完成它。

跳过block_start 的初始化5因为这是安全的,就到了产生线程的循环中的操作6、7、8。一旦在7中创造了第一个线程,如果抛出异常的话就会很麻烦,你的新sta::thread 对象的析构函数会调用

std::terminate 然后中程序,

调用accumulate_block 9也可能会抛出异常,你的线程对象将被销毁并且调用std:terminate ;另一方面,最后调用std::accumulate 10的时候也可能抛出异常并且不导致任何困难,因为所有线程将在此处汇合。

这不是对于主线程来说的,但是也可能抛出异常,在新线程上调用 accumulate_block 可能抛出异常1。这里没有任何catch块,因此该异常将被稍后处理并且导致库调用sta::terminate()来中止程序。

即使不是显而易见的,这段代码也不是异常安全的。

1·增加异常安全性

好了,我们识别出了所有可能抛出异常的地方以及异常所造成的不好影响。那么如何处理它呢?我们先来解决在新线程上抛出异常的问题。

在第4章中介绍了完成此工作的工具。如果你仔细考虑在新线程中想获得什么,那么很明显当允许代码抛出异常的时候,你试图计算结果来返回。std: :packaged_task 和std:future 的组合设计是恰好的。如果你重新设计代码来使用 std::packaged_task ,就以清单8.3中的代码结束。

清单8.3使用std::packaged_task的std::accumulate的并行版本

第一个改变就是,函数调用accumulate_block操作直接返回结果,而不是返回存储地址的引用1。你使用std::packaged_task 和std::future来保证异常安全,因此你也可以使用它来转移结果。这就需要你调用std::accumulate 2明确使用默认构造函数T而不是重新使用提供的result值,不过这只是一个小小的改变。

下一个改变就是你用futures 向量3,而不是用结果为每个生成的线程存储一个 std:future 。在生成戈程的循环中,你首先为 accumulate_block 创造一个任务4。std:packaged_task

既然你已经使用了future ,就不再有结果数组了,因此必须将最后一块的结果存储在一个变量中7而不是存储在数组的一个位置中。同样,因为你将从future中得到值,使用基本的for 循环比使用std:accumulate要简单,以提供的初始值开始8,并且将每个future的结果累加起来9。如果相应的任务抛出异常,就会在future中捕捉到并且调用get() 时会再次抛出异常。最后,在返回总的结果给调用者之前要加上最后一个块的结果10。

因此,这就去除了一个可能的问题,工作线程中抛出的异常会在主线程中再次被抛出。如果多于一个工作线程抛出异常,只有一个异常会被传播,但是这也不是一个大问题。如果确实有关的话,可以使用类似

std::nested_exception来捕捉所有的异常然后抛出它。

如果在你产生第一个线程和你加入它们之间抛出异常的话,那么剩下的问题就是线程泄漏。最简单的方法就是捕获所有异常,并且将它们融合到调用joinable()的线程中,然后再次抛出异常。

现在它起作用了。所有线程将被联合起来,无论代码是如何离开块的。尽管如此, try-catch块是令人讨厌的,并且你有复制代码。你将加入正常的控制流以及捕获块的线程中。复制代码不是一个好事情,因为这意味着需要改变更多的地方。我们在一个对象的析构函数中检查它,毕竟,这是C++中惯用的清除资源的方法。下面是你的类。

这与清单2.3中的thread_guard类是相似的，除了它扩展为适合所有线程。你可以如清单8.4所示简化代码。

清单8.4 std：：accumulate的异常安全并行版本

一旦你创建了你的线程容器,也就创建了一个新类的实例1来加入所有在退出的线程。你可以去除你的联合循环,只要你知道无论函数是否退出,这些线程都将被联合起来。注意调用 futures[i].get() 2将被阻塞直到结果出来,因此在这一点并不需要明确地与线程融合起来。这与清单8.2中的原型不一样,在清单8.2中你必须与线程联合起来确保正确复制了结果向量。你不仅得到了异常安全代码,而且你的函数也更短了,因为将联合代码提取到你的新(可再用的)类中了。

2. STD::ASYNC()的异常安全

你已经知道了当处理线程时需要什么来实现异常安全,我们来看看使用std::async() 时需要做的同样的事情。你已经看到了,在这种情况下库为你处理这些线程,并且当future是就绪的时候,产生的任何线程都完成了。需要注意到关键事情就是异常安全,如果销毁future的时候没有等待它,析构函数将等待线程完成。这就避免了仍然在执行以及持有数据引用的泄漏线程的问题。清单8.5所示就是使用std::async ()的异常安全实现。

清单8.5 使用std::async的std::accumulate的异常安全并行版本

这个版本使用递归将数据划分为块而不是重新计算将数据划分为块,但是它比之前的版本要简单一些,并且是异常安全的。如以前一样,你以计算序列长度开始1,如果它比最大的块尺寸小的话,就直接调用

std::accumuate 2。如果它的元素比块尺寸大的话,就找到中点3然后产生一个异步任务来处理前半部分![序号4。范围内的第二部分就用一个直接的递归调用来处理5。,然后将这两个块的结果累加6。库确保了std::async调用使用了可获得的硬件线程,并且没有创造很多线程。一些"异步”调用将在调用get()的时候被异步执行6。

这种做法的好处在于它不仅可以利用硬件并发,而且它也是异常安全的。如果递归调用抛出异常5,当异常传播时,调用std::async 4创造的future就会被销毁。它会轮流等待异步线程结束,因此避免了悬挂线程。另一方面,如果异步调用抛出异常,就会被future捕捉,并且调用get() 6将再次抛出异常。

当设计并发代码的时候还需要考虑哪些方面呢?让我们来看看可扩展性。如果将你的代码迁移到更多处理器系统中会提高多少性能呢?

8.4.2可扩展性和阿姆达尔定律

可扩展性是关于确保你的应用可以利用系统中增加的处理器。一种极端情况就是你有一个完全不能扩展的单线程应用,即使你在系统中增加100个处理器也不会改变性能。另一种极端情况是你有类似SETI@Home[3]的项目,被设计用来利用成千上万的附加的处理器(以用户将个人计算机增加到网络中的形式)。

对于任何给定的多线程程序,当程序运行时,执行有用工作的线程的数量会发生变化。即使每个线程都在做有用的操作,初始化应用的时候可能只有一个线程,然后就有一个任务生成其他的线程。但是即使那样也是一个不太可能发生的方案。线程经常花时间等待彼此或者等待I/O操作完成

除非每次线程等待事情(无论是什么事情)的时候都有另一个线程在处理器上代替它,否则就有一个可以进行有用工作的处理器处于闲置状态

一种简单的方法就是将程序划分为只有一个线程在做有用的工作"串行的"部分和所有可以获得的处理器都在做有用工作的"并行的"部分。如果你在有更多处理器的系统上运行你的应用,理论上就可以更快地完成"并行"部分,因为可以在更多的处理器间划分工作,但是"串行的"部分仍然是串行的。在这样一种简单假设下,你可以通过增加处理器数量来估计可以获得的性能。如果“连续的"部分组成程序的一个部分fs,那么使用N个处理器获得的性能P就可以估计为

这就是阿姆达尔定律(Amdahl'slaw ) ,当谈论并发代码性能的时候经常被引用。如果所有事情都能被并行,那么串行部分就为0,加速就是N,或者,如果串行部分是三分之一,即使有无限多的处理器,你也不会得到超过3的加速

尽管如此,这是一种很理想的情况。因为任务很少可以像方程式所需要的那样被无穷划分,并且所有事情都达到它所假设的CPU界限是很少出现的。正如你看到的,线程执行的时候会等待很多事情。

阿姆达尔定律中有一点是明确的,那就是当你为性能使用并发的时候,值得考虑总体应用的设计来最大化并发的可能性,并且确保处理器始终有有用的工作来完成。如果你可以减少“串行”部分或者减少线程等待的可能性,你就可以提高在有更多处理器的系统上的性能。或者,如果你可以为系统提供更多的数据,并且保持并行部分准备工作,就可以减少串行部分,增加性能P的值。

从根本上说,可扩展性就是当增加更多的处理器的时候,可以减少它执行操作的时间或者增加在一段时间内处理的数据数量。有时这两点是相同的(如果每个元素可以处理得更快,那么你就可以处理更多数据) ,但是并不总是一样的。在选择在线程间划分工作的方法之前,识别出可扩展性的哪些方面对你很重要是很必要的。

在这部分的开始我就提到过线程并不是总有有用的工作来做。有时它们必须等待别的线程,或者等待I/O操作完成,或者别的事情。如果在等待中你给系统一些有用的事情,你就可以有效的"隐藏等待。

8.4.3用多线程隐藏迟

在很多关于多线程代码性能的讨论中,我们都假设当它们真正在处理器上运行时,线程在"全力以赴的运行并且总是有有用的工作来做。这当然不是正确的,在应用代码中,线程在等待的时候总是频繁地被阻塞。例如,它们可能在等待一些I/O操作的完成,等待获得互斥元,等待另一个线程完成一些操作并且通知一个条件变量,或者只是休眠一段时间。

无论等待的原因是什么,如果你只有和系统中物理处理单元一样多的线程,那么有阻塞的线程就意味着你在浪费CPU时间。运行一个被阻塞的线程的处理器不做任何事情。因此,如果你知道一个线程将会有相当一部分时间在等待,那么你就可以通过运行一个或多个附加线程来使用那个空闲的CPU时间。

考虑一个病毒扫描应用,它使用管道在线程间划分工作。第一个线程搜索文件系统来检查文件并且将它们放到队列中。同时,另一个线程获得队列中的文件名,载入文件,并且扫描它们的病毒。你知道搜索文件系统的文件来扫描的线程肯定会达到I/O界限,因此你就可以通过运行一个附加的扫描线程来使用“空闲的"CPU时间。那么你就有一个搜索文件线程,以及与系统中的物理核或者处理器相同数量的扫描线程。因为扫描线程可能也不得不从磁盘读取文件的重要部分来扫描它们,拥有更多扫描线程也是很有意义的。但是在某个时刻会有太多线程,系统会再次慢下来因为它花了更多时间切换程序,正如8.2.5节所描述的。

仍然,这是一个最优化问题,因此测量线程数量改变前后的性能时很重要的;最有的线程数量将很大程度上取决于工作的性质和线程等待的时间所占的比例。

取决于应用,它也可能用完空闲的CPU时间而没有运行附加的线程。例如,如果一个线程因为等待I/O操作的完成而被阻塞,那么使用可获得的异步I/O操作就很有意义了。那么当在背后执行I/O操作的时候,线程就可以执行别的有用的工作了。在别的情况下,如果一个线程在等待另一个线程执行一个操作,那么等待的线程就可以自己执行那个操作而不是被阻塞,正如第7章中的无锁队列。在一个极端的情况下,如果线程等待完成一个任务并且该任务没有被其他线程执行,等待的线程可以在它内部或者另一个未完成的任务中执行这个任务。清单8.1中你看到了这个例子,在排序程序中只要它需要的块没有排好序就不停地排序它。

有时它增加线程来确保外部事件及时被处理来增加系统响应性,而不是增加线程来确保所有可得到的处理器都被应用了。

8.4.4 用并发提高响应性

很多现代图形用户接口框架是事件驱动的,使用者通过键盘输入或者移动鼠标在用户接口上执行操作,这会产生一系列的事件或者消息,稍后应用就会处理它。系统自己也会产生消息或者事件。为了确保所有事件和消息都能被正确处理,通常应用都有下面所示的一个事件循环。

显然, API的细节是不同的,但是结构通常是一样的,等待一个事件,做需要的操作来处理它,然后等待下一个事件。如果你有单线程应用,就会导致长时间运行的任务很难被写入,如8.1.3节描述的。为了确保用户输入能被及时处理, get_event()和process()必须以合理的频率被调用,无论应用在做什么操作。这就意味着要么任务必须定期暂停并且将控制交给事件循环,要么方便的时候在代码中调用get_event()/

process()代码。每一种选择都将任务的实现变得复杂了。

通过用并发分离关注点,你就可以将长任务在一个新线程上执行,并且用一个专用的GUI线程来处理事件。线程可以通过简单的方法互相访问,而不是必须以某种方式将事件处理代码放到任务代码中。清单8.6列出了这种分离的简单概括。

清单8.6从任务线程中分离GUI线程