DPDK的提出以及设计思想是什么？-电子发烧友网

vhost-user

DPDK的提出以及设计思想

随着各种互联网应用的不断出现，网络设备以及服务器的带宽也快速提升，从千兆到万兆再到25G，40G，100G快速演变。

与此同时，在云计算和系统虚拟化威廉希尔官方网站的快速发展的推动下，虚拟机的网络处理能力需求也逐年增强。

另外，不仅是在服务器领域，很多网络服务也都逐渐向虚拟化，资源池化，云化的方向发展，比如路由器，交换机，防火墙，基站等常用网络设备原本都是硬件解决方案，现在也逐渐向虚拟化方向发展，业界急需适合高性能网络的软件解决方案。

既往的基于Linux实现的网络服务是基于内核的，也就是说所有数据包的发送和接收都要经过内核协议栈。在处理大流量和高并发的情况下，频繁的硬件中断会降低内核数据包的处理能力，同时内核空间和用户空间之间的数据拷贝也会产生比较大的负载。

为了进一步提升网络数据处理能力，Linux UIO (user-space drivers)威廉希尔官方网站把硬件操作映射到用户空间，实现了在用户空间直接操作网卡硬件。

这样网络数据包的处理就可以不经过Linux内核了，避免了高速网络数据处理时内核中断爆炸和大量数据复制的问题，进而让网络处理能力得以大幅度的提升。

绕过内核直接在用户态处理网络数据包，不仅可以解决内核的性能瓶颈，而且还易于开发和维护，和基于内核模块的实现相比也更灵活。

另外，因为程序运行在用户空间，即使有问题也不影响Linux内核和系统的其他模块，也增强了整个系统的可用性。

针对高性能网络软件开发框架的需求，基于Linux UIO威廉希尔官方网站在应用层处理网络数据包，这也正是IntelDPDK的主要设计思想。

DPDK应用初识

图1Linux内核处理路径(慢路径)和DPDK处理路径(快路径)对比

如图1所表示的就是基于Linux内核的数据处理路径(慢路径)和基于DPDK的用户空间数据处理路径(快路径)的对比。

这里需要特别说明的是，Intel DPDK被编译之后其实是一系列的库文件，供应用程序调用和链接。基于DPDK机制开发的应用程序在编译和链接DPDK的库文件之后，就可以进行高速的网络处理了。

这些DPDK应用都是运行在用户空间的应用程序。通过Linux UIO的机制以及少量内核模块(例如Intel x86平台上需要预先加载igb_uio等内核驱动模块)，这些运行在用户空间的DPDK应用程序能够旁路Linux内核，直接操作物理网卡进而完成高速网络数据的发送和接收。

截止到目前为止DPDK已经支持多种硬件体系结构，包括Intel x86，ARM，PowerPC等等，并且被网络设备厂商和互联网厂商广泛接受，已经有很多基于DPDK开发的产品和服务投入到生产环境使用了。

IntelDPDK的核心威廉希尔官方网站之一是PMD (用户态的轮询模式驱动)。通过非中断以及数据进出应用缓冲区内存的零拷贝机制，进一步提高发送接受数据的效率。用户态模式的PMD驱动，去除中断，避免内核态和用户态内存拷贝，减少系统开销，从而提升I/O吞吐能力。

我们可以通过分析DPDK中的L2fw程序，大致了解一下DPDK应用程序的基本结构。

具体的代码可以参考dpdk-stable-18.11.11/examples/l2fwd/main.c。l2fwd是一个简单的DPDK应用程序，可以看出在main函数中，调用rte_eal_init函数对DPDK运行环境进行初始化之后，调用l2fwd_launch_one_lcore函数在每个CPU的核上都运行l2fwd_main_loop函数。

l2fwd_main_loop函数就是在无限循环中不断的轮询该CPU核绑定的网卡的所有端口，调用rte_eth_tx_buffer_flush和rte_eth_rx_burst进行数据包的发送和接收，进而再完成转发。

