第10讲：实战与展望——写一个能在双核上跑的“Hello World”调度器

经过前9讲的跋涉，我们从单核定时器调度走到了多核缓存一致性。
理论讲得再多，不如亲手写一个能在双核上跑起来的微型内核。

这一讲，我们将动手在 QEMU 模拟的双核 RISC-V 平台上，实现一个极简调度器。
它会启动两个核心，创建两个任务，每个任务各自打印“Hello from Core X”，并且轮换执行。
你会亲眼看到两个核心同时输出，也会体会到自旋锁、内存屏障、缓存对齐如何影响正确性和性能。

代码总量不超过 500 行，但包含了多核 OS 的核心精髓。

---

1. 目标与运行环境

1.1 最终效果

在 QEMU 终端中，你会看到类似这样的输出（顺序可能不同）：

Core 0: Hello from Task A
Core 1: Hello from Task B
Core 0: Hello from Task A
Core 1: Hello from Task B
...

两个核心各自独立运行一个任务，任务间通过一个全局自旋锁保护共享打印资源，避免输出交错。

1.2 开发环境

• 模拟器：QEMU 6.0+，支持 smp 多核
• 架构：RISC‑V 64 (RV64GC)，也可选 ARM Cortex‑A，但 RISC‑V 的汇编更简洁
• 工具链：riscv64‑unknown‑elf‑gcc
• 调试：GDB + QEMU 的 -s -S

我们使用 RISC‑V 教学，因为它的原子指令和内存屏障清晰，没有 x86 那么多历史包袱。

---

2. 整体架构设计

我们的微型内核包含以下模块：

• 启动代码 (start.s)：识别核心 ID，BSP 初始化全局数据结构，AP 等待唤醒。
• 任务控制块 (TCB)：栈指针、状态、优先级（简化为轮转）。
• 调度器：全局就绪队列（受自旋锁保护），schedule() 在任务主动让出或定时器中断时调用。
• 任务创建：task_create() 分配栈空间，构造初始上下文。
• 自旋锁 + 内存屏障：原子交换实现 spin_lock/spin_unlock。
• 核间中断 (IPI)：用于唤醒空闲核心（可选简化版，本讲只用轮询就绪队列，不强制 IPI）。
• 系统心跳：使用 Core 0 的定时器中断触发重调度（可选，本讲让任务主动 yield 简化）。

为了聚焦多核调度，本讲采用协作式轮转：每个任务运行一段时间后主动调用 yield() 让出 CPU。
这样我们不需要处理抢占和定时器中断，可以更清晰地展示核心间任务切换。

---

3. 关键代码实现

3.1 多核启动与核心识别

RISC‑V 的每个核心（hart）有一个 mhartid 寄存器。我们在启动汇编中读取，core 0 执行初始化，其他 core 跳转到等待循环。

# start.s
.section .text
.globl _start
_start:
    csrr t0, mhartid      # 读取当前核心 ID
    li   t1, 0
    beq  t0, t1, bsp_init
    # 非 BSP 核心：直接等待被唤醒（简单轮询就绪队列）
    j    ap_wait

bsp_init:
    # 初始化全局就绪队列、自旋锁、全局变量
    call kernel_init
    # 创建两个任务
    call setup_tasks
    # 启动调度器（当前核心进入调度循环）
    call scheduler_enter

ap_wait:
    # AP 核心也进入调度循环（但此时就绪队列中已有任务）
    call scheduler_enter

在 scheduler_enter 中，每个核心会不断从全局就绪队列取出任务并运行。

3.2 任务控制块与就绪队列

// task.h
#define STACK_SIZE 4096
#define MAX_TASKS   16

typedef enum { TASK_READY, TASK_RUNNING, TASK_BLOCKED } task_state_t;

typedef struct tcb {
    uint64_t sp;                // 栈指针（任务自己的栈）
    uint64_t pc;                // 入口地址（实际上由上下文保存）
    task_state_t state;
    int core_id;                // 当前运行在哪个核心（-1 表示未运行）
    struct tcb *next;
    char name[32];
    uint8_t stack[STACK_SIZE];  // 简单起见，栈内嵌在 TCB 中（实际应分开）
} tcb_t;

extern tcb_t *ready_queue;      // 全局就绪队列（单向循环链表）
extern spinlock_t ready_lock;   // 保护就绪队列的自旋锁

3.3 自旋锁与内存屏障（RISC‑V）

RISC‑V 提供 amoswap.w 原子交换指令，但默认不包含内存屏障。我们需要用 fence 指令实现 acquire/release 语义。

