背景XX 团队发现，当多个虚拟机（子机）部署在同一台 S6 标卡母机上时，即使 IO 压力未达到物理机瓶颈，多子机同时下发 IO 会导致性能下降。SPDK 侧定位发现，4 个核（线程）中仅有 2 个核处理大量 IO 请求，而另外 2 个核几乎空闲。问题的直接原因在 SPDK 单核性能达到瓶颈，最终导致子机 IO 性能劣化。问题现象和初步分析XX 团队使用的子机挂载的数据盘为 ​​2 队列盘​​，初步怀疑子机仅通过盘的第一个队列下发 IO，导致多个盘的 IO 集中到 SPDK 的单个核上。简化模型如下：​​SPDK 使用 2 个核​​（CPU 0 和 CPU 1）。​​所有盘的第一个队列（mq0）均映射到 SPDK 的 CPU 0 ​​，第二个队列（mq1）映射到 CPU 1。​​多个子机的 IO 请求均通过 mq0 下发​​，导致 CPU 0 过载，而 CPU 1 空闲。Linux 多队列（Multi-Queue）机制​Linux Kernel 会为硬盘的每个队列（mq）各自分配一个 cpu_list，如下所示： mq0 绑定到 CPU 0-7，mq1 绑定到 CPU 8-15 。运行于指定 CPU 的进程提交 IO 请求就会发送到 CPU 关联的 mq 上。ubuntu@VM-16-202-ubuntu:~$ cat /sys/block/vda/mq/0/cpu_list 0, 1, 2, 3, 4, 5, 6, 7 ubuntu@VM-16-202-ubuntu:~$ cat /sys/block/vda/mq/1/cpu_list 8, 9, 10, 11, 12, 13, 14, 15因此，首先解决 IO 请求不均问题的方案是尝试将 XX 团队的子机中的一半业务进程/线程绑定到 mq1 对应的 CPU 上。然而，在 XX 团队的子机上查看 cpu_list 却得到了以下结果：只有 CPU 1 分配给了 mq1 。这明显不符合预期，并且在这种情况下也不可能通过将多个线程绑定到单个 CPU1 上来解决问题。[root@VM-0-3-centos ~]# cat /sys/block/vda/mq/0/cpu_list 0, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15 [root@VM-0-3-centos ~]# cat /sys/block/vda/mq/1/cpu_list 1问题深入定位​不同母机机型的影响很自然的，我们首先怀疑是 Centos 子机内核有 bug，但中间有一个小插曲：XX 团队的同学发现在 S5 母机上的子机，mq 对应的 cpu_list 却又是均匀的，如下所示。（这里有一个隐藏点，CentOS 上的 mq 对应的 cpu_list 和我前面贴的 ubuntu 子机的 cpu_list 不一样， CentOS 的某个 mq 对应的 cpu 是间隔的，ubuntu 的某个 mq 对应的 cpu 是连续的。）[root@VM-0-19-centos ~]# cat /sys/block/vda/mq/0/cpu_list 0, 2, 4, 6, 8, 10, 12, 14 [root@VM-0-19-centos ~]# cat /sys/block/vda/mq/1/cpu_list 1, 3, 5, 7, 9, 11, 13, 15不同的母机机型会导致相同的子机内核产生不同的 cpu_list 分配逻辑，这是一个很有意思的点，难道和虚拟化还相关？从这一点出发，对比两种机型上的子机的 xml 配置，发现了一个可疑的点：两者对子机 CPU 拓扑结构配置不同，同样 16 核的子机，S6 机型配置了超线程。进一步，通过在 S6 机型上将配置从 8 核 2 超线程模式修改成 16 独立核模式，重启子机后发现盘队列的 cpu_list 也就变均匀了。# S6 机型 <topology sockets='1' cores='8' threads='2'/> # S5 机型 <topology sockets='1' cores='16' threads='1'/>这表明 ​​问题与超线程（Hyper-Threading）相关​​，而非虚拟化（QEMU）本身的问题。进一步将 bug 范围缩小到子机内核为硬盘队列映射 cpu 的处理函数中。Linux Kernel 多队列映射逻辑分析​CentOS 7 的长期支持（LTS）内核基于 Linux 3.10，而 Multi-Queue（blk-mq）特性在 3.13 版本引入。对比 ​​Linux Upstream 3.13​​ 和 ​​CentOS 定制内核​​ 为盘队列映射 cpu 的处理函数实现，发现了关键差异。​Linux Kernel Upstream 3.13 逻辑首先可以看到blk_mq_update_queue_map() 函数中确实感知了超线程，一开始就怀疑是超线程处理逻辑有问题，该函数的主要逻辑如下：遍历所有 online 的 cpu，并统计所有 CPU （nr_cpus）和非超线程 CPU （nr_uniq_cpus）的数量；map 数组用于记录 cpu 核到队列的映射；若 ​​队列数 > CPU 数​​ 或 ​​无超线程​​，则直接将 CPU 按顺序映射到盘的队列上；当存在超线程核时，使用 nr_uniq_cpus 作为参数去计算 CPU 到队列号的映射关系，并且将同一个超线程的 cpu 映射到第一个超线程核已经映射的队列号上。例如，CPU 0 和 CPU 1 互为超线程，首先 CPU 0 会通过 cpu_to_queue_index计算出来映射到 mq0，继续循环到 CPU 1，发现该 CPU 对应于超线程 CPU 0，因此执行map[i] = map[first_sibling]; 将当前 CPU 1 也映射到 mq0；在详细分析 blk_mq_update_queue_map 函数的逻辑后，我们发现就算存在超线程，CPU 在多队列间应该也是均衡的。并且在不存在超线程的情况下，最后的结果应该也不是 CentOS 上 cpu_list 分配的表现，也应该是前 8 个 CPU 映射到 mq0，后 8 个 CPU 映射到 mq1（这里回到了我们最前面的发现，正常情况下，ubuntu 和 CentOS 不同的 cpu 分配表现）。CentOS 定制内核逻辑​那么只有一种可能，CentOS 没有采用 Linux upstream queue map 的方法。我在 Github 上找到了以下代码，最后找 XX 团队的同学确认，他们使用的内核源码中的函数实现确实和这段代码一致：首先为每个队列分配一个 CPU；然后感知超线程，将线程核分配给同一个队列；这段代码有一个明显的问题，在盘的队列数为 2 时（简化一下条件），并且当 CPU 按顺序互为超线程时（如 CPU 0/1、CPU 2/3…），除前 2 个 CPU 外，其余 CPU 均被映射到同一队列（mq0），导致 mq1 仅绑定到单个 CPU（如 CPU 1）。 static int cpu_to_queue_index(unsigned int nr_queues, const int cpu, const struct cpumask *online_mask) { // .... return cpu % nr_queues; } int blk_mq_map_queues(struct blk_mq_tag_set *set) { // ... for_each_possible_cpu(cpu) { /* * First do sequential mapping between CPUs and queues. * In case we still have CPUs to map, and we have some number of * threads per cores then map sibling threads to the same queue for * performace optimizations. */ if (cpu < nr_queues) { map[cpu] = cpu_to_queue_index(nr_queues, cpu, online_mask); } else { first_sibling = get_first_sibling(cpu); if (first_sibling == cpu) map[cpu] = cpu_to_queue_index(nr_queues, cpu, online_mask); else map[cpu] = map[first_sibling]; } } return 0; }解决方案​​问题基本定位，主要问题在于 CentOS kernel 感知了超线程，但是又没有像 upstream 代码感知超线程核数量来建立 CPU 到队列的映射。​​短期规避​​在 S6 机型上，将子机配置改为 ​​独立核模式​​（threads='1'），避免超线程影响。​​长期修复​​修改 CentOS 内核的 blk_mq_map_queues()，使其与 Linux Upstream 逻辑一致，​​基于 nr_uniq_cpus 计算队列映射​​，确保超线程核均衡分布。总结从多盘性能问题定位到 SPDK 核间压力不均，再子机中盘队列映射的 CPU 异常；从怀疑虚拟化的影响，到进一步分析 Linux Kernel 感知超线程核的优化，为多队列盘分配 CPU 的逻辑；最终定位到 CentOS 内核在某些超线程场景下多队列盘 CPU 映射异常的原因。
引言CPU 热插拔[63] 是 Linux 中一种广泛使用于支持处理器集缩放的机制。它们被设计用于两种用途：移除出现故障的 CPU 或动态增加 CPU 计算资源和动态更改虚拟机配置的 CPU 数量。现代先进的系统架构已经引进了具有先进的错误报告和纠错能力的处理器。有一些原始设备生产商（Original Equipment Manufacturer，OEM）也支持可热插拔的NUMA硬件，其中物理节点的插入与移除要求 CPU 热插拔的支持。这样的升级需要删除 kernel 当前可用的 CPU，要么是出于配置（Provisioning）的原因，要么是出于RAS（Reliability、Availability、Serviceability）的目的，以使有问题的 CPU 远离系统执行路径。因此，Linux内核需要CPU热插拔支持。CPU热插拔支持在系统启动后，关闭任意一个secondary cpu（在 ARM64 架构下，CPU0 为boot cpu，不能被关闭），并在需要时重新启动它。例如，在内核选项CONFIG_HOTPLUG_CPU使能的前提下，通过如下命令可以将 CPUx 移除（offline）或加入（online）到当前内核的管理下。echo 0 > /sys/devices/system/cpu/cpu1/online echo 1 > /sys/devices/system/cpu/cpu1/online毕业设计涉及到 Linux 内核 CPU 热插拔的实现修改，所以当时在 CPU 热插拔花了许多时间，现将基本的知识总结一下。CPU 热插拔的基本功能具体来说，CPU 热插拔需要完成以下基本功能。 状态迁移。当 CPU 被移除时，与该 CPU 相关的重要软件状态必须迁移到另一个活跃的 CPU 上。软件状态通常包括软中断以及 CPU 运行队列中的线程。系统中的软中断和当前 CPU 上运行的线程都保存在 per-CPU 队列中， 软中断需要被移动到其他活跃的 CPU 以进行处理，运行队列中的线程也会同步 迁移以避免运行程序的性能下降。但在 per-CPU 的实现下，每个 CPU 都存在一份副本，当发生 CPU 热插拔时就需要为目标 CPU 重新分配或释放 per-CPU 结构 实例，这一定程度上增大了 CPU 热插拔的时间开销。硬件管理与中断迁移。在 CPU 热插拔事件期间，需要禁用或启用某些硬件 相关功能。例如，CPU 进入离线状态前必须禁用机器检查（例如，停止看门狗线程的执行），因为离线 CPU 中的任何故障都不应影响其余系统。同时，当 CPU 被移除时，关联到当前 CPU 的中断需要被移动到其他活跃的 CPU 上进行处理以 保证系统的正常运行。 系统状态更新。关键全局变量的更新。例如，Linux 维护着一些重要的 CPU 状态全局位掩码，它们在系统中被频繁访问。这些掩码应在缩放事件期间 立即更新，以确保系统正常运行。系统文件系统（sysfs）的更新。sysfs 和 procfs 等易失性文件系统通过 Linux 中的虚拟文件系统接口提供对内核配置和获 取内核的状态。sysfs 中包含的基于 CPU 核的文件或目录节点在 CPU 集缩放期间 需要被删除或创建。内存管理。多数内核子系统在收到缩放事件通知后会释 放或分配内存结构，例如 per-CPU 变量或缓冲区队列。线程停放（Parking）。 新版本的 Linux 使用线程停放来无限期地挂起线程。在 CPU 缩放期间，系统线 程（例如看门狗线程）会被置为停放状态。CPU 的移除过程CPU 热插拔子系统为其他内核子系统提供接口来订阅处理器集更改的通知。当 CPU 热插拔事件发生时，其他子系统注册的通知函数将被调用从而进行相应的处理以保证系统在 CPU 移除或加入时系统能感知并维持正常的运行。通过读取 proc 文件/sys/devices/system/cpu/hotplug/states的内容可查看系统内所有注册的回调函数。tail /sys/devices/system/cpu/hotplug/states一次 CPU 移除过程需要经过 CPU_DOWN_PREPARE 、CPU_DYING 、CPU_DEAD 和 CPU_POST_DEAD 总共四个阶段。具体来说，CPU 热插拔过程由各个内核子系统注册的回调函数组成。当 CPU 移除事件发生时，各子系统注册的回调函数被调用来为系统 CPU 集的变化做相 应的操作。CPU 热插拔每个阶段对应多个步骤，每个步骤主要由子系统注册回 调函数构成。不同平台的硬件差异性和不同 Linux 内核版本可能导致系统内各子 系统注册的回调函数存在一些区别。在本文的实验环境下（硬件平台树莓派 4B 和 Linux 内核 5.10 版本），CPU 热插拔各阶段所对应的回调函数按执行顺序（从左至右再顺序向下）排列如下图所示。内核源码分析热插拔线程CPU 热插拔的关键实现源码分为两个部分：通用实现include/linux/cpuhotplug.hkernel/smpboot.ckernel/smpboot.hkernel/cpu.c处理器相关arch/arm64/kernel/smp.c前面提到，CPU 热插拔执行回调过程部分回调函数由待移除的目标 CPU 完成。该实现的原理是在系统启动时，内核调用cpuhp_threads_init()函数在每个 CPU 上都创建了一个热插拔线程。void __init cpuhp_threads_init(void) { BUG_ON(smpboot_register_percpu_thread(&cpuhp_threads)); kthread_unpark(this_cpu_read(cpuhp_state.thread)); }具体来说，smpboot_register_percpu_thread() 函数调用kthread_create_on_cpu 创建了名如cpuhp/%u的线程，如下所示：pi@raspberrypi:~ $ ps -ef | grep cpuhp root 15 2 0 Oct07 ? 00:00:00 [cpuhp/0] root 16 2 0 Oct07 ? 00:00:00 [cpuhp/1] root 21 2 0 Oct07 ? 00:00:00 [cpuhp/2] root 26 2 0 Oct07 ? 00:00:00 [cpuhp/3] CPU Down接下来分析一下 CPU offline 的过程。如上所示，我们可以通过在用户态写入 proc 文件/sys/devices/system/cpu/cpu1/online来触发 CPU offline 事件。对应的内核函数调用为cpu_down()->cpu_down_maps_locked()-> _cpu_down()。_cpu_down()的函数定义如下：首先唤醒对应 CPU 热插拔线程完成CPUHP_TEARDOWN_CPU之前的所有回调函数，通过设定热插拔 target 为CPUHP_TEARDOWN_CPU实现。CPU_DOWN_PREPARE和CPU_DYING 阶段对应的回调函数。若目标 CPU 执行发生错误就返回错误码；目标 CPU 执行其需要做的回调函数成功则继续；由当前 CPU 完成剩余的 CPU 热插拔的回调函数，CPU_DEAD 和 CPU_POST_DEAD 阶段对应的回调函数。static int __ref _cpu_down(unsigned int cpu, int tasks_frozen, enum cpuhp_state target) { struct cpuhp_cpu_state *st = per_cpu_ptr(&cpuhp_state, cpu); int prev_state, ret = 0; cpus_write_lock(); cpuhp_tasks_frozen = tasks_frozen; prev_state = cpuhp_set_state(st, target); // 如果当前状态在目标CPU待执行热插拔线程的范围内，我们就需要唤醒 CPU 热插拔线程 if (st->state > CPUHP_TEARDOWN_CPU) { st->target = max((int)target, CPUHP_TEARDOWN_CPU); ret = cpuhp_kick_ap_work(cpu); // 目标 CPU 执行发生错误的回滚，就返回错误码 if (ret) goto out; // 可能已经停止在 AP hotplug 线程范围内，即 target 设置为大于 CPUHP_TEARDOWN_CPU 步骤 if (st->state > CPUHP_TEARDOWN_CPU) goto out; st->target = target; } // 目标 CPU 将自己带到了 CPUHP_TEARDOWN_CPU 状态了，我们需要进一步cleanups ret = cpuhp_down_callbacks(cpu, st, target); if (ret && st->state == CPUHP_TEARDOWN_CPU && st->state < prev_state) { cpuhp_reset_state(st, prev_state); __cpuhp_kick_ap(st); } out: ... }Linux 内核 CPU 热插拔的实现中，前两个阶段（CPU_DOWN_PREPARE 和 CPU_DYING）由待移除的目标 CPU 完成，然后目标 CPU 进入到离线状态。系统中其他在线的 CPU 完成剩余的两个阶段来结束 CPU 移除的完整过程。参考链接CPU hotplug in the Kernel图解 CPU hot plug 流程CPU hotplug in the KernelLinux 内核 | CPU 状态管理Linux 内核 | Per CPU 变量
纪念自己第一次成为一个大开源项目的 contributor，尽管只是一个很小的PR。什么是CPU mask内核使用CPU mask 来记录 CPU 的状态，cpumasks 提供了代表系统中 CPU 集合的位图，一个 bit 的0/1代表了一个 CPU在每种cpu mask 时的状态。实现（Linux kernel 5.10.20）主要包含在以下文件中：include/linux/cpumask.hkernel/cpu.cCPU 的状态基本可分为以下四种：cpu_possible ：是一个在系统启动时任意时刻都可插入的 cpu ID 集合。它将等于通过CONFIG_NR_CPUS内核配置选项静态设置的NR_CPUS的值；cpu_present ：表示当前哪个CPU是插入的；cpu_online ： 是 cpu_present 的一个子集，表示哪些CPU 是调度器可访问的；cpu_active ： 该掩码的某些 bit 告诉Linux内核，一个任务可能已移至某个处理器。内核声明了结构体cpumask ，并实例化 4 个cpumask 分别用来记录 CPU 的各种状态。以__cpu_possible_mask为例，其定义如下代码所示(定义在文件 kernel/cpu.c 中)。其中__read_mostly原语将定义的变量存放在 .data.read_mostly 段中，其定义在/arch/arm64/include/asm/cache.h文件中，原型为#define __read_mostly __section(.data..read_mostly)，将经常需要被读取的数据定义为__read_mostly类型， 这样Linux内核被加载时,该数据将自动被存放到Cache中,以提高整个系统的执行效率。struct cpumask __cpu_possible_mask __read_mostly; EXPORT_SYMBOL(__cpu_possible_mask);cpumask的定义如下，DECLARE_BITMAP宏用来计算 cpumask所需的大小。NR_CPUS为内核支持的最大 CPU 数，在内核编译时 make menuconfig 指定。typedef struct cpumask { DECLARE_BITMAP(bits, NR_CPUS); } cpumask_t;DECLARE_BITMAP宏定义在include/linux/types.h文件中，其接收两个参数：name，位图名；bits，位图所使用的bit数量。其展开也就是定义了一个以 long为元素的数组。#define DECLARE_BITMAP(name,bits) \ unsigned long name[BITS_TO_LONGS(bits)]最后看一下BITS_TO_LONGS是如何根据 bits 计算出占用多少个 long 的。此宏定义在include/linux/bitops.h文件中，如下：#define BITS_TO_LONGS(nr) DIV_ROUND_UP(nr, BITS_PER_TYPE(long))DIV_ROUND_UP宏展开如下：// include/linux/kernel.h #define DIV_ROUND_UP __KERNEL_DIV_ROUND_UP // include/uapi/linux/const.h #define __KERNEL_DIV_ROUND_UP(n, d) (((n) + (d) - 1) / (d))假设NR_CPUS为8，long 占 8 个字节，cpumask结构体展开如下。也就是说，内核为每个 CPU 分配 long 变量中的一个 bit 来保存各个 CPU 的状态。 BITS_TO_LONGS(8) -> DIV_ROUND_UP(8, BITS_PER_TYPE(long)) -> DIV_ROUND_UP(8, 64) -> (((8) + (64) - 1) / (64)) -> 1 typedef struct cpumask { unsigned long bits[1] } cpumask_t; [注意！] 参考文档1按以下计算方式存在问题。若此时 CPU 数量是 9 个，计算得出需要 2 个 long 变量来保存CPU状态，这样就会很容易理解为一个 CPU的状态需要 1 个 btye 而不是 1 个 bit。 我向源Github仓库提交了 PR，不知道是否能合并进主线已经合并进了主线——首次成为一个大项目的 Contributor。(((8) + (8) - 1) / (8)) = 1 cpumask API对 CPU mask 的访问通过函数set_cpu_xxxx()完成，其接收两个参数（cpuid ，true or false）。当第二个参数为 true 时，set_cpu的功能将被调用；当为 false 时， clear_cpu 的功能将被调用。例如，在系统初始化时，boot CPU 的API 接口调用如下代码所示：// in kernel/cpu.c void __init boot_cpu_init(void) { int cpu = smp_processor_id(); set_cpu_online(cpu, true); set_cpu_active(cpu, true); set_cpu_present(cpu, true); set_cpu_possible(cpu, true); #ifdef CONFIG_SMP __boot_cpu_id = cpu; #endif }set_cpu_online()接收两个参数（cpuid ，true or false），其实现如下。若第二个参数为 true， cpumask_test_and_set_cpu将被调用。void set_cpu_online(unsigned int cpu, bool online) { if (online) { if (!cpumask_test_and_set_cpu(cpu, &__cpu_online_mask)) atomic_inc(&__num_online_cpus); } else { if (cpumask_test_and_clear_cpu(cpu, &__cpu_online_mask)) atomic_dec(&__num_online_cpus); } }cpumask_test_and_set_cpu作为内联函数定义在include/linux/cpumask.h文件中，接收CPU id cpu 以及相应的 CPU mask __cpu_online_mask。static inline int cpumask_test_and_set_cpu(int cpu, struct cpumask *cpumask) { return test_and_set_bit(cpumask_check(cpu), cpumask_bits(cpumask)); }在调用test_and_set_bit之前，有两个宏对 cpu 和 cpumask 进行了处理。cpumask_check是一个内联函数，检查传入的 cpu 参数是否大于了NR_CPUS，不展开cpumask_bits获取了 cpumask 的 bits，即上文叙述的unsigned long bits[1]的内容。#define cpumask_bits(maskp) ((maskp)->bits)test_and_set_bit()函数定义在include/asm-generic/bitops/atomic.h文件中，实现如下。BIT_MASK将 1 左移 nr 位获得 mask，再与 (maskp->bits)相与。若为 1 则返回直接返回1;不为 1 则需要修改 cpumask。因此调用atomic_long_fetch_or()函数将 cpumask 与 mask 相或置位，并返回原 cpumask 给 old 变量；old 变量与 mask 相与为0，经过两次 ！，最后依然返回0；static inline int test_and_set_bit(unsigned int nr, volatile unsigned long *p) { long old; unsigned long mask = BIT_MASK(nr); p += BIT_WORD(nr); if (READ_ONCE(*p) & mask) return 1; old = atomic_long_fetch_or(mask, (atomic_long_t *)p); return !!(old & mask); } #define BIT_MASK(nr) (UL(1) << ((nr) % BITS_PER_LONG)) #define BIT_WORD(nr) ((nr) / BITS_PER_LONG)其他关于 cpu mask 的 API 不一一详细例举，包括但不限于：for_each_cpu：遍历一个mask的所有 cpu;for_each_cpu_not：遍历所有补集的 cpu;cpumask_clear_cpu：清除一个 cpumask 的 cpu;cpumask_test_cpu：测试一个 mask 中的 cpu;cpumask_setall ：设置 mask 的所有 cpu;cpumask_size ： 返回分配 struct cpumask 字节数大小;参考文档# CPU masks
Linux 内核 | 内核启动 cmdline, early_param 和 initcall
本文主要记录 Linux kernel 获取 cmdline 、Linux early_param 启动参数解析以及 init_call 过程。基于ARM64架构，Linux kernel 5.10.20。cmdline 的初始化cmdline的初始化：start_kernel() -> setup_arch() 将 commad_line 指向 boot_command_line。很明显，这个变量将记录 cmdline。*cmdline_p = boot_command_line;但是内核从哪里获取 bootloader 传递的 cmdline 以及在哪里对boot_command_line初始化呢？boot_command_line 定义在 init/main.c 文件中 char __initdata boot_command_line[COMMAND_LINE_SIZE];。ARM64 架构下，设备树被来描述系统中所有的硬件信息，并在启动内核时向内核传递设备树文件（bootloader将设备树文件所在的地址保存在X0寄存器），内核解析设备树来对系统中存在的硬件进行初始化等操作。与 X86 架构传递 cmdline的方式不同（X86方式可参考链接[4]），ARM64架构下使用设备树传递 cmdline。设备树文件中的 chosen 节点通常包含许多信息，启动参数 cmdline 就包含在内，一个简单的例子如下：chosen { bootargs = "console=ttyS0,115200 loglevel=8"; initrd-start = <0xc8000000>; initrd-end = <0xc8200000>; };BTW：但是在使用树莓派时，我们可以通过修改/boot目录下的 cmdline.txt 文件直接修改树莓派4b的启动boot comand line。我估计是在树莓派的启动流程[2]中读取了cmdline.txt文件并修改了设备树的chosen节点，替换了启动参数。设备树解析在内核的启动流程中，early_init_dt_scan_nodes()函数解析设备树文件，提取chosen节点中bootargs保存到 boot_command_line 中。void __init early_init_dt_scan_nodes(void) { /* Retrieve various information from the /chosen node */ rc = of_scan_flat_dt(early_init_dt_scan_chosen, boot_command_line); ... } 函数调用关系总结函数调用关系如下：setup_arch() // arch/arm64/kernel/setup.c *cmdline_p = boot_command_line; ->setup_machine_fdt(__fdt_pointer) ->early_init_dt_scan() // driver/of/fdt.c ->early_init_dt_scan_nodes() -> early_init_dt_scan_chosen()Cmdline 参数解析与 early_param获取到 cmdline 后，Linux kernel 在 start_kernel() 函数中调用parse_early_param()对系统启动早期所需的配置参数进行解析并调用相应的回调。例如，系统启动时可能需要开启一个 earlycon 打印 log 或从设备树文件中获取指定 mem 大小的配置。parse_early_options()函数最终会调用do_early_param()函数。为了能在do_early_param()函数中统一处理所有需 early 配置的选项，Linux kernel 设计了类似注册回调的机制。我们可以看到，do_early_param()函数使用一个循环判断 cmdline 中是否存在 early option，若存在则调用其注册的回调函数p->setup_func(val)。void __init parse_early_options(char *cmdline) { parse_args("early options", cmdline, NULL, 0, 0, 0, NULL, do_early_param); } static int __init do_early_param(char *param, char *val, const char *unused, void *arg) { for (p = __setup_start; p < __setup_end; p++) { if ((p->early && parameq(param, p->str)) || (strcmp(param, "console") == 0 && strcmp(p->str, "earlycon") == 0) ) { if (p->setup_func(val) != 0) pr_warn("Malformed early option '%s'\n", param); } } return 0; }那么如何注册一个 early option 以及指定回调处理函数呢？Linux kernel 使用宏early_param（定义在文件 include/linux/init.h ，不具体展开）注册一个 early option 并指定回调。以早期内核从设备树文件获取内存大小为例，回调函数和注册都在 arch/arm64/mm/init.c 文件中，如下所示。/* * Limit the memory size that was specified via FDT. */ static int __init early_mem(char *p) { if (!p) return 1; memory_limit = memparse(p, &p) & PAGE_MASK; pr_notice("Memory limited to %lldMB\n", memory_limit >> 20); return 0; } early_param("mem", early_mem);添加新 cmdline 参数并解析仿照 quiet boot comandline 就很容易实现了。若需要添加自己的 cmdline 选项，只需要使用宏early_param注册，并实现对应的函数即可。如下所示（在文件 init/main.c 中）。static int __init quiet_kernel(char *str) { console_loglevel = CONSOLE_LOGLEVEL_QUIET; return 0; } early_param("quiet", quiet_kernel);init_call 机制early_param 相关的是 init_call，对应于在系统启动过程对系统进行初始化的过程。底层实现上，在内核镜像文件中，自定义一个段，这个段里面专门用来存放这些初始化函数的地址，内核启动时，只需要在这个段地址处取出函数指针，一个个执行即可。 Linux内核提供xxx_initcall_sync(fn)宏定义接口，驱动开发者只需要将驱动程序的 init_func 使用宏将驱动的初始化函数添加到了上述的段中即可，开发者完全不需要关心实现细节。 对于各种各样的驱动而言，可能存在一定的依赖关系，需要遵循先后顺序来进行初始化，考虑到这个，linux也对这一部分做了分级（Level）处理。如下所示，可选择以下宏来定义各 initcall 的level来达到被依赖的先被调用的目的。#define pure_initcall(fn) __define_initcall(fn, 0) #define core_initcall(fn) __define_initcall(fn, 1) #define core_initcall_sync(fn) __define_initcall(fn, 1s) #define postcore_initcall(fn) __define_initcall(fn, 2) #define postcore_initcall_sync(fn) __define_initcall(fn, 2s) #define arch_initcall(fn) __define_initcall(fn, 3) #define arch_initcall_sync(fn) __define_initcall(fn, 3s) #define subsys_initcall(fn) __define_initcall(fn, 4) #define subsys_initcall_sync(fn) __define_initcall(fn, 4s) #define fs_initcall(fn) __define_initcall(fn, 5) #define fs_initcall_sync(fn) __define_initcall(fn, 5s) #define rootfs_initcall(fn) __define_initcall(fn, rootfs) #define device_initcall(fn) __define_initcall(fn, 6) #define device_initcall_sync(fn) __define_initcall(fn, 6s) #define late_initcall(fn) __define_initcall(fn, 7) #define late_initcall_sync(fn) __define_initcall(fn, 7s)init_call 的函数调用路径如下：kernel_init() -> kernel_init_freeable() -> do_basic_setup() -> do_initcalls() -> do_initcall_level() -> do_one_initcall()参考资料[1]. how does the bootloader pass the kernel command line to the kernel?[2]. 树莓派启动那些事（四）[3]. elixir.bootlin[4]. Last preparations before the kernel entry point[5]. linux之early_param()和__setup[6]. linux的initcall机制(针对编译进内核的驱动)[7]. The initcall mechanism
简介在Linux中，伙伴分配器（buddy allocator）是以页为单位管理和分配内存。但在内核中的需求却以字节为单位（在内核中面临频繁的结构体内存分配问题）。假如我们需要动态申请一个内核结构体（占 20 字节），若仍然分配一页内存，这将严重浪费内存。那么该如何分配呢？slab 分配器专为小内存分配而生，由Sun公司的一个雇员Jeff Bonwick在Solaris 2.4中设计并实现。slab分配器分配内存以字节为单位，基于伙伴分配器的大内存进一步细分成小内存分配。换句话说，slab 分配器仍然从 Buddy 分配器中申请内存，之后自己对申请来的内存细分管理。除了提供小内存外，slab 分配器的第二个任务是维护常用对象的缓存。对于内核中使用的许多结构，初始化对象所需的时间可等于或超过为其分配空间的成本。当创建一个新的slab 时，许多对象将被打包到其中并使用构造函数（如果有）进行初始化。释放对象后，它会保持其初始化状态，这样可以快速分配对象。举例来说, 为管理与进程关联的文件系统数据, 内核必须经常生成struct fs_struct的新实例. 此类型实例占据的内存块同样需要经常回收(在进程结束时). 换句话说, 内核趋向于非常有规律地分配并释放大小为sizeof(fs_struct)的内存块. slab分配器将释放的内存块保存在一个内部列表中. 并不马上返回给伙伴系统. 在请求为该类对象分配一个新实例时, 会使用最近释放的内存块。S这有两个优点. 首先, 由于内核不必使用伙伴系统算法, 处理时间会变短. 其次, 由于该内存块仍然是”新”的，因此其仍然驻留在CPU硬件缓存的概率较高.[3]SLAB分配器的最后一项任务是提高CPU硬件缓存的利用率。 如果将对象包装到SLAB中后仍有剩余空间，则将剩余空间用于为SLAB着色。 SLAB着色是一种尝试使不同SLAB中的对象使用CPU硬件缓存中不同行的方案。 通过将对象放置在SLAB中的不同起始偏移处，对象可能会在CPU缓存中使用不同的行，从而有助于确保来自同一SLAB缓存的对象不太可能相互刷新。 通过这种方案，原本被浪费掉的空间可以实现一项新功能。通过命令sudo cat /proc/slabinfo可查看系统当前 slab 使用情况。以vm_area_struct结构体为例，当前系统已分配了 13014 个vm_area_struct缓存，每个大小为 216 字节，其中 active 的有 12392 个。[root@VM-8-9-centos]# cat /proc/slabinfo slabinfo - version: 2.1 # name <active_objs> <num_objs> <objsize> <objperslab> <pagesperslab> : tunables <limit> <batchcount> <sharedfactor> : slabdata <active_slabs> <num_slabs> <sharedavail> ... vm_area_struct 12392 13014 216 18 1 : tunables 0 0 0 : slabdata 723 723 0 mm_struct 184 200 1600 20 8 : tunables 0 0 0 : slabdata 10 10 0 shared_policy_node 5015 5015 48 85 1 : tunables 0 0 0 : slabdata 59 59 0 numa_policy 15 15 264 15 1 : tunables 0 0 0 : slabdata 1 1 0 radix_tree_node 15392 17234 584 14 2 : tunables 0 0 0 : slabdata 1231 1231 0 idr_layer_cache 240 240 2112 15 8 : tunables 0 0 0 : slabdata 16 16 0 kmalloc-8192 39 44 8192 4 8 : tunables 0 0 0 : slabdata 11 11 0 kmalloc-4096 117 144 4096 8 8 : tunables 0 0 0 : slabdata 18 18 0 kmalloc-2048 458 528 2048 16 8 : tunables 0 0 0 : slabdata 33 33 0 kmalloc-1024 1399 1424 1024 16 4 : tunables 0 0 0 : slabdata 89 89 0 kmalloc-512 763 800 512 16 2 : tunables 0 0 0 : slabdata 50 50 0 kmalloc-256 3132 3376 256 16 1 : tunables 0 0 0 : slabdata 211 211 0 kmalloc-192 2300 2352 192 21 1 : tunables 0 0 0 : slabdata 112 112 0 kmalloc-128 1376 1376 128 32 1 : tunables 0 0 0 : slabdata 43 43 0 kmalloc-96 1596 1596 96 42 1 : tunables 0 0 0 : slabdata 38 38 0 kmalloc-64 16632 20800 64 64 1 : tunables 0 0 0 : slabdata 325 325 0 kmalloc-32 1664 1664 32 128 1 : tunables 0 0 0 : slabdata 13 13 0 kmalloc-16 4608 4608 16 256 1 : tunables 0 0 0 : slabdata 18 18 0 kmalloc-8 4096 4096 8 512 1 : tunables 0 0 0 : slabdata 8 8 0 ...可以看到，系统中存在的 slab 有些形如 kmalloc-xxx 的 slab，我们称其为通用型 slab，用来满足分配通用内存。其他含有具体名字的 slab 我们称其为 专用 slab，用来为特定结构体分配内存，如 vm_area_struct、mm_struct 等。为什么要分专用和通用 slab ？ 最直观的一个原因就是通用 slab 会造成内存浪费：出于 slab 管理的方便，每个 slab 管理的对象大小都是一致的，当我们需要分配一个处于 64-96字节中间大小的对象时，就必须从保存 96 字节的 slab 中分配。而对于专用的 slab，其管理的都是同一个结构体实例，申请一个就给一个恰好内存大小的对象，这就可以充分利用空间。slab 分配器的实现源码（Linux kernel 5.10）：include/linux/slab_def.hinclude/linux/slab.hmm/slab.cSlab API与 libc 提供的内存申请 API （malloc 和 free ）类似，Slab 分配器提供的 API 为 kmalloc()和 kfree()。kmalloc()函数定义在include/linux/slab.h文件中，接收两个参数，并调用__kmalloc()函数。最终分配内存的实现是 __do_kmalloc()函数。size：申请的内存大小flags：参数使用 GFP_XXX来指定分配内存的具体内存域，例如 GFP_DMA 指定分配适合于DMA的内存区.static __always_inline void *kmalloc(size_t size, gfp_t flags) { ... return __kmalloc(size, flags); } // in mm/slab.c void *__kmalloc(size_t size, gfp_t flags) { return __do_kmalloc(size, flags, _RET_IP_); } static __always_inline void *__do_kmalloc(size_t size, gfp_t flags, unsigned long caller) { ... }实现原理由于slab的缓存特性，slab 分配器从 buddy 分配器中获取的物理内存称为 内存缓存（与 CPU 的硬件缓存进行区别，下文都称为缓存），使用结构体struct kmem_cache （定义在include/linux/slab_def.h文件中）描述。struct kmem_cache { /* 2) touched by every alloc & free from the backend */ slab_flags_t flags; /* constant flags */ unsigned int num; /* 每个 slab 中的 objs 数量 */ /* 3) cache_grow/shrink */ /* 每个 slab 所使用的页框数量 order */ unsigned int gfporder; /* force GFP flags, e.g. GFP_DMA */ gfp_t allocflags; size_t colour; /* cache colouring range */ unsigned int colour_off; /* colour offset */ struct kmem_cache *freelist_cache; unsigned int freelist_size; /* 4) cache creation/removal */ const char *name; struct list_head list; int refcount; int object_size; int align; struct kmem_cache_node *node[MAX_NUMNODES]; };SLAB分配器由可变数量的缓存组成，这些缓存由称为“缓存链”的双向循环链表链接在一起（如下图中的 kmem_cache 链表）。 在slab分配器的上下文中，缓存是特定类型的多个对象的管理器，例如使用cat /proc/slabinfo命令输出的mm_struct 或 fs_cache缓存，其名字保存在kmem_cache->name中（Linux 支持单个最大的 slab 缓存大小为32MB ）。kmem_cache 中所有对象的大小是相同的(object_size)，并且此 kmem_cache 中所有SLAB的大小也是相同的(gfporder、num)。每个缓存节点在内存中维护称为slab的连续页块，这些页面被切成小块，用于缓存数据结构和对象。 kmem_cache的 kmem_cache_node 成员记录了该kmem_cache 下的所有 slabs 列表。形成的结构如下图所示。kmem_cache_node 定义在mm/slab.h文件中，如下所示 struct kmem_cache_node { spinlock_t list_lock; #ifdef CONFIG_SLAB struct list_head slabs_partial; /* partial list first, better asm code */ struct list_head slabs_full; struct list_head slabs_free; unsigned long total_slabs; /* 所有 slab list 的长度 */ unsigned long free_slabs; /* 仅 free slab list 的长度*/ unsigned long free_objects; ... #endif };kmem_cache_noed 记录了3种slab：slabs_full ：已经完全分配的 slabslabs_partial： 部分分配的slabslabs_free：空slab，或者没有对象被分配以上3个链表保存的是slab 描述符，Linux kernel 使用 struct page 来描述一个slab。单个slab可以在slab链表之间移动，例如如果一个半满slab被分配了对象后变满了，就要从 slabs_partial 中被删除，同时插入到 slabs_full 中去。为了进一步解释，这里举个例子来说明（此处基于[2]所用的例子进行了一些修改）。用 struct kmem_cache 结构描述的一段内存就称作一个 slab 缓存池。一个slab缓存池就像是一箱牛奶，一箱牛奶中有很多瓶牛奶，每瓶牛奶就是一个 slab 。在初始时，所有牛奶都是新的，都存在 slabs_free 链表中。分配内存的时候，就相当于从牛奶箱中拿一瓶牛奶，若把一瓶牛奶喝完，该牛奶就会被加入到 slabs_full 中不能再分配；若只喝了一半就放回到 slabs_partial 中。牛奶可能会喝完（不考虑 slab 内存 free 的情况，因为自己不能生产牛奶（捂脸）），当箱子空的时候，你就需要去超市再买一箱回来。超市就相当于buddy 分配器。[注意]上述例子有点不严谨，实际上操作系统启动创建kmem_cache完成后，这三个链表都为空，只有在申请对象时发现没有可用的 slab 时才会创建一个新的SLAB，并加入到这三个链表中的一个中。也就是说kmem_cache中的SLAB数量是动态变化的，当SLAB数量太多时，kmem_cache会将一些SLAB释放回页框分配器中。继续来看 page 结构体关于 slab 的部分。struct page定义在include/linux/mm_types.h文件中，与slab相关的结构体成员如下所示：struct page { union { struct { /* Page cache and anonymous pages */ ... }; struct { /* slab, slob and slub */ union { struct list_head slab_list; struct { /* Partial pages */ struct page *next; int pages; /* Nr of pages left */ int pobjects; /* Approximate count */ }; }; struct kmem_cache *slab_cache; /* Double-word boundary */ void *freelist; /* first free object */ union { void *s_mem; /* slab: first object */ unsigned long counters; /* SLUB */ ... }; }; ... }; };void *s_mem: 指向该页框中第一个object 的地址 。struct kmem_cache *slab_cache: struct kmem_cache_node结构体用其追踪所有 page 的链表。struct list_head slab_list: 用于跟踪此页框属于哪个slab链表（full, free, partial），即使用此成员将list串联起来。void *freelist: 用于指向页框中空闲对象链表。空闲对象链表包含页框中每个空闲对象的索引。空闲对象链表是一个由数组制成的简单链表，它保存的地方有两种情况[6]： 1.保存在外部，会从SLAB中分配一个对象用于保存新的SLAB的空闲对象链表。 2.保存在内部，保存在这个SLAB所代表的连续页框的头部。在空闲对象链表都是保存在内部（通常也是保存在内部）的情况下，slab 描述符描述一块连续的内存关键结构如下：借助 freelist 以及 struct page 中的 active成员可以实现优先分配刚释放的内存。内存分配的过程如下图所示（图片引自此链接）。Slab缓存建立过程第一个 kmem_cache 结构体实例是定义的全局变量kmem_cache_boot（mm/slab.c）。static struct kmem_cache kmem_cache_boot = { .batchcount = 1, .limit = BOOT_CPUCACHE_ENTRIES, .shared = 1, .size = sizeof(struct kmem_cache), .name = "kmem_cache", };kmem_cache_init()函数在初始化过程中调用create_boot_cache()对 kmem_cache_boot进行了进一步初始化。第一个kmem cache实例用于为创建其他kmem cache实例分配空间，其大小与系统内的 node 有关。offsetof求出 kmem_cache 结构体中 node的偏移地址，再加上系统内 node 的个数乘以sizeof(struct kmem_cache_node *)，这真是把内存省到极致。因kmem_cache 的成员 node 是用最大的node数定义的数组大小，这里最后只使用系统内实际的node数量来申请内存。void __init kmem_cache_init(void) { int i; kmem_cache = &kmem_cache_boot; // NUM_INIT_LISTS 为 2 * 系统内的节点数量 // 初始化 kmem_cache_node 结构体 for (i = 0; i < NUM_INIT_LISTS; i++) kmem_cache_node_init(&init_kmem_cache_node[i]); // 完成第一个kmem cache实例kmem_cache的初始化 // 第一个kmem cache实例用于为创建其他kmem cache实例分配空间 create_boot_cache(kmem_cache, "kmem_cache", offsetof(struct kmem_cache, node) + nr_node_ids * sizeof(struct kmem_cache_node *), SLAB_HWCACHE_ALIGN, 0, 0); // kmem cache实例加入slab_caches链表 list_add(&kmem_cache->list, &slab_caches); slab_state = PARTIAL; kmalloc_caches[KMALLOC_NORMAL][INDEX_NODE] = create_kmalloc_cache( kmalloc_info[INDEX_NODE].name[KMALLOC_NORMAL], kmalloc_info[INDEX_NODE].size, ARCH_KMALLOC_FLAGS, 0, kmalloc_info[INDEX_NODE].size); slab_state = PARTIAL_NODE; setup_kmalloc_cache_index_table(); slab_early_init = 0; /* 5) Replace the bootstrap kmem_cache_node */ { int nid; for_each_online_node(nid) { init_list(kmem_cache, &init_kmem_cache_node[CACHE_CACHE + nid], nid); init_list(kmalloc_caches[KMALLOC_NORMAL][INDEX_NODE], &init_kmem_cache_node[SIZE_NODE + nid], nid); } } create_kmalloc_caches(ARCH_KMALLOC_FLAGS); } slab_cache 作为链接所有 kmem_cache 的链表头，使用宏LIST_HEAD(slab_caches)定义在文件mm/slab_common.c中；在 kmem_cache 初始化后，都将通过 list_add 将其添加到链表中。接下来 kmalloc_caches 是 struct kmem_cache *的一个二维数组指针（定义在mm/slab_common.c），其使用 kmalloc_info 中保存的信息生成不同 size 的 kmem_cache。kmalloc_info 的定义如下：类似一个 map，例如其中一个元素是{“kmalloc-96”，96}，“kmalloc-96”为 kmem_cache 的 name，96 为 object 的大小。此处调用 create_kmalloc_cache是创建一个 记录kmem_cache_node的 kmem_cache（对 INDEX_NODE 展开为 INDEX_NODE kmalloc_index(sizeof(struct kmem_cache_node))）#define INIT_KMALLOC_INFO(__size, __short_size) \ { \ .name[KMALLOC_NORMAL] = "kmalloc-" #__short_size, \ .name[KMALLOC_RECLAIM] = "kmalloc-rcl-" #__short_size, \ .size = __size, \ } #endif const struct kmalloc_info_struct kmalloc_info[] __initconst = { INIT_KMALLOC_INFO(0, 0), INIT_KMALLOC_INFO(96, 96), INIT_KMALLOC_INFO(192, 192), INIT_KMALLOC_INFO(8, 8), ... INIT_KMALLOC_INFO(67108864, 64M) };create_kmalloc_cache()函数首先调用函数 kmem_cache_zalloc(kmem_cache, GFP_NOWAIT);从 kmem_cache 中申请一个 slab 用来保存记录 struct kmem_cache_node实例的 kmem_cache。之后再调用create_boot_cache()函数对从 kmem_cacha slab 中申请来的 kmem_cache 进行初始化，并再添加到 slab_caches 链表中。struct kmem_cache *__init create_kmalloc_cache(const char *name, unsigned int size, slab_flags_t flags, unsigned int useroffset, unsigned int usersize) { struct kmem_cache *s = kmem_cache_zalloc(kmem_cache, GFP_NOWAIT); ... create_boot_cache(s, name, size, flags, useroffset, usersize); list_add(&s->list, &slab_caches); s->refcount = 1; return s; }此时 kmem_cache 形成的结构如下图。这部分可能有些绕，为了便于区分，此处额外添加了一个名为 vm_area_struct 的 kmem_cache。[问题] 图中有一点不严谨的地方。按照代码逻辑来看，记录 kmem_cache_node的 kmem_cache 的 name 并不为 “kmem_cache_node”，而是 "kmalloc-sizeof(struct kmem_cache_node)"，为什么这里不使用专用的 slab 来为 kmem_cache_node 来分配内存呢？ 按道理有多少 kmem_cache 实例就需要多少 kmem_cache_node 实例，为什么 kmem_cache 有专用的 slab，而 kmem_cache_node 没有？ 百思不得其解，望大佬赐教，万分感激！专用的 slab 的初始化分布在不同子系统所在的源文件中。以 vm_area_struct 为例，其初始化在__init proc_caches_init(void) 函数（定义在文件kernel/fork.c中）中被调用，代码如下所示。void __init proc_caches_init(void) { unsigned int mm_size; sighand_cachep = kmem_cache_create("sighand_cache", sizeof(struct sighand_struct), 0, SLAB_HWCACHE_ALIGN|SLAB_PANIC|SLAB_TYPESAFE_BY_RCU| SLAB_ACCOUNT, sighand_ctor); signal_cachep = kmem_cache_create("signal_cache", sizeof(struct signal_struct), 0, SLAB_HWCACHE_ALIGN|SLAB_PANIC|SLAB_ACCOUNT, NULL); files_cachep = kmem_cache_create("files_cache", sizeof(struct files_struct), 0, SLAB_HWCACHE_ALIGN|SLAB_PANIC|SLAB_ACCOUNT, NULL); fs_cachep = kmem_cache_create("fs_cache", sizeof(struct fs_struct), 0, SLAB_HWCACHE_ALIGN|SLAB_PANIC|SLAB_ACCOUNT, NULL); ... vm_area_cachep = KMEM_CACHE(vm_area_struct, SLAB_PANIC|SLAB_ACCOUNT); }其他通用的 kmem_cache（name 为 kmalloc-xxx）的初始化在函数new_kmalloc_cache()（定义在mm/slab_common.c文件中）中调用，如下。static void __init new_kmalloc_cache(int idx, enum kmalloc_cache_type type, slab_flags_t flags) { if (type == KMALLOC_RECLAIM) flags |= SLAB_RECLAIM_ACCOUNT; kmalloc_caches[type][idx] = create_kmalloc_cache( kmalloc_info[idx].name[type], kmalloc_info[idx].size, flags, 0, kmalloc_info[idx].size); }Slab缓存分配过程根据以上的分析可推断，kmem_cache_zalloc()函数实现了从 slab 中分配内存的功能。其调用流程如下kmem_cache_zalloc(include/linux/slab.h) kmem_cache_alloc(mm/slab.c) slab_alloc(mm/slab.c) __do_cache_alloc(mm/slab.c) ____cache_alloc(mm/slab.c) cpu_cache_get(mm/slab.c) cache_alloc_refill(mm/slab.c)____cache_alloc()函数实现如下，slab 分配器首先会从 array_cache中查找是否存在可用 obj （下面会详细介绍 struct array_cache），若有直接分配，若无则调用 cache_alloc_refill()函数尝试从 3 个 slabs_(free/partial/full)中寻找可用的内存，并将其填充到 array_cache 中，再次进行分配。static inline void *____cache_alloc(struct kmem_cache *cachep, gfp_t flags) { void *objp; struct array_cache *ac; ... ac = cpu_cache_get(cachep); if (likely(ac->avail)) { ac->touched = 1; objp = ac->entry[--ac->avail]; STATS_INC_ALLOCHIT(cachep); goto out; } STATS_INC_ALLOCMISS(cachep); objp = cache_alloc_refill(cachep, flags); ac = cpu_cache_get(cachep); out: if (objp) kmemleak_erase(&ac->entry[ac->avail]); return objp; }struct array_cache 本地 CPU 空闲对象链表在 kmem_cache 结构体中还有一个成员struct array_cache __percpu *cpu_cache;。其用来为每个CPU维护一个空间对象链表（Per CPU 相关内容参考此链接）。这能带来如下好处：提高硬件缓存的命中率。每个CPU都有它们自己的硬件高速缓存（在多核CPU下，通常 L1/L2 为 CPU 单独占有，L2/L3为所有CPU核共享的）。当此CPU上释放对象时，又再次申请一个相同大小的对象时，原对象很可能还在这个CPU的硬件高速缓存中。内核为每个CPU维护一个这样的链表，当需要新的对象时，会优先尝试从当前CPU的本地CPU空闲对象链表获取相应大小的对象，这样就能更快地分配对象。减少锁的竞争。假设多个CPU同时申请一个大小的slab，这时候如果没有本地CPU空闲对象链表，就会导致分配流程是互斥的，需要上锁，就导致分配效率低。struct array_cache定义在文件mm/slab.c文件中，如下所示：struct array_cache { unsigned int avail; unsigned int limit; unsigned int batchcount; unsigned int touched; void *entry[]; };avail： 表示当前可用对象的数量limit：可拥有的最大对象数batchcount：要从 slab_list 转移进本地高速缓存对象的数量，或从本地高速缓存中转移出去的 obj 数量touched： 是否在收缩后访问过entry: 伪数组，保存释放的对象指针本地CPU空闲对象链表在系统初始化完成后是一个空的链表，只有释放对象时才会将对象加入这个链表。链表对象个数是有限制的（最大值就是limit），链表数超过这个值时，会将 batchcount 个数的对象返回到所有CPU共享的空闲对象链表(也是这样一个结构)中。所有 CPU 共享的空闲对象链表在 struct kmem_cache_node结构体中同样存在一个 array_cache的成员 struct array_cache shared;*，其构成一个所有 CPU 共享的缓存。在此基础上，一个常规的对象申请流程是这样的：内核首先会从本地 CPU 空闲对象链表中尝试获取一个对象用于分配：如果失败，则检查所有CPU共享的空闲对象链表链表中是否存在，并且空闲链表中是否存在空闲对象，若有就转移 batchcount 个空闲对象到本地 CPU空闲对象链表中；如果第 1 步失败，就尝试从 SLAB中分配；这时如果还失败，kmem_cache会尝试从页框分配器中获取一组连续的页框建立一个新的SLAB，然后从新的SLAB中获取一个对象。对象释放流程如下[6]：首先会先将对象释放到本地CPU空闲对象链表中，如果本地CPU空闲对象链表中对象过多，kmem_cache 会将本地CPU空闲对象链表中的batchcount个对象移动到所有CPU共享的空闲对象链表链表中，如果所有CPU共享的空闲对象链表链表的对象也太多了，kmem_cache也会把所有CPU共享的空闲对象链表链表中 batchcount 个数的对象移回它们自己所属的SLAB中，这时如果SLAB中空闲对象太多，kmem_cache会整理出一些空闲的SLAB，将这些SLAB所占用的页框释放回页框分配器中。array_cache 填充在了解上述空闲对象分配的过程后，我们参考代码来进一步理解，cache_alloc_refill()函数实现如下：static void *cache_alloc_refill(struct kmem_cache *cachep, gfp_t flags) { int batchcount; struct kmem_cache_node *n; struct array_cache *ac, *shared; int node; void *list = NULL; struct page *page; ac = cpu_cache_get(cachep); batchcount = ac->batchcount; if (!ac->touched && batchcount > BATCHREFILL_LIMIT) { batchcount = BATCHREFILL_LIMIT; } n = get_node(cachep, node); BUG_ON(ac->avail > 0 || !n); shared = READ_ONCE(n->shared); if (!n->free_objects && (!shared || !shared->avail)) goto direct_grow; // 没有空闲对象，并且 没有共享空闲对象 可分配 spin_lock(&n->list_lock); shared = READ_ONCE(n->shared); /* See if we can refill from the shared array */ if (shared && transfer_objects(ac, shared, batchcount)) { shared->touched = 1; goto alloc_done; } while (batchcount > 0) { /* Get slab alloc is to come from. */ page = get_first_slab(n, false); if (!page) goto must_grow; check_spinlock_acquired(cachep); batchcount = alloc_block(cachep, ac, page, batchcount); fixup_slab_list(cachep, n, page, &list); } must_grow: n->free_objects -= ac->avail; alloc_done: spin_unlock(&n->list_lock); fixup_objfreelist_debug(cachep, &list); direct_grow: if (unlikely(!ac->avail)) { /* Check if we can use obj in pfmemalloc slab */ if (sk_memalloc_socks()) { void *obj = cache_alloc_pfmemalloc(cachep, n, flags); if (obj) return obj; } page = cache_grow_begin(cachep, gfp_exact_node(flags), node); ac = cpu_cache_get(cachep); if (!ac->avail && page) alloc_block(cachep, ac, page, batchcount); cache_grow_end(cachep, page); if (!ac->avail) return NULL; } ac->touched = 1; return ac->entry[--ac->avail]; }transfer_objects()：从全局共享空闲对象中转移空闲对象到本地 CPU 空闲对象链表上get_first_slab() : 从 slab 中获取空闲 slaballoc_block() : 从空闲 slab 中获取 空闲对象，转移到 本地 CPU 空闲对象链表上cache_grow_begin()：当 slab 中都没有空闲对象时，开始从 buddy 系统中获取内存上面描述的对象申请的过程与cache_alloc_refill()函数实现的功能基本是一致的。继续 kmalloc API了解 slab 基本原理后，我们再回头看当 kmalloc API 的处理流程。我比较关心的是 kmalloc 并没有传入 name，slab系统怎么知道选择哪个 kmem_cache？还是每个_ kmem_cache_ 实例的大小都会不一样？ 因为专用的 slab 通过 kmem_cache_alloc()函数进行内存申请。例如，在kernel/fork.c文件中，vm_area_alloc()函数的功能是申请一个 vm_area_struct结构体实例，通过vma = kmem_cache_alloc(vm_area_cachep, GFP_KERNEL);来完成该过程。而 kmalloc() 只是适用于分配通用的 slab。之前已经提到 kmalloc 会调用到 __do_kmalloc()函数，其实现如下：static __always_inline void *__do_kmalloc(size_t size, gfp_t flags, unsigned long caller) { struct kmem_cache *cachep; void *ret; if (unlikely(size > KMALLOC_MAX_CACHE_SIZE)) return NULL; cachep = kmalloc_slab(size, flags); if (unlikely(ZERO_OR_NULL_PTR(cachep))) return cachep; ret = slab_alloc(cachep, flags, caller); ret = kasan_kmalloc(cachep, ret, size, flags); trace_kmalloc(caller, ret, size, cachep->size, flags); return ret; }kmalloc_slab()根据申请的 size 获取到对应的 kmem_cache。继续深入，kmalloc_slab()函数实现如下。struct kmem_cache *kmalloc_slab(size_t size, gfp_t flags) { unsigned int index; if (size <= 192) { if (!size) return ZERO_SIZE_PTR; index = size_index[size_index_elem(size)]; } else { if (WARN_ON_ONCE(size > KMALLOC_MAX_CACHE_SIZE)) return NULL; index = fls(size - 1); } return kmalloc_caches[kmalloc_type(flags)][index]; }首先计算出在对应的 index，用于索引之前介绍的全局二维数组 _kmalloc_caches_。 流程如下：首先，对于size 小于 192B的申请，则从 size_index 中查找索引。size_index_elem()函数和 size_index 数组的定义如下，为了方便比对，这里再次给出 kmalloc_info 数组的前 7 个元素。通过分析可以发现，在分配小于 192 的 obj 时，会向上对齐，例如，9~16 字节的 obj 保存到 16 字节的kmem_cache中，65-96字节的 obj 保存到 96字节的 mem_cache中。static u8 size_index[24] __ro_after_init = { 3, /* 8 */ 4, /* 16 */ 5, /* 24 */ 5, /* 32 */ 6, /* 40 */ 6, /* 48 */ 6, /* 56 */ 6, /* 64 */ 1, /* 72 */ 1, /* 80 */ 1, /* 88 */ 1, /* 96 */ 7, /* 104 */ 7, /* 112 */ 7, /* 120 */ 7, /* 128 */ 2, /* 136 */ 2, /* 144 */ 2, /* 152 */ 2, /* 160 */ 2, /* 168 */ 2, /* 176 */ 2, /* 184 */ 2 /* 192 */ }; static inline unsigned int size_index_elem(unsigned int bytes) { return (bytes - 1) / 8; } const struct kmalloc_info_struct kmalloc_info[] __initconst = { INIT_KMALLOC_INFO(0, 0), INIT_KMALLOC_INFO(96, 96), INIT_KMALLOC_INFO(192, 192), INIT_KMALLOC_INFO(8, 8), INIT_KMALLOC_INFO(16, 16), INIT_KMALLOC_INFO(32, 32), INIT_KMALLOC_INFO(64, 64), INIT_KMALLOC_INFO(128, 128), ... }对于高于 192 字节的 obj 分配，则调用fls()函数，其作用是得出 最高有效位（二进制下，最低向高位，最后一个 1的位置）。例如 fls(0) = 0, fls(1) = 1, fls(0x80000000) = 32。此处不进一步分析。kfree接下来简单分析一下释放通用型 slab 对象时 如何根据一个指针找到待释放的对象，并对其进行释放操作的过程。kfree()函数如下，根据一个指针找到待释放的对象由函数virt_to_cache()函数实现。void kfree(const void *objp) { struct kmem_cache *c; unsigned long flags; trace_kfree(_RET_IP_, objp); if (unlikely(ZERO_OR_NULL_PTR(objp))) return; local_irq_save(flags); kfree_debugcheck(objp); c = virt_to_cache(objp); if (!c) { local_irq_restore(flags); return; } debug_check_no_locks_freed(objp, c->object_size); debug_check_no_obj_freed(objp, c->object_size); __cache_free(c, (void *)objp, _RET_IP_); local_irq_restore(flags); }virt_to_cache()函数定义在mm/slab.h文件中，调用 virt_to_head_page()函数获得该虚拟地址对应的 page 描述符，然后由 page->slab_cache 反向获取到对应的 kmem_cache。static inline struct kmem_cache *virt_to_cache(const void *obj) { struct page *page; page = virt_to_head_page(obj); if (WARN_ONCE(!PageSlab(page), "%s: Object is not a Slab page!\n", __func__)) return NULL; return page->slab_cache; }虚拟地址到 page 描述符的相关实现如下（include/linux/mm.h）。static inline struct page *virt_to_head_page(const void *x) { struct page *page = virt_to_page(x); return compound_head(page); } // arch/arm64/include/asm/memory.h #define virt_to_page(x) pfn_to_page(virt_to_pfn(x)) #define virt_to_pfn(x) __phys_to_pfn(__virt_to_phys((unsigned long)(x))) #define __virt_to_phys(x) __virt_to_phys_nodebug(x) // __is_lm_address 检查是否为线性映射（linear map） // 0xffff_8000_0000_0000 - 0xffff_8008_0000_0000：线性映射区域映射内存 // 0xffff_8000_0000_0000向下的 kernel space virtual addr 是给 kernel image 使用的[8] // 这对应 __lm_to_phys 和 __kimg_to_phys 两个宏。 #define __virt_to_phys_nodebug(x) ({ \ phys_addr_t __x = (phys_addr_t)(__tag_reset(x)); \ __is_lm_address(__x) ? __lm_to_phys(__x) : __kimg_to_phys(__x); \ }) #define __is_lm_address(addr) (((u64)(addr) ^ PAGE_OFFSET) < (PAGE_END - PAGE_OFFSET)) #define __lm_to_phys(addr) (((addr) & ~PAGE_OFFSET) + PHYS_OFFSET) #define __kimg_to_phys(addr) ((addr) - kimage_voffset)虚拟地址与物理地址转换的过程这里不详细介绍，之后会专门用一篇文章来梳理。继续回到kfree()函数，virt_to_cache()函数获取到了虚拟地址所对应的 kmem_cache，接下来调用__cache_free()函数来释放内存。static __always_inline void __cache_free(struct kmem_cache *cachep, void *objp, unsigned long caller) { ... ___cache_free(cachep, objp, caller); } void ___cache_free(struct kmem_cache *cachep, void *objp, unsigned long caller) { struct array_cache *ac = cpu_cache_get(cachep); check_irq_off(); if (unlikely(slab_want_init_on_free(cachep))) // 释放时初始化对象 memset(objp, 0, cachep->object_size); ... if (nr_online_nodes > 1 && cache_free_alien(cachep, objp)) return; if (ac->avail < ac->limit) { STATS_INC_FREEHIT(cachep); } else { STATS_INC_FREEMISS(cachep); cache_flusharray(cachep, ac); } if (sk_memalloc_socks()) { struct page *page = virt_to_head_page(objp); if (unlikely(PageSlabPfmemalloc(page))) { cache_free_pfmemalloc(cachep, page, objp); return; } } __free_one(ac, objp); }以上代码实现与我们之前分析的对象回收过程是一致的本地 CPU 缓存数量未超过 limit 则直接回收到本地 CPU 空闲缓存中，调用 __free_one()函数若超过则调用cache_flusharray()向全局 CPU 共享空闲缓存中移动cache_flusharray()函数首先判断是否存在全局共享缓存，若存在并且全局共享缓存空闲对象未超过限制，则向全局缓存中移动若不满足上述条件则调用free_block()函数向 slab 中释放空闲对象。同时根据释放情况，调整对应的slab 到 slabs_free 或 slabs_partial 链表中。static void cache_flusharray(struct kmem_cache *cachep, struct array_cache *ac) { ... if (n->shared) { struct array_cache *shared_array = n->shared; int max = shared_array->limit - shared_array->avail; if (max) { if (batchcount > max) batchcount = max; memcpy(&(shared_array->entry[shared_array->avail]), ac->entry, sizeof(void *) * batchcount); shared_array->avail += batchcount; goto free_done; } } free_block(cachep, ac->entry, batchcount, node, &list); ... } 参考资料[1]. Linux kernel 5.10.20 源码[2]. Linux内核内存管理算法Buddy和Slab[3]. slab分配器--Linux内存管理 (二十二) [4]. The Slab Allocator in the Linux kernel[5]. kernel doc # Chapter 8 Slab Allocator[6]. linux内存源码分析 - SLAB分配器概述[7]. 优雅的slab内存分配器（二）——slab初始化kmem_cache_init[8]. linux kernel中的virt_to_phys代码解读
简介Linux内核中存在不同的子系统，例如内存子系统、I/O子系统。每个子系统需要可能需要获取其他子系统的某些事件以进行相应的处理。通知链（notification chains）机制允许子系统订阅某个事件，当该事件发生时就通知该子系统。这类似于MQTT[2]订阅/发布的方式。通知机制Linux源码实现主要在include/linux/notifier.h和kernel/notifier.c文件中。相关结构体当一个事件发生时，该事件对应的一连串回调函数将被调用，从而实现事件的通知。在Linux中，回调函数的定义如下：typedef int (*notifier_fn_t)(struct notifier_block *nb, unsigned long action, void *data);其接收 3 个参数：nb：回调函数组成的函数指针链表，下面会详细解析该结构体action: 事件的类型。通知链可能支持多个事件，因此我们需要此参数来将一个事件与其他事件区分开data：保存事件的私有信息，用来提供有关事件的其他数据信息。返回值标识通知执行的结果，分为#define NOTIFY_DONE 0x0000 /* 订阅者对该事件不感兴趣 */ #define NOTIFY_OK 0x0001 /* 通知正常完成 */ #define NOTIFY_STOP_MASK 0x8000 /* 通知已完成，但不应为此事件调用进一步的回调。 */ #define NOTIFY_BAD (NOTIFY_STOP_MASK|0x0002) /*通知失败*/希望在某个事件上得到通知的Linux内核子系统都应提供自己的notifier_fn_t回调函数。 通知链机制的主要作用是在发生异步事件时调用某些回调。struct notifier_block { notifier_fn_t notifier_call; // 回调函数 struct notifier_block __rcu *next; // 下一个回调块 int priority; // 优先级 };Linux通知链机制提供了 4 种通知类型，包括Blocking notifier chains、SRCU notifier chainsAtomic notifier chains和Raw notifier chains。每种类型的通知链都类似，主要区别在于通知链执行过程的保护机制不同，暂不展开。以阻塞通知链为例，阻塞通知链会在进程上下文中调用/执行回调函数， 这意味着通知链中的调用过程可能会被阻塞。struct blocking_notifier_head { struct rw_semaphore rwsem; struct notifier_block __rcu *head; };实现前面提到，通知机制类似一种订阅/发布的关系。订阅者向信息代理者注册自己的信息（例如，我们在报亭写下自己的地址），当信息代理者接收到信息发布者的信息后（如人民日报生成报纸并寄送给报亭），将就信息传递给该信息的所有订阅者（报亭将报纸分发给我们）。“代理者”通过以上例子，要实现事件的订阅与发布，首先需要建立一个代理者用来记录订阅者的“地址信息”。如下在文件mm/oom_kill.c中建立一个 oom_notify_list的链表，为需要关注内存不足的子系统提供一个记录点。static BLOCKING_NOTIFIER_HEAD(oom_notify_list);BLOCKING_NOTIFIER_HEAD宏定义在文件include/linux/notifier.h中，展开为定义一个blocking_notifier_head结构体，并进行简单初始化。#define BLOCKING_NOTIFIER_HEAD(name) \ struct blocking_notifier_head name = \ BLOCKING_NOTIFIER_INIT(name) 为方便订阅者注册自身对OOM时间发生时的处理，代理者还提供相应API。register_xxx和unregister_xxx分别将notifier_block *nb 插入到 list 中和从 list 中移除，从而建立/解除订阅者关系。当然订阅者也可以直接调用include/linux/notifier.h文件中提供的blocking_notifier_chain_register()函数也可。int register_oom_notifier(struct notifier_block *nb) { return blocking_notifier_chain_register(&oom_notify_list, nb); } EXPORT_SYMBOL_GPL(register_oom_notifier); int unregister_oom_notifier(struct notifier_block *nb) { return blocking_notifier_chain_unregister(&oom_notify_list, nb); } EXPORT_SYMBOL_GPL(unregister_oom_notifier); “订阅者”例如，RCU 子系统为接收 OOM 事件，其在kernel/rcu/rcutorture.c文件中定义好回调函数并将初始化struct notifier_block结构体，最后调用 API register_oom_notifier来将回调函数添加到 list 中。static struct notifier_block rcutorture_oom_nb = { .notifier_call = rcutorture_oom_notify }; register_oom_notifier(&rcutorture_oom_nb);“发布者”当事件产生时，系统只需要获取保存回调函数的 list，并一一调用相应的回调函数即可。事件订阅者回调函数的调用实现定义在kernel/notifier.c文件中。blocking_notifier_call_chain()接收 3 个参数：nh - 通知链链表头val - 通知的类型v - 回调函数可能用到的参数int blocking_notifier_call_chain(struct blocking_notifier_head *nh, unsigned long val, void *v) { int ret = NOTIFY_DONE; if (rcu_access_pointer(nh->head)) { down_read(&nh->rwsem); ret = notifier_call_chain(&nh->head, val, v, -1, NULL); up_read(&nh->rwsem); } return ret; } EXPORT_SYMBOL_GPL(blocking_notifier_call_chain);blocking_notifier_call_chain()函数调用notifier_call_chain，其接收额外两个参数nr_to_call和nr_calls表示是要调用的通知程序函数的数量和已发送的通知的数量。nr_to_call 为 -1 以及nr_calls 为 NULL 时则忽略这两个参数。static int notifier_call_chain(struct notifier_block **nl, unsigned long val, void *v, int nr_to_call, int *nr_calls) { ... }同样以内存管理中的OOM事件为例。当内存不足时，定义在mm/oom_kill.c文件中的out_of_memory()函数将被调用。通知链中回调函数的调用将会进行内存回收，并通过 freed 返回是否回收了内存。// mm/oom_kill.c bool out_of_memory(struct oom_control *oc) { ... if (!is_memcg_oom(oc)) { blocking_notifier_call_chain(&oom_notify_list, 0, &freed); if (freed > 0) /* Got some memory back in the last second. */ return true; } } 参考链接0xax. Notification Chains in Linux KernelMQTT
简介Linux内核编译完成后生成的通常是被压缩过的内核文件，例如在 Centos 和Ubuntu的 /boot目录下就保存了 vmlinuz 文件。vmlinuz是可引导的、压缩的内核，“vm”代表“Virtual Memory”。# ls -l /boot -rwxr-xr-x 1 root root 6765160 Aug 26 2020 vmlinuz-3.10.0-1127.19.1.el7.x86_64 -rwxr-xr-x 1 root root 6769256 Feb 3 23:10 vmlinuz-3.10.0-1160.15.2.el7.x86_64vmlinuz由ELF文件vmlinux 经过OBJCOPY，并压缩后的文件。Linux下大量工具都是对 ELF 文件进行解析，因此，若我们想要逆向对 Linux 内核镜像进行二进制的分析，就需要先将 vmlinuz 文件还原成 vmlinux 文件。Linux内核镜像分析工具准备Linux内核提供了脚本来实现vmlinuz 到 vmlinux 的转换——下载地址 extract-vmlinux。但在 Centos 和 Ubuntu 下可以通过包管理器直接安装，如下。# in centos yum install kernel-devel # in ubuntu apt-get install linux-headers-$(uname -r)安装完成后，extract-linux脚本将保存在Centos: /usr/src/kernels/$(uname -r)/scripts/extract-vmlinuxUbuntu: /usr/src/linux-headers-$(uname -r)/scripts/extract-vmlinux文件转换/usr/src/kernels/$(uname -r)/scripts/extract-vmlinux vmlinuz-$(uname -r) > vmlinuxELF 文件解析Linux 下 常用的 ELF 文件解析工具是 readelf和objdump，以下记录一些常用的例子：(1) readelf 查看 ELF 文件的信息：- -S --section-headers 查看段头 - -h --file-header 查看文件头 - -r --relocs 查看重定位相关信息 - -x --hex-dump 查看16进制 (2) objdump读取 ELF文件中的内容- -d 查看反编译结果 - -j 指定查看的段 - -r 查看重定位相关信息 - 例子： `objdump -d -j .text hello.o` 符号名通过 objdump 命令则可反汇编 vmlinux 文件。 objdump -D vmlinux | less但该文件中没有保存符号名（例如函数名），这样分析不太方便。符号名相关的内容保存在/boot/System.map文件中。例如查找函数tcp_v4_do_rcv的地址，可执行以下命令获取到函数tcp_v4_do_rcv的地址为ffffffff816c62e0，再结合反汇编结果进行分析。grep "tcp_v4_do_rcv" /boot/System.map-3.10.0-1160.15.2.el7.x86_64 ffffffff816c62e0 T tcp_v4_do_rcv参考链接# Extracting the Linux kernel image (vmlinuz)
# 简介Per CPU 变量，顾名思义指每个CPU核都有对某一个变量的一份拷贝。Per CPU 变量的应用之一是保存每个 CPU 的 ID，这是因为内核有时需要判断自己当前是运行在哪个CPU核上，因此需要获取当前执行代码的 CPU ID。实现为每个CPU定义一个变量的拷贝的宏定义在文件include/linux/percpu-defs.h中，如下：#define DEFINE_PER_CPU(type, name) \ DEFINE_PER_CPU_SECTION(type, name, "")若我们使用DEFINE_PER_CPU(int, per_cpu_n)为每个CPU定义变量，其展开宏如下。__attribute__((section(".data..percpu"))) int per_cpu_n #define DEFINE_PER_CPU_SECTION(type, name, sec) \ __PCPU_ATTRS(sec) __typeof__(type) name #define __PCPU_ATTRS(sec) \ __percpu __attribute__((section(PER_CPU_BASE_SECTION sec))) #define PER_CPU_BASE_SECTION ".data..percpu"在链接过程中，所有通过DEFINE_PER_CPU宏定义的变量都将链接到一起。在操作系统启动时，Linux 将为该段分配一段内存。 查看编译出的内核镜像可找到.data..percpu段（Linux镜像分析方法请先参考此链接）。# readelf -S vmlinux [Nr] Name Type Address Offset Size EntSize Flags Link Info Align [21] .data..percpu PROGBITS 0000000000000000 01000000 000000000001d000 0000000000000000 WA 0 0 4096per CPU 变量的访问通过宏get_cpu_var完成 。Linux内核是可抢占的，并且在访问访问 per cpu变量时我们需要知道当前代码运行在哪个CPU核上。 因此，在访问每个cpu变量时，应当不允许抢占当前代码并将其移至另一个CPU。例如，若在获取到 CPU id 为 1 后，该任务被抢占而移动到了 CPU 2上继续运行，这时访问的将仍然是 CPU 1的per cpu 变量。因此，在 get_cpu_var 宏中，首先要调用preempt_disable()函数禁止任务抢占。// in include/linux/percpu-defs.h #define get_cpu_var(var) \ (*({ \ preempt_disable(); \ this_cpu_ptr(&var); \ }))我比较好奇的是 this_cpu_ptr是如何实现的。内核如何将该变量对应到属于该CPU的 per CPU 变量内存呢？ 在初始化时，内核会使用一个数组__per_cpu_offset[cpu]记录每个CPU静态per cpu 变量的偏移地址。在ARM64架构下， OS 启动时将 per cpu 偏移地址写入到 TPDIR_EL1 和 TPDIR_EL2 寄存器中。void __init setup_per_cpu_areas(void) { unsigned long delta; unsigned int cpu; ... delta = (unsigned long)pcpu_base_addr - (unsigned long)__per_cpu_start; for_each_possible_cpu(cpu) __per_cpu_offset[cpu] = delta + pcpu_unit_offsets[cpu]; } /* arch/arm64/include/asm/percpu.h */ static inline void set_my_cpu_offset(unsigned long off) { asm volatile(ALTERNATIVE("msr tpidr_el1, %0", "msr tpidr_el2, %0", ARM64_HAS_VIRT_HOST_EXTN) :: "r" (off) : "memory"); }this_cpu_ptr的宏展开如下：即相当于 percpu 变量指针 ptr 加上__my_cpu_offset。#define arch_raw_cpu_ptr(ptr) SHIFT_PERCPU_PTR(ptr, __my_cpu_offset) #define raw_cpu_ptr(ptr) \ ({ \ __verify_pcpu_ptr(ptr); \ arch_raw_cpu_ptr(ptr); \ }) #define this_cpu_ptr(ptr) raw_cpu_ptr(ptr)__my_cpu_offset宏即是从当前cpu的tpidr_el1、tpidr_el2寄存器中取出此前设置的__per_cpu_offset[cpu]值，实现如下：static inline unsigned long __my_cpu_offset(void) { unsigned long off; /* * We want to allow caching the value, so avoid using volatile and * instead use a fake stack read to hazard against barrier(). */ asm(ALTERNATIVE("mrs %0, tpidr_el1", "mrs %0, tpidr_el2", ARM64_HAS_VIRT_HOST_EXTN) : "=r" (off) : "Q" (*(const unsigned long *)current_stack_pointer)); return off; }有时会需要指定某个 CPU 获取其某个 per cpu 变量的地址，通过宏per_cpu_ptr实现，源码如下：#define SHIFT_PERCPU_PTR(__p, __offset) \ RELOC_HIDE((typeof(*(__p)) __kernel __force *)(__p), (__offset)) #define per_cpu_ptr(ptr, cpu) \ ({ \ __verify_pcpu_ptr(ptr); \ SHIFT_PERCPU_PTR((ptr), per_cpu_offset((cpu))); \ })__user，__kernel，__safe，__force等定义在compiler_type.h头文件中。看到两个很奇怪的现象，一个是只有在__CHECKER__宏打开的情况下，他们的定义才会被实现，否则他们的定义是空的。第二个是它们的attribute的定义，并不是gcc支持的属性。那到底是哪里使用到了呢？原来linux的作者们自己开发了一套编译期代码检查的工具Sparse，可以用于在编译阶段快速发现代码中隐含的问题。[1]address_space 定义了指针能指向的内存的类型，0代表kernel space，1代表user space，2代表设备地址空间，3代表cpu局部的内存空间safe 表示变量可以为空force 表示变量可以强制类型转换Per CPU 变量的应用记录每个CPU 的 id 是 per CPU 变量的应用之一。那么有了 Per CPU变量之后，如何获得当前执行代码的CPU 编号？ 内核函数smp_processor_id()用来获取当前 CPU 的 id 。CPU id 的存储依赖于 per CPU 变量（DEFINE_PER_CPU宏用来定义 cpu_number 变量）。// 每个CPU的cpuid是放置在cpu_number这个percpu变量中 DEFINE_PER_CPU(int, cpu_number);在内核初始化时，smp_prepare_cpus()函数执行per_cpu(cpu_number, cpu) = cpu;设定每个核的编号。// in /arch/arm64/kernel/smp.c void __init smp_prepare_cpus(unsigned int max_cpus) { const struct cpu_operations *ops; int err; unsigned int cpu; unsigned int this_cpu; init_cpu_topology(); this_cpu = smp_processor_id(); for_each_possible_cpu(cpu) { // 设置 CPU id per_cpu(cpu_number, cpu) = cpu; // 确定在哪个核上执行的，若是本身则跳过。 if (cpu == smp_processor_id()) continue; ops = get_cpu_ops(cpu); if (!ops) continue; err = ops->cpu_prepare(cpu); if (err) continue; set_cpu_present(cpu, true); numa_store_cpu_info(cpu); } }smp_processer_id ()函数(定义在 include/linux/smp.h)展开如下。# define smp_processor_id() __smp_processor_id() #define __smp_processor_id(x) raw_smp_processor_id(x)raw_smp_processor_id与处理器架构相关（下例为ARM64）的实现如下，raw_cpu_ptr 获取到 cpu_number 的地址，在解引用得到 cpu id。#define raw_smp_processor_id() (*raw_cpu_ptr(&cpu_number))per CPU 变量在多文件下的用法声明一个 per cpu 变量并在另一个文件中引用，以获取当前 task_struct 为例（x86下 current 宏的实现）。定义方式如下：DEFINE_PER_CPU(struct task_struct *, current_task) ____cacheline_aligned = &init_task; EXPORT_PER_CPU_SYMBOL(current_task);在另一个文件中引用方式如下：DECLARE_PER_CPU(struct task_struct *, current_task); static __always_inline struct task_struct *get_current(void) { return this_cpu_read_stable(current_task); } #define current get_current()参考资料[1]. 聊一聊linux内核中的基础工具库(2) Sparse