mirror of
https://github.com/DragonOS-Community/DragonOS.git
synced 2026-09-08 23:57:59 +08:00
refactor(rcu): segment callback queues per CPU (#2222)
* refactor(rcu): segment callback queues per CPU Replace the globally serialized callback tracker and FIFO with cache-line-aligned per-CPU queues split into done, wait, next-ready, and next segments. Keep raw callback admission allocation-free and restrict its shared work to the local queue plus a coalesced worker kick. Advance callback generations as whole segments, execute bounded round-robin batches outside RCU locks, and implement rcu_barrier() with per-CPU tail markers that cover queued and executing callbacks. Serialize barrier scans with CPU lifecycle migration while preserving the upstream participating-CPU state as the single hotplug authority. Add per-CPU callback queue debugfs snapshots and extend RCU selftests for segment transitions, migration, callback requeue, multi-batch floods, all-online-CPU admission, concurrent barriers, and the upstream CPU lifecycle protocol. Tests: make kernel Tests: make run-nographic (2 vCPUs) Tests: /opt/tests/dunitest/bin/normal/rcu_selftest_test (3/3 passed) Signed-off-by: longjin <longjin@dragonos.org> * docs(rcu): document segmented callback queues Explain the per-CPU four-segment callback model, grace-period advancement, bounded execution, barrier markers, CPU lifecycle migration, and concurrency invariants without coupling the design to issue history or implementation details. Add a matching English translation and register both documents in their respective RCU documentation indexes. Signed-off-by: longjin <longjin@dragonos.org> * docs(rcu): clarify callback lifecycle Update the Chinese and English RCU architecture overviews with the per-CPU segmented callback lifecycle, its ordering boundaries, and the distinction between grace-period waits and callback barriers. Keep the overview concise and link both languages to the detailed segmented callback queue design. Signed-off-by: longjin <longjin@dragonos.org> * style(rcu): format callback imports Signed-off-by: longjin <longjin@dragonos.org> * fix(rcu): hand off barriers when worker exits Wake barrier waiters unconditionally after publishing that the callback worker has stopped. This lets a barrier that observed the old worker state recheck its predicate and take over bounded inline execution instead of sleeping forever with queued markers. Drain a small per-CPU callback quantum before resuming round-robin selection. This avoids rescanning every possible CPU for each callback in a single-CPU backlog while preserving a strict fairness bound for other CPUs. Tests: make fmt Tests: make kernel Tests: make run-nographic Tests: dunitest normal/rcu_selftest (3/3 passed) Signed-off-by: longjin <longjin@dragonos.org> --------- Signed-off-by: longjin <longjin@dragonos.org>
This commit is contained in:
LoGin
@@ -166,21 +166,25 @@ sequenceDiagram
|
||||
|
||||
## 7. 回调生命周期
|
||||
|
||||
回调在提交时绑定到一个尚未完成的宽限期世代。其生命周期如下：
|
||||
回调先进入当前 CPU 的局部队列，再由宽限期协调器按段归类。对应的宽限期完成后，回调进入可执行段，由回调执行器分批处理。
|
||||
|
||||
```mermaid
|
||||
flowchart LR
|
||||
Admit[提交回调] --> Pending[等待目标宽限期]
|
||||
Pending --> Ready[目标宽限期完成]
|
||||
Ready --> Execute[回调执行器处理]
|
||||
Admit[本 CPU 提交] --> Next[next: 尚未归类]
|
||||
Next --> Wait[wait: 等待当前 GP]
|
||||
Next --> NextReady[next-ready: 等待下一 GP]
|
||||
NextReady --> Wait
|
||||
Wait --> Done[done: GP 已完成]
|
||||
Done --> Execute[有界批次执行]
|
||||
```
|
||||
|
||||
需要保持的原则是：
|
||||
这一过程遵循三个原则：
|
||||
|
||||
- 回调不得早于目标宽限期完成；
|
||||
- 同一执行通道中的回调顺序必须稳定；
|
||||
- 上下文转换路径只负责报告进展，不直接执行任意回调；
|
||||
- 回调执行与上下文状态机解耦，避免在 IRQ-off、Idle 边界或非 watching 状态中运行未知逻辑。
|
||||
- 每个 CPU 的队列保持先入先出，CPU 之间不保证全局执行顺序；
|
||||
- 上下文转换只报告宽限期进展，不直接执行回调。
|
||||
|
||||
需要等待此前回调全部执行完毕时，应使用 `rcu_barrier()`；`synchronize_rcu()` 只等待宽限期结束。分段模型、屏障和 CPU 迁移的详细设计见 [RCU 分段回调队列](segmented-callback-queues.md)。
|
||||
|
||||
## 8. 架构接入原则
|
||||
|
||||
|
||||
@@ -9,3 +9,4 @@ DragonOS RCU 的核心原理、上下文模型、宽限期判定和组件职责
|
||||
:maxdepth: 1
|
||||
|
||||
architecture
|
||||
segmented-callback-queues
|
||||
|
||||
@@ -0,0 +1,300 @@
|
||||
# RCU 分段回调队列
|
||||
|
||||
## 1. 设计目标
|
||||
|
||||
RCU 回调用于把资源释放等延迟操作安排到宽限期之后执行。一个可扩展的回调系统需要同时满足两类要求：
|
||||
|
||||
- 提交回调是高频操作，不应让所有 CPU 持续争用同一把全局锁；
|
||||
- 回调只有在它所依赖的宽限期结束后才能执行，屏障和 CPU 生命周期变化也不能破坏这一语义。
|
||||
|
||||
DragonOS 使用每 CPU 分段队列组织回调。每个 CPU 在自己的局部队列中提交工作，宽限期协调器以“段”为单位推进回调状态，执行器则以有界批次消费已经就绪的回调。
|
||||
|
||||
这套设计借鉴 Linux RCU 分段回调链表的核心思想，但与 DragonOS 当前的宽限期模型相匹配：保留单一宽限期协调器和单一回调执行器，不引入多级 RCU 层级、回调卸载或额外执行通道。
|
||||
|
||||
## 2. 为什么需要分段
|
||||
|
||||
最直接的回调队列是一条全局链表：提交者把回调加入链表，宽限期结束后再扫描链表，找出可以执行的节点。这种模型简单，但存在两个问题：
|
||||
|
||||
1. 所有 CPU 的提交都会竞争全局队列；
|
||||
2. 宽限期变化时，需要逐个检查或更新回调的目标世代。
|
||||
|
||||
分段队列将“等待相同宽限期进展的回调”组织在一起。段的边界本身表示回调状态，因此宽限期推进时只需移动链表边界，不需要遍历其中的每个回调。
|
||||
|
||||
```mermaid
|
||||
flowchart LR
|
||||
Submit[提交回调] --> Next[next<br/>尚未归类]
|
||||
Next --> Wait[wait<br/>等待当前宽限期]
|
||||
Next --> NextReady[next-ready<br/>等待下一宽限期]
|
||||
NextReady --> Wait
|
||||
Wait --> Done[done<br/>可以执行]
|
||||
Done --> Run[执行回调]
|
||||
```
|
||||
|
||||
分段带来的关键收益是：
|
||||
|
||||
- 提交操作主要访问本 CPU 数据；
|
||||
- 宽限期推进的成本与 CPU 数量相关，而不是与积压回调数量相关；
|
||||
- 回调所处的生命周期阶段可以直接从段位置判断；
|
||||
- 屏障、迁移和调试都可以围绕同一状态模型实现。
|
||||
|
||||
## 3. 整体架构
|
||||
|
||||
回调处理分为提交、协调和执行三部分。
|
||||
|
||||
```mermaid
|
||||
flowchart TB
|
||||
subgraph Producers[回调提交者]
|
||||
C0[CPU 0]
|
||||
C1[CPU 1]
|
||||
CN[CPU N]
|
||||
end
|
||||
|
||||
subgraph Queues[每 CPU 分段队列]
|
||||
Q0[CPU 0<br/>done / wait / next-ready / next]
|
||||
Q1[CPU 1<br/>done / wait / next-ready / next]
|
||||
QN[CPU N<br/>done / wait / next-ready / next]
|
||||
end
|
||||
|
||||
GP[宽限期协调器]
|
||||
Worker[回调执行器]
|
||||
|
||||
C0 --> Q0
|
||||
C1 --> Q1
|
||||
CN --> QN
|
||||
Q0 <--> GP
|
||||
Q1 <--> GP
|
||||
QN <--> GP
|
||||
Q0 --> Worker
|
||||
Q1 --> Worker
|
||||
QN --> Worker
|
||||
```
|
||||
|
||||
各部分职责如下：
|
||||
|
||||
- **提交者**只负责把回调放入当前 CPU 的 `next` 段，并通知系统存在新工作；
|
||||
- **每 CPU 队列**保存回调及其宽限期阶段，不负责决定宽限期何时开始或结束；
|
||||
- **宽限期协调器**维护全局宽限期状态，并在状态变化时推进各 CPU 的队列段；
|
||||
- **回调执行器**只消费 `done` 段，在不持有 RCU 内部锁的情况下调用回调函数。
|
||||
|
||||
每 CPU 队列隔离了高频写入，而宽限期状态仍由唯一协调器维护。这样既避免复制全局状态，也不会为了优化提交路径而引入多套相互竞争的宽限期判断。
|
||||
|
||||
## 4. 四段状态模型
|
||||
|
||||
每个 CPU 的回调队列由四个保持 FIFO 顺序的逻辑段组成。
|
||||
|
||||
| 段 | 含义 | 是否可以执行 |
|
||||
|---|---|---|
|
||||
| `next` | 新提交、尚未绑定到某次宽限期 | 否 |
|
||||
| `next-ready` | 已确定需要等待下一次宽限期 | 否 |
|
||||
| `wait` | 正在等待当前宽限期完成 | 否 |
|
||||
| `done` | 所依赖的宽限期已经完成 | 是 |
|
||||
|
||||
段之间的转换由宽限期事件驱动：
|
||||
|
||||
```mermaid
|
||||
stateDiagram-v2
|
||||
[*] --> Next: 提交
|
||||
Next --> Wait: 绑定当前或新启动的宽限期
|
||||
Next --> NextReady: 当前宽限期已开始
|
||||
NextReady --> Wait: 下一宽限期开始
|
||||
Wait --> Done: 对应宽限期完成
|
||||
Done --> Executing: 执行器取出
|
||||
Executing --> [*]: 回调返回
|
||||
Executing --> Next: 回调重新提交自身
|
||||
```
|
||||
|
||||
需要始终成立的不变量是：
|
||||
|
||||
- 一个已提交回调只能位于一个段中，或正由执行器执行；
|
||||
- 只有 `done` 段中的回调可以被执行；
|
||||
- 同一段内保持提交顺序；
|
||||
- 段的宽限期含义一致，不能把等待不同世代的回调混入同一等待段；
|
||||
- 跨 CPU 不提供全局回调执行顺序保证；
|
||||
- 回调函数运行期间不持有队列锁或宽限期锁。
|
||||
|
||||
这里的四段不是四份独立工作流，而是同一条回调生命周期的四个阶段。实现可以用多条局部链表表达这些段，只要段合并和移动保持常数时间，并维持上述语义。
|
||||
|
||||
## 5. 提交与宽限期推进
|
||||
|
||||
### 5.1 局部提交
|
||||
|
||||
提交回调时，CPU 先稳定自己的执行位置，再把回调追加到本地 `next` 段。提交完成后发布一个持久的“需要扫描”条件，并在必要时唤醒回调执行器。
|
||||
|
||||
```mermaid
|
||||
sequenceDiagram
|
||||
participant Caller as 提交者
|
||||
participant Local as 本 CPU 队列
|
||||
participant Signal as 工作通知
|
||||
participant Worker as 回调执行器
|
||||
|
||||
Caller->>Caller: 稳定当前 CPU
|
||||
Caller->>Local: 追加到 next
|
||||
Caller->>Signal: 发布需要扫描
|
||||
Signal-->>Worker: 必要时唤醒
|
||||
```
|
||||
|
||||
本地提交不负责启动或推进宽限期。这样可以让高频路径保持短小，并避免所有 CPU 在每次提交时访问全局宽限期状态。
|
||||
|
||||
工作通知只是减少无效扫描和唤醒的性能机制。队列状态和宽限期状态才是正确性的依据；即使通知被合并，执行器也必须能通过持久状态重新发现尚未处理的工作。
|
||||
|
||||
### 5.2 整段推进
|
||||
|
||||
宽限期协调器观察到新回调后，将 `next` 段绑定到合适的宽限期：
|
||||
|
||||
- 没有活动宽限期时，回调进入 `wait`，并请求一次新的宽限期；
|
||||
- 已有活动宽限期且回调来不及被其覆盖时，回调进入 `next-ready`，等待下一次宽限期；
|
||||
- 宽限期完成时，对应的 `wait` 整段进入 `done`；
|
||||
- 后续宽限期开始时，`next-ready` 整段进入 `wait`。
|
||||
|
||||
协调器必须先明确宽限期事件，再据此移动各队列的段。并发提交到达扫描边界时，可以保守地归入后一宽限期，但绝不能被错误地归入已经无法覆盖它的宽限期。
|
||||
|
||||
这种“允许多等、禁止少等”的边界规则保持了 RCU 安全性，同时不要求提交者与全局协调器进行昂贵同步。
|
||||
|
||||
## 6. 回调执行与公平性
|
||||
|
||||
DragonOS 使用单一逻辑执行权消费所有 CPU 的 `done` 段。单一执行权使回调之间不会因多个执行通道而产生新的并行语义，也简化了屏障对“正在执行的回调”的判断。
|
||||
|
||||
执行器采用轮转和有界批次：
|
||||
|
||||
1. 从上一次停止位置之后的 CPU 开始扫描；
|
||||
2. 从某个 `done` 段取出一个回调，并记录该队列有回调正在执行；
|
||||
3. 释放所有 RCU 内部锁后调用回调；
|
||||
4. 回调返回后发布执行完成状态；
|
||||
5. 达到批次上限时提供调度机会，再继续处理剩余工作。
|
||||
|
||||
```mermaid
|
||||
flowchart TD
|
||||
Scan[轮转扫描每 CPU 队列] --> Ready{存在 done 回调?}
|
||||
Ready -->|否| NextCPU[检查下一个 CPU]
|
||||
Ready -->|是| Pop[摘取一个回调]
|
||||
Pop --> Unlock[释放 RCU 内部锁]
|
||||
Unlock --> Invoke[执行回调]
|
||||
Invoke --> Complete[发布执行完成]
|
||||
Complete --> Limit{达到批次上限?}
|
||||
Limit -->|否| Scan
|
||||
Limit -->|是| Yield[提供调度机会]
|
||||
Yield --> Scan
|
||||
```
|
||||
|
||||
轮转避免某个持续产生回调的 CPU 长期占用执行器；批次上限避免大量回调让内核任务长时间失去调度机会。回调本身仍应遵守其执行上下文约束，分批机制不能消除单个慢回调造成的延迟。
|
||||
|
||||
## 7. RCU 屏障
|
||||
|
||||
等待一个宽限期结束，与等待此前提交的回调全部执行完成，是两个不同的操作。前者只证明旧读者已经离开；回调可能仍停留在 `done` 段中。RCU 屏障必须额外等待回调执行。
|
||||
|
||||
分段队列使用每队列尾标记实现屏障：屏障在每个包含未完成工作的队列中追加一个特殊回调，然后等待所有标记被执行。标记之前的队列前缀因此必然已经执行完毕。
|
||||
|
||||
```mermaid
|
||||
sequenceDiagram
|
||||
participant Barrier as 屏障调用者
|
||||
participant Queues as 每 CPU 队列
|
||||
participant Worker as 回调执行器
|
||||
|
||||
Barrier->>Queues: 稳定回调所有权
|
||||
loop 每个队列
|
||||
Barrier->>Queues: 在已有工作之后追加尾标记
|
||||
end
|
||||
Barrier->>Worker: 唤醒执行器
|
||||
Worker->>Queues: 按正常宽限期规则推进和执行
|
||||
Worker-->>Barrier: 每个尾标记完成一次计数
|
||||
Barrier->>Barrier: 所有标记完成后返回
|
||||
```
|
||||
|
||||
屏障快照的边界，是它在各队列锁保护下插入标记或确认队列无未完成工作的时刻：
|
||||
|
||||
- 标记之前的回调属于本次屏障，标记之后的新提交不属于本次屏障；
|
||||
- 已经从队列摘取但尚未返回的回调也属于未完成工作，不能因为链表暂时为空而漏掉；
|
||||
- 标记跟随普通回调经历相同的分段推进，因此不会绕过宽限期；
|
||||
- 标记应接在已有工作的最后，而不是无条件等待一次额外宽限期。
|
||||
|
||||
多个屏障调用需要串行化其快照和标记生命周期。屏障扫描还必须与 CPU 队列迁移互斥，否则旧回调可能从尚未扫描的源队列移动到已经扫描过的目标队列，造成漏等待。
|
||||
|
||||
## 8. CPU 生命周期与队列迁移
|
||||
|
||||
CPU 下线会改变回调队列的所有权，但不能改变回调原有的宽限期含义。下线流程应先阻止目标 CPU 接收新的普通工作，再把它从后续宽限期参与集合中移除，最后把尚未执行的回调迁移到仍参与 RCU 的 CPU。
|
||||
|
||||
```mermaid
|
||||
flowchart TD
|
||||
Dying[CPU 进入下线阶段] --> Stop[停止接收新的普通提交]
|
||||
Stop --> GP[结清该 CPU 的宽限期责任]
|
||||
GP --> Owner[稳定回调队列所有权]
|
||||
Owner --> Align[按同一宽限期状态对齐源和目标队列]
|
||||
Align --> Merge[分别合并四个同名段]
|
||||
Merge --> Wake[通知回调执行器]
|
||||
Wake --> Dead[完成 CPU 下线]
|
||||
```
|
||||
|
||||
迁移遵循以下原则：
|
||||
|
||||
- `done`、`wait`、`next-ready` 和 `next` 分别合并，不能把待完成回调直接变为可执行；
|
||||
- 源队列和目标队列必须基于同一个宽限期状态完成对齐；
|
||||
- 队列迁移与屏障扫描共享同一所有权同步域；
|
||||
- 正在执行的回调无需迁移，其完成状态仍由原队列记录；
|
||||
- 队列存储的生命周期必须长于 CPU 在线周期，使下线后仍可安全完成记账；
|
||||
- 迁移目标必须来自 RCU 自己的参与 CPU 集合，不能使用语义更宽泛、更新时序不同的在线状态替代。
|
||||
|
||||
CPU 生命周期状态是队列归属的唯一事实来源。回调子系统不维护第二套 online/offline 状态，以免两套状态在并发下线时产生分歧。
|
||||
|
||||
## 9. 并发模型
|
||||
|
||||
分段队列包含三类同步域：
|
||||
|
||||
- **宽限期状态**：保护全局宽限期的开始、完成和参与 CPU 集合；
|
||||
- **回调所有权**：协调屏障全局快照与 CPU 队列迁移；
|
||||
- **局部队列**：保护单个 CPU 的分段链表和执行状态。
|
||||
|
||||
需要同时进入多个同步域的控制路径采用统一方向：先稳定宽限期和所有权，再按确定顺序访问局部队列。普通提交只访问一个局部队列，不反向进入全局同步域。
|
||||
|
||||
```mermaid
|
||||
flowchart LR
|
||||
GP[宽限期状态] --> Ownership[回调所有权]
|
||||
Ownership --> Q0[较小 CPU 编号队列]
|
||||
Q0 --> Q1[较大 CPU 编号队列]
|
||||
|
||||
Submit[普通提交] --> Local[单个本地队列]
|
||||
```
|
||||
|
||||
这一方向约束同时服务于正确性和可维护性：新增控制路径时，只要遵守相同的层次，就不会与已有路径形成反向加锁。
|
||||
|
||||
工作发布还必须遵循“先发布持久条件，后发送唤醒”的原则。唤醒可以合并，也可能发生在执行器检查睡眠条件附近；执行器因此必须在睡眠前复查队列和宽限期是否仍有可推进工作，闭合丢唤醒窗口。尚被活动宽限期阻塞的等待段不属于当前可运行工作，不能让执行器空转。
|
||||
|
||||
## 10. 正确性边界
|
||||
|
||||
### 10.1 回调重复提交
|
||||
|
||||
同一个回调节点在入队期间不能再次提交。执行器应在调用回调函数之前解除节点的入队状态，因此回调可以在自身执行过程中再次提交同一节点；新的提交属于新的队列位置和宽限期判断。
|
||||
|
||||
### 10.2 顺序保证
|
||||
|
||||
同一局部段保持 FIFO 顺序，段的推进也保持段内顺序。不同 CPU 的回调没有全局提交或完成顺序。需要全局完成边界的调用者应使用 RCU 屏障，而不是依赖偶然的执行次序。
|
||||
|
||||
### 10.3 内存可见性
|
||||
|
||||
队列锁负责发布回调节点内容和链表关系；宽限期协议负责建立读侧临界区结束与回调可执行之间的顺序；屏障标记的完成发布则保证屏障返回发生在其覆盖的回调返回之后。
|
||||
|
||||
用于唤醒或减少扫描的原子标志只是提示，不能单独承担节点生命周期、宽限期完成或屏障完成的证明。
|
||||
|
||||
### 10.4 失败与退化
|
||||
|
||||
回调提交路径不依赖动态分配，因此内存压力不会使已经准备好的回调无法入队。大量回调会增加排队延迟，但有界批次和轮转保证系统仍有调度机会。CPU 下线和执行器启动阶段也必须沿用相同的状态机，不能通过提前完成回调来绕过正常宽限期。
|
||||
|
||||
## 11. 可观测性
|
||||
|
||||
调试接口可以按 CPU 展示四个段的长度和是否存在正在执行的回调，并提供聚合视图。这样的快照用于回答两类问题：
|
||||
|
||||
- 回调积压集中在哪些 CPU；
|
||||
- 积压发生在等待宽限期、尚未归类，还是已经可执行的阶段。
|
||||
|
||||
调试快照不参与正确性判断，也不要求全系统线性一致。读取时逐个短暂观察局部队列，可以避免为了诊断而在所有 CPU 之间建立新的全局停顿。一次打开操作应看到稳定文本，避免分段读取时把不同时间点的数据拼接在一起。
|
||||
|
||||
## 12. 设计取舍
|
||||
|
||||
当前设计有意保持以下边界：
|
||||
|
||||
- 使用每 CPU 队列减少提交争用，但保留单一宽限期协调器；
|
||||
- 使用单一逻辑回调执行权，避免引入回调并行执行语义；
|
||||
- 使用固定的有界批次和轮转公平性，不额外引入时间预算调度器；
|
||||
- 不实现多级 RCU 节点层次、回调卸载、延迟回调或多执行器；
|
||||
- 不维护全局回调序号，跨队列完成边界由尾标记屏障表达。
|
||||
|
||||
这些边界让回调提交成本随 CPU 局部化，同时把跨 CPU 协调限制在宽限期变化、屏障和 CPU 生命周期等低频路径。若未来工作负载证明单一协调器或执行器成为瓶颈，应基于测量结果扩展对应层次，而不是预先引入 Linux Tree RCU 的全部复杂度。
|
||||
@@ -166,21 +166,25 @@ Publishing context state, taking snapshots, and publishing grace-period activity
|
||||
|
||||
## 7. Callback Lifecycle
|
||||
|
||||
A callback is associated at admission time with a grace-period generation that has not yet completed.
|
||||
A callback first enters the current CPU's local queue, where the grace-period coordinator classifies it by segment. After the required grace period completes, the callback moves to an executable segment and is processed in a bounded batch.
|
||||
|
||||
```mermaid
|
||||
flowchart LR
|
||||
Admit[Admit callback] --> Pending[Wait for target grace period]
|
||||
Pending --> Ready[Target grace period completes]
|
||||
Ready --> Execute[Callback executor runs it]
|
||||
Admit[Submit on this CPU] --> Next[next: unclassified]
|
||||
Next --> Wait[wait: waiting for current GP]
|
||||
Next --> NextReady[next-ready: waiting for next GP]
|
||||
NextReady --> Wait
|
||||
Wait --> Done[done: GP completed]
|
||||
Done --> Execute[Execute in bounded batches]
|
||||
```
|
||||
|
||||
The following principles must hold:
|
||||
This process follows three principles:
|
||||
|
||||
- A callback must not run before its target grace period completes.
|
||||
- Callback order within one execution channel must remain stable.
|
||||
- Context-transition paths report progress but do not directly execute arbitrary callbacks.
|
||||
- Callback execution remains separate from the context state machine, preventing unknown logic from running at IRQ-disabled boundaries, idle boundaries, or while the CPU is not watching.
|
||||
- Each per-CPU queue preserves FIFO order; no global execution order is guaranteed across CPUs.
|
||||
- Context-transition paths report grace-period progress but do not execute callbacks directly.
|
||||
|
||||
Use `rcu_barrier()` when all previously submitted callbacks must finish; `synchronize_rcu()` waits only for a grace period to end. See [RCU Segmented Callback Queues](segmented-callback-queues.md) for the detailed segment, barrier, and CPU migration design.
|
||||
|
||||
## 8. Architecture Integration Principles
|
||||
|
||||
|
||||
@@ -11,3 +11,4 @@ RCU.
|
||||
:maxdepth: 1
|
||||
|
||||
architecture
|
||||
segmented-callback-queues
|
||||
|
||||
@@ -0,0 +1,300 @@
|
||||
# RCU Segmented Callback Queues
|
||||
|
||||
## 1. Design Goals
|
||||
|
||||
RCU callbacks schedule deferred operations, such as resource reclamation, to run after a grace period. A scalable callback system must satisfy two requirements:
|
||||
|
||||
- callback submission is frequent and should not make every CPU contend for one global lock;
|
||||
- a callback may run only after its required grace period has ended, including during barriers and CPU lifecycle changes.
|
||||
|
||||
DragonOS organizes callbacks in per-CPU segmented queues. Each CPU submits work to its local queue, the grace-period coordinator advances callback state one segment at a time, and the executor consumes ready callbacks in bounded batches.
|
||||
|
||||
This design adopts the central idea of Linux RCU segmented callback lists while matching the current DragonOS grace-period model. It retains one grace-period coordinator and one callback executor without introducing a multilevel RCU hierarchy, callback offloading, or additional execution paths.
|
||||
|
||||
## 2. Why Segmentation Is Needed
|
||||
|
||||
The simplest callback queue is a single global list: producers append callbacks, and the system scans the list after a grace period to find callbacks that may run. Although straightforward, this model has two problems:
|
||||
|
||||
1. callback submission from every CPU contends for the global queue;
|
||||
2. a grace-period transition requires examining or updating callbacks individually.
|
||||
|
||||
A segmented queue groups callbacks that are waiting for the same grace-period progress. Segment boundaries represent callback state, so a grace-period transition moves list boundaries instead of traversing every callback in the segment.
|
||||
|
||||
```mermaid
|
||||
flowchart LR
|
||||
Submit[Submit callback] --> Next[next<br/>unclassified]
|
||||
Next --> Wait[wait<br/>waiting for current grace period]
|
||||
Next --> NextReady[next-ready<br/>waiting for next grace period]
|
||||
NextReady --> Wait
|
||||
Wait --> Done[done<br/>ready to execute]
|
||||
Done --> Run[Execute callback]
|
||||
```
|
||||
|
||||
Segmentation provides several important properties:
|
||||
|
||||
- submission primarily accesses data local to the current CPU;
|
||||
- grace-period advancement costs scale with the number of CPUs rather than the callback backlog;
|
||||
- a callback's lifecycle stage follows directly from its segment;
|
||||
- barriers, migration, and observability use the same state model.
|
||||
|
||||
## 3. Overall Architecture
|
||||
|
||||
Callback processing consists of submission, coordination, and execution.
|
||||
|
||||
```mermaid
|
||||
flowchart TB
|
||||
subgraph Producers[Callback producers]
|
||||
C0[CPU 0]
|
||||
C1[CPU 1]
|
||||
CN[CPU N]
|
||||
end
|
||||
|
||||
subgraph Queues[Per-CPU segmented queues]
|
||||
Q0[CPU 0<br/>done / wait / next-ready / next]
|
||||
Q1[CPU 1<br/>done / wait / next-ready / next]
|
||||
QN[CPU N<br/>done / wait / next-ready / next]
|
||||
end
|
||||
|
||||
GP[Grace-period coordinator]
|
||||
Worker[Callback executor]
|
||||
|
||||
C0 --> Q0
|
||||
C1 --> Q1
|
||||
CN --> QN
|
||||
Q0 <--> GP
|
||||
Q1 <--> GP
|
||||
QN <--> GP
|
||||
Q0 --> Worker
|
||||
Q1 --> Worker
|
||||
QN --> Worker
|
||||
```
|
||||
|
||||
Each part has a distinct responsibility:
|
||||
|
||||
- **Producers** append callbacks to the current CPU's `next` segment and announce that work is available.
|
||||
- **Per-CPU queues** retain callbacks and their grace-period stage, but do not decide when a grace period starts or ends.
|
||||
- **The grace-period coordinator** maintains global grace-period state and advances per-CPU segments when that state changes.
|
||||
- **The callback executor** consumes only the `done` segments and invokes callbacks without holding internal RCU locks.
|
||||
|
||||
Per-CPU queues isolate frequent writes while the single coordinator remains the authority for grace-period state. This avoids duplicating global state or introducing competing grace-period decisions merely to optimize submission.
|
||||
|
||||
## 4. The Four-Segment State Model
|
||||
|
||||
Each per-CPU callback queue contains four logical FIFO segments.
|
||||
|
||||
| Segment | Meaning | Executable |
|
||||
|---|---|---|
|
||||
| `next` | Newly submitted and not yet associated with a grace period | No |
|
||||
| `next-ready` | Known to require the next grace period | No |
|
||||
| `wait` | Waiting for the current grace period to complete | No |
|
||||
| `done` | Its required grace period has completed | Yes |
|
||||
|
||||
Grace-period events drive transitions between segments:
|
||||
|
||||
```mermaid
|
||||
stateDiagram-v2
|
||||
[*] --> Next: Submit
|
||||
Next --> Wait: Associate with current or newly started grace period
|
||||
Next --> NextReady: Current grace period has already started
|
||||
NextReady --> Wait: Next grace period starts
|
||||
Wait --> Done: Required grace period completes
|
||||
Done --> Executing: Executor removes callback
|
||||
Executing --> [*]: Callback returns
|
||||
Executing --> Next: Callback resubmits itself
|
||||
```
|
||||
|
||||
The following invariants must always hold:
|
||||
|
||||
- a submitted callback is in exactly one segment or is being executed;
|
||||
- only callbacks in `done` may be executed;
|
||||
- submission order is preserved within a segment;
|
||||
- callbacks waiting for different grace-period generations are not mixed in one waiting segment;
|
||||
- no global callback execution order is guaranteed across CPUs;
|
||||
- callback functions run without queue or grace-period locks held.
|
||||
|
||||
The four segments are four stages of one callback lifecycle, not four independent workflows. An implementation may represent them with separate local lists as long as segment movement and merging remain constant-time operations and preserve these semantics.
|
||||
|
||||
## 5. Submission and Grace-Period Advancement
|
||||
|
||||
### 5.1 Local Submission
|
||||
|
||||
To submit a callback, a CPU first stabilizes its execution location and then appends the callback to its local `next` segment. It subsequently publishes a persistent “scan needed” condition and wakes the callback executor when necessary.
|
||||
|
||||
```mermaid
|
||||
sequenceDiagram
|
||||
participant Caller as Producer
|
||||
participant Local as Local CPU queue
|
||||
participant Signal as Work notification
|
||||
participant Worker as Callback executor
|
||||
|
||||
Caller->>Caller: Stabilize current CPU
|
||||
Caller->>Local: Append to next
|
||||
Caller->>Signal: Publish scan-needed state
|
||||
Signal-->>Worker: Wake if necessary
|
||||
```
|
||||
|
||||
Local submission neither starts nor advances a grace period. This keeps the frequent path short and avoids accessing global grace-period state for every callback.
|
||||
|
||||
Work notification is a performance mechanism that reduces unnecessary scans and wakeups. Queue and grace-period state remain the source of correctness. Even when notifications are coalesced, the executor must be able to rediscover outstanding work from persistent state.
|
||||
|
||||
### 5.2 Whole-Segment Advancement
|
||||
|
||||
After the grace-period coordinator observes new callbacks, it associates the `next` segment with an appropriate grace period:
|
||||
|
||||
- if no grace period is active, callbacks move to `wait` and a new grace period is requested;
|
||||
- if an active grace period can no longer cover the callbacks, they move to `next-ready` for the following grace period;
|
||||
- when a grace period completes, its `wait` segment moves as a whole to `done`;
|
||||
- when the following grace period starts, `next-ready` moves as a whole to `wait`.
|
||||
|
||||
The coordinator must establish the grace-period event before moving queue segments according to that event. A callback racing with a scan boundary may conservatively wait for a later grace period, but it must never be associated with a grace period that cannot cover it.
|
||||
|
||||
This rule—waiting longer is safe, waiting too little is not—preserves RCU safety without expensive synchronization between every producer and the global coordinator.
|
||||
|
||||
## 6. Callback Execution and Fairness
|
||||
|
||||
DragonOS uses one logical execution owner to consume the `done` segments of all CPUs. A single execution owner avoids introducing parallel callback semantics and simplifies the barrier's treatment of callbacks that are already executing.
|
||||
|
||||
The executor combines round-robin selection with bounded batches:
|
||||
|
||||
1. begin scanning after the CPU at which the previous scan stopped;
|
||||
2. remove one callback from a `done` segment and record that the queue has a callback in progress;
|
||||
3. release all internal RCU locks before invoking the callback;
|
||||
4. publish completion after the callback returns;
|
||||
5. provide a scheduling opportunity after reaching the batch limit, then continue with remaining work.
|
||||
|
||||
```mermaid
|
||||
flowchart TD
|
||||
Scan[Scan per-CPU queues round-robin] --> Ready{A done callback exists?}
|
||||
Ready -->|No| NextCPU[Check the next CPU]
|
||||
Ready -->|Yes| Pop[Remove one callback]
|
||||
Pop --> Unlock[Release internal RCU locks]
|
||||
Unlock --> Invoke[Invoke callback]
|
||||
Invoke --> Complete[Publish completion]
|
||||
Complete --> Limit{Batch limit reached?}
|
||||
Limit -->|No| Scan
|
||||
Limit -->|Yes| Yield[Provide a scheduling opportunity]
|
||||
Yield --> Scan
|
||||
```
|
||||
|
||||
Round-robin selection prevents a CPU with a continuous callback stream from monopolizing the executor. Bounded batches prevent large backlogs from denying other kernel tasks a scheduling opportunity. A callback must still obey the constraints of its execution context; batching cannot eliminate latency caused by one slow callback.
|
||||
|
||||
## 7. RCU Barriers
|
||||
|
||||
Waiting for a grace period to end is different from waiting for all previously submitted callbacks to finish. A completed grace period proves that old readers have left, but callbacks may still remain in `done`. An RCU barrier must additionally wait for callback execution.
|
||||
|
||||
Segmented queues implement a barrier with a tail marker for each queue. The barrier appends a special callback after the unfinished work in every relevant queue, then waits until all markers execute. The queue prefix before each marker must therefore have finished.
|
||||
|
||||
```mermaid
|
||||
sequenceDiagram
|
||||
participant Barrier as Barrier caller
|
||||
participant Queues as Per-CPU queues
|
||||
participant Worker as Callback executor
|
||||
|
||||
Barrier->>Queues: Stabilize callback ownership
|
||||
loop Each queue
|
||||
Barrier->>Queues: Append a tail marker after existing work
|
||||
end
|
||||
Barrier->>Worker: Wake the executor
|
||||
Worker->>Queues: Advance and execute using normal grace-period rules
|
||||
Worker-->>Barrier: Account for each completed marker
|
||||
Barrier->>Barrier: Return after every marker completes
|
||||
```
|
||||
|
||||
The barrier snapshot boundary for a queue is the instant, under that queue's synchronization, at which the barrier inserts a marker or confirms that no unfinished work exists:
|
||||
|
||||
- callbacks before the marker belong to the barrier; new submissions after it do not;
|
||||
- a callback already removed from a queue but not yet returned is still unfinished and cannot be ignored merely because the list is empty;
|
||||
- markers follow the same segment transitions as ordinary callbacks and cannot bypass a grace period;
|
||||
- a marker is placed after existing work instead of unconditionally requiring an extra grace period.
|
||||
|
||||
Concurrent barrier calls must serialize their snapshots and marker lifetimes. Barrier scanning must also exclude CPU queue migration. Otherwise, old callbacks could move from an unscanned source queue to a destination queue that the barrier has already scanned, causing the barrier to miss them.
|
||||
|
||||
## 8. CPU Lifecycle and Queue Migration
|
||||
|
||||
Taking a CPU offline changes callback queue ownership but must not change the grace-period requirements of its callbacks. The offline sequence first prevents the CPU from accepting new ordinary work, removes its responsibility from future grace periods, and then migrates its unexecuted callbacks to a CPU that still participates in RCU.
|
||||
|
||||
```mermaid
|
||||
flowchart TD
|
||||
Dying[CPU enters offline transition] --> Stop[Stop accepting ordinary submissions]
|
||||
Stop --> GP[Settle the CPU's grace-period responsibility]
|
||||
GP --> Owner[Stabilize callback queue ownership]
|
||||
Owner --> Align[Align source and destination to one grace-period state]
|
||||
Align --> Merge[Merge the four corresponding segments]
|
||||
Merge --> Wake[Notify the callback executor]
|
||||
Wake --> Dead[Complete CPU offline transition]
|
||||
```
|
||||
|
||||
Migration follows these principles:
|
||||
|
||||
- `done`, `wait`, `next-ready`, and `next` are merged with their corresponding segments; a pending callback never becomes executable merely because it moved;
|
||||
- source and destination queues are aligned against the same grace-period state;
|
||||
- queue migration and barrier scanning share one callback-ownership synchronization domain;
|
||||
- a callback already executing does not migrate, and its source queue continues to record its completion;
|
||||
- queue storage outlives the CPU's online lifetime, so completion bookkeeping remains safe after the CPU goes offline;
|
||||
- the migration destination comes from RCU's participating CPU set, not from a broader online state with different semantics or update timing.
|
||||
|
||||
CPU lifecycle state is the sole source of truth for queue ownership. The callback subsystem does not maintain a second online/offline state that could diverge during a concurrent offline transition.
|
||||
|
||||
## 9. Concurrency Model
|
||||
|
||||
Segmented callback queues use three synchronization domains:
|
||||
|
||||
- **Grace-period state** protects grace-period start, completion, and the participating CPU set.
|
||||
- **Callback ownership** coordinates barrier-wide snapshots with CPU queue migration.
|
||||
- **Local queues** protect each CPU's segmented lists and execution state.
|
||||
|
||||
A control path that enters multiple domains follows one direction: stabilize grace-period state and callback ownership before accessing local queues in a deterministic order. Ordinary submission accesses only one local queue and never reverses into a global domain.
|
||||
|
||||
```mermaid
|
||||
flowchart LR
|
||||
GP[Grace-period state] --> Ownership[Callback ownership]
|
||||
Ownership --> Q0[Lower-numbered CPU queue]
|
||||
Q0 --> Q1[Higher-numbered CPU queue]
|
||||
|
||||
Submit[Ordinary submission] --> Local[One local queue]
|
||||
```
|
||||
|
||||
This ordering serves both correctness and maintainability. A new control path that follows the same hierarchy cannot create a reverse lock dependency with an existing path.
|
||||
|
||||
Work publication also follows the rule “publish a persistent condition before sending a wakeup.” Wakeups may be coalesced or race with the executor's sleep check. The executor therefore rechecks queue and grace-period state before sleeping to close the lost-wakeup window. Waiting segments blocked by an active grace period are not currently runnable work and must not make the executor spin.
|
||||
|
||||
## 10. Correctness Boundaries
|
||||
|
||||
### 10.1 Duplicate Submission
|
||||
|
||||
The same callback node cannot be submitted again while it is queued. The executor clears the queued state before invoking the callback, allowing a callback to resubmit its own node while it runs. The new submission receives a new queue position and grace-period classification.
|
||||
|
||||
### 10.2 Ordering Guarantees
|
||||
|
||||
FIFO order is retained within each local segment, and segment advancement preserves that order. Callbacks on different CPUs have no global submission or completion order. A caller that requires a global completion boundary uses an RCU barrier rather than relying on incidental execution order.
|
||||
|
||||
### 10.3 Memory Visibility
|
||||
|
||||
Queue synchronization publishes callback contents and list relationships. The grace-period protocol orders the end of read-side critical sections before callback eligibility. Completion of barrier markers ensures that a barrier returns after the callbacks it covers have returned.
|
||||
|
||||
Atomic state used to reduce scans or wakeups is only a hint. It cannot by itself prove callback lifetime, grace-period completion, or barrier completion.
|
||||
|
||||
### 10.4 Failure and Degradation
|
||||
|
||||
Callback submission does not require dynamic allocation, so memory pressure cannot prevent a prepared callback from entering a queue. A large backlog increases queueing latency, but bounded batches and round-robin selection preserve scheduling opportunities. CPU offline and executor startup paths follow the same state machine and must not bypass normal grace-period rules by marking pending callbacks complete early.
|
||||
|
||||
## 11. Observability
|
||||
|
||||
Debugging facilities can expose the lengths of all four segments and whether a callback is currently executing for each CPU, together with an aggregate view. Such a snapshot answers two questions:
|
||||
|
||||
- which CPUs hold most of the callback backlog;
|
||||
- whether the backlog is waiting for a grace period, unclassified, or already executable.
|
||||
|
||||
Debug snapshots do not participate in correctness and need not be globally linearizable. Observing one local queue at a time avoids introducing a system-wide pause solely for diagnostics. A single open operation should present stable text so that partial reads do not combine data from different points in time.
|
||||
|
||||
## 12. Design Trade-offs
|
||||
|
||||
The current design deliberately retains the following boundaries:
|
||||
|
||||
- per-CPU queues reduce submission contention, while one grace-period coordinator remains authoritative;
|
||||
- one logical callback execution owner avoids parallel callback semantics;
|
||||
- fixed bounded batches and round-robin fairness are used without an additional time-budget scheduler;
|
||||
- multilevel RCU nodes, callback offloading, lazy callbacks, and multiple executors are not introduced;
|
||||
- no global callback sequence is maintained; tail-marker barriers express cross-queue completion boundaries.
|
||||
|
||||
These boundaries localize callback submission costs while restricting cross-CPU coordination to lower-frequency grace-period transitions, barriers, and CPU lifecycle events. If measurements later show that the single coordinator or executor is a bottleneck, the corresponding layer can be extended based on evidence rather than adopting the full complexity of Linux Tree RCU in advance.
|
||||
@@ -40,6 +40,9 @@ impl KernFSCallback for RcuDirCallBack {
|
||||
#[derive(Debug)]
|
||||
struct RcuSelftestCallBack;
|
||||
|
||||
#[derive(Debug)]
|
||||
struct RcuCallbacksCallBack;
|
||||
|
||||
impl KernFSCallback for RcuSelftestCallBack {
|
||||
fn open(&self, mut data: KernCallbackData) -> Result<(), SystemError> {
|
||||
let report = crate::rcu::run_debug_selftests();
|
||||
@@ -82,6 +85,48 @@ impl KernFSCallback for RcuSelftestCallBack {
|
||||
}
|
||||
}
|
||||
|
||||
impl KernFSCallback for RcuCallbacksCallBack {
|
||||
fn open(&self, mut data: KernCallbackData) -> Result<(), SystemError> {
|
||||
data.file_private_data_mut()
|
||||
.replace(KernFilePrivateData::DebugTextSnapshot(
|
||||
crate::rcu::callback_queue_debug_report(),
|
||||
));
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn read(
|
||||
&self,
|
||||
data: KernCallbackData,
|
||||
buf: &mut [u8],
|
||||
offset: usize,
|
||||
) -> Result<usize, SystemError> {
|
||||
let report = match data.file_private_data() {
|
||||
Some(KernFilePrivateData::DebugTextSnapshot(report)) => report,
|
||||
_ => return Err(SystemError::EINVAL),
|
||||
};
|
||||
let bytes = report.as_bytes();
|
||||
if offset >= bytes.len() {
|
||||
return Ok(0);
|
||||
}
|
||||
let len = buf.len().min(bytes.len() - offset);
|
||||
buf[..len].copy_from_slice(&bytes[offset..offset + len]);
|
||||
Ok(len)
|
||||
}
|
||||
|
||||
fn write(
|
||||
&self,
|
||||
_data: KernCallbackData,
|
||||
_buf: &[u8],
|
||||
_offset: usize,
|
||||
) -> Result<usize, SystemError> {
|
||||
Err(SystemError::EPERM)
|
||||
}
|
||||
|
||||
fn poll(&self, _data: KernCallbackData) -> Result<PollStatus, SystemError> {
|
||||
Ok(PollStatus::READ)
|
||||
}
|
||||
}
|
||||
|
||||
pub fn init_debugfs_rcu() -> Result<(), SystemError> {
|
||||
let debugfs = debugfs_kobj();
|
||||
let root_dir = debugfs.inode().ok_or(SystemError::ENOENT)?;
|
||||
@@ -99,6 +144,13 @@ pub fn init_debugfs_rcu() -> Result<(), SystemError> {
|
||||
None,
|
||||
Some(&RcuSelftestCallBack),
|
||||
)?;
|
||||
rcu_root.add_file(
|
||||
"callbacks".to_string(),
|
||||
InodeMode::S_IRUGO,
|
||||
Some(32 * 1024),
|
||||
None,
|
||||
Some(&RcuCallbacksCallBack),
|
||||
)?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
+319
-62
@@ -9,8 +9,6 @@ use super::{gp::RcuSequence, RcuRawCallback};
|
||||
struct RcuHeadNode {
|
||||
next: Option<NonNull<RcuHead>>,
|
||||
func: Option<RcuRawCallback>,
|
||||
target_gp: Option<RcuSequence>,
|
||||
callback_seq: Option<RcuSequence>,
|
||||
}
|
||||
|
||||
impl RcuHeadNode {
|
||||
@@ -18,8 +16,6 @@ impl RcuHeadNode {
|
||||
Self {
|
||||
next: None,
|
||||
func: None,
|
||||
target_gp: None,
|
||||
callback_seq: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -42,7 +38,6 @@ impl RcuHead {
|
||||
}
|
||||
}
|
||||
|
||||
/// Claims this head for one admission without modifying its queue node.
|
||||
pub(super) fn try_claim(&self) -> bool {
|
||||
self.queued
|
||||
.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
|
||||
@@ -62,31 +57,25 @@ impl core::fmt::Debug for RcuHead {
|
||||
}
|
||||
}
|
||||
|
||||
// SAFETY: `queued` is atomic. All access to `node` is encapsulated by
|
||||
// `RcuCallbackList` and serialized by `RcuState::inner`. The unsafe admission
|
||||
// contract keeps a queued head initialized at a stable address.
|
||||
// SAFETY: `queued` is atomic. All access to `node` is serialized by a
|
||||
// per-CPU callback-state lock, and admission keeps a queued head stable.
|
||||
unsafe impl Send for RcuHead {}
|
||||
// SAFETY: Shared references only expose atomic duplicate admission. The
|
||||
// non-atomic node remains protected by the global RCU state lock.
|
||||
unsafe impl Sync for RcuHead {}
|
||||
|
||||
pub(super) struct ReadyRcuCallback {
|
||||
pub(super) head: NonNull<RcuHead>,
|
||||
pub(super) func: RcuRawCallback,
|
||||
pub(super) seq: RcuSequence,
|
||||
}
|
||||
|
||||
/// The one global intrusive callback FIFO.
|
||||
///
|
||||
/// The containing `RcuState::inner` lock must be held for every operation.
|
||||
pub(super) struct RcuCallbackList {
|
||||
/// One intrusive FIFO used as a segment of a per-CPU callback queue.
|
||||
struct RcuCallbackList {
|
||||
head: Option<NonNull<RcuHead>>,
|
||||
tail: Option<NonNull<RcuHead>>,
|
||||
len: usize,
|
||||
}
|
||||
|
||||
impl RcuCallbackList {
|
||||
pub(super) const fn new() -> Self {
|
||||
const fn new() -> Self {
|
||||
Self {
|
||||
head: None,
|
||||
tail: None,
|
||||
@@ -94,42 +83,23 @@ impl RcuCallbackList {
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) fn is_empty(&self) -> bool {
|
||||
fn is_empty(&self) -> bool {
|
||||
self.head.is_none()
|
||||
}
|
||||
|
||||
pub(super) fn len(&self) -> usize {
|
||||
fn len(&self) -> usize {
|
||||
self.len
|
||||
}
|
||||
|
||||
pub(super) fn front_target(&self) -> Option<RcuSequence> {
|
||||
let head = self.head?;
|
||||
// SAFETY: Queue operations are serialized by `RcuState::inner`, and
|
||||
// admission keeps every linked head alive at a stable address.
|
||||
unsafe { (*head.as_ref().node.get()).target_gp }
|
||||
}
|
||||
|
||||
pub(super) fn push(
|
||||
&mut self,
|
||||
head: NonNull<RcuHead>,
|
||||
func: RcuRawCallback,
|
||||
target_gp: RcuSequence,
|
||||
callback_seq: RcuSequence,
|
||||
) {
|
||||
// SAFETY: The caller holds `RcuState::inner`, has exclusively claimed
|
||||
// `head`, and guarantees its address and lifetime through callback
|
||||
// start. A queued node is mutated only through this list.
|
||||
fn push(&mut self, head: NonNull<RcuHead>, func: RcuRawCallback) {
|
||||
// SAFETY: The containing state lock is held, `head` is exclusively
|
||||
// claimed, and its address remains stable until callback start.
|
||||
unsafe {
|
||||
let node = &mut *head.as_ref().node.get();
|
||||
debug_assert!(head.as_ref().queued.load(Ordering::Acquire));
|
||||
debug_assert!(node.next.is_none());
|
||||
debug_assert!(node.func.is_none());
|
||||
debug_assert!(node.target_gp.is_none());
|
||||
debug_assert!(node.callback_seq.is_none());
|
||||
|
||||
node.func = Some(func);
|
||||
node.target_gp = Some(target_gp);
|
||||
node.callback_seq = Some(callback_seq);
|
||||
|
||||
if let Some(tail) = self.tail {
|
||||
let tail_node = &mut *tail.as_ref().node.get();
|
||||
@@ -149,42 +119,54 @@ impl RcuCallbackList {
|
||||
self.assert_invariants();
|
||||
}
|
||||
|
||||
pub(super) fn pop_ready(&mut self, completed_gp: RcuSequence) -> Option<ReadyRcuCallback> {
|
||||
fn pop(&mut self) -> Option<ReadyRcuCallback> {
|
||||
let head = self.head?;
|
||||
// SAFETY: The global RCU state lock serializes the list. The head is
|
||||
// still linked, so the admission lifetime contract keeps it valid.
|
||||
// SAFETY: The containing state lock serializes list access and the
|
||||
// admission contract keeps every linked head valid.
|
||||
let node = unsafe { &mut *head.as_ref().node.get() };
|
||||
let target_gp = node
|
||||
.target_gp
|
||||
.expect("queued RCU callback has no target grace period");
|
||||
if !completed_gp.has_reached(target_gp) {
|
||||
return None;
|
||||
}
|
||||
|
||||
self.head = node.next.take();
|
||||
if self.head.is_none() {
|
||||
self.tail = None;
|
||||
}
|
||||
self.len -= 1;
|
||||
|
||||
let func = node
|
||||
.func
|
||||
.take()
|
||||
.expect("queued RCU callback has no function");
|
||||
let seq = node
|
||||
.callback_seq
|
||||
.take()
|
||||
.expect("queued RCU callback has no admission sequence");
|
||||
node.target_gp = None;
|
||||
|
||||
// Publish complete detachment before the callback receives ownership.
|
||||
// After this store the drainer must not dereference `head` again.
|
||||
// Publish complete detachment before callback ownership starts.
|
||||
unsafe { head.as_ref() }
|
||||
.queued
|
||||
.store(false, Ordering::Release);
|
||||
self.assert_invariants();
|
||||
Some(ReadyRcuCallback { head, func })
|
||||
}
|
||||
|
||||
Some(ReadyRcuCallback { head, func, seq })
|
||||
/// Appends `source` as one whole FIFO and leaves it empty.
|
||||
fn append(&mut self, source: &mut Self) {
|
||||
if source.is_empty() {
|
||||
return;
|
||||
}
|
||||
|
||||
// SAFETY: Both lists are exclusively protected, and their nodes are
|
||||
// disjoint because a head can be queued only once.
|
||||
unsafe {
|
||||
if let Some(tail) = self.tail {
|
||||
(*tail.as_ref().node.get()).next = source.head;
|
||||
} else {
|
||||
self.head = source.head;
|
||||
}
|
||||
}
|
||||
self.tail = source.tail;
|
||||
self.len = self
|
||||
.len
|
||||
.checked_add(source.len)
|
||||
.expect("RCU callback count overflow");
|
||||
source.head = None;
|
||||
source.tail = None;
|
||||
source.len = 0;
|
||||
self.assert_invariants();
|
||||
source.assert_invariants();
|
||||
}
|
||||
|
||||
fn assert_invariants(&self) {
|
||||
@@ -193,7 +175,282 @@ impl RcuCallbackList {
|
||||
}
|
||||
}
|
||||
|
||||
// SAFETY: The list's raw pointers are non-owning tokens transferred between
|
||||
// CPUs only as part of `RcuStateInner`. Every dereference is serialized by its
|
||||
// global spin lock, and admission guarantees stable pointee addresses.
|
||||
// SAFETY: Raw links are dereferenced only while the containing state lock is
|
||||
// held, and admission guarantees stable pointee addresses.
|
||||
unsafe impl Send for RcuCallbackList {}
|
||||
|
||||
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
|
||||
pub(crate) struct RcuCallbackQueueDepth {
|
||||
pub(crate) done: usize,
|
||||
pub(crate) wait: usize,
|
||||
pub(crate) next_ready: usize,
|
||||
pub(crate) next: usize,
|
||||
pub(crate) executing: bool,
|
||||
}
|
||||
|
||||
impl RcuCallbackQueueDepth {
|
||||
pub(crate) fn total(self) -> usize {
|
||||
self.done + self.wait + self.next_ready + self.next
|
||||
}
|
||||
|
||||
pub(super) fn add_assign(&mut self, other: Self) {
|
||||
self.done += other.done;
|
||||
self.wait += other.wait;
|
||||
self.next_ready += other.next_ready;
|
||||
self.next += other.next;
|
||||
self.executing |= other.executing;
|
||||
}
|
||||
}
|
||||
|
||||
/// Four explicit grace-period states for one CPU's callbacks.
|
||||
pub(super) struct RcuSegmentedCallbacks {
|
||||
done: RcuCallbackList,
|
||||
wait: RcuCallbackList,
|
||||
next_ready: RcuCallbackList,
|
||||
next: RcuCallbackList,
|
||||
wait_target: Option<RcuSequence>,
|
||||
next_ready_target: Option<RcuSequence>,
|
||||
}
|
||||
|
||||
impl RcuSegmentedCallbacks {
|
||||
pub(super) const fn new() -> Self {
|
||||
Self {
|
||||
done: RcuCallbackList::new(),
|
||||
wait: RcuCallbackList::new(),
|
||||
next_ready: RcuCallbackList::new(),
|
||||
next: RcuCallbackList::new(),
|
||||
wait_target: None,
|
||||
next_ready_target: None,
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) fn enqueue(&mut self, head: NonNull<RcuHead>, func: RcuRawCallback) {
|
||||
self.next.push(head, func);
|
||||
self.assert_invariants();
|
||||
}
|
||||
|
||||
pub(super) fn has_ready(&self) -> bool {
|
||||
!self.done.is_empty()
|
||||
}
|
||||
|
||||
pub(super) fn has_unclassified(&self) -> bool {
|
||||
!self.next.is_empty()
|
||||
}
|
||||
|
||||
pub(super) fn is_empty(&self) -> bool {
|
||||
self.depth().total() == 0
|
||||
}
|
||||
|
||||
pub(super) fn depth(&self) -> RcuCallbackQueueDepth {
|
||||
RcuCallbackQueueDepth {
|
||||
done: self.done.len(),
|
||||
wait: self.wait.len(),
|
||||
next_ready: self.next_ready.len(),
|
||||
next: self.next.len(),
|
||||
executing: false,
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) fn classify_next(&mut self, target: RcuSequence, gp_active: bool) -> bool {
|
||||
if self.next.is_empty() {
|
||||
return false;
|
||||
}
|
||||
if gp_active {
|
||||
Self::set_or_check_target(&mut self.next_ready_target, target);
|
||||
self.next_ready.append(&mut self.next);
|
||||
} else {
|
||||
Self::set_or_check_target(&mut self.wait_target, target);
|
||||
self.wait.append(&mut self.next);
|
||||
}
|
||||
self.assert_invariants();
|
||||
true
|
||||
}
|
||||
|
||||
/// Absorbs callbacks known to precede an imminent GP start.
|
||||
pub(super) fn prepare_gp_start(&mut self, seq: RcuSequence) {
|
||||
// `next_ready` predates `next`: it was classified while the previous
|
||||
// GP was active, whereas `next` contains later, not-yet-classified
|
||||
// admissions. Preserve that global FIFO order when both become
|
||||
// covered by this imminent GP.
|
||||
if !self.next_ready.is_empty() {
|
||||
debug_assert_eq!(self.next_ready_target, Some(seq));
|
||||
Self::set_or_check_target(&mut self.wait_target, seq);
|
||||
self.wait.append(&mut self.next_ready);
|
||||
self.next_ready_target = None;
|
||||
}
|
||||
if !self.next.is_empty() {
|
||||
Self::set_or_check_target(&mut self.wait_target, seq);
|
||||
self.wait.append(&mut self.next);
|
||||
}
|
||||
self.assert_invariants();
|
||||
}
|
||||
|
||||
pub(super) fn complete_gp(&mut self, completed: RcuSequence) -> bool {
|
||||
if self.wait.is_empty() {
|
||||
return false;
|
||||
}
|
||||
let target = self
|
||||
.wait_target
|
||||
.expect("non-empty RCU wait segment has no GP");
|
||||
if !completed.has_reached(target) {
|
||||
return false;
|
||||
}
|
||||
self.done.append(&mut self.wait);
|
||||
self.wait_target = None;
|
||||
self.assert_invariants();
|
||||
true
|
||||
}
|
||||
|
||||
pub(super) fn pop_ready(&mut self) -> Option<ReadyRcuCallback> {
|
||||
let callback = self.done.pop();
|
||||
self.assert_invariants();
|
||||
callback
|
||||
}
|
||||
|
||||
/// Entrains a barrier marker after the last currently queued callback.
|
||||
pub(super) fn entrain(&mut self, head: NonNull<RcuHead>, func: RcuRawCallback) -> bool {
|
||||
if !self.next.is_empty() {
|
||||
self.next.push(head, func);
|
||||
} else if !self.next_ready.is_empty() {
|
||||
self.next_ready.push(head, func);
|
||||
} else if !self.wait.is_empty() {
|
||||
self.wait.push(head, func);
|
||||
} else if !self.done.is_empty() {
|
||||
self.done.push(head, func);
|
||||
} else {
|
||||
return false;
|
||||
}
|
||||
self.assert_invariants();
|
||||
true
|
||||
}
|
||||
|
||||
pub(super) fn push_done(&mut self, head: NonNull<RcuHead>, func: RcuRawCallback) {
|
||||
self.done.push(head, func);
|
||||
self.assert_invariants();
|
||||
}
|
||||
|
||||
pub(super) fn merge_from(&mut self, source: &mut Self) {
|
||||
Self::merge_target_segment(
|
||||
&mut self.wait,
|
||||
&mut self.wait_target,
|
||||
&mut source.wait,
|
||||
&mut source.wait_target,
|
||||
);
|
||||
Self::merge_target_segment(
|
||||
&mut self.next_ready,
|
||||
&mut self.next_ready_target,
|
||||
&mut source.next_ready,
|
||||
&mut source.next_ready_target,
|
||||
);
|
||||
self.done.append(&mut source.done);
|
||||
self.next.append(&mut source.next);
|
||||
self.assert_invariants();
|
||||
source.assert_invariants();
|
||||
}
|
||||
|
||||
fn merge_target_segment(
|
||||
destination: &mut RcuCallbackList,
|
||||
destination_target: &mut Option<RcuSequence>,
|
||||
source: &mut RcuCallbackList,
|
||||
source_target: &mut Option<RcuSequence>,
|
||||
) {
|
||||
if source.is_empty() {
|
||||
return;
|
||||
}
|
||||
let target = source_target.expect("non-empty RCU segment has no GP target");
|
||||
Self::set_or_check_target(destination_target, target);
|
||||
destination.append(source);
|
||||
*source_target = None;
|
||||
}
|
||||
|
||||
fn set_or_check_target(slot: &mut Option<RcuSequence>, target: RcuSequence) {
|
||||
if let Some(current) = *slot {
|
||||
assert_eq!(current, target, "merged incompatible RCU callback segments");
|
||||
} else {
|
||||
*slot = Some(target);
|
||||
}
|
||||
}
|
||||
|
||||
fn assert_invariants(&self) {
|
||||
debug_assert_eq!(self.wait.is_empty(), self.wait_target.is_none());
|
||||
debug_assert_eq!(self.next_ready.is_empty(), self.next_ready_target.is_none());
|
||||
self.done.assert_invariants();
|
||||
self.wait.assert_invariants();
|
||||
self.next_ready.assert_invariants();
|
||||
self.next.assert_invariants();
|
||||
}
|
||||
}
|
||||
|
||||
// SAFETY: The segmented queue is accessed only through its containing
|
||||
// per-CPU callback-state lock.
|
||||
unsafe impl Send for RcuSegmentedCallbacks {}
|
||||
|
||||
pub(super) fn run_segmented_callback_selftests() -> Result<(), &'static str> {
|
||||
unsafe fn noop(_head: NonNull<RcuHead>) {}
|
||||
|
||||
let heads = [const { RcuHead::new() }; 5];
|
||||
for head in &heads {
|
||||
if !head.try_claim() {
|
||||
return Err("fresh segmented callback head could not be claimed");
|
||||
}
|
||||
}
|
||||
|
||||
let gp1 = RcuSequence::from_raw(1);
|
||||
let gp2 = RcuSequence::from_raw(2);
|
||||
let gp3 = RcuSequence::from_raw(3);
|
||||
let gp4 = RcuSequence::from_raw(4);
|
||||
let mut source = RcuSegmentedCallbacks::new();
|
||||
let mut destination = RcuSegmentedCallbacks::new();
|
||||
|
||||
source.enqueue(NonNull::from(&heads[0]), noop);
|
||||
source.classify_next(gp1, false);
|
||||
source.complete_gp(gp1);
|
||||
source.enqueue(NonNull::from(&heads[1]), noop);
|
||||
source.classify_next(gp2, false);
|
||||
source.enqueue(NonNull::from(&heads[2]), noop);
|
||||
source.classify_next(gp3, true);
|
||||
source.enqueue(NonNull::from(&heads[3]), noop);
|
||||
|
||||
if source.depth()
|
||||
!= (RcuCallbackQueueDepth {
|
||||
done: 1,
|
||||
wait: 1,
|
||||
next_ready: 1,
|
||||
next: 1,
|
||||
executing: false,
|
||||
})
|
||||
{
|
||||
return Err("could not construct all four RCU callback segments");
|
||||
}
|
||||
|
||||
destination.merge_from(&mut source);
|
||||
if !source.is_empty() || destination.depth().total() != 4 {
|
||||
return Err("RCU callback migration did not move every segment");
|
||||
}
|
||||
if !destination.entrain(NonNull::from(&heads[4]), noop) {
|
||||
return Err("RCU barrier marker could not entrain behind pending callbacks");
|
||||
}
|
||||
|
||||
if destination.pop_ready().map(|callback| callback.head) != Some(NonNull::from(&heads[0])) {
|
||||
return Err("RCU done segment lost FIFO order during migration");
|
||||
}
|
||||
destination.complete_gp(gp2);
|
||||
if destination.pop_ready().map(|callback| callback.head) != Some(NonNull::from(&heads[1])) {
|
||||
return Err("RCU wait segment did not advance as a whole");
|
||||
}
|
||||
destination.prepare_gp_start(gp3);
|
||||
destination.complete_gp(gp3);
|
||||
if destination.pop_ready().map(|callback| callback.head) != Some(NonNull::from(&heads[2])) {
|
||||
return Err("RCU next-ready segment did not advance as a whole");
|
||||
}
|
||||
destination.classify_next(gp4, false);
|
||||
destination.complete_gp(gp4);
|
||||
if destination.pop_ready().map(|callback| callback.head) != Some(NonNull::from(&heads[3]))
|
||||
|| destination.pop_ready().map(|callback| callback.head) != Some(NonNull::from(&heads[4]))
|
||||
|| !destination.is_empty()
|
||||
{
|
||||
return Err("RCU barrier marker did not remain behind its queue prefix");
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
+4
-103
@@ -19,13 +19,15 @@ pub(super) struct RcuSequence(u64);
|
||||
|
||||
impl RcuSequence {
|
||||
const ZERO: Self = Self(0);
|
||||
const FIRST_CALLBACK: Self = Self(1);
|
||||
|
||||
#[inline]
|
||||
pub(super) const fn raw(self) -> u64 {
|
||||
self.0
|
||||
}
|
||||
|
||||
pub(super) const fn from_raw(value: u64) -> Self {
|
||||
Self(value)
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub(super) const fn next(self) -> Self {
|
||||
Self(self.0.wrapping_add(1))
|
||||
@@ -207,84 +209,6 @@ impl GracePeriodState {
|
||||
}
|
||||
}
|
||||
|
||||
/// Callback admission/completion tickets plus unique drainer ownership.
|
||||
///
|
||||
/// Callback storage remains in `RcuStateInner`; this type only centralizes the
|
||||
/// ordering facts required by `rcu_barrier()`.
|
||||
pub(super) struct CallbackTracker {
|
||||
next: RcuSequence,
|
||||
next_completion: RcuSequence,
|
||||
last_admitted: Option<RcuSequence>,
|
||||
completed: Option<RcuSequence>,
|
||||
draining: bool,
|
||||
}
|
||||
|
||||
impl CallbackTracker {
|
||||
pub(super) fn new() -> Self {
|
||||
Self::with_next(RcuSequence::FIRST_CALLBACK)
|
||||
}
|
||||
|
||||
fn with_next(next: RcuSequence) -> Self {
|
||||
Self {
|
||||
next,
|
||||
next_completion: next,
|
||||
last_admitted: None,
|
||||
completed: None,
|
||||
draining: false,
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) fn admit(&mut self) -> RcuSequence {
|
||||
let seq = self.next;
|
||||
self.next = self.next.next();
|
||||
self.last_admitted = Some(seq);
|
||||
seq
|
||||
}
|
||||
|
||||
pub(super) fn barrier_target(&self) -> Option<RcuSequence> {
|
||||
self.last_admitted
|
||||
}
|
||||
|
||||
pub(super) fn has_completed(&self, target: RcuSequence) -> bool {
|
||||
self.completed
|
||||
.is_some_and(|completed| completed.has_reached(target))
|
||||
}
|
||||
|
||||
pub(super) fn complete(&mut self, seq: RcuSequence) {
|
||||
debug_assert!(
|
||||
self.last_admitted.is_some_and(|last| last.has_reached(seq)),
|
||||
"completed a callback that was not admitted"
|
||||
);
|
||||
debug_assert_eq!(
|
||||
seq, self.next_completion,
|
||||
"RCU callbacks must complete in admission order"
|
||||
);
|
||||
self.completed = Some(seq);
|
||||
self.next_completion = self.next_completion.next();
|
||||
}
|
||||
|
||||
pub(super) fn try_claim_drainer(&mut self) -> bool {
|
||||
if self.draining {
|
||||
return false;
|
||||
}
|
||||
self.draining = true;
|
||||
true
|
||||
}
|
||||
|
||||
pub(super) fn release_drainer(&mut self) {
|
||||
debug_assert!(self.draining, "released an unclaimed RCU callback drainer");
|
||||
self.draining = false;
|
||||
}
|
||||
|
||||
pub(super) fn drainer_available(&self) -> bool {
|
||||
!self.draining
|
||||
}
|
||||
|
||||
pub(super) fn completed_raw(&self) -> u64 {
|
||||
self.completed.map_or(0, RcuSequence::raw)
|
||||
}
|
||||
}
|
||||
|
||||
fn one_cpu_mask(cpu: u32) -> CpuMask {
|
||||
CpuMask::from_cpu(ProcessorId::new(cpu))
|
||||
}
|
||||
@@ -372,28 +296,5 @@ pub(super) fn run_state_machine_selftests() -> Result<(), &'static str> {
|
||||
return Err("RCU GP sequence did not wrap from u64::MAX to zero");
|
||||
}
|
||||
|
||||
let mut callbacks = CallbackTracker::with_next(RcuSequence(u64::MAX));
|
||||
let max_callback = callbacks.admit();
|
||||
if callbacks.barrier_target() != Some(max_callback) || callbacks.has_completed(max_callback) {
|
||||
return Err("RCU callback tracker corrupted its pre-wrap barrier target");
|
||||
}
|
||||
callbacks.complete(max_callback);
|
||||
let zero_callback = callbacks.admit();
|
||||
if zero_callback.raw() != 0 || callbacks.barrier_target() != Some(zero_callback) {
|
||||
return Err("RCU callback tracker treated wrapped zero as an empty target");
|
||||
}
|
||||
callbacks.complete(zero_callback);
|
||||
if !callbacks.has_completed(max_callback) || !callbacks.has_completed(zero_callback) {
|
||||
return Err("RCU callback completion comparison failed across sequence wrap");
|
||||
}
|
||||
if !callbacks.try_claim_drainer() || callbacks.try_claim_drainer() {
|
||||
return Err("RCU callback tracker allowed two simultaneous drainers");
|
||||
}
|
||||
callbacks.release_drainer();
|
||||
if !callbacks.try_claim_drainer() {
|
||||
return Err("RCU callback tracker did not release drainer ownership");
|
||||
}
|
||||
callbacks.release_drainer();
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
+494
-119
@@ -14,34 +14,47 @@
|
||||
//! the happens-before chain from callback admission to callback invocation or
|
||||
//! `synchronize_rcu()` return.
|
||||
|
||||
use alloc::{boxed::Box, rc::Rc, string::ToString, sync::Arc};
|
||||
use alloc::{
|
||||
boxed::Box,
|
||||
rc::Rc,
|
||||
string::{String, ToString},
|
||||
sync::Arc,
|
||||
vec::Vec,
|
||||
};
|
||||
use core::{
|
||||
fmt::Write,
|
||||
marker::PhantomData,
|
||||
ptr::{self, NonNull},
|
||||
sync::atomic::{fence, AtomicBool, AtomicPtr, Ordering},
|
||||
sync::atomic::{fence, AtomicBool, AtomicPtr, AtomicUsize, Ordering},
|
||||
};
|
||||
|
||||
use log::warn;
|
||||
|
||||
use crate::{
|
||||
libs::{cpumask::CpuMask, spinlock::SpinLock, wait_queue::WaitQueue},
|
||||
libs::{cpumask::CpuMask, mutex::Mutex, spinlock::SpinLock, wait_queue::WaitQueue},
|
||||
mm::percpu::PerCpu,
|
||||
process::{kthread::KernelThreadClosure, kthread::KernelThreadMechanism, ProcessManager},
|
||||
sched::{sched_yield, SchedPolicy},
|
||||
smp::{core::smp_get_processor_id, cpu::ProcessorId},
|
||||
process::{
|
||||
kthread::KernelThreadClosure, kthread::KernelThreadMechanism, preempt::PreemptGuard,
|
||||
ProcessManager,
|
||||
},
|
||||
sched::{cond_resched, sched_yield, SchedPolicy},
|
||||
smp::{
|
||||
core::smp_get_processor_id,
|
||||
cpu::{smp_cpu_manager, smp_cpu_manager_initialized, ProcessorId},
|
||||
},
|
||||
};
|
||||
|
||||
mod callback;
|
||||
mod context;
|
||||
mod gp;
|
||||
mod selftest;
|
||||
use callback::RcuCallbackList;
|
||||
pub use callback::RcuHead;
|
||||
use callback::{RcuCallbackQueueDepth, RcuSegmentedCallbacks};
|
||||
use context::{
|
||||
BaseContext, ContextTransition, ContextTransitionError, IdleTransition, IrqDisposition,
|
||||
IrqEntry, RcuContextTracker,
|
||||
};
|
||||
use gp::{CallbackTracker, GracePeriodState};
|
||||
use gp::{GracePeriodState, RcuSequence};
|
||||
pub use selftest::run_debug_selftests;
|
||||
|
||||
pub(crate) type RcuRawCallback = unsafe fn(NonNull<RcuHead>);
|
||||
@@ -349,8 +362,6 @@ struct RcuStateInner {
|
||||
/// CPUs that have executed the RCU starting hook and are eligible for
|
||||
/// future GP snapshots. The active GP keeps its own immutable snapshot.
|
||||
participating_cpus: CpuMask,
|
||||
callbacks: CallbackTracker,
|
||||
callback_queue: RcuCallbackList,
|
||||
}
|
||||
|
||||
impl RcuStateInner {
|
||||
@@ -358,28 +369,54 @@ impl RcuStateInner {
|
||||
Self {
|
||||
gp: GracePeriodState::new(),
|
||||
participating_cpus: CpuMask::new(),
|
||||
callbacks: CallbackTracker::new(),
|
||||
callback_queue: RcuCallbackList::new(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
struct RcuCpuCallbackState {
|
||||
segments: RcuSegmentedCallbacks,
|
||||
executing: bool,
|
||||
}
|
||||
|
||||
impl RcuCpuCallbackState {
|
||||
const fn new() -> Self {
|
||||
Self {
|
||||
segments: RcuSegmentedCallbacks::new(),
|
||||
executing: false,
|
||||
}
|
||||
}
|
||||
|
||||
fn has_ready_work(&self) -> bool {
|
||||
self.callback_queue
|
||||
.front_target()
|
||||
.is_some_and(|target| self.gp.has_completed(target))
|
||||
}
|
||||
|
||||
fn has_drainable_work(&self) -> bool {
|
||||
self.has_ready_work() && self.callbacks.drainer_available()
|
||||
}
|
||||
|
||||
fn has_worker_work(&self) -> bool {
|
||||
self.has_drainable_work()
|
||||
|| self.gp.ready_to_complete()
|
||||
|| (!self.gp.is_active() && self.gp.has_request())
|
||||
fn depth(&self) -> RcuCallbackQueueDepth {
|
||||
let mut depth = self.segments.depth();
|
||||
depth.executing = self.executing;
|
||||
depth
|
||||
}
|
||||
}
|
||||
|
||||
#[repr(align(64))]
|
||||
struct RcuCpuCallbacks {
|
||||
state: SpinLock<RcuCpuCallbackState>,
|
||||
needs_scan: AtomicBool,
|
||||
barrier_head: RcuHead,
|
||||
}
|
||||
|
||||
impl RcuCpuCallbacks {
|
||||
const fn new() -> Self {
|
||||
Self {
|
||||
state: SpinLock::new(RcuCpuCallbackState::new()),
|
||||
needs_scan: AtomicBool::new(false),
|
||||
barrier_head: RcuHead::new(),
|
||||
}
|
||||
}
|
||||
|
||||
fn publish_work(&self) -> bool {
|
||||
!self.needs_scan.swap(true, Ordering::AcqRel)
|
||||
}
|
||||
}
|
||||
|
||||
const RCU_CALLBACK_BATCH_LIMIT: usize = 64;
|
||||
const RCU_CALLBACK_CPU_QUANTUM: usize = 8;
|
||||
|
||||
struct RcuState {
|
||||
initialized: AtomicBool,
|
||||
worker_starting: AtomicBool,
|
||||
@@ -387,13 +424,28 @@ struct RcuState {
|
||||
worker_should_stop: AtomicBool,
|
||||
gp_active: AtomicBool,
|
||||
contexts: [RcuContextTracker; PerCpu::MAX_CPU_NUM as usize],
|
||||
cpu_callbacks: Box<[RcuCpuCallbacks]>,
|
||||
inner: SpinLock<RcuStateInner>,
|
||||
callback_ownership: SpinLock<()>,
|
||||
executor_claimed: AtomicBool,
|
||||
worker_kick_pending: AtomicBool,
|
||||
next_scan_cpu: AtomicUsize,
|
||||
callbacks_invoked: AtomicUsize,
|
||||
barrier_mutex: Mutex<()>,
|
||||
barrier_remaining: AtomicUsize,
|
||||
barrier_wait: WaitQueue,
|
||||
state_wait: WaitQueue,
|
||||
worker_wait: WaitQueue,
|
||||
}
|
||||
|
||||
impl RcuState {
|
||||
fn new() -> Self {
|
||||
// Construct the cache-line-aligned per-CPU records directly in heap
|
||||
// storage. Materializing the complete array in this function's stack
|
||||
// frame can exhaust the BSP's fixed 32-KiB boot stack.
|
||||
let mut cpu_callbacks = Vec::with_capacity(PerCpu::MAX_CPU_NUM as usize);
|
||||
cpu_callbacks.resize_with(PerCpu::MAX_CPU_NUM as usize, RcuCpuCallbacks::new);
|
||||
|
||||
Self {
|
||||
initialized: AtomicBool::new(false),
|
||||
worker_starting: AtomicBool::new(false),
|
||||
@@ -401,7 +453,16 @@ impl RcuState {
|
||||
worker_should_stop: AtomicBool::new(false),
|
||||
gp_active: AtomicBool::new(false),
|
||||
contexts: [const { RcuContextTracker::new() }; PerCpu::MAX_CPU_NUM as usize],
|
||||
cpu_callbacks: cpu_callbacks.into_boxed_slice(),
|
||||
inner: SpinLock::new(RcuStateInner::new()),
|
||||
callback_ownership: SpinLock::new(()),
|
||||
executor_claimed: AtomicBool::new(false),
|
||||
worker_kick_pending: AtomicBool::new(false),
|
||||
next_scan_cpu: AtomicUsize::new(0),
|
||||
callbacks_invoked: AtomicUsize::new(0),
|
||||
barrier_mutex: Mutex::new(()),
|
||||
barrier_remaining: AtomicUsize::new(0),
|
||||
barrier_wait: WaitQueue::default(),
|
||||
state_wait: WaitQueue::default(),
|
||||
worker_wait: WaitQueue::default(),
|
||||
}
|
||||
@@ -417,6 +478,8 @@ impl RcuState {
|
||||
participating_context_snapshot,
|
||||
credit_context_progress_locked,
|
||||
publish_gp_active,
|
||||
|completed| RCU_STATE.complete_callback_gp(completed),
|
||||
|starting| RCU_STATE.prepare_callback_gp_start(starting),
|
||||
)
|
||||
}
|
||||
|
||||
@@ -425,6 +488,8 @@ impl RcuState {
|
||||
mut waiting_snapshot: impl FnMut(&CpuMask) -> (CpuMask, [u64; PerCpu::MAX_CPU_NUM as usize]),
|
||||
mut credit_context: impl FnMut(&mut GracePeriodState),
|
||||
mut publish_active: impl FnMut(bool),
|
||||
mut complete_callbacks: impl FnMut(RcuSequence) -> bool,
|
||||
mut prepare_callbacks: impl FnMut(RcuSequence),
|
||||
) -> bool {
|
||||
let mut ready_changed = false;
|
||||
loop {
|
||||
@@ -434,8 +499,8 @@ impl RcuState {
|
||||
// Pair all real quiescent-state reports with GP completion.
|
||||
// This is a GP slow path and does not affect RCU readers.
|
||||
fence(Ordering::SeqCst);
|
||||
inner.gp.complete_ready();
|
||||
ready_changed |= inner.has_ready_work();
|
||||
let completed = inner.gp.complete_ready();
|
||||
ready_changed |= complete_callbacks(completed);
|
||||
continue;
|
||||
}
|
||||
|
||||
@@ -449,79 +514,283 @@ impl RcuState {
|
||||
|
||||
// Admission/request operations preceding this point must be
|
||||
// ordered before the fresh CPU snapshot that defines GP start.
|
||||
let starting = inner.gp.current().next();
|
||||
prepare_callbacks(starting);
|
||||
publish_active(true);
|
||||
fence(Ordering::SeqCst);
|
||||
let (waiting_cpus, context_generations) = waiting_snapshot(&inner.participating_cpus);
|
||||
inner.gp.start_requested(waiting_cpus, context_generations);
|
||||
let started = inner.gp.start_requested(waiting_cpus, context_generations);
|
||||
debug_assert_eq!(started, starting);
|
||||
}
|
||||
|
||||
publish_active(inner.gp.is_active());
|
||||
|
||||
debug_assert!(inner.gp.is_active() || !inner.gp.has_request());
|
||||
debug_assert!(
|
||||
inner.callback_queue.is_empty()
|
||||
|| inner.has_ready_work()
|
||||
|| inner.gp.is_active()
|
||||
|| inner.gp.has_request(),
|
||||
"RCU callbacks exist without ready, active, or requested GP work"
|
||||
);
|
||||
ready_changed
|
||||
}
|
||||
|
||||
fn complete_callback_gp(&self, completed: RcuSequence) -> bool {
|
||||
let _ownership = self.callback_ownership.lock_irqsave();
|
||||
let mut became_ready = false;
|
||||
for callbacks in &self.cpu_callbacks {
|
||||
let ready = callbacks
|
||||
.state
|
||||
.lock_irqsave()
|
||||
.segments
|
||||
.complete_gp(completed);
|
||||
if ready {
|
||||
callbacks.publish_work();
|
||||
became_ready = true;
|
||||
}
|
||||
}
|
||||
became_ready
|
||||
}
|
||||
|
||||
fn prepare_callback_gp_start(&self, starting: RcuSequence) {
|
||||
let _ownership = self.callback_ownership.lock_irqsave();
|
||||
for callbacks in &self.cpu_callbacks {
|
||||
let mut state = callbacks.state.lock_irqsave();
|
||||
state.segments.prepare_gp_start(starting);
|
||||
let runnable = state.segments.has_ready() || state.segments.has_unclassified();
|
||||
callbacks.needs_scan.store(runnable, Ordering::Release);
|
||||
}
|
||||
}
|
||||
|
||||
fn classify_new_callbacks(&self, inner: &mut RcuStateInner) -> bool {
|
||||
let _ownership = self.callback_ownership.lock_irqsave();
|
||||
let mut classified = false;
|
||||
for callbacks in &self.cpu_callbacks {
|
||||
// Only `next` admissions require classification, and enqueue
|
||||
// publishes them through this persistent per-CPU predicate.
|
||||
// Ready-only queues may still pass the filter, but idle and
|
||||
// GP-blocked possible CPUs avoid an irqsave lock entirely.
|
||||
if !callbacks.needs_scan.load(Ordering::Acquire) {
|
||||
continue;
|
||||
}
|
||||
let mut state = callbacks.state.lock_irqsave();
|
||||
if state.segments.has_unclassified() {
|
||||
let target = inner.gp.request_future();
|
||||
let active = inner.gp.is_active();
|
||||
classified |= state.segments.classify_next(target, active);
|
||||
}
|
||||
let runnable = state.segments.has_ready() || state.segments.has_unclassified();
|
||||
callbacks.needs_scan.store(runnable, Ordering::Release);
|
||||
}
|
||||
classified
|
||||
}
|
||||
|
||||
/// Moves all queued callback segments while the caller holds the GP lock.
|
||||
fn migrate_callback_segments(&self, source: usize, destination: usize) {
|
||||
if source == destination {
|
||||
return;
|
||||
}
|
||||
|
||||
let _ownership = self.callback_ownership.lock_irqsave();
|
||||
let (low, high) = if source < destination {
|
||||
(source, destination)
|
||||
} else {
|
||||
(destination, source)
|
||||
};
|
||||
let mut low_state = self.cpu_callbacks[low].state.lock_irqsave();
|
||||
let mut high_state = self.cpu_callbacks[high].state.lock_irqsave();
|
||||
|
||||
if source < destination {
|
||||
high_state.segments.merge_from(&mut low_state.segments);
|
||||
} else {
|
||||
low_state.segments.merge_from(&mut high_state.segments);
|
||||
}
|
||||
|
||||
let destination_state = if destination == low {
|
||||
&*low_state
|
||||
} else {
|
||||
&*high_state
|
||||
};
|
||||
let destination_runnable =
|
||||
destination_state.segments.has_ready() || destination_state.segments.has_unclassified();
|
||||
self.cpu_callbacks[destination]
|
||||
.needs_scan
|
||||
.store(destination_runnable, Ordering::Release);
|
||||
|
||||
let source_state = if source == low {
|
||||
&*low_state
|
||||
} else {
|
||||
&*high_state
|
||||
};
|
||||
let source_runnable =
|
||||
source_state.segments.has_ready() || source_state.segments.has_unclassified();
|
||||
self.cpu_callbacks[source]
|
||||
.needs_scan
|
||||
.store(source_runnable, Ordering::Release);
|
||||
}
|
||||
|
||||
fn progress_callbacks_and_gps(&self) -> bool {
|
||||
let mut inner = self.inner.lock_irqsave();
|
||||
let mut ready_changed = Self::pump_grace_periods(&mut inner);
|
||||
let classified = self.classify_new_callbacks(&mut inner);
|
||||
if classified {
|
||||
ready_changed |= Self::pump_grace_periods(&mut inner);
|
||||
}
|
||||
ready_changed
|
||||
}
|
||||
|
||||
fn has_worker_work(&self) -> bool {
|
||||
if self
|
||||
.cpu_callbacks
|
||||
.iter()
|
||||
.any(|callbacks| callbacks.needs_scan.load(Ordering::Acquire))
|
||||
{
|
||||
return true;
|
||||
}
|
||||
let inner = self.inner.lock_irqsave();
|
||||
inner.gp.ready_to_complete() || (!inner.gp.is_active() && inner.gp.has_request())
|
||||
}
|
||||
|
||||
fn wake_state_waiters(&self) {
|
||||
self.state_wait.wake_all();
|
||||
}
|
||||
|
||||
fn wake_worker(&self) {
|
||||
self.worker_wait.wake_all();
|
||||
// A single worker only needs one outstanding kick. Coalescing here
|
||||
// keeps simultaneous per-CPU admissions from all serializing on the
|
||||
// waitqueue's internal lock.
|
||||
if !self.worker_kick_pending.swap(true, Ordering::AcqRel) {
|
||||
self.worker_wait.wake_all();
|
||||
}
|
||||
}
|
||||
|
||||
fn wake_barrier_waiter_if_pending(&self) {
|
||||
if self.barrier_remaining.load(Ordering::Acquire) != 0 {
|
||||
self.barrier_wait.wake_all();
|
||||
}
|
||||
}
|
||||
|
||||
fn progress_and_drain_inline_if_no_worker(&self) {
|
||||
if self.worker_started.load(Ordering::Acquire) {
|
||||
return;
|
||||
}
|
||||
|
||||
self.process_ready_callbacks();
|
||||
self.process_callback_batch();
|
||||
}
|
||||
|
||||
fn process_ready_callbacks(&self) {
|
||||
{
|
||||
let mut inner = self.inner.lock_irqsave();
|
||||
Self::pump_grace_periods(&mut inner);
|
||||
if !inner.has_ready_work() || !inner.callbacks.try_claim_drainer() {
|
||||
return;
|
||||
fn try_claim_executor(&self) -> Option<RcuExecutorGuard<'_>> {
|
||||
self.executor_claimed
|
||||
.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
|
||||
.ok()
|
||||
.map(|_| RcuExecutorGuard {
|
||||
claimed: &self.executor_claimed,
|
||||
})
|
||||
}
|
||||
|
||||
fn pop_ready_from_cpu(&self, cpu: usize) -> Option<callback::ReadyRcuCallback> {
|
||||
let callbacks = &self.cpu_callbacks[cpu];
|
||||
// Most possible CPUs have no runnable callback. Avoid taking their
|
||||
// queue locks during a scan; publishers store queue state before
|
||||
// making this persistent predicate visible.
|
||||
if !callbacks.needs_scan.load(Ordering::Acquire) {
|
||||
return None;
|
||||
}
|
||||
|
||||
let mut state = callbacks.state.lock_irqsave();
|
||||
let callback = state.segments.pop_ready()?;
|
||||
debug_assert!(!state.executing);
|
||||
state.executing = true;
|
||||
let more = state.segments.has_ready() || state.segments.has_unclassified();
|
||||
callbacks.needs_scan.store(more, Ordering::Release);
|
||||
Some(callback)
|
||||
}
|
||||
|
||||
fn pop_ready_round_robin(
|
||||
&self,
|
||||
preferred_cpu: Option<usize>,
|
||||
) -> Option<(usize, callback::ReadyRcuCallback)> {
|
||||
let cpu_count = PerCpu::MAX_CPU_NUM as usize;
|
||||
if let Some(cpu) = preferred_cpu {
|
||||
if let Some(callback) = self.pop_ready_from_cpu(cpu) {
|
||||
return Some((cpu, callback));
|
||||
}
|
||||
}
|
||||
|
||||
loop {
|
||||
let next = {
|
||||
let mut inner = self.inner.lock_irqsave();
|
||||
let completed_gp = inner.gp.completed();
|
||||
match inner.callback_queue.pop_ready(completed_gp) {
|
||||
Some(callback) => Some(callback),
|
||||
None => {
|
||||
inner.callbacks.release_drainer();
|
||||
None
|
||||
}
|
||||
}
|
||||
};
|
||||
let start = self.next_scan_cpu.load(Ordering::Relaxed) % cpu_count;
|
||||
for offset in 0..cpu_count {
|
||||
let cpu = (start + offset) % cpu_count;
|
||||
if let Some(callback) = self.pop_ready_from_cpu(cpu) {
|
||||
self.next_scan_cpu
|
||||
.store((cpu + 1) % cpu_count, Ordering::Relaxed);
|
||||
return Some((cpu, callback));
|
||||
}
|
||||
}
|
||||
None
|
||||
}
|
||||
|
||||
let Some(callback) = next else {
|
||||
fn finish_callback(&self, cpu: usize) {
|
||||
let callbacks = &self.cpu_callbacks[cpu];
|
||||
let mut state = callbacks.state.lock_irqsave();
|
||||
debug_assert!(state.executing);
|
||||
state.executing = false;
|
||||
let more = state.segments.has_ready() || state.segments.has_unclassified();
|
||||
callbacks.needs_scan.store(more, Ordering::Release);
|
||||
}
|
||||
|
||||
fn process_callback_batch(&self) -> usize {
|
||||
let Some(_executor) = self.try_claim_executor() else {
|
||||
self.wake_worker();
|
||||
return 0;
|
||||
};
|
||||
|
||||
self.progress_callbacks_and_gps();
|
||||
// GP completion, including a completion credited from a context
|
||||
// snapshot, may satisfy synchronize_rcu() even when no callback is
|
||||
// ready. One wake per batch is sufficient.
|
||||
self.wake_state_waiters();
|
||||
let mut count = 0;
|
||||
let mut preferred_cpu = None;
|
||||
let mut cpu_quantum = 0;
|
||||
while count < RCU_CALLBACK_BATCH_LIMIT {
|
||||
let Some((cpu, callback)) = self.pop_ready_round_robin(preferred_cpu) else {
|
||||
break;
|
||||
};
|
||||
|
||||
// SAFETY: `pop_ready()` detached the head, copied all state needed
|
||||
// after invocation, and released duplicate ownership. The unsafe
|
||||
// admission contract keeps the head valid until this call starts.
|
||||
unsafe { (callback.func)(callback.head) };
|
||||
self.callbacks_invoked.fetch_add(1, Ordering::Relaxed);
|
||||
self.finish_callback(cpu);
|
||||
count += 1;
|
||||
|
||||
{
|
||||
let mut inner = self.inner.lock_irqsave();
|
||||
inner.callbacks.complete(callback.seq);
|
||||
if preferred_cpu == Some(cpu) {
|
||||
cpu_quantum += 1;
|
||||
} else {
|
||||
preferred_cpu = Some(cpu);
|
||||
cpu_quantum = 1;
|
||||
}
|
||||
if cpu_quantum == RCU_CALLBACK_CPU_QUANTUM {
|
||||
preferred_cpu = None;
|
||||
cpu_quantum = 0;
|
||||
}
|
||||
|
||||
self.wake_state_waiters();
|
||||
}
|
||||
|
||||
let more = self.has_worker_work();
|
||||
drop(_executor);
|
||||
if more {
|
||||
self.wake_worker();
|
||||
}
|
||||
// In the pre-worker boot window, the barrier waiter is itself the
|
||||
// bounded inline executor. Wake it once per batch so markers behind
|
||||
// more than one batch cannot stall indefinitely.
|
||||
self.wake_barrier_waiter_if_pending();
|
||||
if count != 0 {
|
||||
cond_resched();
|
||||
}
|
||||
count
|
||||
}
|
||||
}
|
||||
|
||||
struct RcuExecutorGuard<'a> {
|
||||
claimed: &'a AtomicBool,
|
||||
}
|
||||
|
||||
impl Drop for RcuExecutorGuard<'_> {
|
||||
fn drop(&mut self) {
|
||||
self.claimed.store(false, Ordering::Release);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -633,7 +902,7 @@ fn report_quiescent_state(cpu: ProcessorId) {
|
||||
let mut inner = RCU_STATE.inner.lock_irqsave();
|
||||
report_quiescent_state_locked(&mut inner, cpu);
|
||||
let ready_changed = RcuState::pump_grace_periods(&mut inner);
|
||||
(ready_changed || inner.has_ready_work(), true)
|
||||
(ready_changed, true)
|
||||
};
|
||||
|
||||
if wake_waiters {
|
||||
@@ -642,24 +911,18 @@ fn report_quiescent_state(cpu: ProcessorId) {
|
||||
if wake_worker {
|
||||
RCU_STATE.wake_worker();
|
||||
}
|
||||
}
|
||||
|
||||
fn enqueue_callback_locked(
|
||||
inner: &mut RcuStateInner,
|
||||
head: NonNull<RcuHead>,
|
||||
func: RcuRawCallback,
|
||||
) {
|
||||
let target_gp = inner.gp.request_future();
|
||||
let seq = inner.callbacks.admit();
|
||||
inner.callback_queue.push(head, func, target_gp, seq);
|
||||
RCU_STATE.wake_barrier_waiter_if_pending();
|
||||
}
|
||||
|
||||
fn queue_raw_callback(head: NonNull<RcuHead>, func: RcuRawCallback) {
|
||||
let wake_worker = {
|
||||
let mut inner = RCU_STATE.inner.lock_irqsave();
|
||||
enqueue_callback_locked(&mut inner, head, func);
|
||||
inner.has_worker_work()
|
||||
};
|
||||
// Pin before selecting the queue. `lock_irqsave()` only disables
|
||||
// preemption after a particular lock has already been selected.
|
||||
let pin = PreemptGuard::new();
|
||||
let cpu = smp_get_processor_id().data() as usize;
|
||||
let callbacks = &RCU_STATE.cpu_callbacks[cpu];
|
||||
callbacks.state.lock_irqsave().segments.enqueue(head, func);
|
||||
let wake_worker = callbacks.publish_work();
|
||||
drop(pin);
|
||||
|
||||
if wake_worker {
|
||||
RCU_STATE.wake_worker();
|
||||
@@ -734,7 +997,20 @@ fn worker_main() -> i32 {
|
||||
return Some(());
|
||||
}
|
||||
|
||||
if RCU_STATE.inner.lock_irqsave().has_worker_work() {
|
||||
if RCU_STATE.has_worker_work() {
|
||||
return Some(());
|
||||
}
|
||||
|
||||
// Retire the coalesced kick only after observing no work, then
|
||||
// recheck. A publisher racing either side of this store will
|
||||
// therefore be observed by the predicate or issue a fresh wake.
|
||||
RCU_STATE
|
||||
.worker_kick_pending
|
||||
.store(false, Ordering::Release);
|
||||
if RCU_STATE.worker_should_stop.load(Ordering::Acquire) {
|
||||
return Some(());
|
||||
}
|
||||
if RCU_STATE.has_worker_work() {
|
||||
return Some(());
|
||||
}
|
||||
|
||||
@@ -745,7 +1021,7 @@ fn worker_main() -> i32 {
|
||||
break;
|
||||
}
|
||||
|
||||
RCU_STATE.process_ready_callbacks();
|
||||
RCU_STATE.process_callback_batch();
|
||||
}
|
||||
|
||||
{
|
||||
@@ -753,6 +1029,11 @@ fn worker_main() -> i32 {
|
||||
RCU_STATE.worker_started.store(false, Ordering::Release);
|
||||
}
|
||||
RCU_STATE.wake_state_waiters();
|
||||
// Publish the executor handoff to a barrier that observed the worker
|
||||
// before shutdown. This wake must be unconditional: the exit path has no
|
||||
// synchronization that requires it to observe a concurrent barrier's
|
||||
// counter publication before deciding whether to wake.
|
||||
RCU_STATE.barrier_wait.wake_all();
|
||||
0
|
||||
}
|
||||
|
||||
@@ -1059,39 +1340,55 @@ pub fn rcu_barrier() {
|
||||
debug_assert!(!rcu_read_lock_held());
|
||||
}
|
||||
|
||||
let target_cb = {
|
||||
let inner = RCU_STATE.inner.lock_irqsave();
|
||||
inner.callbacks.barrier_target()
|
||||
};
|
||||
let _barrier = RCU_STATE.barrier_mutex.lock();
|
||||
debug_assert_eq!(RCU_STATE.barrier_remaining.load(Ordering::Acquire), 0);
|
||||
RCU_STATE.barrier_remaining.store(1, Ordering::Release);
|
||||
|
||||
let Some(target_cb) = target_cb else {
|
||||
return;
|
||||
};
|
||||
|
||||
loop {
|
||||
RCU_STATE.progress_and_drain_inline_if_no_worker();
|
||||
|
||||
let done = {
|
||||
let inner = RCU_STATE.inner.lock_irqsave();
|
||||
inner.callbacks.has_completed(target_cb)
|
||||
};
|
||||
if done {
|
||||
return;
|
||||
}
|
||||
|
||||
RCU_STATE.state_wait.wait_until(|| {
|
||||
RCU_STATE.progress_and_drain_inline_if_no_worker();
|
||||
let completed = RCU_STATE
|
||||
.inner
|
||||
.lock_irqsave()
|
||||
.callbacks
|
||||
.has_completed(target_cb);
|
||||
if completed {
|
||||
Some(())
|
||||
} else {
|
||||
None
|
||||
{
|
||||
// Keep ownership stable across the complete scan. Otherwise a CPU
|
||||
// migration could move callbacks from an unscanned source into an
|
||||
// already-scanned destination.
|
||||
let _ownership = RCU_STATE.callback_ownership.lock_irqsave();
|
||||
for callbacks in &RCU_STATE.cpu_callbacks {
|
||||
let mut state = callbacks.state.lock_irqsave();
|
||||
if state.segments.is_empty() && !state.executing {
|
||||
continue;
|
||||
}
|
||||
});
|
||||
|
||||
if !callbacks.barrier_head.try_claim() {
|
||||
panic!("RCU barrier head was already queued");
|
||||
}
|
||||
RCU_STATE.barrier_remaining.fetch_add(1, Ordering::AcqRel);
|
||||
let marker = NonNull::from(&callbacks.barrier_head);
|
||||
if !state.segments.entrain(marker, rcu_barrier_callback) {
|
||||
debug_assert!(state.executing);
|
||||
state.segments.push_done(marker, rcu_barrier_callback);
|
||||
}
|
||||
if state.segments.has_ready() || state.segments.has_unclassified() {
|
||||
callbacks.publish_work();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Drop the setup sentinel only after every queue has been inspected.
|
||||
if RCU_STATE.barrier_remaining.fetch_sub(1, Ordering::AcqRel) == 1 {
|
||||
fence(Ordering::SeqCst);
|
||||
return;
|
||||
}
|
||||
|
||||
RCU_STATE.wake_worker();
|
||||
RCU_STATE.barrier_wait.wait_until(|| {
|
||||
RCU_STATE.progress_and_drain_inline_if_no_worker();
|
||||
(RCU_STATE.barrier_remaining.load(Ordering::Acquire) == 0).then_some(())
|
||||
});
|
||||
fence(Ordering::SeqCst);
|
||||
}
|
||||
|
||||
unsafe fn rcu_barrier_callback(_head: NonNull<RcuHead>) {
|
||||
let previous = RCU_STATE.barrier_remaining.fetch_sub(1, Ordering::AcqRel);
|
||||
debug_assert!(previous > 0, "RCU barrier callback count underflow");
|
||||
if previous == 1 {
|
||||
RCU_STATE.barrier_wait.wake_all();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1142,12 +1439,16 @@ fn note_context_eqs_transition(transition: ContextTransition) {
|
||||
(false, false)
|
||||
} else {
|
||||
let ready_changed = RcuState::pump_grace_periods(&mut inner);
|
||||
(true, ready_changed || inner.has_ready_work())
|
||||
(true, ready_changed)
|
||||
}
|
||||
};
|
||||
|
||||
if wake_waiters {
|
||||
RCU_STATE.wake_state_waiters();
|
||||
// Before the callback worker starts, rcu_barrier() is the bounded
|
||||
// inline executor. An EQS transition may be the final GP holdout, so
|
||||
// it must wake that executor just like report_qs() and cpu_dying().
|
||||
RCU_STATE.wake_barrier_waiter_if_pending();
|
||||
}
|
||||
if wake_worker {
|
||||
RCU_STATE.wake_worker();
|
||||
@@ -1280,28 +1581,102 @@ pub fn cpu_dying(cpu: ProcessorId) {
|
||||
let mut inner = RCU_STATE.inner.lock_irqsave();
|
||||
cpu_dying_locked(&mut inner, cpu);
|
||||
RCU_STATE.contexts[cpu.data() as usize].clear_gp_report();
|
||||
let ready_changed = RcuState::pump_grace_periods(&mut inner);
|
||||
ready_changed || inner.has_ready_work()
|
||||
let mut ready_changed = RcuState::pump_grace_periods(&mut inner);
|
||||
let classified = RCU_STATE.classify_new_callbacks(&mut inner);
|
||||
if classified {
|
||||
ready_changed |= RcuState::pump_grace_periods(&mut inner);
|
||||
}
|
||||
|
||||
// `cpu_dying_locked()` removed the source from the authoritative RCU
|
||||
// admission set. Select the destination from that same set while the
|
||||
// GP lock is held, so lifecycle state and queue ownership cannot
|
||||
// disagree. If this is the final participant, the global executor can
|
||||
// still drain the stable source record without migrating it.
|
||||
if let Some(destination) = inner.participating_cpus.iter_cpu().next() {
|
||||
RCU_STATE.migrate_callback_segments(cpu.data() as usize, destination.data() as usize);
|
||||
}
|
||||
|
||||
ready_changed
|
||||
|| RCU_STATE
|
||||
.cpu_callbacks
|
||||
.iter()
|
||||
.any(|callbacks| callbacks.needs_scan.load(Ordering::Acquire))
|
||||
};
|
||||
|
||||
RCU_STATE.wake_state_waiters();
|
||||
if wake_worker {
|
||||
RCU_STATE.wake_worker();
|
||||
}
|
||||
RCU_STATE.wake_barrier_waiter_if_pending();
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
pub fn debug_snapshot() -> (u64, u64, u64, usize, bool) {
|
||||
let inner = RCU_STATE.inner.lock_irqsave();
|
||||
let (aggregate, _) = callback_queue_depth_snapshot();
|
||||
(
|
||||
inner.gp.current().raw(),
|
||||
inner.gp.completed().raw(),
|
||||
inner.callbacks.completed_raw(),
|
||||
inner.callback_queue.len(),
|
||||
inner.has_ready_work(),
|
||||
RCU_STATE.callbacks_invoked.load(Ordering::Relaxed) as u64,
|
||||
aggregate.total(),
|
||||
aggregate.done != 0,
|
||||
)
|
||||
}
|
||||
|
||||
pub(crate) fn callback_queue_depth_snapshot() -> (
|
||||
RcuCallbackQueueDepth,
|
||||
[RcuCallbackQueueDepth; PerCpu::MAX_CPU_NUM as usize],
|
||||
) {
|
||||
let mut aggregate = RcuCallbackQueueDepth::default();
|
||||
let mut per_cpu = [RcuCallbackQueueDepth::default(); PerCpu::MAX_CPU_NUM as usize];
|
||||
for (cpu, callbacks) in RCU_STATE.cpu_callbacks.iter().enumerate() {
|
||||
let depth = callbacks.state.lock_irqsave().depth();
|
||||
per_cpu[cpu] = depth;
|
||||
aggregate.add_assign(depth);
|
||||
}
|
||||
(aggregate, per_cpu)
|
||||
}
|
||||
|
||||
pub(crate) fn callback_queue_debug_report() -> String {
|
||||
let (aggregate, per_cpu) = callback_queue_depth_snapshot();
|
||||
let mut report = String::new();
|
||||
writeln!(
|
||||
report,
|
||||
"aggregate total={} done={} wait={} next_ready={} next={} executing={}",
|
||||
aggregate.total(),
|
||||
aggregate.done,
|
||||
aggregate.wait,
|
||||
aggregate.next_ready,
|
||||
aggregate.next,
|
||||
usize::from(aggregate.executing),
|
||||
)
|
||||
.expect("writing RCU callback snapshot to String failed");
|
||||
|
||||
for (cpu, depth) in per_cpu.iter().copied().enumerate() {
|
||||
let present = !smp_cpu_manager_initialized()
|
||||
|| smp_cpu_manager()
|
||||
.present_cpus()
|
||||
.get(ProcessorId::new(cpu as u32))
|
||||
.unwrap_or(false);
|
||||
if !present && depth.total() == 0 && !depth.executing {
|
||||
continue;
|
||||
}
|
||||
writeln!(
|
||||
report,
|
||||
"cpu={} total={} done={} wait={} next_ready={} next={} executing={}",
|
||||
cpu,
|
||||
depth.total(),
|
||||
depth.done,
|
||||
depth.wait,
|
||||
depth.next_ready,
|
||||
depth.next,
|
||||
usize::from(depth.executing),
|
||||
)
|
||||
.expect("writing RCU per-CPU callback snapshot to String failed");
|
||||
}
|
||||
report
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
pub fn debug_force_quiescent_state() {
|
||||
report_quiescent_state(smp_get_processor_id());
|
||||
|
||||
+265
-68
@@ -10,10 +10,14 @@ use crate::{
|
||||
exception::InterruptArch,
|
||||
ipc::sighand::SigHand,
|
||||
libs::{
|
||||
cpumask::CpuMask,
|
||||
notifier::{AtomicNotifierChain, NotifierBlock, NotifyResult},
|
||||
spinlock::SpinLock,
|
||||
},
|
||||
process::kthread::{KernelThreadClosure, KernelThreadMechanism},
|
||||
process::{
|
||||
kthread::{KernelThreadClosure, KernelThreadMechanism},
|
||||
ProcessManager,
|
||||
},
|
||||
sched::completion::Completion,
|
||||
smp::cpu::{smp_cpu_manager, ProcessorId},
|
||||
};
|
||||
@@ -125,6 +129,217 @@ unsafe fn rcu_selftest_callback(head: NonNull<RcuHead>) {
|
||||
probe.hits.fetch_add(1, Ordering::SeqCst);
|
||||
}
|
||||
|
||||
fn queue_callback_probe(hits: Arc<AtomicUsize>) {
|
||||
let probe = Box::into_raw(Box::new(RcuSelftestCallbackProbe {
|
||||
head: RcuHead::new(),
|
||||
hits,
|
||||
}));
|
||||
// SAFETY: the callback owns the stable Box and reconstructs it exactly
|
||||
// once after admission has detached the embedded head.
|
||||
unsafe {
|
||||
call_rcu_raw(
|
||||
NonNull::new_unchecked(ptr::addr_of_mut!((*probe).head)),
|
||||
rcu_selftest_callback,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
fn run_callback_flood_selftest() -> Result<(), &'static str> {
|
||||
let hits = Arc::new(AtomicUsize::new(0));
|
||||
let callbacks = RCU_CALLBACK_BATCH_LIMIT * 2 + 1;
|
||||
for _ in 0..callbacks {
|
||||
queue_callback_probe(hits.clone());
|
||||
}
|
||||
rcu_barrier();
|
||||
if hits.load(Ordering::SeqCst) != callbacks {
|
||||
return Err("RCU callback flood did not drain across bounded batches");
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn run_smp_callback_barrier_selftest() -> Result<(), &'static str> {
|
||||
let cpus: Vec<ProcessorId> = smp_cpu_manager()
|
||||
.present_cpus()
|
||||
.iter_cpu()
|
||||
.filter(|cpu| smp_cpu_manager().is_online_cpu(*cpu))
|
||||
.collect();
|
||||
if cpus.is_empty() {
|
||||
return Err("RCU SMP callback selftest found no online CPU");
|
||||
}
|
||||
|
||||
const CALLBACKS_PER_CPU: usize = 9;
|
||||
let hits = Arc::new(AtomicUsize::new(0));
|
||||
let ready = Arc::new(Completion::new());
|
||||
let start = Arc::new(Completion::new());
|
||||
let first_admitted = Arc::new(Completion::new());
|
||||
let continue_enqueue = Arc::new(Completion::new());
|
||||
let done = Arc::new(Completion::new());
|
||||
let mut workers = Vec::new();
|
||||
|
||||
for cpu in cpus.iter().copied() {
|
||||
let thread_hits = hits.clone();
|
||||
let thread_ready = ready.clone();
|
||||
let thread_start = start.clone();
|
||||
let thread_first_admitted = first_admitted.clone();
|
||||
let thread_continue = continue_enqueue.clone();
|
||||
let thread_done = done.clone();
|
||||
let closure = KernelThreadClosure::EmptyClosure((
|
||||
Box::new(move || {
|
||||
thread_ready.complete();
|
||||
if thread_start.wait_for_completion().is_err() {
|
||||
thread_done.complete();
|
||||
return 1;
|
||||
}
|
||||
|
||||
queue_callback_probe(thread_hits.clone());
|
||||
thread_first_admitted.complete();
|
||||
if thread_continue.wait_for_completion().is_err() {
|
||||
thread_done.complete();
|
||||
return 1;
|
||||
}
|
||||
for _ in 1..CALLBACKS_PER_CPU {
|
||||
queue_callback_probe(thread_hits.clone());
|
||||
}
|
||||
thread_done.complete();
|
||||
0
|
||||
}),
|
||||
(),
|
||||
));
|
||||
let Some(worker) = KernelThreadMechanism::create_on_cpu(
|
||||
closure,
|
||||
format!("rcu-callback-cpu{}", cpu.data()),
|
||||
cpu,
|
||||
) else {
|
||||
start.complete_all();
|
||||
continue_enqueue.complete_all();
|
||||
for worker in &workers {
|
||||
let _ = KernelThreadMechanism::stop(worker);
|
||||
}
|
||||
return Err("RCU SMP callback selftest could not create a worker");
|
||||
};
|
||||
if ProcessManager::wakeup(&worker).is_err() {
|
||||
start.complete_all();
|
||||
continue_enqueue.complete_all();
|
||||
let _ = KernelThreadMechanism::stop(&worker);
|
||||
for worker in &workers {
|
||||
let _ = KernelThreadMechanism::stop(worker);
|
||||
}
|
||||
return Err("RCU SMP callback selftest could not wake a worker");
|
||||
}
|
||||
workers.push(worker);
|
||||
}
|
||||
|
||||
for _ in &workers {
|
||||
if ready.wait_for_completion().is_err() {
|
||||
start.complete_all();
|
||||
continue_enqueue.complete_all();
|
||||
for worker in &workers {
|
||||
let _ = KernelThreadMechanism::stop(worker);
|
||||
}
|
||||
return Err("RCU SMP callback workers did not reach their start gate");
|
||||
}
|
||||
}
|
||||
start.complete_all();
|
||||
for _ in &workers {
|
||||
if first_admitted.wait_for_completion().is_err() {
|
||||
continue_enqueue.complete_all();
|
||||
for worker in &workers {
|
||||
let _ = KernelThreadMechanism::stop(worker);
|
||||
}
|
||||
return Err("RCU SMP callback workers did not admit their first callback");
|
||||
}
|
||||
}
|
||||
|
||||
// Every CPU now owns a callback ahead of its barrier marker. Release the
|
||||
// remaining admissions immediately before the barrier to exercise its
|
||||
// ownership scan concurrently with enqueue.
|
||||
continue_enqueue.complete_all();
|
||||
rcu_barrier();
|
||||
for _ in &workers {
|
||||
if done.wait_for_completion().is_err() {
|
||||
for worker in &workers {
|
||||
let _ = KernelThreadMechanism::stop(worker);
|
||||
}
|
||||
return Err("RCU SMP callback workers did not finish enqueue");
|
||||
}
|
||||
}
|
||||
rcu_barrier();
|
||||
|
||||
let mut stopped = true;
|
||||
for worker in &workers {
|
||||
stopped &= KernelThreadMechanism::stop(worker).is_ok();
|
||||
}
|
||||
if !stopped {
|
||||
return Err("RCU SMP callback selftest could not stop its workers");
|
||||
}
|
||||
if hits.load(Ordering::SeqCst) != workers.len() * CALLBACKS_PER_CPU {
|
||||
return Err("RCU SMP callback/barrier selftest lost an admitted callback");
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn run_concurrent_barrier_selftest() -> Result<(), &'static str> {
|
||||
let hits = Arc::new(AtomicUsize::new(0));
|
||||
for _ in 0..(RCU_CALLBACK_BATCH_LIMIT + 1) {
|
||||
queue_callback_probe(hits.clone());
|
||||
}
|
||||
|
||||
let ready = Arc::new(Completion::new());
|
||||
let start = Arc::new(Completion::new());
|
||||
let done = Arc::new(Completion::new());
|
||||
let spawn = |name: &'static str| {
|
||||
let ready = ready.clone();
|
||||
let start = start.clone();
|
||||
let done = done.clone();
|
||||
KernelThreadMechanism::create_and_run(
|
||||
KernelThreadClosure::EmptyClosure((
|
||||
Box::new(move || {
|
||||
ready.complete();
|
||||
if start.wait_for_completion().is_err() {
|
||||
done.complete();
|
||||
return 1;
|
||||
}
|
||||
rcu_barrier();
|
||||
done.complete();
|
||||
0
|
||||
}),
|
||||
(),
|
||||
)),
|
||||
name.into(),
|
||||
)
|
||||
};
|
||||
|
||||
let first = spawn("rcu-barrier-a")
|
||||
.ok_or("concurrent barrier selftest could not create its first worker")?;
|
||||
let Some(second) = spawn("rcu-barrier-b") else {
|
||||
start.complete_all();
|
||||
let _ = KernelThreadMechanism::stop(&first);
|
||||
return Err("concurrent barrier selftest could not create its second worker");
|
||||
};
|
||||
if ready.wait_for_completion().is_err() || ready.wait_for_completion().is_err() {
|
||||
start.complete_all();
|
||||
let _ = KernelThreadMechanism::stop(&first);
|
||||
let _ = KernelThreadMechanism::stop(&second);
|
||||
return Err("concurrent barrier workers did not reach their start gate");
|
||||
}
|
||||
start.complete_all();
|
||||
if done.wait_for_completion().is_err() || done.wait_for_completion().is_err() {
|
||||
let _ = KernelThreadMechanism::stop(&first);
|
||||
let _ = KernelThreadMechanism::stop(&second);
|
||||
return Err("concurrent barrier workers did not finish");
|
||||
}
|
||||
|
||||
let first_stopped = KernelThreadMechanism::stop(&first).is_ok();
|
||||
let second_stopped = KernelThreadMechanism::stop(&second).is_ok();
|
||||
if !first_stopped || !second_stopped {
|
||||
return Err("concurrent barrier selftest could not stop its workers");
|
||||
}
|
||||
if hits.load(Ordering::SeqCst) != RCU_CALLBACK_BATCH_LIMIT + 1 {
|
||||
return Err("concurrent barriers returned before their callback prefixes drained");
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[repr(C)]
|
||||
struct RcuSelftestRequeueProbe {
|
||||
head: RcuHead,
|
||||
@@ -212,10 +427,14 @@ fn run_duplicate_claim_selftest() -> Result<(), &'static str> {
|
||||
fn run_pr1_selftest() -> Result<(), &'static str> {
|
||||
gp::run_state_machine_selftests()?;
|
||||
context::run_context_selftests()?;
|
||||
callback::run_segmented_callback_selftests()?;
|
||||
run_callback_generation_selftest()?;
|
||||
run_duplicate_claim_selftest()?;
|
||||
run_cpu_hotplug_lifecycle_selftest()?;
|
||||
run_cpu_hotplug_concurrent_selftest()?;
|
||||
run_callback_flood_selftest()?;
|
||||
run_smp_callback_barrier_selftest()?;
|
||||
run_concurrent_barrier_selftest()?;
|
||||
|
||||
if ProcessManager::current_pcb().rcu_read_depth() != 0 {
|
||||
return Err("initial rcu_read_depth was not zero");
|
||||
@@ -403,6 +622,7 @@ fn run_callback_generation_selftest() -> Result<(), &'static str> {
|
||||
let cpu0 = ProcessorId::new(0);
|
||||
let cpu1 = ProcessorId::new(1);
|
||||
let mut inner = RcuStateInner::new();
|
||||
let mut callbacks = RcuSegmentedCallbacks::new();
|
||||
let first_head = RcuHead::new();
|
||||
let second_head = RcuHead::new();
|
||||
|
||||
@@ -418,33 +638,34 @@ fn run_callback_generation_selftest() -> Result<(), &'static str> {
|
||||
if !first_head.try_claim() || !second_head.try_claim() {
|
||||
return Err("fresh callback generation heads could not be claimed");
|
||||
}
|
||||
enqueue_callback_locked(&mut inner, NonNull::from(&first_head), noop_callback);
|
||||
enqueue_callback_locked(&mut inner, NonNull::from(&second_head), noop_callback);
|
||||
|
||||
callbacks.enqueue(NonNull::from(&first_head), noop_callback);
|
||||
callbacks.enqueue(NonNull::from(&second_head), noop_callback);
|
||||
let expected_second_gp = first_gp.next();
|
||||
if inner.callback_queue.len() != 2
|
||||
|| inner.callback_queue.front_target() != Some(expected_second_gp)
|
||||
let requested_second_gp = inner.gp.request_future();
|
||||
if requested_second_gp != expected_second_gp
|
||||
|| !callbacks.classify_next(requested_second_gp, true)
|
||||
|| callbacks.depth().next_ready != 2
|
||||
{
|
||||
return Err("callback admitted during an active GP targeted the wrong generation");
|
||||
}
|
||||
if inner.has_worker_work() {
|
||||
if callbacks.has_ready() || callbacks.has_unclassified() {
|
||||
return Err("worker would spin on a future GP blocked by the active waiting mask");
|
||||
}
|
||||
|
||||
if !inner.gp.report_quiescent_state(cpu0) {
|
||||
return Err("callback generation selftest could not complete its first waiting mask");
|
||||
}
|
||||
let first_ready_changed = RcuState::pump_grace_periods_with(
|
||||
&mut inner,
|
||||
|_| (CpuMask::from_cpu(cpu1), [0; PerCpu::MAX_CPU_NUM as usize]),
|
||||
|_| {},
|
||||
|_| {},
|
||||
);
|
||||
if first_ready_changed
|
||||
|| inner.gp.completed() != first_gp
|
||||
if inner.gp.complete_ready() != first_gp || callbacks.complete_gp(first_gp) {
|
||||
return Err("callbacks became ready during the GP that preceded admission");
|
||||
}
|
||||
callbacks.prepare_gp_start(expected_second_gp);
|
||||
if inner
|
||||
.gp
|
||||
.start_requested(CpuMask::from_cpu(cpu1), [0; PerCpu::MAX_CPU_NUM as usize])
|
||||
!= expected_second_gp
|
||||
|| !inner.gp.is_waiting_for(cpu1)
|
||||
|| inner.callback_queue.len() != 2
|
||||
|| inner.has_ready_work()
|
||||
|| callbacks.depth().wait != 2
|
||||
|| callbacks.has_ready()
|
||||
{
|
||||
return Err("callbacks became ready before their post-admission GP completed");
|
||||
}
|
||||
@@ -452,41 +673,42 @@ fn run_callback_generation_selftest() -> Result<(), &'static str> {
|
||||
if !inner.gp.report_quiescent_state(cpu1) {
|
||||
return Err("callback generation selftest could not complete its second waiting mask");
|
||||
}
|
||||
let second_ready_changed = RcuState::pump_grace_periods_with(
|
||||
&mut inner,
|
||||
|_| (CpuMask::new(), [0; PerCpu::MAX_CPU_NUM as usize]),
|
||||
|_| {},
|
||||
|_| {},
|
||||
);
|
||||
if !second_ready_changed
|
||||
|| inner.gp.completed() != expected_second_gp
|
||||
|| inner.callback_queue.len() != 2
|
||||
|| !inner.has_ready_work()
|
||||
if inner.gp.complete_ready() != expected_second_gp
|
||||
|| !callbacks.complete_gp(expected_second_gp)
|
||||
|| callbacks.depth().done != 2
|
||||
|| !callbacks.has_ready()
|
||||
{
|
||||
return Err("callbacks did not become ready after their target GP completed");
|
||||
}
|
||||
if !inner.has_worker_work() {
|
||||
return Err("worker did not recognize a ready intrusive callback");
|
||||
}
|
||||
|
||||
let completed_gp = inner.gp.completed();
|
||||
let first_callback = inner.callback_queue.pop_ready(completed_gp).unwrap();
|
||||
let second_callback = inner.callback_queue.pop_ready(completed_gp).unwrap();
|
||||
if second_callback.seq != first_callback.seq.next() {
|
||||
let first_callback = callbacks.pop_ready().unwrap();
|
||||
let second_callback = callbacks.pop_ready().unwrap();
|
||||
if first_callback.head != NonNull::from(&first_head)
|
||||
|| second_callback.head != NonNull::from(&second_head)
|
||||
{
|
||||
return Err("callback generation transition did not preserve admission FIFO");
|
||||
}
|
||||
if !inner.callback_queue.is_empty() {
|
||||
if !callbacks.is_empty() {
|
||||
return Err("callback generation transition did not drain its FIFO");
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn pump_cpu_lifecycle_model(inner: &mut RcuStateInner) -> bool {
|
||||
RcuState::pump_grace_periods_with(
|
||||
inner,
|
||||
|participants| (participants.clone(), [0; PerCpu::MAX_CPU_NUM as usize]),
|
||||
|_| {},
|
||||
|_| {},
|
||||
|_| false,
|
||||
|_| {},
|
||||
)
|
||||
}
|
||||
|
||||
fn run_cpu_hotplug_lifecycle_selftest() -> Result<(), &'static str> {
|
||||
let cpu = ProcessorId::new(PerCpu::MAX_CPU_NUM - 1);
|
||||
let incoming = ProcessorId::new(PerCpu::MAX_CPU_NUM - 2);
|
||||
let snapshot =
|
||||
|participants: &CpuMask| (participants.clone(), [0; PerCpu::MAX_CPU_NUM as usize]);
|
||||
let mut inner = RcuStateInner::new();
|
||||
|
||||
if !prepare_cpu_starting_locked(&inner, cpu) {
|
||||
@@ -496,14 +718,14 @@ fn run_cpu_hotplug_lifecycle_selftest() -> Result<(), &'static str> {
|
||||
// BSP publication of Starting does not admit an AP that has not executed
|
||||
// its local RCU starting hook.
|
||||
let before_ap_runs = inner.gp.request_future();
|
||||
RcuState::pump_grace_periods_with(&mut inner, snapshot, |_| {}, |_| {});
|
||||
pump_cpu_lifecycle_model(&mut inner);
|
||||
if !inner.gp.has_completed(before_ap_runs) || inner.gp.is_waiting_for(cpu) {
|
||||
return Err("RCU waited for a Starting CPU before its AP-side hook");
|
||||
}
|
||||
|
||||
cpu_starting_locked(&mut inner, cpu);
|
||||
let after_ap_runs = inner.gp.request_future();
|
||||
RcuState::pump_grace_periods_with(&mut inner, snapshot, |_| {}, |_| {});
|
||||
pump_cpu_lifecycle_model(&mut inner);
|
||||
if !inner.gp.is_waiting_for(cpu) || inner.gp.has_completed(after_ap_runs) {
|
||||
return Err("RCU did not admit a CPU after its AP-side starting hook");
|
||||
}
|
||||
@@ -527,7 +749,7 @@ fn run_cpu_hotplug_lifecycle_selftest() -> Result<(), &'static str> {
|
||||
// existing responsibility under the same lock, and the next GP snapshot
|
||||
// must not add it back.
|
||||
cpu_dying_locked(&mut inner, cpu);
|
||||
RcuState::pump_grace_periods_with(&mut inner, snapshot, |_| {}, |_| {});
|
||||
pump_cpu_lifecycle_model(&mut inner);
|
||||
if inner.gp.is_waiting_for(cpu)
|
||||
|| !inner.gp.is_waiting_for(incoming)
|
||||
|| inner.gp.has_completed(next_gp)
|
||||
@@ -536,7 +758,7 @@ fn run_cpu_hotplug_lifecycle_selftest() -> Result<(), &'static str> {
|
||||
}
|
||||
|
||||
cpu_dying_locked(&mut inner, incoming);
|
||||
RcuState::pump_grace_periods_with(&mut inner, snapshot, |_| {}, |_| {});
|
||||
pump_cpu_lifecycle_model(&mut inner);
|
||||
if !inner.gp.has_completed(next_gp) {
|
||||
return Err("CPU Dying did not complete the GP it was responsible for");
|
||||
}
|
||||
@@ -550,42 +772,17 @@ fn run_cpu_hotplug_lifecycle_selftest() -> Result<(), &'static str> {
|
||||
}
|
||||
cpu_starting_locked(&mut inner, cpu);
|
||||
let target = inner.gp.request_future();
|
||||
RcuState::pump_grace_periods_with(&mut inner, snapshot, |_| {}, |_| {});
|
||||
pump_cpu_lifecycle_model(&mut inner);
|
||||
if !inner.gp.is_waiting_for(cpu) {
|
||||
return Err("repeated RCU lifecycle did not admit its online CPU");
|
||||
}
|
||||
cpu_dying_locked(&mut inner, cpu);
|
||||
RcuState::pump_grace_periods_with(&mut inner, snapshot, |_| {}, |_| {});
|
||||
pump_cpu_lifecycle_model(&mut inner);
|
||||
if !inner.gp.has_completed(target) || inner.gp.is_waiting_for(cpu) {
|
||||
return Err("repeated RCU lifecycle leaked a GP holdout");
|
||||
}
|
||||
}
|
||||
|
||||
// Queue ownership and barrier tickets remain global while Dying advances
|
||||
// the target GP. The bound worker invariant covers the later execution.
|
||||
cpu_starting_locked(&mut inner, cpu);
|
||||
unsafe fn noop_callback(_head: NonNull<RcuHead>) {}
|
||||
let callback_head = RcuHead::new();
|
||||
if !callback_head.try_claim() {
|
||||
return Err("fresh hotplug callback head could not be claimed");
|
||||
}
|
||||
enqueue_callback_locked(&mut inner, NonNull::from(&callback_head), noop_callback);
|
||||
let barrier_target = inner
|
||||
.callbacks
|
||||
.barrier_target()
|
||||
.ok_or("hotplug callback did not create a barrier target")?;
|
||||
RcuState::pump_grace_periods_with(&mut inner, snapshot, |_| {}, |_| {});
|
||||
cpu_dying_locked(&mut inner, cpu);
|
||||
RcuState::pump_grace_periods_with(&mut inner, snapshot, |_| {}, |_| {});
|
||||
let callback = inner
|
||||
.callback_queue
|
||||
.pop_ready(inner.gp.completed())
|
||||
.ok_or("Dying CPU did not advance its pending callback")?;
|
||||
inner.callbacks.complete(callback.seq);
|
||||
if !inner.callbacks.has_completed(barrier_target) {
|
||||
return Err("callback barrier ticket was lost across CPU Dying");
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
@@ -10,6 +10,7 @@
|
||||
namespace {
|
||||
|
||||
constexpr const char* kRcuSelftestPath = "/sys/kernel/debug/rcu/selftest";
|
||||
constexpr const char* kRcuCallbacksPath = "/sys/kernel/debug/rcu/callbacks";
|
||||
|
||||
std::string ReadAll(const char* path) {
|
||||
int fd = open(path, O_RDONLY);
|
||||
@@ -67,6 +68,18 @@ TEST(RcuSelftest, ReportIsStableAcrossReads) {
|
||||
EXPECT_EQ(first, second);
|
||||
}
|
||||
|
||||
TEST(RcuSelftest, CallbackQueueSnapshotIsPresentAndStablePerOpen) {
|
||||
const std::string report = ReadAll(kRcuCallbacksPath);
|
||||
ASSERT_FALSE(report.empty());
|
||||
EXPECT_NE(std::string::npos, report.find("aggregate total=")) << report;
|
||||
EXPECT_NE(std::string::npos, report.find(" done=")) << report;
|
||||
EXPECT_NE(std::string::npos, report.find(" wait=")) << report;
|
||||
EXPECT_NE(std::string::npos, report.find(" next_ready=")) << report;
|
||||
EXPECT_NE(std::string::npos, report.find(" next=")) << report;
|
||||
EXPECT_NE(std::string::npos, report.find(" executing=")) << report;
|
||||
EXPECT_NE(std::string::npos, report.find("cpu=0 total=")) << report;
|
||||
}
|
||||
|
||||
int main(int argc, char** argv) {
|
||||
::testing::InitGoogleTest(&argc, argv);
|
||||
return RUN_ALL_TESTS();
|
||||
|
||||
Reference in New Issue
Block a user