C++并发编程实战:如何在线程间划分数据?

设想你被安排建造一所房子。为了完成这项工作,你需要挖地基、筑墙、布线等。理论上说,你可以在足够的训练下全部自己完成,但是将耗费很多时间,并且会一直切换任务。或者,你可以雇佣别人来帮助你。你只需要选择雇佣多少人并且决定他们需要哪些技能。例如,你可以雇佣一些具有一般技能的人,并且让每个人做所有事。你仍然需要切换任务,但是因为人数更多所以能够更快地完成。

或者,你可以雇佣一些专家。例如,砖匠,木匠,电工和管道工。这些专家只做他们专长的事情,因此如果没有管道工程的时候,管道工就不需要做任何事情。事情会比之前完成得更快,因为有更多的人,并且当电工给厨房布线的时候,管道工可以安装卫生间。但是当专家没有工作的时候就会处于等待状态。即使有空闲时间,你也会发现雇佣专家比雇佣一般技能的人工作进展得更快。专家不需要改变工具,并且他们完成任务比一般人要快。无论这种情况是否取决于特殊情况-你都需要尝试一下。

即使你雇佣专家,你仍然可以选择每种专家的数量。例如,雇佣比电工数量更多的砖匠就很合理。如果你要建造不止一座房子的话,你的队伍的构成以及总体效率都会发生改变。即使管道工在给定的房子上不会有太多的工作,你也可以一次建造很多房子,这样他就会始终有工作了。并且,如果当他们没有工作的时候不需要付钱的话,那么就可以负担更大的团队了,即使同一时间只有相同数量的人在工作。

那么线程需要做哪些呢?同样的问题也适用于线程。你需要决定使用多少线程以及它们需要完成什么任务。你需要决定是使用"通用型"线程来在需要的时候执行操作,还是使用“专家型"线程来做好一件事或一些合作的事。你需要决定无论是为了使用并发性而划分的原因,还是如何去做都会对代码的性能和清晰度产生很大影响。因此理解这些选择是很重要的,这样当设计你的应用中的数据结构的时候,就可以做出合适的决定。在这部分,我们将会看到一些划分任务的方法。在我们做别的工作前先来看看如何在线程间划分数据。

8.1.1处理开始前在线程间划分数据

最早并行化的算法是简单的算法,如std:for_each在数据集合中对每个元素进行操作。为了并行化这样的算法,就需要将每个元素划分到一个处理线程中。如何划分元素来得到最优性能在很大程度上决定于数据结构的细节,稍后我们分析性能问题的时候就可以看到了。

划分数据最简单的方法就是将第一个N元素分配给一个线程,将下一个N元素分配给另一个线程,以此类推,正E如图8.1所示,但是也可以使用别的模式。无论如何划分数据,每个线程只能处理分配它的元素,并且直到它完成任务的时候才能与别的线程通信。

这种结构与使用消息传递接口(Message Passing Interface, MPI )或者OpenMP框架编程的结构是类似的。一个任务被分成一个并行任务集,工作的线程独立运行这些任务,并且在最后的化简步骤中合并这些结果。这是2.4节中 accumulate例子使用的方法;在这种情况下,并行任务和最后的步骤都是累加。对一个简单的for_each 来说,最后的步骤是不需要的,因为不需要化简结果

确定将最后的步骤作为化简步骤是很重要的,清单2.8中的简单实现将执行此化简作为最后的线性步骤。尽管如此,这个步骤也可以并行化。累加过程本身就是化简操作,因此可以修改清单2.8,当线程的数量比线程处理的的最少数据的数量还要多,就可以递归地调用它本身。或者,当每个线程完成它的任务后,工作线程可以执行一些化简操作步骤，而不是每次产一些新线程。

尽管这种方法是很有效的,但是并不适用于所有情况。有时数据不能被事先划分,因为只有当处理数据的时候才知道如何划分。在化简算法例如快速排序中,这种情况更明显。因此需要一个不同的方法。

8.1.2递归地划分数据

快速排序算法有两个基本步骤,基于其中一个元素(关键值)将数据划分为两部分,一部分在关键值之前,一部分在关键值之后。然后递归地排序这两部分。你无法通过预先划分数据来实行并行,因为只有当处理元素的时候才知道他们属于哪一个“部分”。如果你打算并行这个算法,就需要把握递归的本质。每次递归的时候,会调用更多的quick_sort 函数来排序关键点之前和关键点之后的元素。这些递归调用是完全相互独立的,因为它们读取完全不同的元素集合。这种划分可以作为我们初步的候选方案。此递归划分如图8.2所示。

在第4章中,有这样一个实现。不仅只是为这两部分执行两个递归调用,你还在每一步都在前一部分使用std::async()来生成异步任务。通过使用sta::async() ,你让C++线程库来决定何时在一个新线程上运行这个任务,以及何时同步运行它们。

