前言:从“黑盒调试”到“可观测性驱动开发”

        在前面的章节中,我们实现了安全加固、内存管理、网络协议栈等核心功能。但当系统变得复杂时,调试不再是“加几个printk”就能解决的:性能瓶颈在哪里?哪个进程占用了CPU?哪条代码路径触发了死锁?传统调试手段(printk、gdb)要么侵入性太强(改变时序掩盖bug),要么无法在生产环境使用(gdb会暂停系统)。

        可观测性(Observability)的核心是“在不干扰系统正常运行的前提下,采集、聚合、可视化系统状态”。现代OS提供三层可观测性基础设施:

  • perf_events:硬件性能计数器 + 软件事件采样,定位CPU/内存/IO瓶颈
  • eBPF:内核字节码虚拟机,允许用户空间安全注入观测代码(类似DTrace但更安全)
  • ftrace/kprobe/uprobe:内核函数追踪、动态插桩、用户空间函数追踪

本章里程碑:

  • perf_events:perf_event_open系统调用、环形缓冲区采样、硬件计数器读取(PMU)
  • eBPF:字节码验证器、JIT编译、map数据结构、kprobe/uprobe挂载点
  • ftrace:tracefs文件系统、function tracer、event tracer
  • 集成:perf record/report工作流、BCC/bpftrace脚本示例

核心概念:可观测性不是“事后分析”,而是“设计时内置”
perf_events的本质:采样而非轮询

        传统性能分析工具(如top)通过轮询/proc获取进程状态,采样频率低(~1Hz)且无法深入内核。perf_events采用基于中断的采样:硬件PMU(Performance Monitoring Unit)在指定事件(如CPU周期、缓存缺失、分支预测失败)发生时触发中断,内核记录当前上下文(PID、PC、调用栈)并写入环形缓冲区。

️ 关键洞察:perf_events的核心是“低开销采样”。通过调整采样频率(如每100万次CPU周期采样一次),可将开销控制在1%以内。教学实现中必须先实现sys_perf_event_open(),创建perf_event结构并关联到当前任务,在定时器中断或PMU中断中采集数据并写入mmap映射的环形缓冲区。

eBPF:在内核中安全运行用户代码

        eBPF(extended Berkeley Packet Filter)最初用于网络包过滤,现已扩展为通用内核观测与 hook 机制。用户空间编写eBPF字节码(类似BPF但扩展了寄存器和指令),通过bpf()系统调用加载到内核。内核验证器(verifier)确保程序:①不会无限循环(有界执行);②不会访问非法内存;③不会崩溃内核。验证通过后,JIT编译器将字节码转为原生代码执行。

        关键约束:①eBPF程序必须终止(有界循环);②只能访问白名单中的内核辅助函数(bpf_helper);③map是eBPF程序与用户空间通信的唯一通道。教学实现中先实现验证器核心逻辑(控制流图分析、寄存器状态追踪),再实现简单map类型(hash map、array map)。

ftrace:内核函数的“时间机器”

        ftrace(function tracer)是内核内置的函数追踪框架,通过编译时插入mcount调用(类似gprof),在运行时动态启用/禁用追踪。与perf_events不同,ftrace记录每个函数的进入/退出,提供完整的调用链。但开销较高(每函数调用增加~0.5μs),仅适合短时间调试。

        关键约束:①ftrace通过/sys/kernel/debug/tracing暴露接口;②支持function/function_graph/event等多种tracer;③可过滤特定函数/进程。教学实现中先实现function tracer,在编译时通过-pg插入mcount,运行时通过ftrace_stub钩子调用注册的tracer回调。


实战代码
perf_events系统调用与环形缓冲区
// kernel/events/core.c

struct perf_event {
    struct list_head        entry;      // 链接到task_struct或cpu
    struct perf_event_attr  attr;       // 事件属性(类型、采样周期等)
    atomic_t                count;      // 事件计数
    struct ring_buffer      *rb;        // 环形缓冲区
    struct task_struct      *owner;     // 所有者进程
    // ...
};

