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