VPP的dpo机制跟路由紧密结合在一起。路由表查找（ip4_lookup）的最后结果是一个load_balance_t结构。该结构可以看做是一个hash表，里面包含了很多dpo，指向为下一步处理动作。每个dpo都是添加路由时的一个path的结果。 dpo标准类型有： DPO_DROP, DPO_IP_NULL, DPO_PUNT, DPO_LOAD_BALANCE, DPO_ADJACENCY, DPO_ADJACENCY_INCOMPLETE, DPO_ADJACENCY_MIDCHAIN, DPO_ADJACENCY_GLEAN, DPO_RECEIVE, DPO_LOOKUP, DPO_LISP_CP, DPO_CLASSIFY, DPO_MPLS_LABEL, DPO_LAST, 可以理解为每个类型都可以有自己的私有数据（可能有也可能没有），它们都继承自标准的dpo_id_t结构。 比如DPO_LOAD_BALANCE有自己的私有数据结构：load_balance_t。可以通过dpo_id_t中的dpoi_index来索引到具体的实例。 load_balance_t比较特殊，既是路由的查找最终结果，也是一个dpo。好绕。在lb插件中利用了该特性，处理完自己的修改数据包的逻辑后，又把数据包丢给了load_balance_t，让修改后的数据包能正确的进行下一步处理。
typedef struct dpo_id_t_ { /** * the type */ //dpo的类型，可以是上文标准类型之一，也可以是自定义的 dpo_type_t dpoi_type; /** * the data-path PRotocol of the type. */ dpo_proto_t dpoi_proto; /** * The next VLIB node to follow. */ //下一跳，使用该dpo的node的下一跳slot的索引值 u16 dpoi_next_node; /** * the index of objects of that type */ //本dpo对应的私有数据在其内存池中的索引号 index_t dpoi_index;} __attribute__ ((aligned(sizeof(u64)))) dpo_id_t;DPO_DROP 将数据包送往”XXX-drop” node。简单处理后再传给”error-drop” node，完成最后数据包丢弃回收工作。 DPO_IP_NULL 将数据包送往”ipx-null” node。该node将决定是否针对数据包回icmp不可达包或者icmp禁止包。 DPO_PUNT
这两个函数并不是字面意义上的加锁/解锁操作。上文提到，某些dpo类型有自己的私有数据，这对函数即是增加私有数据结构的引用计数用，如果dpo类型没有自己的私有数据，则这对函数为空。其内部具体实现，调用的是dpo类型注册时提供的函数指针。
void dpo_lock(dpo_id_t *dpo);void dpo_unlock(dpo_id_t *dpo);dpo的设置操作
voiddpo_set (dpo_id_t *dpo, dpo_type_t type, dpo_proto_t proto, index_t index){ //原有的dpo可能含有私有数据，当原有dpo被覆盖时，原有的私有数据需要减少引用计数。 dpo_id_t tmp = *dpo; dpo->dpoi_type = type; dpo->dpoi_proto = proto, dpo->dpoi_index = index; //邻接表类型的dpo有点特殊，邻接表类型的dpo之后会详细分析 if (DPO_ADJACENCY == type) { /* * set the adj subtype */ ip_adjacency_t *adj; adj = adj_get(index); switch (adj->lookup_next_index) { case IP_LOOKUP_NEXT_ARP: dpo->dpoi_type = DPO_ADJACENCY_INCOMPLETE; break; case IP_LOOKUP_NEXT_MIDCHAIN: dpo->dpoi_type = DPO_ADJACENCY_MIDCHAIN; break; default: break; } } //增加引用计数 dpo_lock(dpo); //减少引用计数 dpo_unlock(&tmp);}假设数据包到了dpo中(child)，处理完后我希望数据包交给下一个dpo(parent)来处理，那么该函数就派上用场了。本dpo对应的node中增加一个slot，指向下一个dpo(parent)对应的node。slot的索引保存在本dpo(child)的dpoi_next_node中。
voiddpo_stack (dpo_type_t child_type, dpo_proto_t child_proto, dpo_id_t *dpo, const dpo_id_t *parent){ dpo_stack_i(dpo_get_next_node(child_type, child_proto, parent), dpo, parent);}dpo_edges是一个四重指针，看起来很绕很复杂，它其实就是一个缓存功能，记录了dpo(child)对应的node的指向下一跳dpo(parent)对应node的slot索引号。如果没有则新建一个。
static u32dpo_get_next_node (dpo_type_t child_type, dpo_proto_t child_proto, const dpo_id_t *parent_dpo){ dpo_proto_t parent_proto; dpo_type_t parent_type; parent_type = parent_dpo->dpoi_type; parent_proto = parent_dpo->dpoi_proto; vec_validate(dpo_edges, child_type); vec_validate(dpo_edges[child_type], child_proto); vec_validate(dpo_edges[child_type][child_proto], parent_type); vec_validate_init_empty( dpo_edges[child_type][child_proto][parent_type], parent_proto, ~0); /* * if the edge index has not yet been created for this node to node transistion */ if (~0 == dpo_edges[child_type][child_proto][parent_type][parent_proto]) { vlib_node_t *parent_node, *child_node; vlib_main_t *vm; u32 edge ,pp, cc; vm = vlib_get_main(); ASSERT(NULL != dpo_nodes[child_type]); ASSERT(NULL != dpo_nodes[child_type][child_proto]); ASSERT(NULL != dpo_nodes[parent_type]); ASSERT(NULL != dpo_nodes[parent_type][parent_proto]); cc = 0; /* * create a graph arc from each of the parent's registered node types, * to each of the childs. */ while (NULL != dpo_nodes[child_type][child_proto][cc]) { child_node = vlib_get_node_by_name(vm, (u8*) dpo_nodes[child_type][child_proto][cc]); pp = 0; while (NULL != dpo_nodes[parent_type][parent_proto][pp]) { parent_node = vlib_get_node_by_name(vm, (u8*) dpo_nodes[parent_type][parent_proto][pp]); edge = vlib_node_add_next(vm, child_node->index, parent_node->index); if (~0 == dpo_edges[child_type][child_proto][parent_type][parent_proto]) { dpo_edges[child_type][child_proto][parent_type][parent_proto] = edge; } else { ASSERT(dpo_edges[child_type][child_proto][parent_type][parent_proto] == edge); } pp++; } cc++; } } return (dpo_edges[child_type][child_proto][parent_type][parent_proto]);}新闻热点
疑难解答