被DPDK应用绑定的CPU核不再被Linux内核调度，而是被l2fwd_main_loop函数独占。所以和原来基于Linux内核的实现相比，网络数据包的处理能力得以大幅度提升。 DPDK中的PMD威廉希尔官方网站的具体实现，可以参考dpdk-stable-18.11.11/drivers/net/e1000/igb_ethdev.c中的函数eth_igb_dev_init，PMD相关的实现是DPDK的内部实现，限于篇幅，我们这里就不展开讲解了。

int
main(intargc,char**argv)
{
……
/*initEAL*/
ret=rte_eal_init(argc,argv);
if(ret< 0)
……
/*launchper-lcoreinitoneverylcore*/
rte_eal_mp_remote_launch(l2fwd_launch_one_lcore,NULL,CALL_MASTER);
……
}

staticint
l2fwd_launch_one_lcore(__attribute__((unused))void*dummy)
{
l2fwd_main_loop();
return0;
}

/*mainprocessingloop*/
staticvoid
l2fwd_main_loop(void)
{
structrte_mbuf*pkts_burst[MAX_PKT_BURST];
structrte_mbuf*m;
intsent;
unsignedlcore_id;
uint64_tprev_tsc,diff_tsc,cur_tsc,timer_tsc;
unsignedi,j,portid,nb_rx;
structlcore_queue_conf*qconf;
constuint64_tdrain_tsc=(rte_get_tsc_hz()+US_PER_S-1)/US_PER_S*
BURST_TX_DRAIN_US;
structrte_eth_dev_tx_buffer*buffer;
prev_tsc=0;
timer_tsc=0;
lcore_id=rte_lcore_id();
qconf=&lcore_queue_conf[lcore_id];
if(qconf->n_rx_port==0){
RTE_LOG(INFO,L2FWD,"lcore%uhasnothingtodo ",lcore_id);
return;
}
RTE_LOG(INFO,L2FWD,"enteringmainlooponlcore%u ",lcore_id);
for(i=0;i< qconf->n_rx_port;i++){
portid=qconf->rx_port_list[i];
RTE_LOG(INFO,L2FWD,"--lcoreid=%uportid=%u ",lcore_id,
portid);
}
while(!force_quit){
cur_tsc=rte_rdtsc();
/*
*TXburstqueuedrain
*/
diff_tsc=cur_tsc-prev_tsc;
if(unlikely(diff_tsc>drain_tsc)){
for(i=0;i< qconf->n_rx_port;i++){
portid=l2fwd_dst_ports[qconf->rx_port_list[i]];
buffer=tx_buffer[portid];
sent=rte_eth_tx_buffer_flush(portid,0,buffer);
if(sent)
port_statistics[portid].tx+=sent;
}
/*iftimerisenabled*/
if(timer_period>0){
/*advancethetimer*/
timer_tsc+=diff_tsc;
/*iftimerhasreacheditstimeout*/
if(unlikely(timer_tsc>=timer_period)){
/*dothisonlyonmastercore*/
if(lcore_id==rte_get_master_lcore()){
print_stats();
/*resetthetimer*/
timer_tsc=0;
}
}
}
prev_tsc=cur_tsc;
}
/*
*Readpacketf romRXqueues
*/
for(i=0;i< qconf->n_rx_port;i++){
portid=qconf->rx_port_list[i];
nb_rx=rte_eth_rx_burst(portid,0,
pkts_burst,MAX_PKT_BURST);
port_statistics[portid].rx+=nb_rx;
for(j=0;j< nb_rx; j++) {
m=pkts_burst[j];
rte_prefetch0(rte_pktmbuf_mtod(m,void*));
l2fwd_simple_forward(m,portid);
}
}
}
}

vhost-user与DPDK

在对DPDK应用程序有了一个基本认识，并理解了基本原理和核心威廉希尔官方网站的基础上，下面我们进入到DPDK场景下用户态的virtio-net的实现，也就是vhost-user的相关介绍。

DPDK场景下，virtio-net后端和virtio-net前端的关系图，其中涉及到OVS-DPDK，QEMU，Linux内核和物理网卡硬件。我们接下来会先讲一下各个模块之间的关系，以及它们和virtio-net前端和后端的关系。

