【eBPF 内核实现深度拆解】Map 内核实现（下）：ringbuf / perfbuf / bloom / queue-stack / LPM

/* kernel/bpf/ringbuf.c — Linux 6.8 */
struct bpf_ringbuf {
    wait_queue_head_t waitq;             /* epoll 等待队列 */
    struct irq_work work;                /* 跨 CPU 唤醒机制 */
    u64 mask;                            /* 缓冲区大小掩码（2 的幂 - 1） */
    struct page **pages;                 /* 数据页面数组 */
    int nr_pages;                        /* 页面数 */
    spinlock_t spinlock;                 /* 保留操作的自旋锁 */
    atomic_t busy;                       /* 忙位：标记记录正在写入 */
    unsigned long consumer_pos;          /* 消费者已读位置 */
    unsigned long producer_pos;          /* 生产者已写入位置 */
    unsigned long pending_pos;           /* 已提交但未通知消费者的位置 */
    struct bpf_ringbuf_map *ringbuf_map; /* 关联的 ringbuf map */
};

struct bpf_ringbuf_map {
    struct bpf_map map;
    struct bpf_ringbuf *rb;
};

/* 用户态 mmap 的伪结构 */
struct ringbuf_consumer {
    unsigned long *consumer_pos;    /* mmap page 0 */
    unsigned long *producer_pos;    /* mmap page 0 */
    void *data;                     /* mmap page 1+ */
    unsigned long mask;             /* data 区域大小 - 1 */
};

/* kernel/bpf/ringbuf.c — 忙位定义 */
#define BPF_RINGBUF_BUSY_BIT  (1 << 31)
#define BPF_RINGBUF_DISCARD_BIT (1 << 30)

/* kernel/bpf/ringbuf.c — bpf_ringbuf_reserve() 简化 */
static void *bpf_ringbuf_reserve(struct bpf_map *map, u64 size, u64 flags)
{
    struct bpf_ringbuf_map *rb_map = container_of(map, ...);
    struct bpf_ringbuf *rb = rb_map->rb;
    u64 cons_pos, prod_pos, new_prod_pos;
    void *rec;

    if (size > RINGBUF_MAX_RECORD_SZ)
        return NULL;

    /* 1. 计算对齐后的大小（8 字节对齐） */
    u32 total_size = round_up(size + BPF_RINGBUF_HDR_SZ, 8);

    /* 2. 尝试在缓冲区中保留空间 */
    prod_pos = smp_load_acquire(&rb->producer_pos);
    new_prod_pos = prod_pos + total_size;

    /* 3. 检查空间：如果 new_prod_pos 越过 consumer_pos，缓冲区满 */
    cons_pos = smp_load_acquire(&rb->consumer_pos);
    if (new_prod_pos - cons_pos > rb->mask + 1)
        return NULL;  /* 缓冲区满 */

    /* 4. 获取 record 指针并设置头部 */
    rec = (void *)rb->pages[1] + (prod_pos & rb->mask);
    struct bpf_ringbuf_hdr *hdr = rec;
    hdr->len = size;
    hdr->pg_off = 0;
    hdr->len_discard = BPF_RINGBUF_BUSY_BIT;  /* 设置忙位！ */

    /* 5. 更新 producer_pos（带 release 语义） */
    smp_store_release(&rb->producer_pos, new_prod_pos);

    return rec + BPF_RINGBUF_HDR_SZ;          /* 返回 payload 指针 */
}

/* kernel/bpf/ringbuf.c — bpf_ringbuf_submit() 简化 */
static void bpf_ringbuf_submit(void *data, u64 flags)
{
    struct bpf_ringbuf_hdr *hdr = data - BPF_RINGBUF_HDR_SZ;

    /* 清除忙位：此操作使记录对消费者可见 */
    smp_store_release(&hdr->len_discard,
                       hdr->len_discard & ~BPF_RINGBUF_BUSY_BIT);
}

/* 用户态消费逻辑（伪代码） */
static int consume_ringbuf(struct ringbuf_consumer *c,
                           void (*callback)(void *data, size_t len))
{
    unsigned long cons_pos = *c->consumer_pos;
    unsigned long prod_pos = smp_load_acquire(c->producer_pos);

    while (cons_pos < prod_pos) {
        struct bpf_ringbuf_hdr *hdr = c->data + (cons_pos & c->mask);

        /* 检查忙位：如果设置了，记录还在写入中，跳过 */
        if (hdr->len_discard & BPF_RINGBUF_BUSY_BIT) {
            /* 更新 consumer_pos 以跳过忙的记录 */
            smp_store_release(c->consumer_pos, cons_pos);
            return 0;
        }

        /* 检查丢弃位 */
        if (hdr->len_discard & BPF_RINGBUF_DISCARD_BIT) {
            /* 此记录被丢弃（bpf_ringbuf_discard），跳过 */
            cons_pos += round_up(hdr->len_discard & ~BPF_RINGBUF_DISCARD_BIT, 8);
            continue;
        }

        /* 调用回调处理数据 */
        callback(hdr + 1, hdr->len);

        cons_pos += round_up(hdr->len + BPF_RINGBUF_HDR_SZ, 8);
    }

    /* 更新 consumer_pos */
    smp_store_release(c->consumer_pos, cons_pos);
    return 0;
}