struct perf_event_attr {
    __u32 type;             // PERF_TYPE_HARDWARE/SOFTWARE
    __u32 size;
    __u64 config;           // 具体事件(如PERF_COUNT_HW_CPU_CYCLES)
    __u64 sample_period;    // 采样周期
    __u64 sample_type;      // 采样数据类型(IP、TID、CALLCHAIN等)
    // ...
};

// ★ sys_perf_event_open:创建perf事件
SYSCALL_DEFINE5(perf_event_open,
                struct perf_event_attr __user *, attr_uptr,
                pid_t, pid, int, cpu, int, group_fd, unsigned long, flags) {
    struct perf_event_attr attr;
    struct perf_event *event;
    struct task_struct *task;
    
    if (copy_from_user(&attr, attr_uptr, sizeof(attr)))
        return -EFAULT;
    
    // ★ 验证属性
    if (attr.size < PERF_ATTR_SIZE_VER0 || attr.size > sizeof(attr))
        return -EINVAL;
    
    // ★ 查找目标进程(pid=0表示当前进程,pid=-1表示所有进程)
    if (pid == 0) {
        task = current;
    } else {
        task = find_task_by_vpid(pid);
        if (!task)
            return -ESRCH;
    }
    
    // ★ 创建perf_event
    event = kzalloc(sizeof(*event), GFP_KERNEL);
    if (!event)
        return -ENOMEM;
    
    event->attr = attr;
    atomic_set(&event->count, 0);
    
    // ★ 分配环形缓冲区(用户空间通过mmap访问)
    event->rb = ring_buffer_alloc(attr.watermark ? attr.wakeup_watermark : 4096,
                                  GFP_KERNEL);
    if (!event->rb) {
        kfree(event);
        return -ENOMEM;
    }
    
    // ★ 关联到任务或CPU
    if (cpu == -1) {
        // 进程级别事件:链接到task_struct
        list_add_tail(&event->entry, &task->perf_event_list);
    } else {
        // CPU级别事件:链接到cpu变量
        list_add_tail(&event->entry, &per_cpu(cpu_perf_events, cpu));
    }
    
    event->owner = task;
    
    // ★ 返回event fd
    int fd = get_unused_fd_flags(flags);
    if (fd < 0) {
        ring_buffer_free(event->rb);
        kfree(event);
        return fd;
    }
    
    // 将event关联到fd...
    return fd;
}

// ★ 硬件计数器中断处理(PMU触发)
void perf_event_interrupt(struct pt_regs *regs) {
    struct perf_event *event;
    
    // 遍历当前CPU的所有perf事件
    list_for_each_entry(event, &this_cpu_ptr(&cpu_perf_events), entry) {
        if (event->attr.type != PERF_TYPE_HARDWARE)
            continue;
        
        // ★ 读取硬件计数器
        u64 count = pmu_read_counter(event);
        atomic64_add(count, &event->count);
        
        // ★ 检查是否达到采样周期
        if (atomic64_read(&event->count) >= event->attr.sample_period) {
            atomic64_sub(event->attr.sample_period, &event->count);
            
            // ★ 采样:记录PC、PID、时间戳等
            perf_sample_data data;
            perf_prepare_sample(&data, event, regs);
            
            // ★ 写入环形缓冲区
            ring_buffer_write(event->rb, &data, sizeof(data));
            
            // ★ 唤醒等待的用户空间进程(通过poll/epoll)
            wake_up(&event->waitq);
        }
    }
}

// ★ 环形缓冲区实现(无锁,单生产者多消费者)
struct ring_buffer {
    void            *data_pages;    // 数据页(mmap映射到用户空间)
    struct page     *metadata_page; // 元数据页(head、tail、overrun等)
    atomic_t        head;           // 写指针
    atomic_t        tail;           // 读指针(用户空间更新)
    // ...
};

