Shared address space 与 message passing
这次课来讨论通信
直到目前, 我们总是假设所有处理器连接到同一个内存系统, 所以它们可以有一个共享内存地址空间的抽象 但这种抽象的 implementation 是复杂的, 不同处理器通常有不同的架构
通信抽象:
- 线程在共享内存空间读/写
- 线程操控同步语句, 如 lock, atomic ops 等
- 是单线程编程的自然拓展
需要硬件支持:
- 任何处理器可以在任何地址在读/写
- 拓展到大规模的处理器集群是 costly 的
共享内存地址空间是一种抽象模型, 当然我们也可以有不同的抽象: 消息传递 Message passing
消息传递, 总而言之, 包含以下抽象
- 线程在各自私有的内存地址上操作
- 线程之间通过发送/接收消息通信(这是它们唯一通信的方式)
硬件上的实现也更为简单: 不需要所有处理器共享同一块内存, 只需要处理器之间有通信的方式, 比如网络通信等 这就是 MPI 的设计思路, 现代大型计算集群通常采用这样的思路
我们同样回到前面的例子: 用消息传递的思路, 如何编程一个 PDE 求解器 消息传递的架构如下
Computer 1 Computer 2
Processor Processor
| |
Local Cache Local Cache
| |
Memory Memory
| |
+----------- Network ---------+所以并行化求解格点的方式就是把格点划分为每块处理器私有的地址, 即处理器之间是看不到其他人处理的格点的
但是下一个问题是, 在计算某个区域时, 我们需要区域外边界的格点值才能更新内部的格点, 这也就是为什么需要线程之间的消息传递 在这个问题中, 我们称这些处理格点外的附带区域为 ghost cell, 它们通过消息传递更新
Thread 1: ███████████
↓ send
Thread 2: ○○○○○○○○○○ <- ghost row，实际属于 Thread 1
███████████
███████████
███████████
○○○○○○○○○○ <- ghost row，实际属于 Thread 3
↑ send
Thread 3: ███████████
███████████
Thread 4: ███████████
...线程 2 的逻辑即为如下
// Thread 2 logic:
float* local_data = allocate(N+2, rows_per_thread+2);
int tid = get_thread_id();
int bytes = sizeof(float) * (N+2);
// receive ghost row cells (white dots)
recv(&local_data[0], bytes, tid-1);
recv(&local_data[rows_per_thread+1], bytes, tid+1);
// Thread 2 now has data necessary to perform
// its future computation当接受到邻居线程传递的 ghost cell 的消息后才会开始更新自己的格点 同样, 每个线程更新完格点后也应该把消息发送给邻居线程
每轮 iteration
┌────────────────────────┐
│ 1. exchange ghost rows │
└───────────┬────────────┘
↓
┌────────────────────────┐
│ 2. local computation │
│ compute my_diff │
└───────────┬────────────┘
↓
Thread 1 ─┐
Thread 2 ─┼─ my_diff ─→ Thread 0
Thread 3 ─┘
↓
┌────────────────────────┐
│ 3. compute global diff │
│ decide done │
└───────────┬────────────┘
↓
Thread 0 broadcasts done
↓
next iteration具体的代码如下
int N;
int tid = get_thread_id();
int rows_per_thread = N / get_num_threads();
float* localA = allocate(rows_per_thread+2, N+2);
// assume localA is initialized with starting values
// assume MSG_ID_ROW, MSG_ID_DONE, MSG_ID_DIFF are constants used as msg ids
////////////////////////////////
void solve() {
bool done = false;
while (!done) {
float my_diff = 0.0f;
if (tid != 0)
send(&localA[1,0], sizeof(float)*(N+2), tid-1, MSG_ID_ROW);
if (tid != get_num_threads()-1)
send(&localA[rows_per_thread,0], sizeof(float)*(N+2),
tid+1, MSG_ID_ROW);
if (tid != 0)
recv(&localA[0,0], sizeof(float)*(N+2), tid-1, MSG_ID_ROW);
if (tid != get_num_threads()-1)
recv(&localA[rows_per_thread+1,0], sizeof(float)*(N+2),
tid+1, MSG_ID_ROW);
for (int i=1; i<rows_per_thread+1; i++) {
for (int j=1; j<N+1; j++) {
float prev = localA[i,j];
localA[i,j] = 0.2 * (localA[i-1,j] + localA[i,j] +
localA[i+1,j] + localA[i,j-1] +
localA[i,j+1]);
my_diff += fabs(localA[i,j] - prev);
}
}
if (tid != 0) {
send(&my_diff, sizeof(float), 0, MSG_ID_DIFF);
recv(&done, sizeof(bool), 0, MSG_ID_DONE);
} else {
float remote_diff;
for (int i=1; i<get_num_threads(); i++) {
recv(&remote_diff, sizeof(float), i, MSG_ID_DIFF);
my_diff += remote_diff;
}
if (my_diff/(N*N) < TOLERANCE)
done = true;
for (int i=1; i<get_num_threads(); i++)
send(&done, sizeof(bool), i, MSG_ID_DONE);
}
}
}这个消息传递的例子告诉我们:
计算:
- 数组索引是属于局域的线程地址的
通信:
- 由发送和接收消息实现
- Bulk transfer: 一次传递一整行的消息
同步:
- 由发送和接收消息实现
- 考虑如何用消息实现互斥, barriers 和 flags
实际上, 同步发送和接收会带来发送者与接收者之间的 blocking synchronization:
send()当发送者收到消息已经被放在接收者内存中的 acknowledgement 时才会返回recv()当消息数据被复制到接收者的内存中, 并把 acknowledgement 送回发送者之后才会返回
但这种设计实际上会带来 deadlock, 如果所有线程都先 send() , 没有线程执行 recv() , 那么所有线程全都卡死了!
一种处理方式是, 相邻的线程应该发送和接收交替执行, 防止没有人接收, 比如分奇偶
if (tid % 2 == 0) {
sendDown(); recvDown();
sendUp(); recvUp();
} else {
recvUp(); sendUp();
recvDown(); sendDown();
}当然另一种处理就是使用 non-blocking asynchronous send/recv: 调用发起发送和接收后直接返回, 再用 checksend() / checkrecv() 确认是否已经发送/接收完成
Sender Receiver
| |
| send(foo) | recv(bar)
| -> return h1 | -> return h2
| |
| [communication happens asynchronously]
| |
| ----------- message ---------->|
| | copy data -> bar
| |
| checksend(h1) | checkrecv(h2)
| |
| foo can now be modified | bar can now be used
v v当然在发送/接收的过程中时, 对应值是不能被修改的
实际上通信并不限于机器之间的通信, 线程之间, 核心之间, 核心与缓存之间, 核心与内存之间都是通信
一个处理器系统的层级是这样的
Proc
│
Reg
│
Local L1
│
Local L2
│
L2 from another core
│
L3 cache
│
Local memory
│
Remote memory (1 hop)
│
Remote memory (N hops)
↑ Lower latency, higher bandwidth,
│ smaller capacity
│
│
↓ Higher latency, lower bandwidth,
larger capacity每次之间的数据流都叫通信! 所以在每个层级利用 locality 来降低通信成本是重要的
甚至我们之前也看到, 带宽是限制许多并行计算性能的主要因素 对于前面讨论的 bandwidth-limited 计算任务, 在有足够多并发内存请求隐藏 latency 时, 计算处理器大部分时间都在等待数据, 但内存系统一直在工作; 稳态吞吐量主要受内存带宽限制, 而不是单次数据传输的 latency
Arithmetic intensity
这给出我们对于 Arithmetic Intensity 的定义
意义即为搬运的单位数据量能做的计算量, 一般写作
当然, 它的倒数的意思就是单位计算量所需要的搬运数据量
前面我们讲到并行加速的限制是带宽, 即数据吞吐量, 越大, 那么单位计算量需要的搬运数据量越少, 所以计算核心工作的时间越长
所以 可以是评判一个并行计算的指标, 它越大越好
inherent vs. artifactual communication
在前面用消息传递并行化 PDE 求解的时候, 我们面临的就是 inherent communication, 这是算法本身依赖的通信成本
一些好的分配方式可以降低, 但没办法消除, 因为必须有这些通信
例如, 如果加密分配块, 则处理器之间需要的通信次数就越多, 导致算术强度降低 当然在这个例子里面: 通信相当于各个块之间的边界接触, 也可以不用行来分配线程, 也可以用块分配线程, 如
┌──────┬──────┬──────┐
│ P1 │ P2 │ P3 │
│ │ │ │
├──────┼──────┼──────┤
│ P4 │ P5 │ P6 │
│ │ │ │
├──────┼──────┼──────┤
│ P7 │ P8 │ P9 │
│ │ │ │
└──────┴──────┴──────┘给出算术强度
其他的通信实际上可以归于 Artifactual communication, 就是实际上的算法不需要的部分
一个例子就是 cache line , 通信方式强制我们用一个 cache line 传输数据, 而不是只传一个数据
Techniques for reducing communication
利用 locality 可以降低通信成本, 即利用 cache line 造成的额外通信机制: 让多次计算到的点尽可能处于一个 cache line, 尽可能利用一个 cache line 传输的开销
比如格点上的计算可以走 zigzag 形状
另外一种就是算子的 fusing, 对于复合算子, 我们可以把它们写在一起, 复合成一个算术操作
一个例子是 E = D + ((A + B) * C) 这个操作
做 fusing 前, 我们的操作可能是
Before fusion:
A ──┐
├─ (+) ──> tmp1 ──┐
B ──┘ ├─ (*) ──> tmp2 ──┐
C ─────────────────────┘ ├─ (+) ──> E
D ────────────────────────────────────────┘
Memory traffic:
A,B → compute → tmp1 → memory
↓
load tmp1
↓
C ───────────→ compute → tmp2 → memory
↓
load tmp2
↓
D ─────────────────→ compute → E中间的每次调用都需要通信, 但是通过复合
A ──┐
├─ (+) ──┐
B ──┘ │
├─ (*) ──┐
C ───────────┘ │
├─ (+) ──> E
D ────────────────────┘直接完成计算, 没有中间结果导致的通信
还有一种方式是同样利用 locality, 如果多个计算要使用同一份数据, 就把这些任务放在同一个处理器上进行, 这可以降低 inherent communication
Contention
一个资源单位时间只能处理有限数量的请求；如果很多线程/处理器在很短时间内同时访问它，请求就会排队 这种现象称为 contention
一种方式是把访问的对象分散, 但是这会带来通信成本, 这又是一个 trade off
Flat: Tree:
○ ○ ○ ○ ○ ○ ○ ○ ○
\|/|/ \ / \ /
○ ○ ○
\ /
○一个例子是我们上次提到的, 各个处理器有各自的工作队列, 防止所有处理器挤在一个队列上抢任务, 这就是一种降低 contention 的方式
降低通信成本
- 降低发送/接收者的多余开销
- 降低通信延迟
- 降低 contention
- 提高通信/计算的交叠
Performance analysis
Always, always, always try the simplest parallel solution first, then measure performance to see where you stand.
一个表现良好的分析策略是
- 先确定当前性能是被什么限制: 计算, 带宽还是同步?
- 确定 high watermarks, 明确现实中实现的最好性能, 再决定是否要优化
Roofline model
通常来说, 我们有如下的经验图
Performance
^
|
| ─────────────────── Peak compute
| __/
| __/
| __/
| __/
| __/
| __/
| __/
| __/
+-------------------------------------------------> Arithmetic Intensity
memory bandwidth-bound compute-bound即 Performance = min(BW × AI, P_peak)
明确我们处在哪个位置, 才能知道我们是否还有优化空间
建立 high watermarks
- 增加计算 (保持内存访问), 确认程序是否对计算量敏感
- 清除所有计算, 加载同样的数据, 检查瓶颈是否是内存访问
- 将所有数组改为
A[0], 即制造完美的 locality, 检查 locality 最理想时的速度 - 去除所有的 atomic 和 lock, 检查是否是同步/contention 的问题
真正的程序当然不是简单的这些因素, 可能是综合的, 这只是一个用来估计的方法
从硬件层面, 我们可以看 monitor tools 这些工作检查硬件实际的工作 不要只看硬件的占用率等, 它们不能真实反映硬件实际的工作, 现代处理器通常有更底层的性能计数器, 能更具体看到硬件在做什么
例如 Intel 处理器提供了 Cpp 用的 api 来获取 registers 的数据
PCM *m = PCM::getInstance();
SystemCounterState begin = getSystemCounterState();
// code to analyze goes here
SystemCounterState end = getSystemCounterState();
printf("Instructions per clock: %f\n", getIPC(begin, end));
printf("L3 cache hit ratio: %f\n", getL3CacheHitRatio(begin, end));
printf("Bytes read: %ld\n", getBytesReadFromMC(begin, end));Bonus
Understanding problem size issues can be very helpful when assessing program performance
problem size
↓
┌───────────┴───────────┐
↓ ↓
work per processor working-set size
↓ ↓
communication/computation cache/memory fit
↓ ↓
scaling locality真正的计算问题不仅与问题的大小有关, 也和每个处理器分配到的任务有关, 这些问题是互相纠缠的, 在一个固定问题上作为 benchmark 来评估计算性能是有问题的
买更大的机器是为了算的更多, 而不只是算的更快
- Measure, measure, measure…
- Establish high watermarks for your program
- Be aware of scaling issues. Is the problem “well sized” for the machine?