lib/dpif-netdev.c

   1 /*
   2  * Copyright (c) 2009, 2010, 2011, 2012, 2013, 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 #include "dpif.h"
  19
  20 #include <ctype.h>
  21 #include <errno.h>
  22 #include <fcntl.h>
  23 #include <inttypes.h>
  24 #include <netinet/in.h>
  25 #include <sys/socket.h>
  26 #include <net/if.h>
  27 #include <stdint.h>
  28 #include <stdlib.h>
  29 #include <string.h>
  30 #include <sys/ioctl.h>
  31 #include <sys/stat.h>
  32 #include <unistd.h>
  33
  34 #include "classifier.h"
  35 #include "csum.h"
  36 #include "dpif.h"
  37 #include "dpif-provider.h"
  38 #include "dummy.h"
  39 #include "dynamic-string.h"
  40 #include "flow.h"
  41 #include "hmap.h"
  42 #include "latch.h"
  43 #include "list.h"
  44 #include "meta-flow.h"
  45 #include "netdev.h"
  46 #include "netdev-vport.h"
  47 #include "netlink.h"
  48 #include "odp-execute.h"
  49 #include "odp-util.h"
  50 #include "ofp-print.h"
  51 #include "ofpbuf.h"
  52 #include "ovs-rcu.h"
  53 #include "packets.h"
  54 #include "poll-loop.h"
  55 #include "random.h"
  56 #include "seq.h"
  57 #include "shash.h"
  58 #include "sset.h"
  59 #include "timeval.h"
  60 #include "unixctl.h"
  61 #include "util.h"
  62 #include "vlog.h"
  63
  64 VLOG_DEFINE_THIS_MODULE(dpif_netdev);
  65
  66 /* By default, choose a priority in the middle. */
  67 #define NETDEV_RULE_PRIORITY 0x8000
  68
  69 #define NR_THREADS 1
  70
  71 /* Configuration parameters. */
  72 enum { MAX_FLOWS = 65536 };     /* Maximum number of flows in flow table. */
  73
  74 /* Queues. */
  75 enum { MAX_QUEUE_LEN = 128 };   /* Maximum number of packets per queue. */
  76 enum { QUEUE_MASK = MAX_QUEUE_LEN - 1 };
  77 BUILD_ASSERT_DECL(IS_POW2(MAX_QUEUE_LEN));
  78
  79 /* Protects against changes to 'dp_netdevs'. */
  80 static struct ovs_mutex dp_netdev_mutex = OVS_MUTEX_INITIALIZER;
  81
  82 /* Contains all 'struct dp_netdev's. */
  83 static struct shash dp_netdevs OVS_GUARDED_BY(dp_netdev_mutex)
  84     = SHASH_INITIALIZER(&dp_netdevs);
  85
  86 struct dp_netdev_upcall {
  87     struct dpif_upcall upcall;  /* Queued upcall information. */
  88     struct ofpbuf buf;          /* ofpbuf instance for upcall.packet. */
  89 };
  90
  91 /* A queue passing packets from a struct dp_netdev to its clients (handlers).
  92  *
  93  *
  94  * Thread-safety
  95  * =============
  96  *
  97  * Any access at all requires the owning 'dp_netdev''s queue_rwlock and
  98  * its own mutex. */
  99 struct dp_netdev_queue {
 100     struct ovs_mutex mutex;
 101     struct seq *seq;      /* Incremented whenever a packet is queued. */
 102     struct dp_netdev_upcall upcalls[MAX_QUEUE_LEN] OVS_GUARDED;
 103     unsigned int head OVS_GUARDED;
 104     unsigned int tail OVS_GUARDED;
 105 };
 106
 107 /* Datapath based on the network device interface from netdev.h.
 108  *
 109  *
 110  * Thread-safety
 111  * =============
 112  *
 113  * Some members, marked 'const', are immutable.  Accessing other members
 114  * requires synchronization, as noted in more detail below.
 115  *
 116  * Acquisition order is, from outermost to innermost:
 117  *
 118  *    dp_netdev_mutex (global)
 119  *    port_rwlock
 120  *    flow_mutex
 121  *    cls.rwlock
 122  *    queue_rwlock
 123  */
 124 struct dp_netdev {
 125     const struct dpif_class *const class;
 126     const char *const name;
 127     struct ovs_refcount ref_cnt;
 128     atomic_flag destroyed;
 129
 130     /* Flows.
 131      *
 132      * Readers of 'cls' and 'flow_table' must take a 'cls->rwlock' read lock.
 133      *
 134      * Writers of 'cls' and 'flow_table' must take the 'flow_mutex' and then
 135      * the 'cls->rwlock' write lock.  (The outer 'flow_mutex' allows writers to
 136      * atomically perform multiple operations on 'cls' and 'flow_table'.)
 137      */
 138     struct ovs_mutex flow_mutex;
 139     struct classifier cls;      /* Classifier.  Protected by cls.rwlock. */
 140     struct hmap flow_table OVS_GUARDED; /* Flow table. */
 141
 142     /* Queues.
 143      *
 144      * 'queue_rwlock' protects the modification of 'handler_queues' and
 145      * 'n_handlers'.  The queue elements are protected by its
 146      * 'handler_queues''s mutex. */
 147     struct fat_rwlock queue_rwlock;
 148     struct dp_netdev_queue *handler_queues;
 149     uint32_t n_handlers;
 150
 151     /* Statistics.
 152      *
 153      * ovsthread_stats is internally synchronized. */
 154     struct ovsthread_stats stats; /* Contains 'struct dp_netdev_stats *'. */
 155
 156     /* Ports.
 157      *
 158      * Any lookup into 'ports' or any access to the dp_netdev_ports found
 159      * through 'ports' requires taking 'port_rwlock'. */
 160     struct ovs_rwlock port_rwlock;
 161     struct hmap ports OVS_GUARDED;
 162     struct seq *port_seq;       /* Incremented whenever a port changes. */
 163
 164     /* Forwarding threads. */
 165     struct latch exit_latch;
 166     struct pmd_thread *pmd_threads;
 167     size_t n_pmd_threads;
 168     int pmd_count;
 169 };
 170
 171 static struct dp_netdev_port *dp_netdev_lookup_port(const struct dp_netdev *dp,
 172                                                     odp_port_t)
 173     OVS_REQ_RDLOCK(dp->port_rwlock);
 174
 175 enum dp_stat_type {
 176     DP_STAT_HIT,                /* Packets that matched in the flow table. */
 177     DP_STAT_MISS,               /* Packets that did not match. */
 178     DP_STAT_LOST,               /* Packets not passed up to the client. */
 179     DP_N_STATS
 180 };
 181
 182 /* Contained by struct dp_netdev's 'stats' member.  */
 183 struct dp_netdev_stats {
 184     struct ovs_mutex mutex;          /* Protects 'n'. */
 185
 186     /* Indexed by DP_STAT_*, protected by 'mutex'. */
 187     unsigned long long int n[DP_N_STATS] OVS_GUARDED;
 188 };
 189
 190
 191 /* A port in a netdev-based datapath. */
 192 struct dp_netdev_port {
 193     struct hmap_node node;      /* Node in dp_netdev's 'ports'. */
 194     odp_port_t port_no;
 195     struct netdev *netdev;
 196     struct netdev_saved_flags *sf;
 197     struct netdev_rxq **rxq;
 198     struct ovs_refcount ref_cnt;
 199     char *type;                 /* Port type as requested by user. */
 200 };
 201
 202 /* A flow in dp_netdev's 'flow_table'.
 203  *
 204  *
 205  * Thread-safety
 206  * =============
 207  *
 208  * Except near the beginning or ending of its lifespan, rule 'rule' belongs to
 209  * its dp_netdev's classifier.  The text below calls this classifier 'cls'.
 210  *
 211  * Motivation
 212  * ----------
 213  *
 214  * The thread safety rules described here for "struct dp_netdev_flow" are
 215  * motivated by two goals:
 216  *
 217  *    - Prevent threads that read members of "struct dp_netdev_flow" from
 218  *      reading bad data due to changes by some thread concurrently modifying
 219  *      those members.
 220  *
 221  *    - Prevent two threads making changes to members of a given "struct
 222  *      dp_netdev_flow" from interfering with each other.
 223  *
 224  *
 225  * Rules
 226  * -----
 227  *
 228  * A flow 'flow' may be accessed without a risk of being freed by code that
 229  * holds a read-lock or write-lock on 'cls->rwlock' or that owns a reference to
 230  * 'flow->ref_cnt' (or both).  Code that needs to hold onto a flow for a while
 231  * should take 'cls->rwlock', find the flow it needs, increment 'flow->ref_cnt'
 232  * with dpif_netdev_flow_ref(), and drop 'cls->rwlock'.
 233  *
 234  * 'flow->ref_cnt' protects 'flow' from being freed.  It doesn't protect the
 235  * flow from being deleted from 'cls' (that's 'cls->rwlock') and it doesn't
 236  * protect members of 'flow' from modification (that's 'flow->mutex').
 237  *
 238  * 'flow->mutex' protects the members of 'flow' from modification.  It doesn't
 239  * protect the flow from being deleted from 'cls' (that's 'cls->rwlock') and it
 240  * doesn't prevent the flow from being freed (that's 'flow->ref_cnt').
 241  *
 242  * Some members, marked 'const', are immutable.  Accessing other members
 243  * requires synchronization, as noted in more detail below.
 244  */
 245 struct dp_netdev_flow {
 246     /* Packet classification. */
 247     const struct cls_rule cr;   /* In owning dp_netdev's 'cls'. */
 248
 249     /* Hash table index by unmasked flow. */
 250     const struct hmap_node node; /* In owning dp_netdev's 'flow_table'. */
 251     const struct flow flow;      /* The flow that created this entry. */
 252
 253     /* Protects members marked OVS_GUARDED.
 254      *
 255      * Acquire after datapath's flow_mutex. */
 256     struct ovs_mutex mutex OVS_ACQ_AFTER(dp_netdev_mutex);
 257
 258     /* Statistics.
 259      *
 260      * Reading or writing these members requires 'mutex'. */
 261     struct ovsthread_stats stats; /* Contains "struct dp_netdev_flow_stats". */
 262
 263     /* Actions.
 264      *
 265      * Reading 'actions' requires 'mutex'.
 266      * Writing 'actions' requires 'mutex' and (to allow for transactions) the
 267      * datapath's flow_mutex. */
 268     OVSRCU_TYPE(struct dp_netdev_actions *) actions;
 269 };
 270
 271 static void dp_netdev_flow_free(struct dp_netdev_flow *);
 272
 273 /* Contained by struct dp_netdev_flow's 'stats' member.  */
 274 struct dp_netdev_flow_stats {
 275     struct ovs_mutex mutex;         /* Guards all the other members. */
 276
 277     long long int used OVS_GUARDED; /* Last used time, in monotonic msecs. */
 278     long long int packet_count OVS_GUARDED; /* Number of packets matched. */
 279     long long int byte_count OVS_GUARDED;   /* Number of bytes matched. */
 280     uint16_t tcp_flags OVS_GUARDED; /* Bitwise-OR of seen tcp_flags values. */
 281 };
 282
 283 /* A set of datapath actions within a "struct dp_netdev_flow".
 284  *
 285  *
 286  * Thread-safety
 287  * =============
 288  *
 289  * A struct dp_netdev_actions 'actions' may be accessed without a risk of being
 290  * freed by code that holds a read-lock or write-lock on 'flow->mutex' (where
 291  * 'flow' is the dp_netdev_flow for which 'flow->actions == actions') or that
 292  * owns a reference to 'actions->ref_cnt' (or both). */
 293 struct dp_netdev_actions {
 294     /* These members are immutable: they do not change during the struct's
 295      * lifetime.  */
 296     struct nlattr *actions;     /* Sequence of OVS_ACTION_ATTR_* attributes. */
 297     unsigned int size;          /* Size of 'actions', in bytes. */
 298 };
 299
 300 struct dp_netdev_actions *dp_netdev_actions_create(const struct nlattr *,
 301                                                    size_t);
 302 struct dp_netdev_actions *dp_netdev_flow_get_actions(
 303     const struct dp_netdev_flow *);
 304 static void dp_netdev_actions_free(struct dp_netdev_actions *);
 305
 306 /* PMD: Poll modes drivers.  PMD accesses devices via polling to eliminate
 307  * the performance overhead of interrupt processing.  Therefore netdev can
 308  * not implement rx-wait for these devices.  dpif-netdev needs to poll
 309  * these device to check for recv buffer.  pmd-thread does polling for
 310  * devices assigned to itself thread.
 311  *
 312  * DPDK used PMD for accessing NIC.
 313  *
 314  * A thread that receives packets from PMD ports, looks them up in the flow
 315  * table, and executes the actions it finds.
 316  **/
 317 struct pmd_thread {
 318     struct dp_netdev *dp;
 319     pthread_t thread;
 320     int id;
 321     atomic_uint change_seq;
 322     char *name;
 323 };
 324
 325 /* Interface to netdev-based datapath. */
 326 struct dpif_netdev {
 327     struct dpif dpif;
 328     struct dp_netdev *dp;
 329     uint64_t last_port_seq;
 330 };
 331
 332 static int get_port_by_number(struct dp_netdev *dp, odp_port_t port_no,
 333                               struct dp_netdev_port **portp)
 334     OVS_REQ_RDLOCK(dp->port_rwlock);
 335 static int get_port_by_name(struct dp_netdev *dp, const char *devname,
 336                             struct dp_netdev_port **portp)
 337     OVS_REQ_RDLOCK(dp->port_rwlock);
 338 static void dp_netdev_free(struct dp_netdev *)
 339     OVS_REQUIRES(dp_netdev_mutex);
 340 static void dp_netdev_flow_flush(struct dp_netdev *);
 341 static int do_add_port(struct dp_netdev *dp, const char *devname,
 342                        const char *type, odp_port_t port_no)
 343     OVS_REQ_WRLOCK(dp->port_rwlock);
 344 static int do_del_port(struct dp_netdev *dp, odp_port_t port_no)
 345     OVS_REQ_WRLOCK(dp->port_rwlock);
 346 static void dp_netdev_destroy_all_queues(struct dp_netdev *dp)
 347     OVS_REQ_WRLOCK(dp->queue_rwlock);
 348 static int dpif_netdev_open(const struct dpif_class *, const char *name,
 349                             bool create, struct dpif **);
 350 static int dp_netdev_output_userspace(struct dp_netdev *dp, struct ofpbuf *,
 351                                       int queue_no, int type,
 352                                       const struct flow *,
 353                                       const struct nlattr *userdata);
 354 static void dp_netdev_execute_actions(struct dp_netdev *dp,
 355                                       const struct flow *, struct ofpbuf *, bool may_steal,
 356                                       struct pkt_metadata *,
 357                                       const struct nlattr *actions,
 358                                       size_t actions_len);
 359 static void dp_netdev_port_input(struct dp_netdev *dp, struct ofpbuf *packet,
 360                                  struct pkt_metadata *);
 361
 362 static void dp_netdev_set_pmd_threads(struct dp_netdev *, int n);
 363
 364 static struct dpif_netdev *
 365 dpif_netdev_cast(const struct dpif *dpif)
 366 {
 367     ovs_assert(dpif->dpif_class->open == dpif_netdev_open);
 368     return CONTAINER_OF(dpif, struct dpif_netdev, dpif);
 369 }
 370
 371 static struct dp_netdev *
 372 get_dp_netdev(const struct dpif *dpif)
 373 {
 374     return dpif_netdev_cast(dpif)->dp;
 375 }
 376
 377 static int
 378 dpif_netdev_enumerate(struct sset *all_dps)
 379 {
 380     struct shash_node *node;
 381
 382     ovs_mutex_lock(&dp_netdev_mutex);
 383     SHASH_FOR_EACH(node, &dp_netdevs) {
 384         sset_add(all_dps, node->name);
 385     }
 386     ovs_mutex_unlock(&dp_netdev_mutex);
 387
 388     return 0;
 389 }
 390
 391 static bool
 392 dpif_netdev_class_is_dummy(const struct dpif_class *class)
 393 {
 394     return class != &dpif_netdev_class;
 395 }
 396
 397 static const char *
 398 dpif_netdev_port_open_type(const struct dpif_class *class, const char *type)
 399 {
 400     return strcmp(type, "internal") ? type
 401                   : dpif_netdev_class_is_dummy(class) ? "dummy"
 402                   : "tap";
 403 }
 404
 405 static struct dpif *
 406 create_dpif_netdev(struct dp_netdev *dp)
 407 {
 408     uint16_t netflow_id = hash_string(dp->name, 0);
 409     struct dpif_netdev *dpif;
 410
 411     ovs_refcount_ref(&dp->ref_cnt);
 412
 413     dpif = xmalloc(sizeof *dpif);
 414     dpif_init(&dpif->dpif, dp->class, dp->name, netflow_id >> 8, netflow_id);
 415     dpif->dp = dp;
 416     dpif->last_port_seq = seq_read(dp->port_seq);
 417
 418     return &dpif->dpif;
 419 }
 420
 421 /* Choose an unused, non-zero port number and return it on success.
 422  * Return ODPP_NONE on failure. */
 423 static odp_port_t
 424 choose_port(struct dp_netdev *dp, const char *name)
 425     OVS_REQ_RDLOCK(dp->port_rwlock)
 426 {
 427     uint32_t port_no;
 428
 429     if (dp->class != &dpif_netdev_class) {
 430         const char *p;
 431         int start_no = 0;
 432
 433         /* If the port name begins with "br", start the number search at
 434          * 100 to make writing tests easier. */
 435         if (!strncmp(name, "br", 2)) {
 436             start_no = 100;
 437         }
 438
 439         /* If the port name contains a number, try to assign that port number.
 440          * This can make writing unit tests easier because port numbers are
 441          * predictable. */
 442         for (p = name; *p != '\0'; p++) {
 443             if (isdigit((unsigned char) *p)) {
 444                 port_no = start_no + strtol(p, NULL, 10);
 445                 if (port_no > 0 && port_no != odp_to_u32(ODPP_NONE)
 446                     && !dp_netdev_lookup_port(dp, u32_to_odp(port_no))) {
 447                     return u32_to_odp(port_no);
 448                 }
 449                 break;
 450             }
 451         }
 452     }
 453
 454     for (port_no = 1; port_no <= UINT16_MAX; port_no++) {
 455         if (!dp_netdev_lookup_port(dp, u32_to_odp(port_no))) {
 456             return u32_to_odp(port_no);
 457         }
 458     }
 459
 460     return ODPP_NONE;
 461 }
 462
 463 static int
 464 create_dp_netdev(const char *name, const struct dpif_class *class,
 465                  struct dp_netdev **dpp)
 466     OVS_REQUIRES(dp_netdev_mutex)
 467 {
 468     struct dp_netdev *dp;
 469     int error;
 470
 471     dp = xzalloc(sizeof *dp);
 472     shash_add(&dp_netdevs, name, dp);
 473
 474     *CONST_CAST(const struct dpif_class **, &dp->class) = class;
 475     *CONST_CAST(const char **, &dp->name) = xstrdup(name);
 476     ovs_refcount_init(&dp->ref_cnt);
 477     atomic_flag_clear(&dp->destroyed);
 478
 479     ovs_mutex_init(&dp->flow_mutex);
 480     classifier_init(&dp->cls, NULL);
 481     hmap_init(&dp->flow_table);
 482
 483     fat_rwlock_init(&dp->queue_rwlock);
 484
 485     ovsthread_stats_init(&dp->stats);
 486
 487     ovs_rwlock_init(&dp->port_rwlock);
 488     hmap_init(&dp->ports);
 489     dp->port_seq = seq_create();
 490     latch_init(&dp->exit_latch);
 491
 492     ovs_rwlock_wrlock(&dp->port_rwlock);
 493     error = do_add_port(dp, name, "internal", ODPP_LOCAL);
 494     ovs_rwlock_unlock(&dp->port_rwlock);
 495     if (error) {
 496         dp_netdev_free(dp);
 497         return error;
 498     }
 499
 500     *dpp = dp;
 501     return 0;
 502 }
 503
 504 static int
 505 dpif_netdev_open(const struct dpif_class *class, const char *name,
 506                  bool create, struct dpif **dpifp)
 507 {
 508     struct dp_netdev *dp;
 509     int error;
 510
 511     ovs_mutex_lock(&dp_netdev_mutex);
 512     dp = shash_find_data(&dp_netdevs, name);
 513     if (!dp) {
 514         error = create ? create_dp_netdev(name, class, &dp) : ENODEV;
 515     } else {
 516         error = (dp->class != class ? EINVAL
 517                  : create ? EEXIST
 518                  : 0);
 519     }
 520     if (!error) {
 521         *dpifp = create_dpif_netdev(dp);
 522     }
 523     ovs_mutex_unlock(&dp_netdev_mutex);
 524
 525     return error;
 526 }
 527
 528 static void
 529 dp_netdev_purge_queues(struct dp_netdev *dp)
 530     OVS_REQ_WRLOCK(dp->queue_rwlock)
 531 {
 532     int i;
 533
 534     for (i = 0; i < dp->n_handlers; i++) {
 535         struct dp_netdev_queue *q = &dp->handler_queues[i];
 536
 537         ovs_mutex_lock(&q->mutex);
 538         while (q->tail != q->head) {
 539             struct dp_netdev_upcall *u = &q->upcalls[q->tail++ & QUEUE_MASK];
 540             ofpbuf_uninit(&u->upcall.packet);
 541             ofpbuf_uninit(&u->buf);
 542         }
 543         ovs_mutex_unlock(&q->mutex);
 544     }
 545 }
 546
 547 /* Requires dp_netdev_mutex so that we can't get a new reference to 'dp'
 548  * through the 'dp_netdevs' shash while freeing 'dp'. */
 549 static void
 550 dp_netdev_free(struct dp_netdev *dp)
 551     OVS_REQUIRES(dp_netdev_mutex)
 552 {
 553     struct dp_netdev_port *port, *next;
 554     struct dp_netdev_stats *bucket;
 555     int i;
 556
 557     shash_find_and_delete(&dp_netdevs, dp->name);
 558
 559     dp_netdev_set_pmd_threads(dp, 0);
 560     free(dp->pmd_threads);
 561
 562     dp_netdev_flow_flush(dp);
 563     ovs_rwlock_wrlock(&dp->port_rwlock);
 564     HMAP_FOR_EACH_SAFE (port, next, node, &dp->ports) {
 565         do_del_port(dp, port->port_no);
 566     }
 567     ovs_rwlock_unlock(&dp->port_rwlock);
 568
 569     OVSTHREAD_STATS_FOR_EACH_BUCKET (bucket, i, &dp->stats) {
 570         ovs_mutex_destroy(&bucket->mutex);
 571         free_cacheline(bucket);
 572     }
 573     ovsthread_stats_destroy(&dp->stats);
 574
 575     fat_rwlock_wrlock(&dp->queue_rwlock);
 576     dp_netdev_destroy_all_queues(dp);
 577     fat_rwlock_unlock(&dp->queue_rwlock);
 578
 579     fat_rwlock_destroy(&dp->queue_rwlock);
 580
 581     classifier_destroy(&dp->cls);
 582     hmap_destroy(&dp->flow_table);
 583     ovs_mutex_destroy(&dp->flow_mutex);
 584     seq_destroy(dp->port_seq);
 585     hmap_destroy(&dp->ports);
 586     latch_destroy(&dp->exit_latch);
 587     free(CONST_CAST(char *, dp->name));
 588     free(dp);
 589 }
 590
 591 static void
 592 dp_netdev_unref(struct dp_netdev *dp)
 593 {
 594     if (dp) {
 595         /* Take dp_netdev_mutex so that, if dp->ref_cnt falls to zero, we can't
 596          * get a new reference to 'dp' through the 'dp_netdevs' shash. */
 597         ovs_mutex_lock(&dp_netdev_mutex);
 598         if (ovs_refcount_unref(&dp->ref_cnt) == 1) {
 599             dp_netdev_free(dp);
 600         }
 601         ovs_mutex_unlock(&dp_netdev_mutex);
 602     }
 603 }
 604
 605 static void
 606 dpif_netdev_close(struct dpif *dpif)
 607 {
 608     struct dp_netdev *dp = get_dp_netdev(dpif);
 609
 610     dp_netdev_unref(dp);
 611     free(dpif);
 612 }
 613
 614 static int
 615 dpif_netdev_destroy(struct dpif *dpif)
 616 {
 617     struct dp_netdev *dp = get_dp_netdev(dpif);
 618
 619     if (!atomic_flag_test_and_set(&dp->destroyed)) {
 620         if (ovs_refcount_unref(&dp->ref_cnt) == 1) {
 621             /* Can't happen: 'dpif' still owns a reference to 'dp'. */
 622             OVS_NOT_REACHED();
 623         }
 624     }
 625
 626     return 0;
 627 }
 628
 629 static int
 630 dpif_netdev_get_stats(const struct dpif *dpif, struct dpif_dp_stats *stats)
 631 {
 632     struct dp_netdev *dp = get_dp_netdev(dpif);
 633     struct dp_netdev_stats *bucket;
 634     size_t i;
 635
 636     fat_rwlock_rdlock(&dp->cls.rwlock);
 637     stats->n_flows = hmap_count(&dp->flow_table);
 638     fat_rwlock_unlock(&dp->cls.rwlock);
 639
 640     stats->n_hit = stats->n_missed = stats->n_lost = 0;
 641     OVSTHREAD_STATS_FOR_EACH_BUCKET (bucket, i, &dp->stats) {
 642         ovs_mutex_lock(&bucket->mutex);
 643         stats->n_hit += bucket->n[DP_STAT_HIT];
 644         stats->n_missed += bucket->n[DP_STAT_MISS];
 645         stats->n_lost += bucket->n[DP_STAT_LOST];
 646         ovs_mutex_unlock(&bucket->mutex);
 647     }
 648     stats->n_masks = UINT32_MAX;
 649     stats->n_mask_hit = UINT64_MAX;
 650
 651     return 0;
 652 }
 653
 654 static void
 655 dp_netdev_reload_pmd_threads(struct dp_netdev *dp)
 656 {
 657     int i;
 658
 659     for (i = 0; i < dp->n_pmd_threads; i++) {
 660         struct pmd_thread *f = &dp->pmd_threads[i];
 661         int id;
 662
 663         atomic_add(&f->change_seq, 1, &id);
 664    }
 665 }
 666
 667 static int
 668 do_add_port(struct dp_netdev *dp, const char *devname, const char *type,
 669             odp_port_t port_no)
 670     OVS_REQ_WRLOCK(dp->port_rwlock)
 671 {
 672     struct netdev_saved_flags *sf;
 673     struct dp_netdev_port *port;
 674     struct netdev *netdev;
 675     enum netdev_flags flags;
 676     const char *open_type;
 677     int error;
 678     int i;
 679
 680     /* XXX reject devices already in some dp_netdev. */
 681
 682     /* Open and validate network device. */
 683     open_type = dpif_netdev_port_open_type(dp->class, type);
 684     error = netdev_open(devname, open_type, &netdev);
 685     if (error) {
 686         return error;
 687     }
 688     /* XXX reject non-Ethernet devices */
 689
 690     netdev_get_flags(netdev, &flags);
 691     if (flags & NETDEV_LOOPBACK) {
 692         VLOG_ERR("%s: cannot add a loopback device", devname);
 693         netdev_close(netdev);
 694         return EINVAL;
 695     }
 696
 697     port = xzalloc(sizeof *port);
 698     port->port_no = port_no;
 699     port->netdev = netdev;
 700     port->rxq = xmalloc(sizeof *port->rxq * netdev_n_rxq(netdev));
 701     port->type = xstrdup(type);
 702     for (i = 0; i < netdev_n_rxq(netdev); i++) {
 703         error = netdev_rxq_open(netdev, &port->rxq[i], i);
 704         if (error
 705             && !(error == EOPNOTSUPP && dpif_netdev_class_is_dummy(dp->class))) {
 706             VLOG_ERR("%s: cannot receive packets on this network device (%s)",
 707                      devname, ovs_strerror(errno));
 708             netdev_close(netdev);
 709             return error;
 710         }
 711     }
 712
 713     error = netdev_turn_flags_on(netdev, NETDEV_PROMISC, &sf);
 714     if (error) {
 715         for (i = 0; i < netdev_n_rxq(netdev); i++) {
 716             netdev_rxq_close(port->rxq[i]);
 717         }
 718         netdev_close(netdev);
 719         free(port->rxq);
 720         free(port);
 721         return error;
 722     }
 723     port->sf = sf;
 724
 725     if (netdev_is_pmd(netdev)) {
 726         dp->pmd_count++;
 727         dp_netdev_set_pmd_threads(dp, NR_THREADS);
 728         dp_netdev_reload_pmd_threads(dp);
 729     }
 730     ovs_refcount_init(&port->ref_cnt);
 731
 732     hmap_insert(&dp->ports, &port->node, hash_int(odp_to_u32(port_no), 0));
 733     seq_change(dp->port_seq);
 734
 735     return 0;
 736 }
 737
 738 static int
 739 dpif_netdev_port_add(struct dpif *dpif, struct netdev *netdev,
 740                      odp_port_t *port_nop)
 741 {
 742     struct dp_netdev *dp = get_dp_netdev(dpif);
 743     char namebuf[NETDEV_VPORT_NAME_BUFSIZE];
 744     const char *dpif_port;
 745     odp_port_t port_no;
 746     int error;
 747
 748     ovs_rwlock_wrlock(&dp->port_rwlock);
 749     dpif_port = netdev_vport_get_dpif_port(netdev, namebuf, sizeof namebuf);
 750     if (*port_nop != ODPP_NONE) {
 751         port_no = *port_nop;
 752         error = dp_netdev_lookup_port(dp, *port_nop) ? EBUSY : 0;
 753     } else {
 754         port_no = choose_port(dp, dpif_port);
 755         error = port_no == ODPP_NONE ? EFBIG : 0;
 756     }
 757     if (!error) {
 758         *port_nop = port_no;
 759         error = do_add_port(dp, dpif_port, netdev_get_type(netdev), port_no);
 760     }
 761     ovs_rwlock_unlock(&dp->port_rwlock);
 762
 763     return error;
 764 }
 765
 766 static int
 767 dpif_netdev_port_del(struct dpif *dpif, odp_port_t port_no)
 768 {
 769     struct dp_netdev *dp = get_dp_netdev(dpif);
 770     int error;
 771
 772     ovs_rwlock_wrlock(&dp->port_rwlock);
 773     error = port_no == ODPP_LOCAL ? EINVAL : do_del_port(dp, port_no);
 774     ovs_rwlock_unlock(&dp->port_rwlock);
 775
 776     return error;
 777 }
 778
 779 static bool
 780 is_valid_port_number(odp_port_t port_no)
 781 {
 782     return port_no != ODPP_NONE;
 783 }
 784
 785 static struct dp_netdev_port *
 786 dp_netdev_lookup_port(const struct dp_netdev *dp, odp_port_t port_no)
 787     OVS_REQ_RDLOCK(dp->port_rwlock)
 788 {
 789     struct dp_netdev_port *port;
 790
 791     HMAP_FOR_EACH_IN_BUCKET (port, node, hash_int(odp_to_u32(port_no), 0),
 792                              &dp->ports) {
 793         if (port->port_no == port_no) {
 794             return port;
 795         }
 796     }
 797     return NULL;
 798 }
 799
 800 static int
 801 get_port_by_number(struct dp_netdev *dp,
 802                    odp_port_t port_no, struct dp_netdev_port **portp)
 803     OVS_REQ_RDLOCK(dp->port_rwlock)
 804 {
 805     if (!is_valid_port_number(port_no)) {
 806         *portp = NULL;
 807         return EINVAL;
 808     } else {
 809         *portp = dp_netdev_lookup_port(dp, port_no);
 810         return *portp ? 0 : ENOENT;
 811     }
 812 }
 813
 814 static void
 815 port_ref(struct dp_netdev_port *port)
 816 {
 817     if (port) {
 818         ovs_refcount_ref(&port->ref_cnt);
 819     }
 820 }
 821
 822 static void
 823 port_unref(struct dp_netdev_port *port)
 824 {
 825     if (port && ovs_refcount_unref(&port->ref_cnt) == 1) {
 826         int i;
 827
 828         netdev_close(port->netdev);
 829         netdev_restore_flags(port->sf);
 830
 831         for (i = 0; i < netdev_n_rxq(port->netdev); i++) {
 832             netdev_rxq_close(port->rxq[i]);
 833         }
 834         free(port->type);
 835         free(port);
 836     }
 837 }
 838
 839 static int
 840 get_port_by_name(struct dp_netdev *dp,
 841                  const char *devname, struct dp_netdev_port **portp)
 842     OVS_REQ_RDLOCK(dp->port_rwlock)
 843 {
 844     struct dp_netdev_port *port;
 845
 846     HMAP_FOR_EACH (port, node, &dp->ports) {
 847         if (!strcmp(netdev_get_name(port->netdev), devname)) {
 848             *portp = port;
 849             return 0;
 850         }
 851     }
 852     return ENOENT;
 853 }
 854
 855 static int
 856 do_del_port(struct dp_netdev *dp, odp_port_t port_no)
 857     OVS_REQ_WRLOCK(dp->port_rwlock)
 858 {
 859     struct dp_netdev_port *port;
 860     int error;
 861
 862     error = get_port_by_number(dp, port_no, &port);
 863     if (error) {
 864         return error;
 865     }
 866
 867     hmap_remove(&dp->ports, &port->node);
 868     seq_change(dp->port_seq);
 869     if (netdev_is_pmd(port->netdev)) {
 870         dp_netdev_reload_pmd_threads(dp);
 871     }
 872
 873     port_unref(port);
 874     return 0;
 875 }
 876
 877 static void
 878 answer_port_query(const struct dp_netdev_port *port,
 879                   struct dpif_port *dpif_port)
 880 {
 881     dpif_port->name = xstrdup(netdev_get_name(port->netdev));
 882     dpif_port->type = xstrdup(port->type);
 883     dpif_port->port_no = port->port_no;
 884 }
 885
 886 static int
 887 dpif_netdev_port_query_by_number(const struct dpif *dpif, odp_port_t port_no,
 888                                  struct dpif_port *dpif_port)
 889 {
 890     struct dp_netdev *dp = get_dp_netdev(dpif);
 891     struct dp_netdev_port *port;
 892     int error;
 893
 894     ovs_rwlock_rdlock(&dp->port_rwlock);
 895     error = get_port_by_number(dp, port_no, &port);
 896     if (!error && dpif_port) {
 897         answer_port_query(port, dpif_port);
 898     }
 899     ovs_rwlock_unlock(&dp->port_rwlock);
 900
 901     return error;
 902 }
 903
 904 static int
 905 dpif_netdev_port_query_by_name(const struct dpif *dpif, const char *devname,
 906                                struct dpif_port *dpif_port)
 907 {
 908     struct dp_netdev *dp = get_dp_netdev(dpif);
 909     struct dp_netdev_port *port;
 910     int error;
 911
 912     ovs_rwlock_rdlock(&dp->port_rwlock);
 913     error = get_port_by_name(dp, devname, &port);
 914     if (!error && dpif_port) {
 915         answer_port_query(port, dpif_port);
 916     }
 917     ovs_rwlock_unlock(&dp->port_rwlock);
 918
 919     return error;
 920 }
 921
 922 static void
 923 dp_netdev_flow_free(struct dp_netdev_flow *flow)
 924 {
 925     struct dp_netdev_flow_stats *bucket;
 926     size_t i;
 927
 928     OVSTHREAD_STATS_FOR_EACH_BUCKET (bucket, i, &flow->stats) {
 929         ovs_mutex_destroy(&bucket->mutex);
 930         free_cacheline(bucket);
 931     }
 932     ovsthread_stats_destroy(&flow->stats);
 933
 934     cls_rule_destroy(CONST_CAST(struct cls_rule *, &flow->cr));
 935     dp_netdev_actions_free(dp_netdev_flow_get_actions(flow));
 936     ovs_mutex_destroy(&flow->mutex);
 937     free(flow);
 938 }
 939
 940 static void
 941 dp_netdev_remove_flow(struct dp_netdev *dp, struct dp_netdev_flow *flow)
 942     OVS_REQ_WRLOCK(dp->cls.rwlock)
 943     OVS_REQUIRES(dp->flow_mutex)
 944 {
 945     struct cls_rule *cr = CONST_CAST(struct cls_rule *, &flow->cr);
 946     struct hmap_node *node = CONST_CAST(struct hmap_node *, &flow->node);
 947
 948     classifier_remove(&dp->cls, cr);
 949     hmap_remove(&dp->flow_table, node);
 950     ovsrcu_postpone(dp_netdev_flow_free, flow);
 951 }
 952
 953 static void
 954 dp_netdev_flow_flush(struct dp_netdev *dp)
 955 {
 956     struct dp_netdev_flow *netdev_flow, *next;
 957
 958     ovs_mutex_lock(&dp->flow_mutex);
 959     fat_rwlock_wrlock(&dp->cls.rwlock);
 960     HMAP_FOR_EACH_SAFE (netdev_flow, next, node, &dp->flow_table) {
 961         dp_netdev_remove_flow(dp, netdev_flow);
 962     }
 963     fat_rwlock_unlock(&dp->cls.rwlock);
 964     ovs_mutex_unlock(&dp->flow_mutex);
 965 }
 966
 967 static int
 968 dpif_netdev_flow_flush(struct dpif *dpif)
 969 {
 970     struct dp_netdev *dp = get_dp_netdev(dpif);
 971
 972     dp_netdev_flow_flush(dp);
 973     return 0;
 974 }
 975
 976 struct dp_netdev_port_state {
 977     uint32_t bucket;
 978     uint32_t offset;
 979     char *name;
 980 };
 981
 982 static int
 983 dpif_netdev_port_dump_start(const struct dpif *dpif OVS_UNUSED, void **statep)
 984 {
 985     *statep = xzalloc(sizeof(struct dp_netdev_port_state));
 986     return 0;
 987 }
 988
 989 static int
 990 dpif_netdev_port_dump_next(const struct dpif *dpif, void *state_,
 991                            struct dpif_port *dpif_port)
 992 {
 993     struct dp_netdev_port_state *state = state_;
 994     struct dp_netdev *dp = get_dp_netdev(dpif);
 995     struct hmap_node *node;
 996     int retval;
 997
 998     ovs_rwlock_rdlock(&dp->port_rwlock);
 999     node = hmap_at_position(&dp->ports, &state->bucket, &state->offset);
1000     if (node) {
1001         struct dp_netdev_port *port;
1002
1003         port = CONTAINER_OF(node, struct dp_netdev_port, node);
1004
1005         free(state->name);
1006         state->name = xstrdup(netdev_get_name(port->netdev));
1007         dpif_port->name = state->name;
1008         dpif_port->type = port->type;
1009         dpif_port->port_no = port->port_no;
1010
1011         retval = 0;
1012     } else {
1013         retval = EOF;
1014     }
1015     ovs_rwlock_unlock(&dp->port_rwlock);
1016
1017     return retval;
1018 }
1019
1020 static int
1021 dpif_netdev_port_dump_done(const struct dpif *dpif OVS_UNUSED, void *state_)
1022 {
1023     struct dp_netdev_port_state *state = state_;
1024     free(state->name);
1025     free(state);
1026     return 0;
1027 }
1028
1029 static int
1030 dpif_netdev_port_poll(const struct dpif *dpif_, char **devnamep OVS_UNUSED)
1031 {
1032     struct dpif_netdev *dpif = dpif_netdev_cast(dpif_);
1033     uint64_t new_port_seq;
1034     int error;
1035
1036     new_port_seq = seq_read(dpif->dp->port_seq);
1037     if (dpif->last_port_seq != new_port_seq) {
1038         dpif->last_port_seq = new_port_seq;
1039         error = ENOBUFS;
1040     } else {
1041         error = EAGAIN;
1042     }
1043
1044     return error;
1045 }
1046
1047 static void
1048 dpif_netdev_port_poll_wait(const struct dpif *dpif_)
1049 {
1050     struct dpif_netdev *dpif = dpif_netdev_cast(dpif_);
1051
1052     seq_wait(dpif->dp->port_seq, dpif->last_port_seq);
1053 }
1054
1055 static struct dp_netdev_flow *
1056 dp_netdev_flow_cast(const struct cls_rule *cr)
1057 {
1058     return cr ? CONTAINER_OF(cr, struct dp_netdev_flow, cr) : NULL;
1059 }
1060
1061 static struct dp_netdev_flow *
1062 dp_netdev_lookup_flow(const struct dp_netdev *dp, const struct flow *flow)
1063     OVS_EXCLUDED(dp->cls.rwlock)
1064 {
1065     struct dp_netdev_flow *netdev_flow;
1066
1067     fat_rwlock_rdlock(&dp->cls.rwlock);
1068     netdev_flow = dp_netdev_flow_cast(classifier_lookup(&dp->cls, flow, NULL));
1069     fat_rwlock_unlock(&dp->cls.rwlock);
1070
1071     return netdev_flow;
1072 }
1073
1074 static struct dp_netdev_flow *
1075 dp_netdev_find_flow(const struct dp_netdev *dp, const struct flow *flow)
1076     OVS_REQ_RDLOCK(dp->cls.rwlock)
1077 {
1078     struct dp_netdev_flow *netdev_flow;
1079
1080     HMAP_FOR_EACH_WITH_HASH (netdev_flow, node, flow_hash(flow, 0),
1081                              &dp->flow_table) {
1082         if (flow_equal(&netdev_flow->flow, flow)) {
1083             return netdev_flow;
1084         }
1085     }
1086
1087     return NULL;
1088 }
1089
1090 static void
1091 get_dpif_flow_stats(struct dp_netdev_flow *netdev_flow,
1092                     struct dpif_flow_stats *stats)
1093 {
1094     struct dp_netdev_flow_stats *bucket;
1095     size_t i;
1096
1097     memset(stats, 0, sizeof *stats);
1098     OVSTHREAD_STATS_FOR_EACH_BUCKET (bucket, i, &netdev_flow->stats) {
1099         ovs_mutex_lock(&bucket->mutex);
1100         stats->n_packets += bucket->packet_count;
1101         stats->n_bytes += bucket->byte_count;
1102         stats->used = MAX(stats->used, bucket->used);
1103         stats->tcp_flags |= bucket->tcp_flags;
1104         ovs_mutex_unlock(&bucket->mutex);
1105     }
1106 }
1107
1108 static int
1109 dpif_netdev_mask_from_nlattrs(const struct nlattr *key, uint32_t key_len,
1110                               const struct nlattr *mask_key,
1111                               uint32_t mask_key_len, const struct flow *flow,
1112                               struct flow *mask)
1113 {
1114     if (mask_key_len) {
1115         enum odp_key_fitness fitness;
1116
1117         fitness = odp_flow_key_to_mask(mask_key, mask_key_len, mask, flow);
1118         if (fitness) {
1119             /* This should not happen: it indicates that
1120              * odp_flow_key_from_mask() and odp_flow_key_to_mask()
1121              * disagree on the acceptable form of a mask.  Log the problem
1122              * as an error, with enough details to enable debugging. */
1123             static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(1, 5);
1124
1125             if (!VLOG_DROP_ERR(&rl)) {
1126                 struct ds s;
1127
1128                 ds_init(&s);
1129                 odp_flow_format(key, key_len, mask_key, mask_key_len, NULL, &s,
1130                                 true);
1131                 VLOG_ERR("internal error parsing flow mask %s (%s)",
1132                          ds_cstr(&s), odp_key_fitness_to_string(fitness));
1133                 ds_destroy(&s);
1134             }
1135
1136             return EINVAL;
1137         }
1138         /* Force unwildcard the in_port. */
1139         mask->in_port.odp_port = u32_to_odp(UINT32_MAX);
1140     } else {
1141         enum mf_field_id id;
1142         /* No mask key, unwildcard everything except fields whose
1143          * prerequisities are not met. */
1144         memset(mask, 0x0, sizeof *mask);
1145
1146         for (id = 0; id < MFF_N_IDS; ++id) {
1147             /* Skip registers and metadata. */
1148             if (!(id >= MFF_REG0 && id < MFF_REG0 + FLOW_N_REGS)
1149                 && id != MFF_METADATA) {
1150                 const struct mf_field *mf = mf_from_id(id);
1151                 if (mf_are_prereqs_ok(mf, flow)) {
1152                     mf_mask_field(mf, mask);
1153                 }
1154             }
1155         }
1156     }
1157
1158     return 0;
1159 }
1160
1161 static int
1162 dpif_netdev_flow_from_nlattrs(const struct nlattr *key, uint32_t key_len,
1163                               struct flow *flow)
1164 {
1165     odp_port_t in_port;
1166
1167     if (odp_flow_key_to_flow(key, key_len, flow)) {
1168         /* This should not happen: it indicates that odp_flow_key_from_flow()
1169          * and odp_flow_key_to_flow() disagree on the acceptable form of a
1170          * flow.  Log the problem as an error, with enough details to enable
1171          * debugging. */
1172         static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(1, 5);
1173
1174         if (!VLOG_DROP_ERR(&rl)) {
1175             struct ds s;
1176
1177             ds_init(&s);
1178             odp_flow_format(key, key_len, NULL, 0, NULL, &s, true);
1179             VLOG_ERR("internal error parsing flow key %s", ds_cstr(&s));
1180             ds_destroy(&s);
1181         }
1182
1183         return EINVAL;
1184     }
1185
1186     in_port = flow->in_port.odp_port;
1187     if (!is_valid_port_number(in_port) && in_port != ODPP_NONE) {
1188         return EINVAL;
1189     }
1190
1191     return 0;
1192 }
1193
1194 static int
1195 dpif_netdev_flow_get(const struct dpif *dpif,
1196                      const struct nlattr *nl_key, size_t nl_key_len,
1197                      struct ofpbuf **actionsp, struct dpif_flow_stats *stats)
1198 {
1199     struct dp_netdev *dp = get_dp_netdev(dpif);
1200     struct dp_netdev_flow *netdev_flow;
1201     struct flow key;
1202     int error;
1203
1204     error = dpif_netdev_flow_from_nlattrs(nl_key, nl_key_len, &key);
1205     if (error) {
1206         return error;
1207     }
1208
1209     fat_rwlock_rdlock(&dp->cls.rwlock);
1210     netdev_flow = dp_netdev_find_flow(dp, &key);
1211     fat_rwlock_unlock(&dp->cls.rwlock);
1212
1213     if (netdev_flow) {
1214         if (stats) {
1215             get_dpif_flow_stats(netdev_flow, stats);
1216         }
1217
1218         if (actionsp) {
1219             struct dp_netdev_actions *actions;
1220
1221             actions = dp_netdev_flow_get_actions(netdev_flow);
1222             *actionsp = ofpbuf_clone_data(actions->actions, actions->size);
1223         }
1224      } else {
1225         error = ENOENT;
1226     }
1227
1228     return error;
1229 }
1230
1231 static int
1232 dp_netdev_flow_add(struct dp_netdev *dp, const struct flow *flow,
1233                    const struct flow_wildcards *wc,
1234                    const struct nlattr *actions,
1235                    size_t actions_len)
1236     OVS_REQUIRES(dp->flow_mutex)
1237 {
1238     struct dp_netdev_flow *netdev_flow;
1239     struct match match;
1240
1241     netdev_flow = xzalloc(sizeof *netdev_flow);
1242     *CONST_CAST(struct flow *, &netdev_flow->flow) = *flow;
1243
1244     ovs_mutex_init(&netdev_flow->mutex);
1245
1246     ovsthread_stats_init(&netdev_flow->stats);
1247
1248     ovsrcu_set(&netdev_flow->actions,
1249                dp_netdev_actions_create(actions, actions_len));
1250
1251     match_init(&match, flow, wc);
1252     cls_rule_init(CONST_CAST(struct cls_rule *, &netdev_flow->cr),
1253                   &match, NETDEV_RULE_PRIORITY);
1254     fat_rwlock_wrlock(&dp->cls.rwlock);
1255     classifier_insert(&dp->cls,
1256                       CONST_CAST(struct cls_rule *, &netdev_flow->cr));
1257     hmap_insert(&dp->flow_table,
1258                 CONST_CAST(struct hmap_node *, &netdev_flow->node),
1259                 flow_hash(flow, 0));
1260     fat_rwlock_unlock(&dp->cls.rwlock);
1261
1262     return 0;
1263 }
1264
1265 static void
1266 clear_stats(struct dp_netdev_flow *netdev_flow)
1267 {
1268     struct dp_netdev_flow_stats *bucket;
1269     size_t i;
1270
1271     OVSTHREAD_STATS_FOR_EACH_BUCKET (bucket, i, &netdev_flow->stats) {
1272         ovs_mutex_lock(&bucket->mutex);
1273         bucket->used = 0;
1274         bucket->packet_count = 0;
1275         bucket->byte_count = 0;
1276         bucket->tcp_flags = 0;
1277         ovs_mutex_unlock(&bucket->mutex);
1278     }
1279 }
1280
1281 static int
1282 dpif_netdev_flow_put(struct dpif *dpif, const struct dpif_flow_put *put)
1283 {
1284     struct dp_netdev *dp = get_dp_netdev(dpif);
1285     struct dp_netdev_flow *netdev_flow;
1286     struct flow flow;
1287     struct flow_wildcards wc;
1288     int error;
1289
1290     error = dpif_netdev_flow_from_nlattrs(put->key, put->key_len, &flow);
1291     if (error) {
1292         return error;
1293     }
1294     error = dpif_netdev_mask_from_nlattrs(put->key, put->key_len,
1295                                           put->mask, put->mask_len,
1296                                           &flow, &wc.masks);
1297     if (error) {
1298         return error;
1299     }
1300
1301     ovs_mutex_lock(&dp->flow_mutex);
1302     netdev_flow = dp_netdev_lookup_flow(dp, &flow);
1303     if (!netdev_flow) {
1304         if (put->flags & DPIF_FP_CREATE) {
1305             if (hmap_count(&dp->flow_table) < MAX_FLOWS) {
1306                 if (put->stats) {
1307                     memset(put->stats, 0, sizeof *put->stats);
1308                 }
1309                 error = dp_netdev_flow_add(dp, &flow, &wc, put->actions,
1310                                            put->actions_len);
1311             } else {
1312                 error = EFBIG;
1313             }
1314         } else {
1315             error = ENOENT;
1316         }
1317     } else {
1318         if (put->flags & DPIF_FP_MODIFY
1319             && flow_equal(&flow, &netdev_flow->flow)) {
1320             struct dp_netdev_actions *new_actions;
1321             struct dp_netdev_actions *old_actions;
1322
1323             new_actions = dp_netdev_actions_create(put->actions,
1324                                                    put->actions_len);
1325
1326             old_actions = dp_netdev_flow_get_actions(netdev_flow);
1327             ovsrcu_set(&netdev_flow->actions, new_actions);
1328
1329             if (put->stats) {
1330                 get_dpif_flow_stats(netdev_flow, put->stats);
1331             }
1332             if (put->flags & DPIF_FP_ZERO_STATS) {
1333                 clear_stats(netdev_flow);
1334             }
1335
1336             ovsrcu_postpone(dp_netdev_actions_free, old_actions);
1337         } else if (put->flags & DPIF_FP_CREATE) {
1338             error = EEXIST;
1339         } else {
1340             /* Overlapping flow. */
1341             error = EINVAL;
1342         }
1343     }
1344     ovs_mutex_unlock(&dp->flow_mutex);
1345
1346     return error;
1347 }
1348
1349 static int
1350 dpif_netdev_flow_del(struct dpif *dpif, const struct dpif_flow_del *del)
1351 {
1352     struct dp_netdev *dp = get_dp_netdev(dpif);
1353     struct dp_netdev_flow *netdev_flow;
1354     struct flow key;
1355     int error;
1356
1357     error = dpif_netdev_flow_from_nlattrs(del->key, del->key_len, &key);
1358     if (error) {
1359         return error;
1360     }
1361
1362     ovs_mutex_lock(&dp->flow_mutex);
1363     fat_rwlock_wrlock(&dp->cls.rwlock);
1364     netdev_flow = dp_netdev_find_flow(dp, &key);
1365     if (netdev_flow) {
1366         if (del->stats) {
1367             get_dpif_flow_stats(netdev_flow, del->stats);
1368         }
1369         dp_netdev_remove_flow(dp, netdev_flow);
1370     } else {
1371         error = ENOENT;
1372     }
1373     fat_rwlock_unlock(&dp->cls.rwlock);
1374     ovs_mutex_unlock(&dp->flow_mutex);
1375
1376     return error;
1377 }
1378
1379 struct dp_netdev_flow_state {
1380     struct dp_netdev_actions *actions;
1381     struct odputil_keybuf keybuf;
1382     struct odputil_keybuf maskbuf;
1383     struct dpif_flow_stats stats;
1384 };
1385
1386 struct dp_netdev_flow_iter {
1387     uint32_t bucket;
1388     uint32_t offset;
1389     int status;
1390     struct ovs_mutex mutex;
1391 };
1392
1393 static void
1394 dpif_netdev_flow_dump_state_init(void **statep)
1395 {
1396     struct dp_netdev_flow_state *state;
1397
1398     *statep = state = xmalloc(sizeof *state);
1399     state->actions = NULL;
1400 }
1401
1402 static void
1403 dpif_netdev_flow_dump_state_uninit(void *state_)
1404 {
1405     struct dp_netdev_flow_state *state = state_;
1406
1407     free(state);
1408 }
1409
1410 static int
1411 dpif_netdev_flow_dump_start(const struct dpif *dpif OVS_UNUSED, void **iterp)
1412 {
1413     struct dp_netdev_flow_iter *iter;
1414
1415     *iterp = iter = xmalloc(sizeof *iter);
1416     iter->bucket = 0;
1417     iter->offset = 0;
1418     iter->status = 0;
1419     ovs_mutex_init(&iter->mutex);
1420     return 0;
1421 }
1422
1423 /* XXX the caller must use 'actions' without quiescing */
1424 static int
1425 dpif_netdev_flow_dump_next(const struct dpif *dpif, void *iter_, void *state_,
1426                            const struct nlattr **key, size_t *key_len,
1427                            const struct nlattr **mask, size_t *mask_len,
1428                            const struct nlattr **actions, size_t *actions_len,
1429                            const struct dpif_flow_stats **stats)
1430 {
1431     struct dp_netdev_flow_iter *iter = iter_;
1432     struct dp_netdev_flow_state *state = state_;
1433     struct dp_netdev *dp = get_dp_netdev(dpif);
1434     struct dp_netdev_flow *netdev_flow;
1435     int error;
1436
1437     ovs_mutex_lock(&iter->mutex);
1438     error = iter->status;
1439     if (!error) {
1440         struct hmap_node *node;
1441
1442         fat_rwlock_rdlock(&dp->cls.rwlock);
1443         node = hmap_at_position(&dp->flow_table, &iter->bucket, &iter->offset);
1444         if (node) {
1445             netdev_flow = CONTAINER_OF(node, struct dp_netdev_flow, node);
1446         }
1447         fat_rwlock_unlock(&dp->cls.rwlock);
1448         if (!node) {
1449             iter->status = error = EOF;
1450         }
1451     }
1452     ovs_mutex_unlock(&iter->mutex);
1453     if (error) {
1454         return error;
1455     }
1456
1457     if (key) {
1458         struct ofpbuf buf;
1459
1460         ofpbuf_use_stack(&buf, &state->keybuf, sizeof state->keybuf);
1461         odp_flow_key_from_flow(&buf, &netdev_flow->flow,
1462                                netdev_flow->flow.in_port.odp_port);
1463
1464         *key = buf.data;
1465         *key_len = buf.size;
1466     }
1467
1468     if (key && mask) {
1469         struct ofpbuf buf;
1470         struct flow_wildcards wc;
1471
1472         ofpbuf_use_stack(&buf, &state->maskbuf, sizeof state->maskbuf);
1473         minimask_expand(&netdev_flow->cr.match.mask, &wc);
1474         odp_flow_key_from_mask(&buf, &wc.masks, &netdev_flow->flow,
1475                                odp_to_u32(wc.masks.in_port.odp_port),
1476                                SIZE_MAX);
1477
1478         *mask = buf.data;
1479         *mask_len = buf.size;
1480     }
1481
1482     if (actions || stats) {
1483         state->actions = NULL;
1484
1485         if (actions) {
1486             state->actions = dp_netdev_flow_get_actions(netdev_flow);
1487             *actions = state->actions->actions;
1488             *actions_len = state->actions->size;
1489         }
1490
1491         if (stats) {
1492             get_dpif_flow_stats(netdev_flow, &state->stats);
1493             *stats = &state->stats;
1494         }
1495     }
1496
1497     return 0;
1498 }
1499
1500 static int
1501 dpif_netdev_flow_dump_done(const struct dpif *dpif OVS_UNUSED, void *iter_)
1502 {
1503     struct dp_netdev_flow_iter *iter = iter_;
1504
1505     ovs_mutex_destroy(&iter->mutex);
1506     free(iter);
1507     return 0;
1508 }
1509
1510 static int
1511 dpif_netdev_execute(struct dpif *dpif, struct dpif_execute *execute)
1512 {
1513     struct dp_netdev *dp = get_dp_netdev(dpif);
1514     struct pkt_metadata *md = &execute->md;
1515     struct flow key;
1516
1517     if (execute->packet->size < ETH_HEADER_LEN ||
1518         execute->packet->size > UINT16_MAX) {
1519         return EINVAL;
1520     }
1521
1522     /* Extract flow key. */
1523     flow_extract(execute->packet, md, &key);
1524
1525     ovs_rwlock_rdlock(&dp->port_rwlock);
1526     dp_netdev_execute_actions(dp, &key, execute->packet, false, md,
1527                               execute->actions, execute->actions_len);
1528     ovs_rwlock_unlock(&dp->port_rwlock);
1529
1530     return 0;
1531 }
1532
1533 static void
1534 dp_netdev_destroy_all_queues(struct dp_netdev *dp)
1535     OVS_REQ_WRLOCK(dp->queue_rwlock)
1536 {
1537     size_t i;
1538
1539     dp_netdev_purge_queues(dp);
1540
1541     for (i = 0; i < dp->n_handlers; i++) {
1542         struct dp_netdev_queue *q = &dp->handler_queues[i];
1543
1544         ovs_mutex_destroy(&q->mutex);
1545         seq_destroy(q->seq);
1546     }
1547     free(dp->handler_queues);
1548     dp->handler_queues = NULL;
1549     dp->n_handlers = 0;
1550 }
1551
1552 static void
1553 dp_netdev_refresh_queues(struct dp_netdev *dp, uint32_t n_handlers)
1554     OVS_REQ_WRLOCK(dp->queue_rwlock)
1555 {
1556     if (dp->n_handlers != n_handlers) {
1557         size_t i;
1558
1559         dp_netdev_destroy_all_queues(dp);
1560
1561         dp->n_handlers = n_handlers;
1562         dp->handler_queues = xzalloc(n_handlers * sizeof *dp->handler_queues);
1563
1564         for (i = 0; i < n_handlers; i++) {
1565             struct dp_netdev_queue *q = &dp->handler_queues[i];
1566
1567             ovs_mutex_init(&q->mutex);
1568             q->seq = seq_create();
1569         }
1570     }
1571 }
1572
1573 static int
1574 dpif_netdev_recv_set(struct dpif *dpif, bool enable)
1575 {
1576     struct dp_netdev *dp = get_dp_netdev(dpif);
1577
1578     if ((dp->handler_queues != NULL) == enable) {
1579         return 0;
1580     }
1581
1582     fat_rwlock_wrlock(&dp->queue_rwlock);
1583     if (!enable) {
1584         dp_netdev_destroy_all_queues(dp);
1585     } else {
1586         dp_netdev_refresh_queues(dp, 1);
1587     }
1588     fat_rwlock_unlock(&dp->queue_rwlock);
1589
1590     return 0;
1591 }
1592
1593 static int
1594 dpif_netdev_handlers_set(struct dpif *dpif, uint32_t n_handlers)
1595 {
1596     struct dp_netdev *dp = get_dp_netdev(dpif);
1597
1598     fat_rwlock_wrlock(&dp->queue_rwlock);
1599     if (dp->handler_queues) {
1600         dp_netdev_refresh_queues(dp, n_handlers);
1601     }
1602     fat_rwlock_unlock(&dp->queue_rwlock);
1603
1604     return 0;
1605 }
1606
1607 static int
1608 dpif_netdev_queue_to_priority(const struct dpif *dpif OVS_UNUSED,
1609                               uint32_t queue_id, uint32_t *priority)
1610 {
1611     *priority = queue_id;
1612     return 0;
1613 }
1614
1615 static bool
1616 dp_netdev_recv_check(const struct dp_netdev *dp, const uint32_t handler_id)
1617     OVS_REQ_RDLOCK(dp->queue_rwlock)
1618 {
1619     static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(1, 5);
1620
1621     if (!dp->handler_queues) {
1622         VLOG_WARN_RL(&rl, "receiving upcall disabled");
1623         return false;
1624     }
1625
1626     if (handler_id >= dp->n_handlers) {
1627         VLOG_WARN_RL(&rl, "handler index out of bound");
1628         return false;
1629     }
1630
1631     return true;
1632 }
1633
1634 static int
1635 dpif_netdev_recv(struct dpif *dpif, uint32_t handler_id,
1636                  struct dpif_upcall *upcall, struct ofpbuf *buf)
1637 {
1638     struct dp_netdev *dp = get_dp_netdev(dpif);
1639     struct dp_netdev_queue *q;
1640     int error = 0;
1641
1642     fat_rwlock_rdlock(&dp->queue_rwlock);
1643
1644     if (!dp_netdev_recv_check(dp, handler_id)) {
1645         error = EAGAIN;
1646         goto out;
1647     }
1648
1649     q = &dp->handler_queues[handler_id];
1650     ovs_mutex_lock(&q->mutex);
1651     if (q->head != q->tail) {
1652         struct dp_netdev_upcall *u = &q->upcalls[q->tail++ & QUEUE_MASK];
1653
1654         *upcall = u->upcall;
1655
1656         ofpbuf_uninit(buf);
1657         *buf = u->buf;
1658     } else {
1659         error = EAGAIN;
1660     }
1661     ovs_mutex_unlock(&q->mutex);
1662
1663 out:
1664     fat_rwlock_unlock(&dp->queue_rwlock);
1665
1666     return error;
1667 }
1668
1669 static void
1670 dpif_netdev_recv_wait(struct dpif *dpif, uint32_t handler_id)
1671 {
1672     struct dp_netdev *dp = get_dp_netdev(dpif);
1673     struct dp_netdev_queue *q;
1674     uint64_t seq;
1675
1676     fat_rwlock_rdlock(&dp->queue_rwlock);
1677
1678     if (!dp_netdev_recv_check(dp, handler_id)) {
1679         goto out;
1680     }
1681
1682     q = &dp->handler_queues[handler_id];
1683     ovs_mutex_lock(&q->mutex);
1684     seq = seq_read(q->seq);
1685     if (q->head != q->tail) {
1686         poll_immediate_wake();
1687     } else {
1688         seq_wait(q->seq, seq);
1689     }
1690
1691     ovs_mutex_unlock(&q->mutex);
1692
1693 out:
1694     fat_rwlock_unlock(&dp->queue_rwlock);
1695 }
1696
1697 static void
1698 dpif_netdev_recv_purge(struct dpif *dpif)
1699 {
1700     struct dpif_netdev *dpif_netdev = dpif_netdev_cast(dpif);
1701
1702     fat_rwlock_wrlock(&dpif_netdev->dp->queue_rwlock);
1703     dp_netdev_purge_queues(dpif_netdev->dp);
1704     fat_rwlock_unlock(&dpif_netdev->dp->queue_rwlock);
1705 }
1706 \f
1707 /* Creates and returns a new 'struct dp_netdev_actions', with a reference count
1708  * of 1, whose actions are a copy of from the 'ofpacts_len' bytes of
1709  * 'ofpacts'. */
1710 struct dp_netdev_actions *
1711 dp_netdev_actions_create(const struct nlattr *actions, size_t size)
1712 {
1713     struct dp_netdev_actions *netdev_actions;
1714
1715     netdev_actions = xmalloc(sizeof *netdev_actions);
1716     netdev_actions->actions = xmemdup(actions, size);
1717     netdev_actions->size = size;
1718
1719     return netdev_actions;
1720 }
1721
1722 struct dp_netdev_actions *
1723 dp_netdev_flow_get_actions(const struct dp_netdev_flow *flow)
1724 {
1725     return ovsrcu_get(struct dp_netdev_actions *, &flow->actions);
1726 }
1727
1728 static void
1729 dp_netdev_actions_free(struct dp_netdev_actions *actions)
1730 {
1731     free(actions->actions);
1732     free(actions);
1733 }
1734 \f
1735
1736 inline static void
1737 dp_netdev_process_rxq_port(struct dp_netdev *dp,
1738                           struct dp_netdev_port *port,
1739                           struct netdev_rxq *rxq)
1740 {
1741     struct ofpbuf *packet[NETDEV_MAX_RX_BATCH];
1742     int error, c;
1743
1744     error = netdev_rxq_recv(rxq, packet, &c);
1745     if (!error) {
1746         struct pkt_metadata md = PKT_METADATA_INITIALIZER(port->port_no);
1747         int i;
1748
1749         for (i = 0; i < c; i++) {
1750             dp_netdev_port_input(dp, packet[i], &md);
1751         }
1752     } else if (error != EAGAIN && error != EOPNOTSUPP) {
1753         static struct vlog_rate_limit rl
1754             = VLOG_RATE_LIMIT_INIT(1, 5);
1755
1756         VLOG_ERR_RL(&rl, "error receiving data from %s: %s",
1757                     netdev_get_name(port->netdev),
1758                     ovs_strerror(error));
1759     }
1760 }
1761
1762 static void
1763 dpif_netdev_run(struct dpif *dpif)
1764 {
1765     struct dp_netdev_port *port;
1766     struct dp_netdev *dp = get_dp_netdev(dpif);
1767
1768     ovs_rwlock_rdlock(&dp->port_rwlock);
1769
1770     HMAP_FOR_EACH (port, node, &dp->ports) {
1771         if (!netdev_is_pmd(port->netdev)) {
1772             int i;
1773
1774             for (i = 0; i < netdev_n_rxq(port->netdev); i++) {
1775                 dp_netdev_process_rxq_port(dp, port, port->rxq[i]);
1776             }
1777         }
1778     }
1779
1780     ovs_rwlock_unlock(&dp->port_rwlock);
1781 }
1782
1783 static void
1784 dpif_netdev_wait(struct dpif *dpif)
1785 {
1786     struct dp_netdev_port *port;
1787     struct dp_netdev *dp = get_dp_netdev(dpif);
1788
1789     ovs_rwlock_rdlock(&dp->port_rwlock);
1790
1791     HMAP_FOR_EACH (port, node, &dp->ports) {
1792         if (!netdev_is_pmd(port->netdev)) {
1793             int i;
1794
1795             for (i = 0; i < netdev_n_rxq(port->netdev); i++) {
1796                 netdev_rxq_wait(port->rxq[i]);
1797             }
1798         }
1799     }
1800     ovs_rwlock_unlock(&dp->port_rwlock);
1801 }
1802
1803 struct rxq_poll {
1804     struct dp_netdev_port *port;
1805     struct netdev_rxq *rx;
1806 };
1807
1808 static int
1809 pmd_load_queues(struct pmd_thread *f,
1810                 struct rxq_poll **ppoll_list, int poll_cnt)
1811 {
1812     struct dp_netdev *dp = f->dp;
1813     struct rxq_poll *poll_list = *ppoll_list;
1814     struct dp_netdev_port *port;
1815     int id = f->id;
1816     int index;
1817     int i;
1818
1819     /* Simple scheduler for netdev rx polling. */
1820     ovs_rwlock_rdlock(&dp->port_rwlock);
1821     for (i = 0; i < poll_cnt; i++) {
1822          port_unref(poll_list[i].port);
1823     }
1824
1825     poll_cnt = 0;
1826     index = 0;
1827
1828     HMAP_FOR_EACH (port, node, &f->dp->ports) {
1829         if (netdev_is_pmd(port->netdev)) {
1830             int i;
1831
1832             for (i = 0; i < netdev_n_rxq(port->netdev); i++) {
1833                 if ((index % dp->n_pmd_threads) == id) {
1834                     poll_list = xrealloc(poll_list, sizeof *poll_list * (poll_cnt + 1));
1835
1836                     port_ref(port);
1837                     poll_list[poll_cnt].port = port;
1838                     poll_list[poll_cnt].rx = port->rxq[i];
1839                     poll_cnt++;
1840                 }
1841                 index++;
1842             }
1843         }
1844     }
1845
1846     ovs_rwlock_unlock(&dp->port_rwlock);
1847     *ppoll_list = poll_list;
1848     return poll_cnt;
1849 }
1850
1851 static void *
1852 pmd_thread_main(void *f_)
1853 {
1854     struct pmd_thread *f = f_;
1855     struct dp_netdev *dp = f->dp;
1856     unsigned int lc = 0;
1857     struct rxq_poll *poll_list;
1858     unsigned int port_seq;
1859     int poll_cnt;
1860     int i;
1861
1862     f->name = xasprintf("pmd_%u", ovsthread_id_self());
1863     set_subprogram_name("%s", f->name);
1864     poll_cnt = 0;
1865     poll_list = NULL;
1866
1867 reload:
1868     poll_cnt = pmd_load_queues(f, &poll_list, poll_cnt);
1869     atomic_read(&f->change_seq, &port_seq);
1870
1871     for (;;) {
1872         unsigned int c_port_seq;
1873         int i;
1874
1875         for (i = 0; i < poll_cnt; i++) {
1876             dp_netdev_process_rxq_port(dp,  poll_list[i].port, poll_list[i].rx);
1877         }
1878
1879         if (lc++ > 1024) {
1880             ovsrcu_quiesce();
1881
1882             /* TODO: need completely userspace based signaling method.
1883              * to keep this thread entirely in userspace.
1884              * For now using atomic counter. */
1885             lc = 0;
1886             atomic_read_explicit(&f->change_seq, &c_port_seq, memory_order_consume);
1887             if (c_port_seq != port_seq) {
1888                 break;
1889             }
1890         }
1891     }
1892
1893     if (!latch_is_set(&f->dp->exit_latch)){
1894         goto reload;
1895     }
1896
1897     for (i = 0; i < poll_cnt; i++) {
1898          port_unref(poll_list[i].port);
1899     }
1900
1901     free(poll_list);
1902     free(f->name);
1903     return NULL;
1904 }
1905
1906 static void
1907 dp_netdev_set_pmd_threads(struct dp_netdev *dp, int n)
1908 {
1909     int i;
1910
1911     if (n == dp->n_pmd_threads) {
1912         return;
1913     }
1914
1915     /* Stop existing threads. */
1916     latch_set(&dp->exit_latch);
1917     dp_netdev_reload_pmd_threads(dp);
1918     for (i = 0; i < dp->n_pmd_threads; i++) {
1919         struct pmd_thread *f = &dp->pmd_threads[i];
1920
1921         xpthread_join(f->thread, NULL);
1922     }
1923     latch_poll(&dp->exit_latch);
1924     free(dp->pmd_threads);
1925
1926     /* Start new threads. */
1927     dp->pmd_threads = xmalloc(n * sizeof *dp->pmd_threads);
1928     dp->n_pmd_threads = n;
1929
1930     for (i = 0; i < n; i++) {
1931         struct pmd_thread *f = &dp->pmd_threads[i];
1932
1933         f->dp = dp;
1934         f->id = i;
1935         atomic_store(&f->change_seq, 1);
1936
1937         /* Each thread will distribute all devices rx-queues among
1938          * themselves. */
1939         xpthread_create(&f->thread, NULL, pmd_thread_main, f);
1940     }
1941 }
1942
1943 \f
1944 static void *
1945 dp_netdev_flow_stats_new_cb(void)
1946 {
1947     struct dp_netdev_flow_stats *bucket = xzalloc_cacheline(sizeof *bucket);
1948     ovs_mutex_init(&bucket->mutex);
1949     return bucket;
1950 }
1951
1952 static void
1953 dp_netdev_flow_used(struct dp_netdev_flow *netdev_flow,
1954                     const struct ofpbuf *packet,
1955                     const struct flow *key)
1956 {
1957     uint16_t tcp_flags = ntohs(key->tcp_flags);
1958     long long int now = time_msec();
1959     struct dp_netdev_flow_stats *bucket;
1960
1961     bucket = ovsthread_stats_bucket_get(&netdev_flow->stats,
1962                                         dp_netdev_flow_stats_new_cb);
1963
1964     ovs_mutex_lock(&bucket->mutex);
1965     bucket->used = MAX(now, bucket->used);
1966     bucket->packet_count++;
1967     bucket->byte_count += packet->size;
1968     bucket->tcp_flags |= tcp_flags;
1969     ovs_mutex_unlock(&bucket->mutex);
1970 }
1971
1972 static void *
1973 dp_netdev_stats_new_cb(void)
1974 {
1975     struct dp_netdev_stats *bucket = xzalloc_cacheline(sizeof *bucket);
1976     ovs_mutex_init(&bucket->mutex);
1977     return bucket;
1978 }
1979
1980 static void
1981 dp_netdev_count_packet(struct dp_netdev *dp, enum dp_stat_type type)
1982 {
1983     struct dp_netdev_stats *bucket;
1984
1985     bucket = ovsthread_stats_bucket_get(&dp->stats, dp_netdev_stats_new_cb);
1986     ovs_mutex_lock(&bucket->mutex);
1987     bucket->n[type]++;
1988     ovs_mutex_unlock(&bucket->mutex);
1989 }
1990
1991 static void
1992 dp_netdev_port_input(struct dp_netdev *dp, struct ofpbuf *packet,
1993                      struct pkt_metadata *md)
1994 {
1995     struct dp_netdev_flow *netdev_flow;
1996     struct flow key;
1997
1998     if (packet->size < ETH_HEADER_LEN) {
1999         ofpbuf_delete(packet);
2000         return;
2001     }
2002     flow_extract(packet, md, &key);
2003     netdev_flow = dp_netdev_lookup_flow(dp, &key);
2004     if (netdev_flow) {
2005         struct dp_netdev_actions *actions;
2006
2007         dp_netdev_flow_used(netdev_flow, packet, &key);
2008
2009         actions = dp_netdev_flow_get_actions(netdev_flow);
2010         dp_netdev_execute_actions(dp, &key, packet, true, md,
2011                                   actions->actions, actions->size);
2012         dp_netdev_count_packet(dp, DP_STAT_HIT);
2013     } else if (dp->handler_queues) {
2014         dp_netdev_count_packet(dp, DP_STAT_MISS);
2015         dp_netdev_output_userspace(dp, packet,
2016                                    flow_hash_5tuple(&key, 0) % dp->n_handlers,
2017                                    DPIF_UC_MISS, &key, NULL);
2018         ofpbuf_delete(packet);
2019     }
2020 }
2021
2022 static int
2023 dp_netdev_output_userspace(struct dp_netdev *dp, struct ofpbuf *packet,
2024                            int queue_no, int type, const struct flow *flow,
2025                            const struct nlattr *userdata)
2026 {
2027     struct dp_netdev_queue *q;
2028     int error;
2029
2030     fat_rwlock_rdlock(&dp->queue_rwlock);
2031     q = &dp->handler_queues[queue_no];
2032     ovs_mutex_lock(&q->mutex);
2033     if (q->head - q->tail < MAX_QUEUE_LEN) {
2034         struct dp_netdev_upcall *u = &q->upcalls[q->head++ & QUEUE_MASK];
2035         struct dpif_upcall *upcall = &u->upcall;
2036         struct ofpbuf *buf = &u->buf;
2037         size_t buf_size;
2038
2039         upcall->type = type;
2040
2041         /* Allocate buffer big enough for everything. */
2042         buf_size = ODPUTIL_FLOW_KEY_BYTES;
2043         if (userdata) {
2044             buf_size += NLA_ALIGN(userdata->nla_len);
2045         }
2046         buf_size += packet->size;
2047         ofpbuf_init(buf, buf_size);
2048
2049         /* Put ODP flow. */
2050         odp_flow_key_from_flow(buf, flow, flow->in_port.odp_port);
2051         upcall->key = buf->data;
2052         upcall->key_len = buf->size;
2053
2054         /* Put userdata. */
2055         if (userdata) {
2056             upcall->userdata = ofpbuf_put(buf, userdata,
2057                                           NLA_ALIGN(userdata->nla_len));
2058         }
2059
2060         upcall->packet.data = ofpbuf_put(buf, packet->data, packet->size);
2061         upcall->packet.size = packet->size;
2062
2063         seq_change(q->seq);
2064
2065         error = 0;
2066     } else {
2067         dp_netdev_count_packet(dp, DP_STAT_LOST);
2068         error = ENOBUFS;
2069     }
2070     ovs_mutex_unlock(&q->mutex);
2071     fat_rwlock_unlock(&dp->queue_rwlock);
2072
2073     return error;
2074 }
2075
2076 struct dp_netdev_execute_aux {
2077     struct dp_netdev *dp;
2078     const struct flow *key;
2079 };
2080
2081 static void
2082 dp_execute_cb(void *aux_, struct ofpbuf *packet,
2083               const struct pkt_metadata *md OVS_UNUSED,
2084               const struct nlattr *a, bool may_steal)
2085     OVS_NO_THREAD_SAFETY_ANALYSIS
2086 {
2087     struct dp_netdev_execute_aux *aux = aux_;
2088     int type = nl_attr_type(a);
2089     struct dp_netdev_port *p;
2090
2091     switch ((enum ovs_action_attr)type) {
2092     case OVS_ACTION_ATTR_OUTPUT:
2093         p = dp_netdev_lookup_port(aux->dp, u32_to_odp(nl_attr_get_u32(a)));
2094         if (p) {
2095             netdev_send(p->netdev, packet, may_steal);
2096         }
2097         break;
2098
2099     case OVS_ACTION_ATTR_USERSPACE: {
2100         const struct nlattr *userdata;
2101
2102         userdata = nl_attr_find_nested(a, OVS_USERSPACE_ATTR_USERDATA);
2103
2104         dp_netdev_output_userspace(aux->dp, packet,
2105                                    flow_hash_5tuple(aux->key, 0)
2106                                        % aux->dp->n_handlers,
2107                                    DPIF_UC_ACTION, aux->key,
2108                                    userdata);
2109
2110         if (may_steal) {
2111             ofpbuf_delete(packet);
2112         }
2113         break;
2114     }
2115     case OVS_ACTION_ATTR_PUSH_VLAN:
2116     case OVS_ACTION_ATTR_POP_VLAN:
2117     case OVS_ACTION_ATTR_PUSH_MPLS:
2118     case OVS_ACTION_ATTR_POP_MPLS:
2119     case OVS_ACTION_ATTR_SET:
2120     case OVS_ACTION_ATTR_SAMPLE:
2121     case OVS_ACTION_ATTR_UNSPEC:
2122     case __OVS_ACTION_ATTR_MAX:
2123         OVS_NOT_REACHED();
2124     }
2125
2126 }
2127
2128 static void
2129 dp_netdev_execute_actions(struct dp_netdev *dp, const struct flow *key,
2130                           struct ofpbuf *packet, bool may_steal,
2131                           struct pkt_metadata *md,
2132                           const struct nlattr *actions, size_t actions_len)
2133 {
2134     struct dp_netdev_execute_aux aux = {dp, key};
2135
2136     odp_execute_actions(&aux, packet, may_steal, md,
2137                         actions, actions_len, dp_execute_cb);
2138 }
2139
2140 const struct dpif_class dpif_netdev_class = {
2141     "netdev",
2142     dpif_netdev_enumerate,
2143     dpif_netdev_port_open_type,
2144     dpif_netdev_open,
2145     dpif_netdev_close,
2146     dpif_netdev_destroy,
2147     dpif_netdev_run,
2148     dpif_netdev_wait,
2149     dpif_netdev_get_stats,
2150     dpif_netdev_port_add,
2151     dpif_netdev_port_del,
2152     dpif_netdev_port_query_by_number,
2153     dpif_netdev_port_query_by_name,
2154     NULL,                       /* port_get_pid */
2155     dpif_netdev_port_dump_start,
2156     dpif_netdev_port_dump_next,
2157     dpif_netdev_port_dump_done,
2158     dpif_netdev_port_poll,
2159     dpif_netdev_port_poll_wait,
2160     dpif_netdev_flow_get,
2161     dpif_netdev_flow_put,
2162     dpif_netdev_flow_del,
2163     dpif_netdev_flow_flush,
2164     dpif_netdev_flow_dump_state_init,
2165     dpif_netdev_flow_dump_start,
2166     dpif_netdev_flow_dump_next,
2167     NULL,
2168     dpif_netdev_flow_dump_done,
2169     dpif_netdev_flow_dump_state_uninit,
2170     dpif_netdev_execute,
2171     NULL,                       /* operate */
2172     dpif_netdev_recv_set,
2173     dpif_netdev_handlers_set,
2174     dpif_netdev_queue_to_priority,
2175     dpif_netdev_recv,
2176     dpif_netdev_recv_wait,
2177     dpif_netdev_recv_purge,
2178 };
2179
2180 static void
2181 dpif_dummy_change_port_number(struct unixctl_conn *conn, int argc OVS_UNUSED,
2182                               const char *argv[], void *aux OVS_UNUSED)
2183 {
2184     struct dp_netdev_port *port;
2185     struct dp_netdev *dp;
2186     odp_port_t port_no;
2187
2188     ovs_mutex_lock(&dp_netdev_mutex);
2189     dp = shash_find_data(&dp_netdevs, argv[1]);
2190     if (!dp || !dpif_netdev_class_is_dummy(dp->class)) {
2191         ovs_mutex_unlock(&dp_netdev_mutex);
2192         unixctl_command_reply_error(conn, "unknown datapath or not a dummy");
2193         return;
2194     }
2195     ovs_refcount_ref(&dp->ref_cnt);
2196     ovs_mutex_unlock(&dp_netdev_mutex);
2197
2198     ovs_rwlock_wrlock(&dp->port_rwlock);
2199     if (get_port_by_name(dp, argv[2], &port)) {
2200         unixctl_command_reply_error(conn, "unknown port");
2201         goto exit;
2202     }
2203
2204     port_no = u32_to_odp(atoi(argv[3]));
2205     if (!port_no || port_no == ODPP_NONE) {
2206         unixctl_command_reply_error(conn, "bad port number");
2207         goto exit;
2208     }
2209     if (dp_netdev_lookup_port(dp, port_no)) {
2210         unixctl_command_reply_error(conn, "port number already in use");
2211         goto exit;
2212     }
2213     hmap_remove(&dp->ports, &port->node);
2214     port->port_no = port_no;
2215     hmap_insert(&dp->ports, &port->node, hash_int(odp_to_u32(port_no), 0));
2216     seq_change(dp->port_seq);
2217     unixctl_command_reply(conn, NULL);
2218
2219 exit:
2220     ovs_rwlock_unlock(&dp->port_rwlock);
2221     dp_netdev_unref(dp);
2222 }
2223
2224 static void
2225 dpif_dummy_register__(const char *type)
2226 {
2227     struct dpif_class *class;
2228
2229     class = xmalloc(sizeof *class);
2230     *class = dpif_netdev_class;
2231     class->type = xstrdup(type);
2232     dp_register_provider(class);
2233 }
2234
2235 void
2236 dpif_dummy_register(bool override)
2237 {
2238     if (override) {
2239         struct sset types;
2240         const char *type;
2241
2242         sset_init(&types);
2243         dp_enumerate_types(&types);
2244         SSET_FOR_EACH (type, &types) {
2245             if (!dp_unregister_provider(type)) {
2246                 dpif_dummy_register__(type);
2247             }
2248         }
2249         sset_destroy(&types);
2250     }
2251
2252     dpif_dummy_register__("dummy");
2253
2254     unixctl_command_register("dpif-dummy/change-port-number",
2255                              "DP PORT NEW-NUMBER",
2256                              3, 3, dpif_dummy_change_port_number, NULL);
2257 }