-OVS-DPDK：实现了三部分功能。1). 用户空间的PMD，不断地和物理网卡通信，发送接收数据包；2). 虚拟交换机的功能，这部分和我们平时熟悉的OVS是一样的；3). DPDK场景下的virtio-net后端的实现，和virtio-net前端通信。

-QEMU：是一个运行在用户空间的多线程程序，有运行虚拟机的vCPU线程，有处理中断的KVM线程和eventfd。不同之处在于：在DPDK场景下，virtio-net前端是运行在虚拟机线程内的DPDK应用程序。这时，虚拟机的网络接口的类型是vhost-user，它会基于UNIX domain socket和virtio-net后端(实现在OVS-DPDK中)通信。

可以看出，在DPDK场景下，virtio-net前端和后端的实现都在DPDK代码中。

在运行态的时候，就像图2所显示的那样，virtio-net前端是QEMU虚拟机内运行的DPDK应用程序，virtio-net后端运行在OVS-DPDK中。

两者都运行在用户态，通过UNIX domain socket进行通信。vhost-user是将virtio-net驱动在用户态的实现。

下面我们结合结合DPDK 18.11的代码，针对virtio设备的初始化，发送数据的处理流程，做进一步的分析和讲解。

设备发现和初始化

在QEMU虚拟机中运行DPDK应用时：virtio-net前端的初始化的具体实现是在dpdk-stable-18.11.11/drivers/net/virtio/virtio_ethdev.c文件中，其中rte_virtio_pmd是驱动的主要数据结构。

QEMU中的虚拟机运行DPDK应用程序的时候，在指定vhost-user端口的时候，会注册rte_virtio_pmd驱动程序。驱动中的probe函数注册为eth_virtio_pci_probe函数。DPDK驱动程序这样的写法和Linux内核中的PCI网络驱动是一样的，比较容易理解。

函数eth_virtio_dev_init会调用virtio_init_device函数，其中可以看到我们之前介绍过的复合virtio规范的协商特性位(feature bits)，配置MTU等网卡相关的配置，调用virtio_alloc_queues配置虚拟队列等操作。

staticstructrte_pci_driverrte_virtio_pmd={
.driver={
.name="net_virtio",
},
.id_table=pci_id_virtio_map,
.drv_flags=0,
.probe=eth_virtio_pci_probe,
.remove=eth_virtio_pci_remove,
};
RTE_INIT(rte_virtio_pmd_init)
{
rte_eal_iopl_init();
rte_pci_register(&rte_virtio_pmd);
}

staticinteth_virtio_pci_probe(structrte_pci_driver*pci_drv__rte_unused,
structrte_pci_device*pci_dev)
{
if(rte_eal_iopl_init()!=0){
PMD_INIT_LOG(ERR,"IOPLcallfailed-cannotusevirtioPMD");
return1;
}
/*virtiopmdskipsprobeifdeviceneedstoworkinvdpamode*/
if(vdpa_mode_selected(pci_dev->device.devargs))
return1;
returnrte_eth_dev_pci_generic_probe(pci_dev,sizeof(structvirtio_hw),eth_virtio_dev_init);
}

