分类：ebpf

xdp 使用ebpf 做 包过滤，相对于dpdk将数据包直接送到用户态，用用户态当做快速数据处理平面，xdp是在驱动层创建了一个数据快速平面。在数据被网卡硬件dma到内存，分配skb之前，对数据包进行处理。由于完全不存在锁操作。且bypass了协议栈，非常适合用修改数据包并转发，数据探针，执行丢包。
xdp 程序执行的位置实在ixgbe_clean_rx_irq函数中。网卡刚把数据dma到内存时。

        while (likely(total_rx_packets < budget)) {
                union ixgbe_adv_rx_desc *rx_desc;
                struct ixgbe_rx_buffer *rx_buffer;
                struct sk_buff *skb;
                unsigned int size;

                /* return some buffers to hardware, one at a time is too slow */
                if (cleaned_count >= IXGBE_RX_BUFFER_WRITE) {
                        ixgbe_alloc_rx_buffers(rx_ring, cleaned_count);
                        cleaned_count = 0;
                }

                rx_desc = IXGBE_RX_DESC(rx_ring, rx_ring->next_to_clean);
                size = le16_to_cpu(rx_desc->wb.upper.length);
                if (!size)
                        break;

                /* This memory barrier is needed to keep us from reading
                 * any other fields out of the rx_desc until we know the
                 * descriptor has been written back
                 */
                dma_rmb();

                rx_buffer = ixgbe_get_rx_buffer(rx_ring, rx_desc, &skb, size);

                /* retrieve a buffer from the ring */
                if (!skb) {
                        xdp.data = page_address(rx_buffer->page) +
                                   rx_buffer->page_offset;
                        xdp.data_meta = xdp.data;
                        xdp.data_hard_start = xdp.data -
                                              ixgbe_rx_offset(rx_ring);
                        xdp.data_end = xdp.data + size;

                        skb = ixgbe_run_xdp(adapter, rx_ring, &xdp);            //执行run_xdp函数
                }

ixgbe_run_xdp 函数。

static struct sk_buff *ixgbe_run_xdp(struct ixgbe_adapter *adapter,
                                     struct ixgbe_ring *rx_ring,
                                     struct xdp_buff *xdp)
{
        int err, result = IXGBE_XDP_PASS;
        struct bpf_prog *xdp_prog;
        u32 act;

        rcu_read_lock();
        xdp_prog = READ_ONCE(rx_ring->xdp_prog);

        if (!xdp_prog)              //如果没有注册xdp程序，则直接跳出处理流程
                goto xdp_out;

        act = bpf_prog_run_xdp(xdp_prog, xdp);                  //执行xdp虚拟机。 xdp_prog 是 ebpf程序体， xdp是出入给ebpf程序的参数
        switch (act) {
        case XDP_PASS:              //xdp处理结果是将数据上送协议栈
                break;
        case XDP_TX:                //xdp处理结果是将数据通过同一个口转发出去
                result = ixgbe_xmit_xdp_ring(adapter, xdp);
                break;
        case XDP_REDIRECT:          //xdp处理结果是从另一个网口转发出去
                err = xdp_do_redirect(adapter->netdev, xdp, xdp_prog);
                if (!err)
                        result = IXGBE_XDP_TX;
                else    
                        result = IXGBE_XDP_CONSUMED;
                break;
        default:
                bpf_warn_invalid_xdp_action(act);
                /* fallthrough */
        case XDP_ABORTED:               //xdp程序异常丢掉数据包
                trace_xdp_exception(rx_ring->netdev, xdp_prog, act);
                /* fallthrough -- handle aborts by dropping packet */
        case XDP_DROP:                  //数据包被丢弃。
                result = IXGBE_XDP_CONSUMED;
                break;
        }
xdp_out:
        rcu_read_unlock();
        return ERR_PTR(-result);
}

设置xdp程序函数。在intel 82599网卡中使用的是 ixgbe_xdp函数。 ixgbe_xdp函数最后赋值给 net_device_ops结构体的ndo_bpf函数变量中。

static int ixgbe_xdp(struct net_device *dev, struct netdev_bpf *xdp)
{       
        struct ixgbe_adapter *adapter = netdev_priv(dev);

        switch (xdp->command) {
        case XDP_SETUP_PROG:
                return ixgbe_xdp_setup(dev, xdp->prog);         //设置xdp程序，把xdp程序设置到rx_ring的 xdp_prog变量中。
        case XDP_QUERY_PROG:
                xdp->prog_attached = !!(adapter->xdp_prog);
                xdp->prog_id = adapter->xdp_prog ?
                        adapter->xdp_prog->aux->id : 0;
                return 0;
        default:
                return -EINVAL;
        }
}

在 net/core/rtnetlink.c文件中的rtnl_setlink函数中会调用最后会调用 do_setlink函数。 rtnl_setlink 关联到 PF_UNSPEC netlink 的 RTM_SETLINK 中，用户态发送这类消息时，直接调用rtnl_setlink函数处理。
do_setlink函数 最后会调用dev_change_xdp_fd函数

        if (tb[IFLA_XDP]) {
                struct nlattr *xdp[IFLA_XDP_MAX + 1];
                u32 xdp_flags = 0;

                err = nla_parse_nested(xdp, IFLA_XDP_MAX, tb[IFLA_XDP],
                                       ifla_xdp_policy, NULL);
                if (err < 0)
                        goto errout;

                if (xdp[IFLA_XDP_ATTACHED] || xdp[IFLA_XDP_PROG_ID]) {
                        err = -EINVAL;
                        goto errout;
                }

                if (xdp[IFLA_XDP_FLAGS]) {
                        xdp_flags = nla_get_u32(xdp[IFLA_XDP_FLAGS]);
                        if (xdp_flags & ~XDP_FLAGS_MASK) {
                                err = -EINVAL;
                                goto errout;
                        }
                        if (hweight32(xdp_flags & XDP_FLAGS_MODES) > 1) {
                                err = -EINVAL;
                                goto errout;
                        }
                }

                if (xdp[IFLA_XDP_FD]) {
                        err = dev_change_xdp_fd(dev, extack,
                                                nla_get_s32(xdp[IFLA_XDP_FD]),
                                                xdp_flags);         //修改指定dev的 prog
                        if (err)
                                goto errout;
                        status |= DO_SETLINK_NOTIFY;
                }
        }

dev_change_xdp_fd函数做了一系列检查后最后调用 dev_xdp_install函数

int dev_change_xdp_fd(struct net_device *dev, struct netlink_ext_ack *extack,
                      int fd, u32 flags)
{       
        const struct net_device_ops *ops = dev->netdev_ops;
        struct bpf_prog *prog = NULL;
        bpf_op_t bpf_op, bpf_chk;
        int err;

        ASSERT_RTNL();

        bpf_op = bpf_chk = ops->ndo_bpf;                //获取网卡驱动ndo_bpf程序 ixgbe网卡的函数是 ixgbe_bpf
        if (!bpf_op && (flags & (XDP_FLAGS_DRV_MODE | XDP_FLAGS_HW_MODE)))
                return -EOPNOTSUPP;
        if (!bpf_op || (flags & XDP_FLAGS_SKB_MODE))
                bpf_op = generic_xdp_install;
        if (bpf_op == bpf_chk)
                bpf_chk = generic_xdp_install;

        if (fd >= 0) {
                if (bpf_chk && __dev_xdp_attached(dev, bpf_chk))
                        return -EEXIST;
                if ((flags & XDP_FLAGS_UPDATE_IF_NOEXIST) &&
                    __dev_xdp_attached(dev, bpf_op))
                        return -EBUSY;

                prog = bpf_prog_get_type_dev(fd, BPF_PROG_TYPE_XDP,
                                             bpf_op == ops->ndo_bpf);
                if (IS_ERR(prog))
                        return PTR_ERR(prog);

                if (!(flags & XDP_FLAGS_HW_MODE) &&
                    bpf_prog_is_dev_bound(prog->aux)) {
                        NL_SET_ERR_MSG(extack, "using device-bound program without HW_MODE flag is not supported");
                        bpf_prog_put(prog);
                        return -EINVAL;
                }
        }

        err = dev_xdp_install(dev, bpf_op, extack, flags, prog);        //将xdp ebpf程序安装到网卡的 rx_ring中
        if (err < 0 && prog)
                bpf_prog_put(prog);

        return err;
}

dev_xdp_install 函数代码如下

static int dev_xdp_install(struct net_device *dev, bpf_op_t bpf_op,
                           struct netlink_ext_ack *extack, u32 flags,
                           struct bpf_prog *prog)
{               
        struct netdev_bpf xdp;

        memset(&xdp, 0, sizeof(xdp));
        if (flags & XDP_FLAGS_HW_MODE)
                xdp.command = XDP_SETUP_PROG_HW;
        else    
                xdp.command = XDP_SETUP_PROG;
        xdp.extack = extack;
        xdp.flags = flags;
        xdp.prog = prog;

        return bpf_op(dev, &xdp);           //执行ndo_xdp函数， ixgbe驱动的函数为ixgbe_bpf
} 

至此，xdp 运行位置及用户程序将ebpf代码attach到网卡的流程都分析完毕。

ebpf maps 内核态代码分析
创建maps BPF_MAP_CREATE

          map = find_and_alloc_map(attr);                   //根据map类型创建map对象
          if (IS_ERR(map))
                  return PTR_ERR(map);

          err = bpf_obj_name_cpy(map->name, attr->map_name);
          if (err)
                  goto free_map_nouncharge;

          atomic_set(&map->refcnt, 1);
          atomic_set(&map->usercnt, 1);

          err = security_bpf_map_alloc(map);
          if (err)
                  goto free_map_nouncharge;

          err = bpf_map_charge_memlock(map);
          if (err)
                  goto free_map_sec;

          err = bpf_map_alloc_id(map);              //将map存放在idr数据结构中
          if (err)
                  goto free_map;

          err = bpf_map_new_fd(map, f_flags);       //将map映射成fd文件返回给用户态的fd文件描述符
          if (err < 0) {
                  /* failed to allocate fd.
                   * bpf_map_put() is needed because the above
                   * bpf_map_alloc_id() has published the map
                   * to the userspace and the userspace may
                   * have refcnt-ed it through BPF_MAP_GET_FD_BY_ID.
                   */
                  bpf_map_put(map);
                  return err;
          }

          trace_bpf_map_create(map, err);
          return err;               //返回文件描述符

查找更新和删除，没什么好说的，代码写的很清楚。代码在kernel/bpf/syscall.c文件中，入口函数分别是map_lookup_elem map_update_elem map_delete_elem。
比较关键是，内核创建的map 对象怎么跟ebpf程序关联起来。这个关联在ebpf程序加载的时候进行的。如果代码时bpf_prog_load函数

        err = find_prog_type(type, prog);
        if (err < 0)
                goto free_prog;

        prog->aux->load_time = ktime_get_boot_ns();
        err = bpf_obj_name_cpy(prog->aux->name, attr->prog_name);
        if (err)
                goto free_prog;

        /* run eBPF verifier */
        err = bpf_check(&prog, attr);                       //这个函数对ebpf程序进行检查并将与fd关联的map对象替换到ebpf程序中
        if (err < 0)
                goto free_used_maps;

        /* eBPF program is ready to be JITed */
        if (!prog->bpf_func)
                prog = bpf_prog_select_runtime(prog, &err);
        if (err < 0)
                goto free_used_maps;

        err = bpf_prog_alloc_id(prog);
        if (err)
                goto free_used_maps;

        err = bpf_prog_new_fd(prog);

关键函数是replace_map_fd_with_map_ptr.

        env->strict_alignment = !!(attr->prog_flags & BPF_F_STRICT_ALIGNMENT);
        if (!IS_ENABLED(CONFIG_HAVE_EFFICIENT_UNALIGNED_ACCESS))
                env->strict_alignment = true;

        if (bpf_prog_is_dev_bound(env->prog->aux)) {
                ret = bpf_prog_offload_verifier_prep(env);
                if (ret)
                        goto err_unlock;
        }

        ret = replace_map_fd_with_map_ptr(env);                 //将文件描述符替换成map指针地址
        if (ret < 0)
                goto skip_full_check;

        env->explored_states = kcalloc(env->prog->len,
                                       sizeof(struct bpf_verifier_state_list *),
                                       GFP_USER);
        ret = -ENOMEM;
        if (!env->explored_states)
                goto skip_full_check;

        env->allow_ptr_leaks = capable(CAP_SYS_ADMIN);

        ret = check_cfg(env);

replace_map_fd_with_map_ptr函数中可以看到会获取文件描述符，并根据文件描述符获取文件，最后获取文件的private_data即 map的地址。

                        f = fdget(insn->imm);
                        map = __bpf_map_get(f);             //获取map对象地址
                        if (IS_ERR(map)) {
                                verbose(env, "fd %d is not pointing to valid bpf_map\n",
                                        insn->imm);
                                return PTR_ERR(map);
                        }

                        err = check_map_prog_compatibility(env, map, env->prog);
                        if (err) {
                                fdput(f);
                                return err;
                        }

                        /* store map pointer inside BPF_LD_IMM64 instruction */
                        insn[0].imm = (u32) (unsigned long) map;
                        insn[1].imm = ((u64) (unsigned long) map) >> 32;            //将地址赋值到insn中，这样ebpf虚拟机可以直接访问map。

                        /* check whether we recorded this map already */
                        for (j = 0; j < env->used_map_cnt; j++)
                                if (env->used_maps[j] == map) {
                                        fdput(f);
                                        goto next_insn;
                                }

                        if (env->used_map_cnt >= MAX_USED_MAPS) {
                                fdput(f);
                                return -E2BIG;
                        }

至此， ebpf map 在用户态和内核传递数据的方式基本上已经搞情况了。

最近 facebook 开源了一份 L4LB katran 基于 xdp做的。 里面大量使用了 ebpf maps 进行用户态和内核态进行数据交互。今天dig了一下代码大概了解一下原理。 使用的代码时 kernel/sample/bpf/中的代码。
ebpf 代码部分：
在ebpf代码中需要申明map

struct bpf_map_def SEC("maps") port_a = {           //SEC（"maps"） 这个是llvm生成代码的section标记，表示为这段代码单独生成一个maps code setion。 在load_bpf代码时，会根据这个标记找到这个section 然后在内核创建相应的内存。
        .type = BPF_MAP_TYPE_ARRAY,                 //map的类型  有很多当前为array类型
        .key_size = sizeof(u32),                    //key 的长度
        .value_size = sizeof(int),                  //value 的长度
        .max_entries = MAX_NR_PORTS,                //这个array的最大长度。
};

编译完成后 会生成一个 elf的.o文件，然后在进程上下文中解析elf文件获取maps信息，代码如下

        fd = open(path, O_RDONLY, 0);               //打开llvm 编译后的test_map_in_kernel.o文件
        if (fd < 0)
                return 1;

        elf = elf_begin(fd, ELF_C_READ, NULL);

        if (!elf)
                return 1;

        if (gelf_getehdr(elf, &ehdr) != &ehdr)      //获取所有section head信息
                return 1;

        /* clear all kprobes */
        i = system("echo \"\" > /sys/kernel/debug/tracing/kprobe_events");

        /* scan over all elf sections to get license and map info */
        for (i = 1; i < ehdr.e_shnum; i++) {

                if (get_sec(elf, i, &ehdr, &shname, &shdr, &data))
                        continue;

                if (0) /* helpful for llvm debugging */
                        printf("section %d:%s data %p size %zd link %d flags %d\n",
                               i, shname, data->d_buf, data->d_size,
                               shdr.sh_link, (int) shdr.sh_flags);

                if (strcmp(shname, "license") == 0) {
                        processed_sec[i] = true;
                        memcpy(license, data->d_buf, data->d_size);
                } else if (strcmp(shname, "version") == 0) {
                        processed_sec[i] = true;
                        if (data->d_size != sizeof(int)) {
                                printf("invalid size of version section %zd\n",
                                       data->d_size);
                                return 1;
                        }
                        memcpy(&kern_version, data->d_buf, sizeof(int));
                } else if (strcmp(shname, "maps") == 0) {           //如果section 头部名字是maps，则说明 test_map_in_kernel中有map
                        int j;

                        maps_shndx = i;
                        data_maps = data;
                        for (j = 0; j < MAX_MAPS; j++)
                                map_data[j].fd = -1;
                } else if (shdr.sh_type == SHT_SYMTAB) {
                        strtabidx = shdr.sh_link;
                        symbols = data;
                }
        }

分析 maps section 并在内核创建map 对象，对外表现为文件类型

        if (data_maps) {
                nr_maps = load_elf_maps_section(map_data, maps_shndx,
                                                elf, symbols, strtabidx);           //解析maps section 获取每个定义map的对象
                if (nr_maps < 0) {
                        printf("Error: Failed loading ELF maps (errno:%d):%s\n",
                               nr_maps, strerror(-nr_maps));
                        ret = 1;
                        goto done;
                }
                if (load_maps(map_data, nr_maps, fixup_map))                //加载map并在内核创建相应的对象。返回文件描述符
                        goto done;
                map_data_count = nr_maps;

                processed_sec[maps_shndx] = true;
        }

load_elf_maps_section 主要是讲maps 对象的定义读取出来，到load_maps中使用，关键代码如下

        for (i = 0; i < nr_maps; i++) {
                unsigned char *addr, *end;
                struct bpf_map_def *def;
                const char *map_name;
                size_t offset;

                map_name = elf_strptr(elf, strtabidx, sym[i].st_name);
                maps[i].name = strdup(map_name);
                if (!maps[i].name) {
                        printf("strdup(%s): %s(%d)\n", map_name,
                               strerror(errno), errno);
                        free(sym);
                        return -errno;
                }

                /* Symbol value is offset into ELF maps section data area */
                offset = sym[i].st_value;                               //获取变量的偏移位置
                def = (struct bpf_map_def *)(data_maps->d_buf + offset);
                maps[i].elf_offset = offset;
                memset(&maps[i].def, 0, sizeof(struct bpf_map_def));
                memcpy(&maps[i].def, def, map_sz_copy);             //拷贝变量的定义到maps[i].def中后面会使用

                /* Verify no newer features were requested */
                if (validate_zero) {
                        addr = (unsigned char*) def + map_sz_copy;
                        end  = (unsigned char*) def + map_sz_elf;
                        for (; addr < end; addr++) {
                                if (*addr != 0) {
                                        free(sym);
                                        return -EFBIG;
                                }
                        }
                }
        }

load_maps函数，根据上一个函数获取的信息将信息下发到内核并创建文件描述符，关键代码如下

                if (maps[i].def.type == BPF_MAP_TYPE_ARRAY_OF_MAPS ||
                    maps[i].def.type == BPF_MAP_TYPE_HASH_OF_MAPS) {
                        int inner_map_fd = map_fd[maps[i].def.inner_map_idx];

                        map_fd[i] = bpf_create_map_in_map_node(maps[i].def.type,        //元素的值是map的
                                                        maps[i].name,
                                                        maps[i].def.key_size,
                                                        inner_map_fd,
                                                        maps[i].def.max_entries,
                                                        maps[i].def.map_flags,
                                                        numa_node);
                } else {
                        map_fd[i] = bpf_create_map_node(maps[i].def.type,           //元素的值是普通类型的
                                                        maps[i].name,
                                                        maps[i].def.key_size,
                                                        maps[i].def.value_size,
                                                        maps[i].def.max_entries,
                                                        maps[i].def.map_flags,
                                                        numa_node);
                }

bpf_create_map_node函数就是将这些信息组织成attr 结构体 并调用 bpf系统调用，将信息下发到内核，并返回map的fd信息给用户态，在加载程序时需要用到。
接下来是将代码段中的 maps 与相应的fd 关联起来，内核中会根据fd，给代码中使用符号赋地址

        for (i = 1; i < ehdr.e_shnum; i++) {
                if (processed_sec[i])
                        continue;

                if (get_sec(elf, i, &ehdr, &shname, &shdr, &data))
                        continue;

                if (shdr.sh_type == SHT_REL) {
                        struct bpf_insn *insns;

                        /* locate prog sec that need map fixup (relocations) */
                        if (get_sec(elf, shdr.sh_info, &ehdr, &shname_prog,
                                    &shdr_prog, &data_prog))
                                continue;

                        if (shdr_prog.sh_type != SHT_PROGBITS ||
                            !(shdr_prog.sh_flags & SHF_EXECINSTR))
                                continue;

                        insns = (struct bpf_insn *) data_prog->d_buf;
                        processed_sec[i] = true; /* relo section */

                        if (parse_relo_and_apply(data, symbols, &shdr, insns,           //这个函数是将fd与insns关联起来。内核中根据fd将map地址赋值到程序中
                                                 map_data, nr_maps))
                                continue;
                }
        }

最后是通过加载EBPF程序到内核中

        for (i = 1; i < ehdr.e_shnum; i++) {

                if (processed_sec[i])
                        continue;

                if (get_sec(elf, i, &ehdr, &shname, &shdr, &data))
                        continue;

                if (memcmp(shname, "kprobe/", 7) == 0 ||
                    memcmp(shname, "kretprobe/", 10) == 0 ||
                    memcmp(shname, "tracepoint/", 11) == 0 ||
                    memcmp(shname, "raw_tracepoint/", 15) == 0 ||
                    memcmp(shname, "xdp", 3) == 0 ||
                    memcmp(shname, "perf_event", 10) == 0 ||
                    memcmp(shname, "socket", 6) == 0 ||
                    memcmp(shname, "cgroup/", 7) == 0 ||
                    memcmp(shname, "sockops", 7) == 0 ||
                    memcmp(shname, "sk_skb", 6) == 0 ||
                    memcmp(shname, "sk_msg", 6) == 0) {
                        ret = load_and_attach(shname, data->d_buf,
                                              data->d_size);                    //这个函数根据不同的ebpf类型讲代码加载到内核中。
                        if (ret != 0)
                                goto done;
                }
        }