引言

这篇文章的初衷是解决以下几个问题：

Map的原理到底是什么，如何做到用户态内核态之间互相交互？
Map的内存分配是怎么做的？
在操作queue的时候需要使用栈上变量，并传入一个栈上指针，这样做是否有危险呢？

当然读者可能会认为有一些文不对题的嫌疑，事实是kernel中这部分的代码非常简单(queue和stack的代码还是一样的)[2]，所以其实是在问题三的基础上去扩展到问题一二，以此搞清楚我们在操作map的时候其实是在做什么事情，我不求达到像欢神一样看到C语言脑中就浮现出汇编一样看一眼BPF程序就可以知道其字节码的境界，但是每个函数到底对我程序做了哪些手脚的境界按理说还是应该触碰的到的。

map

重要结构体

对于QUEUE来说其实就是代表 queue/stack 结构的bpf_queue_stack以及普适的 map 结构bpf_map：

struct bpf_queue_stack {
   
	struct bpf_map map;		// map的基础数据
	raw_spinlock_t lock;	// 用于在get和push的时候保证线程安全
	u32 head, tail;			// 一个循环队列的头尾
	u32 size; 				/* max_entries + 1 */

	char elements[0] __aligned(8);	// 实际数据存储
};

struct bpf_map {
   
	/* The first two cachelines with read-mostly members of which some
	 * are also accessed in fast-path (e.g. ops, max_entries).
	 * 这里的设计也是比较秀，前两个cache-line均为只读成员
	 */
	// 各种增删改查函数的实现
	const struct bpf_map_ops *ops ____cacheline_aligned;
	struct bpf_map *inner_map_meta;
#ifdef CONFIG_SECURITY
	void *security;
#endif
	// 这四项相信大家非常熟悉了
	enum bpf_map_type map_type;	
	u32 key_size;
	u32 value_size;
	u32 max_entries;
	u32 map_flags;		//[6]中是flag相关的标记位，BPF_F_NO_PREALLOC，BPF_F_NUMA_NODE之类的
	
	int spin_lock_off; 	/* >=0 valid offset, <0 error */
	u32 id;				// 唯一ID，调用idr_alloc_cyclic[11]分配
	int numa_node;		// bpf程序创建时可以指定其numa节点是多少，为了避免创建和执行的numa节点不同而导致性能较差[5]
	u32 btf_key_type_id;
	u32 btf_value_type_id;
	struct btf *btf;
	struct bpf_map_memory memory;
	bool unpriv_array;
	bool frozen; /* write-once */
	/* 48 bytes hole */

	/* The 3rd and 4th cacheline with misc members to avoid false sharing
	 * particularly with refcounting.
	 * 第三第四cache-line中使得redcnt独占一个cache-line，以避免false-shareing[4]
	 */
	atomic_t refcnt ____cacheline_aligned;	// 引用计数相关
	atomic_t usercnt;						// 用户态，比如pin了以后就会递增这个，当然上面也会
	struct work_struct work;
	char name[BPF_OBJ_NAME_LEN];
};

const struct bpf_map_ops queue_map_ops = {
   
	.map_alloc_check = queue_stack_map_alloc_check,
	.map_alloc = queue_stack_map_alloc,
	.map_free = queue_stack_map_free,
	.map_lookup_elem = queue_stack_map_lookup_elem,
	.map_update_elem = queue_stack_map_update_elem,
	.map_delete_elem = queue_stack_map_delete_elem,
	.map_push_elem = queue_stack_map_push_elem,
	.map_pop_elem = queue_map_pop_elem,
	.map_peek_elem = queue_map_peek_elem,
	.map_get_next_key = queue_stack_map_get_next_key,
};

现在才知道一个类型支持的操作从哪里看是最全的，没错，就是内核代码，事实上Linux内核观测技术这本书对于queue的接口描述已经过时了。

map创建

我们会通过如下栈帧到达map_create：

-bpf_create_map
-bpf_create_map_xattr
-sys_bpf(BPF_MAP_CREATE)
-map_create