/*resetdeviceandrenegotiatefeaturesifneeded*/
staticint
virtio_init_device(structrte_eth_dev*eth_dev,uint64_treq_features)
{
structvirtio_hw*hw=eth_dev->data->dev_private;
structvirtio_net_config*config;
structvirtio_net_configlocal_config;
structrte_pci_device*pci_dev=NULL;
intret;
/*Resetthedevicealthoughnotnecessaryatstartup*/
vtpci_reset(hw);
if(hw->vqs){
virtio_dev_free_mbufs(eth_dev);
virtio_free_queues(hw);
}
/*Tellthehostwe'venoticedthisdevice.*/
vtpci_set_status(hw,VIRTIO_CONFIG_STATUS_ACK);
/*Tellthehostwe'veknownhowtodrivethedevice.*/
vtpci_set_status(hw,VIRTIO_CONFIG_STATUS_DRIVER);
if(virtio_negotiate_features(hw,req_features)< 0)
return-1;
if(!hw->virtio_user_dev){
pci_dev=RTE_ETH_DEV_TO_PCI(eth_dev);
rte_eth_copy_pci_info(eth_dev,pci_dev);
}
/*IfhostdoesnotsupportbothstatusandMSI-XthendisableLSC*/
if(vtpci_with_feature(hw,VIRTIO_NET_F_STATUS)&&
hw->use_msix!=VIRTIO_MSIX_NONE)
eth_dev->data->dev_flags|=RTE_ETH_DEV_INTR_LSC;
else
eth_dev->data->dev_flags&=~RTE_ETH_DEV_INTR_LSC;
/*Settinguprx_headersizeforthedevice*/
if(vtpci_with_feature(hw,VIRTIO_NET_F_MRG_RXBUF)||
vtpci_with_feature(hw,VIRTIO_F_VERSION_1))
hw->vtnet_hdr_size=sizeof(structvirtio_net_hdr_mrg_rxbuf);
else
hw->vtnet_hdr_size=sizeof(structvirtio_net_hdr);
/*CopythepermanentMACaddressto:virtio_hw*/
virtio_get_hwaddr(hw);
ether_addr_copy((structether_addr*)hw->mac_addr,
ð_dev->data->mac_addrs[0]);
PMD_INIT_LOG(DEBUG,
"PORTMAC:%02X:%02X:%02X:%02X:%02X:%02X",
hw->mac_addr[0],hw->mac_addr[1],hw->mac_addr[2],
hw->mac_addr[3],hw->mac_addr[4],hw->mac_addr[5]);
if(vtpci_with_feature(hw,VIRTIO_NET_F_CTRL_VQ)){
config=&local_config;
vtpci_read_dev_config(hw,
offsetof(structvirtio_net_config,mac),
&config->mac,sizeof(config->mac));
if(vtpci_with_feature(hw,VIRTIO_NET_F_STATUS)){
vtpci_read_dev_config(hw,
offsetof(structvirtio_net_config,status),
&config->status,sizeof(config->status));
}else{
PMD_INIT_LOG(DEBUG,
"VIRTIO_NET_F_STATUSisnotsupported");
config->status=0;
}
if(vtpci_with_feature(hw,VIRTIO_NET_F_MQ)){
vtpci_read_dev_config(hw,
offsetof(structvirtio_net_config,max_virtqueue_pairs),
&config->max_virtqueue_pairs,
sizeof(config->max_virtqueue_pairs));
}else{
PMD_INIT_LOG(DEBUG,
"VIRTIO_NET_F_MQisnotsupported");
config->max_virtqueue_pairs=1;
}
hw->max_queue_pairs=config->max_virtqueue_pairs;
if(vtpci_with_feature(hw,VIRTIO_NET_F_MTU)){
vtpci_read_dev_config(hw,
offsetof(structvirtio_net_config,mtu),
&config->mtu,
sizeof(config->mtu));
/*
*MTUvaluehasalreadybeencheckedatnegotiation
*time,butcheckagainincaseithaschangedsince
*then,whichshouldnothappen.
*/
if(config->mtu< ETHER_MIN_MTU) {
PMD_INIT_LOG(ERR,"invalidmaxMTUvalue(%u)",
config->mtu);
return-1;
}
hw->max_mtu=config->mtu;
/*SetinitialMTUtomaximumonesupportedbyvhost*/
eth_dev->data->mtu=config->mtu;
}else{
hw->max_mtu=VIRTIO_MAX_RX_PKTLEN-ETHER_HDR_LEN-
VLAN_TAG_LEN-hw->vtnet_hdr_size;
}
PMD_INIT_LOG(DEBUG,"config->max_virtqueue_pairs=%d",
config->max_virtqueue_pairs);
PMD_INIT_LOG(DEBUG,"config->status=%d",config->status);
PMD_INIT_LOG(DEBUG,
"PORTMAC:%02X:%02X:%02X:%02X:%02X:%02X",
config->mac[0],config->mac[1],
config->mac[2],config->mac[3],
config->mac[4],config->mac[5]);
}else{
PMD_INIT_LOG(DEBUG,"config->max_virtqueue_pairs=1");
hw->max_queue_pairs=1;
hw->max_mtu=VIRTIO_MAX_RX_PKTLEN-ETHER_HDR_LEN-
VLAN_TAG_LEN-hw->vtnet_hdr_size;
}
ret=virtio_alloc_queues(eth_dev);
if(ret< 0)
returnret;
if(eth_dev->data->dev_conf.intr_conf.rxq){
if(virtio_configure_intr(eth_dev)< 0) {
PMD_INIT_LOG(ERR,"failedtoconfigureinterrupt");
virtio_free_queues(hw);
return-1;
}
}
vtpci_reinit_complete(hw);
if(pci_dev)
PMD_INIT_LOG(DEBUG,"port%dvendorID=0x%xdeviceID=0x%x",
eth_dev->data->port_id,pci_dev->id.vendor_id,
pci_dev->id.device_id);
return0;
}