/* kernel/bpf/ringbuf.c — bpf_ringbuf_discard() 简化 */
static void bpf_ringbuf_discard(void *data, u64 flags)
{
    struct bpf_ringbuf_hdr *hdr = data - BPF_RINGBUF_HDR_SZ;

    /* 设置丢弃位并清除忙位 */
    hdr->len_discard |= BPF_RINGBUF_DISCARD_BIT;
    smp_store_release(&hdr->len_discard,
                       hdr->len_discard & ~BPF_RINGBUF_BUSY_BIT);
}

/* libbpf: ring_buffer__new() 创建消费者 */
struct ring_buffer *ring_buffer__new(int map_fd,
                                     ring_buffer_sample_fn sample_cb,
                                     void *ctx,
                                     const struct ring_buffer_opts *opts);

/* 在事件循环中调用 */
int ring_buffer__consume(struct ring_buffer *rb);
int ring_buffer__poll(struct ring_buffer *rb, int timeout_ms);

/* kernel/trace/bpf_trace.c — perf_event_array_map_update_elem() 简化 */
static long perf_event_array_map_update_elem(struct bpf_map *map, void *key,
                                              void *value, u64 map_flags)
{
    struct bpf_array *array = container_of(map, struct bpf_array, map);
    u32 index = *(u32 *)key;

    /* value 是一个 fd，内核将其转换为 struct perf_event * */
    struct perf_event *event = perf_event_get_from_fd(*(int *)value);
    if (IS_ERR(event))
        return PTR_ERR(event);

    WRITE_ONCE(array->ptrs[index], event);  /* 存储 perf_event 指针 */
    return 0;
}

/* kernel/trace/bpf_trace.c — bpf_perf_event_output() 简化逻辑 */
/*
 * BPF_CALL_5(bpf_perf_event_output, struct pt_regs *, regs,
 *            struct bpf_map *, map, u64, flags, void *, data, u64, size)
 *
 * 执行路径:
 * 1. 从 map 获取 perf_event *（通过 index 或 key）
 * 2. 调用 perf_event_output(event, data, size)
 *    → perf_output_begin(&handle, event, size)  // 保留空间
 *    → perf_output_copy(&handle, data, size)    // 复制数据
 *    → perf_output_end(&handle)                  // 提交
 * 3. perf_output_begin 内部:
 *    - 检查缓冲区空间
 *    - 写入 perf_event_header（type, size, misc）
 *    - 如果缓冲区满，返回 NULL
 */

/* include/uapi/linux/perf_event.h */
struct perf_event_header {
    __u32   type;         /* PERF_RECORD_SAMPLE 等 */
    __u16   misc;         /* 混杂标志（CPU、时间戳模式等） */
    __u16   size;         /* 记录总大小 */
};

/* kernel/bpf/bloom_filter.c — Linux 6.8 */
struct bpf_bloom_filter {
    struct bpf_map map;
    u32 bitset_mask;                     /* 位图大小掩码：nr_hash_funcs * BITS_PER_LONG - 1 */
    u32 hash_seed;                       /* 哈希种子 */
    u32 nr_hash_funcs;                   /* 哈希函数数量 */
    /* 紧随: unsigned long bitset[]  ——实际位图 */
    unsigned long bitset[];
};

/* kernel/bpf/bloom_filter.c */
__bpf_kfunc u64 bpf_bloom_filter_lookup(void *map, void *data,
                                         u32 data_len, void *data2,
                                         u32 data2_len)
{
    struct bpf_bloom_filter *bloom = container_of(*(struct bpf_map **)map, ...);
    u32 n = bloom->nr_hash_funcs;
    u32 i;

    for (i = 0; i < n; i++) {
        /* 使用 jhash2 计算第 i 个 hash */
        u32 hash = jhash2((const u32 *)data, data_len / 4,
                          bloom->hash_seed + i);
        /* 检查对应位是否设置 */
        if (!test_bit(hash & bloom->bitset_mask, bloom->bitset))
            return 0;  /* 确定不存在 */
    }

    return 1;  /* 可能存在（false positive 可能） */
}