当你对很大规模的数据进行排序的时候这是很重要的。为每一个递归调用生成一个新线程会很快产生大量线程。当我们考虑性能的时候,你就会发现如果线程太多的时候,就可能降低了应用。如果数据集很大的时候也可能会用完所有的线程。像这样用递归来划分总体任务的方法是很好的,只需要严格控制线程数量就可以了。在简单情况下, std:async()就可以处理它,但是这并不是唯一选择

或者使用sta::thread: :hardware_concurrency() 函数来选择线程数量,正如清单2.8中 accumulate()的并行版本所做的一样。然后,你只是将此块存储到第6章和第7章描述线程安全栈中,而不是为递归调用创建一个新线程。如果线程不在工作,就说明它已经处理完所有块,或者等待存储在栈中的块。此时可以从栈中得到一个块并将它排序。

清单8.1列出了使用这种方法的简单实现,

这里, parallel_quick_sort 函数19代表了sorter类1的大部分功能,提供了一种简单的方法将未排序块2所在的栈进行分组,以及将线程集3分组。主要的工作是在do_sort成员函数9中完成的,它是用来完成通常的数据排序10。这次,它将块压入栈中11,而不是为一个块产生一个新的线程,并且当你仍然有处理器可以分配的时候就产生一个新进程12。因为后一部分可能是被另一个线程处理,你就必须等待它处理完成13。为了处理这种情况(即只有一个线程或者别的线程都在工作) ,你就试图在等待的时候,让这个线程处理栈中的块14。 try_sort_chunk 只是将一个块出栈7并且将它排序8,将结果存储在promise 中,此结果可以被将此块放入楼中的线程取得15。

当未设置 end_of_data标志的时候17 ,可以在循环中产生线程将栈中的块先出栈然后排序16。在检查的时候,它们让别的线程先对栈进行操作。这段代码依靠sorter类析构函数4来结束这些线程。当所有数据都被排序了,就会返回do_sort (即使线程仍然在运行) ,因此主线程将从 parallel_quici_sort 20中返回,并且销毁你的sorter 对象。这就设置了 end_of_data标志位5并且等待线程结束6。设置标志位结束了线程函数中的循环16。

使用这种方法就不会和使用spawn_task来产生一个新线程一样导致无穷多个线程这样的问题了,并且不再和std::async()一样依赖C++线程库来选择线程数量。现在我们将线程数量限制到std::thread::hardware_concurrency()来避免过多的任务切换。尽管如此,你有另一个潜在的问题,处理这些线程和线程间通信给代码增加了很多复杂性。同时,尽管这些线程在处理不相关的数据元素,它们都访问栈来移入和移出所操作的块。即使使用无锁(因此是无阻塞的)栈,这种竞争会降低性能。你稍后会看到原因。

这种方法是一种特殊版本的线程池--有一个线程集,每个线程处理等待列表中的工作,然后回到线程池。线程池存在的问题(包括列表上的竞争) ,第9章中有解决这些问题的方法。本章稍后将讨论如何将程序扩展到多处理器上执行(参见8.2.1节)。

在处理开始前划分数据和递归划分数据都是假设数据是固定不变的,然后你寻找划分它的方法,但是情况并不总是这样。如果数据是动态生成的或者是外部输入的,那么这种方法就不行了。在这种情况下,通过任务类型来划分工作比基于数据划分更合适。

8.1.3以任务类型划分工作

通过给每个线程分配不同数据块在线程间划分工作(无论是事先划分还是处理过程中递归划分)仍然是基于这样的假设,即线程将会基于每个数据块做同样的工作。划分工作的另一种方法是使得线程变得专业化,即每个线程执行不同的任务,就如同建造房子的时候管道工和电工执行不同的任务一样。线程可以基于也可以不基于同样的数据来工作,但是如果基于同样的数据,那也是有不同的目的。

这种划分工作的方式源自于将并发中的关注点分离。每个线程都有不同的任务,并且独立于别的线程来工作。偶尔别的线程可能给它数据或者有触发事件需要处理,但是通常每个线程只做一件事情。这本质上是一个好设计,每块任务都应该只负责一个单一的任务。

1,以任务类型划分工作来分离关注点

当在一段时间内需要持续运行多个任务的时候,或者需要此应用能够及时处理有输入的事件(例如用户键盘输入或者输入网络数据)而不影响别的线程继续执行的时候,单线程应用需要处理与单一任务原则间的矛盾。在单线程环境中,你手工写代码来执行任务A的一部分,执行任务B的一部分,检查键盘输入,检查输入的网络包,然后继续循环执行A的另一部分。这就意味着任务A代码的结束部分会变复杂,因为需要保留它的状态以及周期性地返回控制给主循环。如果你给循环增加了太多任务,运行就会变得很慢,并且使用者会发现键盘输入的响应时间太长。你肯定见过一些应用采取过极端的方式。你设置它处理一些任务,然后接口保持不变直到它完成任务。