int ring_buffer_write(struct ring_buffer *rb, void *data, size_t size) {
    int head = atomic_read(&rb->head);
    int tail = atomic_read(&rb->tail);
    
    // ★ 检查空间
    if (CIRC_SPACE(head, tail, rb->nr_pages * PAGE_SIZE) < size) {
        atomic_inc(&rb->overrun); // 记录溢出
        return -ENOBUFS;
    }
    
    // ★ 写入数据(单生产者,无需锁)
    memcpy(rb->data_pages + head, data, size);
    smp_wmb(); // 内存屏障,确保数据写入后更新head
    atomic_set(&rb->head, (head + size) % (rb->nr_pages * PAGE_SIZE));
    
    return 0;
}
eBPF验证器与JIT编译
// kernel/bpf/core.c

// ★ eBPF指令集(简化版)
struct bpf_insn {
    __u8 code;        // 操作码
    __u8 dst_reg:4;   // 目标寄存器
    __u8 src_reg:4;   // 源寄存器
    __s16 off;        // 偏移(用于分支/内存访问)
    __s32 imm;        // 立即数
};

// ★ eBPF寄存器状态(验证器追踪)
struct bpf_reg_state {
    enum bpf_reg_type type; // PTR_TO_STACK/CONST_IMM/PTR_TO_MAP等
    u64 min_value;          // 最小值(用于边界检查)
    u64 max_value;          // 最大值
    // ...
};

struct bpf_verifier_env {
    struct bpf_prog *prog;
    struct bpf_reg_state regs[MAX_BPF_REG]; // 每个寄存器的状态
    struct bpf_stack_state stack[MAX_BPF_STACK]; // 栈状态
    // ...
};

// ★ 验证器:控制流图分析 + 寄存器状态追踪
int bpf_check(struct bpf_prog *prog) {
    struct bpf_verifier_env env;
    
    // ★ 第一步:基本块划分与控制流图构建
    if (build_control_flow_graph(prog, &env.cfg))
        return -EINVAL;
    
    // ★ 第二步:寄存器状态模拟执行
    for (int i = 0; i < env.cfg.num_blocks; i++) {
        struct bpf_basic_block *block = &env.cfg.blocks[i];
        
        // 初始化入口寄存器状态(从predecessor继承)
        init_reg_state_from_predecessors(&env, block);
        
        // 逐条指令模拟执行
        for (int j = 0; j < block->num_insns; j++) {
            struct bpf_insn *insn = &block->insns[j];
            
            if (simulate_insn(&env, insn)) {
                pr_err("bpf: instruction %d failed verification\n", j);
                return -EINVAL;
            }
        }
    }
    
    // ★ 第三步:检查所有路径是否终止
    if (!all_paths_terminate(&env.cfg)) {
        pr_err("bpf: program has infinite loop\n");
        return -EINVAL;
    }
    
    return 0;
}

// ★ 模拟执行单条指令(验证器核心)
static int simulate_insn(struct bpf_verifier_env *env, struct bpf_insn *insn) {
    struct bpf_reg_state *dst = &env->regs[insn->dst_reg];
    struct bpf_reg_state *src = &env->regs[insn->src_reg];
    
    switch (BPF_OP(insn->code)) {
    case BPF_MOV | BPF_K:
        // dst = imm(常量)
        dst->type = CONST_IMM;
        dst->min_value = dst->max_value = insn->imm;
        break;
        
    case BPF_ADD | BPF_X:
        // dst = dst + src
        if (dst->type == CONST_IMM && src->type == CONST_IMM) {
            dst->min_value += src->min_value;
            dst->max_value += src->max_value;
        } else {
            dst->type = UNKNOWN;
        }
        break;
        
    case BPF_LDX | BPF_MEM | BPF_W:
        // dst = *(u32 *)(src + off)
        if (src->type != PTR_TO_STACK && src->type != PTR_TO_MAP_VALUE) {
            pr_err("bpf: invalid memory access\n");
            return -EINVAL;
        }
        // ★ 边界检查:确保访问在合法范围内
        if (src->type == PTR_TO_STACK) {
            if (insn->off < -MAX_BPF_STACK || insn->off >= 0) {
                pr_err("bpf: stack access out of bounds\n");
                return -EINVAL;
            }
        }
        dst->type = UNKNOWN; // 从内存读取的值类型未知
        break;
        
    case BPF_JMP | BPF_JGT | BPF_K:
        // if dst > imm goto +off
        // ★ 分支后寄存器状态分化
        // ...
        break;
        
    // ... 其他指令
    }
    
    return 0;
}

