Linux O(1) 调度器核心数据结构 prio_array 完全拆解 + 命令行参数原理
为什么 Linux 2.6 的调度器叫 O(1)?queue[140]、bitmap[5]、nr_active 三个成员如何用一条 CPU 指令找到最高优先级进程?这篇文章带你从源码级彻底搞懂。文末附命令行参数(argc/argv)的原理详解。
目录
- 第一部分:queue[140] —— 优先级队列数组
- 第二部分:bitmap[5] —— 优先级位图
- 第三部分:nr_active —— 就绪进程计数器
- 第四部分:三者协同 —— 一次完整调度流程
- 第五部分:prio_array 完整结构与 runqueue
- 第六部分:命令行参数 argc/argv 原理
- 总结
第一部分:queue[140] —— 优先级队列数组
1.1 queue[140] 是什么?
直接看 Linux 2.6 内核源码:
// include/linux/sched.h (Linux 2.6 早期)
#define MAX_USER_RT_PRIO 100 // 实时优先级:0~99,共 100 级
#define MAX_RT_PRIO MAX_USER_RT_PRIO
#define MAX_PRIO (MAX_RT_PRIO + 40) // 100 + 40 = 140
#define DEFAULT_PRIO (MAX_RT_PRIO + 20) // 100 + 20 = 120
struct prio_array {
unsigned int nr_active; // 数组内就绪进程总数
unsigned long bitmap[BITMAP_SIZE]; // 优先级位图(5 个 unsigned long)
struct list_head queue[MAX_PRIO]; // 140 个链表头(哨兵)
};
queue[140] 的本质:queue 是一个数组,每个元素不是进程,而是一个"哨兵"(链表头)。
queue[0] → 优先级 0(最高实时)的进程链表
queue[1] → 优先级 1 的进程链表
...
queue[99] → 优先级 99 的进程链表(最低实时)
────────────────── 分界线 ──────────────────
queue[100] → 优先级 100(nice=-20,最高普通)的进程链表
queue[101] → 优先级 101
...
queue[120] → 优先级 120(nice=0,默认)的进程链表
...
queue[139] → 优先级 139(nice=+19,最低普通)的进程链表
140 这个数字怎么来的?
实时优先级:0 ~ 99 = 100 个级别(POSIX 实时调度规范要求)
普通优先级:100 ~ 139 = 40 个级别(对应 nice 值 -20 ~ +19 的 40 个梯度)
100 + 40 = 140
重要澄清:queue[i] 是哨兵(链表头),不是进程指针。对于空队列,queue[i] 自己指自己(next == prev == &queue[i]),形成一个空环。
1.2 queue[i] 内部长什么样?
假设优先级 120 有 3 个就绪进程:
queue[120](哨兵)
┌──────────┐
│ next ────┼──→ task_A.run_list
│ prev ────┼──→ task_C.run_list
└──────────┘
task_A.run_list task_B.run_list task_C.run_list
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ next ────────┼──→ B │ next ────────┼──→ C │ next ────────┼──→ queue[120]
│ prev ────────┼──→ Q │ prev ────────┼──→ A │ prev ────────┼──→ B
└──────────────┘ └──────────────┘ └──────────────┘
关键:queue[i] 挂的是 task_struct 内部某个 list_head 成员(通常是 task_struct → sched_entity → run_node),不是挂在 task_struct 的起始地址。要通过 container_of() 才能从 list_head 找回 task_struct——这就是之前讲的侵入式链表的威力。
为什么队列是 FIFO? 入队时把新进程插在 queue[i].prev 前面(队尾),出队时取 queue[i].next(队头)。同优先级的进程按到达顺序轮流运行,实现时间片轮转(Round-Robin)。
1.3 为什么是数组 + 链表,而不是纯数组或纯链表?
这涉及三组权衡:
方案 A:纯数组(每个槽一个 task_struct*)
struct prio_array_bad {
struct task_struct *queue[140]; // 每个优先级只能存一个进程!
};
致命缺陷:同一优先级只能放一个进程。实际系统中 nice=0 可能有几十个进程,只留一个槽,其他全丢了。
方案 B:纯链表(所有优先级混在一起)
struct task_struct *all_tasks; // 每次选进程需要遍历全部,比较优先级字段 → O(n)
系统有 1 万个进程时走一遍链表就太慢了。
方案 C:数组 + 链表(Linux 实际方案)
- 数组提供按优先级的一级索引(O(1) 定位)
- 链表提供同一优先级内的二级存储(O(1) 入队/出队)
两者结合 = "哈希表"一样的思路:优先级当哈希键,链表解决冲突。
| 方案 | 定位目标优先级 | 取下一个进程 | 插入新进程 |
|---|---|---|---|
| 纯数组(一槽一进程) | O(1) | O(n) | O(1) |
| 纯链表(全混在一起) | O(n) | O(n) | O(1) |
| 数组+链表(Linux 实际) | O(1) | O(1)(bitmap 辅助) | O(1) |
1.4 queue[140] 的操作复杂度逐一分析
入队(enqueue_task):
static void enqueue_task(struct task_struct *p, struct prio_array *array) {
list_add_tail(&p->run_list, &array->queue[p->prio]); // ① 定位 + 入队 → O(1)
__set_bit(p->prio, array->bitmap); // ② 置位 → O(1)
array->nr_active++; // ③ 计数 → O(1)
}
// 总计:3 步,全部 O(1)
出队(dequeue_task):
static void dequeue_task(struct task_struct *p, struct prio_array *array) {
list_del(&p->run_list); // ① 摘除 → O(1)
if (list_empty(&array->queue[p->prio]))
__clear_bit(p->prio, array->bitmap); // ② 链表变空才清 bit → O(1)
array->nr_active--; // ③ 计数 → O(1)
}
// 总计:3 步,全部 O(1)
查找最高优先级(O(1) 的核心):
int idx = sched_find_first_bit(array->bitmap); // ① bsfl 指令 → O(1),约 1 个 CPU 周期!
struct task_struct *next = list_entry(
array->queue[idx].next, struct task_struct, run_list);// ② container_of → O(1)
第二部分:bitmap[5] —— 优先级位图
2.1 bitmap[5] 是什么?
#define BITMAP_SIZE 5 // 5 个 unsigned long
unsigned long bitmap[BITMAP_SIZE]; // 5 × 32 = 160 位,覆盖 140 个优先级
2.2 位图操作的三个核心函数
① sched_find_first_bit() —— 调度器的灵魂
// include/asm-i386/bitops.h (x86 32)
static inline int sched_find_first_bit(const unsigned long *b) {
if (unlikely(b[0])) return __ffs(b[0]); // bitmap[0] 有非零 → 在 0~31 内找
if (unlikely(b[1])) return __ffs(b[1]) + 32; // bitmap[1] 有非零 → 32~63
if (unlikely(b[2])) return __ffs(b[2]) + 64; // bitmap[2] 有非零 → 64~95
if (unlikely(b[3])) return __ffs(b[3]) + 96; // bitmap[3] 有非零 → 96~127
return __ffs(b[4]) + 128; // 最后一定在 bitmap[4]
}
__ffs(find first set bit)的真身:
static inline int __ffs(unsigned long word) {
int result;
__asm__("bsfl %1, %0" : "=r"(result) : "r"(word));
return result;
}
bsfl 指令干了什么?
bitmap[0] = 0b00000000_00000001_00000000_00000000
bit31 ←──────────────────────→ bit0
bsfl(bitmap[0]) → 8(bit8 是第一个为 1 的位)
这对应 queue[8] 有就绪进程。
为什么是 forward(从低到高)?
bit 编号越小 = 优先级越高 = 越应该先运行
bsfl 从 bit0 开始扫,碰到的第一个 1 = 最高优先级的非空队列
——恰好就是我们要的结果。设计者和 CPU 架构师想到一块去了。
没有 bsfl 怎么办? 某些非 x86 架构没有 bsfl,Linux 提供 fallback 二分查找:
static inline int generic_ffs(unsigned long word) {
int r = 1;
if (!word) return 0;
if (!(word & 0xffff)) { word >>= 16; r += 16; }
if (!(word & 0xff)) { word >>= 8; r += 8; }
if (!(word & 0xf)) { word >>= 4; r += 4; }
if (!(word & 0x3)) { word >>= 2; r += 2; }
if (!(word & 0x1)) { r += 1; }
return r;
}
// 二分查找,依然是 O(1)(常数次 if),只是比 bsfl 多几个周期
② __set_bit() —— 入队时置位
static inline void __set_bit(int nr, volatile unsigned long *addr) {
__asm__("btsl %1, %0" : "+m"(*addr) : "Ir"(nr) : "memory");
// btsl = Bit Test and Set Long
// 把 addr 指向的 32 位值中第 nr 位 置 1
// 这是一条原子指令!多核下不需要额外加锁
}
// 调用:__set_bit(p->prio, array->bitmap);
// 例如进程优先级 120:
// nr = 120 → 120 / 32 = 3(在 bitmap[3] 里)
// → 120 % 32 = 24(在 bitmap[3] 的第 24 位)
// btsl 指令 → bitmap[3] 的 bit24 置 1
一条 CPU 指令完成,不需要关中断、不需要自旋锁——硬件保证了原子性。
③ __clear_bit() —— 队列变空时清除
static inline void __clear_bit(int nr, volatile unsigned long *addr) {
__asm__("btrl %1, %0" : "+m"(*addr) : "Ir"(nr) : "memory");
// btrl = Bit Test and Reset Long
// 把 addr 的第 nr 位 清 0
}
// 有条件的 clear——只有该优先级链表变空时才清:
if (list_empty(&array->queue[p->prio]))
__clear_bit(p->prio, array->bitmap);
为什么空的时候才清? 一个优先级下可能有多个进程,摘走一个还剩,bit 应该继续标记"这个优先级还有人"。只有链表彻底变空,才把这个优先级的"有人"标志摘掉。
2.3 为什么是位图,而不是其他数据结构?
| 方案 | 内存开销 | 找最高优先级 | 入队 | 出队 |
|---|---|---|---|---|
| 遍历 140 个 queue | 0 | O(140),~140 次循环 | O(1) | O(1) |
| 最小堆 | ~几十字节 | O(log n),~7 次比较 | O(log n) | O(log n) |
| 位图 | 20 字节 | bsfl,~1 周期 | btsl,1 指令 | btrl,1 指令 |
2.4 bsfl 指令的硬件级原理
这条指令是为编译器和操作系统设计者准备的"礼物"。
bsf = Bit Scan Forward
输入:一个 32 位(或 64 位)的通用寄存器值
输出:第一个为 1 的 bit 的位置编号(从 0 开始)
如果全为 0,则 ZF 标志位置 1
内部实现(硬件层面):CPU 内部有一个"前导零检测器"(priority encoder),它可以在一个时钟周期内完成所有 bit 的并行检测——32 根线 → 组合逻辑电路 → 5 bit 二进制结果。
这不是"一个 bit 一个 bit 地扫",而是 32 个 bit 同时被看。
这和 bitmap 设计的默契:
"低 bit 编号 = 高优先级" + "bsfl 从低到高扫" = 一条指令得出最高非空优先级
2.5 位图的"跨字"逻辑
当 bitmap[0] 全为 0(高 32 个优先级都空),需要跳到下一个 unsigned long:
bitmap = [0x00000000, 0x00000000, 0x00000001, 0x00000000, ...]
bitmap[0] bitmap[1] bitmap[2] bitmap[3]
① if (b[0]) → 否,bitmap[0] 是 0,跳过
② if (b[1]) → 否,bitmap[1] 是 0,跳过
③ if (b[2]) → 是!__ffs(b[2]) = 0(bit 0 为 1)→ 返回值 = 0 + 64 = 64
→ queue[64] 有就绪进程,这是最高的非空优先级
最坏情况:5 次 if 判断 + 1 次 bsfl = 约 6 个 CPU 周期。进程数量再多也不变——这就是 O(1) 的真正含义。
第三部分:nr_active —— 就绪进程计数器
3.1 nr_active 是什么?
struct prio_array {
unsigned int nr_active; // 这个数组里一共有多少个就绪进程
unsigned long bitmap[5];
struct list_head queue[140];
};
生命周期:
进程创建/唤醒 → 入队 → nr_active++
进程选中运行 → 出队 → nr_active--
进程时间片用完 → 出队(active) → 入队(expired) → 各自 nr_active 增减
进程退出 → 出队 → nr_active--
3.2 没有它行不行?
没有 nr_active 的世界:
bool is_active_empty(struct prio_array *array) {
for (int i = 0; i < 140; i++) { // 每次调度都要把 140 个链表全扫一遍!
if (!list_empty(&array->queue[i]))
return false;
}
return true;
}
// → O(140),每次都走 140 次循环
有 nr_active 的世界:
bool is_active_empty(struct prio_array *array) {
return array->nr_active == 0; // 一条 cmp 指令!
}
// → O(1)
nr_active == 0 是 active↔expired 交换的触发条件。每次调度都要检查,所以这个检查必须 O(1)——nr_active 就是这个目的。
3.3 nr_active 和 bitmap 的分工
它们互补,不是冗余:
| bitmap | nr_active | |
|---|---|---|
| 功能 | 定位非空优先级 | 计数就绪进程总数 |
| 更新 | set_bit / clear_bit(有条件) | ++ / --(每次出入队) |
| 使用 | bsfl 找第一个非空队列 | == 0 判断是否触发交换 |
| 如果只有它 | 无法判断 active 是否全空 | 找不到具体哪个优先级有进程 |
第四部分:三者协同 —— 一次完整调度流程
4.1 入队全流程(enqueue)
enqueue_task(task, array)
│
├── ① queue[task->prio] 定位目标优先级队列 → O(1)
├── ② list_add_tail → 挂到该链表末尾 → O(1)
├── ③ __set_bit(task->prio, array->bitmap) → O(1), btsl 指令
└── ④ array->nr_active++ → O(1)
四步走完,进程正式进入"可被调度"状态
4.2 出队全流程(dequeue)
dequeue_task(task, array)
│
├── ① list_del → 从链表中摘除 → O(1)
├── ② if (list_empty(queue[task->prio]))
│ __clear_bit(task->prio, array->bitmap) → O(1), btrl 指令
└── ③ array->nr_active-- → O(1)
4.3 active↔expired 交换全流程
场景:进程 A(prio=120)在 CPU 上跑,时间片用完了。
asmlinkage void __sched schedule(void) {
struct task_struct *prev, *next;
// ...
// ① 如果当前进程还在运行态,移入 expired
if (prev->state == TASK_RUNNING) {
dequeue_task(prev, active); // active->nr_active--
enqueue_task(prev, expired); // expired->nr_active++
}
// ② 检查 active 是否取空
if (active->nr_active == 0) {
// 交换指针(不是复制数据!)
struct prio_array *tmp = active;
rq->active = expired;
rq->expired = tmp;
}
// ③ 从 active 取最高优先级进程
int idx = sched_find_first_bit(active->bitmap); // bsfl 指令
next = list_entry(active->queue[idx].next, ...); // container_of
dequeue_task(next, active); // active->nr_active--
// ④ 上下文切换到 next
}
关键洞察:交换两个指针而不是复制整个数组。prio_array 包含 140 个链表头 + 5 个 unsigned long + 1 个计数器——交换指针只需两条赋值,这也是 O(1) 的一部分。
4.4 为什么只在 nr_active == 0 时判断一次?
逻辑很简单:只要 nr_active > 0,active 就还有库存,直接从 active 取下一个。只有库存归零时才交换。
就像从冰箱拿东西——拿一次看一眼"还有没有",全拿完了才补货,而不是每拿一个就去超市买一趟。
4.5 追踪一个进程的一生
进程 X(nice=0, prio=120)的一生:
创建 → fork():
enqueue_task(X, active)
→ queue[120].next = X.run_list
→ bitmap bit120 置 1
→ active->nr_active 变为 N+1
就绪排队 → 同优先级另有 2 个进程在它前面,FIFO 等待
被调度 → schedule():
sched_find_first_bit → 120
X = queue[120].next
dequeue_task(X, active) → active->nr_active 变为 N
运行 → 100ms 时间片用完 → schedule() 又调进来:
dequeue_task(X, active) → active->nr_active 再 -1
enqueue_task(X, expired) → expired->nr_active +1
→ expired->queue[120] 里有 X
→ expired->bitmap bit120 置 1
轮回 → 当 active 取空:
swap(active, expired)
→ X 又回到 active!
→ X 重新获得 100ms 时间片,等待被调度
退出 → exit():
dequeue_task(X, active/expired)
→ nr_active 相应 -1
→ 如果 queue[120] 变空 → clear_bit 120
一个进程在 active 中执行过它的时间片之后,放入 expired 中。在放入之前,重新计算时间片 + 动态优先级(根据 nice 值调整优先级就发生在这个时候)。直到 active 的 nr_active 为 0 后,交换 active 和 expired,进程重新等待被执行。
4.6 为什么需要两个数组(active + expired)?
场景:没有 expired,只有 active。
进程 A 时间片用完 → 移入 active 末尾 → 马上又能被调度到!
结果:可能 A 立刻又被选出来跑(如果它优先级最高)
→ 其他同优先级的进程可能饿死
有了 expired:
进程 A 时间片用完 → 移入 expired → 它进"隔离区"
→ 只有等 active 全部取空、交换后,A 才能回来
→ 在 active 清空之前,其他同优先级进程有机会运行
两个数组的目的 = 严格的时间片隔离。当前时间片周期内的所有进程轮完,才开始下一轮。
第五部分:prio_array 完整结构与 runqueue
5.1 源码真身
// kernel/sched.c (Linux 2.6.0 ~ 2.6.22)
// 简化版,保留核心结构
#define MAX_PRIO 140
#define BITMAP_SIZE ((((MAX_PRIO + 7) / 8) + sizeof(long) - 1) / sizeof(long))
// (140 + 7) / 8 = 18 字节 → 对齐到 unsigned long → 5 个(32位)
struct prio_array {
unsigned int nr_active; // 就绪进程数
unsigned long bitmap[BITMAP_SIZE]; // 优先级位图(5 或 3 个 unsigned long)
struct list_head queue[MAX_PRIO]; // 140 个链表头
};
// 全局运行队列(per-CPU)
struct runqueue {
spinlock_t lock; // 自旋锁保护
unsigned long nr_running; // 该 CPU 上所有就绪进程
unsigned long nr_switches; // 上下文切换次数
unsigned long nr_uninterruptible; // 不可中断的进程数
struct task_struct *curr; // 当前正在运行的进程
struct prio_array *active; // 指着活跃队列
struct prio_array *expired; // 指着过期队列
struct prio_array arrays[2]; // 实际存储:arrays[0] 和 arrays[1]
// ...
};
5.2 为什么 active 和 expired 各有一个 prio_array?
active 和 expired 不是两个独立的全局变量,而是两个指针,分别指向 runqueue 里的 arrays[0] 和 arrays[1]。
交换时只交换两个指针的指向:
交换前: active → arrays[0], expired → arrays[1]
交换后: active → arrays[1], expired → arrays[0]
这就是 O(1) 调度器的优雅之处——不复制数据,只交换指针。
第六部分:命令行参数 argc/argv 原理
6.1 为什么要有命令行参数?
命令行参数的本质应用:让一个命令根据不同的选项实现不同的子功能。这也是 Linux 中所有命令选项功能的实现方式。
ls -l -a /home
grep -r "hello" ./src
gcc -o main main.c
6.2 三个关键细节
| 细节 | 说明 |
|---|---|
| argc 至少是 1 | argv[0] 一定指向程序名,所以 argc > 1 才有额外参数 |
| 选项是字符串 | 以空格分隔,"一个字符也是字符串"(-l 是字符串 "-l") |
| NULL 哨兵 | argv[argc] == NULL,这是标准规定的哨兵 |
int main(int argc, char *argv[]) {
return 0;
}
6.3 char *argv[] 到底是什么?
int main(int argc, char *argv[])
// ~~~~ ~~~~~~~~
// | └── char* [] = char** = 指向"指向char的指针"的指针
// └── int = 有多少个字符串
// 等价写法:
int main(int argc, char **argv) // 完全等价
int main(int argc, char *argv[]) // 语法糖,编译器眼中就是 char**
argv 是一个指针,指向一个指针数组的第一个元素。这个数组里的每个元素(char*)又指向一个 C 字符串。
argv (char**)
│
│ 值 = 0x7fff1000(指针数组的起始地址)
│
▼
┌──────────────────┐
│ argv[0] = 0x2000 │ ──→ "./a.out\0"
│ argv[1] = 0x2008 │ ──→ "hello\0"
│ argv[2] = 0x200e │ ──→ "world\0"
│ argv[3] = NULL │ ──→ (哨兵,argc=3)
└──────────────────┘
地址:0x7fff1000
6.4 两种遍历方式
#include <stdio.h>
int main(int argc, char *argv[]) {
// ===== 方式一:用 argc 计数遍历 =====
printf("=== 方式一:argc 索引 ===\n");
for (int i = 0; i < argc; i++) {
printf("argv[%d] = \"%s\"\n", i, argv[i]);
}
// ===== 方式二:利用 NULL 哨兵遍历(不需要 argc) =====
printf("\n=== 方式二:NULL 哨兵 ===\n");
for (char **p = argv; *p != NULL; p++) {
printf("argv[%td] = \"%s\"\n", p - argv, *p);
}
// argv[argc] 一定是 NULL,这就是哨兵
printf("\nargv[argc] == NULL ? %s\n",
argv[argc] == NULL ? "是" : "否");
return 0;
}
运行 ./a.out hello world 输出:
=== 方式一:argc 索引 ===
argv[0] = "./a.out"
argv[1] = "hello"
argv[2] = "world"
=== 方式二:NULL 哨兵 ===
argv[0] = "./a.out"
argv[1] = "hello"
argv[2] = "world"
argv[argc] == NULL ? 是
6.5 两种方式对比
argc 索引 argv[i] | NULL 哨兵 *p != NULL | |
|---|---|---|
| 需要 argc | 需要 | 不需要 |
| 安全边界 | argc 保证不越界 | argv[argc] == NULL 保证不越界 |
| 代码风格 | 更常见 | 更"指针味",更贴近底层 |
| 适用场景 | 需要知道位置 i 时 | 只需遍历每个值时 |
| 通用性 | 仅用于 argv(argc 已知) | 通用模式,任何 NULL-terminated 指针数组都能用 |
总结
O(1) 调度器三大核心数据结构的协作关系
┌─────────────────────────────────────────────────┐
│ prio_array │
│ │
│ nr_active ←── "还有多少进程?"(一条 cmp 判断交换)│
│ bitmap[5] ←── "哪个优先级有进程?"(一条 bsfl 定位)│
│ queue[140] ←── "具体是哪些进程?"(FIFO 链表轮转) │
└─────────────────────────────────────────────────┘
为什么叫 O(1)?
无论系统中有多少进程(10 个还是 10 万个),以下操作的时间都是恒定的:
| 操作 | 耗时 | 手段 |
|---|---|---|
| 找到最高优先级 | ~1 CPU 周期 | bsfl 指令查位图 |
| 入队 | 3 步固定操作 | 数组下标直接定位 + btsl |
| 出队 | 3 步固定操作 | 链表摘除 + btrl |
| 判断是否交换 | 1 条 cmp | nr_active == 0 |
| 执行交换 | 2 条赋值 | 交换 active/expired 指针 |
一句话记住
优先级数字,本质就是数组的下标。根据优先级选择进程,本质是一个哈希的过程。一旦确定是哪个队列,剩下的就是 FIFO。
本文重点拆解 prio_array 数据结构的设计原理与源码实现,适合对操作系统内核调度机制感兴趣的开发者。
如果觉得有帮助,欢迎点赞收藏,评论区交流讨论!
DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。
更多推荐
所有评论(0)