这就是线程发生作用的地方。如果你在独立的线程里运行每个任务,操作系统就可以帮你处理这种问题。在任务A的代码中,你可以致力于执行任务,并且不需要担心保留状态以及返回到主循环或者在此之前你花费了多长时间。操作系统将自动地保留状态,然后在适当的时候切换到任务B或C,并且如果系统有多个核或者处理器,任务A和B就可以真正地并行运行。处理键盘输入或者网络包的代码将会运行得很及时,并且皆大欢喜。使用者获得及时响应,你作为开发者可以写更简单的代码,因为每个线程都致力于做与任务直接相关的操作,而不是与控制流和用户互动混合在一起。

看上去这是一个很好的版本。它真的如此吗?如同任何事情一样,它取决于细节。如果每件事情都是独立的,并且线程不需要与另一个线程通信,那么它就确实很简单了。可是,事实却并不是如此。这些后台任务经常做一些用户需要的事情,并且它们需要更新用户接口让用户知道是何时完成这些任务的。或者,用户可能要取消任务,这就会要求用户接口以某种方式给后台任务发送一个消息通知它停止此任务。这两种情况都需要仔细的思考,设计以及适当的同步。但是这些关注点都是分离的。用户接口线程仍然只是处理用户接口,但是当别的线程要求时,它需要更新它的接口。同样,运行后台任务的线程仍然致力于那个任务要求的操作,只有当它的操作是"允许另一个线程停止此任务"。在这两个例子中,线程都不关心该要求是从哪里产生的,只关心它是否是为它们准备的以及与它们的任务是否直接相关。

多个线程关键点分离有两个危害。首先就是你将分离错误的关键点。征兆就是线程间有很多共享数据,或者不同的线程都以等待彼此作为结束。这两种情况都可以归结为线程间有太多的通信。如果发生这种情况,就值得查找产生通信原因。如果所有的通信都与同一件事相关,那么可能那就是单线程的关键任务,并且从所有引用它的线程中获得。或者,如果两个线程彼此之间需要很多通信但是与别的线程通信很少,那么它们就应该联合为一个线程。

当根据任务类型在线程间划分工作的时候,你就不需要局限于完全独立的任务。如果多个数据集合需要应用同样的操作序列,那么就可以划分工作使每个线程执行整个序列中一个步骤。

2.划分线程间的任务序列

如果你的任务是由在很多独立数据项上运行同样的操作序列组成的话,就可以使用管道来开发系统可能的并发性。可以将它类比为管道,数据通过一系列操作(管子)从一端流入,并且从另一端流出。

为了用这种方式划分工作,你在管道的每一个步骤都创造一个独立的线程--序列中的每个操作都有一个线程。当操作完成时,数据元素被放入队列中供下一个线程获得。这就允许当管道中第二个线程在操作第一个元素的时候,第一个线程执行序列中的第一个操作来开始下一个数据元素。

这是仅仅在线程间划分数据的一种替代方法,正如8.1.1节中描述的一样,并且在操作开始时并不知道所有输入数据的情况下是适用的。例如,数据可能是通过网络输入的,或者序列的第一个操作就是扫描一个文件系统来识别要处理的文件。

当序列中的每个操作都消耗时间的时候,管道也可以很好地工作。通过在线程间划分任务而不是数据,你改变了性能概况。假设你要在四核上处理20个数据项,并且每个数据项需要四个步骤,每个步骤需要3秒。如果你在四个线程中划分数据,那么每个线程要处理5个数据项。假设没有别的影响时间的处理, 12秒后将处理完4个数据项, 24秒后将处理完8个数据项,以此类推。1分钟后将处理完所有的20个数据项。如果使用管道,事情就会不一样了。可以将四个步骤中的每个步骤分配到一个处理核上。现在每个核心都要处理第一个元素,因此需要12秒。实际上, 12秒后你只处理完一个数据项,这就没有比数据划分的方法好。但是,一旦管道被使用,处理事情就会变得不一样了。在第一个核心处理完的第一项以后,它接着处理第二项。因此一旦最后一个核处理完第一项,它就可以在第二项上执行它的步骤。现在每3秒都可以处理完一个数据项,而不是在每12秒能处理完一批四个数据项。

处理整个分批会花费更长的时间,因为在最后一个核开始处理第一个数据项之前你需要等待9秒。但是更平滑,更有规律的处理在某些环境下可能会很有效。例如,考虑用来收看高清数字电视的系统。为了使电视是可看的,你至少需要每秒25帧并且更多的帧得到更理想的效果。同样,观看者需要它们被均匀地分开来得到持续活动的印象;一个可以每秒解码100帧的应用是没用的,如果它暂停一秒,然后显示100帧,然后再停-秒,然后显示另一个100帧;另一方面,观看者可能很高兴接受当他们开始观看电视的时候有几秒的延迟。在这种情况下,使用管道以一个更好、更稳定的速度并行输出帧可能会更好。

已经看过在线程间划分工作的一些方法,我们来看看影响多线程系统性能的因素以及它是如何影响你选择的方法的。

本文节选自《C++并发编程实战》

本书介绍如何编写或者设计、调试、维护、研究多线程C++程序，并提供了技术模式和工具，可以用来分析并发编程以及降低封装并发交互的复杂性。书中还提供了大量的实例、练习、可重用的代码以及用于网络通信程序的简化库。