为什么 Linux 2.6 的调度器叫 O(1)?queue[140]、bitmap[5]、nr_active 三个成员如何用一条 CPU 指令找到最高优先级进程?这篇文章带你从源码级彻底搞懂。文末附命令行参数(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 个 queue0O(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 的分工

它们互补,不是冗余:

bitmapnr_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 至少是 1argv[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 条 cmpnr_active == 0
执行交换2 条赋值交换 active/expired 指针

一句话记住

优先级数字,本质就是数组的下标。根据优先级选择进程,本质是一个哈希的过程。一旦确定是哪个队列,剩下的就是 FIFO。


本文重点拆解 prio_array 数据结构的设计原理与源码实现,适合对操作系统内核调度机制感兴趣的开发者。

如果觉得有帮助,欢迎点赞收藏,评论区交流讨论!

Logo

DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