在OVS-DPDK中添加一个vhost-user网络接口时：OVS-DPDK会调用rte_vhost_driver_register函数，首先根据传入参数path文件路径创建UNIX domain socket，用于后续virtio-net前端(QEMU虚拟机中的DPDK应用程序)和virtio-net后端(OVS-DPDK应用程序)的通信。

这部分相关的源码在dpdk-stable-18.11.11/lib/librte_vhost/socket.c的第824行，主要的函数是rte_vhost_driver_register。

/*
*Registeranewvhost-usersocket;herewecouldactasserver
*(thedefaultcase),orclient(whenRTE_VHOST_USER_CLIENT)flag
*isset.
*/
int
rte_vhost_driver_register(constchar*path,uint64_tflags)
{
intret=-1;
structvhost_user_socket*vsocket;
if(!path)
return-1;
pthread_mutex_lock(&vhost_user.mutex);
if(vhost_user.vsocket_cnt==MAX_VHOST_SOCKET){
RTE_LOG(ERR,VHOST_CONFIG,
"error:thenumberofvhostsocketsreachesmaximum ");
gotoout;
}
vsocket=malloc(sizeof(structvhost_user_socket));
if(!vsocket)
gotoout;
memset(vsocket,0,sizeof(structvhost_user_socket));
vsocket->path=strdup(path);
if(vsocket->path==NULL){
RTE_LOG(ERR,VHOST_CONFIG,
"error:failedtocopysocketpathstring ");
vhost_user_socket_mem_free(vsocket);
gotoout;
}
TAILQ_INIT(&vsocket->conn_list);
ret=pthread_mutex_init(&vsocket->conn_mutex,NULL);
if(ret){
RTE_LOG(ERR,VHOST_CONFIG,
"error:failedtoinitconnectionmutex ");
gotoout_free;
}
vsocket->vdpa_dev_id=-1;
vsocket->dequeue_zero_copy=flags&RTE_VHOST_USER_DEQUEUE_ZERO_COPY;
if(vsocket->dequeue_zero_copy&&
(flags&RTE_VHOST_USER_IOMMU_SUPPORT)){
RTE_LOG(ERR,VHOST_CONFIG,
"error:enablingdequeuezerocopyandIOMMUfeatures"
"simultaneouslyisnotsupported ");
gotoout_mutex;
}
/*
*Setthesupportedfeaturescorrectlyforthebuiltinvhost-user
*netdriver.
*
*Applicationsknownothingaboutfeaturesthebuiltinvirtionet
*driver(virtio_net.c)supports,thusit'snotpossibleforthem
*toinvokerte_vhost_driver_set_features().Toworkaroundit,here
*wesetitunconditionally.Iftheapplicationwanttoimplement
*anothervhost-userdriver(saySCSI),itshouldcallthe
*rte_vhost_driver_set_features(),whichwilloverwritefollowing
*twovalues.
*/
vsocket->use_builtin_virtio_net=true;
vsocket->supported_features=VIRTIO_NET_SUPPORTED_FEATURES;
vsocket->features=VIRTIO_NET_SUPPORTED_FEATURES;
vsocket->protocol_features=VHOST_USER_PROTOCOL_FEATURES;
/*
*Dequeuezerocopycan'tassuredescriptorsreturnedinorder.
*Also,itrequiresthattheguestmemoryispopulated,whichis
*notcompatiblewithpostcopy.
*/
if(vsocket->dequeue_zero_copy){
if((flags&RTE_VHOST_USER_CLIENT)!=0)
RTE_LOG(WARNING,VHOST_CONFIG,
"zerocopymaybeincompatiblewithvhostclientmode ");
vsocket->supported_features&=~(1ULL<< VIRTIO_F_IN_ORDER);
vsocket->features&=~(1ULL<< VIRTIO_F_IN_ORDER);
RTE_LOG(INFO,VHOST_CONFIG,
"Dequeuezerocopyrequested,disablingpostcopysupport ");
vsocket->protocol_features&=
~(1ULL<< VHOST_USER_PROTOCOL_F_PAGEFAULT);
}
if(!(flags&RTE_VHOST_USER_IOMMU_SUPPORT)){
vsocket->supported_features&=~(1ULL<< VIRTIO_F_IOMMU_PLATFORM);
vsocket->features&=~(1ULL<< VIRTIO_F_IOMMU_PLATFORM);
}
if(!(flags&RTE_VHOST_USER_POSTCOPY_SUPPORT)){
vsocket->protocol_features&=
~(1ULL<< VHOST_USER_PROTOCOL_F_PAGEFAULT);
}else{
}
if((flags&RTE_VHOST_USER_CLIENT)!=0){
vsocket->reconnect=!(flags&RTE_VHOST_USER_NO_RECONNECT);
if(vsocket->reconnect&&reconn_tid==0){
if(vhost_user_reconnect_init()!=0)
gotoout_mutex;
}
}else{
vsocket->is_server=true;
}
ret=create_unix_socket(vsocket);
if(ret< 0) {
gotoout_mutex;
}
vhost_user.vsockets[vhost_user.vsocket_cnt++]=vsocket;
pthread_mutex_unlock(&vhost_user.mutex);
returnret;
}

