ff_run详解

作者，风眠-广坤

ff_run(loop_func_t loop, void* arg)

loop为入口函数
arg为user pointer

    void ff_run(loop_func_t loop, void* arg){ff_dpdk_run(loop, arg);}

ff_dpdk_run函数

    void ff_dpdk_run(loop_func_t loop, void *arg){/***保存回调指针和上下文*/struct loop_routine *lr = rte_malloc(NULL,sizeof(struct loop_routine), 0);lr->loop = loop;lr->arg = arg;/*** launch a function on dpdk lcore* 内部会回调用户传入loop函数指针*/rte_eal_mp_remote_launch(main_loop, lr, CALL_MASTER);rte_eal_mp_wait_lcore();rte_free(lr);}

main_loop函数

dpdk lcore上实际入口
功能
- 定时器管理
- pcap处理
- 刷新tx缓冲
- rx数据包
  - 通过dispatch_ring传递过来的数据包
  - 当前queue接收的数据包
- msg_ring消息处理
- 用户loop回调

    static int main_loop(void *arg){struct loop_routine *lr = (struct loop_routine *)arg;struct rte_mbuf *pkts_burst[MAX_PKT_BURST];uint64_t prev_tsc, diff_tsc, cur_tsc, usch_tsc, div_tsc, usr_tsc, sys_tsc, end_tsc, idle_sleep_tsc;int i, j, nb_rx, idle;uint16_t port_id, queue_id;struct lcore_conf *qconf;/***计算间隔时间100us对应tick数*/const uint64_t drain_tsc = (rte_get_tsc_hz() + US_PER_S - 1) /US_PER_S * BURST_TX_DRAIN_US;struct ff_dpdk_if_context *ctx;prev_tsc = 0;usch_tsc = 0;qconf = &lcore_conf; //lcore分配信息while (1) {/*** 每经过一个时钟周期，tsc寄存器就自动加1， rte_rdtsc()只是获得tsc寄存器的值* rte_get_tsc_hz() 返回一秒的tsc数目* CPU MHz为1600，那么tsc的1就是1/1600/1000/1000的时间*/cur_tsc = rte_rdtsc();//获取当前时钟数if (unlikely(freebsd_clock.expire < cur_tsc)) {//定时器到期，管理定时器列表并执行定时器回调函数rte_timer_manage();}/*** idel=1：表示没有新的数据包发送或者接收* idle=0：表示有新的数据达到或者发送，可能有感兴趣事件触发，需要调用用户回调*/idle = 1;sys_tsc = 0;usr_tsc = 0;/*** 数据实际发送至物理网卡不是有数据包就立即发送，而是间隔100us使用burst方式发送*/diff_tsc = cur_tsc - prev_tsc;if (unlikely(diff_tsc > drain_tsc)) {/*** 发送间隔到期，遍历所有的port，发送缓存的需要的数据包*/for (i = 0; i < qconf->nb_tx_port; i++) {port_id = qconf->tx_port_id[i];if (qconf->tx_mbufs[port_id].len == 0)continue;idle = 0;/*** 1. 如果启用pcap功能，记录tx数据包* 2.内部调用rte_eth_tx_burst发送数据包*/send_burst(qconf,qconf->tx_mbufs[port_id].len,port_id);qconf->tx_mbufs[port_id].len = 0;}/*** 更新prev_tsc时钟*/prev_tsc = cur_tsc;}/*** 处理rx队列上数据包，包括该lcore对应queueid上数据包，和通过dispatcher分发的数据包* 遍历proc_id处理的所有网卡队列*/for (i = 0; i < qconf->nb_rx_queue; ++i) {port_id = qconf->rx_queue_list[i].port_id;queue_id = qconf->rx_queue_list[i].queue_id;/*** 获取协议栈相关上下文struct ff_dpdk_if_context**/ctx = veth_ctx[port_id];#ifdef FF_KNI/*** 如果开启了kni功能，primary lcore核上处理需要上传给kni模块数据包*/if (enable_kni && rte_eal_process_type() == RTE_PROC_PRIMARY) {/***1. 从kni_rp上获取数据包，上传至kni模块*2. 将kni模块处理返回数据包发送至物理网卡*/ff_kni_process(port_id, queue_id, pkts_burst, MAX_PKT_BURST);}#endif/***处理其它lcore上分发过来的rx数据包*/process_dispatch_ring(port_id, queue_id, pkts_burst, ctx);/*** 从网卡队列读取数据包*/nb_rx = rte_eth_rx_burst(port_id, queue_id, pkts_burst,MAX_PKT_BURST);if (nb_rx == 0)continue;idle = 0;/*** 处理新收到的每个数据包*//* Prefetch first packets */for (j = 0; j < PREFETCH_OFFSET && j < nb_rx; j++) {rte_prefetch0(rte_pktmbuf_mtod(pkts_burst[j], void *));}/* Prefetch and handle already prefetched packets */for (j = 0; j < (nb_rx - PREFETCH_OFFSET); j++) {rte_prefetch0(rte_pktmbuf_mtod(pkts_burst[j + PREFETCH_OFFSET], void *));//最后一个参数为0，表示数据包不是从dispatch_ring上获取到的process_packets(port_id, queue_id, &pkts_burst[j], 1, ctx, 0);}/* Handle remaining prefetched packets */for (; j < nb_rx; j++) {process_packets(port_id, queue_id, &pkts_burst[j], 1, ctx, 0);}}/***处理控制消息数据*/process_msg_ring(qconf->proc_id);div_tsc = rte_rdtsc();/***调用用户回调函数，调用条件为*1）有新的数据处理过*2）距离上次回调超过100us*/if (likely(lr->loop != NULL && (!idle || cur_tsc - usch_tsc > drain_tsc))) {usch_tsc = cur_tsc;//更新用户上次回调事件，最大间隔为100uslr->loop(lr->arg);}/*** 1.如果设置了idle_sleep，并且当前没有新的数据包，睡眠*/idle_sleep_tsc = rte_rdtsc();if (likely(idle && idle_sleep)) {usleep(idle_sleep);end_tsc = rte_rdtsc();} else {end_tsc = idle_sleep_tsc;}end_tsc = rte_rdtsc();if (usch_tsc == cur_tsc) {usr_tsc = idle_sleep_tsc - div_tsc;}if (!idle) {sys_tsc = div_tsc - cur_tsc;ff_top_status.sys_tsc += sys_tsc;}ff_top_status.usr_tsc += usr_tsc;ff_top_status.work_tsc += end_tsc - cur_tsc;ff_top_status.idle_tsc += end_tsc - cur_tsc - usr_tsc - sys_tsc;ff_top_status.loops++;}return 0;}

process_packets

      static inline void process_packets(uint16_t port_id, uint16_t queue_id, struct rte_mbuf **bufs,uint16_t count, const struct ff_dpdk_if_context *ctx, int pkts_from_ring){struct lcore_conf *qconf = &lcore_conf;uint16_t nb_queues = qconf->nb_queue_list[port_id];//获取该网卡开启的队列数uint16_t i;for (i = 0; i < count; i++) {struct rte_mbuf *rtem = bufs[i];if (unlikely(qconf->pcap[port_id] != NULL)) {/*** 1.如果pkts_from_ring=0，则表示bufs是从lcore对应queueid接收数据包* 2. 如果pcap功能开启的话，记录数据包*/if (!pkts_from_ring) {ff_dump_packets(qconf->pcap[port_id], rtem);}}void *data = rte_pktmbuf_mtod(rtem, void *);//data指向数据包的起始位置uint16_t len = rte_pktmbuf_data_len(rtem);//获取数据包的长度if (!pkts_from_ring) {//更新rx统计信息ff_traffic.rx_packets++;ff_traffic.rx_bytes += len;}/*** packet_dispatcher为用户设置的自定义分发函数* FF_DISPATCH_ERROR：错误数据包，直接丢弃* FF_DISPATCH_RESPONSE：用户已经处理完成* 返回0~nb_queues-1：将该数据包分发到返回的queueid上处理*/if (!pkts_from_ring && packet_dispatcher) {int ret = (*packet_dispatcher)(data, &len, queue_id, nb_queues);/*** 如果用户已经处理完该数据包，直接将该数据包缓存在发送缓冲中*/if (ret == FF_DISPATCH_RESPONSE) {rte_pktmbuf_pkt_len(rtem) = rte_pktmbuf_data_len(rtem) = len;send_single_packet(rtem, port_id);continue;}if (ret == FF_DISPATCH_ERROR || ret >= nb_queues) {/*** 如果返回错误，则该数据包直接丢弃，不会交由freebsd协议栈处理*/rte_pktmbuf_free(rtem);continue;}/*** 如果返回ret与接收queue_id不相等，即该数据包需要交由其它lcore处理，将该数据包加入对应dispatch_ring*/if (ret != queue_id) {ret = rte_ring_enqueue(dispatch_ring[port_id][ret], rtem);if (ret < 0)rte_pktmbuf_free(rtem);continue;}}enum FilterReturn filter = protocol_filter(data, len);if (filter == FILTER_ARP) {/*** 如果数据包是ARP数据包，则将其广播至所有的队列上*/struct rte_mempool *mbuf_pool;struct rte_mbuf *mbuf_clone;if (!pkts_from_ring) {uint16_t j;for (j = 0; j < nb_queues; ++j) {if (j == queue_id)continue;unsigned socket_id = 0;if (numa_on) {uint16_t lcore_id = qconf->port_cfgs[port_id].lcore_list[j];socket_id = rte_lcore_to_socket_id(lcore_id);}mbuf_pool = pktmbuf_pool[socket_id];//深度拷贝数据包，主要需要注意rte_mbuf分段情况mbuf_clone = pktmbuf_deep_clone(rtem, mbuf_pool);if (mbuf_clone) {int ret = rte_ring_enqueue(dispatch_ring[port_id][j],mbuf_clone);if (ret < 0)rte_pktmbuf_free(mbuf_clone);}}}#ifdef FF_KNI//ARP数据包copy至kni模块，因为目前kni模块只区分tcp/udp端口if (enable_kni && rte_eal_process_type() == RTE_PROC_PRIMARY) {mbuf_pool = pktmbuf_pool[qconf->socket_id];mbuf_clone = pktmbuf_deep_clone(rtem, mbuf_pool);if (mbuf_clone) {ff_kni_enqueue(port_id, mbuf_clone);}}#endif//数据包传递至freebsd协议栈处理ff_veth_input(ctx, rtem);#ifdef FF_KNI} else if (enable_kni &&((filter == FILTER_KNI && kni_accept) ||(filter == FILTER_UNKNOWN && !kni_accept))) {//如果是kni处理数据包，push进kni对应ring队列中ff_kni_enqueue(port_id, rtem);#endif} else {//交由freebsd协议栈处理ff_veth_input(ctx, rtem);}}}

ff_veth_input

该函数主要将rte_mbuf转化成freebsd协议栈中mbuf数据结构
将mbuf递交给ether_input处理

    static void ff_veth_input(const struct ff_dpdk_if_context *ctx, struct rte_mbuf *pkt){/*** 如果开启了校验和检验硬件负载功能，检查mbuf的标志*/uint8_t rx_csum = ctx->hw_features.rx_csum;if (rx_csum) {if (pkt->ol_flags & (PKT_RX_IP_CKSUM_BAD | PKT_RX_L4_CKSUM_BAD)) {rte_pktmbuf_free(pkt);return;}}//data指向实际数据开端void *data = rte_pktmbuf_mtod(pkt, void*);//len表示数据包长度uint16_t len = rte_pktmbuf_data_len(pkt);/*** 将dpdk框架下rte_mbuf转换成freebsd协议栈上mbuf结构* 1. 初始化mbuf结构* 2.将rte_mbuf结构与mbuf结构关联起来，mbuf->m_ext实际数据指向data，零copy*/void *hdr = ff_mbuf_gethdr(pkt, pkt->pkt_len, data, len, rx_csum);if (hdr == NULL) {rte_pktmbuf_free(pkt);return;}if (pkt->ol_flags & PKT_RX_VLAN_STRIPPED) {/*** 如果vlan头被剥离，设置mbuf中vlan信息*/ff_mbuf_set_vlan_info(hdr, pkt->vlan_tci);}/*** 处理rte_mbuf分段，在mbuf中将分段数据通过m_next指针连接*/struct rte_mbuf *pn = pkt->next;void *prev = hdr;while (pn != NULL) {data = rte_pktmbuf_mtod(pn, void *);len = rte_pktmbuf_data_len(pn);void *mb = ff_mbuf_get(prev, data, len);if (mb == NULL) {ff_mbuf_free(hdr);rte_pktmbuf_free(pkt);return;}pn = pn->next;prev = mb;}/*** 调用ifnet->if_input函数交由freebsd协议处理，实际处理函数为ether_input*/ff_veth_process_packet(ctx->ifp, hdr);}

F-Stack：ff_run函数详解相关推荐

np.stack()函数详解 ==＞堆叠【类似于torch.stack()】
目录 1.来看看axis=0时,它是如何进行堆叠的:(按矩阵进行堆叠) 2.再来看看axis=1的时候:(按行进行堆叠) 3.当axis=2时(按列的的元素进行堆叠,先堆叠三个矩阵的第0个元素0,12 ...
pytorch函数详解
pytorch函数详解在typora这里写之后复制到简书上 1. torchvision 1.1 transforms.Compose(transforms) 把几个转换组合 example: fr ...
【ES6】Generator函数详解
[ES6]Generator函数详解一.Generator函数简介基本概念函数写法 yield关键字介绍二.next方法的参数三.for...of循环四.关于普通throw()与Gener ...
python平方数迭代器_对python中的高效迭代器函数详解
python中内置的库中有个itertools,可以满足我们在编程中绝大多数需要迭代的场合,当然也可以自己造轮子,但是有现成的好用的轮子不妨也学习一下,看哪个用的顺手~ 首先还是要先import一下: ...
scanf函数详解与缓冲区
1.基本信息函数原型: int scanf( char *format, args, ...); 函数返回值: 读入并赋给args的数据个数,遇到文件结束返回EOF,出错返回0. 函数功能: sca ...
pythonpandas函数详解_对pandas中Series的map函数详解
Series的map方法可以接受一个函数或含有映射关系的字典型对象. 使用map是一种实现元素级转换以及其他数据清理工作的便捷方式. (DataFrame中对应的是applymap()函数,当然Dat ...
python3 format函数_Python学习教程：Python3之字符串格式化format函数详解（上）
Python学习教程:Python3之字符串格式化format函数详解(上) 概述在Python3中,字符串格式化操作通过format()方法或者f'string'实现.而相比于老版的字符串格式化方 ...
QT：常用函数详解--常用操作记录（个人笔记）
QT:常用函数详解(个人笔记) PS:一下内容个人笔记,要求自己看懂,随笔,阅读体验会很差很差! Qt setContentsMargins()函数函数原型:void QLayout::setCon ...
python3 内置函数详解
内置函数详解 abs(x) 返回数字的绝对值,参数可以是整数或浮点数,如果参数是复数,则返回其大小. # 如果参数是复数,则返回其大小.>>> abs(-25) 25>> ...

F-Stack：ff_run函数详解

ff_run详解

F-Stack：ff_run函数详解相关推荐

最新文章

热门文章