// ★ JIT编译:eBPF → x86_64原生代码
int bpf_jit_compile(struct bpf_prog *prog) {
    struct jit_context ctx;
    
    for (int i = 0; i < prog->len; i++) {
        struct bpf_insn *insn = &prog->insns[i];
        
        switch (BPF_OP(insn->code)) {
        case BPF_MOV | BPF_K:
            // mov $imm, %dst
            emit_mov_imm(&ctx, insn->dst_reg, insn->imm);
            break;
            
        case BPF_ADD | BPF_X:
            // add %src, %dst
            emit_add_reg(&ctx, insn->dst_reg, insn->src_reg);
            break;
            
        case BPF_JMP | BPF_JEQ | BPF_K:
            // cmp $imm, %dst; je +off
            emit_cmp_imm(&ctx, insn->dst_reg, insn->imm);
            emit_jcc(&ctx, X86_JE, insn->off);
            break;
            
        // ... 其他指令的JIT实现
        }
    }
    
    // ★ 刷新指令缓存
    flush_icache_range(ctx.start, ctx.end);
    prog->bpf_func = ctx.start;
    
    return 0;
}
ftrace函数追踪
// kernel/trace/ftrace.c

// ★ ftrace钩子:每个函数入口调用
notrace void ftrace_stub(unsigned long ip, unsigned long parent_ip,
                         struct ftrace_ops *ops, struct pt_regs *regs) {
    struct ftrace_ops *op;
    
    // ★ 遍历注册的tracer
    for_each_ftrace_op(op) {
        if (op->func)
            op->func(ip, parent_ip, op, regs);
    }
}

// ★ function tracer回调
static void function_trace_call(unsigned long ip, unsigned long parent_ip,
                                struct ftrace_ops *ops, struct pt_regs *regs) {
    struct trace_array *tr = ops->private;
    struct ring_buffer *buffer = tr->trace_buffer.buffer;
    struct ring_buffer_event *event;
    struct ftrace_entry *entry;
    
    // ★ 分配trace事件
    event = trace_buffer_lock_reserve(buffer, TRACE_FN,
                                      sizeof(*entry), 0, 0);
    if (!event)
        return;
    
    entry = ring_buffer_event_data(event);
    entry->ip = ip;
    entry->parent_ip = parent_ip;
    entry->pid = current->pid;
    
    // ★ 提交事件到环形缓冲区
    trace_buffer_unlock_commit(buffer, event, 0, 0);
    
    // ★ 唤醒读取tracefs的进程
    trace_wake_up();
}

// ★ 编译时插入mcount(通过-pg编译选项)
// 每个函数编译后:
//   func:
//     call mcount    ← 编译器插入
//     ...
//   mcount:
//     call ftrace_stub  ← 运行时patch为nop或ftrace_stub

// ★ 运行时启用/禁用追踪
static ssize_t tracing_on_write(struct file *filp, const char __user *ubuf,
                                size_t cnt, loff_t *ppos) {
    char buf;
    
    if (copy_from_user(&buf, ubuf, 1))
        return -EFAULT;
    
    if (buf == '1')
        ftrace_enable();   // patch mcount → ftrace_stub
    else
        ftrace_disable();  // patch mcount → nop
    
    return cnt;
}