__bpf_kfunc void bpf_bloom_filter_push(void *map, void *data,
                                        u32 data_len, void *data2,
                                        u32 data2_len)
{
    struct bpf_bloom_filter *bloom = ...;
    u32 n = bloom->nr_hash_funcs;
    u32 i;

    for (i = 0; i < n; i++) {
        u32 hash = jhash2(...);
        set_bit(hash & bloom->bitset_mask, bloom->bitset);
    }
}

/* BPF 程序：用 bloom filter 去重连接 */
struct {
    __uint(type, BPF_MAP_TYPE_BLOOM_FILTER);
    __type(value, __u32);
    __uint(max_entries, 1 << 20);       /* 1M 位 */
    __uint(map_extra, 4);               /* 4 个哈希函数 */
    __uint(nr_hash_funcs, 4);
} conn_seen SEC(".maps");

SEC("kprobe/tcp_connect")
int filter_new_conn(struct pt_regs *ctx)
{
    struct sock *sk = (struct sock *)PT_REGS_PARM1(ctx);
    __u32 daddr = BPF_CORE_READ(sk, __sk_common.skc_daddr);
    __u16 dport = BPF_CORE_READ(sk, __sk_common.skc_dport);

    /* 先查 bloom filter：这个连接之前见过吗？ */
    if (bpf_bloom_filter_lookup(&conn_seen, &daddr, sizeof(daddr),
                                 &dport, sizeof(dport)))
        return 0;  /* 可能见过，跳过 */

    /* 确实没见过，加入并处理 */
    bpf_bloom_filter_push(&conn_seen, &daddr, sizeof(daddr),
                           &dport, sizeof(dport));
    /* ... 记录新连接事件 ... */
    return 0;
}

/* kernel/bpf/queue_stack_maps.c — Linux 6.8 */
struct bpf_queue_stack {
    struct bpf_map map;
    raw_spinlock_t lock;
    u64 head;                            /* 队列头 / 栈顶 */
    u64 tail;                            /* 队列尾 */
    u64 size;                            /* 当前元素数 */
    u32 elem_size;                       /* 每个元素的大小（含填充） */
    /* 紧随：value 数组 */
    char elements[] __aligned(8);
};

/* kernel/bpf/queue_stack_maps.c — push / pop / peek */
static long queue_stack_map_push_elem(struct bpf_map *map, void *value, u64 flags)
{
    struct bpf_queue_stack *qs = container_of(map, ...);
    raw_spin_lock_irqsave(&qs->lock, flags);

    if (qs->size >= qs->map.max_entries) {
        ret = -ENOSPC;  /* 队列满 */
        goto out;
    }

    /* 复制 value 到环形缓冲区的尾部 */
    memcpy(&qs->elements[qs->tail * qs->elem_size], value, qs->map.value_size);
    qs->tail = (qs->tail + 1) % qs->map.max_entries;
    qs->size++;

out:
    raw_spin_unlock_irqrestore(&qs->lock, flags);
    return ret;
}

/* 例：BPF 程序 A 收集事件，尾调用程序 B 批量处理 */
struct {
    __uint(type, BPF_MAP_TYPE_QUEUE);
    __type(value, struct event);
    __uint(max_entries, 1024);
} event_queue SEC(".maps");

/* 程序 A：推入事件 */
SEC("kprobe/some_func")
int producer(struct pt_regs *ctx)
{
    struct event evt = {};
    evt.pid = bpf_get_current_pid_tgid() >> 32;
    /* ... 填充其他字段 ... */
    bpf_map_push_elem(&event_queue, &evt, BPF_EXIST);
    return 0;
}

/* 程序 B：批处理事件（通过尾调用触发） */
SEC("kprobe/another_func")
int consumer(struct pt_regs *ctx)
{
    struct event evt;
    while (bpf_map_pop_elem(&event_queue, &evt) == 0) {
        /* 处理事件 */
    }
    return 0;
}

/* include/uapi/linux/bpf.h */
struct bpf_lpm_trie_key {
    __u32   prefixlen;   /* 前缀长度（位数） */
    __u8    data[];      /* 实际 key 数据（仅前 prefixlen 位有效） */
};

/* kernel/bpf/lpm_trie.c — Linux 6.8 */
struct lpm_trie_node {
    struct rcu_head rcu;
    struct lpm_trie_node __rcu *child[2];  /* 两个子节点（bit 0 和 bit 1） */
    u32 prefixlen;                          /* 该节点的前缀长度 */
    u32 flags;
    u8 data[];                              /* key 前缀 + value */
};