// spinlock.h
typedef struct { volatile int lock; } spinlock_t;

static inline void spin_lock(spinlock_t *lk) {
    int expected = 0;
    int desired = 1;
    __asm__ volatile (
        "1: lr.w t0, (%0)\n"          // 加载保留
        "   bnez t0, 1b\n"            // 若已锁，自旋
        "   sc.w t1, %1, (%0)\n"      // 条件存储
        "   bnez t1, 1b\n"
        : : "r"(&lk->lock), "r"(desired) : "t0", "t1", "memory"
    );
    __asm__ volatile ("fence rw, rw" : : : "memory"); // acquire 屏障
}

static inline void spin_unlock(spinlock_t *lk) {
    __asm__ volatile ("fence rw, rw" : : : "memory"); // release 屏障
    lk->lock = 0;
}

更简洁的可使用 GCC 内置原子函数 __sync_lock_test_and_set，它会自动生成合适的屏障。

3.4 任务创建（构造初始上下文）

任务第一次被调度时，需要有一个伪造的栈帧，使得当我们从调度器 switch_to(task) 时，它能正确返回到任务入口函数。
RISC‑V 的上下文切换需要保存：ra（返回地址）、sp、s0-s11（被调用者保存寄存器）。我们简化：只保存 ra 和 sp，因为我们的任务不涉及复杂计算。

void task_create(tcb_t *task, void (*entry)(void *), void *arg, const char *name) {
    // 栈从高地址向低地址增长；预留空间给上下文（16字节）
    uint64_t *sp = (uint64_t *)(task->stack + STACK_SIZE);
    // 构造上下文：模拟被调度出去的场景
    *(--sp) = (uint64_t)entry;      // ra（返回地址）
    *(--sp) = (uint64_t)arg;        // a0（第一个参数）
    // 更多寄存器可留白
    task->sp = (uint64_t)sp;
    task->state = TASK_READY;
    task->next = NULL;
    strcpy(task->name, name);
}

3.5 调度器核心：任务切换

调度器使用一个全局 current_task 数组记录每个核心当前运行的任务。

tcb_t *current_task[NR_CPUS];

void scheduler_enter(void) {
    int core = get_core_id();
    while (1) {
        tcb_t *next = NULL;
        spin_lock(&ready_lock);
        if (ready_queue) {
            next = ready_queue;
            ready_queue = ready_queue->next;
            if (ready_queue == next) ready_queue = NULL; // 只有自己
            else next->next = NULL;
        }
        spin_unlock(&ready_lock);
        
        if (next == NULL) {
            // 无任务可运行，进入低功耗等待（wfi）
            __asm__ volatile("wfi");
            continue;
        }
        
        next->state = TASK_RUNNING;
        next->core_id = core;
        tcb_t *prev = current_task[core];
        current_task[core] = next;
        
        if (prev == NULL) {
            // 首次切换，直接进入任务
            switch_to_first(next);
        } else {
            // 保存当前任务上下文，再恢复 next 的上下文
            switch_to(prev, next);
        }
        // 当任务再次让出 CPU 时会回到这里，然后循环重新取任务
    }
}

switch_to 用汇编实现（省略，完整代码见文末链接）。

3.6 任务主动让出：yield()

void yield(void) {
    int core = get_core_id();
    tcb_t *self = current_task[core];
    if (self == NULL) return;
    
    spin_lock(&ready_lock);
    // 将自己放回就绪队列尾部
    self->state = TASK_READY;
    self->core_id = -1;
    if (ready_queue == NULL) {
        ready_queue = self;
        self->next = self;
    } else {
        self->next = ready_queue->next;
        ready_queue->next = self;
        ready_queue = self;          // 移动到尾部
    }
    spin_unlock(&ready_lock);
    
    // 触发重新调度（会切换到队列中的下一个任务）
    // 注意：我们仍在当前任务的上下文中，需要跳出到调度器
    // 最简单的方法：直接长跳转到 scheduler_enter 的调度循环起点
    // 实际使用 setjmp/longjmp 风格或内联汇编保存/恢复
    // 这里简化：调用 schedule() 函数（内部会做切换）
    schedule();
}

schedule() 内部会从就绪队列取下一个任务，并与当前任务交换上下文。

3.7 任务示例：打印 Hello World

void task_a(void *arg) {
    char *msg = (char *)arg;
    int core = get_core_id();
    while (1) {
        spin_lock(&uart_lock);   // 保护串口打印
        printf("Core %d: %s\n", core, msg);
        spin_unlock(&uart_lock);
        for (volatile int i = 0; i < 1000000; i++); // 模拟工作
        yield();                 // 主动让出 CPU
    }
}

