注：本文是转载，但不是100%的转载，可能稍微有些出入，原文地址点击这里基本概念VPP里的Bihash全名为Bounded-index extensible hash。它的最大特点是，在查找时是无锁并且线程安全的。修改操作之间会有互斥，但是修改操作时仍然可以进行查找操作。vpp里的Bihash优化成了两种，bihash_kv_8_8和bihash_kv_24_8,区别在于hash key是8字节还是24字 节。最大限度的利用SSE4.2指令集中的_mm_crc32_u64来进行hash计算。核心函数在bihash_template.c中。根据包含的头文件是bihash_8_8.h还是bihash_24_8.h，BV宏和BTV 宏将把名字做出对应扩展。例如:BV (clib_bihash_init)扩展为clib_bihash_init_8_8()或者clib_bihash_init_24_8()。BVT (clib_bihash)扩展为clib_bihash_8_8_t或者为clib_bihash_24_8_t。clib_bihash_bucket_thash桶typedef struct { union { struct { //这个桶中记录得kv对，在heap中的起始位置 u32 offset; u8 pad[3]; //这个桶中记录的kv对，一共占用了1 << log2_pages个page u8 log2_pages; }; u64 as_u64; }; } clib_bihash_bucket_t;page数据结构，中间包含了kv对。typedef struct BV (clib_bihash_value) { union { BVT (clib_bihash_kv) kvp[BIHASH_KVP_PER_PAGE]; struct BV (clib_bihash_value) * next_free; }; } BVT (clib_bihash_value);核心函数clib_bihash_init_8_8()和clib_bihash_init_24_8()初始化bihash，并分配一个独占的内存heap给它。void BV (clib_bihash_init)(BVT (clib_bihash) * h, char *name, u32 nbuckets, uword memory_size) { void *oldheap; //方便之后hash值映射到bucket，用&替代昂贵的%操作。 nbuckets = 1 << (max_log2 (nbuckets)); h->name = (u8 *) name; h->nbuckets = nbuckets; h->log2_nbuckets = max_log2 (nbuckets); h->mheap = mheap_alloc (0 /* use VM */ , memory_size); //常用操作，这样才能使用内存操作函数， //注意这里单核心上只能有一个线程独占此操作，不同核心可以并发 oldheap = clib_mem_set_heap (h->mheap); vec_validate_aligned (h->buckets, nbuckets - 1, CLIB_CACHE_LINE_BYTES); h->writer_lock = clib_mem_alloc_aligned (CLIB_CACHE_LINE_BYTES, CLIB_CACHE_LINE_BYTES); clib_mem_set_heap (oldheap); }clib_bihash_search_8_8()和clib_bihash_search_24_8()给定key，查找value。int BV (clib_bihash_search)(const BVT (clib_bihash) * h, BVT (clib_bihash_kv) * search_key, BVT (clib_bihash_kv) * valuep) { //hash计算，者利用了SSE4.2指令集特性。值得记住 hash = BV (clib_bihash_hash) (search_key); //hash值的低log2_nbuckets bit用来索引桶号 bucket_index = hash & (h->nbuckets - 1); b = &h->buckets[bucket_index]; if (b->offset == 0) return -1; hash >>= h->log2_nbuckets; //在heap中的offset字节开始，属于桶。 v = BV (clib_bihash_get_value) (h, b->offset); //hash值的中间log2_pages bit用来索引桶中的page号 value_index = hash & ((1 << b->log2_pages) - 1); v += value_index; //遍历page，查找key值。 for (i = 0; i < BIHASH_KVP_PER_PAGE; i++) { if (BV (clib_bihash_key_compare) (v->kvp[i].key, search_key->key)) { *valuep = v->kvp[i]; return 0; } } return -1; }clib_bihash_add_del_8_8()和clib_bihash_add_del_24_8()添加删除kv对。int BV (clib_bihash_add_del)(BVT (clib_bihash) * h, BVT (clib_bihash_kv) * add_v, int is_add) { u32 cpu_number = os_get_cpu_number (); hash = BV (clib_bihash_hash) (add_v); //hash值的低log2_nbuckets bit用来索引桶号 bucket_index = hash & (h->nbuckets - 1); b = &h->buckets[bucket_index]; hash >>= h->log2_nbuckets; while (__sync_lock_test_and_set (h->writer_lock, 1)) ; /* First elt in the bucket? */ if (b->offset == 0) { if (is_add == 0) { rv = -1; goto unlock; } //桶中最初没有kv对，现在分配一个page v = BV (value_alloc) (h, 0); *v->kvp = *add_v; tmp_b.as_u64 = 0; tmp_b.offset = BV (clib_bihash_get_offset) (h, v); b->as_u64 = tmp_b.as_u64; goto unlock; } //把桶b中的kv对拷贝到缓存中，缓存加入到b中，b中原来的kv对留给下文修改。 BV (make_working_copy) (h, b); //b中原来的kv对page v = BV (clib_bihash_get_value) (h, h->saved_bucket.offset); //hash值的中间log2_pages bit用来索引桶中的page号 value_index = hash & ((1 << h->saved_bucket.log2_pages) - 1); v += value_index; if (is_add) { /* * For obvious (in hindsight) reasons, see if we're supposed to * replace an existing key, then look for an empty slot. */ for (i = 0; i < BIHASH_KVP_PER_PAGE; i++) { if (!memcmp (&(v->kvp[i]), &add_v->key, sizeof (add_v->key))) { //有重复的key值，把value覆盖旧的 clib_memcpy (&(v->kvp[i]), add_v, sizeof (*add_v)); CLIB_MEMORY_BARRIER (); /* Restore the previous (k,v) pairs */ //修改完的page重新保存回b中 b->as_u64 = h->saved_bucket.as_u64; goto unlock; } } for (i = 0; i < BIHASH_KVP_PER_PAGE; i++) { if (BV (clib_bihash_is_free) (&(v->kvp[i]))) { //要保存的kv内容拷贝到第一个空闲的空间中 clib_memcpy (&(v->kvp[i]), add_v, sizeof (*add_v)); CLIB_MEMORY_BARRIER (); b->as_u64 = h->saved_bucket.as_u64; goto unlock; } } /* no room at the inn... split case... */ } else { for (i = 0; i < BIHASH_KVP_PER_PAGE; i++) { if (!memcmp (&(v->kvp[i]), &add_v->key, sizeof (add_v->key))) { memset (&(v->kvp[i]), 0xff, sizeof (*(add_v))); CLIB_MEMORY_BARRIER (); b->as_u64 = h->saved_bucket.as_u64; goto unlock; } } rv = -3; b->as_u64 = h->saved_bucket.as_u64; goto unlock; } //添加kv发现空间不够了，该桶的page数量增加一倍 new_log2_pages = h->saved_bucket.log2_pages + 1; expand_again: working_copy = h->working_copies[cpu_number]; //扩充page，其中的kv对需要重新排列下，因为hash值中需要用new_log2_pages个bit来确定page位置 new_v = BV (split_and_rehash) (h, working_copy, new_log2_pages); if (new_v == 0) { new_log2_pages++; goto expand_again; } /* Try to add the new entry */ save_new_v = new_v; new_hash = BV (clib_bihash_hash) (add_v); new_hash >>= h->log2_nbuckets; new_hash &= (1 << min_log2 (vec_len (new_v))) - 1; new_v += new_hash; for (i = 0; i < BIHASH_KVP_PER_PAGE; i++) { if (BV (clib_bihash_is_free) (&(new_v->kvp[i]))) { clib_memcpy (&(new_v->kvp[i]), add_v, sizeof (*add_v)); goto expand_ok; } } /* Crap. Try again */ new_log2_pages++; BV (value_free) (h, save_new_v); goto expand_again; expand_ok: tmp_b.log2_pages = min_log2 (vec_len (save_new_v)); tmp_b.offset = BV (clib_bihash_get_offset) (h, save_new_v); CLIB_MEMORY_BARRIER (); b->as_u64 = tmp_b.as_u64; v = BV (clib_bihash_get_value) (h, h->saved_bucket.offset); BV (value_free) (h, v); unlock: CLIB_MEMORY_BARRIER (); h->writer_lock[0] = 0; return rv; }make_working_copy_8_8()和make_working_copy_24_8()用来生成桶内pages的副本，供添加删除修改使用。static inline void BV (make_working_copy) (BVT (clib_bihash) * h, clib_bihash_bucket_t * b) { BVT (clib_bihash_value) * v; clib_bihash_bucket_t working_bucket __attribute__ ((aligned (8))); void *oldheap; BVT (clib_bihash_value) * working_copy; u32 cpu_number = os_get_cpu_number (); //working_copies是per-cpu的 if (cpu_number >= vec_len (h->working_copies)) { oldheap = clib_mem_set_heap (h->mheap); vec_validate (h->working_copies, cpu_number); clib_mem_set_heap (oldheap); } /* * working_copies are per-cpu so that near-simultaneous * updates from multiple threads will not result in sporadic, spurious * lookup failures. */ working_copy = h->working_copies[cpu_number]; //博主觉得saved_bucket又不是per-cpu的，那么working_copies就没必要做成per-cpu了 h->saved_bucket.as_u64 = b->as_u64; oldheap = clib_mem_set_heap (h->mheap); if ((1 << b->log2_pages) > vec_len (working_copy)) { vec_validate_aligned (working_copy, (1 << b->log2_pages) - 1, sizeof (u64)); h->working_copies[cpu_number] = working_copy; } _vec_len (working_copy) = 1 << b->log2_pages; clib_mem_set_heap (oldheap); v = BV (clib_bihash_get_value) (h, b->offset); //b中原有的kv内容拷贝到working_copy中，然后把b的page指向working_copy中的。这样b中的kv其实是副本。 clib_memcpy (working_copy, v, sizeof (*v) * (1 << b->log2_pages)); working_bucket.as_u64 = b->as_u64; working_bucket.offset = BV (clib_bihash_get_offset) (h, working_copy); CLIB_MEMORY_BARRIER (); b->as_u64 = working_bucket.as_u64; h->working_copies[cpu_number] = working_copy; }split_and_rehash_8_8()和split_and_rehash_24_8()桶中的page数量扩张后，原有的kv需要重新插入一边。 static BVT (clib_bihash_value) *BV (split_and_rehash)(BVT (clib_bihash) * h, BVT (clib_bihash_value) * old_values, u32 new_log2_pages) { BVT (clib_bihash_value) * new_values, *v, *new_v; int i, j, k; new_values = BV (value_alloc) (h, new_log2_pages); //v会遍历原有的每个page v = old_values; for (i = 0; i < vec_len (old_values); i++) { u64 new_hash; //遍历原有桶中特定page中的kv for (j = 0; j < BIHASH_KVP_PER_PAGE; j++) { if (BV (clib_bihash_is_free) (&(v->kvp[j])) == 0) { new_hash = BV (clib_bihash_hash) (&(v->kvp[j])); new_hash >>= h->log2_nbuckets; new_hash &= (1 << new_log2_pages) - 1; new_v = &new_values[new_hash]; for (k = 0; k < BIHASH_KVP_PER_PAGE; k++) { if (BV (clib_bihash_is_free) (&(new_v->kvp[k]))) { clib_memcpy (&(new_v->kvp[k]), &(v->kvp[j]), sizeof (new_v->kvp[k])); goto doublebreak; } } /* Crap. Tell caller to try again */ BV (value_free) (h, new_values); return 0; } doublebreak: ; } v++; } return new_values; }clib_bihash_value_8_8()和clib_bihash_value_24_8()分配page用，page用来保存kv对。内存分配以page为单位，分配1 << log2_pages个page，并且对回收的page做了缓存。但是没有用伙伴算法进行碎片内存合并。static BVT (clib_bihash_value) *BV (value_alloc) (BVT (clib_bihash) * h, u32 log2_pages) { BVT (clib_bihash_value) * rv = 0; void *oldheap; ASSERT (h->writer_lock[0]); //h->freelists用log2_pages值来索引空闲page if (log2_pages >= vec_len (h->freelists) || h->freelists[log2_pages] == 0) { oldheap = clib_mem_set_heap (h->mheap); vec_validate (h->freelists, log2_pages); //分配1 << log2_pages 个page，方便从hash值中计算出page编号。 vec_validate_aligned (rv, (1 << log2_pages) - 1, CLIB_CACHE_LINE_BYTES); clib_mem_set_heap (oldheap); goto initialize; } rv = h->freelists[log2_pages]; h->freelists[log2_pages] = rv->next_free; initialize: ASSERT (rv); ASSERT (vec_len (rv) == (1 << log2_pages)); /* * Latest gcc complains that the length arg is zero * if we replace (1<<log2_pages) with vec_len(rv). * No clue. */ memset (rv, 0xff, sizeof (*rv) * (1 << log2_pages)); return rv; }
注：本文是转载，但不是100%的转载，可能稍微有些出入，原文地址点击这里基本概念早期的VPP本身的node框架比较固定，各个node之间逻辑连接已经固化。为此新版本增加了feature机制，每个feature是一个node，用户可以启用/停止某个或某些feature。用户也可以自己写插件，把自定义node（自己的业务逻辑）加入到指定位置。vnet_feature_arc_registration_t vpp将feature分成不同的组，每组feature称为一个arc。arc中的feature按照代码指定的顺序串接起来。arc结构将记录这组feature的起始node和结束node。系统初始化时完成每个feature的连接。VNET_FEATURE_ARC_INIT宏用来注册arc。在arc中指定的起始node中，必须调用vnet_feature_arc_start函数，才能正式进入feature机制业务流程，该函数会将下一跳强行指定为arc中的下一个feature。vnet_feature_registration_t一个feature等价于一个node，用户通过VNET_FEATURE_INIT宏定义自己的feature，指定需要加入哪个arc，以及在arc中的哪个相对位置。每个feature都可以通过外部命令行启用/停止。关键函数VNET_FEATURE_ARC_INIT和VNET_FEATURE_ARC_INIT宏将arc和feature保存到全局变量feature_main中。此时arc和feature还未被”加工”，只是基本的原料。Feature初始化clib_error_t *vnet_feature_init将把arc和feature初始化，并组织成最终的数据结构填充到vnet_feature_main_t feature_main中，代码比较简单。这里只对vnet_feature_main_t描述即可。typedef struct { // feature arc configuration list //VNET_FEATURE_ARC_INIT宏中注册的arc保存在这里，顺序不定，也没有意义。 vnet_feature_arc_registration_t *next_arc; /* * vnet_feature_init()中来初始化该成员， * 把arc按名字组织成hash表,为何要用双重指针？？？ * 博主认为是bug，后面明明是当hash表用么。又不是hash数组 */ uword **arc_index_by_name; /* * feature path configuration lists * VNET_FEATURE_ARC_INIT宏中注册的feature保存在这里，顺序不定，也没有意义。 */ vnet_feature_registration_t *next_feature; /* * vnet_feature_init()中来初始化该成员，把feature分配到不同arc中， * 并且根据before和after变量来决定feature顺序。 * 这里看作是一个数组，用arc的index做索引 */ vnet_feature_registration_t **next_feature_by_arc; /* * vnet_feature_init()中来初始化该成员，把feature按名字组织成hash表数组， * arc的index索引到对应hash表，node name来做hash表的key */ uword **next_feature_by_name; // feature config main objects vnet_feature_config_main_t *feature_config_mains; // Save partial order results for show command char ***feature_nodes; /* * bitmap of interfaces which have driver rx features configured * bitmap数组，arc的index索引对应bitmap表,决定该网卡是否启用了该arc中的feature */ uword **sw_if_index_has_features; // feature reference counts by interface i16 **feature_count_by_sw_if_index; /* * Feature arc index for device-input * 从收包开始就使用feature机制了，这个变量给收包驱动用。 * 追踪了下代码，这个变量属于device_input_node这个node， * 但是这个node业务函数为空，而且状态是VLIB_NODE_STATE_DISABLED。 * 那这是干什么用的呢？层层追踪，原来这个node纯粹是个dump node， * 但是它是一个arc起点。这个arc的index保存在了下面变量中。 * 然后真正的网卡收包node中会借用这个index当成自己的， * 调用vnet_feature_start_device_input_x1这类函数， * 开始了feature之旅。真TMD恶心，代码还能写的更难懂点么。 */ u8 device_input_feature_arc_index; // convenience vlib_main_t *vlib_main; vnet_main_t *vnet_main; } vnet_feature_main_t;arc初始化vnet_feature_arc_init把属于某arc的无序的feature排序成有序状态。feature注册时可以指定在某些node前面，和某些node后面。clib_error_t *vnet_feature_arc_init (vlib_main_t * vm, vnet_config_main_t * vcm, char **feature_start_nodes, int num_feature_start_nodes, vnet_feature_registration_t * first_reg, char ***in_feature_nodes) { ... // pass 1, collect feature node names, construct a before b pairs while (this_reg) { node_name = format (0, "%s%c", this_reg->node_name, 0); //vec_len (node_names) 指的是当前node_names数组中元素个数 hash_set (reg_by_index, vec_len (node_names), (uword) this_reg); //现在feature用index来代表了 hash_set_mem (index_by_name, node_name, vec_len (node_names)); vec_add1 (node_names, node_name); //runs_before是个数组，可以保存有多个node。 these_constraints = this_reg->runs_before; while (these_constraints && these_constraints[0]) { ... } //runs_after是个数组，可以保存有多个node。 these_constraints = this_reg->runs_after; while (these_constraints && these_constraints[0]) { ... } this_reg = this_reg->next; } //至此，把feature之间顺序关系表达为"A,B"字符串，代表A在B前面。 //其实就是该arc中所有feature个数 n_features = vec_len (node_names); //可以看作生成一个(n_features X n_features)的二维数组 orig = clib_ptclosure_alloc (n_features); for (i = 0; i < vec_len (constraints); i++) { ... } ... /* see if we got a partial order... */ if (vec_len (result) != n_features) return clib_error_return (0, "%d feature_init_cast no partial order!"); //到这里，feature的顺序关系就计算成功了。保存在result中，每个feature用index代表。 /* * We win. * Bind the index variables, and output the feature node name vector * using the partial order we just computed. Result is in stack * order, because the entry with the fewest constraints (e.g. none) * is output first, etc. */ //把feature索引按顺序转换成对应的vnet_feature_registration_t信息，保存在feature_nodes中 for (i = n_features - 1; i >= 0; i--) { p = hash_get (reg_by_index, result[i]); ASSERT (p != 0); this_reg = (vnet_feature_registration_t *) p[0]; if (this_reg->feature_index_ptr) *this_reg->feature_index_ptr = n_features - (i + 1); this_reg->feature_index = n_features - (i + 1); vec_add1 (feature_nodes, this_reg->node_name); } /* Set up the config infrastructure */ /*feature顺序现在保存在feature_nodes中，现在要把这个关系保存到vnet_config_main_t中了。 注意每个arc都有一份自己的vnet_config_main_t*/ vnet_config_init (vm, vcm, feature_start_nodes, num_feature_start_nodes, feature_nodes, vec_len (feature_nodes)); /* Save a copy for show command */ *in_feature_nodes = feature_nodes; /* Finally, clean up all the shit we allocated */ /* *INDENT-OFF* */ ... }至此，feature已经计算好了顺序，但是还是没有真正把各个feature连接起来。连接Feature连接各个feature的真正工作，最终交由vnet_config_add_feature和vnet_config_del_feature完成。u32 vnet_config_add_feature (vlib_main_t * vm, vnet_config_main_t * cm, u32 config_string_heap_index, u32 feature_index, void *feature_config, u32 n_feature_config_bytes) { ... //现有的feature基础上增加参数中传来的新feature。 //注意每个feature都是这么一个个添加进来的。 vec_add2 (new_features, f, 1); f->feature_index = feature_index; f->node_index = node_index; /* * 不同feature可能有自己独特的config内容， * 因此参数中的feature_config和n_feature_config_bytes就是用来干这事的， * 保存在f->feature_config中，没有私有内容，则计算出来为0。 */ n_feature_config_u32s = round_pow2 (n_feature_config_bytes, sizeof (f->feature_config[0])) / sizeof (f->feature_config[0]); vec_add (f->feature_config, feature_config, n_feature_config_u32s); //上文所知，feature的index也代表着在arc中的顺序，这里排下序。 if (vec_len (new_features) > 1) vec_sort_with_function (new_features, feature_cmp); if (old) remove_reference (cm, old); //开始干活了，连接每个feature。config内存也会更新 new = find_config_with_features (vm, cm, new_features); new->reference_count += 1; /* * User gets pointer to config string first element * (which defines the pool index * this config string comes from). */ vec_validate (cm->config_pool_index_by_user_index, new->config_string_heap_index + 1); cm->config_pool_index_by_user_index[new->config_string_heap_index + 1] = new - cm->config_pool; return new->config_string_heap_index + 1; } u32 vnet_config_del_feature (vlib_main_t * vm, vnet_config_main_t * cm, u32 config_string_heap_index, u32 feature_index, void *feature_config, u32 n_feature_config_bytes) { vnet_config_t *old, *new; vnet_config_feature_t *new_features, *f; u32 n_feature_config_u32s; //每个feature组合都有一个对应的config内存。 u32 *p = vnet_get_config_heap (cm, config_string_heap_index); //config内存第一个32bit，保存了该config的索引号。 //这代码很恶心，作者为啥不弄个结构出来，还好懂些。 old = pool_elt_at_index (cm->config_pool, p[-1]); n_feature_config_u32s = round_pow2 (n_feature_config_bytes, sizeof (f->feature_config[0])) / sizeof (f->feature_config[0]); /* Find feature with same index and opaque data. */ //找到要删除的目标feature vec_foreach (f, old->features) { if (f->feature_index == feature_index && vec_len (f->feature_config) == n_feature_config_u32s && (n_feature_config_u32s == 0 || !memcmp (f->feature_config, feature_config, n_feature_config_bytes))) break; } /* Feature not found. */ if (f >= vec_end (old->features)) return config_string_heap_index; // return original config index //生成一组新feature，不包含目标feature。 new_features = duplicate_feature_vector (old->features); f = new_features + (f - old->features); vnet_config_feature_free (f); vec_delete (new_features, 1, f - new_features); /* * must remove old from config_pool now as it may be expanded and change * memory location if the following function find_config_with_features() * adds a new config because none of existing config's has matching features * and so can be reused */ remove_reference (cm, old); //新生成的feature重新连接下，config内存也会更新 new = find_config_with_features (vm, cm, new_features); new->reference_count += 1; vec_validate (cm->config_pool_index_by_user_index, new->config_string_heap_index + 1); cm->config_pool_index_by_user_index[new->config_string_heap_index + 1] = new - cm->config_pool; return new->config_string_heap_index + 1; } static vnet_config_t *find_config_with_features (vlib_main_t * vm, vnet_config_main_t * cm, vnet_config_feature_t * feature_vector) { u32 last_node_index = ~0; vnet_config_feature_t *f; u32 *config_string; uword *p; vnet_config_t *c; config_string = cm->config_string_temp; cm->config_string_temp = 0; if (config_string) _vec_len (config_string) = 0; vec_foreach (f, feature_vector) { //按顺序连接各个node f->next_index = add_next (vm, cm, last_node_index, f->node_index); last_node_index = f->node_index; //下一个node在本node中的slot号，保存进config_string vec_add1 (config_string, f->next_index); //slot号后面还可以保存本node特有的config信息，可以为0大小 vec_add (config_string, f->feature_config, vec_len (f->feature_config)); } //确保最终连接到end node if (last_node_index == ~0 || last_node_index != cm->end_node_index) { u32 next_index = add_next (vm, cm, last_node_index, cm->end_node_index); vec_add1 (config_string, next_index); } //看看hash表里是不是有同样的config_string，注意这里比较的是config_string内容，不是地址 p = hash_get_mem (cm->config_string_hash, config_string); if (p) { //大多数情况下config_string不会重复，除非旧的还被别的模块引用了没释放， //那这里就继续引用旧的释放新的 cm->config_string_temp = config_string; /* we'll use it again later. */ free_feature_vector (feature_vector); c = pool_elt_at_index (cm->config_pool, p[0]); } else { u32 *d; //分配新的config结构 pool_get (cm->config_pool, c); c->index = c - cm->config_pool; c->features = feature_vector; c->config_string_vector = config_string; /* * Allocate copy of config string in heap. * VLIB buffers will maintain pointers to heap as they read out * configuration data. */ c->config_string_heap_index = heap_alloc (cm->config_string_heap, vec_len (config_string) + 1, c->config_string_heap_handle); /* First element in heap points back to pool index. */ d = vec_elt_at_index (cm->config_string_heap, c->config_string_heap_index); //注意这里，config第一个4字节保存的是config索引号， //之后才是slot号-私有config-slot号-私有config ...... d[0] = c->index; clib_memcpy (d + 1, config_string, vec_bytes (config_string)); hash_set_mem (cm->config_string_hash, config_string, c->index); c->reference_count = 0; /* will be incremented by caller. */ } return c; }现在各个feature node已经连接好了，连接信息也保存到了config中，接下来就是在业务node中使用了。启动arc中的Feature在需要开始使用feature机制的业务node中调用如下函数即可，注意该业务node必须是arc中的起始node。static_always_inline void vnet_feature_arc_start (u8 arc, u32 sw_if_index, u32 * next0, vlib_buffer_t * b0) { vnet_feature_arc_start_with_data (arc, sw_if_index, next0, b0, 0); }此函数会将next0修改为下一跳feature。feature的顺序在上文中已经确定。获取下一跳feature的逻辑很简单，在函数：always_inline void *vnet_get_config_data (vnet_config_main_t * cm, u32 * config_index, u32 * next_index, u32 n_data_bytes) { u32 i, n, *d; //vlib_buffer_t->current_config_index用来记录执行到了哪个feature了，注意最初值为1。 i = *config_index; d = heap_elt_at_index (cm->config_string_heap, i); n = round_pow2 (n_data_bytes, sizeof (d[0])) / sizeof (d[0]); //本feature下一跳的slot号 *next_index = d[n]; //记录下一跳feature的config索引 *config_index = (i + n + 1); return (void *) d; }
注：本文是转载，但不是100%的转载，可能稍微有些出入，原文地址点击这里基本概念VPP路由可以看作是思科iOS CEF的翻版。将路由分为路由表和转发表。路由表的实现基于hash，转发表的实现基于8-8-8-8 mtrie树。路由表存储了所有可能的路由表项(其中最优路由将安装到转发表中)，转发表存储了实际使用的路由项(数据包转发真正查找的是这个转发表)。VPP转发表查找的最后结果将得到一个DPO（data path object），DPO将指示数据包的下一步动作(分类，ARP查找，丢弃等等，请查看DPO_TYPES枚举类型)。以下皆以ipv4来分析，只分析了我看懂了的一部分。路由表typedef struct ip4_fib_t { //根据掩码大小对路由项作hash uword *fib_entry_by_dst_address[33]; //8-8-8-8 mtrie转发表 ip4_fib_mtrie_t mtrie; // Table ID (hash key) for this FIB. u32 table_id; // Index into FIB vector. u32 index; // flow hash configuration flow_hash_config_t flow_hash_config; // N-tuple classifier indices u32 fwd_classify_table_index; u32 rev_classify_table_index; } ip4_fib_t;路由表项hash链入fib_entry_by_dst_address中。转发表项安装在mtrie中，这是实际被数据平面使用的查找数据库。路由表的任何变化，将导致转发表的更新。路由表项typedef struct fib_entry_t_ { //所有路由关键数据结构都已这个打头，可以看做是他们的基类。 //用来将这些不同数据结构连在一起，又能用统一的接口来遍历。 fib_node_t fe_node; //可以理解为路由目的地址+掩码 const fib_prefix_t fe_prefix; //The index of the FIB table this entry is in u32 fe_fib_index; /* * The load-balance used for forwarding. * * We don't share the EOS and non-EOS even in case when they could be * because: * - complexity & reliability v. memory * determining the conditions where sharing is possible is non-trivial. * - separate LBs means we can get the EOS bit right in the MPLS label DPO * and so save a few clock cycles in the DP imposition node since we can * paint the header straight on without the need to check the packet * type to derive the EOS bit value. */ dpo_id_t fe_lb; // [FIB_FORW_CHAIN_MPLS_NUM]; /* * Vector of source infos. * Most entries will only have 1 source. So we optimise for memory usage, * which is preferable since we have many entries. * * 同一个路由目的，可以是由不同的来源添加的， * 所有来源类型查看fib_source_t_枚举类型，数值越小的来源， * 优先级越高，最高优先级的来源的路由将安装到转发表， * 低优先级来源的路由项默默地保存在最高优先级来源后面。 * 这个数组是有序的，添加删除都将导致排序，0号位置为最优路由。 */ fib_entry_src_t *fe_srcs; /* * the path-list for which this entry is a child. This is also the path-list * that is contributing forwarding for this entry. */ fib_node_index_t fe_parent; /* * index of this entry in the parent's child list. * This is set when this entry is added as a child, but can also * be changed by the parent as it manages its list. */ u32 fe_sibling; //A vector of delegates. fib_entry_delegate_t *fe_delegates; } fib_entry_t;source表明该路由项的来源，每个source结构中包含了一个path_list，这个表包含了所有可能的下一跳。typedef struct fib_entry_src_t_ { // A vector of path extensions struct fib_path_ext_t_ *fes_path_exts; /* * The path-list created by the source * 通过它可以找到该source的path_list */ fib_node_index_t fes_pl; // Which source this info block is for fib_source_t fes_src; // Flags on the source fib_entry_src_flag_t fes_flags; /* * 1 bytes ref count. This is not the number of users of the Entry * (which is itself not large, due to path-list sharing), but the number * of times a given source has been added. Which is even fewer */ u8 fes_ref_count; // Flags the source contributes to the entry fib_entry_flag_t fes_entry_flags; // Source specific info union { struct { // the index of the FIB entry that is the covering entry fib_node_index_t fesr_cover; // This source's index in the cover's list u32 fesr_sibling; } rr; struct { // the index of the FIB entry that is the covering entry fib_node_index_t fesa_cover; // This source's index in the cover's list u32 fesa_sibling; } adj; struct { // the index of the FIB entry that is the covering entry fib_node_index_t fesi_cover; // This source's index in the cover's list u32 fesi_sibling; } interface; struct { // This MPLS local label associated with the prefix. mpls_label_t fesm_label; // the indicies of the LFIB entries created fib_node_index_t fesm_lfes[2]; } mpls; struct { // The source FIB index. fib_node_index_t fesl_fib_index; } lisp; }; } fib_entry_src_t;path path_list:为了到达目的路由的下一跳称作path，所有可能的下一跳组合成path_list。typedef struct fib_path_list_t_ { // A path-list is a node in the FIB graph. fib_node_t fpl_node; // Flags on the path-list fib_path_list_flags_t fpl_flags; /* * The next-hop protocol for the paths in this path list. * Note that fixing the proto here means we don't support a mix of * v4 and v6 paths. ho hum. */ fib_protocol_t fpl_nh_proto; /* * Vector of paths indicies for all configured paths. * For shareable path-lists this list MUST not change. * path数组，记录了所有下一跳 */ fib_node_index_t *fpl_paths; // the RPF list calculated for this path list fib_node_index_t fpl_urpf; } fib_path_list_t;关键函数路由的添加从vnet_ip_route_cmd函数入手，这是通过CLI添加路由的入口函数。从中找到路由添加关键函数fib_table_entry_path_add2。fib_node_index_t fib_table_entry_path_add2 (u32 fib_index, const fib_prefix_t *prefix, fib_source_t source, fib_entry_flag_t flags, fib_route_path_t *rpath) { fib_node_index_t fib_entry_index; fib_table_t *fib_table; u32 ii; //找到路由表 fib_table = fib_table_get(fib_index, prefix->fp_proto); //精确匹配到路由表项，注意不是最长匹配。 fib_entry_index = fib_table_lookup_exact_match_i(fib_table, prefix); for (ii = 0; ii < vec_len(rpath); ii++) { fib_table_route_path_fixup(prefix, &rpath[ii]); } if (FIB_NODE_INDEX_INVALID == fib_entry_index) { //创建路由表项，包括souces，path_list等，比较复杂，将详细分析。 fib_entry_index = fib_entry_create(fib_index, prefix, source, flags, rpath); //路由表项链入hash表，很简单。 fib_table_entry_insert(fib_table, prefix, fib_entry_index); fib_table->ft_src_route_counts[source]++; } else { int was_sourced; was_sourced = fib_entry_is_sourced(fib_entry_index, source); //添加一个下一跳到路由表项，可能会新建source，也可能只是在现有source的path_list中增加一个path fib_entry_path_add(fib_entry_index, source, flags, rpath);; if (was_sourced != fib_entry_is_sourced(fib_entry_index, source)) { //如果这个source是从无到有，更新下统计值 fib_table->ft_src_route_counts[source]++; } } return (fib_entry_index); }新建路由项，并插入mtrie树 fib_node_index_t fib_entry_create (u32 fib_index, const fib_prefix_t *prefix, fib_source_t source, fib_entry_flag_t flags, const fib_route_path_t *paths) { fib_node_index_t fib_entry_index; fib_entry_t *fib_entry; ASSERT(0 < vec_len(paths)); //简单的分配一个初始化了的路由表项，很简单。 fib_entry = fib_entry_alloc(fib_index, prefix, &fib_entry_index); /* * since this is a new entry create, we don't need to check for winning * sources - there is only one. * 不同的source有不同的fesv_add函数。这里就是为了调用它。 */ fib_entry = fib_entry_src_action_add(fib_entry, source, flags, drop_dpo_get( fib_proto_to_dpo( fib_entry_get_proto(fib_entry)))); //不同的source有不同的fesv_path_swap函数。这里就是为了调用它。 fib_entry_src_action_path_swap(fib_entry, source, flags, paths); // handle possible realloc's by refetching the pointer fib_entry = fib_entry_get(fib_entry_index); //不同的source有不同的fesv_activate函数。这里就是为了调用它。 //如果该source没有fesv_activate函数，则调用fib_entry_src_action_install fib_entry_src_action_activate(fib_entry, source); //调用source的fesv_installed函数和fesv_fwd_update函数 fib_entry_post_install_actions(fib_entry, source, FIB_ENTRY_FLAG_NONE); return (fib_entry_index); }fib_entry_t *fib_entry_src_action_add (fib_entry_t *fib_entry, fib_source_t source, fib_entry_flag_t flags, const dpo_id_t *dpo) { fib_node_index_t fib_entry_index; fib_entry_src_t *esrc; esrc = fib_entry_src_find_or_create(fib_entry, source, NULL); esrc->fes_ref_count++; //对应source必须是从无到有。否则就直接返回了。 if (1 != esrc->fes_ref_count) { // we only want to add the source on the 0->1 transition return (fib_entry); } esrc->fes_entry_flags = flags; /* * save variable so we can recover from a fib_entry realloc. * 接下来的函数调用可能会导致fib_entry pool重新分配（扩张），所以只有索引号肯定是保证不变 */ fib_entry_index = fib_entry_get_index(fib_entry); //这个场景调用的是fib_entry_src_api_add if (NULL != fib_entry_src_vft[source].fesv_add) { fib_entry_src_vft[source].fesv_add(esrc, fib_entry, flags, fib_entry_get_proto(fib_entry), dpo); } //路由表项pool可能重新分配了，所以这里通过索引重新获取下。 fib_entry = fib_entry_get(fib_entry_index); esrc->fes_flags |= FIB_ENTRY_SRC_FLAG_ADDED; //这个lock理解为引用计数，不是多线程中的互斥lock fib_path_list_lock(esrc->fes_pl); // the source owns a lock on the entry fib_entry_lock(fib_entry_get_index(fib_entry)); return (fib_entry); }fib_entry_src_api_add-->fib_path_list_create_specialfib_node_index_t fib_path_list_create_special (fib_protocol_t nh_proto, fib_path_list_flags_t flags, const dpo_id_t *dpo) { fib_node_index_t path_index, path_list_index; fib_path_list_t *path_list; path_list = fib_path_list_alloc(&path_list_index); path_list->fpl_flags = flags; path_list->fpl_nh_proto = nh_proto; //创建path，这个场景里创建的是FIB_PATH_LIST_FLAG_NONE，drop dpo path_index = fib_path_create_special(path_list_index, path_list->fpl_nh_proto, fib_path_list_flags_2_path_flags(flags), dpo); //path加入path_list中 vec_add1(path_list->fpl_paths, path_index); /* * we don't share path-lists. we can do PIC on them so why bother. * 最重要的部分，分析path_list中的path，生成path->fp_dpo。 */ path_list = fib_path_list_resolve(path_list); return (path_list_index); } fib_node_index_t fib_path_create_special (fib_node_index_t pl_index, fib_protocol_t nh_proto, fib_path_cfg_flags_t flags, const dpo_id_t *dpo) { fib_path_t *path; pool_get(fib_path_pool, path); memset(path, 0, sizeof(*path)); fib_node_init(&path->fp_node, FIB_NODE_TYPE_PATH); dpo_reset(&path->fp_dpo); path->fp_pl_index = pl_index; path->fp_weight = 1; path->fp_nh_proto = nh_proto; path->fp_via_fib = FIB_NODE_INDEX_INVALID; path->fp_cfg_flags = flags; if (FIB_PATH_CFG_FLAG_DROP & flags) { path->fp_type = FIB_PATH_TYPE_SPECIAL; } else if (FIB_PATH_CFG_FLAG_LOCAL & flags) { path->fp_type = FIB_PATH_TYPE_RECEIVE; path->attached.fp_interface = FIB_NODE_INDEX_INVALID; } else { //本场景使用的是这里 path->fp_type = FIB_PATH_TYPE_EXCLUSIVE; ASSERT(NULL != dpo); dpo_copy(&path->exclusive.fp_ex_dpo, dpo); } return (fib_path_get_index(path)); } static fib_path_list_t *fib_path_list_resolve (fib_path_list_t *path_list) { fib_node_index_t *path_index, *paths, path_list_index; ASSERT(!(path_list->fpl_flags & FIB_PATH_LIST_FLAG_RESOLVED)); /* * resolving a path-list is a recursive action. this means more path * lists can be created during this call, and hence this path-list * can be realloc'd. so we work with copies. * this function is called only once per-path list, so its no great overhead. */ path_list_index = fib_path_list_get_index(path_list); paths = vec_dup(path_list->fpl_paths); vec_foreach (path_index, paths) { //FIB_PATH_TYPE_RECURSIVE类型的path会导致递归调用，可能导致path_list pool重分配。 fib_path_resolve(*path_index); } vec_free(paths); path_list = fib_path_list_get(path_list_index); FIB_PATH_LIST_DBG(path_list, "resovled"); fib_path_list_mk_urpf(path_list); return (path_list); } //本函数最终目的是为path分析出DPO，保存到path->fp_dpo成员中，细节太多，博主已晕，以后再补 int fib_path_resolve (fib_node_index_t path_index) { fib_path_t *path; path = fib_path_get(path_index); // hope for the best. path->fp_oper_flags |= FIB_PATH_OPER_FLAG_RESOLVED; // the forced drop path resolves via the drop adj if (fib_path_is_permanent_drop(path)) { dpo_copy(&path->fp_dpo, drop_dpo_get(fib_proto_to_dpo(path->fp_nh_proto))); path->fp_oper_flags &= ~FIB_PATH_OPER_FLAG_RESOLVED; return (fib_path_is_resolved(path_index)); } switch (path->fp_type) { case FIB_PATH_TYPE_ATTACHED_NEXT_HOP: fib_path_attached_next_hop_set(path); break; case FIB_PATH_TYPE_ATTACHED: // path->attached.fp_interface if (!vnet_sw_interface_is_admin_up(vnet_get_main(), path->attached.fp_interface)) { path->fp_oper_flags &= ~FIB_PATH_OPER_FLAG_RESOLVED; } if (vnet_sw_interface_is_p2p(vnet_get_main(), path->attached.fp_interface)) { /* * point-2-point interfaces do not require a glean, since * there is nothing to ARP. Install a rewrite/nbr adj instead */ dpo_set(&path->fp_dpo, DPO_ADJACENCY, fib_proto_to_dpo(path->fp_nh_proto), adj_nbr_add_or_lock( path->fp_nh_proto, fib_proto_to_link(path->fp_nh_proto), &zero_addr, path->attached.fp_interface)); } else { dpo_set(&path->fp_dpo, DPO_ADJACENCY_GLEAN, fib_proto_to_dpo(path->fp_nh_proto), adj_glean_add_or_lock(path->fp_nh_proto, path->attached.fp_interface, NULL)); } /* * become a child of the adjacency so we receive updates * when the interface state changes */ path->fp_sibling = adj_child_add(path->fp_dpo.dpoi_index, FIB_NODE_TYPE_PATH, fib_path_get_index(path)); break; case FIB_PATH_TYPE_RECURSIVE: { /* * Create a RR source entry in the table for the address * that this path recurses through. * This resolve action is recursive, hence we may create * more paths in the process. more creates mean maybe realloc * of this path. */ fib_node_index_t fei; fib_prefix_t pfx; ASSERT(FIB_NODE_INDEX_INVALID == path->fp_via_fib); if (FIB_PROTOCOL_MPLS == path->fp_nh_proto) { fib_prefix_from_mpls_label(path->recursive.fp_nh.fp_local_label, &pfx); } else { fib_prefix_from_ip46_addr(&path->recursive.fp_nh.fp_ip, &pfx); } //把下一跳也插入路由表，开始递归了 fei = fib_table_entry_special_add(path->recursive.fp_tbl_id, &pfx, FIB_SOURCE_RR, FIB_ENTRY_FLAG_NONE, ADJ_INDEX_INVALID); path = fib_path_get(path_index); path->fp_via_fib = fei; /* * become a dependent child of the entry so the path is * informed when the forwarding for the entry changes. */ path->fp_sibling = fib_entry_child_add(path->fp_via_fib, FIB_NODE_TYPE_PATH, fib_path_get_index(path)); // create and configure the IP DPO fib_path_recursive_adj_update( path, fib_path_proto_to_chain_type(path->fp_nh_proto), &path->fp_dpo); break; } case FIB_PATH_TYPE_SPECIAL: // Resolve via the drop dpo_copy(&path->fp_dpo, drop_dpo_get(fib_proto_to_dpo(path->fp_nh_proto))); break; case FIB_PATH_TYPE_DEAG: /* * Resolve via a lookup DPO. * FIXME. control plane should add routes with a table ID */ lookup_dpo_add_or_lock_w_fib_index(path->deag.fp_tbl_id, fib_proto_to_dpo(path->fp_nh_proto), LOOKUP_INPUT_DST_ADDR, LOOKUP_TABLE_FROM_CONFIG, &path->fp_dpo); break; case FIB_PATH_TYPE_RECEIVE: // Resolve via a receive DPO. receive_dpo_add_or_lock(fib_proto_to_dpo(path->fp_nh_proto), path->receive.fp_interface, &path->receive.fp_addr, &path->fp_dpo); break; case FIB_PATH_TYPE_EXCLUSIVE: // Resolve via the user provided DPO dpo_copy(&path->fp_dpo, &path->exclusive.fp_ex_dpo); break; } return (fib_path_is_resolved(path_index)); }static void fib_entry_src_api_path_swap (fib_entry_src_t *src, const fib_entry_t *entry, fib_path_list_flags_t pl_flags, const fib_route_path_t *paths) { //这里是为了减少path_list重复，看看hash表里有没有一样的path_list，没有就加进hash表里 src->fes_pl = fib_path_list_create((FIB_PATH_LIST_FLAG_SHARED | pl_flags), paths); } fib_node_index_t fib_path_list_create (fib_path_list_flags_t flags, const fib_route_path_t *rpaths) { fib_node_index_t path_list_index, old_path_list_index; fib_path_list_t *path_list; int i; flags = fib_path_list_flags_fixup(flags); path_list = fib_path_list_alloc(&path_list_index); path_list->fpl_flags = flags; // we'll assume for now all paths are the same next-hop protocol path_list->fpl_nh_proto = rpaths[0].frp_proto; //创建path_list中的所有path vec_foreach_index(i, rpaths) { vec_add1(path_list->fpl_paths, fib_path_create(path_list_index, path_list->fpl_nh_proto, fib_path_list_flags_2_path_flags(flags), &rpaths[i])); } // If a shared path list is requested, consult the DB for a match if (flags & FIB_PATH_LIST_FLAG_SHARED) { /* * check for a matching path-list in the DB. * If we find one then we can return the existing one and destroy the * new one just created. * fib_path_list_db_hash_key_equal。 * 可以知道，key其实是把path_list中每一个path的部分内容， * 累积成 key，可以无脑理解为就是确保两个path_list中每个path内容一样， * 才能认为这两个path_list相等。*/ old_path_list_index = fib_path_list_db_find(path_list); if (FIB_NODE_INDEX_INVALID != old_path_list_index) { fib_path_list_destroy(path_list); path_list_index = old_path_list_index; } else { /* * if there was not a matching path-list, then this * new one will need inserting into the DB and resolving. */ fib_path_list_db_insert(path_list_index); path_list = fib_path_list_resolve(path_list); } } else { /* * no shared path list requested. resolve and use the one * just created. */ path_list = fib_path_list_resolve(path_list); } return (path_list_index); }转发表安装路由表，路由表项，source，path_list，path都准备好了。接下来要安装到转发表中了，安装后才是真正对数据包转发生效void fib_entry_src_action_activate (fib_entry_t *fib_entry, fib_source_t source) { int houston_we_are_go_for_install; fib_entry_src_t *esrc; esrc = fib_entry_src_find(fib_entry, source, NULL); ASSERT(!(esrc->fes_flags & FIB_ENTRY_SRC_FLAG_ACTIVE)); ASSERT(esrc->fes_flags & FIB_ENTRY_SRC_FLAG_ADDED); esrc->fes_flags |= FIB_ENTRY_SRC_FLAG_ACTIVE; if (NULL != fib_entry_src_vft[source].fesv_activate) { houston_we_are_go_for_install = fib_entry_src_vft[source].fesv_activate(esrc, fib_entry); } else { /* * the source is not providing an activate function, we'll assume * therefore it has no objection to installing the entry */ houston_we_are_go_for_install = !0; } /* * link to the path-list provided by the source, and go check * if that forms any loops in the graph. */ fib_entry->fe_parent = esrc->fes_pl; fib_entry->fe_sibling = fib_path_list_child_add(fib_entry->fe_parent, FIB_NODE_TYPE_ENTRY, fib_entry_get_index(fib_entry)); fib_entry_recursive_loop_detect_i(fib_entry->fe_parent); FIB_ENTRY_DBG(fib_entry, "activate: %d", fib_entry->fe_parent); if (0 != houston_we_are_go_for_install) { fib_entry_src_action_install(fib_entry, source); } else { fib_entry_src_action_uninstall(fib_entry); } } void fib_entry_src_action_install (fib_entry_t *fib_entry, fib_source_t source) { /* * Install the forwarding chain for the given source into the forwarding * tables */ fib_forward_chain_type_t fct; fib_entry_src_t *esrc; int insert; fct = fib_entry_get_default_chain_type(fib_entry); esrc = fib_entry_src_find(fib_entry, source, NULL); /* * Every entry has its own load-balance object. All changes to the entry's * forwarding result in an inplace modify of the load-balance. This means * the load-balance object only needs to be added to the forwarding * DB once, when it is created. */ insert = !dpo_id_is_valid(&fib_entry->fe_lb); //准备转发表中的最终DPO。利用上文提到的path->fp_dpo，来构建最终DPO fib_entry_src_mk_lb(fib_entry, esrc, fct, &fib_entry->fe_lb); ASSERT(dpo_id_is_valid(&fib_entry->fe_lb)); FIB_ENTRY_DBG(fib_entry, "install: %d", fib_entry->fe_lb); // insert the adj into the data-plane forwarding trie if (insert) { //安装到8-8-8-8 mtrie转发表 fib_table_fwding_dpo_update(fib_entry->fe_fib_index, &fib_entry->fe_prefix, &fib_entry->fe_lb); } /* * if any of the other chain types are already created they will need * updating too */ fib_entry_delegate_type_t fdt; fib_entry_delegate_t *fed; FOR_EACH_DELEGATE_CHAIN(fib_entry, fdt, fed, { fib_entry_src_mk_lb(fib_entry, esrc, fib_entry_delegate_type_to_chain_type(fdt), &fed->fd_dpo); }); } void fib_entry_src_mk_lb (fib_entry_t *fib_entry, const fib_entry_src_t *esrc, fib_forward_chain_type_t fct, dpo_id_t *dpo_lb) { dpo_proto_t lb_proto; /* * If the entry has path extensions then we construct a load-balance * by stacking the extensions on the forwarding chains of the paths. * Otherwise we use the load-balance of the path-list */ fib_entry_src_collect_forwarding_ctx_t ctx = { .esrc = esrc, .fib_entry = fib_entry, .next_hops = NULL, .is_recursive = 0, .fct = fct, }; /* * As an optimisation we allocate the vector of next-hops to be sized * equal to the maximum nuber of paths we will need, which is also the * most likely number we will need, since in most cases the paths are 'up'. */ vec_validate(ctx.next_hops, fib_path_list_get_n_paths(esrc->fes_pl)); vec_reset_length(ctx.next_hops); lb_proto = fib_proto_to_dpo(fib_entry->fe_prefix.fp_proto); fib_path_list_walk(esrc->fes_pl, fib_entry_src_collect_forwarding, &ctx); if (esrc->fes_entry_flags & FIB_ENTRY_FLAG_EXCLUSIVE) { /* * the client provided the DPO that the entry should link to. * all entries must link to a LB, so if it is an LB already * then we can use it. */ if ((1 == vec_len(ctx.next_hops)) && (DPO_LOAD_BALANCE == ctx.next_hops[0].path_dpo.dpoi_type)) { dpo_copy(dpo_lb, &ctx.next_hops[0].path_dpo); dpo_reset(&ctx.next_hops[0].path_dpo); return; } } if (!dpo_id_is_valid(dpo_lb)) { // first time create flow_hash_config_t fhc; fhc = fib_table_get_flow_hash_config(fib_entry->fe_fib_index, dpo_proto_to_fib(lb_proto)); dpo_set(dpo_lb, DPO_LOAD_BALANCE, lb_proto, load_balance_create(0, lb_proto, fhc)); } load_balance_multipath_update(dpo_lb, ctx.next_hops, fib_entry_calc_lb_flags(&ctx)); vec_free(ctx.next_hops); /* * if this entry is sourced by the uRPF-exempt source then we * append the always present local0 interface (index 0) to the * uRPF list so it is not empty. that way packets pass the loose check. */ index_t ui = fib_path_list_get_urpf(esrc->fes_pl); if (fib_entry_is_sourced(fib_entry_get_index(fib_entry), FIB_SOURCE_URPF_EXEMPT) && (0 == fib_urpf_check_size(ui))) { /* * The uRPF list we get from the path-list is shared by all * other users of the list, but the uRPF exemption applies * only to this prefix. So we need our own list. */ ui = fib_urpf_list_alloc_and_lock(); fib_urpf_list_append(ui, 0); fib_urpf_list_bake(ui); load_balance_set_urpf(dpo_lb->dpoi_index, ui); fib_urpf_list_unlock(ui); } else { load_balance_set_urpf(dpo_lb->dpoi_index, ui); } }
注：本文是转载，但不是100%的转载，可能稍微有些出入，原文地址点击这里基本概念VPP作为思科开源项目，路由系统基本继承了Cisco快速转发（Cisco Express Forwarding，CEF）设计思路。作为理解VPP路由源码第一步，首先要掌握起理论基础。Cisco Express Forwarding内容摘自异步社区。虽然基于缓存的交换机制相对进程交换而言提高了转发的性能，但是它们的性能是不确定的。进程交换和基于缓存的交换都是数据驱动的（data-driven），或者说是需求驱动的（demand-driven）。换句话说，只有当数据包进入路由器的时候交换组件才会被布置到位，一旦这些数据包不被路由器转发，交换组件就会被清除。如果存在大量的不可预知流量模式的数据包，交换性能就会显著地下降。显然，在Internet的级别上，这些交换路线是不可扩展的。创建CEF是为了避免基于缓存交换机制所固有的问题。它的设计最好地适应了频繁变化的网络态势和流量特征，这些都是由于不断增长的较短持续时间的流所产生的。典型地，这些较短持续时间的流与基于Web的应用和交互式TCP会话相关联。CEF具有以下几个优点：可扩展性（Scalability）——CEF是拓扑驱动的（topology-driven），并与路由选择表紧密相关。当激活了分布式CEF（Distributed CEF）模式时，CEF在每一块线卡（line card）上也提供了全部的交换能力。CEF支持硬件辅助的（hardware-assisted）转发方式，这是在高容量的线卡上，提供线速交换（line rate switching）能力所必需的。增强的性能——CEF的CPU密集程度比路由缓存机制的CPU密集程度低。更多的CPU处理能力可以专注于第3层服务，例如，处理BGP更新。弹性（Resilience）——在大型动态网络中，CEF提供了更好的交换一致性和稳定性。在这样一些网络中，由于路由选择的变化，快速交换缓存表项会频繁地失效。这些变化导致路由器使用路由选择表进程交换流量，而不是使用路由缓存快速交换流量。由于CEF查找表包含了存在于路由选择表中所有已知的路由，因此它消除了路由缓存的维护需要，并使快速交换/进程交换转发不再适用。CEF能够比典型的按需缓存机制更有效地交换流量。注意：IP RIB中所有路由的表项不管是否被使用，它们都需要被维护，因而CEF可能比其他交换方式需要更多的内存。CEF是拓扑驱动的交换机制，它的转发表和路由选择表是紧密关联的。无论何时路由选择表发生变化，CEF转发表也将被更新。在表项被创建的时候，数据包被交换到更慢的交换路线上。CEF把路由缓存的功能分割成两部分：转发信息库(Forwarding Information Base，FIB)邻接表(ajdancency table)FIBFIB包含了来自于路由选择表的所有IP前缀。如果不同的路由选择表被维护着，例如在MPLS VPN环境中，那么每个VPN都有自己的FIB。FIB不是数据驱动的。更确切地说，它是通过路由选择表来创建和更新的。FIB子系统负责确保所有的递归路由（是指那些没有与直接下一跳相联系的路由）被解析。为了增强一致性和减少查找时间，FIB被组织成一个被称为mtrie的多路（multiway）数据结构。在mtrie数据结构中，树结构用来定位所要找的数据，但是数据本身存储在其他地方。相反，mtree数据结构在树结构本身中存储了实际的数据。例如，在最优交换的mtree缓存中，用来转发数据包的MAC头的数据就是实际存储于mtree中的。Cisco路由器通常使用两种类型的mtrie结构：8-8-8-8——这种格式也被称为256-way mtrie，因为4个八位组的IPv4地址被映射到4个8bit的结构中。因此，一条前缀的最大查找次数是4次。这种格式用在大多数Cisco路由器中。16-8-8——这是一个3级的mtrie，它的根级有65536个表项。因此，一条前缀的最大查找次数是3次。换句话说，第一次查找解析了前面的两个八位组，接着最多再需要2次查找就可以解析最后的两个八位组。这种格式只用于Cisco 12000系列路由器。mtrie的每一级都被称为一个节点（node）。最后的节点被称为叶节点（leaf）。叶节点指向邻接表（adjacency table），或者当到达相同目的地的多条路径存在时，指向另一个负载分担的结构。IP FIB的内容可以使用show ip cef命令来显示。下面列举了一些FIB表项：附接的（attached）——这种前缀被配置为可以通过接口直接到达，不需要由IP下一跳来创建邻接关系。这种前缀是指路由器本地接口所属的网络。连接的（connected）——这种接口是由IP addressaddress mask配置命令来配置的。所有连接的FIB表项都是附接的，但不是所有附接的表项都是连接的。收到（receive）——这种前缀是一个32位掩码的主机地址，它是路由器始终接收到的主机地址之一。每个接口通常有3种这样的地址：实际的接口地址、全0的子网和全1的广播地址。（博主理解为就是本机local地址）递归的（recursive）——当前缀的输出接口不能通过路由选择协议或静态配置指定时，它就被标记为递归的。当找不到递归FIB表项的下一跳IP地址时，这个递归FIB表项也许就不能被解析。因此，递归标记实际上与下一跳地址，而不是与FIB表项相关联。邻接表邻接表被创建，以包含所有连接的下一跳地址。邻接节点（adjacency node）是指通过链路层一跳就可以到达的节点。一旦邻居变成了邻接的关系，用来到达那个邻居的链路层帧头——被称为MAC字符串或者MAC改写字符串——就会被创建，并且被保存在邻接表中。例如，在以太网段上，帧头信息就是目的地MAC地址、源MAC地址和以太类型（EtherType），并按以太网的定义来排列。例2-6显示了以太网的MAC头。在这个例子中，00044EB31838是目的地MAC地址，0003E4BB2000是源MAC地址，0800是IP的以太类型。一旦路由被解析，它就会指向一个邻接的下一跳地址。如果在邻接表中发现了一个邻接，那么与这个邻接相对应的一个指针就会被缓存到FIB单元中。如果存在到达相同目的地的多条路径（也就是说，存在多个下一跳或者邻接），那么每一个邻接的指针都会被增添到负载分担的结构中去。在CEF中，基于每数据包的负载分担在中断级别上是有用的。除了上述情况，还存在几种异常邻接的类型。当前缀被增加到FIB表中时，需要异常处理（exception handling）的前缀被缓存为特殊的邻接。下面列举了一些特殊的邻接：Null——这种邻接针对那些指向Null 0接口的、要被丢弃的数据包。Glean——这种邻接针对那些通过广播网络附接的目的地，但是其广播网络又没有对应可用的MAC改写字符串。可以设想路由器直连到一个包含几台主机的子网的情形。路由器的FIB表维护着这个子网的前缀，而不是单个的主机前缀。这个子网前缀就指向一个glean邻接。当数据包需要被转发到一台特定的主机时，为这个特定的前缀就需要收集（glean）邻接数据库。这将导致额外的查找成本。（博主理解为，就是同一个网络中的主机，目前mac地址不知道，所以用glean表项标记其连接表项，然后再由arp协议来填充）Punt——如果数据包不被CEF支持，那么它们就会被转发给下一级慢速的交换路线来处理。Drop——是指要丢弃这种数据包，因为这些数据包不能被CEF交换，或者不能被踢给（punt）其他交换路线处理。Discard——类似于Drop邻接，但是只应用于Cisco 12000系列路由器。下图显示了CEF的所有组件之间的关联关系。
注：本文是转载，但不是100%的转载，可能稍微有些出入，原文地址点击这里vpp的功能逻辑被划分为一个个node，node之间通过下一跳传递处理完的数据包集合，从而组成整个业务图谱。本文将分析node调度框架源码。基本概念vlib_node_type_tnode分为四种类型：VLIB_NODE_TYPE_INTERNAL对数据包真正处理的业务node。VLIB_NODE_TYPE_INPUT收包逻辑node，比如：dpdk，pcap等。VLIB_NODE_TYPE_PRE_INPUT目前只有一个epoll node，对socket相关逻辑提供服务，主要使用在控制业务上。VLIB_NODE_TYPE_PROCESS该类型node可以被挂起也可以被恢复，有独立的分配在heap上的运行时栈。类似于在一个线程中实现了多任务调度机制。主要用来修改vpp node内部参数。vlib_node_main_t记录各种全局信息，比如各种数据结构集合，数据结构内存池之类。vlib_node_t注册node时将业务逻辑，几乎不怎么修改的参数，状态信息保存在这里。vlib_node_runtime_t这是调度框架实际频繁使用的结构，从vlib_node_t拷贝了部分信息，以及私有的频繁变动的信息。vlib_process_tVLIB_NODE_TYPE_PROCESS类型node专用结构，记录用于模拟task的基础结构：heap上的运行时栈，2种返回时寄存器备份，等。vlib_frame_t每个node都有一个对应的vlib_frame_t，用来保存供node使用的数据包集合。这是每个node最终处理数据的内存所在地。vlib_pending_frame_t当一个node处理完数据包，则填充该数据结构，并加入到全局链表，调度框架便能在下一次调度时找到需要接收该数据包的下一个node。vlib_next_frame_t主要是node内部逻辑使用，定位该node的下一条信息。关键结构vlib_node_main_ttypedef struct { //一块连续内存，头部是vec_header_t，数据部分是node指针数组 vlib_node_t **nodes; //node按名字组成hash表 uword *node_by_name; //目前只有VLIB_NODE_MAIN_RUNTIME_STARTED一个状态，暂时没啥用，忽略之。 u32 flags; #define VLIB_NODE_MAIN_RUNTIME_STARTED (1 << 0) /* * node有三种类型:VLIB_NODE_TYPE_INTERNAL,VLIB_NODE_TYPE_INPUT, * VLIB_NODE_TYPE_PRE_INPUT,按类型分类索引 */ vlib_node_runtime_t *nodes_by_type[VLIB_N_NODE_TYPE]; //以下三个用于类似于网卡napi收包模式，dpdk没用到，忽略之 u32 *pending_interrupt_node_runtime_indices; //假设node1有n1个下一跳，node2有n2个下一跳....共n1 + n2 +....+ni个下一跳 //连续保存在next_frames指向的vec中。 vlib_next_frame_t *next_frames; /* * 数据包从node输出到下一跳，那么下一跳node即是pending frame，会加入pending_frames指向的vec。 * 主循环会遍历该vec，对每个node来调用处理逻辑 */ vlib_pending_frame_t *pending_frames; //定时器，信号相关，源码及其恶心，博主暂时放弃阅读 timing_wheel_t timing_wheel; vlib_signal_timed_event_data_t *signal_timed_event_data_pool; /* * VLIB_NODE_TYPE_PROCESS类型的node，每个node有分配在heap上的运行栈， * 没错就是通常说的在顶部那种栈，但是这里分配在了heap上。 * 这里在单线程中模拟了类似多线程的效果，以后会详细分析。 * processes保存了所有VLIB_NODE_TYPE_PROCESS类型node的描述结构指针。 */ vlib_process_t **processes; //VLIB_NODE_TYPE_PROCESS类型的node,执行时都会把 //current_process_index赋值为本node的runtime_index u32 current_process_index; //VLIB_NODE_TYPE_PROCESS类型node专用，挂起时在其中保存信息。 vlib_pending_frame_t *suspended_process_frames; //hash表，把node的scalar_size，vector_size值组合成key，查找对应的vlib_frame_size_t结构。 uword *frame_size_hash; //通过hash查找到node对应的vlib_frame_size_t值，从中的内存池来分配vlib_frame_t。 vlib_frame_size_t *frame_sizes; //注册node函数提交的注册信息链接在这里，仅仅初始化时使用 vlib_node_registration_t *node_registrations; } vlib_node_main_t;vlib_node_ttypedef struct vlib_node_t { //业务逻辑 vlib_node_function_t *function; //node类型，之前提到的那四种之一 vlib_node_type_t type; //vlib_node_t和vlib_node_runtime_t是一一对应的好基友 u32 runtime_index; //这两个成员组合成key，在vlib_node_main_t->frame_size_hash中查找， //确定本node相关的frame的内存池 u16 scalar_size, vector_size; //初始化时用用 char **next_node_names; //根据next_node_names来生成next_nodes，vec结构，记录了每个可选下一条的index。 u32 *next_nodes; //统计发给下一条node的数据包总数 u64 *n_vectors_by_next_node; } vlib_node_t;vlib_node_runtime_ttypedef struct vlib_node_runtime_t { //本node的多个下一条中，第一个下一条在vlib_node_main_t->next_frames中的索引 u32 next_frame_index; //vlib_node_t和vlib_node_runtime_t是一一对应的好基友 u32 node_index; }vlib_process_ttypedef struct { //对应的PROSESS类型node索引号 vlib_node_runtime_t node_runtime; //如下return_longjmp，resume_longjmp用于保存当前寄存器内容， //可以参考C库的setjump，longjump原理 clib_longjmp_t return_longjmp; //PROSESS类型node的业务逻辑运行时栈地址，在heap上 #define VLIB_PROCESS_STACK_MAGIC (0xdead7ead) u32 stack[0] ALIGN_ON_MULTIPLE_PAGE_BOUNDARY_FOR_MPROTECT; } vlib_process_t __attribute__ ((aligned (CLIB_CACHE_LINE_BYTES)));调度逻辑VPP支持多工作线程模型，这里不考虑多线程，以单线程模型来分析源码。核心调度函数：vlib_main_loopstatic void vlib_main_loop (vlib_main_t * vm) { ... //dpdk收包时下面用不到。它们是用来模拟标准网卡NAPI机制的 if (!nm->polling_threshold_vector_length) nm->polling_threshold_vector_length = 10; if (!nm->interrupt_threshold_vector_length) nm->interrupt_threshold_vector_length = 5; nm->current_process_index = ~0; /* * 执行所有VLIB_NODE_TYPE_PROCESS类型node，利用setjump，longjump机制， * 把node挂起来，等待之后唤醒。 * 可以理解为一种多任务模型。该类型node主要时用在作运行时配置相关。 * 之后会详细论述该类型node。 */ uword i; for (i = 0; i < vec_len (nm->processes); i++) cpu_time_now = dispatch_process (vm, nm->processes[i], /* frame */ 0, cpu_time_now); while (1) { vlib_node_runtime_t *n; //目前只有一个epoll相关的node，监听socket，辅助功能 vec_foreach (n, nm->nodes_by_type[VLIB_NODE_TYPE_PRE_INPUT]) cpu_time_now = dispatch_node (vm, n, VLIB_NODE_TYPE_PRE_INPUT, VLIB_NODE_STATE_POLLING, /* frame */ 0, cpu_time_now); //收包node，假设使用dpdk的node vec_foreach (n, nm->nodes_by_type[VLIB_NODE_TYPE_INPUT]) cpu_time_now = dispatch_node (vm, n, VLIB_NODE_TYPE_INPUT, VLIB_NODE_STATE_POLLING, /* frame */ 0, cpu_time_now); //memclnt_node会用到，但是博主发现该node没有使用。无视之。 if (PREDICT_TRUE (vm->queue_signal_pending == 0)) vm->queue_signal_callback (vm); //dpdk没有中断机制，这里不会执行 { ... } /* * 处理超时事件，通过api发生的唤醒node事件。 * 只对VLIB_NODE_TYPE_PROCESS类型node有效。 * VLIB_NODE_TYPE_PROCESS类型node处理需要单独描述， * 但不是开发人员关注的重点，只是配置相关。 */ nm->data_from_advancing_timing_wheel = timing_wheel_advance (&nm->timing_wheel, cpu_time_now, nm->data_from_advancing_timing_wheel, &nm->cpu_time_next_process_ready); ... /* * 真正开发人员关注的重点，数据包处理的核心逻辑都在VLIB_NODE_TYPE_INTERNAL类型node中。 * nm->pending_frames记录了上一个node转给下一个node信息， * 最后可以找到传递给下一个node使用的数据包 */ for (i = 0; i < _vec_len (nm->pending_frames); i++) cpu_time_now = dispatch_pending_node (vm, nm->pending_frames + i, cpu_time_now); } }u64 dispatch_pending_node (vlib_main_t * vm, vlib_pending_frame_t * p, u64 last_time_stamp) { vlib_node_main_t *nm = &vm->node_main; vlib_frame_t *f; vlib_next_frame_t *nf, nf_dummy; vlib_node_runtime_t *n; u32 restore_frame_index; //之后将执行n的业务逻辑 n = vec_elt_at_index (nm->nodes_by_type[VLIB_NODE_TYPE_INTERNAL], p->node_runtime_index); //node n的对应frame，里面包含了node需要处理的数据包 f = vlib_get_frame (vm, p->frame_index); /* * 不是下一跳传过来的数据包，比如自己生成的包注入某个node，此时if判断为真。 * vlib_put_frame_to_node()完成注入逻辑。 */ if (p->next_frame_index == VLIB_PENDING_FRAME_NO_NEXT_FRAME) { nf = &nf_dummy; nf->flags = f->flags & VLIB_NODE_FLAG_TRACE; nf->frame_index = ~p->frame_index; } else //通过node下一跳传递来的数据包 nf = vec_elt_at_index (nm->next_frames, p->next_frame_index); /* * node传递数据包到下一个node是通过调用vlib_put_next_frame()接口: * p->frame_index = nf->frame_index; * p->node_runtime_index = nf->node_runtime_index; * p->next_frame_index = nf - nm->next_frames; */ if (nf->frame_index == p->frame_index) { //有两个位置同时引用了frame，把nf中的清掉，以免干扰本线程对frame的使用。 nf->frame_index = ~0; nf->flags &= ~VLIB_FRAME_IS_ALLOCATED; if (!(n->flags & VLIB_NODE_FLAG_FRAME_NO_FREE_AFTER_DISPATCH)) restore_frame_index = p->frame_index; } ... //调度node业务逻辑 last_time_stamp = dispatch_node (vm, n, VLIB_NODE_TYPE_INTERNAL, VLIB_NODE_STATE_POLLING, f, last_time_stamp); f->flags &= ~VLIB_FRAME_PENDING; //frame已经在业务逻辑中使用完了，重新保存到nf中 if (restore_frame_index != ~0) { //vlib_next_frame_change_ownership()可能会修改p->next_frame_index nf = vec_elt_at_index (nm->next_frames, p->next_frame_index); nf->frame_index = restore_frame_index; nf->flags |= VLIB_FRAME_IS_ALLOCATED; } return last_time_stamp; }//对DPDK驱动，该函数基本没有什么复杂处理。 u64 dispatch_node (vlib_main_t * vm, vlib_node_runtime_t * node, vlib_node_type_t type, vlib_node_state_t dispatch_state, vlib_frame_t * frame, u64 last_time_stamp) { ... if (node->n_next_nodes > 0) { //node的业务逻辑肯定会用到nf nf = vec_elt_at_index (nm->next_frames, node->next_frame_index); CLIB_PREFETCH (nf, 4 * sizeof (nf[0]), WRITE); } ... //业务逻辑 n = node->function (vm, node, frame); ... //跟dpdk没关系了 if ((DPDK == 0 && dispatch_state == VLIB_NODE_STATE_INTERRUPT) || (DPDK == 0 && dispatch_state == VLIB_NODE_STATE_POLLING && (node->flags & VLIB_NODE_FLAG_SWITCH_FROM_INTERRUPT_TO_POLLING_MODE))) { ... } return t; }有时间再详细分析，该函数背后机制比较复杂，但是跟vpp框架使用者关系不大。static u64 dispatch_process (vlib_main_t * vm, vlib_process_t * p, vlib_frame_t * f, u64 last_time_stamp) { ... }
注：本文是转载，但不是100%的转载，可能稍微有些出入，原文地址点击这里基本概念这个只是辅助部分，大多数人不关心，我就随便写写。核心函数//主线程会调用unix_cli_config，开始cli部分的初始化工作 VLIB_CONFIG_FUNCTION (unix_cli_config, "unix-cli");/** Handle configuration directives in the @em unix section. */ static clib_error_t *unix_cli_config (vlib_main_t * vm, unformat_input_t * input) { ... //确保unix_config执行且只执行了一次 if ((error = vlib_call_config_function (vm, unix_config))) return error; if (um->flags & UNIX_FLAG_INTERACTIVE) { //支持交互模式 //非阻塞模式，UNIX_CLI_STDIN_FD会交由VLIB_NODE_TYPE_PRE_INPUT类型的 //unix-epoll-input node来监视 if ((flags = fcntl (UNIX_CLI_STDIN_FD, F_GETFL, 0)) < 0) flags = 0; (void) fcntl (UNIX_CLI_STDIN_FD, F_SETFL, flags | O_NONBLOCK); //生成或者复用一个VLIB_NODE_TYPE_PROCESS类型node， //并调度激活node，等待来处理该cli session cf_index = unix_cli_file_add (cm, "stdin", UNIX_CLI_STDIN_FD); cf = pool_elt_at_index (cm->cli_file_pool, cf_index); cm->stdin_cli_file_index = cf_index; //下面是调整cli界面参数的，暂时不分析了。 if (isatty (UNIX_CLI_STDIN_FD) && um->cli_line_mode == 0) { ... } //一些欢迎信息拷贝到cf->output_vector中，更新epoll状态，等待时机输出。 unix_cli_file_welcome (cm, cf); } //下面是CLI socket支持 /* If we have socket config, LISTEN, otherwise, don't */ clib_socket_t *s = &um->cli_listen_socket; if (s->config && s->config[0] != 0) { /* CLI listen. */ unix_file_t template = { 0 }; s->flags = SOCKET_IS_SERVER; /* listen, don't connect */ //socket初始化，监听模式，代码不难 error = clib_socket_init (s); if (error) return error; template.read_function = unix_cli_listen_read_ready; template.file_descriptor = s->fd; //socket加入到epoll unix_file_add (um, &template); } /* Set CLI prompt. */ if (!cm->cli_prompt) cm->cli_prompt = format (0, "VLIB: "); return 0; }//该函数启动一个CLI session，关键是要提前理解VLIB_NODE_TYPE_PROCESS类型node运行机制 /** Store a new CLI session. * @param name The name of the session. * @param fd The file descriptor for the session I/O. * @return The session ID. */ static u32 unix_cli_file_add (unix_cli_main_t * cm, char *name, int fd) { ... //CLI退出时，会把处理该CLI的node放入这个复用链表池，以待继续使用 if (vec_len (cm->unused_cli_process_node_indices) > 0) { uword l = vec_len (cm->unused_cli_process_node_indices); /* Find node and give it new name. */ n = vlib_get_node (vm, cm->unused_cli_process_node_indices[l - 1]); vec_free (n->name); n->name = (u8 *) name; vlib_node_set_state (vm, n->index, VLIB_NODE_STATE_POLLING); _vec_len (cm->unused_cli_process_node_indices) = l - 1; } else { //池子里没有，就新生成一个 static vlib_node_registration_t r = { .function = unix_cli_process, .type = VLIB_NODE_TYPE_PROCESS, .process_log2_n_stack_bytes = 16, }; r.name = name; vlib_register_node (vm, &r); vec_free (name); n = vlib_get_node (vm, r.index); } pool_get (cm->cli_file_pool, cf); memset (cf, 0, sizeof (*cf)); template.read_function = unix_cli_read_ready; template.write_function = unix_cli_write_ready; template.file_descriptor = fd; template.private_data = cf - cm->cli_file_pool; cf->process_node_index = n->index; //该文件fd加入epoll cf->unix_file_index = unix_file_add (um, &template); cf->output_vector = 0; cf->input_vector = 0; //启动该VLIB_NODE_TYPE_PROCESS类型node，它会进入“睡眠”状态，等待外部命令行输入来激活node vlib_start_process (vm, n->runtime_index); vlib_process_t *p = vlib_get_process_from_node (vm, n); p->output_function = unix_vlib_cli_output; p->output_function_arg = cf - cm->cli_file_pool; return cf - cm->cli_file_pool; }当VLIB_NODE_TYPE_PROCESS类型node被CLI输入激活时，会从vlib_process_wait_for_event (vm)下一行开始执行，理解这里需要VLIB_NODE_TYPE_PROCESS类型node调度的前置知识。/** Handle system events. */ static uword unix_cli_process (vlib_main_t * vm, vlib_node_runtime_t * rt, vlib_frame_t * f) { unix_cli_main_t *cm = &unix_cli_main; uword i, *data = 0; while (1) { unix_cli_process_event_type_t event_type; vlib_process_wait_for_event (vm); //激活该node的事件类型 event_type = vlib_process_get_events (vm, &data); switch (event_type) { //有CLI 命令来了，开始解析 case UNIX_CLI_PROCESS_EVENT_READ_READY: for (i = 0; i < vec_len (data); i++) //解析命令行 unix_cli_process_input (cm, data[i]); break; //CLI要退出了 case UNIX_CLI_PROCESS_EVENT_QUIT: /* Kill this process. */ for (i = 0; i < vec_len (data); i++) //如果是标准输入上来的退出命令，那就是退出进程，否则只需要退出该CLI session unix_cli_kill (cm, data[i]); goto done; } if (data) _vec_len (data) = 0; } done: vec_free (data); vlib_node_set_state (vm, rt->node_index, VLIB_NODE_STATE_DISABLED); /* Add node index so we can re-use this process later. */ //该node完成使命，不需要在主循环上“睡眠”了，丢给复用池 vec_add1 (cm->unused_cli_process_node_indices, rt->node_index); return 0; }//该函数用在epoll监听到有数据到来时，调用它接受数据。逻辑不难。 /** Called when a CLI session file descriptor has data to be read. */ static clib_error_t *unix_cli_read_ready (unix_file_t * uf) { unix_main_t *um = &unix_main; unix_cli_main_t *cm = &unix_cli_main; unix_cli_file_t *cf; uword l; int n, n_read, n_try; cf = pool_elt_at_index (cm->cli_file_pool, uf->private_data); n = n_try = 4096; while (n == n_try) { l = vec_len (cf->input_vector); vec_resize (cf->input_vector, l + n_try); n = read (uf->file_descriptor, cf->input_vector + l, n_try); /* Error? */ if (n < 0 && errno != EAGAIN) return clib_error_return_unix (0, "read"); n_read = n < 0 ? 0 : n; _vec_len (cf->input_vector) = l + n_read; } if (!(n < 0)) vlib_process_signal_event (um->vlib_main, cf->process_node_index, (n_read == 0 ? UNIX_CLI_PROCESS_EVENT_QUIT : UNIX_CLI_PROCESS_EVENT_READ_READY), /* event data */ uf->private_data); return /* no error */ 0; } //epoll把需要回馈给用户的CLI输出，写入。 /** Called when a CLI session file descriptor can be written to without * blocking. */ static clib_error_t *unix_cli_write_ready (unix_file_t * uf) { unix_cli_main_t *cm = &unix_cli_main; unix_cli_file_t *cf; int n; cf = pool_elt_at_index (cm->cli_file_pool, uf->private_data); /* Flush output vector. */ n = write (uf->file_descriptor, cf->output_vector, vec_len (cf->output_vector)); if (n < 0 && errno != EAGAIN) return clib_error_return_unix (0, "write"); else if (n > 0) unix_cli_del_pending_output (uf, cf, n); return /* no error */ 0; }
注：本文是转载，但不是100%的转载，可能稍微有些出入，原文地址点击这里基本概念VPP支持多线程模式，其中区分为主线程和工作线程。主线程可以调用所有类型的node，工作线程只能调用。VLIB_NODE_TYPE_INTERNAL和VLIB_NODE_TYPE_INPUT类型的node。VPP多线程之间同步采用的是自旋锁，如果滥用vlib_worker_thread_barrier_sync和vlib_worker_thread_barrier_release会导致性能下降。博主认为如果使用RCU锁会有更好的性能。核心函数start_workers建立多线程框架。static clib_error_t *start_workers (vlib_main_t * vm) { /* * VLIB_REGISTER_THREAD会增加vlib_thread_main_t->registrations中数量。 * cpu_config可以配置占用核心位图或者指定核心个数，这将增加 * vlib_thread_registration_t->count计数。 * 这两个值将在vlib_thread_init用于计算vlib_thread_main_t->n_vlib_mains */ u32 n_vlib_mains = tm->n_vlib_mains; //调用该函数的是在主线程中 u8 *main_heap = clib_mem_get_per_cpu_heap (); mheap_t *main_heap_header = mheap_header (main_heap); ... //不开启工作线程时，n_vlib_mains为1，否则，每个工作线程都要单独的一份vlib_mains if (n_vlib_mains > 1) { //确保vlib_mains中能保存tm->n_vlib_mains个数据结构，可以产生内存重新分配 vec_validate (vlib_mains, tm->n_vlib_mains - 1); _vec_len (vlib_mains) = 0; //主线程的配置信息 vec_add1 (vlib_mains, vm); //博主发现根本没用到这个队列。无视之。 vec_validate (vlib_frame_queues, tm->n_vlib_mains - 1); _vec_len (vlib_frame_queues) = 0; fq = vlib_frame_queue_alloc (FRAME_QUEUE_NELTS); vec_add1 (vlib_frame_queues, fq); //如果wait_at_barrier为1，则工作线程主循环会自旋等待。 vlib_worker_threads->wait_at_barrier = clib_mem_alloc_aligned (sizeof (u32), CLIB_CACHE_LINE_BYTES); /* * 记录有多少工作线程进入自旋等待状态，只有所有工作线程都自旋等待了， * vlib_worker_thread_barrier_sync才能返回。 * 该机制滥用的话，性能会下降很快，希望以后能用rcu锁替代自旋锁 */ vlib_worker_threads->workers_at_barrier = clib_mem_alloc_aligned (sizeof (u32), CLIB_CACHE_LINE_BYTES); ... //tm->registrations就是调用VLIB_REGISTER_THREAD个数 for (i = 0; i < vec_len (tm->registrations); i++) { ... //配置工作线程个数为0的话，自然不用管了 if (tr->count == 0) continue; //工作线程个数要么指定core bitmap，要么指定个数自动分配核心 for (k = 0; k < tr->count; k++) { vec_add2 (vlib_worker_threads, w, 1); //博主没有找到mheap_size赋值的地方，应该都是0，一脸懵逼 if (tr->mheap_size) w->thread_mheap = mheap_alloc (0 /* use VM */ , tr->mheap_size); else w->thread_mheap = main_heap; ... //VLIB_REGISTER_THREAD处指定该值。"stats"工作线程这里为1，"workers"工作线程为0 if (tr->no_data_structure_clone) continue; ... //工作线程都是从主线程复制配置信息 vm_clone = clib_mem_alloc (sizeof (*vm_clone)); clib_memcpy (vm_clone, vlib_mains[0], sizeof (*vm_clone)); vm_clone->cpu_index = worker_thread_index; vm_clone->heap_base = w->thread_mheap; vm_clone->mbuf_alloc_list = 0; memset (&vm_clone->random_buffer, 0, sizeof (vm_clone->random_buffer)); nm = &vlib_mains[0]->node_main; nm_clone = &vm_clone->node_main; /* fork next frames array, preserving node runtime indices */ nm_clone->next_frames = vec_dup (nm->next_frames); for (j = 0; j < vec_len (nm_clone->next_frames); j++) { vlib_next_frame_t *nf = &nm_clone->next_frames[j]; u32 save_node_runtime_index; u32 save_flags; save_node_runtime_index = nf->node_runtime_index; save_flags = nf->flags & VLIB_FRAME_NO_FREE_AFTER_DISPATCH; vlib_next_frame_init (nf); nf->node_runtime_index = save_node_runtime_index; nf->flags = save_flags; } /* fork the frame dispatch queue */ nm_clone->pending_frames = 0; vec_validate (nm_clone->pending_frames, 10); /* $$$$$?????? */ _vec_len (nm_clone->pending_frames) = 0; /* fork nodes */ nm_clone->nodes = 0; for (j = 0; j < vec_len (nm->nodes); j++) { vlib_node_t *n; n = clib_mem_alloc_no_fail (sizeof (*n)); clib_memcpy (n, nm->nodes[j], sizeof (*n)); /* none of the copied nodes have enqueue rights given out */ n->owner_node_index = VLIB_INVALID_NODE_INDEX; memset (&n->stats_total, 0, sizeof (n->stats_total)); memset (&n->stats_last_clear, 0, sizeof (n->stats_last_clear)); vec_add1 (nm_clone->nodes, n); } nm_clone->nodes_by_type[VLIB_NODE_TYPE_INTERNAL] = vec_dup (nm->nodes_by_type[VLIB_NODE_TYPE_INTERNAL]); nm_clone->nodes_by_type[VLIB_NODE_TYPE_INPUT] = vec_dup (nm->nodes_by_type[VLIB_NODE_TYPE_INPUT]); vec_foreach (rt, nm_clone->nodes_by_type[VLIB_NODE_TYPE_INPUT]) rt->cpu_index = vm_clone->cpu_index; nm_clone->processes = vec_dup (nm->processes); /* zap the (per worker) frame freelists, etc */ nm_clone->frame_sizes = 0; nm_clone->frame_size_hash = 0; /* * vpp默认mheap_size为0，那么将使用main heap，那么多线程下会有锁开销。 * 懵逼，应该是不同核心不同heap才对。博主觉得这是个bug */ clib_mem_set_heap (oldheap); vec_add1 (vlib_mains, vm_clone); vm_clone->error_main.counters = vec_dup (vlib_mains[0]->error_main.counters); vm_clone->error_main.counters_last_clear = vec_dup (vlib_mains[0]->error_main.counters_last_clear); /* Fork the vlib_buffer_main_t free lists, etc. */ bm_clone = vec_dup (vm_clone->buffer_main); vm_clone->buffer_main = bm_clone; orig_freelist_pool = bm_clone->buffer_free_list_pool; bm_clone->buffer_free_list_pool = 0; /* *INDENT-OFF* */ pool_foreach (fl_orig, orig_freelist_pool, ({ pool_get_aligned (bm_clone->buffer_free_list_pool, fl_clone, CLIB_CACHE_LINE_BYTES); ASSERT (fl_orig - orig_freelist_pool == fl_clone - bm_clone->buffer_free_list_pool); fl_clone[0] = fl_orig[0]; fl_clone->aligned_buffers = 0; fl_clone->unaligned_buffers = 0; fl_clone->n_alloc = 0; })); /* *INDENT-ON* */ worker_thread_index++; } } } else { //"stats"工作线程依旧开启，"workers"工作线程没启用 for (i = 0; i < vec_len (tm->registrations); i++) { tr = tm->registrations[i]; for (j = 0; j < tr->count; j++) { vec_add2 (vlib_worker_threads, w, 1); if (tr->mheap_size) w->thread_mheap = mheap_alloc (0 /* use VM */ , tr->mheap_size); else w->thread_mheap = main_heap; w->thread_stack = vlib_thread_stacks[w - vlib_worker_threads]; w->thread_function = tr->function; w->thread_function_arg = w; w->instance_id = j; w->elog_track.name = (char *) format (0, "%s %d", tr->name, j + 1); w->registration = tr; vec_add1 (w->elog_track.name, 0); elog_track_register (&vm->elog_main, &w->elog_track); } } } worker_thread_index = 1; for (i = 0; i < vec_len (tm->registrations); i++) { int j; tr = tm->registrations[i]; //支持pthread或者dpdk线程启动。线程入口函数调用vlib_worker_thread_bootstrap_fn if (tr->use_pthreads || tm->use_pthreads) { for (j = 0; j < tr->count; j++) { w = vlib_worker_threads + worker_thread_index++; if (vlib_launch_thread (vlib_worker_thread_bootstrap_fn, w, 0) < 0) clib_warning ("Couldn't start '%s' pthread ", tr->name); } } else { uword c; /* *INDENT-OFF* */ clib_bitmap_foreach (c, tr->coremask, ({ w = vlib_worker_threads + worker_thread_index++; if (vlib_launch_thread (vlib_worker_thread_bootstrap_fn, w, c) < 0) clib_warning ("Couldn't start DPDK lcore %d", c); })); /* *INDENT-ON* */ } } //同步 vlib_worker_thread_barrier_sync (vm); vlib_worker_thread_barrier_release (vm); return 0; }//工作线程业务逻辑，使用单独分配的栈 void vlib_worker_thread_fn (void *arg) { vlib_worker_thread_t *w = (vlib_worker_thread_t *) arg; vlib_thread_main_t *tm = vlib_get_thread_main (); vlib_main_t *vm = vlib_get_main (); ASSERT (vm->cpu_index == os_get_cpu_number ()); //如果使用的是pthread，则里面有一个同步点 vlib_worker_thread_init (w); clib_time_init (&vm->clib_time); clib_mem_set_heap (w->thread_mheap); /* Wait until the dpdk init sequence is complete */ //主线程的dpdk_process中释放worker_thread_release while (tm->worker_thread_release == 0) vlib_worker_thread_barrier_check (); //业务主循环 vlib_worker_thread_internal (vm); }static_always_inline void vlib_worker_thread_internal (vlib_main_t * vm) { vlib_node_main_t *nm = &vm->node_main; u64 cpu_time_now = clib_cpu_time_now (); while (1) { /* * 每次循环都检查有没有同步请求，如果主线程vlib_worker_thread_barrier_sync调用， * 则所有工作线程都自旋等待时，主线程的vlib_worker_thread_barrier_sync返回，开始同步逻辑， * 此时所有工作线程都在自旋等待中，主线程可以操作共有数据。 * vlib_worker_thread_barrier_release调用后，工作线程退出自旋等待，继续主业务逻辑循环 */ vlib_worker_thread_barrier_check (); //博主没有找到enqeue操作，那么这个函数等于没用。 vlib_frame_queue_dequeue_internal (vm); //工作线程只处理VLIB_NODE_TYPE_INPUT和VLIB_NODE_TYPE_INTERNAL两种类型 vlib_node_runtime_t *n; vec_foreach (n, nm->nodes_by_type[VLIB_NODE_TYPE_INPUT]) { //cli会配置网卡与特定核心绑定，因此这里只会收取分配给本核心的网卡的数据包 cpu_time_now = dispatch_node (vm, n, VLIB_NODE_TYPE_INPUT, VLIB_NODE_STATE_POLLING, /* frame */ 0, cpu_time_now); } if (_vec_len (nm->pending_frames)) { int i; cpu_time_now = clib_cpu_time_now (); for (i = 0; i < _vec_len (nm->pending_frames); i++) { vlib_pending_frame_t *p; p = nm->pending_frames + i; cpu_time_now = dispatch_pending_node (vm, p, cpu_time_now); } _vec_len (nm->pending_frames) = 0; } vlib_increment_main_loop_counter (vm); /* Record time stamp in case there are no enabled nodes and above calls do not update time stamp. */ cpu_time_now = clib_cpu_time_now (); } }
注：本文是转载，但不是100%的转载，可能稍微有些出入，原文地址点击这里核心函数ethernet_input_init初始化函数，主循环之前会调用。static clib_error_t *ethernet_input_init (vlib_main_t * vm) { //支持vlan，和qinq协议 ethernet_main_t *em = &ethernet_main; __attribute__ ((unused)) vlan_table_t *invalid_vlan_table; __attribute__ ((unused)) qinq_table_t *invalid_qinq_table; //只是给format_buffer，unformat_buffer赋值， //值得注意的是有对packet generate初始化，基本协议都有自己的pg实现。 ethernet_setup_node (vm, ethernet_input_node.index); ethernet_setup_node (vm, ethernet_input_type_node.index); ethernet_setup_node (vm, ethernet_input_not_l2_node.index); //初始化sparse_vec，用于根据3层协议来区分下一跳node这个目的。 next_by_ethertype_init (&em->l3_next); ... }ethernet_input_inline完成了该node业务逻辑功能。static_always_inline uword ethernet_input_inline (vlib_main_t * vm, vlib_node_runtime_t * node, vlib_frame_t * from_frame, ethernet_input_variant_t variant) { u32 cpu_index = os_get_cpu_number (); /* * ETHERNET_INPUT_VARIANT_ETHERNET_TYPE， * ETHERNET_INPUT_VARIANT_NOT_L2， * ETHERNET_INPUT_VARIANT_ETHERNET三种模式下， * 公用ethernet_input_node的error信息。 * 博主没有看出这里有什么特殊的含义。 */ if (variant != ETHERNET_INPUT_VARIANT_ETHERNET) error_node = vlib_node_get_runtime (vm, ethernet_input_node.index); else error_node = node; //返回frame尾部保存数据包信息内存的起始地址 from = vlib_frame_vector_args (from_frame); //frame中的数据包个数 n_left_from = from_frame->n_vectors; //上次数据包的下一跳这里直接使用，后面有机会修正 next_index = node->cached_next_index; stats_sw_if_index = node->runtime_data[0]; while (n_left_from > 0) { //获取传给下一跳node的保存数据包的缓存 vlib_get_next_frame (vm, node, next_index, to_next, n_left_to_next); while (n_left_from >= 4 && n_left_to_next >= 2) { //操作两个数据包，再预取两个数据包，以下省略，因为下面操作一个数据包和这个方法一致 ... } while (n_left_from > 0 && n_left_to_next > 0) { //预取下一个报文 if (n_left_from > 1) { vlib_buffer_t *p2; p2 = vlib_get_buffer (vm, from[1]); vlib_prefetch_buffer_header (p2, STORE); CLIB_PREFETCH (p2->data, CLIB_CACHE_LINE_BYTES, LOAD); } bi0 = from[0]; to_next[0] = bi0; from += 1; to_next += 1; n_left_from -= 1; n_left_to_next -= 1; b0 = vlib_get_buffer (vm, bi0); error0 = ETHERNET_ERROR_NONE; //解析2层信息，有多重封装的也解封， //最终把vlib_buffer_t->current_data指向三层头部。 parse_header (variant, b0, &type0, &orig_type0, &outer_id0, &inner_id0, &match_flags0); old_sw_if_index0 = vnet_buffer (b0)->sw_if_index[VLIB_RX]; eth_vlan_table_lookups (em, vnm, old_sw_if_index0, orig_type0, outer_id0, inner_id0, &hi0, &main_intf0, &vlan_intf0, &qinq_intf0); identify_subint (hi0, b0, match_flags0, main_intf0, vlan_intf0, qinq_intf0, &new_sw_if_index0, &error0, &is_l20); // Save RX sw_if_index for later nodes vnet_buffer (b0)->sw_if_index[VLIB_RX] = error0 != ETHERNET_ERROR_NONE ? old_sw_if_index0 : new_sw_if_index0; /* * Increment subinterface stats * Note that interface-level counters have already been incremented * prior to calling this function. Thus only subinterface counters * are incremented here. * Interface level counters include packets received on the main * interface and all subinterfaces. Subinterface level counters * include only those packets received on that subinterface * Increment stats if the subint is valid and it is not the main intf * 更新统计信息，vpp中大量代码都是先按照预测执行逻辑，随后再修正， * 或许对代码流水线有帮助，有空再仔细琢磨下。 */ if ((new_sw_if_index0 != ~0) && (new_sw_if_index0 != old_sw_if_index0)) { len0 = vlib_buffer_length_in_chain (vm, b0) + b0->current_data - vnet_buffer (b0)->ethernet.start_of_ethernet_header; stats_n_packets += 1; stats_n_bytes += len0; // Batch stat increments from the same subinterface so counters // don't need to be incremented for every packet. if (PREDICT_FALSE (new_sw_if_index0 != stats_sw_if_index)) { stats_n_packets -= 1; stats_n_bytes -= len0; if (new_sw_if_index0 != ~0) vlib_increment_combined_counter (vnm->interface_main.combined_sw_if_counters + VNET_INTERFACE_COUNTER_RX, cpu_index, new_sw_if_index0, 1, len0); if (stats_n_packets > 0) { vlib_increment_combined_counter (vnm->interface_main.combined_sw_if_counters + VNET_INTERFACE_COUNTER_RX, cpu_index, stats_sw_if_index, stats_n_packets, stats_n_bytes); stats_n_packets = stats_n_bytes = 0; } stats_sw_if_index = new_sw_if_index0; } } if (variant == ETHERNET_INPUT_VARIANT_NOT_L2) is_l20 = 0; //决定下一跳node，根据设置可以支持按照协议决定下一跳 determine_next_node (em, variant, is_l20, type0, b0, &error0, &next0); b0->error = error_node->errors[error0]; // verify speculative enqueue //修正这两个数据包下一跳node vlib_validate_buffer_enqueue_x1 (vm, node, next_index, to_next, n_left_to_next, bi0, next0); } vlib_put_next_frame (vm, node, next_index, n_left_to_next); } // Increment any remaining batched stats if (stats_n_packets > 0) { vlib_increment_combined_counter (vnm->interface_main.combined_sw_if_counters + VNET_INTERFACE_COUNTER_RX, cpu_index, stats_sw_if_index, stats_n_packets, stats_n_bytes); node->runtime_data[0] = stats_sw_if_index; } return from_frame->n_vectors; }以下几个函数用于hook该node，替代determine_next_node中默认的下一跳node挑选机制。默认有ETHERNET_TYPE_IP4，ETHERNET_TYPE_IP6，ETHERNET_TYPE_MPLS_UNICAST三种协议，用户可以自己添加更多协议。可以根据协议来做不同下一跳。void ethernet_register_input_type (vlib_main_t * vm, ethernet_type_t type, u32 node_index) { ethernet_main_t *em = &ethernet_main; ethernet_type_info_t *ti; u32 i; ti = ethernet_get_type_info (em, type); ti->node_index = node_index; ti->next_index = vlib_node_add_next (vm, ethernet_input_node.index, node_index); i = vlib_node_add_next (vm, ethernet_input_type_node.index, node_index); ASSERT (i == ti->next_index); i = vlib_node_add_next (vm, ethernet_input_not_l2_node.index, node_index); ASSERT (i == ti->next_index); // Add the L3 node for this ethertype to the next nodes structure next_by_ethertype_register (&em->l3_next, type, ti->next_index); // Call the registration functions for other nodes that want a mapping l2bvi_register_input_type (vm, type, node_index); }vlan包下一跳判定机制，可以还原ethernet头部，根据这里注册的值作跳转。void ethernet_register_l2_input (vlib_main_t * vm, u32 node_index) { ethernet_main_t *em = &ethernet_main; u32 i; em->l2_next = vlib_node_add_next (vm, ethernet_input_node.index, node_index); // Even if we never use these arcs, we have to align the next indices... i = vlib_node_add_next (vm, ethernet_input_type_node.index, node_index); ASSERT (i == em->l2_next); i = vlib_node_add_next (vm, ethernet_input_not_l2_node.index, node_index); ASSERT (i == em->l2_next); }调用该函数后，大多数下一跳基本就由这里注册的值决定了。// Register a next node for L3 redirect, and enable L3 redirect void ethernet_register_l3_redirect (vlib_main_t * vm, u32 node_index) { ethernet_main_t *em = &ethernet_main; u32 i; em->redirect_l3 = 1; em->redirect_l3_next = vlib_node_add_next (vm, ethernet_input_node.index, node_index); //Change the cached next nodes to the redirect node em->l3_next.input_next_ip4 = em->redirect_l3_next; em->l3_next.input_next_ip6 = em->redirect_l3_next; em->l3_next.input_next_mpls = em->redirect_l3_next; // Even if we never use these arcs, we have to align the next indices... i = vlib_node_add_next (vm, ethernet_input_type_node.index, node_index); ASSERT (i == em->redirect_l3_next); }
注：本文是转载，但不是100%的转载，可能稍微有些出入，原文地址点击这里VPP使用者几乎都会使用dpdk node作为收包驱动，本文将分析其源码。基本概念vlib_buffer_tdpdk收到的数据包用rte_mbuf结构描述。vpp为了兼容其它收包node（netmap，pcap等）改为使用vlib_buffer_t来描述数据包。vlib_buffer_t紧跟在rte_mbuf后面，headroom空间中。vlib_buffer_from_rte_mbuf和rte_mbuf_from_vlib_buffer完成相互转化。值得注意的是，该结构使用了三个CLIB_CACHE_LINE_ALIGN_MARK宏，把结构分成了3段，确保每段都是CLIB_CACHE_LINE_BYTES对齐。作为一个被频繁使用的结构，这样设计可以更好的更具需要进行内存预取。vlib_get_next_frame，vlib_put_next_frame几乎每个node中必定出现的一对好基友。vlib_get_next_frame获取传递给下一个node的数据包将驻留的内存结构。vlib_put_next_frame把传递给下一个node的数据包写入特定位置。这样下一个node将正式可以被调度框架调度，并处理传给他的数据包.EFDearly-fast-discard没有找到相关资料，估计也是一个性能优化特性，之后再补个人对作者意图的猜测。核心函数dpdk_device_input//use_efd指定是否启用early-fast-discard特性。 static inline u32 dpdk_device_input (dpdk_main_t * dm, dpdk_device_t * xd, vlib_node_runtime_t * node, u32 cpu_index, u16 queue_id, int use_efd) { u32 n_buffers; /* * 缺省情况下一跳是ethernet node，之后可以最终改变下一条， * 博主认为这里使用next_index = node->cached_next_index会更合理 */ u32 next_index = DPDK_RX_NEXT_ETHERNET_INPUT; ... //网卡必须up了 if (xd->admin_up == 0) return 0; //指定网卡的指定队列上收一批包，返回收到的包个数， //数据包地址存在xd->rx_vectors[queue_id]中。 n_buffers = dpdk_rx_burst (dm, xd, queue_id); if (n_buffers == 0) { //看来EFD是默认不使用EFD。 if (PREDICT_FALSE (use_efd && dm->efd.enabled)) { xd->efd_agent.last_poll_time = 0; xd->efd_agent.last_burst_sz = 0; } return 0; } /* * vlib_buffer有多个pool，可以用在自己构建的数据包， * 这里的fl仅仅用来作为模板初始化ret_mbuf后面的vlib_buffer结构。 */ fl = vlib_buffer_get_free_list (vm, VLIB_BUFFER_DEFAULT_FREE_LIST_INDEX); if (PREDICT_FALSE (use_efd && dm->efd.enabled)) { //一次收包到了VLIB_FRAME_SIZE个，如果这种情况次数很多， //说明发生了收包端拥塞，可以在后面直接丢弃数据包。 if (PREDICT_FALSE (xd->efd_agent.consec_full_frames_cnt >= dm->efd.consec_full_frames_hi_thresh)) { u32 device_queue_sz = rte_eth_rx_queue_count (xd->device_index, queue_id); if (device_queue_sz >= dm->efd.queue_hi_thresh) { /* dpdk device queue has reached the critical threshold */ xd->efd_agent.congestion_cnt++; /* apply EFD to packets from the burst */ efd_discard_burst = 1; } } } mb_index = 0; while (n_buffers > 0) { //to_next指向的内存用来保存数据包地址，n_left_to_next保存数据包地址的内存最大剩余空间 vlib_get_next_frame (vm, node, next_index, to_next, n_left_to_next); while (n_buffers > 0 && n_left_to_next > 0) { if (PREDICT_TRUE (n_buffers > 2)) { /* * 把之后需要使用的数据包内存预取一下。 * 为什么预取2个包以后的而不是下一个包？疑惑。 * 内存预取所花费的cpu周期,博主有时间在定量分析下。 */ struct rte_mbuf *pfmb = xd->rx_vectors[queue_id][mb_index + 2]; vlib_buffer_t *bp = vlib_buffer_from_rte_mbuf (pfmb); CLIB_PREFETCH (pfmb, CLIB_CACHE_LINE_BYTES, STORE); CLIB_PREFETCH (bp, CLIB_CACHE_LINE_BYTES, STORE); } //拥塞发生了，丢包 if (PREDICT_FALSE (efd_discard_burst)) { vlib_thread_main_t *tm = vlib_get_thread_main (); if (PREDICT_TRUE (cntr_type = is_efd_discardable (tm, b0, mb))) { rte_pktmbuf_free (mb); xd->efd_agent.discard_cnt++; increment_efd_drop_counter (vm, cntr_type, 1); n_buffers--; mb_index++; continue; } } if (PREDICT_FALSE (mb->nb_segs > 1)) { //又是内存预取 struct rte_mbuf *pfmb = mb->next; vlib_buffer_t *bp = vlib_buffer_from_rte_mbuf (pfmb); CLIB_PREFETCH (pfmb, CLIB_CACHE_LINE_BYTES, LOAD); CLIB_PREFETCH (bp, CLIB_CACHE_LINE_BYTES, STORE); b_chain = b0; } //取用fl中的模板给b0初始化,这个函数在赋值时， //如果cpu支持128bit向量，则可以每次赋值拷贝16个字节(u8x16类型支持)。 vlib_buffer_init_for_free_list (b0, fl); //vlib_buffer转换为偏移量，注意地址都是1 << CLIB_LOG2_CACHE_LINE_BYTES对齐 bi0 = vlib_get_buffer_index (vm, b0); //数据包写入缓存，下一跳node将会使用 to_next[0] = bi0; to_next++; n_left_to_next--; //决定下一跳node，默认情况下会根据3层协议来决定下一跳。 //如果自己hook了，则条转给指定的node dpdk_rx_next_and_error_from_mb_flags_x1 (xd, mb, b0, &next0, &error0); ... //博主认为这里是个bug， //应该是b0->current_data += mb->data_off - sizeof(struct vlib_buffer_t) b0->current_data += mb->data_off - RTE_PKTMBUF_HEADROOM; b0->current_length = mb->data_len - l3_offset0; ... while ((mb->nb_segs > 1) && (nb_seg < mb->nb_segs)) { ... } vlib_validate_buffer_enqueue_x1 (vm, node, next_index, to_next, n_left_to_next, bi0, next0); n_buffers--; mb_index++; } //调度框架之后就可以调度下一跳node处理上文传入的数据包集合了。 vlib_put_next_frame (vm, node, next_index, n_left_to_next); } ... //多线程模型下，可以使用多个dpdk线程收包。 dpdk_worker_t *dw = vec_elt_at_index (dm->workers, cpu_index); dw->aggregate_rx_packets += mb_index; return mb_index; }//观察这个注册函数，可见它有5个下一跳。 VLIB_REGISTER_NODE (dpdk_input_node) = { .function = dpdk_input, .type = VLIB_NODE_TYPE_INPUT, .name = "dpdk-input", /* Will be enabled if/when hardware is detected. */ .state = VLIB_NODE_STATE_DISABLED, .format_buffer = format_ethernet_header_with_length, .format_trace = format_dpdk_rx_dma_trace, .n_errors = DPDK_N_ERROR, .error_strings = dpdk_error_strings, .n_next_nodes = DPDK_RX_N_NEXT, .next_nodes = { [DPDK_RX_NEXT_DROP] = "error-drop", [DPDK_RX_NEXT_ETHERNET_INPUT] = "ethernet-input", [DPDK_RX_NEXT_IP4_INPUT] = "ip4-input-no-checksum", [DPDK_RX_NEXT_IP6_INPUT] = "ip6-input", [DPDK_RX_NEXT_MPLS_INPUT] = "mpls-gre-input", }, };
注：本文是转载，但不是100%的转载，可能稍微有些出入，原文地址点击这里vec变长数组len是数组元素个数，不是字节长度。每个数组元素都看作是等大小。自定义头部后接vec 组合使用。组合vec将引申出其他内存管理结构。_vec_find(v)v指向数据部分，宏返回vec头部地址。_vec_round_size(s)s按sizeof (uword)大小，向上对齐。vec_header_bytes (uword header_bytes)自定义头部+vec 组合这种模型下，header_bytes代表自定义头部大小。函数返回自定义头部+vec总长度，按sizeof (vec_header_t)向上对齐。vec_header (void *v, uword header_bytes)自定义头部+vec 组合，按sizeof (vec_header_t)向上对齐这种模型下，v指向数据部分地址，header_bytes代表自定义头部大小，返回自定义头部地址。vec_header_end (void *v, uword header_bytes)自定义头部+vec 组合，按sizeof (vec_header_t)向上对齐这种模型下，v指向自定义头部地址，header_bytes代表自定义头部大小，返回数据部分地址。vec_aligned_header (void *v, uword header_bytes, uword align)自定义头部+vec 组合，按align向上对齐这种模型下，v指向数据部分地址，header_bytes代表自定义头部大小，返回自定义头部地址。vec_aligned_header_end (void *v, uword header_bytes, uword align)自定义头部+vec 组合，按align向上对齐这种模型下，v指向自定义头部地址，header_bytes代表自定义头部大小，返回数据部分地址。_vec_len(v)v指向数据部分，返回vec头部的vec_header_t->len。vec_len(v)v指向数据部分，v等于null时返回0，否则返回vec头部的vec_header_t->len。vec_reset_length(v)v指向数据部分，vec头部的vec_header_t->len置0。vec_bytes(v)v指向数据部分，返回数据部分总字节大小。vec_capacity(v,b)v指向data起始地址，b代表user_header大小，返回该mheap_elt_t字节大小，即vec的最大可占用内存。博主觉得该函数放在这里破坏了该头件的独立性。vec_max_len(v)类似vec_capacity(v,b)内存布局，v指向data起始地址，返回该vec可容纳最大数据块数目。vec_end(v)v指向data起始地址，返回vec数据部分末尾的下一个字节。vec_is_member(v,e)v指向data起始地址，返回e是否在该vec地址范围内。vec_elt_at_index(v,i)v指向data起始地址，返回第i个数据块的地址。vec_elt(v,i)v指向data起始地址，返回第i个数据块的内容。vec_foreach(var,vec) vec_foreach_backwards(var,vec) vec_foreach_index(var,v)迭代宏mheapmheap整体视图注意每个elt大小不一定相等。这里的vec_header_t把后面数据看成一个字节一个字节的元素，因此vec->len实际是整个数据部分字节长度。mheap_elt_t视图基本函数操作在mheap_bootstrap.h中。该头文件基本函数，结合上文内容很容易理解，这里不再详细叙述。mheap.c中会对上图基本结构更近一步索引管理，也是mheap的核心部分。mheap_maybe_lock (void *v) mheap_maybe_unlock (void *v)自旋锁加锁解锁函数，如果是持有锁的本CPU再次进入，可以无条件获取锁。mheap_get_aligned (void v, uword n_user_data_bytes, uword align, uword align_offset, uword offset_return)mheap内存分配接口函数。void *mheap_get_aligned (void *v, uword n_user_data_bytes, uword align, uword align_offset, uword * offset_return) { ... //align至少是mheap_elt_t大小，并且是2的幂 align = clib_max (align, STRUCT_SIZE_OF (mheap_elt_t, user_data[0])); align = max_pow2 (align); ... //各种对齐，确保mheap_elt_t头部和数据部分都是STRUCT_SIZE_OF (mheap_elt_t, user_data[0]) //之后再用专门章节详细分析数据结构对齐背后的作者设计目的 n_user_data_bytes = clib_max (n_user_data_bytes, MHEAP_MIN_USER_DATA_BYTES); n_user_data_bytes = round_pow2 (n_user_data_bytes, STRUCT_SIZE_OF (mheap_elt_t, user_data[0])); if (!v) v = mheap_alloc (0, 64 << 20);//分配完整的一个mheap结构 mheap_maybe_lock (v); h = mheap_header (v); if (h->flags & MHEAP_FLAG_VALIDATE) mheap_validate (v);//遍历mheap的空闲块链表，验证是否正确链接。 /* First search free lists for object. */ //从空闲块链表中获取数据块 offset = mheap_get_search_free_list (v, &n_user_data_bytes, align, align_offset); h = mheap_header (v); /* If that fails allocate object at end of heap by extending vector. */ if (offset == MHEAP_GROUNDED && _vec_len (v) < h->max_size) { //如果现有空闲块没有满足要求的块，则从mheap中新分配块 //注意h->max_size代表该mheap最大内存量，_vec_len (v)代表目前使用量 v = mheap_get_extend_vector (v, n_user_data_bytes, align, align_offset, &offset); //mheap_get_extend_vector()并不会导致mheap重新分配，博主认为下面重新计算h毫无必要 h = mheap_header (v); //如果扩充成功则统计值加1 h->stats.n_vector_expands += offset != MHEAP_GROUNDED; } ... }void *mheap_alloc (void *memory, uword size) { ... //这里指的是利用cpu的SIMD向量指令加快缓存块的cache查找，之后会详细介绍 #ifdef CLIB_HAVE_VEC128 flags |= MHEAP_FLAG_SMALL_OBJECT_CACHE; #endif return mheap_alloc_with_flags (memory, size, flags); }void *mheap_alloc_with_flags (void *memory, uword memory_size, uword flags) { ... if (!memory) { //通常用的是UNIX模式，该分配函数将mmap()匿名内存使用 memory = clib_mem_vm_alloc (memory_size); ... } //把memory向上对mheap_page_size 大小对齐 am = pointer_to_uword (memory); av = mheap_page_round (am); //计算对齐后的mheap头部地址 v = uword_to_pointer (av, void *); h = mheap_header (v); ah = pointer_to_uword (h); //mheap头部地址小于memory起始地址，则增加一个mheap_page_size。 //这样mheap头部地址既在mmap()分配的内存范围中，也满足对齐要求。 while (ah < am) ah += mheap_page_size; h = uword_to_pointer (ah, void *); v = mheap_vector (h); //size代表mheap数据部分大小 size = memory + memory_size - v; ... //虽然不解除映射也不会影响性能。但是这样好处是一旦访问了未分配的内存会导致段错误，提示开发人员编码错误 if (!(h->flags & MHEAP_FLAG_DISABLE_VM)) mheap_vm (v, MHEAP_VM_UNMAP | MHEAP_VM_ROUND_UP, (clib_address_t) v, h->max_size); //MHEAP_GROUNDED代表该bin下没有空闲数据块，值为~0。所以这里用0xFF初始化。 memset (h->first_free_elt_uoffset_by_bin, 0xFF, sizeof (h->first_free_elt_uoffset_by_bin)); return v; }static uword mheap_get_search_free_list (void *v, uword * n_user_bytes_arg, uword align, uword align_offset) { ... //计算bin值，用来索引对应大小空闲块链表 bin = user_data_size_to_bin_index (n_user_bytes); //如果请求的是满足条件的块，则从一个额外的cache中获取 //避免了搜索空闲块链表开销，还可以利用CPU的SIMD向量指令来加速匹配。 if (MHEAP_HAVE_SMALL_OBJECT_CACHE && (h->flags & MHEAP_FLAG_SMALL_OBJECT_CACHE) && bin < 255 && align == STRUCT_SIZE_OF (mheap_elt_t, user_data[0]) && align_offset == 0) { //cache中寻找数据块，之后会详细分析 uword r = mheap_get_small_object (h, bin); h->stats.n_small_object_cache_attempts += 1; if (r != MHEAP_GROUNDED) { h->stats.n_small_object_cache_hits += 1; return r; } } //h->non_empty_free_elt_heads是一个位图，记录了对应bin链表是否为空 for (i = bin / BITS (uword); i < ARRAY_LEN (h->non_empty_free_elt_heads); i++) { uword non_empty_bin_mask = h->non_empty_free_elt_heads[i]; //把比bin小的位置的位图略掉 if (i == bin / BITS (uword)) non_empty_bin_mask &= ~pow2_mask (bin % BITS (uword)); //foreach_set_bit()从右往左遍历每个bit，为1的bit调用mheap_get_search_free_bin() //foreach_set_bit()中first_set (uword x)用来计算从右算第一个为1的bit代表的数 foreach_set_bit (bi, non_empty_bin_mask, ( { uword r = mheap_get_search_free_bin (v, bi + i * BITS(uword), n_user_bytes_arg, align, align_offset); if (r != MHEAP_GROUNDED) return r; } )); } return MHEAP_GROUNDED; }/* * 如果每次都是计算bin值再从链表中找空闲块就太麻烦了 * 对满足特定条件块，也应该是vpp使用频率最多的那种类型的块做了个cache * 只需要计算bin值就能直接得到可用空闲块。该函数就是干这事。 */ always_inline uword mheap_get_small_object (mheap_t * h, uword bin) { /* * cache中c->bins.as_u8数组每个元素记录了(bin+1) * 为0代表该位置没有空闲块。c->offsets记录对应bin的空闲块。 * mheap_small_object_cache_t *c = &h->small_object_cache; * 查找c->bins.as_u8数组中，所有值为(bin+1)的索引组成的位图 */ uword mask = mheap_small_object_cache_mask (c, bin + 1); uword offset = MHEAP_GROUNDED; //如果位图不为0，说明有空闲块，查找位图最左边那个为1bit的索引， //返回对应的c->offsets值，即可用空闲块。 if (mask) { uword i = min_log2 (mask); uword o = c->offsets[i]; ASSERT (o != MHEAP_GROUNDED); c->bins.as_u8[i] = 0; offset = o; } return offset; }always_inline uword mheap_small_object_cache_mask ( mheap_small_object_cache_t * c, uword bin) { uword mask; //本函数依赖于CPU支持向量指令。 #if !defined (CLIB_HAVE_VEC128) || defined (__ALTIVEC__) mask = 0; #else //返回一个由16字节组成的向量，每个字节的值都是bin u8x16 b = u8x16_splat (bin); ASSERT (bin < 256); /* * u8x16_is_equal()比较两个128bit位向量，每个字节进行对比， * 如果相等，返回字节为1，一共返回16个字节，组成一个返回值128bit向量。 * u8x16_compare_byte_mask()把返回的128bit向量转化成位图，共16个bit位图 */ #define _(i) ((uword) u8x16_compare_byte_mask ( u8x16_is_equal (b, c->bins.as_u8x16[i])) << (uword) ((i)*16)) mask = _(0) | _(1); if (BITS (uword) > 32) mask |= _(2) | _(3); #undef _ #endif //32位系统返回的是32bit位图，64位系统返回64bit位图 return mask; } /* * 该函数搜索bin对应的空闲块链表。找到一个大小满足要求的空闲块是很容易的， * 本函数额外大量代码花在了满足空闲块的对齐要求上。 * align:首地址相对于heap偏移对齐，align_offset:首地址对齐后，再向左做个偏移。 * 分配内存时从对应的slot中查找，值得注意的是查找算法为了满足参数中对齐要求， * 很有可能从空闲块中间部分分配内存，这样首尾部分空闲块重新链入新的slot。 * 一切的一切都是为了数据对齐。如果分配的内存是一个带头部的数据结构，例如vec+data类型。 * 那么分配内存时需要把头部大小作为align_offset参数，使数据部分满足起始地址align对齐。 */static uword mheap_get_search_free_bin (void *v, uword bin, uword * n_user_data_bytes_arg, uword align, uword align_offset) { ... //o0,f0是用于首部，f0，f1用于尾部，lo_free_usize切分后首部空间，hi_free_usize切分后尾部空间 word o0, o1, f0, f1, search_n_user_data_bytes; word lo_free_usize, hi_free_usize; ... /* Find an object that is large enough with correct alignment at given alignment offset. */ while (1) { ... //f0指向原始块首部地址偏移，f1指向原始块尾部地址偏移 f0 = ((void *) e->user_data - v); f1 = f0 + this_object_n_user_data_bytes; //从尾部往左分配目标块，o0指向首部偏移并且满足对齐要求 o0 = ((f1 - search_n_user_data_bytes) & ~(align - 1)) - align_offset; //o0比f0还小了，右偏移一个align while (o0 < f0) o0 += align; //确保切分后首部空闲块至少有MHEAP_ELT_OVERHEAD_BYTES大小，这是要给切去的块做头部的 while (1) { lo_free_usize = o0 != f0 ? o0 - f0 - MHEAP_ELT_OVERHEAD_BYTES : 0; if (o0 <= f0 || lo_free_usize >= (word) MHEAP_MIN_USER_DATA_BYTES) break; //越偏移lo_free_usize越小，什么鬼，博主强烈认为这是bug，应该是+= o0 -= align; } //o0~o1之间就是我们要找的内存了，大小，对齐性都满足 o1 = o0 + search_n_user_data_bytes; ... } found: ... //从中间切开得到我们要的内存，尾部如果太小， //就不单独做成空闲块了，直接给我们用。 if (hi_free_usize < (word) MHEAP_MIN_USER_DATA_BYTES) { search_n_user_data_bytes += f1 - o1; o1 = f1; hi_free_usize = 0; } //切去的内存mmap下 if (!(h->flags & MHEAP_FLAG_DISABLE_VM)) { ... } ... //切分后首部块重新链入bin表 if (lo_free_usize > 0) { ... } mheap_elt_set_size (v, o0, search_n_user_data_bytes, /* is_free */ 0); //切分后尾部块重新链入bin表 if (hi_free_usize > 0) { ... } //分配的elt的数据部分大小 *n_user_data_bytes_arg = search_n_user_data_bytes; h->stats.free_list.n_objects_found += 1; //返回分配的elt的数据部分偏移值 return o0; }user_data_size_to_bin_index (uword n_user_data_bytes)空闲mheap_elt_t按照数据部分大小分类链接到不同slot中。该函数计算bin值。always_inline uword user_data_size_to_bin_index (uword n_user_data_bytes) { ... //n_user_data_bytes至少要能包含mheap_elt_t头部 n_user_data_bytes = clib_max (n_user_data_bytes, MHEAP_MIN_USER_DATA_BYTES); ... //除去头部的纯数据部分长度，换算成占用word个数，即为bin值 small_bin = n_user_data_words - (MHEAP_MIN_USER_DATA_BYTES / MHEAP_USER_DATA_WORD_BYTES); large_bin = MHEAP_N_SMALL_OBJECT_BINS + max_log2 (n_user_data_bytes) - MHEAP_LOG2_N_SMALL_OBJECT_BINS; //1 ~ MHEAP_N_SMALL_OBJECT_BINS每个值对应单独bin //>= MHEAP_N_SMALL_OBJECT_BINS多个值映射到一个bin中。 //可见，分配内存越小，越能精准快速定位到可选空闲块。 return small_bin < MHEAP_N_SMALL_OBJECT_BINS ? small_bin : large_bin; }vector实际使用的缓存一般是如下结构：user_header指具体数据结构头部，比如：mhash_t，pool_header_t等等。