lib/netdev-dpdk.c

   1 /*
   2  * Copyright (c) 2014 Nicira, Inc.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at:
   7  *
   8  *     http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16
  17 #include <config.h>
  18
  19 #include <string.h>
  20 #include <signal.h>
  21 #include <stdlib.h>
  22 #include <pthread.h>
  23 #include <config.h>
  24 #include <errno.h>
  25 #include <sched.h>
  26 #include <stdlib.h>
  27 #include <unistd.h>
  28 #include <sys/stat.h>
  29 #include <stdio.h>
  30 #include <sys/types.h>
  31 #include <sys/stat.h>
  32
  33 #include "dirs.h"
  34 #include "dp-packet.h"
  35 #include "dpif-netdev.h"
  36 #include "list.h"
  37 #include "netdev-dpdk.h"
  38 #include "netdev-provider.h"
  39 #include "netdev-vport.h"
  40 #include "odp-util.h"
  41 #include "ofp-print.h"
  42 #include "ovs-numa.h"
  43 #include "ovs-thread.h"
  44 #include "ovs-rcu.h"
  45 #include "packets.h"
  46 #include "shash.h"
  47 #include "sset.h"
  48 #include "unaligned.h"
  49 #include "timeval.h"
  50 #include "unixctl.h"
  51 #include "openvswitch/vlog.h"
  52
  53 #include "rte_config.h"
  54 #include "rte_mbuf.h"
  55 #include "rte_virtio_net.h"
  56
  57 VLOG_DEFINE_THIS_MODULE(dpdk);
  58 static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(5, 20);
  59
  60 #define DPDK_PORT_WATCHDOG_INTERVAL 5
  61
  62 #define OVS_CACHE_LINE_SIZE CACHE_LINE_SIZE
  63 #define OVS_VPORT_DPDK "ovs_dpdk"
  64
  65 /*
  66  * need to reserve tons of extra space in the mbufs so we can align the
  67  * DMA addresses to 4KB.
  68  */
  69
  70 #define MTU_TO_MAX_LEN(mtu)  ((mtu) + ETHER_HDR_LEN + ETHER_CRC_LEN)
  71 #define MBUF_SIZE(mtu)       (MTU_TO_MAX_LEN(mtu) + (512) + \
  72                              sizeof(struct rte_mbuf) + RTE_PKTMBUF_HEADROOM)
  73
  74 /* Max and min number of packets in the mempool.  OVS tries to allocate a
  75  * mempool with MAX_NB_MBUF: if this fails (because the system doesn't have
  76  * enough hugepages) we keep halving the number until the allocation succeeds
  77  * or we reach MIN_NB_MBUF */
  78
  79 #define MAX_NB_MBUF          (4096 * 64)
  80 #define MIN_NB_MBUF          (4096 * 4)
  81 #define MP_CACHE_SZ          RTE_MEMPOOL_CACHE_MAX_SIZE
  82
  83 /* MAX_NB_MBUF can be divided by 2 many times, until MIN_NB_MBUF */
  84 BUILD_ASSERT_DECL(MAX_NB_MBUF % ROUND_DOWN_POW2(MAX_NB_MBUF/MIN_NB_MBUF) == 0);
  85
  86 /* The smallest possible NB_MBUF that we're going to try should be a multiple
  87  * of MP_CACHE_SZ. This is advised by DPDK documentation. */
  88 BUILD_ASSERT_DECL((MAX_NB_MBUF / ROUND_DOWN_POW2(MAX_NB_MBUF/MIN_NB_MBUF))
  89                   % MP_CACHE_SZ == 0);
  90
  91 #define SOCKET0              0
  92
  93 #define NIC_PORT_RX_Q_SIZE 2048  /* Size of Physical NIC RX Queue, Max (n+32<=4096)*/
  94 #define NIC_PORT_TX_Q_SIZE 2048  /* Size of Physical NIC TX Queue, Max (n+32<=4096)*/
  95
  96 static char *cuse_dev_name = NULL;    /* Character device cuse_dev_name. */
  97 static char *vhost_sock_dir = NULL;   /* Location of vhost-user sockets */
  98
  99 /*
 100  * Maximum amount of time in micro seconds to try and enqueue to vhost.
 101  */
 102 #define VHOST_ENQ_RETRY_USECS 100
 103
 104 static const struct rte_eth_conf port_conf = {
 105     .rxmode = {
 106         .mq_mode = ETH_MQ_RX_RSS,
 107         .split_hdr_size = 0,
 108         .header_split   = 0, /* Header Split disabled */
 109         .hw_ip_checksum = 0, /* IP checksum offload disabled */
 110         .hw_vlan_filter = 0, /* VLAN filtering disabled */
 111         .jumbo_frame    = 0, /* Jumbo Frame Support disabled */
 112         .hw_strip_crc   = 0,
 113     },
 114     .rx_adv_conf = {
 115         .rss_conf = {
 116             .rss_key = NULL,
 117             .rss_hf = ETH_RSS_IP | ETH_RSS_UDP | ETH_RSS_TCP,
 118         },
 119     },
 120     .txmode = {
 121         .mq_mode = ETH_MQ_TX_NONE,
 122     },
 123 };
 124
 125 enum { MAX_TX_QUEUE_LEN = 384 };
 126 enum { DPDK_RING_SIZE = 256 };
 127 BUILD_ASSERT_DECL(IS_POW2(DPDK_RING_SIZE));
 128 enum { DRAIN_TSC = 200000ULL };
 129
 130 enum dpdk_dev_type {
 131     DPDK_DEV_ETH = 0,
 132     DPDK_DEV_VHOST = 1,
 133 };
 134
 135 static int rte_eal_init_ret = ENODEV;
 136
 137 static struct ovs_mutex dpdk_mutex = OVS_MUTEX_INITIALIZER;
 138
 139 /* Contains all 'struct dpdk_dev's. */
 140 static struct ovs_list dpdk_list OVS_GUARDED_BY(dpdk_mutex)
 141     = OVS_LIST_INITIALIZER(&dpdk_list);
 142
 143 static struct ovs_list dpdk_mp_list OVS_GUARDED_BY(dpdk_mutex)
 144     = OVS_LIST_INITIALIZER(&dpdk_mp_list);
 145
 146 /* This mutex must be used by non pmd threads when allocating or freeing
 147  * mbufs through mempools. Since dpdk_queue_pkts() and dpdk_queue_flush() may
 148  * use mempools, a non pmd thread should hold this mutex while calling them */
 149 static struct ovs_mutex nonpmd_mempool_mutex = OVS_MUTEX_INITIALIZER;
 150
 151 struct dpdk_mp {
 152     struct rte_mempool *mp;
 153     int mtu;
 154     int socket_id;
 155     int refcount;
 156     struct ovs_list list_node OVS_GUARDED_BY(dpdk_mutex);
 157 };
 158
 159 /* There should be one 'struct dpdk_tx_queue' created for
 160  * each cpu core. */
 161 struct dpdk_tx_queue {
 162     bool flush_tx;                 /* Set to true to flush queue everytime */
 163                                    /* pkts are queued. */
 164     int count;
 165     rte_spinlock_t tx_lock;        /* Protects the members and the NIC queue
 166                                     * from concurrent access.  It is used only
 167                                     * if the queue is shared among different
 168                                     * pmd threads (see 'txq_needs_locking'). */
 169     uint64_t tsc;
 170     struct rte_mbuf *burst_pkts[MAX_TX_QUEUE_LEN];
 171 };
 172
 173 /* dpdk has no way to remove dpdk ring ethernet devices
 174    so we have to keep them around once they've been created
 175 */
 176
 177 static struct ovs_list dpdk_ring_list OVS_GUARDED_BY(dpdk_mutex)
 178     = OVS_LIST_INITIALIZER(&dpdk_ring_list);
 179
 180 struct dpdk_ring {
 181     /* For the client rings */
 182     struct rte_ring *cring_tx;
 183     struct rte_ring *cring_rx;
 184     int user_port_id; /* User given port no, parsed from port name */
 185     int eth_port_id; /* ethernet device port id */
 186     struct ovs_list list_node OVS_GUARDED_BY(dpdk_mutex);
 187 };
 188
 189 struct netdev_dpdk {
 190     struct netdev up;
 191     int port_id;
 192     int max_packet_len;
 193     enum dpdk_dev_type type;
 194
 195     struct dpdk_tx_queue *tx_q;
 196
 197     struct ovs_mutex mutex OVS_ACQ_AFTER(dpdk_mutex);
 198
 199     struct dpdk_mp *dpdk_mp;
 200     int mtu;
 201     int socket_id;
 202     int buf_size;
 203     struct netdev_stats stats;
 204     /* Protects stats */
 205     rte_spinlock_t stats_lock;
 206
 207     struct eth_addr hwaddr;
 208     enum netdev_flags flags;
 209
 210     struct rte_eth_link link;
 211     int link_reset_cnt;
 212
 213     /* The user might request more txqs than the NIC has.  We remap those
 214      * ('up.n_txq') on these ('real_n_txq').
 215      * If the numbers match, 'txq_needs_locking' is false, otherwise it is
 216      * true and we will take a spinlock on transmission */
 217     int real_n_txq;
 218     bool txq_needs_locking;
 219
 220     /* Spinlock for vhost transmission.  Other DPDK devices use spinlocks in
 221      * dpdk_tx_queue */
 222     rte_spinlock_t vhost_tx_lock;
 223
 224     /* virtio-net structure for vhost device */
 225     OVSRCU_TYPE(struct virtio_net *) virtio_dev;
 226
 227     /* Identifier used to distinguish vhost devices from each other */
 228     char vhost_id[PATH_MAX];
 229
 230     /* In dpdk_list. */
 231     struct ovs_list list_node OVS_GUARDED_BY(dpdk_mutex);
 232 };
 233
 234 struct netdev_rxq_dpdk {
 235     struct netdev_rxq up;
 236     int port_id;
 237 };
 238
 239 static bool thread_is_pmd(void);
 240
 241 static int netdev_dpdk_construct(struct netdev *);
 242
 243 struct virtio_net * netdev_dpdk_get_virtio(const struct netdev_dpdk *dev);
 244
 245 static bool
 246 is_dpdk_class(const struct netdev_class *class)
 247 {
 248     return class->construct == netdev_dpdk_construct;
 249 }
 250
 251 /* XXX: use dpdk malloc for entire OVS. in fact huge page should be used
 252  * for all other segments data, bss and text. */
 253
 254 static void *
 255 dpdk_rte_mzalloc(size_t sz)
 256 {
 257     void *ptr;
 258
 259     ptr = rte_zmalloc(OVS_VPORT_DPDK, sz, OVS_CACHE_LINE_SIZE);
 260     if (ptr == NULL) {
 261         out_of_memory();
 262     }
 263     return ptr;
 264 }
 265
 266 /* XXX this function should be called only by pmd threads (or by non pmd
 267  * threads holding the nonpmd_mempool_mutex) */
 268 void
 269 free_dpdk_buf(struct dp_packet *p)
 270 {
 271     struct rte_mbuf *pkt = (struct rte_mbuf *) p;
 272
 273     rte_pktmbuf_free_seg(pkt);
 274 }
 275
 276 static void
 277 __rte_pktmbuf_init(struct rte_mempool *mp,
 278                    void *opaque_arg OVS_UNUSED,
 279                    void *_m,
 280                    unsigned i OVS_UNUSED)
 281 {
 282     struct rte_mbuf *m = _m;
 283     uint32_t buf_len = mp->elt_size - sizeof(struct dp_packet);
 284
 285     RTE_MBUF_ASSERT(mp->elt_size >= sizeof(struct dp_packet));
 286
 287     memset(m, 0, mp->elt_size);
 288
 289     /* start of buffer is just after mbuf structure */
 290     m->buf_addr = (char *)m + sizeof(struct dp_packet);
 291     m->buf_physaddr = rte_mempool_virt2phy(mp, m) +
 292                     sizeof(struct dp_packet);
 293     m->buf_len = (uint16_t)buf_len;
 294
 295     /* keep some headroom between start of buffer and data */
 296     m->data_off = RTE_MIN(RTE_PKTMBUF_HEADROOM, m->buf_len);
 297
 298     /* init some constant fields */
 299     m->pool = mp;
 300     m->nb_segs = 1;
 301     m->port = 0xff;
 302 }
 303
 304 static void
 305 ovs_rte_pktmbuf_init(struct rte_mempool *mp,
 306                      void *opaque_arg OVS_UNUSED,
 307                      void *_m,
 308                      unsigned i OVS_UNUSED)
 309 {
 310     struct rte_mbuf *m = _m;
 311
 312     __rte_pktmbuf_init(mp, opaque_arg, _m, i);
 313
 314     dp_packet_init_dpdk((struct dp_packet *) m, m->buf_len);
 315 }
 316
 317 static struct dpdk_mp *
 318 dpdk_mp_get(int socket_id, int mtu) OVS_REQUIRES(dpdk_mutex)
 319 {
 320     struct dpdk_mp *dmp = NULL;
 321     char mp_name[RTE_MEMPOOL_NAMESIZE];
 322     unsigned mp_size;
 323
 324     LIST_FOR_EACH (dmp, list_node, &dpdk_mp_list) {
 325         if (dmp->socket_id == socket_id && dmp->mtu == mtu) {
 326             dmp->refcount++;
 327             return dmp;
 328         }
 329     }
 330
 331     dmp = dpdk_rte_mzalloc(sizeof *dmp);
 332     dmp->socket_id = socket_id;
 333     dmp->mtu = mtu;
 334     dmp->refcount = 1;
 335
 336     mp_size = MAX_NB_MBUF;
 337     do {
 338         if (snprintf(mp_name, RTE_MEMPOOL_NAMESIZE, "ovs_mp_%d_%d_%u",
 339                      dmp->mtu, dmp->socket_id, mp_size) < 0) {
 340             return NULL;
 341         }
 342
 343         dmp->mp = rte_mempool_create(mp_name, mp_size, MBUF_SIZE(mtu),
 344                                      MP_CACHE_SZ,
 345                                      sizeof(struct rte_pktmbuf_pool_private),
 346                                      rte_pktmbuf_pool_init, NULL,
 347                                      ovs_rte_pktmbuf_init, NULL,
 348                                      socket_id, 0);
 349     } while (!dmp->mp && rte_errno == ENOMEM && (mp_size /= 2) >= MIN_NB_MBUF);
 350
 351     if (dmp->mp == NULL) {
 352         return NULL;
 353     } else {
 354         VLOG_DBG("Allocated \"%s\" mempool with %u mbufs", mp_name, mp_size );
 355     }
 356
 357     list_push_back(&dpdk_mp_list, &dmp->list_node);
 358     return dmp;
 359 }
 360
 361 static void
 362 dpdk_mp_put(struct dpdk_mp *dmp)
 363 {
 364
 365     if (!dmp) {
 366         return;
 367     }
 368
 369     dmp->refcount--;
 370     ovs_assert(dmp->refcount >= 0);
 371
 372 #if 0
 373     /* I could not find any API to destroy mp. */
 374     if (dmp->refcount == 0) {
 375         list_delete(dmp->list_node);
 376         /* destroy mp-pool. */
 377     }
 378 #endif
 379 }
 380
 381 static void
 382 check_link_status(struct netdev_dpdk *dev)
 383 {
 384     struct rte_eth_link link;
 385
 386     rte_eth_link_get_nowait(dev->port_id, &link);
 387
 388     if (dev->link.link_status != link.link_status) {
 389         netdev_change_seq_changed(&dev->up);
 390
 391         dev->link_reset_cnt++;
 392         dev->link = link;
 393         if (dev->link.link_status) {
 394             VLOG_DBG_RL(&rl, "Port %d Link Up - speed %u Mbps - %s",
 395                         dev->port_id, (unsigned)dev->link.link_speed,
 396                         (dev->link.link_duplex == ETH_LINK_FULL_DUPLEX) ?
 397                          ("full-duplex") : ("half-duplex"));
 398         } else {
 399             VLOG_DBG_RL(&rl, "Port %d Link Down", dev->port_id);
 400         }
 401     }
 402 }
 403
 404 static void *
 405 dpdk_watchdog(void *dummy OVS_UNUSED)
 406 {
 407     struct netdev_dpdk *dev;
 408
 409     pthread_detach(pthread_self());
 410
 411     for (;;) {
 412         ovs_mutex_lock(&dpdk_mutex);
 413         LIST_FOR_EACH (dev, list_node, &dpdk_list) {
 414             ovs_mutex_lock(&dev->mutex);
 415             check_link_status(dev);
 416             ovs_mutex_unlock(&dev->mutex);
 417         }
 418         ovs_mutex_unlock(&dpdk_mutex);
 419         xsleep(DPDK_PORT_WATCHDOG_INTERVAL);
 420     }
 421
 422     return NULL;
 423 }
 424
 425 static int
 426 dpdk_eth_dev_queue_setup(struct netdev_dpdk *dev, int n_rxq, int n_txq)
 427 {
 428     int diag = 0;
 429     int i;
 430
 431     /* A device may report more queues than it makes available (this has
 432      * been observed for Intel xl710, which reserves some of them for
 433      * SRIOV):  rte_eth_*_queue_setup will fail if a queue is not
 434      * available.  When this happens we can retry the configuration
 435      * and request less queues */
 436     while (n_rxq && n_txq) {
 437         if (diag) {
 438             VLOG_INFO("Retrying setup with (rxq:%d txq:%d)", n_rxq, n_txq);
 439         }
 440
 441         diag = rte_eth_dev_configure(dev->port_id, n_rxq, n_txq, &port_conf);
 442         if (diag) {
 443             break;
 444         }
 445
 446         for (i = 0; i < n_txq; i++) {
 447             diag = rte_eth_tx_queue_setup(dev->port_id, i, NIC_PORT_TX_Q_SIZE,
 448                                           dev->socket_id, NULL);
 449             if (diag) {
 450                 VLOG_INFO("Interface %s txq(%d) setup error: %s",
 451                           dev->up.name, i, rte_strerror(-diag));
 452                 break;
 453             }
 454         }
 455
 456         if (i != n_txq) {
 457             /* Retry with less tx queues */
 458             n_txq = i;
 459             continue;
 460         }
 461
 462         for (i = 0; i < n_rxq; i++) {
 463             diag = rte_eth_rx_queue_setup(dev->port_id, i, NIC_PORT_RX_Q_SIZE,
 464                                           dev->socket_id, NULL,
 465                                           dev->dpdk_mp->mp);
 466             if (diag) {
 467                 VLOG_INFO("Interface %s rxq(%d) setup error: %s",
 468                           dev->up.name, i, rte_strerror(-diag));
 469                 break;
 470             }
 471         }
 472
 473         if (i != n_rxq) {
 474             /* Retry with less rx queues */
 475             n_rxq = i;
 476             continue;
 477         }
 478
 479         dev->up.n_rxq = n_rxq;
 480         dev->real_n_txq = n_txq;
 481
 482         return 0;
 483     }
 484
 485     return diag;
 486 }
 487
 488
 489 static int
 490 dpdk_eth_dev_init(struct netdev_dpdk *dev) OVS_REQUIRES(dpdk_mutex)
 491 {
 492     struct rte_pktmbuf_pool_private *mbp_priv;
 493     struct rte_eth_dev_info info;
 494     struct ether_addr eth_addr;
 495     int diag;
 496     int n_rxq, n_txq;
 497
 498     if (dev->port_id < 0 || dev->port_id >= rte_eth_dev_count()) {
 499         return ENODEV;
 500     }
 501
 502     rte_eth_dev_info_get(dev->port_id, &info);
 503
 504     n_rxq = MIN(info.max_rx_queues, dev->up.n_rxq);
 505     n_txq = MIN(info.max_tx_queues, dev->up.n_txq);
 506
 507     diag = dpdk_eth_dev_queue_setup(dev, n_rxq, n_txq);
 508     if (diag) {
 509         VLOG_ERR("Interface %s(rxq:%d txq:%d) configure error: %s",
 510                  dev->up.name, n_rxq, n_txq, rte_strerror(-diag));
 511         return -diag;
 512     }
 513
 514     diag = rte_eth_dev_start(dev->port_id);
 515     if (diag) {
 516         VLOG_ERR("Interface %s start error: %s", dev->up.name,
 517                  rte_strerror(-diag));
 518         return -diag;
 519     }
 520
 521     rte_eth_promiscuous_enable(dev->port_id);
 522     rte_eth_allmulticast_enable(dev->port_id);
 523
 524     memset(&eth_addr, 0x0, sizeof(eth_addr));
 525     rte_eth_macaddr_get(dev->port_id, &eth_addr);
 526     VLOG_INFO_RL(&rl, "Port %d: "ETH_ADDR_FMT"",
 527                     dev->port_id, ETH_ADDR_BYTES_ARGS(eth_addr.addr_bytes));
 528
 529     memcpy(dev->hwaddr.ea, eth_addr.addr_bytes, ETH_ADDR_LEN);
 530     rte_eth_link_get_nowait(dev->port_id, &dev->link);
 531
 532     mbp_priv = rte_mempool_get_priv(dev->dpdk_mp->mp);
 533     dev->buf_size = mbp_priv->mbuf_data_room_size - RTE_PKTMBUF_HEADROOM;
 534
 535     dev->flags = NETDEV_UP | NETDEV_PROMISC;
 536     return 0;
 537 }
 538
 539 static struct netdev_dpdk *
 540 netdev_dpdk_cast(const struct netdev *netdev)
 541 {
 542     return CONTAINER_OF(netdev, struct netdev_dpdk, up);
 543 }
 544
 545 static struct netdev *
 546 netdev_dpdk_alloc(void)
 547 {
 548     struct netdev_dpdk *netdev = dpdk_rte_mzalloc(sizeof *netdev);
 549     return &netdev->up;
 550 }
 551
 552 static void
 553 netdev_dpdk_alloc_txq(struct netdev_dpdk *netdev, unsigned int n_txqs)
 554 {
 555     unsigned i;
 556
 557     netdev->tx_q = dpdk_rte_mzalloc(n_txqs * sizeof *netdev->tx_q);
 558     for (i = 0; i < n_txqs; i++) {
 559         int numa_id = ovs_numa_get_numa_id(i);
 560
 561         if (!netdev->txq_needs_locking) {
 562             /* Each index is considered as a cpu core id, since there should
 563              * be one tx queue for each cpu core.  If the corresponding core
 564              * is not on the same numa node as 'netdev', flags the
 565              * 'flush_tx'. */
 566             netdev->tx_q[i].flush_tx = netdev->socket_id == numa_id;
 567         } else {
 568             /* Queues are shared among CPUs. Always flush */
 569             netdev->tx_q[i].flush_tx = true;
 570         }
 571         rte_spinlock_init(&netdev->tx_q[i].tx_lock);
 572     }
 573 }
 574
 575 static int
 576 netdev_dpdk_init(struct netdev *netdev_, unsigned int port_no,
 577                  enum dpdk_dev_type type)
 578     OVS_REQUIRES(dpdk_mutex)
 579 {
 580     struct netdev_dpdk *netdev = netdev_dpdk_cast(netdev_);
 581     int sid;
 582     int err = 0;
 583
 584     ovs_mutex_init(&netdev->mutex);
 585     ovs_mutex_lock(&netdev->mutex);
 586
 587     rte_spinlock_init(&netdev->stats_lock);
 588
 589     /* If the 'sid' is negative, it means that the kernel fails
 590      * to obtain the pci numa info.  In that situation, always
 591      * use 'SOCKET0'. */
 592     if (type == DPDK_DEV_ETH) {
 593         sid = rte_eth_dev_socket_id(port_no);
 594     } else {
 595         sid = rte_lcore_to_socket_id(rte_get_master_lcore());
 596     }
 597
 598     netdev->socket_id = sid < 0 ? SOCKET0 : sid;
 599     netdev->port_id = port_no;
 600     netdev->type = type;
 601     netdev->flags = 0;
 602     netdev->mtu = ETHER_MTU;
 603     netdev->max_packet_len = MTU_TO_MAX_LEN(netdev->mtu);
 604
 605     netdev->dpdk_mp = dpdk_mp_get(netdev->socket_id, netdev->mtu);
 606     if (!netdev->dpdk_mp) {
 607         err = ENOMEM;
 608         goto unlock;
 609     }
 610
 611     netdev_->n_txq = NR_QUEUE;
 612     netdev_->n_rxq = NR_QUEUE;
 613     netdev->real_n_txq = NR_QUEUE;
 614
 615     if (type == DPDK_DEV_ETH) {
 616         netdev_dpdk_alloc_txq(netdev, NR_QUEUE);
 617         err = dpdk_eth_dev_init(netdev);
 618         if (err) {
 619             goto unlock;
 620         }
 621     }
 622
 623     list_push_back(&dpdk_list, &netdev->list_node);
 624
 625 unlock:
 626     if (err) {
 627         rte_free(netdev->tx_q);
 628     }
 629     ovs_mutex_unlock(&netdev->mutex);
 630     return err;
 631 }
 632
 633 static int
 634 dpdk_dev_parse_name(const char dev_name[], const char prefix[],
 635                     unsigned int *port_no)
 636 {
 637     const char *cport;
 638
 639     if (strncmp(dev_name, prefix, strlen(prefix))) {
 640         return ENODEV;
 641     }
 642
 643     cport = dev_name + strlen(prefix);
 644     *port_no = strtol(cport, NULL, 0); /* string must be null terminated */
 645     return 0;
 646 }
 647
 648 static int
 649 vhost_construct_helper(struct netdev *netdev_) OVS_REQUIRES(dpdk_mutex)
 650 {
 651     struct netdev_dpdk *netdev = netdev_dpdk_cast(netdev_);
 652
 653     if (rte_eal_init_ret) {
 654         return rte_eal_init_ret;
 655     }
 656
 657     rte_spinlock_init(&netdev->vhost_tx_lock);
 658     return netdev_dpdk_init(netdev_, -1, DPDK_DEV_VHOST);
 659 }
 660
 661 static int
 662 netdev_dpdk_vhost_cuse_construct(struct netdev *netdev_)
 663 {
 664     struct netdev_dpdk *netdev = netdev_dpdk_cast(netdev_);
 665     int err;
 666
 667     ovs_mutex_lock(&dpdk_mutex);
 668     strncpy(netdev->vhost_id, netdev->up.name, sizeof(netdev->vhost_id));
 669     err = vhost_construct_helper(netdev_);
 670     ovs_mutex_unlock(&dpdk_mutex);
 671     return err;
 672 }
 673
 674 static int
 675 netdev_dpdk_vhost_user_construct(struct netdev *netdev_)
 676 {
 677     struct netdev_dpdk *netdev = netdev_dpdk_cast(netdev_);
 678     int err;
 679
 680     ovs_mutex_lock(&dpdk_mutex);
 681     /* Take the name of the vhost-user port and append it to the location where
 682      * the socket is to be created, then register the socket.
 683      */
 684     snprintf(netdev->vhost_id, sizeof(netdev->vhost_id), "%s/%s",
 685             vhost_sock_dir, netdev_->name);
 686     err = rte_vhost_driver_register(netdev->vhost_id);
 687     if (err) {
 688         VLOG_ERR("vhost-user socket device setup failure for socket %s\n",
 689                  netdev->vhost_id);
 690     }
 691     VLOG_INFO("Socket %s created for vhost-user port %s\n", netdev->vhost_id, netdev_->name);
 692     err = vhost_construct_helper(netdev_);
 693     ovs_mutex_unlock(&dpdk_mutex);
 694     return err;
 695 }
 696
 697 static int
 698 netdev_dpdk_construct(struct netdev *netdev)
 699 {
 700     unsigned int port_no;
 701     int err;
 702
 703     if (rte_eal_init_ret) {
 704         return rte_eal_init_ret;
 705     }
 706
 707     /* Names always start with "dpdk" */
 708     err = dpdk_dev_parse_name(netdev->name, "dpdk", &port_no);
 709     if (err) {
 710         return err;
 711     }
 712
 713     ovs_mutex_lock(&dpdk_mutex);
 714     err = netdev_dpdk_init(netdev, port_no, DPDK_DEV_ETH);
 715     ovs_mutex_unlock(&dpdk_mutex);
 716     return err;
 717 }
 718
 719 static void
 720 netdev_dpdk_destruct(struct netdev *netdev_)
 721 {
 722     struct netdev_dpdk *dev = netdev_dpdk_cast(netdev_);
 723
 724     ovs_mutex_lock(&dev->mutex);
 725     rte_eth_dev_stop(dev->port_id);
 726     ovs_mutex_unlock(&dev->mutex);
 727
 728     ovs_mutex_lock(&dpdk_mutex);
 729     rte_free(dev->tx_q);
 730     list_remove(&dev->list_node);
 731     dpdk_mp_put(dev->dpdk_mp);
 732     ovs_mutex_unlock(&dpdk_mutex);
 733 }
 734
 735 static void
 736 netdev_dpdk_vhost_destruct(struct netdev *netdev_)
 737 {
 738     struct netdev_dpdk *dev = netdev_dpdk_cast(netdev_);
 739
 740     /* Can't remove a port while a guest is attached to it. */
 741     if (netdev_dpdk_get_virtio(dev) != NULL) {
 742         VLOG_ERR("Can not remove port, vhost device still attached");
 743                 return;
 744     }
 745
 746     ovs_mutex_lock(&dpdk_mutex);
 747     list_remove(&dev->list_node);
 748     dpdk_mp_put(dev->dpdk_mp);
 749     ovs_mutex_unlock(&dpdk_mutex);
 750 }
 751
 752 static void
 753 netdev_dpdk_dealloc(struct netdev *netdev_)
 754 {
 755     struct netdev_dpdk *netdev = netdev_dpdk_cast(netdev_);
 756
 757     rte_free(netdev);
 758 }
 759
 760 static int
 761 netdev_dpdk_get_config(const struct netdev *netdev_, struct smap *args)
 762 {
 763     struct netdev_dpdk *dev = netdev_dpdk_cast(netdev_);
 764
 765     ovs_mutex_lock(&dev->mutex);
 766
 767     smap_add_format(args, "configured_rx_queues", "%d", netdev_->n_rxq);
 768     smap_add_format(args, "requested_tx_queues", "%d", netdev_->n_txq);
 769     smap_add_format(args, "configured_tx_queues", "%d", dev->real_n_txq);
 770     ovs_mutex_unlock(&dev->mutex);
 771
 772     return 0;
 773 }
 774
 775 static int
 776 netdev_dpdk_get_numa_id(const struct netdev *netdev_)
 777 {
 778     struct netdev_dpdk *netdev = netdev_dpdk_cast(netdev_);
 779
 780     return netdev->socket_id;
 781 }
 782
 783 /* Sets the number of tx queues and rx queues for the dpdk interface.
 784  * If the configuration fails, do not try restoring its old configuration
 785  * and just returns the error. */
 786 static int
 787 netdev_dpdk_set_multiq(struct netdev *netdev_, unsigned int n_txq,
 788                        unsigned int n_rxq)
 789 {
 790     struct netdev_dpdk *netdev = netdev_dpdk_cast(netdev_);
 791     int err = 0;
 792     int old_rxq, old_txq;
 793
 794     if (netdev->up.n_txq == n_txq && netdev->up.n_rxq == n_rxq) {
 795         return err;
 796     }
 797
 798     ovs_mutex_lock(&dpdk_mutex);
 799     ovs_mutex_lock(&netdev->mutex);
 800
 801     rte_eth_dev_stop(netdev->port_id);
 802
 803     old_txq = netdev->up.n_txq;
 804     old_rxq = netdev->up.n_rxq;
 805     netdev->up.n_txq = n_txq;
 806     netdev->up.n_rxq = n_rxq;
 807
 808     rte_free(netdev->tx_q);
 809     err = dpdk_eth_dev_init(netdev);
 810     netdev_dpdk_alloc_txq(netdev, netdev->real_n_txq);
 811     if (err) {
 812         /* If there has been an error, it means that the requested queues
 813          * have not been created.  Restore the old numbers. */
 814         netdev->up.n_txq = old_txq;
 815         netdev->up.n_rxq = old_rxq;
 816     }
 817
 818     netdev->txq_needs_locking = netdev->real_n_txq != netdev->up.n_txq;
 819
 820     ovs_mutex_unlock(&netdev->mutex);
 821     ovs_mutex_unlock(&dpdk_mutex);
 822
 823     return err;
 824 }
 825
 826 static int
 827 netdev_dpdk_vhost_set_multiq(struct netdev *netdev_, unsigned int n_txq,
 828                              unsigned int n_rxq)
 829 {
 830     struct netdev_dpdk *netdev = netdev_dpdk_cast(netdev_);
 831     int err = 0;
 832
 833     if (netdev->up.n_txq == n_txq && netdev->up.n_rxq == n_rxq) {
 834         return err;
 835     }
 836
 837     ovs_mutex_lock(&dpdk_mutex);
 838     ovs_mutex_lock(&netdev->mutex);
 839
 840     netdev->up.n_txq = n_txq;
 841     netdev->real_n_txq = 1;
 842     netdev->up.n_rxq = 1;
 843
 844     ovs_mutex_unlock(&netdev->mutex);
 845     ovs_mutex_unlock(&dpdk_mutex);
 846
 847     return err;
 848 }
 849
 850 static struct netdev_rxq *
 851 netdev_dpdk_rxq_alloc(void)
 852 {
 853     struct netdev_rxq_dpdk *rx = dpdk_rte_mzalloc(sizeof *rx);
 854
 855     return &rx->up;
 856 }
 857
 858 static struct netdev_rxq_dpdk *
 859 netdev_rxq_dpdk_cast(const struct netdev_rxq *rx)
 860 {
 861     return CONTAINER_OF(rx, struct netdev_rxq_dpdk, up);
 862 }
 863
 864 static int
 865 netdev_dpdk_rxq_construct(struct netdev_rxq *rxq_)
 866 {
 867     struct netdev_rxq_dpdk *rx = netdev_rxq_dpdk_cast(rxq_);
 868     struct netdev_dpdk *netdev = netdev_dpdk_cast(rx->up.netdev);
 869
 870     ovs_mutex_lock(&netdev->mutex);
 871     rx->port_id = netdev->port_id;
 872     ovs_mutex_unlock(&netdev->mutex);
 873
 874     return 0;
 875 }
 876
 877 static void
 878 netdev_dpdk_rxq_destruct(struct netdev_rxq *rxq_ OVS_UNUSED)
 879 {
 880 }
 881
 882 static void
 883 netdev_dpdk_rxq_dealloc(struct netdev_rxq *rxq_)
 884 {
 885     struct netdev_rxq_dpdk *rx = netdev_rxq_dpdk_cast(rxq_);
 886
 887     rte_free(rx);
 888 }
 889
 890 static inline void
 891 dpdk_queue_flush__(struct netdev_dpdk *dev, int qid)
 892 {
 893     struct dpdk_tx_queue *txq = &dev->tx_q[qid];
 894     uint32_t nb_tx = 0;
 895
 896     while (nb_tx != txq->count) {
 897         uint32_t ret;
 898
 899         ret = rte_eth_tx_burst(dev->port_id, qid, txq->burst_pkts + nb_tx,
 900                                txq->count - nb_tx);
 901         if (!ret) {
 902             break;
 903         }
 904
 905         nb_tx += ret;
 906     }
 907
 908     if (OVS_UNLIKELY(nb_tx != txq->count)) {
 909         /* free buffers, which we couldn't transmit, one at a time (each
 910          * packet could come from a different mempool) */
 911         int i;
 912
 913         for (i = nb_tx; i < txq->count; i++) {
 914             rte_pktmbuf_free_seg(txq->burst_pkts[i]);
 915         }
 916         rte_spinlock_lock(&dev->stats_lock);
 917         dev->stats.tx_dropped += txq->count-nb_tx;
 918         rte_spinlock_unlock(&dev->stats_lock);
 919     }
 920
 921     txq->count = 0;
 922     txq->tsc = rte_get_timer_cycles();
 923 }
 924
 925 static inline void
 926 dpdk_queue_flush(struct netdev_dpdk *dev, int qid)
 927 {
 928     struct dpdk_tx_queue *txq = &dev->tx_q[qid];
 929
 930     if (txq->count == 0) {
 931         return;
 932     }
 933     dpdk_queue_flush__(dev, qid);
 934 }
 935
 936 static bool
 937 is_vhost_running(struct virtio_net *dev)
 938 {
 939     return (dev != NULL && (dev->flags & VIRTIO_DEV_RUNNING));
 940 }
 941
 942 /*
 943  * The receive path for the vhost port is the TX path out from guest.
 944  */
 945 static int
 946 netdev_dpdk_vhost_rxq_recv(struct netdev_rxq *rxq_,
 947                            struct dp_packet **packets, int *c)
 948 {
 949     struct netdev_rxq_dpdk *rx = netdev_rxq_dpdk_cast(rxq_);
 950     struct netdev *netdev = rx->up.netdev;
 951     struct netdev_dpdk *vhost_dev = netdev_dpdk_cast(netdev);
 952     struct virtio_net *virtio_dev = netdev_dpdk_get_virtio(vhost_dev);
 953     int qid = 1;
 954     uint16_t nb_rx = 0;
 955
 956     if (OVS_UNLIKELY(!is_vhost_running(virtio_dev))) {
 957         return EAGAIN;
 958     }
 959
 960     nb_rx = rte_vhost_dequeue_burst(virtio_dev, qid,
 961                                     vhost_dev->dpdk_mp->mp,
 962                                     (struct rte_mbuf **)packets,
 963                                     NETDEV_MAX_BURST);
 964     if (!nb_rx) {
 965         return EAGAIN;
 966     }
 967
 968     rte_spinlock_lock(&vhost_dev->stats_lock);
 969     vhost_dev->stats.rx_packets += (uint64_t)nb_rx;
 970     rte_spinlock_unlock(&vhost_dev->stats_lock);
 971
 972     *c = (int) nb_rx;
 973     return 0;
 974 }
 975
 976 static int
 977 netdev_dpdk_rxq_recv(struct netdev_rxq *rxq_, struct dp_packet **packets,
 978                      int *c)
 979 {
 980     struct netdev_rxq_dpdk *rx = netdev_rxq_dpdk_cast(rxq_);
 981     struct netdev *netdev = rx->up.netdev;
 982     struct netdev_dpdk *dev = netdev_dpdk_cast(netdev);
 983     int nb_rx;
 984
 985     /* There is only one tx queue for this core.  Do not flush other
 986      * queues.
 987      * Do not flush tx queue which is shared among CPUs
 988      * since it is always flushed */
 989     if (rxq_->queue_id == rte_lcore_id() &&
 990         OVS_LIKELY(!dev->txq_needs_locking)) {
 991         dpdk_queue_flush(dev, rxq_->queue_id);
 992     }
 993
 994     nb_rx = rte_eth_rx_burst(rx->port_id, rxq_->queue_id,
 995                              (struct rte_mbuf **) packets,
 996                              NETDEV_MAX_BURST);
 997     if (!nb_rx) {
 998         return EAGAIN;
 999     }
1000
1001     *c = nb_rx;
1002
1003     return 0;
1004 }
1005
1006 static void
1007 __netdev_dpdk_vhost_send(struct netdev *netdev, struct dp_packet **pkts,
1008                          int cnt, bool may_steal)
1009 {
1010     struct netdev_dpdk *vhost_dev = netdev_dpdk_cast(netdev);
1011     struct virtio_net *virtio_dev = netdev_dpdk_get_virtio(vhost_dev);
1012     struct rte_mbuf **cur_pkts = (struct rte_mbuf **) pkts;
1013     unsigned int total_pkts = cnt;
1014     uint64_t start = 0;
1015
1016     if (OVS_UNLIKELY(!is_vhost_running(virtio_dev))) {
1017         rte_spinlock_lock(&vhost_dev->stats_lock);
1018         vhost_dev->stats.tx_dropped+= cnt;
1019         rte_spinlock_unlock(&vhost_dev->stats_lock);
1020         goto out;
1021     }
1022
1023     /* There is vHost TX single queue, So we need to lock it for TX. */
1024     rte_spinlock_lock(&vhost_dev->vhost_tx_lock);
1025
1026     do {
1027         unsigned int tx_pkts;
1028
1029         tx_pkts = rte_vhost_enqueue_burst(virtio_dev, VIRTIO_RXQ,
1030                                           cur_pkts, cnt);
1031         if (OVS_LIKELY(tx_pkts)) {
1032             /* Packets have been sent.*/
1033             cnt -= tx_pkts;
1034             /* Prepare for possible next iteration.*/
1035             cur_pkts = &cur_pkts[tx_pkts];
1036         } else {
1037             uint64_t timeout = VHOST_ENQ_RETRY_USECS * rte_get_timer_hz() / 1E6;
1038             unsigned int expired = 0;
1039
1040             if (!start) {
1041                 start = rte_get_timer_cycles();
1042             }
1043
1044             /*
1045              * Unable to enqueue packets to vhost interface.
1046              * Check available entries before retrying.
1047              */
1048             while (!rte_vring_available_entries(virtio_dev, VIRTIO_RXQ)) {
1049                 if (OVS_UNLIKELY((rte_get_timer_cycles() - start) > timeout)) {
1050                     expired = 1;
1051                     break;
1052                 }
1053             }
1054             if (expired) {
1055                 /* break out of main loop. */
1056                 break;
1057             }
1058         }
1059     } while (cnt);
1060     rte_spinlock_unlock(&vhost_dev->vhost_tx_lock);
1061
1062     rte_spinlock_lock(&vhost_dev->stats_lock);
1063     vhost_dev->stats.tx_packets += (total_pkts - cnt);
1064     vhost_dev->stats.tx_dropped += cnt;
1065     rte_spinlock_unlock(&vhost_dev->stats_lock);
1066
1067 out:
1068     if (may_steal) {
1069         int i;
1070
1071         for (i = 0; i < total_pkts; i++) {
1072             dp_packet_delete(pkts[i]);
1073         }
1074     }
1075 }
1076
1077 inline static void
1078 dpdk_queue_pkts(struct netdev_dpdk *dev, int qid,
1079                struct rte_mbuf **pkts, int cnt)
1080 {
1081     struct dpdk_tx_queue *txq = &dev->tx_q[qid];
1082     uint64_t diff_tsc;
1083
1084     int i = 0;
1085
1086     while (i < cnt) {
1087         int freeslots = MAX_TX_QUEUE_LEN - txq->count;
1088         int tocopy = MIN(freeslots, cnt-i);
1089
1090         memcpy(&txq->burst_pkts[txq->count], &pkts[i],
1091                tocopy * sizeof (struct rte_mbuf *));
1092
1093         txq->count += tocopy;
1094         i += tocopy;
1095
1096         if (txq->count == MAX_TX_QUEUE_LEN || txq->flush_tx) {
1097             dpdk_queue_flush__(dev, qid);
1098         }
1099         diff_tsc = rte_get_timer_cycles() - txq->tsc;
1100         if (diff_tsc >= DRAIN_TSC) {
1101             dpdk_queue_flush__(dev, qid);
1102         }
1103     }
1104 }
1105
1106 /* Tx function. Transmit packets indefinitely */
1107 static void
1108 dpdk_do_tx_copy(struct netdev *netdev, int qid, struct dp_packet **pkts,
1109                 int cnt)
1110     OVS_NO_THREAD_SAFETY_ANALYSIS
1111 {
1112 #if !defined(__CHECKER__) && !defined(_WIN32)
1113     const size_t PKT_ARRAY_SIZE = cnt;
1114 #else
1115     /* Sparse or MSVC doesn't like variable length array. */
1116     enum { PKT_ARRAY_SIZE = NETDEV_MAX_BURST };
1117 #endif
1118     struct netdev_dpdk *dev = netdev_dpdk_cast(netdev);
1119     struct rte_mbuf *mbufs[PKT_ARRAY_SIZE];
1120     int dropped = 0;
1121     int newcnt = 0;
1122     int i;
1123
1124     /* If we are on a non pmd thread we have to use the mempool mutex, because
1125      * every non pmd thread shares the same mempool cache */
1126
1127     if (!thread_is_pmd()) {
1128         ovs_mutex_lock(&nonpmd_mempool_mutex);
1129     }
1130
1131     for (i = 0; i < cnt; i++) {
1132         int size = dp_packet_size(pkts[i]);
1133
1134         if (OVS_UNLIKELY(size > dev->max_packet_len)) {
1135             VLOG_WARN_RL(&rl, "Too big size %d max_packet_len %d",
1136                          (int)size , dev->max_packet_len);
1137
1138             dropped++;
1139             continue;
1140         }
1141
1142         mbufs[newcnt] = rte_pktmbuf_alloc(dev->dpdk_mp->mp);
1143
1144         if (!mbufs[newcnt]) {
1145             dropped += cnt - i;
1146             break;
1147         }
1148
1149         /* We have to do a copy for now */
1150         memcpy(rte_pktmbuf_mtod(mbufs[newcnt], void *), dp_packet_data(pkts[i]), size);
1151
1152         rte_pktmbuf_data_len(mbufs[newcnt]) = size;
1153         rte_pktmbuf_pkt_len(mbufs[newcnt]) = size;
1154
1155         newcnt++;
1156     }
1157
1158     if (OVS_UNLIKELY(dropped)) {
1159         rte_spinlock_lock(&dev->stats_lock);
1160         dev->stats.tx_dropped += dropped;
1161         rte_spinlock_unlock(&dev->stats_lock);
1162     }
1163
1164     if (dev->type == DPDK_DEV_VHOST) {
1165         __netdev_dpdk_vhost_send(netdev, (struct dp_packet **) mbufs, newcnt, true);
1166     } else {
1167         dpdk_queue_pkts(dev, qid, mbufs, newcnt);
1168         dpdk_queue_flush(dev, qid);
1169     }
1170
1171     if (!thread_is_pmd()) {
1172         ovs_mutex_unlock(&nonpmd_mempool_mutex);
1173     }
1174 }
1175
1176 static int
1177 netdev_dpdk_vhost_send(struct netdev *netdev, int qid OVS_UNUSED, struct dp_packet **pkts,
1178                  int cnt, bool may_steal)
1179 {
1180     if (OVS_UNLIKELY(pkts[0]->source != DPBUF_DPDK)) {
1181         int i;
1182
1183         dpdk_do_tx_copy(netdev, qid, pkts, cnt);
1184         if (may_steal) {
1185             for (i = 0; i < cnt; i++) {
1186                 dp_packet_delete(pkts[i]);
1187             }
1188         }
1189     } else {
1190         __netdev_dpdk_vhost_send(netdev, pkts, cnt, may_steal);
1191     }
1192     return 0;
1193 }
1194
1195 static inline void
1196 netdev_dpdk_send__(struct netdev_dpdk *dev, int qid,
1197                    struct dp_packet **pkts, int cnt, bool may_steal)
1198 {
1199     int i;
1200
1201     if (OVS_UNLIKELY(dev->txq_needs_locking)) {
1202         qid = qid % dev->real_n_txq;
1203         rte_spinlock_lock(&dev->tx_q[qid].tx_lock);
1204     }
1205
1206     if (OVS_UNLIKELY(!may_steal ||
1207                      pkts[0]->source != DPBUF_DPDK)) {
1208         struct netdev *netdev = &dev->up;
1209
1210         dpdk_do_tx_copy(netdev, qid, pkts, cnt);
1211
1212         if (may_steal) {
1213             for (i = 0; i < cnt; i++) {
1214                 dp_packet_delete(pkts[i]);
1215             }
1216         }
1217     } else {
1218         int next_tx_idx = 0;
1219         int dropped = 0;
1220
1221         for (i = 0; i < cnt; i++) {
1222             int size = dp_packet_size(pkts[i]);
1223
1224             if (OVS_UNLIKELY(size > dev->max_packet_len)) {
1225                 if (next_tx_idx != i) {
1226                     dpdk_queue_pkts(dev, qid,
1227                                     (struct rte_mbuf **)&pkts[next_tx_idx],
1228                                     i-next_tx_idx);
1229                 }
1230
1231                 VLOG_WARN_RL(&rl, "Too big size %d max_packet_len %d",
1232                              (int)size , dev->max_packet_len);
1233
1234                 dp_packet_delete(pkts[i]);
1235                 dropped++;
1236                 next_tx_idx = i + 1;
1237             }
1238         }
1239         if (next_tx_idx != cnt) {
1240            dpdk_queue_pkts(dev, qid,
1241                             (struct rte_mbuf **)&pkts[next_tx_idx],
1242                             cnt-next_tx_idx);
1243         }
1244
1245         if (OVS_UNLIKELY(dropped)) {
1246             rte_spinlock_lock(&dev->stats_lock);
1247             dev->stats.tx_dropped += dropped;
1248             rte_spinlock_unlock(&dev->stats_lock);
1249         }
1250     }
1251
1252     if (OVS_UNLIKELY(dev->txq_needs_locking)) {
1253         rte_spinlock_unlock(&dev->tx_q[qid].tx_lock);
1254     }
1255 }
1256
1257 static int
1258 netdev_dpdk_eth_send(struct netdev *netdev, int qid,
1259                      struct dp_packet **pkts, int cnt, bool may_steal)
1260 {
1261     struct netdev_dpdk *dev = netdev_dpdk_cast(netdev);
1262
1263     netdev_dpdk_send__(dev, qid, pkts, cnt, may_steal);
1264     return 0;
1265 }
1266
1267 static int
1268 netdev_dpdk_set_etheraddr(struct netdev *netdev, const struct eth_addr mac)
1269 {
1270     struct netdev_dpdk *dev = netdev_dpdk_cast(netdev);
1271
1272     ovs_mutex_lock(&dev->mutex);
1273     if (!eth_addr_equals(dev->hwaddr, mac)) {
1274         dev->hwaddr = mac;
1275         netdev_change_seq_changed(netdev);
1276     }
1277     ovs_mutex_unlock(&dev->mutex);
1278
1279     return 0;
1280 }
1281
1282 static int
1283 netdev_dpdk_get_etheraddr(const struct netdev *netdev, struct eth_addr *mac)
1284 {
1285     struct netdev_dpdk *dev = netdev_dpdk_cast(netdev);
1286
1287     ovs_mutex_lock(&dev->mutex);
1288     *mac = dev->hwaddr;
1289     ovs_mutex_unlock(&dev->mutex);
1290
1291     return 0;
1292 }
1293
1294 static int
1295 netdev_dpdk_get_mtu(const struct netdev *netdev, int *mtup)
1296 {
1297     struct netdev_dpdk *dev = netdev_dpdk_cast(netdev);
1298
1299     ovs_mutex_lock(&dev->mutex);
1300     *mtup = dev->mtu;
1301     ovs_mutex_unlock(&dev->mutex);
1302
1303     return 0;
1304 }
1305
1306 static int
1307 netdev_dpdk_set_mtu(const struct netdev *netdev, int mtu)
1308 {
1309     struct netdev_dpdk *dev = netdev_dpdk_cast(netdev);
1310     int old_mtu, err;
1311     struct dpdk_mp *old_mp;
1312     struct dpdk_mp *mp;
1313
1314     ovs_mutex_lock(&dpdk_mutex);
1315     ovs_mutex_lock(&dev->mutex);
1316     if (dev->mtu == mtu) {
1317         err = 0;
1318         goto out;
1319     }
1320
1321     mp = dpdk_mp_get(dev->socket_id, dev->mtu);
1322     if (!mp) {
1323         err = ENOMEM;
1324         goto out;
1325     }
1326
1327     rte_eth_dev_stop(dev->port_id);
1328
1329     old_mtu = dev->mtu;
1330     old_mp = dev->dpdk_mp;
1331     dev->dpdk_mp = mp;
1332     dev->mtu = mtu;
1333     dev->max_packet_len = MTU_TO_MAX_LEN(dev->mtu);
1334
1335     err = dpdk_eth_dev_init(dev);
1336     if (err) {
1337         dpdk_mp_put(mp);
1338         dev->mtu = old_mtu;
1339         dev->dpdk_mp = old_mp;
1340         dev->max_packet_len = MTU_TO_MAX_LEN(dev->mtu);
1341         dpdk_eth_dev_init(dev);
1342         goto out;
1343     }
1344
1345     dpdk_mp_put(old_mp);
1346     netdev_change_seq_changed(netdev);
1347 out:
1348     ovs_mutex_unlock(&dev->mutex);
1349     ovs_mutex_unlock(&dpdk_mutex);
1350     return err;
1351 }
1352
1353 static int
1354 netdev_dpdk_get_carrier(const struct netdev *netdev_, bool *carrier);
1355
1356 static int
1357 netdev_dpdk_vhost_get_stats(const struct netdev *netdev,
1358                             struct netdev_stats *stats)
1359 {
1360     struct netdev_dpdk *dev = netdev_dpdk_cast(netdev);
1361
1362     ovs_mutex_lock(&dev->mutex);
1363     memset(stats, 0, sizeof(*stats));
1364     /* Unsupported Stats */
1365     stats->rx_errors = UINT64_MAX;
1366     stats->tx_errors = UINT64_MAX;
1367     stats->multicast = UINT64_MAX;
1368     stats->collisions = UINT64_MAX;
1369     stats->rx_crc_errors = UINT64_MAX;
1370     stats->rx_fifo_errors = UINT64_MAX;
1371     stats->rx_frame_errors = UINT64_MAX;
1372     stats->rx_length_errors = UINT64_MAX;
1373     stats->rx_missed_errors = UINT64_MAX;
1374     stats->rx_over_errors = UINT64_MAX;
1375     stats->tx_aborted_errors = UINT64_MAX;
1376     stats->tx_carrier_errors = UINT64_MAX;
1377     stats->tx_errors = UINT64_MAX;
1378     stats->tx_fifo_errors = UINT64_MAX;
1379     stats->tx_heartbeat_errors = UINT64_MAX;
1380     stats->tx_window_errors = UINT64_MAX;
1381     stats->rx_bytes += UINT64_MAX;
1382     stats->rx_dropped += UINT64_MAX;
1383     stats->tx_bytes += UINT64_MAX;
1384
1385     rte_spinlock_lock(&dev->stats_lock);
1386     /* Supported Stats */
1387     stats->rx_packets += dev->stats.rx_packets;
1388     stats->tx_packets += dev->stats.tx_packets;
1389     stats->tx_dropped += dev->stats.tx_dropped;
1390     rte_spinlock_unlock(&dev->stats_lock);
1391     ovs_mutex_unlock(&dev->mutex);
1392
1393     return 0;
1394 }
1395
1396 static int
1397 netdev_dpdk_get_stats(const struct netdev *netdev, struct netdev_stats *stats)
1398 {
1399     struct netdev_dpdk *dev = netdev_dpdk_cast(netdev);
1400     struct rte_eth_stats rte_stats;
1401     bool gg;
1402
1403     netdev_dpdk_get_carrier(netdev, &gg);
1404     ovs_mutex_lock(&dev->mutex);
1405     rte_eth_stats_get(dev->port_id, &rte_stats);
1406
1407     memset(stats, 0, sizeof(*stats));
1408
1409     stats->rx_packets = rte_stats.ipackets;
1410     stats->tx_packets = rte_stats.opackets;
1411     stats->rx_bytes = rte_stats.ibytes;
1412     stats->tx_bytes = rte_stats.obytes;
1413     stats->rx_errors = rte_stats.ierrors;
1414     stats->tx_errors = rte_stats.oerrors;
1415     stats->multicast = rte_stats.imcasts;
1416
1417     rte_spinlock_lock(&dev->stats_lock);
1418     stats->tx_dropped = dev->stats.tx_dropped;
1419     rte_spinlock_unlock(&dev->stats_lock);
1420     ovs_mutex_unlock(&dev->mutex);
1421
1422     return 0;
1423 }
1424
1425 static int
1426 netdev_dpdk_get_features(const struct netdev *netdev_,
1427                          enum netdev_features *current,
1428                          enum netdev_features *advertised OVS_UNUSED,
1429                          enum netdev_features *supported OVS_UNUSED,
1430                          enum netdev_features *peer OVS_UNUSED)
1431 {
1432     struct netdev_dpdk *dev = netdev_dpdk_cast(netdev_);
1433     struct rte_eth_link link;
1434
1435     ovs_mutex_lock(&dev->mutex);
1436     link = dev->link;
1437     ovs_mutex_unlock(&dev->mutex);
1438
1439     if (link.link_duplex == ETH_LINK_AUTONEG_DUPLEX) {
1440         if (link.link_speed == ETH_LINK_SPEED_AUTONEG) {
1441             *current = NETDEV_F_AUTONEG;
1442         }
1443     } else if (link.link_duplex == ETH_LINK_HALF_DUPLEX) {
1444         if (link.link_speed == ETH_LINK_SPEED_10) {
1445             *current = NETDEV_F_10MB_HD;
1446         }
1447         if (link.link_speed == ETH_LINK_SPEED_100) {
1448             *current = NETDEV_F_100MB_HD;
1449         }
1450         if (link.link_speed == ETH_LINK_SPEED_1000) {
1451             *current = NETDEV_F_1GB_HD;
1452         }
1453     } else if (link.link_duplex == ETH_LINK_FULL_DUPLEX) {
1454         if (link.link_speed == ETH_LINK_SPEED_10) {
1455             *current = NETDEV_F_10MB_FD;
1456         }
1457         if (link.link_speed == ETH_LINK_SPEED_100) {
1458             *current = NETDEV_F_100MB_FD;
1459         }
1460         if (link.link_speed == ETH_LINK_SPEED_1000) {
1461             *current = NETDEV_F_1GB_FD;
1462         }
1463         if (link.link_speed == ETH_LINK_SPEED_10000) {
1464             *current = NETDEV_F_10GB_FD;
1465         }
1466     }
1467
1468     return 0;
1469 }
1470
1471 static int
1472 netdev_dpdk_get_ifindex(const struct netdev *netdev)
1473 {
1474     struct netdev_dpdk *dev = netdev_dpdk_cast(netdev);
1475     int ifindex;
1476
1477     ovs_mutex_lock(&dev->mutex);
1478     ifindex = dev->port_id;
1479     ovs_mutex_unlock(&dev->mutex);
1480
1481     return ifindex;
1482 }
1483
1484 static int
1485 netdev_dpdk_get_carrier(const struct netdev *netdev_, bool *carrier)
1486 {
1487     struct netdev_dpdk *dev = netdev_dpdk_cast(netdev_);
1488
1489     ovs_mutex_lock(&dev->mutex);
1490     check_link_status(dev);
1491     *carrier = dev->link.link_status;
1492
1493     ovs_mutex_unlock(&dev->mutex);
1494
1495     return 0;
1496 }
1497
1498 static int
1499 netdev_dpdk_vhost_get_carrier(const struct netdev *netdev_, bool *carrier)
1500 {
1501     struct netdev_dpdk *dev = netdev_dpdk_cast(netdev_);
1502     struct virtio_net *virtio_dev = netdev_dpdk_get_virtio(dev);
1503
1504     ovs_mutex_lock(&dev->mutex);
1505
1506     if (is_vhost_running(virtio_dev)) {
1507         *carrier = 1;
1508     } else {
1509         *carrier = 0;
1510     }
1511
1512     ovs_mutex_unlock(&dev->mutex);
1513
1514     return 0;
1515 }
1516
1517 static long long int
1518 netdev_dpdk_get_carrier_resets(const struct netdev *netdev_)
1519 {
1520     struct netdev_dpdk *dev = netdev_dpdk_cast(netdev_);
1521     long long int carrier_resets;
1522
1523     ovs_mutex_lock(&dev->mutex);
1524     carrier_resets = dev->link_reset_cnt;
1525     ovs_mutex_unlock(&dev->mutex);
1526
1527     return carrier_resets;
1528 }
1529
1530 static int
1531 netdev_dpdk_set_miimon(struct netdev *netdev_ OVS_UNUSED,
1532                        long long int interval OVS_UNUSED)
1533 {
1534     return EOPNOTSUPP;
1535 }
1536
1537 static int
1538 netdev_dpdk_update_flags__(struct netdev_dpdk *dev,
1539                            enum netdev_flags off, enum netdev_flags on,
1540                            enum netdev_flags *old_flagsp) OVS_REQUIRES(dev->mutex)
1541 {
1542     int err;
1543
1544     if ((off | on) & ~(NETDEV_UP | NETDEV_PROMISC)) {
1545         return EINVAL;
1546     }
1547
1548     *old_flagsp = dev->flags;
1549     dev->flags |= on;
1550     dev->flags &= ~off;
1551
1552     if (dev->flags == *old_flagsp) {
1553         return 0;
1554     }
1555
1556     if (dev->type == DPDK_DEV_ETH) {
1557         if (dev->flags & NETDEV_UP) {
1558             err = rte_eth_dev_start(dev->port_id);
1559             if (err)
1560                 return -err;
1561         }
1562
1563         if (dev->flags & NETDEV_PROMISC) {
1564             rte_eth_promiscuous_enable(dev->port_id);
1565         }
1566
1567         if (!(dev->flags & NETDEV_UP)) {
1568             rte_eth_dev_stop(dev->port_id);
1569         }
1570     }
1571
1572     return 0;
1573 }
1574
1575 static int
1576 netdev_dpdk_update_flags(struct netdev *netdev_,
1577                          enum netdev_flags off, enum netdev_flags on,
1578                          enum netdev_flags *old_flagsp)
1579 {
1580     struct netdev_dpdk *netdev = netdev_dpdk_cast(netdev_);
1581     int error;
1582
1583     ovs_mutex_lock(&netdev->mutex);
1584     error = netdev_dpdk_update_flags__(netdev, off, on, old_flagsp);
1585     ovs_mutex_unlock(&netdev->mutex);
1586
1587     return error;
1588 }
1589
1590 static int
1591 netdev_dpdk_get_status(const struct netdev *netdev_, struct smap *args)
1592 {
1593     struct netdev_dpdk *dev = netdev_dpdk_cast(netdev_);
1594     struct rte_eth_dev_info dev_info;
1595
1596     if (dev->port_id < 0)
1597         return ENODEV;
1598
1599     ovs_mutex_lock(&dev->mutex);
1600     rte_eth_dev_info_get(dev->port_id, &dev_info);
1601     ovs_mutex_unlock(&dev->mutex);
1602
1603     smap_add_format(args, "driver_name", "%s", dev_info.driver_name);
1604
1605     smap_add_format(args, "port_no", "%d", dev->port_id);
1606     smap_add_format(args, "numa_id", "%d", rte_eth_dev_socket_id(dev->port_id));
1607     smap_add_format(args, "driver_name", "%s", dev_info.driver_name);
1608     smap_add_format(args, "min_rx_bufsize", "%u", dev_info.min_rx_bufsize);
1609     smap_add_format(args, "max_rx_pktlen", "%u", dev_info.max_rx_pktlen);
1610     smap_add_format(args, "max_rx_queues", "%u", dev_info.max_rx_queues);
1611     smap_add_format(args, "max_tx_queues", "%u", dev_info.max_tx_queues);
1612     smap_add_format(args, "max_mac_addrs", "%u", dev_info.max_mac_addrs);
1613     smap_add_format(args, "max_hash_mac_addrs", "%u", dev_info.max_hash_mac_addrs);
1614     smap_add_format(args, "max_vfs", "%u", dev_info.max_vfs);
1615     smap_add_format(args, "max_vmdq_pools", "%u", dev_info.max_vmdq_pools);
1616
1617     smap_add_format(args, "pci-vendor_id", "0x%u", dev_info.pci_dev->id.vendor_id);
1618     smap_add_format(args, "pci-device_id", "0x%x", dev_info.pci_dev->id.device_id);
1619
1620     return 0;
1621 }
1622
1623 static void
1624 netdev_dpdk_set_admin_state__(struct netdev_dpdk *dev, bool admin_state)
1625     OVS_REQUIRES(dev->mutex)
1626 {
1627     enum netdev_flags old_flags;
1628
1629     if (admin_state) {
1630         netdev_dpdk_update_flags__(dev, 0, NETDEV_UP, &old_flags);
1631     } else {
1632         netdev_dpdk_update_flags__(dev, NETDEV_UP, 0, &old_flags);
1633     }
1634 }
1635
1636 static void
1637 netdev_dpdk_set_admin_state(struct unixctl_conn *conn, int argc,
1638                             const char *argv[], void *aux OVS_UNUSED)
1639 {
1640     bool up;
1641
1642     if (!strcasecmp(argv[argc - 1], "up")) {
1643         up = true;
1644     } else if ( !strcasecmp(argv[argc - 1], "down")) {
1645         up = false;
1646     } else {
1647         unixctl_command_reply_error(conn, "Invalid Admin State");
1648         return;
1649     }
1650
1651     if (argc > 2) {
1652         struct netdev *netdev = netdev_from_name(argv[1]);
1653         if (netdev && is_dpdk_class(netdev->netdev_class)) {
1654             struct netdev_dpdk *dpdk_dev = netdev_dpdk_cast(netdev);
1655
1656             ovs_mutex_lock(&dpdk_dev->mutex);
1657             netdev_dpdk_set_admin_state__(dpdk_dev, up);
1658             ovs_mutex_unlock(&dpdk_dev->mutex);
1659
1660             netdev_close(netdev);
1661         } else {
1662             unixctl_command_reply_error(conn, "Not a DPDK Interface");
1663             netdev_close(netdev);
1664             return;
1665         }
1666     } else {
1667         struct netdev_dpdk *netdev;
1668
1669         ovs_mutex_lock(&dpdk_mutex);
1670         LIST_FOR_EACH (netdev, list_node, &dpdk_list) {
1671             ovs_mutex_lock(&netdev->mutex);
1672             netdev_dpdk_set_admin_state__(netdev, up);
1673             ovs_mutex_unlock(&netdev->mutex);
1674         }
1675         ovs_mutex_unlock(&dpdk_mutex);
1676     }
1677     unixctl_command_reply(conn, "OK");
1678 }
1679
1680 /*
1681  * Set virtqueue flags so that we do not receive interrupts.
1682  */
1683 static void
1684 set_irq_status(struct virtio_net *dev)
1685 {
1686     dev->virtqueue[VIRTIO_RXQ]->used->flags = VRING_USED_F_NO_NOTIFY;
1687     dev->virtqueue[VIRTIO_TXQ]->used->flags = VRING_USED_F_NO_NOTIFY;
1688 }
1689
1690 /*
1691  * A new virtio-net device is added to a vhost port.
1692  */
1693 static int
1694 new_device(struct virtio_net *dev)
1695 {
1696     struct netdev_dpdk *netdev;
1697     bool exists = false;
1698
1699     ovs_mutex_lock(&dpdk_mutex);
1700     /* Add device to the vhost port with the same name as that passed down. */
1701     LIST_FOR_EACH(netdev, list_node, &dpdk_list) {
1702         if (strncmp(dev->ifname, netdev->vhost_id, IF_NAME_SZ) == 0) {
1703             ovs_mutex_lock(&netdev->mutex);
1704             ovsrcu_set(&netdev->virtio_dev, dev);
1705             ovs_mutex_unlock(&netdev->mutex);
1706             exists = true;
1707             dev->flags |= VIRTIO_DEV_RUNNING;
1708             /* Disable notifications. */
1709             set_irq_status(dev);
1710             break;
1711         }
1712     }
1713     ovs_mutex_unlock(&dpdk_mutex);
1714
1715     if (!exists) {
1716         VLOG_INFO("vHost Device '%s' (%ld) can't be added - name not found",
1717                    dev->ifname, dev->device_fh);
1718
1719         return -1;
1720     }
1721
1722     VLOG_INFO("vHost Device '%s' (%ld) has been added",
1723                dev->ifname, dev->device_fh);
1724     return 0;
1725 }
1726
1727 /*
1728  * Remove a virtio-net device from the specific vhost port.  Use dev->remove
1729  * flag to stop any more packets from being sent or received to/from a VM and
1730  * ensure all currently queued packets have been sent/received before removing
1731  *  the device.
1732  */
1733 static void
1734 destroy_device(volatile struct virtio_net *dev)
1735 {
1736     struct netdev_dpdk *vhost_dev;
1737
1738     ovs_mutex_lock(&dpdk_mutex);
1739     LIST_FOR_EACH (vhost_dev, list_node, &dpdk_list) {
1740         if (netdev_dpdk_get_virtio(vhost_dev) == dev) {
1741
1742             ovs_mutex_lock(&vhost_dev->mutex);
1743             dev->flags &= ~VIRTIO_DEV_RUNNING;
1744             ovsrcu_set(&vhost_dev->virtio_dev, NULL);
1745             ovs_mutex_unlock(&vhost_dev->mutex);
1746
1747             /*
1748              * Wait for other threads to quiesce before
1749              * setting the virtio_dev to NULL.
1750              */
1751             ovsrcu_synchronize();
1752             /*
1753              * As call to ovsrcu_synchronize() will end the quiescent state,
1754              * put thread back into quiescent state before returning.
1755              */
1756             ovsrcu_quiesce_start();
1757         }
1758     }
1759     ovs_mutex_unlock(&dpdk_mutex);
1760
1761     VLOG_INFO("vHost Device '%s' (%ld) has been removed",
1762                dev->ifname, dev->device_fh);
1763 }
1764
1765 struct virtio_net *
1766 netdev_dpdk_get_virtio(const struct netdev_dpdk *dev)
1767 {
1768     return ovsrcu_get(struct virtio_net *, &dev->virtio_dev);
1769 }
1770
1771 /*
1772  * These callbacks allow virtio-net devices to be added to vhost ports when
1773  * configuration has been fully complete.
1774  */
1775 static const struct virtio_net_device_ops virtio_net_device_ops =
1776 {
1777     .new_device =  new_device,
1778     .destroy_device = destroy_device,
1779 };
1780
1781 static void *
1782 start_vhost_loop(void *dummy OVS_UNUSED)
1783 {
1784      pthread_detach(pthread_self());
1785      /* Put the cuse thread into quiescent state. */
1786      ovsrcu_quiesce_start();
1787      rte_vhost_driver_session_start();
1788      return NULL;
1789 }
1790
1791 static int
1792 dpdk_vhost_class_init(void)
1793 {
1794     rte_vhost_driver_callback_register(&virtio_net_device_ops);
1795     ovs_thread_create("vhost_thread", start_vhost_loop, NULL);
1796     return 0;
1797 }
1798
1799 static int
1800 dpdk_vhost_cuse_class_init(void)
1801 {
1802     int err = -1;
1803
1804
1805     /* Register CUSE device to handle IOCTLs.
1806      * Unless otherwise specified on the vswitchd command line, cuse_dev_name
1807      * is set to vhost-net.
1808      */
1809     err = rte_vhost_driver_register(cuse_dev_name);
1810
1811     if (err != 0) {
1812         VLOG_ERR("CUSE device setup failure.");
1813         return -1;
1814     }
1815
1816     dpdk_vhost_class_init();
1817     return 0;
1818 }
1819
1820 static int
1821 dpdk_vhost_user_class_init(void)
1822 {
1823     dpdk_vhost_class_init();
1824     return 0;
1825 }
1826
1827 static void
1828 dpdk_common_init(void)
1829 {
1830     unixctl_command_register("netdev-dpdk/set-admin-state",
1831                              "[netdev] up|down", 1, 2,
1832                              netdev_dpdk_set_admin_state, NULL);
1833
1834     ovs_thread_create("dpdk_watchdog", dpdk_watchdog, NULL);
1835 }
1836
1837 /* Client Rings */
1838
1839 static int
1840 dpdk_ring_create(const char dev_name[], unsigned int port_no,
1841                  unsigned int *eth_port_id)
1842 {
1843     struct dpdk_ring *ivshmem;
1844     char ring_name[10];
1845     int err;
1846
1847     ivshmem = dpdk_rte_mzalloc(sizeof *ivshmem);
1848     if (ivshmem == NULL) {
1849         return ENOMEM;
1850     }
1851
1852     /* XXX: Add support for multiquque ring. */
1853     err = snprintf(ring_name, 10, "%s_tx", dev_name);
1854     if (err < 0) {
1855         return -err;
1856     }
1857
1858     /* Create single consumer/producer rings, netdev does explicit locking. */
1859     ivshmem->cring_tx = rte_ring_create(ring_name, DPDK_RING_SIZE, SOCKET0,
1860                                         RING_F_SP_ENQ | RING_F_SC_DEQ);
1861     if (ivshmem->cring_tx == NULL) {
1862         rte_free(ivshmem);
1863         return ENOMEM;
1864     }
1865
1866     err = snprintf(ring_name, 10, "%s_rx", dev_name);
1867     if (err < 0) {
1868         return -err;
1869     }
1870
1871     /* Create single consumer/producer rings, netdev does explicit locking. */
1872     ivshmem->cring_rx = rte_ring_create(ring_name, DPDK_RING_SIZE, SOCKET0,
1873                                         RING_F_SP_ENQ | RING_F_SC_DEQ);
1874     if (ivshmem->cring_rx == NULL) {
1875         rte_free(ivshmem);
1876         return ENOMEM;
1877     }
1878
1879     err = rte_eth_from_rings(dev_name, &ivshmem->cring_rx, 1,
1880                              &ivshmem->cring_tx, 1, SOCKET0);
1881
1882     if (err < 0) {
1883         rte_free(ivshmem);
1884         return ENODEV;
1885     }
1886
1887     ivshmem->user_port_id = port_no;
1888     ivshmem->eth_port_id = rte_eth_dev_count() - 1;
1889     list_push_back(&dpdk_ring_list, &ivshmem->list_node);
1890
1891     *eth_port_id = ivshmem->eth_port_id;
1892     return 0;
1893 }
1894
1895 static int
1896 dpdk_ring_open(const char dev_name[], unsigned int *eth_port_id) OVS_REQUIRES(dpdk_mutex)
1897 {
1898     struct dpdk_ring *ivshmem;
1899     unsigned int port_no;
1900     int err = 0;
1901
1902     /* Names always start with "dpdkr" */
1903     err = dpdk_dev_parse_name(dev_name, "dpdkr", &port_no);
1904     if (err) {
1905         return err;
1906     }
1907
1908     /* look through our list to find the device */
1909     LIST_FOR_EACH (ivshmem, list_node, &dpdk_ring_list) {
1910          if (ivshmem->user_port_id == port_no) {
1911             VLOG_INFO("Found dpdk ring device %s:", dev_name);
1912             *eth_port_id = ivshmem->eth_port_id; /* really all that is needed */
1913             return 0;
1914          }
1915     }
1916     /* Need to create the device rings */
1917     return dpdk_ring_create(dev_name, port_no, eth_port_id);
1918 }
1919
1920 static int
1921 netdev_dpdk_ring_send(struct netdev *netdev_, int qid,
1922                       struct dp_packet **pkts, int cnt, bool may_steal)
1923 {
1924     struct netdev_dpdk *netdev = netdev_dpdk_cast(netdev_);
1925     unsigned i;
1926
1927     /* When using 'dpdkr' and sending to a DPDK ring, we want to ensure that the
1928      * rss hash field is clear. This is because the same mbuf may be modified by
1929      * the consumer of the ring and return into the datapath without recalculating
1930      * the RSS hash. */
1931     for (i = 0; i < cnt; i++) {
1932         dp_packet_set_rss_hash(pkts[i], 0);
1933     }
1934
1935     netdev_dpdk_send__(netdev, qid, pkts, cnt, may_steal);
1936     return 0;
1937 }
1938
1939 static int
1940 netdev_dpdk_ring_construct(struct netdev *netdev)
1941 {
1942     unsigned int port_no = 0;
1943     int err = 0;
1944
1945     if (rte_eal_init_ret) {
1946         return rte_eal_init_ret;
1947     }
1948
1949     ovs_mutex_lock(&dpdk_mutex);
1950
1951     err = dpdk_ring_open(netdev->name, &port_no);
1952     if (err) {
1953         goto unlock_dpdk;
1954     }
1955
1956     err = netdev_dpdk_init(netdev, port_no, DPDK_DEV_ETH);
1957
1958 unlock_dpdk:
1959     ovs_mutex_unlock(&dpdk_mutex);
1960     return err;
1961 }
1962
1963 #define NETDEV_DPDK_CLASS(NAME, INIT, CONSTRUCT, DESTRUCT, MULTIQ, SEND, \
1964     GET_CARRIER, GET_STATS, GET_FEATURES, GET_STATUS, RXQ_RECV)          \
1965 {                                                             \
1966     NAME,                                                     \
1967     INIT,                       /* init */                    \
1968     NULL,                       /* netdev_dpdk_run */         \
1969     NULL,                       /* netdev_dpdk_wait */        \
1970                                                               \
1971     netdev_dpdk_alloc,                                        \
1972     CONSTRUCT,                                                \
1973     DESTRUCT,                                                 \
1974     netdev_dpdk_dealloc,                                      \
1975     netdev_dpdk_get_config,                                   \
1976     NULL,                       /* netdev_dpdk_set_config */  \
1977     NULL,                       /* get_tunnel_config */       \
1978     NULL,                       /* build header */            \
1979     NULL,                       /* push header */             \
1980     NULL,                       /* pop header */              \
1981     netdev_dpdk_get_numa_id,    /* get_numa_id */             \
1982     MULTIQ,                     /* set_multiq */              \
1983                                                               \
1984     SEND,                       /* send */                    \
1985     NULL,                       /* send_wait */               \
1986                                                               \
1987     netdev_dpdk_set_etheraddr,                                \
1988     netdev_dpdk_get_etheraddr,                                \
1989     netdev_dpdk_get_mtu,                                      \
1990     netdev_dpdk_set_mtu,                                      \
1991     netdev_dpdk_get_ifindex,                                  \
1992     GET_CARRIER,                                              \
1993     netdev_dpdk_get_carrier_resets,                           \
1994     netdev_dpdk_set_miimon,                                   \
1995     GET_STATS,                                                \
1996     GET_FEATURES,                                             \
1997     NULL,                       /* set_advertisements */      \
1998                                                               \
1999     NULL,                       /* set_policing */            \
2000     NULL,                       /* get_qos_types */           \
2001     NULL,                       /* get_qos_capabilities */    \
2002     NULL,                       /* get_qos */                 \
2003     NULL,                       /* set_qos */                 \
2004     NULL,                       /* get_queue */               \
2005     NULL,                       /* set_queue */               \
2006     NULL,                       /* delete_queue */            \
2007     NULL,                       /* get_queue_stats */         \
2008     NULL,                       /* queue_dump_start */        \
2009     NULL,                       /* queue_dump_next */         \
2010     NULL,                       /* queue_dump_done */         \
2011     NULL,                       /* dump_queue_stats */        \
2012                                                               \
2013     NULL,                       /* get_in4 */                 \
2014     NULL,                       /* set_in4 */                 \
2015     NULL,                       /* get_in6 */                 \
2016     NULL,                       /* add_router */              \
2017     NULL,                       /* get_next_hop */            \
2018     GET_STATUS,                                               \
2019     NULL,                       /* arp_lookup */              \
2020                                                               \
2021     netdev_dpdk_update_flags,                                 \
2022                                                               \
2023     netdev_dpdk_rxq_alloc,                                    \
2024     netdev_dpdk_rxq_construct,                                \
2025     netdev_dpdk_rxq_destruct,                                 \
2026     netdev_dpdk_rxq_dealloc,                                  \
2027     RXQ_RECV,                                                 \
2028     NULL,                       /* rx_wait */                 \
2029     NULL,                       /* rxq_drain */               \
2030 }
2031
2032 static int
2033 process_vhost_flags(char *flag, char *default_val, int size,
2034                     char **argv, char **new_val)
2035 {
2036     int changed = 0;
2037
2038     /* Depending on which version of vhost is in use, process the vhost-specific
2039      * flag if it is provided on the vswitchd command line, otherwise resort to
2040      * a default value.
2041      *
2042      * For vhost-user: Process "-cuse_dev_name" to set the custom location of
2043      * the vhost-user socket(s).
2044      * For vhost-cuse: Process "-vhost_sock_dir" to set the custom name of the
2045      * vhost-cuse character device.
2046      */
2047     if (!strcmp(argv[1], flag) && (strlen(argv[2]) <= size)) {
2048         changed = 1;
2049         *new_val = strdup(argv[2]);
2050         VLOG_INFO("User-provided %s in use: %s", flag, *new_val);
2051     } else {
2052         VLOG_INFO("No %s provided - defaulting to %s", flag, default_val);
2053         *new_val = default_val;
2054     }
2055
2056     return changed;
2057 }
2058
2059 int
2060 dpdk_init(int argc, char **argv)
2061 {
2062     int result;
2063     int base = 0;
2064     char *pragram_name = argv[0];
2065
2066     if (argc < 2 || strcmp(argv[1], "--dpdk"))
2067         return 0;
2068
2069     /* Remove the --dpdk argument from arg list.*/
2070     argc--;
2071     argv++;
2072
2073 #ifdef VHOST_CUSE
2074     if (process_vhost_flags("-cuse_dev_name", strdup("vhost-net"),
2075                             PATH_MAX, argv, &cuse_dev_name)) {
2076 #else
2077     if (process_vhost_flags("-vhost_sock_dir", strdup(ovs_rundir()),
2078                             NAME_MAX, argv, &vhost_sock_dir)) {
2079         struct stat s;
2080         int err;
2081
2082         err = stat(vhost_sock_dir, &s);
2083         if (err) {
2084             VLOG_ERR("vHostUser socket DIR '%s' does not exist.",
2085                      vhost_sock_dir);
2086             return err;
2087         }
2088 #endif
2089         /* Remove the vhost flag configuration parameters from the argument
2090          * list, so that the correct elements are passed to the DPDK
2091          * initialization function
2092          */
2093         argc -= 2;
2094         argv += 2;    /* Increment by two to bypass the vhost flag arguments */
2095         base = 2;
2096     }
2097
2098     /* Keep the program name argument as this is needed for call to
2099      * rte_eal_init()
2100      */
2101     argv[0] = pragram_name;
2102
2103     /* Make sure things are initialized ... */
2104     result = rte_eal_init(argc, argv);
2105     if (result < 0) {
2106         ovs_abort(result, "Cannot init EAL");
2107     }
2108
2109     rte_memzone_dump(stdout);
2110     rte_eal_init_ret = 0;
2111
2112     if (argc > result) {
2113         argv[result] = argv[0];
2114     }
2115
2116     /* We are called from the main thread here */
2117     RTE_PER_LCORE(_lcore_id) = NON_PMD_CORE_ID;
2118
2119     return result + 1 + base;
2120 }
2121
2122 static const struct netdev_class dpdk_class =
2123     NETDEV_DPDK_CLASS(
2124         "dpdk",
2125         NULL,
2126         netdev_dpdk_construct,
2127         netdev_dpdk_destruct,
2128         netdev_dpdk_set_multiq,
2129         netdev_dpdk_eth_send,
2130         netdev_dpdk_get_carrier,
2131         netdev_dpdk_get_stats,
2132         netdev_dpdk_get_features,
2133         netdev_dpdk_get_status,
2134         netdev_dpdk_rxq_recv);
2135
2136 static const struct netdev_class dpdk_ring_class =
2137     NETDEV_DPDK_CLASS(
2138         "dpdkr",
2139         NULL,
2140         netdev_dpdk_ring_construct,
2141         netdev_dpdk_destruct,
2142         netdev_dpdk_set_multiq,
2143         netdev_dpdk_ring_send,
2144         netdev_dpdk_get_carrier,
2145         netdev_dpdk_get_stats,
2146         netdev_dpdk_get_features,
2147         netdev_dpdk_get_status,
2148         netdev_dpdk_rxq_recv);
2149
2150 static const struct netdev_class OVS_UNUSED dpdk_vhost_cuse_class =
2151     NETDEV_DPDK_CLASS(
2152         "dpdkvhostcuse",
2153         dpdk_vhost_cuse_class_init,
2154         netdev_dpdk_vhost_cuse_construct,
2155         netdev_dpdk_vhost_destruct,
2156         netdev_dpdk_vhost_set_multiq,
2157         netdev_dpdk_vhost_send,
2158         netdev_dpdk_vhost_get_carrier,
2159         netdev_dpdk_vhost_get_stats,
2160         NULL,
2161         NULL,
2162         netdev_dpdk_vhost_rxq_recv);
2163
2164 static const struct netdev_class OVS_UNUSED dpdk_vhost_user_class =
2165     NETDEV_DPDK_CLASS(
2166         "dpdkvhostuser",
2167         dpdk_vhost_user_class_init,
2168         netdev_dpdk_vhost_user_construct,
2169         netdev_dpdk_vhost_destruct,
2170         netdev_dpdk_vhost_set_multiq,
2171         netdev_dpdk_vhost_send,
2172         netdev_dpdk_vhost_get_carrier,
2173         netdev_dpdk_vhost_get_stats,
2174         NULL,
2175         NULL,
2176         netdev_dpdk_vhost_rxq_recv);
2177
2178 void
2179 netdev_dpdk_register(void)
2180 {
2181     static struct ovsthread_once once = OVSTHREAD_ONCE_INITIALIZER;
2182
2183     if (rte_eal_init_ret) {
2184         return;
2185     }
2186
2187     if (ovsthread_once_start(&once)) {
2188         dpdk_common_init();
2189         netdev_register_provider(&dpdk_class);
2190         netdev_register_provider(&dpdk_ring_class);
2191 #ifdef VHOST_CUSE
2192         netdev_register_provider(&dpdk_vhost_cuse_class);
2193 #else
2194         netdev_register_provider(&dpdk_vhost_user_class);
2195 #endif
2196         ovsthread_once_done(&once);
2197     }
2198 }
2199
2200 int
2201 pmd_thread_setaffinity_cpu(unsigned cpu)
2202 {
2203     cpu_set_t cpuset;
2204     int err;
2205
2206     CPU_ZERO(&cpuset);
2207     CPU_SET(cpu, &cpuset);
2208     err = pthread_setaffinity_np(pthread_self(), sizeof(cpu_set_t), &cpuset);
2209     if (err) {
2210         VLOG_ERR("Thread affinity error %d",err);
2211         return err;
2212     }
2213     /* NON_PMD_CORE_ID is reserved for use by non pmd threads. */
2214     ovs_assert(cpu != NON_PMD_CORE_ID);
2215     RTE_PER_LCORE(_lcore_id) = cpu;
2216
2217     return 0;
2218 }
2219
2220 static bool
2221 thread_is_pmd(void)
2222 {
2223     return rte_lcore_id() != NON_PMD_CORE_ID;
2224 }