// ★ tracefs文件系统接口
static const struct file_operations tracing_on_fops = {
    .read  = tracing_on_read,
    .write = tracing_on_write,
    .open  = tracing_on_open,
};

// 初始化时注册tracefs
static int __init ftrace_init(void) {
    struct dentry *d_tracer;
    
    d_tracer = tracing_init_dentry();
    if (!d_tracer)
        return -ENODEV;
    
    trace_create_file("tracing_on", 0644, d_tracer, NULL,
                      &tracing_on_fops);
    trace_create_file("trace", 0444, d_tracer, NULL,
                      &trace_fops);
    trace_create_file("available_tracers", 0444, d_tracer, NULL,
                      &show_tracers_fops);
    
    return 0;
}

关键细节解析

1. 为什么perf_events使用环形缓冲区而非普通缓冲区?

        因为环形缓冲区支持无锁并发读写。perf_events的场景是:内核中断上下文(生产者)持续写入采样数据,用户空间进程(消费者)持续读取。如果使用普通缓冲区,需要锁保护,而锁在中断上下文中会引入死锁风险。环形缓冲区通过单生产者单消费者模型 + 内存屏障实现无锁:生产者只更新head,消费者只更新tail,两者互不干扰。教学实现中必须使用atomic_t操作head/tail,并在读写之间插入smp_wmb()/smp_rmb()内存屏障。

2. 为什么eBPF验证器需要追踪寄存器状态?

        因为eBPF程序可以间接访问内存(如通过寄存器指针),验证器必须确保:①指针不会越界访问;②不会解引用空指针;③不会将指针泄露到用户空间。验证器通过抽象解释(abstract interpretation)模拟执行每条指令,追踪每个寄存器的类型(常量、栈指针、map指针等)和值范围。例如,若R1 = PTR_TO_STACK + 8,则*(u64 *)(R1 + 0)是合法的(访问栈偏移8),但*(u64 *)(R1 + 100)会被拒绝(越界)。教学实现中必须实现完整的寄存器状态机,包括分支后的状态分化与合并。

3. 为什么ftrace使用mcount而非动态插桩?

        因为mcount在编译时插入,零运行时开销(当追踪关闭时)。ftrace通过编译器-pg选项在每个函数入口插入call mcount。当追踪关闭时,mcount被patch为nop(单字节指令),CPU执行nop几乎无开销(~1周期)。当追踪开启时,mcount被patch为call ftrace_stub,调用tracer回调。相比之下,动态插桩(如kprobe)在运行时修改代码,需要中断所有CPU并同步指令缓存,开销较大。教学实现中必须理解mcount的编译时插入机制与运行时patch机制的分离。


调试Checklist:可观测性工具排查
症状可能原因排查方法
perf record无数据事件未配置/权限不足/PMU未初始化检查perf list可用事件;用perf stat -e cpu-cycles测试;验证PMU驱动加载;检查/proc/sys/kernel/perf_event_paranoid权限
eBPF程序加载失败验证器拒绝/JIT编译错误/map创建失败读取/sys/fs/bpf日志;用bpftool prog load测试;简化程序逐步排查;检查内核版本是否支持对应helper
ftrace无输出tracing_on未启用/filter过滤过严/tracefs未挂载cat /sys/kernel/debug/tracing/tracing_on确认状态;echo 0 > tracing_filter清除过滤;mount -t tracefs none /sys/kernel/debug/tracing
perf采样频率过低采样周期设置过大/PMU计数器被其他事件占用调整-c参数降低采样周期;用perf list确认事件可用性;检查是否多事件复用同一PMU计数器
eBPF map读写失败map类型不支持/键值大小不匹配/权限不足验证map创建参数;用bpftool map检查map状态
Logo

openEuler 是由开放原子开源基金会孵化的全场景开源操作系统项目,面向数字基础设施四大核心场景(服务器、云计算、边缘计算、嵌入式),全面支持 ARM、x86、RISC-V、loongArch、PowerPC、SW-64 等多样性计算架构

更多推荐