C++并发实战-第八章-总学习笔记第6
[TOC]
C++ 并发实战《C++ Concurrency in Action》的学习笔记 6, 记录第八章的部分 Designing concurrent code.
内容是:
- 线程间如何分数据. Techniques for dividing data between threads.
- 影响并发代码的因素. Factors that affect the performance of concurrent code.
- 影响因素如何影响数据结构的设计. How performance factors affect the design of data structures.
- 异常安全. Exception safety in multithreaded code.
- Scalability.
- Example implementations of several parallel algorithms.
Chapter 8 Designing concurrent code
8.1 Techniques for dividing work between threads
类比于建房子请工人 You need to decide whether to have “generalist” threads that do whatever work is necessary at any point in time or “specialist” threads that do one thing well, or some combination.
8.1.1 Dividing data between threads before processing begins
map-reduce 开局就均分法:
实际应用举例: Message Passing Interface (MPI, http://www.mpi-forum.org/) or OpenMP (http://www.openmp.org/) frameworks.
对最后的 reduction step 也可以考虑使用多线程: reduction operation: the worker threads could be made to perform some of the reduction steps as each one completes its task,
rather than spawning new threads each time.
8.1.2 Dividing data recursively
实际例子: Quicksort algorithm, These recursive calls are entirely independent.
决定线程数分配的策略:
std::async()
自动分配决定分配 thread 的策略.- use the
std::thread::hardware_concurrency()
function to choose the number of threads.
If a thread has nothing else to do, either because it has finished processing all its chunks or because it’s waiting for a chunk to be sorted, it can take a chunk from the stack and sort that.
一个非 production-ready code 的示例:
1 |
|
上面做法的问题:
- the management of these threads and the communication between them add quite a lot of complexity to the code.
- threads are processing separate data elements, they all access the stack to add new chunks and to remove chunks for processing. ==> reduce performance
这种做法其实是线程池: This approach is a specialized version of a thread pool.
到目前介绍的两种分数据都是建立在假设: the data itself is fixed beforehand.
对于 dynamically generated or is coming from external input 如何做? ==> it might make more sense to divide the work by task type rather than dividing based on the data.
8.1.3 Dividing work by task type
好处: Occasionally other threads may give it data or trigger events that it needs to handle, but in general each thread focuses on doing one thing well. In itself, this is basic good design; each piece of code should have a single responsibility.
DIVIDING WORK BY TASK TYPE TO SEPARATE CONCERNS
对于单线程下的多任务系统:
- In the single-threaded world you end up manually writing code that performs a bit of task A, a bit of task B, checks for key presses, checks for incoming network packets, and then loops back to perform another bit of task A.
- This means that the code for task A ends up being complicated by the need to save its state and return control to the main loop periodically.
- If you add too many tasks to the loop, things might slow down too much, and the user may find it takes too long to respond to the key press.
多线程下, 分清楚任务在不同的线程上可以提升效率:
- The operating system will automatically save the state and switch to task B or C when appropriate, and if the target system has multiple cores or processors, tasks A and B may be able to run concurrently.
- simpler code because each thread can focus on doing operations related directly to its responsibilities, rather than getting mixed up with control flow and user interaction.
虽然各线程无法实现完全的独立, 但通过线程间的通讯与同步可以解决这个问题.
通信的问题: too much communication between threads, 两种表现:
- there is a lot of data shared between the threads or
- the different threads end up waiting for each other.
解决思路:
- 创建新线程 If all the communication relates to the same issue, maybe that should be the key responsibility of a single thread and extracted from all the threads that refer to it.
- 合并线程 if two threads are communicating a lot with each other but much less with other threads, maybe they should be combined into a single thread.
DIVIDING A SEQUENCE OF TASKS BETWEEN THREADS
pipeline: data flows in at one end through a series of operations (pipes) and out at the other end.
通信的方式: the data element is put in a queue to be picked up by the next thread.
这种方式好处: 除了刚开动时比较慢, 实现的效果是 You now get 1 item processed every 3 seconds rather than having the items processed in batches of 4 every 12 seconds. 单个 item 处理的时间虽然不变但是可以一个个拿到而不是一个 batch 拿到.
But smoother, more regular processing can be beneficial in some circumstances. 例如看视频不是 100 帧加一个延迟, 而是 using a pipeline that outputs frames at a nice steady rate is probably preferable.
8.2 Factors affecting the performance of concurrent code
每个因素的细微的变化都有可能导致重大性能影响: even something as simple as changing which data elements are processed by each thread (while keeping everything else identical) can have a dramatic effect on performance.
8.2.1 How many processors?
To allow applications to scale the number of threads in line with the number of threads the hardware can run concurrently, the C++11 Standard Thread Library provides std::thread::hardware_concurrency()
.
consult your system documentation to see what options are available to you.
并行任务的复杂度与计算单元数目之间的关系: One additional twist to this situation is that the ideal algorithm for a problem can depend on the size of the problem compared to the number of processing units.
具体地, If you have a massively parallel system with many processing units, an algorithm that performs more operations overall may finish more quickly than one that performs fewer operations, because each processor performs only a few operations.
8.2.2 Data contention and cache ping-pong
cache 同步问题: the data will be copied into their respective caches, and both processors can proceed. But if one of the threads modifies the data, this change then has to propagate to the cache on the other core, which takes time.
如下例子:
1 | std::atomic<unsigned long> counter(0); |
high contention: If do_something()
is short enough, or if there are too many processors running this code(前提), the processors might find themselves waiting for each other(注意此处的主体对象是 processors 而不是线程); one processor is ready to update the value, but another processor is currently doing that, so it has to wait until the second processor has completed its update and the change has propagated.
In a loop like this one, the data
for counter will be passed back and forth between the caches many times. This is called cache ping-pong. 对性能的影响: If a processor stalls because it has to wait for a cache transfer, it can’t do any work in the meantime, even if there are other threads waiting that could do useful work. 整个 Processor 的停顿.
这种现象也会出现在 mutex 情境下: extra transfer time: In order to lock the mutex, another thread must transfer the data that makes up the mutex to its processor and modify it. ==> if the data and mutex are accessed by more than one thread, then as you add more cores and processors to the system, it becomes more likely that you will get high contention and one processor having to wait for another.
mutex 下比 atomic operation 情况要好一些的原因: The effects of contention with mutexes are usually different from the effects of contention with atomic operations for the simple reason that the use of a mutex naturally serializes threads at the operating system level rather than at the processor level. If you have enough threads ready to run, the operating system can schedule another thread to run while one thread is waiting for the mutex, whereas a processor stall prevents any threads from running on that processor. 最多阻碍线程跑在一个特定的 processor 上, 但是 OS 还是可以安排一些线程跑在其他 processor 上.
即便是读写锁 a single-writer, multiple-reader mutex 也会在 reading 过程中受到 cache ping-pong 现象的影响.
具体地: Even reader threads still have to modify the mutex itself. As the number of processors accessing the data goes up, the contention on the mutex itself increases, and the cache line holding the mutex must be transferred between cores, potentially increasing the time taken to acquire and release locks to undesirable levels.
解决 cache ping-pong 的原则: do what you can to reduce the potential for two threads competing for the same memory location.
但是也不是那么容易的, 即便 follow 这个原则, Even if a particular memory location is only ever accessed by one thread, you can still get cache ping-pong due to an effect known as false sharing.
8.2.3 False sharing
cache lines: unit by blocks of memory, typically 32 or 64 bytes in size.
Because the cache hardware only deals in cache-line-sized blocks of memory, small data items in adjacent memory locations will be in the same cache line. But if the data items in a cache line are unrelated and need to be accessed by different threads, this can be a major cause of performance problems.
an array of int
values: even though each thread only accesses its own array entry, the cache hardware still has to play cache ping-pong.
定义: The cache line is shared, even though none of the data is, hence the term false sharing.
解决思路: 只把同一个线程处理的数据放在相邻的位置上(cache line里). The solution here is to structure the data so that data items to be accessed by the same thread are close together in memory (and thus more likely to be in the same cache line).
C++ 17 提供了解决工具: The C++17 standard defines std::hardware_destructive_interference_size
in the header <new>
, which specifies the maximum number of consecutive bytes that may be subject to false sharing for the current compilation target. If you ensure that your data is at least this number of bytes apart, then there will be no false sharing.
8.2.4 How close is your data?
data proximity: if the data accessed by a single thread is spread out in memory, it’s likely that it lies on separate cache lines. On the flip side, if the data accessed by a single thread is close together in memory, it’s more likely to lie on the same cache line. 空间相邻性.
空间相邻性差的结果:
- increase memory access latency
- cache miss up
从单线程的此问题到多线程情景下: task switching: when the processor switches threads, it’s more likely to have to reload the cache lines if each thread uses data spread across multiple cache lines than if each thread’s data is close together in the same cache line.(前提 If there are more threads than cores in the system).
还有一种情况下, 不同 processors 之间 threads 切换导致的 cache line transfering: the operating system might also choose to schedule a thread on one core for one time slice and then on another core for the next time slice. This will therefore require transferring the cache lines for that thread’s data from the cache for the first core to the cache for the second; the more cache lines that need transferring, the more time-consuming this will be.
8.2.5 Oversubscription and excessive task switching
Oversubscription can arise when you have a task that repeatedly spawns new threads without limits.
由于此现象所以怎么分任务就显得很重要. choosing the appropriate division may require more knowledge of the target platform than you have available and is only worth doing if performance is unacceptable and it can be demonstrated that changing the division of work does improve performance. 有时候需要通过测试来最终确定.
8.3 Designing data structures for multithreaded performance
三个注意考虑因素: contention, false sharing, and data proximity.
8.3.1 Dividing array elements for complex operations
例子, 矩阵相乘运算如何分配: It may therefore be better to divide the result matrix into small, square or almost square blocks rather than have each thread compute the entirety of a small number of rows. You can adjust the size of each block at runtime, depending on the size of the matrices and the available number of processors.
原则除了细心检查别无他法: The same principles apply to any situation where you have large blocks of data to divide between threads; look at all the aspects of the data access patterns carefully, and identify the potential causes of performance hits.
8.3.2 Data access patterns in other data structures
原则:
- Try to adjust the data distribution between threads so that data that’s close together is worked on by the same thread.
- Try to minimize the data required by any given thread.
- Try to ensure that data accessed by separate threads is sufficiently far apart to avoid false sharing using
std::hardware_destructive_interference_size
as a guide.
一些思路:
- 用指针作为线程处理的第一层数据
if the tree nodes only contain pointers to the real data held at the node, then the processor only has to load the data from memory if it’s needed. - 把 mutex 与其处理的数据放在一起
If the mutex and the data items are close together in memory, this is ideal for a thread that acquires the mutex; the data it needs may already be in the processor cache, because it was loaded in order to modify the mutex.
But there’s also a downside: if other threads try to lock the mutex while it’s held by the first thread, they’ll need access to that memory. Mutex locks are typically implemented as a read-modify-write atomic operation on a memory location within the mutex to try to acquire the mutex, followed by a call to the operating system kernel if the mutex is already locked. This read-modify-write operation may cause the data held in the cache by the thread that owns the mutex to be invalidated. 例如 DCAS 会增加一些数据进去, 改变原先的数据.
检查方法: One way to test whether this kind of false sharing is a problem is to add huge blocks of padding between the data elements that can be concurrently accessed by different threads.
1 | struct protected_data |
If this improves the performance, you know that false sharing was a problem, and you can either leave the padding in or work to eliminate the false sharing in another way by rearranging the data accesses.
8.4 Additional considerations when designing for concurrency
除了上面的因素还有 2 个方面需要考虑:
scalability: Code is said to be scalable if the performance (whether in terms of reduced speed of execution or increased throughput) increases as more processing cores are added to the system.
exception safety is a matter of correctness.
8.4.1 Exception safety in parallel algorithms
与单线程不同, 多线程下的异常无法传递到 caller. By contrast, in a parallel algorithm many of the operations will be running on separate threads. In this case, the exception can’t be allowed to propagate because it’s on the wrong call stack. If a function spawned on a new thread exits with an exception, the application is terminated.
1 |
|
主要在 2 个地方不是异常安全的:
- thread 创建的过程中出现异常会直接
std::terminate
. - the calls to
accumulate_block
on the new threads might throw. 因为没有catch
语句, 异常不会传播.
ADDING EXCEPTION SAFETY
使用 std::packaged_task
and std::future
的组合:
Listing 8.3 A parallel version of std::accumulate
using std::packaged_task
1 |
|
结果: If more than one of the worker threads throws an exception, only one will be propagated. 或者 you can use something like std::nested_exception
to capture all the exceptions and throw that instead.
另一个问题: The remaining problem is the leaking threads if an exception is thrown between when you spawn the first thread and when you’ve joined with them all.
方法: The simplest solution is to catch any exceptions, join with the threads that are still joinable()
, and rethrow the exception.
1 | try |
这种做法有 duplicate code, 更好的方式 let’s extract this out into the destructor of an object.
1 | class join_threads |
修改后的代码, Listing 8.4 An exception-safe parallel version of std::accumulate
:
1 |
|
EXCEPTION SAFETY WITH STD::ASYNC()
The key thing to note for exception safety is that if you destroy the future without waiting for it, the destructor will wait for the thread to complete. This neatly avoids the problem of leaked threads that are still executing and holding references to the data.
Listing 8.5 An exception-safe parallel version of std::accumulate
using std::async
1 |
|
使用的是 a recursive division of the data.
两个地方可能抛出异常需要关注:
- If an exception is thrown by the recursive call, the future created from the call to
std::async
will be destroyed as the exception propagates. This will in turn wait for the asynchronous task to finish, avoiding a dangling thread. - On the other hand, if the asynchronous call throws, this is captured by the future, and the call to
get()
will rethrow the exception.
8.4.2 Scalability and Amdahl’s law
Scalability 底下产生的源泉: Every time one thread has to wait for something (whatever that something is), unless there’s another thread ready to take its place on the processor, you have a processor sitting idle that could be doing useful work.
对任务进行分类: A simplified way of looking at this is to divide the program into “serial” sections where only one thread is doing any useful work and “parallel” sections where all the available processors are doing useful work.
Amdahl’s law: “serial” sections constitute a fraction, $f_s$, of the program.
performance gain, $P$
using $N$ processors
$$
P=\frac{1}{f_s+\frac{1-f_s}{N}}
$$
得到定义: Scalability is about reducing the time it takes to perform an action or increasing the amount of data that can be processed in a given time as more processors are added.
2 种思路:
- reduce the size of the “serial” sections or reduce the potential for threads to wait,
- provide more data for the system to process, and thus keep the parallel sections primed with work.
8.4.3 Hiding latency with multiple threads
一些削减 wait 的思路:
if a thread is blocked because it’s waiting for an I/O operation to complete, it might make sense to use asynchronous I/O if that’s available, and then the thread can perform other useful work while the I/O is performed in the background.
In other cases, if a thread is waiting for another thread to perform an operation, then rather than blocking, the waiting thread might be able to perform that operation itself.
8.4.4 Improving responsiveness with concurrency
event-driven frameworks, 例如 modern graphical user interface.
1 | while(true) |
直观的想法:
- This means that either the task must periodically suspend itself and return control to the event loop,
- or the
get_event()
/process()
code must be called from within the code at convenient points.
把执行任务的耗时较长的部分单独分离为 thread, 通过 event/message 与 interface 的部分通信: you can put the lengthy task on a whole new thread and leave a dedicated GUI thread to process the events. The threads can then communicate through simple mechanisms rather than having to somehow mix the event-handling code in with the task code.
Listing 8.6 Separating GUI thread from task thread
1 |
|
8.5 Designing concurrent code in practice
8.5.1 A parallel implementation of std::for_each
Listing 8.7 A parallel version of std::for_each
1 |
|
Listing 8.8 A parallel version of std::for_each
using std::async
1 |
|
8.5.2 A parallel implementation of std::find
特点: can complete without every element having been processed. ===> to interrupt the other tasks in some way when the answer is known.
Other algorithms in this category include std::equal
and std::any_of
.
思路: atomic variable as a flag and checking the flag after processing every element. ===> downside to this is that atomic loads can be slow operations.
遇到异常时, 需不需要在其他线程上继续完成搜索?
- 终止全部搜索: use
std::promise
to set the final result directly from the worker threads. - 异常的线程终止, 其他线程继续: use
std::packaged_task
, store all the exceptions, and then rethrow one of them if a match isn’t found.
有一点需要注意的是 You need to wait for all the threads to finish before getting the result from the future. If you block on the future, you’ll be waiting forever if the value isn’t there.
Listing 8.9 An implementation of a parallel find algorithm
1 |
|
一个良性的 data race: If multiple threads find a match or throw at the same time, they’ll race to set the result in the promise. But this is a benign race condition; whichever succeeds is nominally “first” and therefore an acceptable result.
注意中间有一段 {}
的 block: you need to wait for all threads to finish before you check the result, because there might not be any matching elements. You do this by enclosing the thread launching-and-joining code in a block.
Listing 8.10 An implementation of a parallel find algorithm using std::async
1 |
|
8.5.3 A parallel implementation of std::partial_sum
特点: you can’t just divide the range into chunks and calculate each chunk independently.
书中展示了一个算法, 总共 $N$ 个元素, 共进行 $\log_N$ 轮, 对于第 $n$ 个元素, 第一轮进行计算 $v_n=v_n+v_{n-1}$, 第二轮进行计算 $v_n=v_n+v_{n-2}$, 以此类推第 $i$ 轮 计算 $v_n=v_n+v_{n-i}$ 直到 $v_n=v_n+v_{0}$.
算法复杂度: Overall, the second approach takes $/log_2N$ steps of approximately $N$ operations (one per processor), where $N$ is the number of elements in the list. $O(N \logN) $
第一种算法 $O(N)$, 分段分别计算并且把上一段的结果加到下一段上(因此存在嵌套的等待关系).
Listing 8.11 Calculating partial sums in parallel by dividing the problem
1 |
|
IMPLEMENTING THE INCREMENTAL PAIRWISE ALGORITHM FOR PARTIAL SUMS
第二种算法.
Listing 8.12 A simple barrier class
1 |
|
this implementation was simple:
- uses a spin wait.
- it doesn’t work if there’s more than
count
threads that can potentially callwait()
at any one time. - stuck to sequentially consistent operations.
- possible cache ping-pong.
Listing 8.13 A parallel implementation of partial_sum
by pairwise updates
1 |
|
Note that this solution isn’t exception-safe.
C++并发实战-第八章-总学习笔记第6
https://www.chuxin911.com/Notes_On_C++_Concurrency_in_Action_6_20221007/