// 返回文件描述符
static int map_create(union bpf_attr *attr)
{
   	
	// 获取attr中的numa标记，在对应类型的map分配内存时使用
	int numa_node = bpf_map_attr_numa_node(attr);
	struct bpf_map_memory mem;
	struct bpf_map *map;
	int f_flags;
	int err;

	err = CHECK_ATTR(BPF_MAP_CREATE);
	if (err)
		return -EINVAL;
	
	// 获取文件访问类型的flag
	f_flags = bpf_get_file_flag(attr->map_flags);
	if (f_flags < 0)
		return f_flags;
	
	// 对于numa_node的有效检查
	if (numa_node != NUMA_NO_NODE &&
	    ((unsigned int)numa_node >= nr_node_ids ||
	     !node_online(numa_node)))
		return -EINVAL;

	/* find map type and init map: hashtable vs rbtree vs bloom vs ... */
	// 根据map的类型分配空间，创建map结构体，并为其编号，以后利用编号寻找生成的map
	// 会利用一个全局的bpf_map_ops * const bpf_map_types[] 去调用其中的map_alloc回调
	// 对于queue来说当然就是 queue_stack_map_alloc 了
	// 可以调用container_of获取实际的map地址
	// map现在已经被创建成功
	map = find_and_alloc_map(attr);
	if (IS_ERR(map))
		return PTR_ERR(map);
	// 把src拷贝到dst，并检查src中是否有不允许的字符
	err = bpf_obj_name_cpy(map->name, attr->map_name);
	if (err)
		goto free_map;
	
	// 设置引用计数
	atomic_set(&map->refcnt, 1);
	atomic_set(&map->usercnt, 1);

	...... BTF相关，暂时不考虑
	
	// 不知道有啥用，函数体是空的
	err = security_bpf_map_alloc(map);
	if (err)
		goto free_map;
	
	// 调用idr_alloc_cyclic为map分配一个id，参考[11]中的注释
	err = bpf_map_alloc_id(map);
	if (err)
		goto free_map_sec;
	
	// 通过将其挂接到单个 inode 上来创建一个新文件。
	// 这对于不需要完整的 inode 即可正常运行的文件很有用。 
	// 使用 anon_inode_getfd() 创建的所有文件都将共享一个 inode，从而节省内存并避免文件/inode/dentry 设置的代码重复。
	// 以前写epoll的源码分析时也见到过anon_inode_getfd
	err = bpf_map_new_fd(map, f_flags);
	if (err < 0) {
   
		/* failed to allocate fd.
		 * bpf_map_put_with_uref() 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_with_uref(map);
		return err;
	}

	return err;

free_map_sec:
	security_bpf_map_free(map);
free_map:
	btf_put(map->btf);
	bpf_map_charge_move(&mem, &map->memory);
	map->ops->map_free(map);
	bpf_map_charge_finish(&mem);
	return err;
}

当map_create创建完成时file结构体已经创建完成了，这里的一个小优化是调用了anon_inode_getfd，所有的map的file结构体实际上inode都是一样的，bpf_map是file的private_data。

object pin

object pin的功能可以查看[8]和cilium document。

栈帧是这样的：

-sys_bpf(BPF_OBJ_PIN)
-bpf_obj_pin
-bpf_obj_pin_user
-bpf_fd_probe_obj
-bpf_map_get_with_uref
-bpf_map_inc

// ufd attr->bpf_fd
// pathname u64_to_user_ptr(attr->pathname)
// pin的时候需要给一个路径
int bpf_obj_pin_user(u32 ufd, const char __user *pathname)
{
   
	struct filename *pname;
	enum bpf_type type;
	void *raw;
	int ret;
	
	pname = getname(pathname);
	if (IS_ERR(pname))
		return PTR_ERR(pname);
	// 还记得我们前面提到特化的map结构被存在fd对应的file中的private_data
	// 这里就是根据fd拿到map结构
	// 调用bpf_map_get_with_uref->bpf_map_inc增加引用计数
	raw = bpf_fd_probe_obj(ufd, &type);
	if (IS_ERR(raw)) {
   
		ret = PTR_ERR(raw);
		goto out;
	}
	// 在其中会调用bpf_mkmap->bpf_mkobj_ops，在虚拟文件系统上创建一个文件
	ret = bpf_obj_do_pin(pname, raw, type);
	if (ret != 0)
		bpf_any_put(raw, type);
out:
	putname(pname);
	return ret;
}

queue实现相关

queue_stack_map_alloc

static struct bpf_map *queue_stack_map_alloc(union bpf_attr *attr)
{
   
	// 从attr中查看是否设置BPF_F_NUMA_NODE，设置的话取attr->numa_node
	int ret, numa_node = bpf_map_attr_numa_node(attr);
	struct bpf_map_memory mem = {
   0};
	struct bpf_queue_stack *qs;
	u64 size, queue_size, cost;

	size = (u64) attr->max_entries + 1;
	// 队列的实际大小就是size*队列中每一项的大小
	cost = queue_size = sizeof(*qs) + size * attr->value_size;
	
	// 确保内存不会溢出，即检查当前用户的是否还有round_up(cost, PAGE_SIZE) >> PAGE_SHIFT的空间
	// 其实这也是用户态调用setrlimit的原因，函数中最后调用bpf_charge_memlock检查
	ret = bpf_map_charge_init(&mem, cost);
	if (ret < 0)
		return ERR_PTR(ret);
	
	// 调用kmalloc来分配内存，所以只会失败而不会出现OOM；分配请求可能会由于未尝试从页面缓存中回收内存而失败，因此设置 __GFP_RETRY_MAYFAIL 来避免这种情况。
	qs = bpf_map_area_alloc(queue_size, numa_node);
	if (!qs) {
   
		bpf_map_charge_finish(&mem);
		return ERR_PTR(-ENOMEM);
	}
	
	// 分配了一大块内存，清空下
	memset(qs, 0, sizeof(*qs));
	
	// 用attr中的数据填充map
	bpf_map_init_from_attr(&qs->map, attr);
	
	// 把mem的指针赋值给qs->map.memory
	bpf_map_charge_move(&qs->map.memory, &mem);
	qs->size = size;
	
	// 初始化自旋锁
	raw_spin_lock_init(&qs->lock);

	return &qs->map;
}

queue_stack_map_push_elem

static struct bpf_queue_stack *bpf_queue_stack(struct bpf_map *map)
{
   	// container_of可以根据一个结构体中某一个成员的地址和名称推断出此结构体的地址，这里根据map推出所属bpf_queue_stack的地址
	// 很秀的一个实现[1]
	return container_of(map, struct bpf_queue_stack, map);
}

static int queue_stack_map_push_elem(struct bpf_map *map, void *value,
				     u64 flags)
{
   
	struct bpf_queue_stack *qs = bpf_queue_stack(map);
	unsigned long irq_flags;
	int err = 0;
	void *dst;

	/* BPF_EXIST is used to force making room for a new element in case the
	 * map is full
	 */
	bool replace = (flags & BPF_EXIST);

	/* Check supported flags for queue and stack maps */
	if (flags & BPF_NOEXIST || flags > BPF_EXIST)
		return -EINVAL;
	
	// 加锁，禁止抢占，且禁止本地中断，这也就意味着中断处理函数中是可以操作这个map的
	raw_spin_lock_irqsave(&qs->lock, irq_flags);
	
	// 检查当前队列是否已满
	if (queue_stack_map_is_full(qs)) {
   
		if (!replace) {
   
			// 已满会返回-E2BIG
			err = -E2BIG;
			goto out;
		}
		/* advance tail pointer to overwrite oldest element */
		if (unlikely(++qs->tail >= qs->size))
			qs->tail = 0;
	}
	// 其实就是用一个每个entry都相同的大数组去模拟queue
	dst = &qs->elements[qs->head * qs->map.value_size];
	// 解释了第三个问题了，就算是栈上的push也没关系，因为会执行一次拷贝
	memcpy(dst, value, qs->map.value_size);
	
	// 循环队列
	if (unlikely(++qs->head >= qs->size))
		qs->head = 0;

out:
	raw_spin_unlock_irqrestore(&qs->lock, irq_flags);
	return err;
}

__queue_map_get

static int __queue_map_get(struct bpf_map *map, void *value, bool delete)
{
   
	struct bpf_queue_stack *qs = bpf_queue_stack(map);
	unsigned long flags;
	int err = 0;
	void *ptr;

	raw_spin_lock_irqsave(&qs->lock, flags);

	if (queue_stack_map_is_empty(qs)) {
   
		memset(value, 0, qs->map.value_size);
		err = -ENOENT;
		goto out;
	}

	ptr = &qs->elements[qs->tail * qs->map.value_size];
	// 又是一次拷贝
	memcpy(value, ptr, qs->map.value_size);
	
	// 这里的delete其实是为了区分pop和pop_and_delete
	if (delete) {
   
		if (unlikely(++qs->tail >= qs->size))
			qs->tail = 0;
	}

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

引言

很粗糙的一篇文章，但是文首提出的三个问题已经解决了。

还有几个有意思的点值得记录一下：

map中对应的file的创建调用anon_inode_getfd，同类型的file共享一个inode，以前在epoll中见过类似的用法[12]
map id 调用idr_alloc_cyclic分配
map其中element的内存分配先调用kmalloc（GFP_USER | __GFP_NORETRY | gfp）,如果不成功调用vmalloc（GFP_KERNEL | __GFP_RETRY_MAYFAIL | gfp）(gfp = __GFP_NOWARN | __GFP_ZERO | __GFP_ACCOUNT;)
queue结构的所有push/pop操作都需要拷贝，其他结构暂时还没看

参考：