简单测试了一下virtio-user的性能。测试拓扑------- | | tap | host0 | host1 ------- | | --------------- ------------ | ------- --------| testpmd(dpdk) |----| eth0(dpdk) |-------|-------| eth0 | --------------- ------------ | -------通过iperf在host0的tap口和host1的eth0口进行收发报文。环境搭建我们只在host0上搭建virtio-user环境，host1依然采用原生的物理口DPDK版本为17.05.2，执行命令如下：testpmd -c 0x40004 -n 4 -d /lib64/librte_pmd_virtio.so --vdev=virtio_user0,path=/dev/vhost-net,queue_size=1024 -- -a --disable-hw-vlan --forward-mode=csum这是会启动一个tap0口，给tap0口配置IP。大包对比测试原生物理口测试为了找一个基准数据，我们在这两台服务器搭建virtio-user环境之前先用iperf进行流量测试，结果如下：TCP性能为 9.42 GbpsUDP性能为发 9.57 Gbps， 收 7.59 Gbps，有 20% 的丢包virtio-user作为发送方性能如下：TCP性能为 9.42 GbpsUDP性能为发 8.86 Gbps, 收 6.66 Gbps， 有 20% 的丢包virtio-user作为接收方性能如下：TCP性能为 9.42 GbpsUDP性能为发 6.33 Gbps，收 6.33 Gbps，有 0% 的丢包小包对比测试使用pktgen进行64Bytes的UDP报文发送，另一端通过sar -n DEV查看端口的kernel收包量，另外使用pktgen-dpdk查看一次端口的收包量，因为收包时过内核网络协议栈会严重降低收包量的这里我们不再使用一个tap口，而是通过virtio-user创建出多个tap口进行性能测试，并且只测试从tap口发送流量到另一端的物理端口。原生物理口测试发包性能为 578w pps， 内核收包 78w pps，dpdk收包 578w ppsvirtio-user生成2个tap口发包性能为 233w pps，内核收包 147w pps， dpdk收包 233w pps，这里感觉应该是两个tap口，两个IP给目的IP发送流量，两条流处理所以是上面单个物理口的两倍包量virtio-user生成4个tap口发包性能为 571w pps，内核收包 222w pps， dpdk收包 571w ppsvirtio-user生成8个tap口发包性能为 678w pps，内核收包 356w pps， dpdk收包 678w pps因为测试使用的是DPDK的例子testpmd，显示丢包都在tap口发送给testpmd的时候，testpmd转发并没有丢包，所以初步怀疑瓶颈是在tap处。
主要查看一下virtio-user的工作思路，个人觉得他主要是用来替换KNI或者OVS的TAP设备，更好的用法应该是给container来用，主要是通过操作/dev/vhost-net创建kernel的tap设备用，然后kernel和virtio-user进行内存映射，便于收发报文。| 更正一下：目前来说container用virtio-user还不太合适，因为如果TAP设备加入到namespace的话，重启virtio-user就因为在当前的宿主机找不到TAP设备，会生成一个新的TAP设备。这块目前没有解决。官方图以上就是virtio-user的结构图，我们可以看到virtio-user驱动会在内核生成一个tap设备，然后另一端连接DPDK，报文通过内核发送到用户态。virtio-user创建tap端口流程接下来详细看一下，函数路径如下，就是初始化的时候，根据参数设置初始化虚拟设备。rte_eal_init-->rte_bus_probe-->vdev_probe-->vdev_probe_all_drivers-->virtio_user_pmd_probe主要是两个操作首先会解析一些参数，用来配置tap口，支持的参数为queues，cq，mac，path，queue_size，iface，这里面path是必备的，但是mac和iface我们可能也常用。如果进程是primary的话，调用virtio_user_dev_init初始化设备如果进程是secondary的话，调用rte_eth_dev_attach_secondary绑定设备，这里所做的就是简单的获取端口，默认就是由primary进行端口的配置了。eth_virtio_dev_init初始化vhost-user或者vhost-net设备virtio_user_dev_init-->virtio_user_dev_setup主要是根据支持对端为vhost-user和vhost-net，并且配置通知和中断，另外我们接下来主要关心一下vhost-net。virtio_user_dev_setup当是vhost-user时，得到vhost-user相关的ops，vhost-net时，得到kernel相关的vhost的ops，接下来就看kernel的vhost-net，不管vhost-user。首先申请vhostfds和tapfds的内存相关空间。调用vhost_kernel_setup打开相关的设备，比如我们打开的是dev/vhost-net。调用virtio_user_dev_init_notify去初始化通知。调用virtio_user_fill_intr_handle创建中断。eth_virtio_dev_initvirtio_init_device初始化设备rte_intr_callback_register注册中断回调virtio_init_devicevtpci_reset重置设备vtpci_set_status告诉host，注意到设备，并且知道怎么去驱动它了virtio_alloc_queues为设备创建队列virtio_configure_intr配置中断vtpci_reinit_complete实际调用vtpci_set_status告诉host，驱动完成。vtpci_set_status-->virtio_user_set_status如果驱动完成了，调用virtio_user_start_device驱动设备如果是重置标志，调用virtio_user_reset重置设备如果其他状态，则将状态记录在设备中virtio_user_start_devicevirtio_user_queue_setup调用virtio_user_create_queue，最终调用ioctl告知kernel的vhost，创建队列。dev->ops->send_request调用vhost_kernel_ioctl配置共享内存virtio_user_queue_setup调用virtio_user_kick_queue配置队列的vringdev->ops->enable_qp调用vhost_kernel_enable_queue_pair创建并且配置tap设备vhost_kernel_enable_queue_pair vhost_kernel_open_tap创建tap设备，并且进行配置。vhost_kernel_set_backend配置tap设备数据传递流程Kernel部分接下来我们看一下数据的传递路径，主要看一个流向，就是pktgen通过tap口进行流量发送，然后最终会传递给virtio-user的接收。pktgen发送直接调用的是tap的ndo_start_xmit，即tun_net_xmit 。tun_net_xmit-->vhost_poll_wakeup-->vhost_poll_queue-->vhost_work_queue-->wake_up_process会唤醒vhost_worker。vhost_worker-->handle_rx_kick-->handle_rx-->tun_recvmsg-->tun_do_read会进行报文的收取，然后发送到用户态，而virtio-user处会轮询收取报文，所以我们略过了消息通知机制，其实是我没有找到。DPDK部分rte_eth_rx_burst-->virtio_recv_pkts_vec会从相应的队列中读取数据。目前问题virtio-user创建的tap设备是down的状态，需要手动up，up之前，virtio-user是不知道设备是down的，如果有别的类型的端口的报文，可能就会向virtio-user端口转发。最后就是tap设备up或者down的状态目前还不能通知virtio-user。mvs重启后tap设备丢失，tap设备重新生成后，是down状态，并且Mac会变化，创建时并不能指定，tap设备丢失的问题会在DPDK-18.05中加入，但是这个功能稳定性和可用性还需要再验证。virtio-user创建的tap设备的mac地址不能指定。
这次因为遇到想要添加其他类型的端口，所以查看一下添加端口的流程。首先查看命令ovs-vsctl的实现，这里我的理解主要是操作ovsdb的，需要通过jsonrpc和ovsdb-server进行通信。现在我们主要是添加端口的，其他的先不看。utilities/ovs-vsctl.cmain-->do_vsctl-->cmd_add_port(c->syntax->run in vsctl_commands)-->add_port函数用来添加端口。add_port如果有--may_exist参数存在，则查看是否已经有该名字的port，有的话退出，没有的话继续。调用check_conflicts查看port和interface是否存在名字的冲突，冲突报错退出。找到相应的桥，然后添加port和interface，因为都是ovsrec操作，我们就不看了。ovsdb-server我们就不看了，因为就是用来连接ovs-vsctl和ovs-vswitchd的中转站。ovs-vswitchd.cmain-->bridge_run-->bridge_reconfigure-->bridge_add_ports-->bridge_add_ports__-->iface_create-->iface_do_create-->netdev_open(rc->class->construct)会根据类型来调用相应的type的端口。所以我们发现，我们只要在这里添加相应类型的一个netdev_class，然后模仿其他的type的一堆接口实现相应的函数即可，ovs-vsctl命令以及ovsdb-server都对type类型不关心，只是负责配置和存储，所以完全可以不用去操作。写到这里我们就先不写了，等真正需要添加端口的时候，再看看netdev_class数据结构各个函数指针的意义和函数。
本文只是针对bond相关的流程，每个函数中只关注bond相关的操作。bond口添加首先是ovs-vsctl add-bond命令添加bond口，这里是配置到ovsdb里面，ovsdb-server进行处理的，我们暂时忽略，我们主要关注的是ovs-vswitchd进程监测到添加了bond口之后的操作。调用路径如下，主要是bridge_run函数从ovsdb-server进程得知端口发生了变化，就会去调用bridge_reconfigure进行重新配置，看样子是配置所有的，我们只关注bond相关的。main-->bridge_run-->bridge_reconfigure先将DPDK的多个端口添加到桥中，在ovs中的概念就是多个interface，一个port。最终会调用到netdev_dpdk_construct.bridge_reconfigure-->bridge_add_ports-->bridge_add_ports__-->iface_create-->iface_do_create-->netdev_open(rc->class->construct)-->netdev_dpdk_construct接着是端口的配置了port_configure开局先获取该port有多少个interface，即多少slave。port_configure_lacp首先是获取LACP的配置，有则调用iface_configure_lacp配置每个interface。slave > 1 表示是bond口，调用port_configure_bond最后调用ofproto_bundle_register-->bundle_set对bond口进行操作来看一下bundle_set调用lacp_configure配置LACP调用bundle_add_port添加端口如果已经创建了bond口了，则调用bond_reconfigure重新配置bond口，没有则调用bond_create创建bond口，里面也包含了配置。调用bond_slave_register注册slave以上主要就是端口添加的时候的操作，大部分都是数据结构添加和配置保存到结构中，已被运行的时候根据配置运行，接下来我们看一下bond口运行。bond口运行主要是非主线程的操作。主线程bond口运行调用流程如下：main-->bridge_run-->bridge_run__-->ofproto_run(p->ofproto_class->run)-->runrun中有两个bond相关的操作bundle_run是bond口的基本操作如果需要rebalance的话，bond_rebalance会做该操作，这块暂时不看了，没看懂，主要是recirc有点吓datapath的混了，不知道是不是一个，主要就是根据统计的报文大小看情况要不要迁移。bundle_run如果是LACP协议的话，lacp_run主要是根据rx定时器更新slave信息，更新lacp的attache信息，tx定时器超时候，生成并且发送LACPDU报文，重置定时器。是bond口的话，bond_run主要是监测连接状态，以防端口down掉不能接收发送报文。如果有需要，bundle_send_learning_packets会发送学习报文，主要是自己的MAC和VLAN信息，当是SLB、主备或者是不带协商的LACP时需要发送。不写了Bond口的运行分析到这里就差不多，本来觉得应该还有接收发送报文时候的bond相关的接口，但是没有查看到相关的操作，猜测应该是该bond是针对流表的balance，当然是可以rebalance，也就是datapath流表还是有多个物理口，下流表的时候，流表balance给bond的几个口，然后收发报文就都是相当于没有bond口那样的操作了，最后如果有口down掉的时候，从数据结构中删除，然后流表部分做相应的调整(不知道是不是流量迁移)，rebalance的时候，也可以流量迁移。官方给出的bond后使用方式，一般都是normal流表，以后有时间可以从这块去看一下，normal流表的时候怎么去balance流表的。
想要看一看DPDK的抓包是怎么做的，我有没有改进的空间。用法这里不讲，可以查看官方文档。通过了解，DPDK的抓包是采用的C/S架构，DPDK有一个librte_pdump的库是server端，用来接收client端的指令去配置抓包，类似于linux kenrel做的工作，而client端就是我们运行的抓包命令，类似于tcpdump命令。librte_pdump由基于DPDK的主程序去调用rte_pdump_init创建server端。int rte_pdump_init(const char *path) { ... /* 设置unix socket domain的server端路径 */ rte_pdump_set_socket_dir(path, RTE_PDUMP_SOCKET_SERVER); /* 创建unix socket domain的socket */ pdump_create_server_socket(); /* 创建线程持续接收client端的信息，以便于在抓包时挂载抓包函数 */ ret = pthread_create(&pdump_thread, NULL, pdump_thread_main, NULL); /* 配置线程名字为pdump-thread，便于调试 */ snprintf(thread_name, RTE_MAX_THREAD_NAME_LEN, "pdump-thread"); ret = rte_thread_setname(pdump_thread, thread_name); return 0; }这里面我们比较重视的是pdump_thread_main，所以看一下该函数。static __attribute__((noreturn)) void * pdump_thread_main(__rte_unused void *arg) { /* 无限循环，没有退出机制 */ for (;;) { /* 从客户端接收信息 */ recvfrom(pdump_socket_fd, &cli_req, sizeof(struct pdump_request), 0, (struct sockaddr *)&cli_addr, &cli_len); /* 根据接收的信息挂载抓包函数 */ set_pdump_rxtx_cbs(&cli_req); /* 给客户端返回一些信息 */ resp.ver = cli_req.ver; resp.res_op = cli_req.op; resp.err_value = ret; sendto(pdump_socket_fd, &resp, sizeof(struct pdump_response), 0, (struct sockaddr *)&cli_addr, cli_len); } }接下来我们关注的是挂载抓包函数，所以看一下函数et_pdump_rxtx_cbs。static int set_pdump_rxtx_cbs(struct pdump_request *p) { /* 通过设备名称找到端口信息 */ if (operation == ENABLE) { rte_eth_dev_get_port_by_name(p->data.en_v1.device, &port); queue = p->data.en_v1.queue; ring = p->data.en_v1.ring; mp = p->data.en_v1.mp; } else { rte_eth_dev_get_port_by_name(p->data.dis_v1.device, &port); queue = p->data.dis_v1.queue; ring = p->data.dis_v1.ring; mp = p->data.dis_v1.mp; } /* 判断是抓所有队列的报文还是只抓一些队列的 */ if (queue == RTE_PDUMP_ALL_QUEUES) { rte_eth_dev_info_get(port, &dev_info); nb_rx_q = dev_info.nb_rx_queues; nb_tx_q = dev_info.nb_tx_queues; } /* register RX callback */ if (flags & RTE_PDUMP_FLAG_RX) { end_q = (queue == RTE_PDUMP_ALL_QUEUES) ? nb_rx_q : queue + 1; pdump_regitser_rx_callbacks(end_q, port, queue, ring, mp, operation); } /* register TX callback */ if (flags & RTE_PDUMP_FLAG_TX) { end_q = (queue == RTE_PDUMP_ALL_QUEUES) ? nb_tx_q : queue + 1; pdump_regitser_tx_callbacks(end_q, port, queue, ring, mp, operation); } }我们主要操作就是挂载函数，分别是RX和TX处挂载回调函数，这块我们只关注一个就行，因为两者操作基本一致。static int pdump_regitser_rx_callbacks(uint16_t end_q, uint8_t port, uint16_t queue, struct rte_ring *ring, struct rte_mempool *mp, uint16_t operation) { uint16_t qid = (queue == RTE_PDUMP_ALL_QUEUES) ? 0 : queue; for (; qid < end_q; qid++) { struct pdump_rxtx_cbs *cbs = &rx_cbs[port][qid]; if (cbs && operation == ENABLE) { cbs->ring = ring; cbs->mp = mp; /* 此处挂载pdump_rx函数，而tx处挂载pdump_tx函数 */ cbs->cb = rte_eth_add_first_rx_callback(port, qid, pdump_rx, cbs); } if (cbs && operation == DISABLE) { int ret = rte_eth_remove_rx_callback(port, qid, cbs->cb); cbs->cb = NULL; } } return 0; }接收和发送分别挂载了函数pdump_rx和pdump_tx，而他们两个都是调用了函数pdump_copy。static inline void pdump_copy(struct rte_mbuf **pkts, uint16_t nb_pkts, void *user_params) { uint16_t d_pkts = 0; struct rte_mbuf *dup_bufs[nb_pkts]; struct pdump_rxtx_cbs *cbs = user_params; struct rte_ring *ring = cbs->ring; struct rte_mempool *mp = cbs->mp; /* 拷贝一份报文 */ for (unsigned i = 0; i < nb_pkts; i++) { struct rte_mbuf *p = pdump_pktmbuf_copy(pkts[i], mp); if (p) { dup_bufs[d_pkts++] = p; } } /* 通过ring发送 */ int ring_enq = rte_ring_enqueue_burst(ring, (void *)dup_bufs, d_pkts); if (unlikely(ring_enq < d_pkts)) { RTE_LOG(DEBUG, PDUMP, "only %d of packets enqueued to ring\n", ring_enq); do { rte_pktmbuf_free(dup_bufs[ring_enq]); } while (++ring_enq < d_pkts); } }至此，报文已经传递出去了，就需要看看client端怎么接收。pdumpapp/pdump/main.cmain-->enable_pdump主要和server进行通信，然后配置抓包，完成后才开始抓包操作。static void enable_pdump(void) { /* 这个路径通常需要参数配置，因为是server指定的 */ if (server_socket_path[0] != 0) { rte_pdump_set_socket_dir(server_socket_path, RTE_PDUMP_SOCKET_SERVER); } /* 这块可以不指定，可以采用默认值，然后通信时告知server */ if (client_socket_path[0] != 0) { rte_pdump_set_socket_dir(client_socket_path, RTE_PDUMP_SOCKET_CLIENT); } /* 启动pdump，因为参数不同，调用的函数和参数不同 * 但是实质调用的函数都一致，就是pdump_create_client_socket */ for (int i = 0; i < num_tuples; i++) { struct pdump_tuples *pt = &pdump_t[i]; if (pt->dir == RTE_PDUMP_FLAG_RXTX) { if (pt->dump_by_type == DEVICE_ID) { rte_pdump_enable_by_deviceid( pt->device_id, pt->queue, RTE_PDUMP_FLAG_RX, pt->rx_ring, pt->mp, NULL); rte_pdump_enable_by_deviceid( pt->device_id, pt->queue, RTE_PDUMP_FLAG_TX, pt->tx_ring, pt->mp, NULL); } else if (pt->dump_by_type == PORT_ID) { rte_pdump_enable(pt->port, pt->queue, RTE_PDUMP_FLAG_RX, pt->rx_ring, pt->mp, NULL); rte_pdump_enable(pt->port, pt->queue, RTE_PDUMP_FLAG_TX, pt->tx_ring, pt->mp, NULL); } } else if (pt->dir == RTE_PDUMP_FLAG_RX) { ... } else if (pt->dir == RTE_PDUMP_FLAG_TX) { ... } } }static int pdump_create_client_socket(struct pdump_request *p) { int n; struct pdump_response server_resp; struct sockaddr_un addr, serv_addr, from; socklen_t addr_len, serv_len; int pid = getpid(); int socket_fd = socket(AF_UNIX, SOCK_DGRAM, 0); pdump_get_socket_path(addr.sun_path, sizeof(addr.sun_path), RTE_PDUMP_SOCKET_CLIENT); addr.sun_family = AF_UNIX; addr_len = sizeof(struct sockaddr_un); bind(socket_fd, (struct sockaddr *) &addr, addr_len); serv_len = sizeof(struct sockaddr_un); memset(&serv_addr, 0, sizeof(serv_addr)); pdump_get_socket_path(serv_addr.sun_path, sizeof(serv_addr.sun_path), RTE_PDUMP_SOCKET_SERVER); serv_addr.sun_family = AF_UNIX; /* * 以上是创建socket，然后将配置内容发送给server * server配置好会给回复，然后就会进入到抓包的流程 * 如果没有收到回复是不能抓包的 */ sendto(socket_fd, p, sizeof(struct pdump_request), 0, (struct sockaddr *)&serv_addr, serv_len); recvfrom(socket_fd, &server_resp, sizeof(struct pdump_response), 0, (struct sockaddr *)&from, &serv_len); close(socket_fd); unlink(addr.sun_path); }main-->dump_packets-->pdump_rxtx就开启了抓包的操作。static inline void pdump_rxtx(struct rte_ring *ring, uint8_t vdev_id, struct pdump_stats *stats) { /* write input packets of port to vdev for pdump */ struct rte_mbuf *rxtx_bufs[BURST_SIZE]; /* 拿到报文 */ const uint16_t nb_in_deq = rte_ring_dequeue_burst(ring, (void *)rxtx_bufs, BURST_SIZE); stats->dequeue_pkts += nb_in_deq; if (nb_in_deq) { /* 发送，可以是设备，可以是pcap文件 */ uint16_t nb_in_txd = rte_eth_tx_burst( vdev_id, 0, rxtx_bufs, nb_in_deq); stats->tx_pkts += nb_in_txd; if (unlikely(nb_in_txd < nb_in_deq)) { do { rte_pktmbuf_free(rxtx_bufs[nb_in_txd]); stats->freed_pkts++; } while (++nb_in_txd < nb_in_deq); } } }TODO希望在server端能加上类似Linux kernel的bfp报文过滤。client可以支持各种filter语法。
一直在纠结这块要不要写出来，毕竟自己没有看懂，后来还是鼓起梁静茹给的勇气，看多少写多少吧。操作方式首先先说我们操作内存热扩容的命令操作，采用的是qemu monitor的方式，执行如下两条命令：两条命令# 此处一定要加share=on，不然会出现qemu扩的内存和dpdk映射的内存不是同一块 # 导致内存浪费和网络通信的问题 object_add memory-backend-file,id=mem1,size=4G,mem-path=/mnt/huge,share=on device_add pc-dimm,id=dimm1,memdev=mem1从文件hmp-commands.hx中我们可以看一下这两条命令针对的函数调用： { .name = "object_add", .args_type = "object:O", .params = "[qom-type=]type,id=str[,prop=value][,...]", .help = "create QOM object", .mhandler.cmd = hmp_object_add, .command_completion = object_add_completion, }, STEXI @item object_add @findex object_add Create QOM object. ETEXI { .name = "device_add", .args_type = "device:O", .params = "driver[,prop=value][,...]", .help = "add device, like -device on the command line", .mhandler.cmd = hmp_device_add, .command_completion = device_add_completion, }, STEXI @item device_add @var{config} @findex device_add Add device. ETEXImonitor_init-->monitor_command_cb-->handle_hmp_command最后会调用到命令行执行的函数，可以看到第一条命令对应的函数hmp_object_add，第二条命令对应的函数hmp_device_add，接下来我们针对这两个函数进行分析。内存扩容void hmp_object_add(Monitor *mon, const QDict *qdict) { /* 解析object的参数 */ QemuOpts *opts = qemu_opts_from_qdict(qemu_find_opts("object"), qdict, &err); /* 创建一些函数指针，下个函数会用到 */ OptsVisitor *ov = opts_visitor_new(opts); /* 创建实例，并且申请配置的实例大小 */ Object *obj = user_creatable_add(qdict, opts_get_visitor(ov), &err); }Object *user_creatable_add(const QDict *qdict, Visitor *v, Error **errp) { char *type = NULL; char *id = NULL; Object *obj = NULL; Error *local_err = NULL, *end_err = NULL; /* clone一份 */ QDict *pdict = qdict_clone_shallow(qdict); /* 调用opts_start_struct创建参数存储 */ visit_start_struct(v, NULL, NULL, 0, &local_err); /* 删除参数，并且解析参数出来 */ qdict_del(pdict, "qom-type"); visit_type_str(v, "qom-type", &type, &local_err); /* 同上 */ qdict_del(pdict, "id"); visit_type_str(v, "id", &id, &local_err); /* 创建目标，并且根据需求做相应的操作，这里我们就是申请内存 */ obj = user_creatable_add_type(type, id, pdict, v, &local_err); return obj; }Object *user_creatable_add_type(const char *type, const char *id, const QDict *qdict, Visitor *v, Error **errp) { /* 根据type类型获取class */ ObjectClass *klass = object_class_by_name(type); const QDictEntry *e; /* 创建目标 */ Object *obj = object_new(type); /* 调用host_memory_backend_memory_complete */ user_creatable_complete(obj, &local_err); return obj; }static void host_memory_backend_memory_complete(UserCreatable *uc, Error **errp) { HostMemoryBackend *backend = MEMORY_BACKEND(uc); HostMemoryBackendClass *bc = MEMORY_BACKEND_GET_CLASS(uc); if (bc->alloc) { /* 申请memory region，我们是通过调用file_backend_memory_alloc */ bc->alloc(backend, &local_err); /* 获取memory region的可用地址 */ void *ptr = memory_region_get_ram_ptr(&backend->mr); /* 获取memory region的大小 */ uint64_t sz = memory_region_size(&backend->mr); /* 对该段地址进行一些特性的设置，比如可以合并，不允许dump */ if (backend->merge) { qemu_madvise(ptr, sz, QEMU_MADV_MERGEABLE); } if (!backend->dump) { qemu_madvise(ptr, sz, QEMU_MADV_DONTDUMP); } #ifdef CONFIG_NUMA /* 没细看，我觉得应该是计算有几个node，我们一般是两个 */ unsigned long lastbit = find_last_bit(backend->host_nodes, MAX_NODES); unsigned long maxnode = (lastbit + 1) % (MAX_NODES + 1); unsigned flags = MPOL_MF_STRICT | MPOL_MF_MOVE; /* 设置内存策略，内核尝试移动所有内存范围内的页面，如果无法移动，则失败报错EIO */ mbind(ptr, sz, backend->policy, maxnode ? backend->host_nodes : NULL, maxnode + 1, flags); #endif /* Preallocate memory after the NUMA policy has been instantiated. * This is necessary to guarantee memory is allocated with * specified NUMA policy in place. */ if (backend->prealloc) { /* 安装信号处理函数，并且清空内存内容 */ os_mem_prealloc(memory_region_get_fd(&backend->mr), ptr, sz); } } }内存设备插入hmp_device_add-->qmp_device_addvoid qmp_device_add(QDict *qdict, QObject **ret_data, Error **errp) { /* 解析device的参数 */ QemuOpts *opts = qemu_opts_from_qdict(qemu_find_opts("device"), qdict, &local_err); DeviceState *dev = qdev_device_add(opts, &local_err); }DeviceState *qdev_device_add(QemuOpts *opts, Error **errp) { BusState *bus = NULL; /* 查找驱动，并根据驱动找到设备的class */ const char *driver = qemu_opt_get(opts, "driver"); DeviceClass *dc = qdev_get_device_class(&driver, errp); /* find bus */ const char *path = qemu_opt_get(opts, "bus"); if (path != NULL) { bus = qbus_find(path, errp); } else if (dc->bus_type != NULL) { bus = qbus_find_recursive(sysbus_get_default(), NULL, dc->bus_type); } /* create device */ DeviceState *dev = DEVICE(object_new(driver)); if (bus) { qdev_set_parent_bus(dev, bus); } const char *id = qemu_opts_id(opts); if (id) { dev->id = id; } if (dev->id) { object_property_add_child(qdev_get_peripheral(), dev->id, OBJECT(dev), NULL); } else { static int anon_count; gchar *name = g_strdup_printf("device[%d]", anon_count++); object_property_add_child(qdev_get_peripheral_anon(), name, OBJECT(dev), NULL); g_free(name); } /* set properties */ qemu_opt_foreach(opts, set_property, dev, &err); dev->opts = opts; object_property_set_bool(OBJECT(dev), true, "realized", &err); return dev; }看不下去了，直接写上路径的了。hmp_device_add-->qmp_device_add-->qdev_device_add-->object_property_set_bool-->object_property_set_qobject-->object_property_set-->property_set_bool-->device_set_realized-->hotplug_handler_plug-->pc_machine_device_plug_cb-->pc_dimm_plug-->pc_dimm_memory_plug-->memory_region_add_subregion-->memory_region_add_subregion_common-->memory_region_update_container_subregions-->memory_region_transaction_commit-->vhost_commit-->vhost_user_set_mem_table此处函数大概的意思是对DPDK进行一些配置，主要是传递过去作为共享内存的文件描述符fd，并且进行之前qemu扩的内存的mmap，这里需要注意的有两点：文件描述符通过unix socket domain传递的时候可能是在DPDK的进程中重新做了一个fd值映射相同的文件，所以qemu端发送的fd的值和DPDK端收到的值不一样，但是映射的是同一个文件。因为每次扩容，都会将每个region的fd都传一次，所以老的region(非扩容部分)也会传来一个新的fd，这块相当于对映射文件做了多次打开，所以需要注意fd的关闭，保证文件的所有fd都关闭，防止内存泄露，做法就是每次新传来老region的fd，要么关闭新的，要么关闭旧的，保证每个region只打开一个fd。
最近OVS-DPDK做分片重组的时候，遇到一些问题，所以现在大致看了一下分片思路和代码，现在记录一下。DPDK版本为16.11.1分片试验因为涉及到封装，所以我的分片不是基于标准的MTU 1500，而是基于1422，我采用ICMP报文，长度5000。以下记录了一下打印信息。从打印的信息可以看到，分片并尽量不使用copy，所以导致于新的分片都都几个段组成，中间记录了一些偏移量。in_packet buf_addr 0x7f49901b8180, data_off 128, buf_len 2184, nb_segs 4, pkt_len 5042, data_len 1514 buf_addr 0x7f49908ac680, data_off 162, buf_len 2184, nb_segs 3, pkt_len 1480, data_len 1480 buf_addr 0x7f49908abb40, data_off 162, buf_len 2184, nb_segs 2, pkt_len 1480, data_len 1480 buf_addr 0x7f49908ab000, data_off 162, buf_len 2184, nb_segs 1, pkt_len 568, data_len 568 out_packet buf_addr 0x7f49908aa4c0, data_off 114, buf_len 2184, nb_segs 2, pkt_len 1456, data_len 34 buf_addr 0x7f49901b8180, data_off 162, buf_len 2184, nb_segs 1, pkt_len 1500, data_len 1422 out_packet buf_addr 0x7f49908a8e40, data_off 114, buf_len 2184, nb_segs 3, pkt_len 1456, data_len 34 buf_addr 0x7f49901b8180, data_off 1584, buf_len 2184, nb_segs 1, pkt_len 1500, data_len 58 buf_addr 0x7f49908ac680, data_off 162, buf_len 2184, nb_segs 1, pkt_len 1480, data_len 1364 out_packet buf_addr 0x7f49908a6c80, data_off 114, buf_len 2184, nb_segs 3, pkt_len 1456, data_len 34 buf_addr 0x7f49908ac680, data_off 1526, buf_len 2184, nb_segs 1, pkt_len 1480, data_len 116 buf_addr 0x7f49908abb40, data_off 162, buf_len 2184, nb_segs 1, pkt_len 1480, data_len 1306 out_packet buf_addr 0x7f49908a4ac0, data_off 114, buf_len 2184, nb_segs 3, pkt_len 776, data_len 34 buf_addr 0x7f49908abb40, data_off 1468, buf_len 2184, nb_segs 1, pkt_len 1480, data_len 174 buf_addr 0x7f49908ab000, data_off 162, buf_len 2184, nb_segs 1, pkt_len 568, data_len 568入报文，长度位5042(icmp 5000+8+20+14)第一个报文是偏移162开始L4数据，长度1480接下来的2个报文也都是偏移162开始，长度1480最后一个报文偏移162开始，长度568mtu 1442出报文4个，总长度5144第一个报文新建一个L2和L3的头，接着使用入报文1的L4开始的数据，长度位1422，总长度34+1422=1456第二个报文新建一个L2和L3的头，接着使用入报文1的偏移1584(1422+162)，长度58(报文长度1422+58=1480)，接着使用入报文2的偏移162，长度1364，所以总长度34+58+1364=1456第三个报文新建一个L2和L3的头，接着使用入报文2的偏移1526(1364+162)，长度116(报文长度1364+116=1480)，接着使用入报文3的偏移162，长度1306，所以总长度34+116+1306=1456第四个报文新建一个L2和L3的头，接着使用入报文3的偏移1468(1306+162)，长度174(报文长度1306+174=1480)，接着使用入报文4的偏移162，长度568，所以总长度34+174+568=776代码查看出现的问题int32_t rte_ipv4_fragment_packet(struct rte_mbuf *pkt_in, struct rte_mbuf **pkts_out, uint16_t nb_pkts_out, uint16_t mtu_size, struct rte_mempool *pool_direct, struct rte_mempool *pool_indirect) { ... //此处问题就是分片大小计算的时候默认IP协议头是20，不包含option //以下还有几处存在这个问题，需要注意 frag_size = (uint16_t)(mtu_size - sizeof(struct ipv4_hdr)); //此处要求分片大小必须是8的倍数，但是非debug版本这块是忽略的 //函数外保证这一点的话，会比较麻烦，因为mtu_size和ip头长度一起决定的 //分片大小反正是这里计算的，所以计算的时候保证是8的倍数比较好 RTE_ASSERT((frag_size & IPV4_HDR_FO_MASK) == 0); //所以以上两个问题推荐合并两行代码为以下一行代码 frag_size = (uint16_t)((mtu_size - ((in_hdr->version_ihl & 0xf) << 2)) & 0xfff8); ... }
注：本文是转载，但不是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; }
注：本文是参照了一些其他文章，原文地址点击这里。前言推荐看原文，因为我是懵逼的。并且这也算是广告贴了。谈到优化应用程序的性能，开发人员不必是性能专家，但应该精通自己的应用程序。许多方面的因素都会对应用程序性能产生影响，需要从硬件平台，代码设计和利用microarchitecture特性对功能代码进行微调。应用程序开发人员了解应用程序如何利用硬件资源的基本要求是，对代码设计和实现有深入理解。这可以通过对硬件microarchitecture有更深入的了解，或者通过使用诸如VTune™ Amplifier的专用分析工具来实现。top-down方法一个比较好的性能调优方法是top-down的方式。 这种方法有三个阶段：顶部的系统调整，中间的应用程序调整和底层的microarchitecture调整。 系统调整涉及硬件和操作系统调整，而应用程序调整包括更好的设计，并行化和提高库的效率。 microarchitecture是最后一个阶段，包括仔细选择编译器标志，向量化和代码重构w.r.t内存/缓存优化以及对CPU陷阱的理解，如下图所示。Microarchitecture和Micro-operations上图描述了第四代Intel® Core™ microarchitecture，其中上绿色部分是前端，下蓝色部分是pipeline的后端。前端是指令(例如ADD，MUL)被取出和解码(转换为较小的操作，称为micro-operations[μ-Ops])的地方。后端是执行所需操作，实际计算的地方。pipeline的每个块的详细描述超出了本文的范围。下图描述了前端和后端处理器流水线的职责。micro-operations(也称为μOps)成功完成整个datapath，称为retired。后端可以在每个cycle达到4个μOps。μOps通过(pass through)处理器pipeline并retired;偶尔由于各种分支错误预测，推测性地提取的几个μOps就会在其间被取消。pipeline slot表示处理一个μOp所需的硬件资源，这意味着在每个时钟周期的CPU核上有四个pipeline slot可用。如前所述，每个cycle可以有四个μOps retired，这意味着每个指令在理论上应该耗费0.25个cycle。在给定的时间点，pipeline slot可以为空或用μOps填充。pipeline slot分为四类，如下图所示。有关pipeline slot分类的更多信息，请参见本文。VTune™ AmplifierVTune Amplifier通过利用作为CPU内核的部分PMU(performance monitoring units)来收集测量结果。专用监视计数器可以收集和公开关于硬件资源消耗的信息。在PMU的帮助下，可以测量和检索关于所处理的指令的效率和缓存使用的度量。熟悉各种可用的指标（例如retired instructions; clock ticks; L2, L3 cache statistics; branch mis-predicts等）非常重要。此外，可以测量非CPU的PMU度量，如从/向存储器控制器读/写的字节，以及由Intel® QPI(Intel® QuickPath Interconnect)传输的数据流量。下面是用于计算不同pipeline slot度量的简要描述和公式。如果后端准备好接受更多μOps，而前端每个周期传送小于4μOps，那么应用程序考虑是front-end bound。这可能是由于提取代码(cache/ITLB问题)或解码指令的延迟所致。front-end bound pipeline slot可以被分类为子类别，如下图所示。公式：IDQ_UOPS_NOT_DELIVERED.CORE / (4 * Clockticks)如果由于在pipeline的后端缺少所需的资源而没有传送μOps，则应用程序考虑是back-end bound。这可能是因为内部结构填充了等待数据的μOps。下图中进一步描绘了front-end bound子类别。公式：Formula: 1 - (Front-end Bound + Bad speculation + Retiring)由于从分支错误预测或机器清除中恢复，μOps从不retire或分配slots被浪费时，pipeline slot被归类为bad speculation。公式：(UOPS_ISSUED.ANY - UOPS_RETIRED.RETIRE_SLOTS + 4* INT_MISC.RECOVERY_CYCLES) / (4* Clockticks)如果成功传递的μOps最终retired，那pipeline slot被归类为retiring公式：UOPS_RETIRED.RETIRE_SLOTS / (4 * Clockticks)每个子类别的详细描述超出了本文的范围。 有关每个子类别的深入描述，请参考CPU指标参考。OVS-DPDK瓶颈分析通过上面的一堆看不懂的东东，终于可以到了我想看到的性能瓶颈分析了，接着往下看吧，骚年。流表的报文批处理OVS-DPDK报文pipeline的一个重要阶段是流量批处理。首先对每个传入的报文进行分类，然后根据其匹配流进行分组，如下图所示。进入流表批处理的报文，根据该流表定义的相应的action（drop，push / pop VLAN）进行处理。为了提高报文转发性能，同一流表的报文将被批处理。偶尔，批处理中可能会有很少的报文。在最坏的情况下，每个获取的报文都匹配不同的流，因此每个批处理将包含一个数据包。当流表的相应action是将报文转发到某个物理端口时，由于DPDK接口上的数据包传输会导致昂贵的内存映射I/O（MMIO）写操作，所以传输报文可能非常低效。下图显示了报文处理关系调用图。对于每个输入的报文，在EMC中执行查找以匹配流表。在EMC命中的情况下，将报文使用dp_netdev_queue_batches()匹配流(请参阅OvS-DPDK源代码中的struct packet_batch_per_flow)。此后，使用packet_batch_per_flow_execute进行更快的数据包处理。如果流表的相应action是将数据包转发到DPDK端口，netdev_send将被调用，如下图所示。64字节UDP报文的性能基准测试用IXIA流量仪搭建测试环境，它发送64字节的UDP报文。在"PHY2PHY"测试用例中，有数十个不同的流时观察到显着的性能下降。请注意，流规则是唯一的，并且根据报文的源IP匹配流表。示例流程规则如下所示; 这将创建四个批次，并且数据包排队到相应的批次。$ ovs-ofctl add-flow br0 in_port=4,dl_type=0x0800,nw_src=2.2.2.1,actions=output:2 $ ovs-ofctl add-flow br0 in_port=4,dl_type=0x0800,nw_src=4.4.4.1,actions=output:2 $ ovs-ofctl add-flow br0 in_port=4,dl_type=0x0800,nw_src=6.6.6.1,actions=output:2 $ ovs-ofctl add-flow br0 in_port=4,dl_type=0x0800,nw_src=8.8.8.1,actions=output:2根据上述流程规则，VTune General Exploration分析显示了对OVS-DPDK传输瓶颈的有趣见解。VTune Amplifier总结当VTune分析运行60秒时，下图显示了VTune Amplifier关于pipeline slots占用情况的总结。请注意，粉红色的高亮slots需要注意，并由VTune根据应用程序类别的默认阈值自动高亮显示。根据VTune文档，下图显示了通过VTune放大器输出的OVS-DPDK结果，HPC(high-performance computing)应用程序中pipeline slots的预期范围之间的比较。如上图所示，对于特定的测试用例，OVS-DPDK是back-end bound的，retiring（只有35％）远远低于预期和健康70％限制的指令。bottom-up的分析显示了一些有趣的细节，如下图所示。netdev_dpdk_eth_send()消耗了总周期的17％。上述功能的每个指令周期（CPI）为4.921，远高于理论极限0.25，在HPC应用情况下可接受的范围为1.0。此功能完全是back-end bound，并且几乎不retiring任何指令（小于6％）。通过扩展back-end Bound列，可以更详细地了解back-end pipeline中发生的情况。 下图描述了back-end pipeline中的瓶颈在何处，并且它指向了L1 cache中的一个重大问题。列出的函数是关于报文传输的。这可能意味着back-end pipeline在I/O操作中遇到长时间的延迟。所以瓶颈可能是在于MMIO操作。当处理的报文（最多32个）匹配许多不同的流表时，情况尤其如此。在每个报文正在触发不同流表的最坏情况下，将触发MMIO操作，以从相应批处理去处理每个报文。降低MMIO成本的解决方案在实际情况下，物理端口的报文可能会遇到很多流表，每个流表的批处理缓存非常少的报文。当报文通过DPDK接口传输时，这变得非常低效。为了摊销MMIO写入的成本，可以使用中间队列来排队"NETDEV_MAX_BURST"（即32个）报文，并以rte_eth_tx_burst的格式发送数据包。中间队列是使用netdev_dpdk_eth_tx_queue()函数实现的，它将对数据包进行排队。当满足以下条件之一时，数据包将被传送。如果txq（txq-> count）> = NETDEV_MAX_BURST中的数据包计数，则调用netdev_dpdk_eth_tx_burst()来突发报文。超时后，任何等待队列的数据包都必须刷新，而不管其数量如何。通过这种解决方案，可以观察到back-end和Retiring类别的改进，如下图所示。下图显示，在应用补丁后，函数netdev_dpdk_eth_send()从列表的顶部到底部大幅度移动。通过减轻MMIO延迟的影响，测量的CPI从4.921下降到1.108。请注意，通过报文传输路径中的更多优化可以进一步提高性能; 但是，这超出了本文的范围。以上是使用VTune Amplifier了解和修复HPC应用程序瓶颈的示例。 在我们使用VTune Amplifier的分析中，我们使用了General Exploration模式，并集中在back-end bound度量上。值得注意的是，其他分析类型在VTune Amplifier工具中可用，如热点，高级热点模式，hpc性能，内存访问和锁定和等待。此外，其他指标可用于显示资源消耗。
注：本文是转载，但不是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; }