发送数据处理过程

vhost-user前端：在vhost-user接口创建之后，会调用rte_vhost_driver_start函数：首先根据UNIX domainsocket的文件路径path找到对应的socket，然后会调用函数创建监听线程，

这个线程的处理函数是fdset_event_dispatch会去监听vhost_user.fdset指定的socket fd的读写操作，进而实现和vhost-user后端的通信。这部分的实现代码在：dpdk-stable-18.11.11/lib/librte_vhost/socket.c的第1094行。

intrte_vhost_driver_start(constchar*path)
{
structvhost_user_socket*vsocket;
staticpthread_tfdset_tid;
pthread_mutex_lock(&vhost_user.mutex);
vsocket=find_vhost_user_socket(path);
pthread_mutex_unlock(&vhost_user.mutex);
if(!vsocket)
return-1;
if(fdset_tid==0){
/**
*createapipewhichwillbewaitedbypollandnotifiedto
*rebuildthewaitlistofpoll.
*/
if(fdset_pipe_init(&vhost_user.fdset)< 0) {
RTE_LOG(ERR,VHOST_CONFIG,
"failedtocreatepipeforvhostfdset ");
return-1;
}
intret=rte_ctrl_thread_create(&fdset_tid,
"vhost-events",NULL,fdset_event_dispatch,
&vhost_user.fdset);
if(ret!=0){
RTE_LOG(ERR,VHOST_CONFIG,
"failedtocreatefdsethandlingthread");
fdset_pipe_uninit(&vhost_user.fdset);
return-1;
}
}
if(vsocket->is_server)
returnvhost_user_start_server(vsocket);
else
returnvhost_user_start_client(vsocket);
}

vhost-user后端：当vhost-user前端(QEMU虚拟机的DPDK应用程序)向UNIX domain socket写入事件的时候，之前所说的那个监听线程会捕获到这个事件，调用vhost_user_read_cb函数进行处理，最终会调用vhost_user_msg_handler函数，根据不同的消息类型做不同的处理。

这部分的实现代码在：dpdk-stable-18.11.11/lib/librte_vhost/socket.c的第293行。