struct lpm_trie {
    struct bpf_map map;
    struct lpm_trie_node __rcu *root;       /* 根节点 */
    size_t n_entries;                        /* 节点数 */
    size_t max_prefixlen;                    /* 最大前缀长度（bits） */
    size_t data_size;                        /* 每个节点的总数据大小 */
    raw_spinlock_t lock;
};

/* kernel/bpf/lpm_trie.c — trie_update_elem() 简化逻辑 */
/*
 * 对于要插入的 key（prefixlen=P, data=D）：
 *
 * 1. 从根节点开始遍历
 * 2. 在每个节点 N：
 *    a. 计算 N 的 key 与新 key 的公共前缀长度（common prefix length）
 *    b. 如果 common >= N->prefixlen：确定下一个分支位
 *       - 用新 key 在那一位的 bit 值决定走 child[0] 或 child[1]
 *       - 继续向下
 *    c. 如果 common < N->prefixlen：
 *       - 需要在此位置插入一个中间节点
 *       - 中间节点的 prefixlen = common
 *       - 根据 common+1 位的值，将 N 和新节点分别挂在中间节点的两个 child 上
 *
 * 3. 到达叶子节点或找到匹配位置：
 *    - 如果 prefixlen 相同：替换 value
 *    - 如果 prefixlen 不同：在正确位置插入
 *
 * 整个操作在 trie->lock 下进行。
 */

/* kernel/bpf/lpm_trie.c — trie_lookup_elem() 简化 */
static void *trie_lookup_elem(struct bpf_map *map, void *_key)
{
    struct lpm_trie *trie = container_of(map, struct lpm_trie, map);
    struct bpf_lpm_trie_key *key = _key;
    struct lpm_trie_node *node, *found = NULL;

    /* RCU 读锁已由 BPF 运行上下文的 preempt_disable 保证 */
    node = rcu_dereference_check(trie->root, rcu_read_lock_bh_held());

    /* 沿 trie 向下，记录沿途所有匹配节点 */
    while (node) {
        /* 计算 public prefix：查找 key 的前缀与该节点 key 的公共部分 */
        u32 limit = min(key->prefixlen, node->prefixlen);
        u32 prefixlen = longest_prefix_match(trie, node, key);

        if (prefixlen >= limit) {
            /* 如果完全匹配该节点的前缀 */
            if (node->prefixlen >= key->prefixlen) {
                /* 节点比查找 key 更具体：可以返回 */
                found = node;
            }
            /* 继续向下查找 */
            u32 next_bit = extract_bit(key->data, node->prefixlen);
            node = rcu_dereference_check(node->child[next_bit], ...);
        } else {
            /* 不匹配：这条路径上没有更多可能 */
            break;
        }
    }

    if (found)
        return found->data + trie->data_size;
    return NULL;
}

/* 例：BPF 路由表 */
struct {
    __uint(type, BPF_MAP_TYPE_LPM_TRIE);
    __type(key, struct ipv4_lpm_key);     /* { u32 prefixlen; u32 addr; } */
    __type(value, struct route_info);      /* { u32 ifindex; u8 dmac[6]; } */
    __uint(max_entries, 65536);
    __uint(map_flags, BPF_F_NO_PREALLOC);
} route_table SEC(".maps");

/* 查找：匹配最长前缀路由 */
SEC("xdp")
int xdp_router(struct xdp_md *ctx)
{
    struct ipv4_lpm_key key = {
        .prefixlen = 32,
        .addr = dst_ip,
    };
    struct route_info *ri;

    ri = bpf_map_lookup_elem(&route_table, &key);  /* LPM lookup */
    if (ri) {
        return bpf_redirect(ri->ifindex, 0);
    }
    return XDP_PASS;
}

/* kernel/bpf/devmap.c — struct bpf_devmap_val */
struct bpf_devmap_val {
    __u32 ifindex;       /* 目标网络接口 */
    union {
        int   fd;        /* bpf_prog FD（可选：在重定向前执行） */
        __u32 id;        /* bpf_prog ID */
    } bpf_prog;
};

/* kernel/bpf/cpumap.c — struct bpf_cpumap_val */
struct bpf_cpumap_val {
    __u32 qsize;         /* 每 CPU 队列大小 */
    union {
        int   fd;        /* 在目标 CPU 上执行的 BPF 程序 */
        __u32 id;
    } bpf_prog;
};

/* kernel/bpf/xskmap.c — struct bpf_xskmap_val */
struct xskmap_entry {
    struct xdp_sock __rcu *xsk;
    struct xdp_sock __rcu *xsk_map_entry;
};