我们在 main 中创建两个任务，分别传入 “Hello from Task A” 和 “Hello from Task B”。

---

4. 运行与实验

4.1 编译与模拟

riscv64-unknown-elf-gcc -march=rv64gc -mabi=lp64 -static -o kernel.elf start.s main.c
qemu-system-riscv64 -machine virt -smp 2 -nographic -kernel kernel.elf

你应该能看到两个核心交替输出。如果没有看到，说明调度或上下文切换有 bug。

4.2 伪共享的威力

修改任务创建，让两个任务共用同一个缓存行上的两个不同计数器。对比有/无缓存对齐的性能差异。你会发现：对齐后吞吐量明显提升。

4.3 增加自旋锁争用测试

在 yield 前后打印时间戳，观察当两个核心同时争用 ready_lock 时的自旋次数（可在 spin_lock 中插入计数器）。

---

5. 你可能遇到的坑与解决方法

现象	可能原因	解决
只有一个核心在工作，另一个永远空闲	AP 未正确进入调度循环，或就绪队列被 BSP 独占	检查 AP 的启动路径，确保它调用了 scheduler_enter
输出乱码或丢失字符	串口打印没有加锁	用自旋锁保护 printf 的整个调用（或者用原子 putchar）
偶发死锁，系统卡住	自旋锁缺少内存屏障，或上下文切换破坏了锁状态	在 spin_lock/unlock 加入 fence；切换任务前释放锁
任务切换后栈溢出	TCB 内嵌栈太小，或栈指针计算错误	增大 STACK_SIZE，检查 sp 初始值
两个任务输出完全相同的 core id	get_core_id() 实现错误	正确读取 mhartid，或者用全局变量传递

---

6. 展望：从迷你内核到真正的多核 OS

我们的双核 Hello World 调度器只有几百行代码，但你已经掌握了现代多核 OS 的核心思想：

• 每个核心独立调度 + 全局就绪队列 是最简单的多核模型，但会引入锁争用。
• 原子操作 + 内存屏障 是正确实现锁的基石。
• 缓存一致性 硬件帮你做，但你需要用对齐和填充避免伪共享。

如果要扩展成一个实用的 RTOS 或通用 OS，还需要实现：

• 每核本地就绪队列：减少锁争用，配合负载均衡算法（如工作窃取）。
• 更丰富的同步原语：在内核中使用自旋锁，用户态使用 futex。
• CPU 亲和性与隔离：绑定关键任务到特定核心，避免缓存抖动。
• 中断管理与重调度：定时器中断触发抢占，IPI 实现跨核重调度请求。
• 内存管理：多核下的页表同步与 TLB 无效化（需要 IPI）。

此外，现代多核系统还引入 NUMA（非统一内存访问）和 SMT（超线程），它们的调度和缓存一致性更加复杂，但根源仍是我们学过的 MESI + 内存屏障。

---

7. 专栏结语

从第 1 讲定时器驱动的“假并发”，到第 10 讲亲手写出的双核调度器，我们走过了操作系统内核最动人的一段进化史。
希望这个专栏能让你真正理解：并发不是魔法，而是一套环环相扣的软硬件契约 —— 从硬件定时器、缓存、中断，到 TCB、调度算法、信号量、屏障，每一层都在为“同时做好多件事”这个朴素愿望服务。

你现在的知识体系，已经足以去阅读 FreeRTOS、Zephyr、Linux 内核的多核调度代码。当你再遇到奇怪的并发 bug 时，你会想起 MESI 的四种状态，会想起那个被乱序执行颠倒的 flag 和 data。

代码即真理，动手是捷径。
祝你继续探索，写出属于自己的内核。

---

✍️ 最终思考与挑战

1. 完整代码仓库：将本讲的代码补充完整（上下文切换汇编、链接脚本、启动流程），放到 GitHub 上，并写一个 README 演示运行。
2. 增加优先级调度：修改就绪队列为每个优先级一个链表，实现固定优先级抢占（不要忘了自旋锁保护每个链表）。
3. 实现跨核心负载均衡：每隔 100ms，检查各核心的就绪任务数（本地队列模型），将任务从忙的核心迁到空闲核心。
4. 思考无锁队列：如果不用全局锁，用无锁的并发队列（如 spsc 或 mpsc）来管理就绪任务，会带来什么好处和挑战？

欢迎在评论区晒出你的双核 Hello World 运行截图，以及你为它增加的新特性！
十讲已毕，内核之路永无止境。感谢你的阅读。