第一次见到 CSP 201903-4 消息传递接口的时候,我的第一反应是:完了,是不是要手写一个小型并行通信协议。读完整道题才发现,它考的根本不是消息传递接口的底层实现,而是一个很纯粹的调度判断:给 n 个进程各自的指令序列,判断这些进程最终能不能全部跑完,会不会卡死。这种题在算法竞赛里被归为模拟,但很多人在上面只能拿三四十分,原因不是不会模拟,而是没搞清楚“哪条指令能跟哪条指令配对”“什么时候该停下来”这两件事。
这题非常适合用来练手,因为它代码量不大,但设计得很有意思。只要你掌握了阻塞收发的基本规则,再用一个 while 循环反复扫描每个进程的当前指令,找到能配对的就同时推进,最后看是不是所有进程都执行完,基本就能拿满分。下面我从题意、匹配逻辑、完整代码和踩坑点四个角度把它彻底拆开。
1. 题目在等一个“配对成功”,不是在模拟操作系统
1.1 S 和 R 是一对同步原语
先看最核心的定义。每个进程有一串指令,指令只有两种,一种是发送,一种是接收:
S x:当前进程向进程 x 发送一个消息。这个消息没被对方接收之前,当前进程不能执行下一条指令。R x:当前进程从进程 x 接收一个消息。消息没到之前,当前进程也不能执行下一条指令。
也就是说,每个进程的执行是阻塞式的。它站在某条指令上,如果指令条件是发送,它就一直等对方来收;如果指令条件是接收,它就一直等对方来发。只有当“当前进程的发送”和“另一个进程的接收”刚好对上时,两边才能一起往前走。
这种规则并不难理解,你可以把它想象成两个人打电话:一个人必须等对方接起来,另一个人必须等对方拨过来。两个人的“当前动作”如果不匹配,电话就永远打不通。
1.2 一个例子:三进程怎么跑通
假设有 3 个进程,0、1、2,指令序列分别是:
| 进程 | 指令序列 |
|---|---|
| 0 | S 1 |
| 1 | R 0 S 2 |
| 2 | R 1 |
一开始,进程 0 的当前指令是S 1,意思是要发送给进程 1。进程 1 的当前指令是R 0,意思是要从进程 0 接收。这两个指令正好配对:进程 0 想发给 1,进程 1 想从 0 收。于是 0 和 1 同时推进。
推进之后,进程 0 执行完毕,进程 1 的下一条指令变成S 2,进程 2 的当前指令是R 1。这又是一对匹配项:1 想发给 2,2 想从 1 收。于是 1 和 2 继续推进。最终全部进程执行完,答案是 0,表示没有死锁。
再看一个死锁例子:
| 进程 | 指令序列 |
|---|---|
| 0 | S 1 |
| 1 | S 0 |
进程 0 想发送给 1,但进程 1 的当前指令不是R 0,而是S 0,进程 1 也想向 0 发送。两边都在等对方先收,结果谁也动不了。这种情况下答案是 1。
1.3 最该避免的误判:把“目标编号相同”当配对
刚开始写这题时,很多人会犯一个直觉错误:以为配对条件是两条指令的目标编号相同。比如进程 0 的S 1和进程 1 的S 1,其实两边都在给进程 1 发消息,根本没有接收方,当然不能配对。
正确的配对条件是:进程 i 的当前指令如果是S j,那么进程 j 的当前指令必须是R i。这里两个编号不一定相等,i 和 j 是不同进程的编号;S指令里的编号是接收方,R指令里的编号是发送方。只有把这两个编号交叉对上,才算一次成功配对。
如果你按“目标编号相等”去匹配,样例也许能蒙对几个,但稍微构造几组数据就会错。这条规则一定要刻在脑子里。
2. 当前指令就是唯一入口:核心匹配逻辑与终止条件
2.1 为什么不同进程的当前指令可以直接配对
阻塞语义带来一个非常重要的性质:每个进程在执行过程中,永远只有一个“卡住”的位置,就是它还没执行的第一条指令。进程不可能先执行后面的指令,再回头处理前面的指令。所以全局状态在任何时刻都可以用 n 个当前位置来表示。
在这种情况下,死锁的判断变得很简单:只要不断检查所有进程的当前位置,如果发现两个进程的当前位置能配对,就同时推进它们。不能配对的那些进程,继续停在原地等待。
有人可能会问:如果第一轮找不到配对,是不是可能后面某一轮才能找到?答案是不是。因为每个进程只会停在当前位置,它不能跳过当前指令去执行后面的指令。当前位置没有匹配项,就意味着这个进程在当前状态下无法进行任何一步。除非某些进程先推进,改变了当前位置,否则不会有新的配对冒出来。因此 while 循环每轮扫描,只要发现有配对就推进;如果一整轮扫描下来没有任何进程推进,说明状态已经凝固,直接判定死锁。
这看起来像贪心,但其实是唯一正确解。原因在于每条指令只有唯一的潜在伙伴:进程 i 的S j只能等进程 j 的R i,没有第二种选择。所以先推进哪一对都不会出现“选错”的情况,因为能推进的配对在当前状态下本来就是强制性的。
2.2 终止条件只有两种:跑完和卡住
模拟题的终止条件特别重要。这道题的终止条件不是“按某个固定圈数跑完”,而是看当前轮次有没有进展:
- 如果某个进程执行完所有指令,它就不再参与匹配。
- 如果当前所有未执行完的进程都找不到可以配对的当前指令,那么整个系统死锁。
- 如果所有进程都执行完毕,那么没有死锁。
实现上,我用一个布尔变量记录这一轮是否发生了“推进”。如果发生了,就继续下一轮扫描;如果一整轮都没有任何推进,就退出循环。退出之后再检查一遍所有进程的当前位置是否都到了末尾。是则输出 0,否则输出 1。
这个“推进”标记是整个算法的灵魂。少了它,代码可能会陷入死循环;乱写它,又可能在还没跑完时就提前退出。
2.3 复杂度估算:为什么暴力扫描也没问题
如果只考虑暴力扫描,每一轮看 n 个进程,每次成功配对会推进两个进程。假设所有指令总数为 M,那么最多会有 M/2 次配对。如果每轮只推进一对,最坏情况可能要跑 O(nM) 的复杂度。
但实际题目中 n 和每个进程的指令长度都不会大到离谱,这个复杂度完全够用。即使想优化,也可以维护一个“还有未执行指令的进程集合”,每次只扫描这些进程,但代码会更复杂。对考试来说,先把正确性写出来,再考虑优化完全来得及。
3. 完整可提交代码:从读入到输出的每一步都放出来
3.1 C++ 实现
我用 vector 保存每个进程的指令,用 cur 记录当前执行位置。这样既方便判断是否执行完,也方便读取当前指令。
#include <bits/stdc++.h> using namespace std; struct Process { vector<pair<char, int>> ops; int cur = 0; bool done() const { return cur == (int)ops.size(); } }; int main() { ios::sync_with_stdio(false); cin.tie(0); int T, n; cin >> T >> n; cin.ignore(); while (T--) { vector<Process> p(n); for (int i = 0; i < n; ++i) { string line; getline(cin, line); istringstream iss(line); string token; while (iss >> token) { char type = token[0]; int target = stoi(token.substr(2)); p[i].ops.push_back({type, target}); } } while (true) { bool moved = false; for (int i = 0; i < n; ++i) { if (p[i].done()) continue; char typeI = p[i].ops[p[i].cur].first; int targetI = p[i].ops[p[i].cur].second; int j = targetI; if (j < 0 || j >= n) continue; if (p[j].done()) continue; char typeJ = p[j].ops[p[j].cur].first; int targetJ = p[j].ops[p[j].cur].second; if (typeI == 'S' && typeJ == 'R' && targetJ == i) { p[i].cur++; p[j].cur++; moved = true; } else if (typeI == 'R' && typeJ == 'S' && targetJ == i) { p[i].cur++; p[j].cur++; moved = true; } } if (!moved) break; } bool ok = true; for (int i = 0; i < n; ++i) { if (!p[i].done()) { ok = false; break; } } cout << (ok ? 0 : 1) << '\n'; } return 0; }3.2 代码注释逐段解释
这段代码最关键的部分是匹配判断:
- 取进程 i 的当前指令类型 typeI 和参数 targetI。
- 把 targetI 当作另一个进程的编号 j,去取进程 j 的当前指令。
- 如果进程 i 是
S targetI,进程 j 必须有R i。注意这里比较的是targetJ == i,而不是targetI == targetJ。 - 如果进程 i 是
R targetI,进程 j 必须有S i,同样比较targetJ == i。
之所以这样写,是因为S指令的编号是接收方,R指令的编号是发送方。两条指令配对时,前者编号是对方进程,后者编号是当前进程,两个人互相指。
另外要注意,不能把 j 直接写成targetI后,再看targetJ是否等于i,这个逻辑本身是对的;但如果不加区分地认为typeI和typeJ不相同就配对,就错了。比如进程 0 是S 1,进程 1 是R 2,type 确实不同,但进程 1 想接收的是进程 2,而不是进程 0,所以它们不能配对。
3.3 顺手给一版 Python 对照
如果平时用 Python 刷题,逻辑完全一样,只是写法更简洁一些:
def check(n, lines): ops = [line.split() for line in lines] cur = [0] * n while True: moved = False for i in range(n): if cur[i] >= len(ops[i]): continue opI = ops[i][cur[i]] typeI, targetI = opI[0], int(opI[2:]) j = targetI if j < 0 or j >= n or cur[j] >= len(ops[j]): continue opJ = ops[j][cur[j]] typeJ, targetJ = opJ[0], int(opJ[2:]) if typeI == 'S' and typeJ == 'R' and targetJ == i: cur[i] += 1 cur[j] += 1 moved = True elif typeI == 'R' and typeJ == 'S' and targetJ == i: cur[i] += 1 cur[j] += 1 moved = True if not moved: break for i in range(n): if cur[i] != len(ops[i]): return 1 return 0 T, n = map(int, input().split()) for _ in range(T): lines = [] for _ in range(n): lines.append(input()) print(check(n, lines))Python 版在数据量较大时可能比 C++ 慢,但这里的瓶颈本来就不是性能,而且line.split()天然处理了空行,写起来很舒服。
4. 这几个边界和踩坑点,才是这题真正的得分点
4.1 空指令行和 getline 的微妙关系
原题第一行是 T 和 n,接下来每个测试用例有 n 行,每行是一条指令序列。有的进程可能一条指令都没有,这一行就是空的。
很多人在 C++ 里习惯用cin >> s一条条读字符串,但遇到空行,cin >> s会直接跳过,导致后面的指令串位。正确做法是读完整行,再用istringstream按空格切分。
尤其注意,读完cin >> T >> n之后,行末还有一个换行符。如果不先cin.ignore()一下,第一次getline读到的就是空字符串,整个输入就全乱了。解决办法也简单,就是上面代码里的cin.ignore()。这道题输入就是这个套路,先统一处理后面对样例就很顺畅。
4.2 方向写反:最隐蔽的 60 分
方向写反是这题最常见、最容易在纯样例测试中漏掉的错误。很多人的判断条件是两边的 type 不同而且 target 相同,这在某些精心构造的样例下会让你怎么都调不对。
举个例子:
| 进程 | 指令序列 |
|---|---|
| 0 | S 1 |
| 1 | R 1 |
进程 0 向 1 发送,进程 1 也从 1 接收。如果只看目标编号,两个指令的 target 都是 1,有些人会误判为配对成功。但实际进程 1 想等的是进程 1 给自己发消息,而不是进程 0。它们根本等不到,应该输出 1。
正确的判断必须保证:S指令的 target 等于接收方进程号,R指令的 target 等于发送方进程号,两边指向的进程互换。用一个简单的口诀记就是:你要发给我,我正好要从你手里收。这句话写进代码就是:
- i 的当前指令是
S j - j 的当前指令是
R i
或者反过来:
- i 的当前指令是
R j - j 的当前指令是
S i
4.3 多组数据的状态重置与输出顺序
有多个测试用例时,每组数据之间要记得重置 cur 和 ops。我上面的 C++ 代码是在 while(T--) 内部重新构造 vector,所以每组数据互不影响。
还有一点是输出。题目要求每个用例输出一行结果,是 0 还是 1,不要把所有结果攒在一起最后输出,也别把 0 和 1 的含义写反。可以先用小样例确认一下:能全部跑完输出 0,卡住输出 1。
4.4 自环指令:不要给特殊情况开后门
如果题目允许进程给自身发消息,比如进程 0 的指令是S 0,那么根据阻塞语义,它不可能凭空完成,因为当前进程没有一条单独的R 0指令可以和自己配对。代码里的检查逻辑是取同一个进程的当前位置,发现它是S 0,不是R 0,于是不会推进,最终判定死锁。
这其实是统一逻辑的优点。千万不要额外写一段“如果 target 是自己,就自动成功”的特殊处理。真的需要自环配对时,必须同时存在一条合适的接收指令,而且那条指令还得正好是同一个进程的当前指令。但当前位置被S 0占住了,后面的R 0根本没机会执行,所以最终还是死锁。
5. 从消息传递接口总结出来的模拟题通用套路
5.1 先定义“一步”,再写循环
做模拟题最常见的错误是一上来就套 while 循环,却说不清每一步到底改哪些状态。消息传递接口这题里,“一步”就是找到两个可以配对的进程,让它们的当前指令都推进一格。把这个动作想清楚,代码结构就自然出来了。
以后遇到其他模拟题,也可以先问自己:全局状态是什么?一次合法操作是什么?操作之后哪些状态变了?这三个问题的答案就是模拟题的主干。
5.2 用“有没有进展”做终止,而不是用固定轮数
这道题的终止条件非常典型:不能用“最多扫 n 轮”来限制,因为有的匹配可能要等好几轮才会出现;也不能无条件 while(true),否则没进展时会死循环。
正确做法是每一轮开始前把moved置为 false,只要发生任意一次推进就置 true。整轮结束后如果还是 false,说明已经没有任何下一步可走,此时再判断所有进程是否全部完成。这几乎可以平移到所有“找配对、做推进”的模拟场景里。
5.3 把样例之外的小数据都试一遍
我自己刷这道题时,专门构造了一组非常小的边界数据来验证:
- n=1,进程 0 的指令为空行:答案应该是 0,因为没有指令,不会死锁。
- n=1,进程 0 的指令是
S 0:答案应该是 1。 - n=2,两个进程都是空指令:答案应该是 0。
- n=2,进程 0 是
S 1,进程 1 是R 0:答案应该是 0。 - n=2,进程 0 是
S 1,进程 1 是S 0:答案应该是 1。
这五组数据基本覆盖了空指令、自环、正常配对、互发死锁四种情况。如果你的代码能全部通过,再交上去基本就是稳稳的满分。刷题时最怕的不是不会写,而是样例过了、交上去错得莫名其妙。把边界数据自己想一遍,很多隐藏问题在提交前就能暴露出来。
消息传递接口这道题,说出来一点也不玄乎。它考的就是“能不能找到顺序让所有收发配对成功”。只要老老实实维护每个进程的当前指令,按匹配规则推进,再用有没有进展做终止判断,100 分并没有想象中那么难。顺手把这套思路留下来,下次遇到同类的模拟题,可以直接照着搭框架。