staticvoid
vhost_user_read_cb(intconnfd,void*dat,int*remove)
{
structvhost_user_connection*conn=dat;
structvhost_user_socket*vsocket=conn->vsocket;
intret;
ret=vhost_user_msg_handler(conn->vid,connfd);
if(ret< 0) {
structvirtio_net*dev=get_device(conn->vid);
close(connfd);
*remove=1;
if(dev)
vhost_destroy_device_notify(dev);
if(vsocket->notify_ops->destroy_connection)
vsocket->notify_ops->destroy_connection(conn->vid);
vhost_destroy_device(conn->vid);
if(vsocket->reconnect){
create_unix_socket(vsocket);
vhost_user_start_client(vsocket);
}
pthread_mutex_lock(&vsocket->conn_mutex);
TAILQ_REMOVE(&vsocket->conn_list,conn,next);
pthread_mutex_unlock(&vsocket->conn_mutex);
free(conn);
}
}

收发数据包的函数分别是eth_vhost_rx和eth_vhost_tx。代码的具体实现在dpdk-stable-18.11.11/drivers/net/vhost/rte_eth_vhost.c中。

函数eth_dev_vhost_create用来创建vhost-user的后端设备的时候，会注册发送和接收数据的回调函数为eth_vhost_rx和eth_vhost_tx。

在DPDK中，发送和接收数据包的代码都是在DPDK中实现的，代码的可读性比较好。只是特别要注意的是同样的DPDK代码，运行的位置不一样：前端是在QEMU虚拟机中运行的DPDK应用程序，后端运行在OVS-DPDK中。

staticint
eth_dev_vhost_create(structrte_vdev_device*dev,char*iface_name,
int16_tqueues,constunsignedintnuma_node,uint64_tflags)
{
constchar*name=rte_vdev_device_name(dev);
structrte_eth_dev_data*data;
structpmd_internal*internal=NULL;
structrte_eth_dev*eth_dev=NULL;
structether_addr*eth_addr=NULL;
structrte_vhost_vring_state*vring_state=NULL;
structinternal_list*list=NULL;
VHOST_LOG(INFO,"CreatingVHOST-USERbackendonnumasocket%u ",
numa_node);
list=rte_zmalloc_socket(name,sizeof(*list),0,numa_node);
if(list==NULL)
gotoerror;
/*reserveanethdeventry*/
eth_dev=rte_eth_vdev_allocate(dev,sizeof(*internal));
if(eth_dev==NULL)
gotoerror;
data=eth_dev->data;
eth_addr=rte_zmalloc_socket(name,sizeof(*eth_addr),0,numa_node);
if(eth_addr==NULL)
gotoerror;
data->mac_addrs=eth_addr;
*eth_addr=base_eth_addr;
eth_addr->addr_bytes[5]=eth_dev->data->port_id;
vring_state=rte_zmalloc_socket(name,
sizeof(*vring_state),0,numa_node);
if(vring_state==NULL)
gotoerror;
/*nowputitalltogether
*-storequeuedataininternal,
*-pointeth_dev_datatointernals
*-andpointeth_devstructuretoneweth_dev_datastructure
*/
internal=eth_dev->data->dev_private;
internal->dev_name=strdup(name);
if(internal->dev_name==NULL)
gotoerror;
internal->iface_name=rte_malloc_socket(name,strlen(iface_name)+1,
0,numa_node);
if(internal->iface_name==NULL)
gotoerror;
strcpy(internal->iface_name,iface_name);
list->eth_dev=eth_dev;
pthread_mutex_lock(&internal_list_lock);
TAILQ_INSERT_TAIL(&internal_list,list,next);
pthread_mutex_unlock(&internal_list_lock);
rte_spinlock_init(&vring_state->lock);
vring_states[eth_dev->data->port_id]=vring_state;
data->nb_rx_queues=queues;
data->nb_tx_queues=queues;
internal->max_queues=queues;
internal->vid=-1;
data->dev_link=pmd_link;
data->dev_flags=RTE_ETH_DEV_INTR_LSC;
eth_dev->dev_ops=&ops;
/*finallyassignrxandtxops*/
eth_dev->rx_pkt_burst=eth_vhost_rx;
eth_dev->tx_pkt_burst=eth_vhost_tx;
if(rte_vhost_driver_register(iface_name,flags))
gotoerror;
if(rte_vhost_driver_callback_register(iface_name,&vhost_ops)< 0) {
VHOST_LOG(ERR,"Can'tregistercallbacks ");
gotoerror;
}
if(rte_vhost_driver_start(iface_name)< 0) {
VHOST_LOG(ERR,"Failedtostartdriverfor%s ",
iface_name);
gotoerror;
}
rte_eth_dev_probing_finish(eth_dev);
return0;
error:
if(internal){
rte_free(internal->iface_name);
free(internal->dev_name);
}
rte_free(vring_state);
rte_eth_dev_release_port(eth_dev);
rte_free(list);
return-1;
}

