思科VPP源码分析（dpo机制源码分析）

思科VPP源码分析（dpo机制源码分析）"/>

思科VPP源码分析（dpo机制源码分析）

基本概念

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的设置操作

void
dpo_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中。

void
dpo_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 u32
dpo_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]);
}