vDPA

DPDK场景下的vhost-user是virtio-net在用户态的实现。和内核的实现相比，提高了网络数据的处理能力。但毕竟还是纯软件层面的实现，在性能上肯定比不过硬件层面的实现。

为了进一步提升性能，Intel还推出了vDPA (vHost Data Path Acceleration) 的硬件解决方案，直接让网卡与虚拟机内的virtio虚拟队列交互，把数据包DMA到虚拟机的缓存内，在支持了virtio标准的基础上实现了真正意义上的零拷贝。

在DPDK 18.05之后的版本中，已经开始支持vDPA的特性了。vDPA这部分的实现主要是在网卡中，相当于把virtio后端实现在网卡中了。

所以，我们这里只关注一下virtio前端和vDPA设备之间的关联。这部分实现的相关代码在dpdk-stable-18.11.11/examples/vdpa/main.c，第143行start_vdpa函数中的rte_vhost_driver_attach_vdpa_device函数中实现了vhost-user的vsocket数据结构和vDPA设备ID的关联。

staticint
start_vdpa(structvdpa_port*vport)
{
intret;
char*socket_path=vport->ifname;
intdid=vport->did;
if(client_mode)
vport->flags|=RTE_VHOST_USER_CLIENT;
if(access(socket_path,F_OK)!=-1&&!client_mode){
RTE_LOG(ERR,VDPA,
"%sexists,pleaseremoveitorspecifyanotherfileandtryagain. ",
socket_path);
return-1;
}
ret=rte_vhost_driver_register(socket_path,vport->flags);
if(ret!=0)
rte_exit(EXIT_FAILURE,
"registerdriverfailed:%s ",
socket_path);
ret=rte_vhost_driver_callback_register(socket_path,
&vdpa_sample_devops);
if(ret!=0)
rte_exit(EXIT_FAILURE,
"registerdriveropsfailed:%s ",
socket_path);
ret=rte_vhost_driver_attach_vdpa_device(socket_path,did);
if(ret!=0)
rte_exit(EXIT_FAILURE,
"attachvdpadevicefailed:%s ",
socket_path);
if(rte_vhost_driver_start(socket_path)< 0)
rte_exit(EXIT_FAILURE,
"startvhostdriverfailed:%s ",
socket_path);
return0;
}

int
rte_vhost_driver_attach_vdpa_device(constchar*path,intdid)
{
structvhost_user_socket*vsocket;
if(rte_vdpa_get_device(did)==NULL||path==NULL)
return-1;
pthread_mutex_lock(&vhost_user.mutex);
vsocket=find_vhost_user_socket(path);
if(vsocket)
vsocket->vdpa_dev_id=did;
pthread_mutex_unlock(&vhost_user.mutex);
returnvsocket?0:-1;
}

原文标题：virtio威廉希尔官方网站的演进和发展 (2/2)

文章出处：【微信公众号：Linux阅码场】欢迎添加关注！文章转载请注明出处。

责任编辑：haq

声明：本文内容及配图由入驻作者撰写或者入驻合作网站授权转载。文章观点仅代表作者本人，不代表电子发烧友网立场。文章及其配图仅供工程师学习之用，如有内容侵权或者其他违规问题，请联系本站处理。举报投诉