ofproto/ofproto-dpif-upcall.c

   1 /* Copyright (c) 2009, 2010, 2011, 2012, 2013, 2014 Nicira, Inc.
   2  *
   3  * Licensed under the Apache License, Version 2.0 (the "License");
   4  * you may not use this file except in compliance with the License.
   5  * You may obtain a copy of the License at:
   6  *
   7  *     http://www.apache.org/licenses/LICENSE-2.0
   8  *
   9  * Unless required by applicable law or agreed to in writing, software
  10  * distributed under the License is distributed on an "AS IS" BASIS,
  11  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12  * See the License for the specific language governing permissions and
  13  * limitations under the License.  */
  14
  15 #include <config.h>
  16 #include "ofproto-dpif-upcall.h"
  17
  18 #include <errno.h>
  19 #include <stdbool.h>
  20 #include <inttypes.h>
  21
  22 #include "connmgr.h"
  23 #include "coverage.h"
  24 #include "cmap.h"
  25 #include "dpif.h"
  26 #include "dynamic-string.h"
  27 #include "fail-open.h"
  28 #include "guarded-list.h"
  29 #include "latch.h"
  30 #include "list.h"
  31 #include "netlink.h"
  32 #include "ofpbuf.h"
  33 #include "ofproto-dpif-ipfix.h"
  34 #include "ofproto-dpif-sflow.h"
  35 #include "ofproto-dpif-xlate.h"
  36 #include "ovs-rcu.h"
  37 #include "packets.h"
  38 #include "poll-loop.h"
  39 #include "seq.h"
  40 #include "unixctl.h"
  41 #include "openvswitch/vlog.h"
  42
  43 #define MAX_QUEUE_LENGTH 512
  44 #define UPCALL_MAX_BATCH 64
  45 #define REVALIDATE_MAX_BATCH 50
  46
  47 VLOG_DEFINE_THIS_MODULE(ofproto_dpif_upcall);
  48
  49 COVERAGE_DEFINE(dumped_duplicate_flow);
  50 COVERAGE_DEFINE(dumped_new_flow);
  51 COVERAGE_DEFINE(handler_duplicate_upcall);
  52 COVERAGE_DEFINE(upcall_ukey_contention);
  53 COVERAGE_DEFINE(revalidate_missed_dp_flow);
  54
  55 /* A thread that reads upcalls from dpif, forwards each upcall's packet,
  56  * and possibly sets up a kernel flow as a cache. */
  57 struct handler {
  58     struct udpif *udpif;               /* Parent udpif. */
  59     pthread_t thread;                  /* Thread ID. */
  60     uint32_t handler_id;               /* Handler id. */
  61 };
  62
  63 /* In the absence of a multiple-writer multiple-reader datastructure for
  64  * storing ukeys, we use a large number of cmaps, each with its own lock for
  65  * writing. */
  66 #define N_UMAPS 512 /* per udpif. */
  67 struct umap {
  68     struct ovs_mutex mutex;            /* Take for writing to the following. */
  69     struct cmap cmap;                  /* Datapath flow keys. */
  70 };
  71
  72 /* A thread that processes datapath flows, updates OpenFlow statistics, and
  73  * updates or removes them if necessary. */
  74 struct revalidator {
  75     struct udpif *udpif;               /* Parent udpif. */
  76     pthread_t thread;                  /* Thread ID. */
  77     unsigned int id;                   /* ovsthread_id_self(). */
  78 };
  79
  80 /* An upcall handler for ofproto_dpif.
  81  *
  82  * udpif keeps records of two kind of logically separate units:
  83  *
  84  * upcall handling
  85  * ---------------
  86  *
  87  *    - An array of 'struct handler's for upcall handling and flow
  88  *      installation.
  89  *
  90  * flow revalidation
  91  * -----------------
  92  *
  93  *    - Revalidation threads which read the datapath flow table and maintains
  94  *      them.
  95  */
  96 struct udpif {
  97     struct ovs_list list_node;         /* In all_udpifs list. */
  98
  99     struct dpif *dpif;                 /* Datapath handle. */
 100     struct dpif_backer *backer;        /* Opaque dpif_backer pointer. */
 101
 102     struct handler *handlers;          /* Upcall handlers. */
 103     size_t n_handlers;
 104
 105     struct revalidator *revalidators;  /* Flow revalidators. */
 106     size_t n_revalidators;
 107
 108     struct latch exit_latch;           /* Tells child threads to exit. */
 109
 110     /* Revalidation. */
 111     struct seq *reval_seq;             /* Incremented to force revalidation. */
 112     bool reval_exit;                   /* Set by leader on 'exit_latch. */
 113     struct ovs_barrier reval_barrier;  /* Barrier used by revalidators. */
 114     struct dpif_flow_dump *dump;       /* DPIF flow dump state. */
 115     long long int dump_duration;       /* Duration of the last flow dump. */
 116     struct seq *dump_seq;              /* Increments each dump iteration. */
 117     atomic_bool enable_ufid;           /* If true, skip dumping flow attrs. */
 118
 119     /* There are 'N_UMAPS' maps containing 'struct udpif_key' elements.
 120      *
 121      * During the flow dump phase, revalidators insert into these with a random
 122      * distribution. During the garbage collection phase, each revalidator
 123      * takes care of garbage collecting a slice of these maps. */
 124     struct umap *ukeys;
 125
 126     /* Datapath flow statistics. */
 127     unsigned int max_n_flows;
 128     unsigned int avg_n_flows;
 129
 130     /* Following fields are accessed and modified by different threads. */
 131     atomic_uint flow_limit;            /* Datapath flow hard limit. */
 132
 133     /* n_flows_mutex prevents multiple threads updating these concurrently. */
 134     atomic_uint n_flows;               /* Number of flows in the datapath. */
 135     atomic_llong n_flows_timestamp;    /* Last time n_flows was updated. */
 136     struct ovs_mutex n_flows_mutex;
 137
 138     /* Following fields are accessed and modified only from the main thread. */
 139     struct unixctl_conn **conns;       /* Connections waiting on dump_seq. */
 140     uint64_t conn_seq;                 /* Corresponds to 'dump_seq' when
 141                                           conns[n_conns-1] was stored. */
 142     size_t n_conns;                    /* Number of connections waiting. */
 143 };
 144
 145 enum upcall_type {
 146     BAD_UPCALL,                 /* Some kind of bug somewhere. */
 147     MISS_UPCALL,                /* A flow miss.  */
 148     SFLOW_UPCALL,               /* sFlow sample. */
 149     FLOW_SAMPLE_UPCALL,         /* Per-flow sampling. */
 150     IPFIX_UPCALL                /* Per-bridge sampling. */
 151 };
 152
 153 struct upcall {
 154     struct ofproto_dpif *ofproto;  /* Parent ofproto. */
 155
 156     /* The flow and packet are only required to be constant when using
 157      * dpif-netdev.  If a modification is absolutely necessary, a const cast
 158      * may be used with other datapaths. */
 159     const struct flow *flow;       /* Parsed representation of the packet. */
 160     const ovs_u128 *ufid;          /* Unique identifier for 'flow'. */
 161     int pmd_id;                    /* Datapath poll mode driver id. */
 162     const struct ofpbuf *packet;   /* Packet associated with this upcall. */
 163     ofp_port_t in_port;            /* OpenFlow in port, or OFPP_NONE. */
 164
 165     enum dpif_upcall_type type;    /* Datapath type of the upcall. */
 166     const struct nlattr *userdata; /* Userdata for DPIF_UC_ACTION Upcalls. */
 167
 168     bool xout_initialized;         /* True if 'xout' must be uninitialized. */
 169     struct xlate_out xout;         /* Result of xlate_actions(). */
 170     struct ofpbuf put_actions;     /* Actions 'put' in the fastapath. */
 171
 172     struct dpif_ipfix *ipfix;      /* IPFIX pointer or NULL. */
 173     struct dpif_sflow *sflow;      /* SFlow pointer or NULL. */
 174
 175     bool vsp_adjusted;             /* 'packet' and 'flow' were adjusted for
 176                                       VLAN splinters if true. */
 177
 178     struct udpif_key *ukey;        /* Revalidator flow cache. */
 179     bool ukey_persists;            /* Set true to keep 'ukey' beyond the
 180                                       lifetime of this upcall. */
 181
 182     uint64_t dump_seq;             /* udpif->dump_seq at translation time. */
 183     uint64_t reval_seq;            /* udpif->reval_seq at translation time. */
 184
 185     /* Not used by the upcall callback interface. */
 186     const struct nlattr *key;      /* Datapath flow key. */
 187     size_t key_len;                /* Datapath flow key length. */
 188     const struct nlattr *out_tun_key;  /* Datapath output tunnel key. */
 189 };
 190
 191 /* 'udpif_key's are responsible for tracking the little bit of state udpif
 192  * needs to do flow expiration which can't be pulled directly from the
 193  * datapath.  They may be created by any handler or revalidator thread at any
 194  * time, and read by any revalidator during the dump phase. They are however
 195  * each owned by a single revalidator which takes care of destroying them
 196  * during the garbage-collection phase.
 197  *
 198  * The mutex within the ukey protects some members of the ukey. The ukey
 199  * itself is protected by RCU and is held within a umap in the parent udpif.
 200  * Adding or removing a ukey from a umap is only safe when holding the
 201  * corresponding umap lock. */
 202 struct udpif_key {
 203     struct cmap_node cmap_node;     /* In parent revalidator 'ukeys' map. */
 204
 205     /* These elements are read only once created, and therefore aren't
 206      * protected by a mutex. */
 207     const struct nlattr *key;      /* Datapath flow key. */
 208     size_t key_len;                /* Length of 'key'. */
 209     const struct nlattr *mask;     /* Datapath flow mask. */
 210     size_t mask_len;               /* Length of 'mask'. */
 211     struct ofpbuf *actions;        /* Datapath flow actions as nlattrs. */
 212     ovs_u128 ufid;                 /* Unique flow identifier. */
 213     bool ufid_present;             /* True if 'ufid' is in datapath. */
 214     uint32_t hash;                 /* Pre-computed hash for 'key'. */
 215     int pmd_id;                    /* Datapath poll mode driver id. */
 216
 217     struct ovs_mutex mutex;                   /* Guards the following. */
 218     struct dpif_flow_stats stats OVS_GUARDED; /* Last known stats.*/
 219     long long int created OVS_GUARDED;        /* Estimate of creation time. */
 220     uint64_t dump_seq OVS_GUARDED;            /* Tracks udpif->dump_seq. */
 221     uint64_t reval_seq OVS_GUARDED;           /* Tracks udpif->reval_seq. */
 222     bool flow_exists OVS_GUARDED;             /* Ensures flows are only deleted
 223                                                  once. */
 224
 225     struct xlate_cache *xcache OVS_GUARDED;   /* Cache for xlate entries that
 226                                                * are affected by this ukey.
 227                                                * Used for stats and learning.*/
 228     union {
 229         struct odputil_keybuf buf;
 230         struct nlattr nla;
 231     } keybuf, maskbuf;
 232 };
 233
 234 /* Datapath operation with optional ukey attached. */
 235 struct ukey_op {
 236     struct udpif_key *ukey;
 237     struct dpif_flow_stats stats; /* Stats for 'op'. */
 238     struct dpif_op dop;           /* Flow operation. */
 239 };
 240
 241 static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(1, 5);
 242 static struct ovs_list all_udpifs = OVS_LIST_INITIALIZER(&all_udpifs);
 243
 244 static size_t recv_upcalls(struct handler *);
 245 static int process_upcall(struct udpif *, struct upcall *,
 246                           struct ofpbuf *odp_actions);
 247 static void handle_upcalls(struct udpif *, struct upcall *, size_t n_upcalls);
 248 static void udpif_stop_threads(struct udpif *);
 249 static void udpif_start_threads(struct udpif *, size_t n_handlers,
 250                                 size_t n_revalidators);
 251 static void *udpif_upcall_handler(void *);
 252 static void *udpif_revalidator(void *);
 253 static unsigned long udpif_get_n_flows(struct udpif *);
 254 static void revalidate(struct revalidator *);
 255 static void revalidator_sweep(struct revalidator *);
 256 static void revalidator_purge(struct revalidator *);
 257 static void upcall_unixctl_show(struct unixctl_conn *conn, int argc,
 258                                 const char *argv[], void *aux);
 259 static void upcall_unixctl_disable_megaflows(struct unixctl_conn *, int argc,
 260                                              const char *argv[], void *aux);
 261 static void upcall_unixctl_enable_megaflows(struct unixctl_conn *, int argc,
 262                                             const char *argv[], void *aux);
 263 static void upcall_unixctl_disable_ufid(struct unixctl_conn *, int argc,
 264                                               const char *argv[], void *aux);
 265 static void upcall_unixctl_enable_ufid(struct unixctl_conn *, int argc,
 266                                              const char *argv[], void *aux);
 267 static void upcall_unixctl_set_flow_limit(struct unixctl_conn *conn, int argc,
 268                                             const char *argv[], void *aux);
 269 static void upcall_unixctl_dump_wait(struct unixctl_conn *conn, int argc,
 270                                      const char *argv[], void *aux);
 271 static void upcall_unixctl_purge(struct unixctl_conn *conn, int argc,
 272                                  const char *argv[], void *aux);
 273
 274 static struct udpif_key *ukey_create_from_upcall(const struct upcall *);
 275 static int ukey_create_from_dpif_flow(const struct udpif *,
 276                                       const struct dpif_flow *,
 277                                       struct udpif_key **);
 278 static bool ukey_install_start(struct udpif *, struct udpif_key *ukey);
 279 static bool ukey_install_finish(struct udpif_key *ukey, int error);
 280 static bool ukey_install(struct udpif *udpif, struct udpif_key *ukey);
 281 static struct udpif_key *ukey_lookup(struct udpif *udpif,
 282                                      const ovs_u128 *ufid);
 283 static int ukey_acquire(struct udpif *, const struct dpif_flow *,
 284                         struct udpif_key **result, int *error);
 285 static void ukey_delete__(struct udpif_key *);
 286 static void ukey_delete(struct umap *, struct udpif_key *);
 287 static enum upcall_type classify_upcall(enum dpif_upcall_type type,
 288                                         const struct nlattr *userdata);
 289
 290 static int upcall_receive(struct upcall *, const struct dpif_backer *,
 291                           const struct ofpbuf *packet, enum dpif_upcall_type,
 292                           const struct nlattr *userdata, const struct flow *,
 293                           const ovs_u128 *ufid, const int pmd_id);
 294 static void upcall_uninit(struct upcall *);
 295
 296 static upcall_callback upcall_cb;
 297
 298 static atomic_bool enable_megaflows = ATOMIC_VAR_INIT(true);
 299
 300 struct udpif *
 301 udpif_create(struct dpif_backer *backer, struct dpif *dpif)
 302 {
 303     static struct ovsthread_once once = OVSTHREAD_ONCE_INITIALIZER;
 304     struct udpif *udpif = xzalloc(sizeof *udpif);
 305
 306     if (ovsthread_once_start(&once)) {
 307         unixctl_command_register("upcall/show", "", 0, 0, upcall_unixctl_show,
 308                                  NULL);
 309         unixctl_command_register("upcall/disable-megaflows", "", 0, 0,
 310                                  upcall_unixctl_disable_megaflows, NULL);
 311         unixctl_command_register("upcall/enable-megaflows", "", 0, 0,
 312                                  upcall_unixctl_enable_megaflows, NULL);
 313         unixctl_command_register("upcall/disable-ufid", "", 0, 0,
 314                                  upcall_unixctl_disable_ufid, NULL);
 315         unixctl_command_register("upcall/enable-ufid", "", 0, 0,
 316                                  upcall_unixctl_enable_ufid, NULL);
 317         unixctl_command_register("upcall/set-flow-limit", "", 1, 1,
 318                                  upcall_unixctl_set_flow_limit, NULL);
 319         unixctl_command_register("revalidator/wait", "", 0, 0,
 320                                  upcall_unixctl_dump_wait, NULL);
 321         unixctl_command_register("revalidator/purge", "", 0, 0,
 322                                  upcall_unixctl_purge, NULL);
 323         ovsthread_once_done(&once);
 324     }
 325
 326     udpif->dpif = dpif;
 327     udpif->backer = backer;
 328     atomic_init(&udpif->flow_limit, MIN(ofproto_flow_limit, 10000));
 329     udpif->reval_seq = seq_create();
 330     udpif->dump_seq = seq_create();
 331     latch_init(&udpif->exit_latch);
 332     list_push_back(&all_udpifs, &udpif->list_node);
 333     atomic_init(&udpif->enable_ufid, false);
 334     atomic_init(&udpif->n_flows, 0);
 335     atomic_init(&udpif->n_flows_timestamp, LLONG_MIN);
 336     ovs_mutex_init(&udpif->n_flows_mutex);
 337     udpif->ukeys = xmalloc(N_UMAPS * sizeof *udpif->ukeys);
 338     for (int i = 0; i < N_UMAPS; i++) {
 339         cmap_init(&udpif->ukeys[i].cmap);
 340         ovs_mutex_init(&udpif->ukeys[i].mutex);
 341     }
 342
 343     dpif_register_upcall_cb(dpif, upcall_cb, udpif);
 344
 345     return udpif;
 346 }
 347
 348 void
 349 udpif_run(struct udpif *udpif)
 350 {
 351     if (udpif->conns && udpif->conn_seq != seq_read(udpif->dump_seq)) {
 352         int i;
 353
 354         for (i = 0; i < udpif->n_conns; i++) {
 355             unixctl_command_reply(udpif->conns[i], NULL);
 356         }
 357         free(udpif->conns);
 358         udpif->conns = NULL;
 359         udpif->n_conns = 0;
 360     }
 361 }
 362
 363 void
 364 udpif_destroy(struct udpif *udpif)
 365 {
 366     udpif_stop_threads(udpif);
 367
 368     for (int i = 0; i < N_UMAPS; i++) {
 369         cmap_destroy(&udpif->ukeys[i].cmap);
 370         ovs_mutex_destroy(&udpif->ukeys[i].mutex);
 371     }
 372     free(udpif->ukeys);
 373     udpif->ukeys = NULL;
 374
 375     list_remove(&udpif->list_node);
 376     latch_destroy(&udpif->exit_latch);
 377     seq_destroy(udpif->reval_seq);
 378     seq_destroy(udpif->dump_seq);
 379     ovs_mutex_destroy(&udpif->n_flows_mutex);
 380     free(udpif);
 381 }
 382
 383 /* Stops the handler and revalidator threads, must be enclosed in
 384  * ovsrcu quiescent state unless when destroying udpif. */
 385 static void
 386 udpif_stop_threads(struct udpif *udpif)
 387 {
 388     if (udpif && (udpif->n_handlers != 0 || udpif->n_revalidators != 0)) {
 389         size_t i;
 390
 391         latch_set(&udpif->exit_latch);
 392
 393         for (i = 0; i < udpif->n_handlers; i++) {
 394             struct handler *handler = &udpif->handlers[i];
 395
 396             xpthread_join(handler->thread, NULL);
 397         }
 398
 399         for (i = 0; i < udpif->n_revalidators; i++) {
 400             xpthread_join(udpif->revalidators[i].thread, NULL);
 401         }
 402
 403         dpif_disable_upcall(udpif->dpif);
 404
 405         for (i = 0; i < udpif->n_revalidators; i++) {
 406             struct revalidator *revalidator = &udpif->revalidators[i];
 407
 408             /* Delete ukeys, and delete all flows from the datapath to prevent
 409              * double-counting stats. */
 410             revalidator_purge(revalidator);
 411         }
 412
 413         latch_poll(&udpif->exit_latch);
 414
 415         ovs_barrier_destroy(&udpif->reval_barrier);
 416
 417         free(udpif->revalidators);
 418         udpif->revalidators = NULL;
 419         udpif->n_revalidators = 0;
 420
 421         free(udpif->handlers);
 422         udpif->handlers = NULL;
 423         udpif->n_handlers = 0;
 424     }
 425 }
 426
 427 /* Starts the handler and revalidator threads, must be enclosed in
 428  * ovsrcu quiescent state. */
 429 static void
 430 udpif_start_threads(struct udpif *udpif, size_t n_handlers,
 431                     size_t n_revalidators)
 432 {
 433     if (udpif && n_handlers && n_revalidators) {
 434         size_t i;
 435         bool enable_ufid;
 436
 437         udpif->n_handlers = n_handlers;
 438         udpif->n_revalidators = n_revalidators;
 439
 440         udpif->handlers = xzalloc(udpif->n_handlers * sizeof *udpif->handlers);
 441         for (i = 0; i < udpif->n_handlers; i++) {
 442             struct handler *handler = &udpif->handlers[i];
 443
 444             handler->udpif = udpif;
 445             handler->handler_id = i;
 446             handler->thread = ovs_thread_create(
 447                 "handler", udpif_upcall_handler, handler);
 448         }
 449
 450         enable_ufid = ofproto_dpif_get_enable_ufid(udpif->backer);
 451         atomic_init(&udpif->enable_ufid, enable_ufid);
 452         dpif_enable_upcall(udpif->dpif);
 453
 454         ovs_barrier_init(&udpif->reval_barrier, udpif->n_revalidators);
 455         udpif->reval_exit = false;
 456         udpif->revalidators = xzalloc(udpif->n_revalidators
 457                                       * sizeof *udpif->revalidators);
 458         for (i = 0; i < udpif->n_revalidators; i++) {
 459             struct revalidator *revalidator = &udpif->revalidators[i];
 460
 461             revalidator->udpif = udpif;
 462             revalidator->thread = ovs_thread_create(
 463                 "revalidator", udpif_revalidator, revalidator);
 464         }
 465     }
 466 }
 467
 468 /* Tells 'udpif' how many threads it should use to handle upcalls.
 469  * 'n_handlers' and 'n_revalidators' can never be zero.  'udpif''s
 470  * datapath handle must have packet reception enabled before starting
 471  * threads. */
 472 void
 473 udpif_set_threads(struct udpif *udpif, size_t n_handlers,
 474                   size_t n_revalidators)
 475 {
 476     ovs_assert(udpif);
 477     ovs_assert(n_handlers && n_revalidators);
 478
 479     ovsrcu_quiesce_start();
 480     if (udpif->n_handlers != n_handlers
 481         || udpif->n_revalidators != n_revalidators) {
 482         udpif_stop_threads(udpif);
 483     }
 484
 485     if (!udpif->handlers && !udpif->revalidators) {
 486         int error;
 487
 488         error = dpif_handlers_set(udpif->dpif, n_handlers);
 489         if (error) {
 490             VLOG_ERR("failed to configure handlers in dpif %s: %s",
 491                      dpif_name(udpif->dpif), ovs_strerror(error));
 492             return;
 493         }
 494
 495         udpif_start_threads(udpif, n_handlers, n_revalidators);
 496     }
 497     ovsrcu_quiesce_end();
 498 }
 499
 500 /* Waits for all ongoing upcall translations to complete.  This ensures that
 501  * there are no transient references to any removed ofprotos (or other
 502  * objects).  In particular, this should be called after an ofproto is removed
 503  * (e.g. via xlate_remove_ofproto()) but before it is destroyed. */
 504 void
 505 udpif_synchronize(struct udpif *udpif)
 506 {
 507     /* This is stronger than necessary.  It would be sufficient to ensure
 508      * (somehow) that each handler and revalidator thread had passed through
 509      * its main loop once. */
 510     size_t n_handlers = udpif->n_handlers;
 511     size_t n_revalidators = udpif->n_revalidators;
 512
 513     ovsrcu_quiesce_start();
 514     udpif_stop_threads(udpif);
 515     udpif_start_threads(udpif, n_handlers, n_revalidators);
 516     ovsrcu_quiesce_end();
 517 }
 518
 519 /* Notifies 'udpif' that something changed which may render previous
 520  * xlate_actions() results invalid. */
 521 void
 522 udpif_revalidate(struct udpif *udpif)
 523 {
 524     seq_change(udpif->reval_seq);
 525 }
 526
 527 /* Returns a seq which increments every time 'udpif' pulls stats from the
 528  * datapath.  Callers can use this to get a sense of when might be a good time
 529  * to do periodic work which relies on relatively up to date statistics. */
 530 struct seq *
 531 udpif_dump_seq(struct udpif *udpif)
 532 {
 533     return udpif->dump_seq;
 534 }
 535
 536 void
 537 udpif_get_memory_usage(struct udpif *udpif, struct simap *usage)
 538 {
 539     size_t i;
 540
 541     simap_increase(usage, "handlers", udpif->n_handlers);
 542
 543     simap_increase(usage, "revalidators", udpif->n_revalidators);
 544     for (i = 0; i < N_UMAPS; i++) {
 545         simap_increase(usage, "udpif keys", cmap_count(&udpif->ukeys[i].cmap));
 546     }
 547 }
 548
 549 /* Remove flows from a single datapath. */
 550 void
 551 udpif_flush(struct udpif *udpif)
 552 {
 553     size_t n_handlers, n_revalidators;
 554
 555     n_handlers = udpif->n_handlers;
 556     n_revalidators = udpif->n_revalidators;
 557
 558     ovsrcu_quiesce_start();
 559
 560     udpif_stop_threads(udpif);
 561     dpif_flow_flush(udpif->dpif);
 562     udpif_start_threads(udpif, n_handlers, n_revalidators);
 563
 564     ovsrcu_quiesce_end();
 565 }
 566
 567 /* Removes all flows from all datapaths. */
 568 static void
 569 udpif_flush_all_datapaths(void)
 570 {
 571     struct udpif *udpif;
 572
 573     LIST_FOR_EACH (udpif, list_node, &all_udpifs) {
 574         udpif_flush(udpif);
 575     }
 576 }
 577
 578 \f
 579 static unsigned long
 580 udpif_get_n_flows(struct udpif *udpif)
 581 {
 582     long long int time, now;
 583     unsigned long flow_count;
 584
 585     now = time_msec();
 586     atomic_read_relaxed(&udpif->n_flows_timestamp, &time);
 587     if (time < now - 100 && !ovs_mutex_trylock(&udpif->n_flows_mutex)) {
 588         struct dpif_dp_stats stats;
 589
 590         atomic_store_relaxed(&udpif->n_flows_timestamp, now);
 591         dpif_get_dp_stats(udpif->dpif, &stats);
 592         flow_count = stats.n_flows;
 593         atomic_store_relaxed(&udpif->n_flows, flow_count);
 594         ovs_mutex_unlock(&udpif->n_flows_mutex);
 595     } else {
 596         atomic_read_relaxed(&udpif->n_flows, &flow_count);
 597     }
 598     return flow_count;
 599 }
 600
 601 /* The upcall handler thread tries to read a batch of UPCALL_MAX_BATCH
 602  * upcalls from dpif, processes the batch and installs corresponding flows
 603  * in dpif. */
 604 static void *
 605 udpif_upcall_handler(void *arg)
 606 {
 607     struct handler *handler = arg;
 608     struct udpif *udpif = handler->udpif;
 609
 610     while (!latch_is_set(&handler->udpif->exit_latch)) {
 611         if (recv_upcalls(handler)) {
 612             poll_immediate_wake();
 613         } else {
 614             dpif_recv_wait(udpif->dpif, handler->handler_id);
 615             latch_wait(&udpif->exit_latch);
 616         }
 617         poll_block();
 618     }
 619
 620     return NULL;
 621 }
 622
 623 static size_t
 624 recv_upcalls(struct handler *handler)
 625 {
 626     struct udpif *udpif = handler->udpif;
 627     uint64_t recv_stubs[UPCALL_MAX_BATCH][512 / 8];
 628     struct ofpbuf recv_bufs[UPCALL_MAX_BATCH];
 629     struct dpif_upcall dupcalls[UPCALL_MAX_BATCH];
 630     struct upcall upcalls[UPCALL_MAX_BATCH];
 631     struct flow flows[UPCALL_MAX_BATCH];
 632     size_t n_upcalls, i;
 633
 634     n_upcalls = 0;
 635     while (n_upcalls < UPCALL_MAX_BATCH) {
 636         struct ofpbuf *recv_buf = &recv_bufs[n_upcalls];
 637         struct dpif_upcall *dupcall = &dupcalls[n_upcalls];
 638         struct upcall *upcall = &upcalls[n_upcalls];
 639         struct flow *flow = &flows[n_upcalls];
 640         struct pkt_metadata md;
 641         int error;
 642
 643         ofpbuf_use_stub(recv_buf, recv_stubs[n_upcalls],
 644                         sizeof recv_stubs[n_upcalls]);
 645         if (dpif_recv(udpif->dpif, handler->handler_id, dupcall, recv_buf)) {
 646             ofpbuf_uninit(recv_buf);
 647             break;
 648         }
 649
 650         if (odp_flow_key_to_flow(dupcall->key, dupcall->key_len, flow)
 651             == ODP_FIT_ERROR) {
 652             goto free_dupcall;
 653         }
 654
 655         error = upcall_receive(upcall, udpif->backer, &dupcall->packet,
 656                                dupcall->type, dupcall->userdata, flow,
 657                                &dupcall->ufid, PMD_ID_NULL);
 658         if (error) {
 659             if (error == ENODEV) {
 660                 /* Received packet on datapath port for which we couldn't
 661                  * associate an ofproto.  This can happen if a port is removed
 662                  * while traffic is being received.  Print a rate-limited
 663                  * message in case it happens frequently. */
 664                 dpif_flow_put(udpif->dpif, DPIF_FP_CREATE, dupcall->key,
 665                               dupcall->key_len, NULL, 0, NULL, 0,
 666                               &dupcall->ufid, PMD_ID_NULL, NULL);
 667                 VLOG_INFO_RL(&rl, "received packet on unassociated datapath "
 668                              "port %"PRIu32, flow->in_port.odp_port);
 669             }
 670             goto free_dupcall;
 671         }
 672
 673         upcall->key = dupcall->key;
 674         upcall->key_len = dupcall->key_len;
 675         upcall->ufid = &dupcall->ufid;
 676
 677         upcall->out_tun_key = dupcall->out_tun_key;
 678
 679         if (vsp_adjust_flow(upcall->ofproto, flow, &dupcall->packet)) {
 680             upcall->vsp_adjusted = true;
 681         }
 682
 683         md = pkt_metadata_from_flow(flow);
 684         flow_extract(&dupcall->packet, &md, flow);
 685
 686         error = process_upcall(udpif, upcall, NULL);
 687         if (error) {
 688             goto cleanup;
 689         }
 690
 691         n_upcalls++;
 692         continue;
 693
 694 cleanup:
 695         upcall_uninit(upcall);
 696 free_dupcall:
 697         ofpbuf_uninit(&dupcall->packet);
 698         ofpbuf_uninit(recv_buf);
 699     }
 700
 701     if (n_upcalls) {
 702         handle_upcalls(handler->udpif, upcalls, n_upcalls);
 703         for (i = 0; i < n_upcalls; i++) {
 704             ofpbuf_uninit(&dupcalls[i].packet);
 705             ofpbuf_uninit(&recv_bufs[i]);
 706             upcall_uninit(&upcalls[i]);
 707         }
 708     }
 709
 710     return n_upcalls;
 711 }
 712
 713 static void *
 714 udpif_revalidator(void *arg)
 715 {
 716     /* Used by all revalidators. */
 717     struct revalidator *revalidator = arg;
 718     struct udpif *udpif = revalidator->udpif;
 719     bool leader = revalidator == &udpif->revalidators[0];
 720
 721     /* Used only by the leader. */
 722     long long int start_time = 0;
 723     uint64_t last_reval_seq = 0;
 724     size_t n_flows = 0;
 725
 726     revalidator->id = ovsthread_id_self();
 727     for (;;) {
 728         if (leader) {
 729             uint64_t reval_seq;
 730
 731             reval_seq = seq_read(udpif->reval_seq);
 732             last_reval_seq = reval_seq;
 733
 734             n_flows = udpif_get_n_flows(udpif);
 735             udpif->max_n_flows = MAX(n_flows, udpif->max_n_flows);
 736             udpif->avg_n_flows = (udpif->avg_n_flows + n_flows) / 2;
 737
 738             /* Only the leader checks the exit latch to prevent a race where
 739              * some threads think it's true and exit and others think it's
 740              * false and block indefinitely on the reval_barrier */
 741             udpif->reval_exit = latch_is_set(&udpif->exit_latch);
 742
 743             start_time = time_msec();
 744             if (!udpif->reval_exit) {
 745                 bool terse_dump;
 746
 747                 atomic_read_relaxed(&udpif->enable_ufid, &terse_dump);
 748                 udpif->dump = dpif_flow_dump_create(udpif->dpif, terse_dump);
 749             }
 750         }
 751
 752         /* Wait for the leader to start the flow dump. */
 753         ovs_barrier_block(&udpif->reval_barrier);
 754         if (udpif->reval_exit) {
 755             break;
 756         }
 757         revalidate(revalidator);
 758
 759         /* Wait for all flows to have been dumped before we garbage collect. */
 760         ovs_barrier_block(&udpif->reval_barrier);
 761         revalidator_sweep(revalidator);
 762
 763         /* Wait for all revalidators to finish garbage collection. */
 764         ovs_barrier_block(&udpif->reval_barrier);
 765
 766         if (leader) {
 767             unsigned int flow_limit;
 768             long long int duration;
 769
 770             atomic_read_relaxed(&udpif->flow_limit, &flow_limit);
 771
 772             dpif_flow_dump_destroy(udpif->dump);
 773             seq_change(udpif->dump_seq);
 774
 775             duration = MAX(time_msec() - start_time, 1);
 776             udpif->dump_duration = duration;
 777             if (duration > 2000) {
 778                 flow_limit /= duration / 1000;
 779             } else if (duration > 1300) {
 780                 flow_limit = flow_limit * 3 / 4;
 781             } else if (duration < 1000 && n_flows > 2000
 782                        && flow_limit < n_flows * 1000 / duration) {
 783                 flow_limit += 1000;
 784             }
 785             flow_limit = MIN(ofproto_flow_limit, MAX(flow_limit, 1000));
 786             atomic_store_relaxed(&udpif->flow_limit, flow_limit);
 787
 788             if (duration > 2000) {
 789                 VLOG_INFO("Spent an unreasonably long %lldms dumping flows",
 790                           duration);
 791             }
 792
 793             poll_timer_wait_until(start_time + MIN(ofproto_max_idle, 500));
 794             seq_wait(udpif->reval_seq, last_reval_seq);
 795             latch_wait(&udpif->exit_latch);
 796             poll_block();
 797         }
 798     }
 799
 800     return NULL;
 801 }
 802 \f
 803 static enum upcall_type
 804 classify_upcall(enum dpif_upcall_type type, const struct nlattr *userdata)
 805 {
 806     union user_action_cookie cookie;
 807     size_t userdata_len;
 808
 809     /* First look at the upcall type. */
 810     switch (type) {
 811     case DPIF_UC_ACTION:
 812         break;
 813
 814     case DPIF_UC_MISS:
 815         return MISS_UPCALL;
 816
 817     case DPIF_N_UC_TYPES:
 818     default:
 819         VLOG_WARN_RL(&rl, "upcall has unexpected type %"PRIu32, type);
 820         return BAD_UPCALL;
 821     }
 822
 823     /* "action" upcalls need a closer look. */
 824     if (!userdata) {
 825         VLOG_WARN_RL(&rl, "action upcall missing cookie");
 826         return BAD_UPCALL;
 827     }
 828     userdata_len = nl_attr_get_size(userdata);
 829     if (userdata_len < sizeof cookie.type
 830         || userdata_len > sizeof cookie) {
 831         VLOG_WARN_RL(&rl, "action upcall cookie has unexpected size %"PRIuSIZE,
 832                      userdata_len);
 833         return BAD_UPCALL;
 834     }
 835     memset(&cookie, 0, sizeof cookie);
 836     memcpy(&cookie, nl_attr_get(userdata), userdata_len);
 837     if (userdata_len == MAX(8, sizeof cookie.sflow)
 838         && cookie.type == USER_ACTION_COOKIE_SFLOW) {
 839         return SFLOW_UPCALL;
 840     } else if (userdata_len == MAX(8, sizeof cookie.slow_path)
 841                && cookie.type == USER_ACTION_COOKIE_SLOW_PATH) {
 842         return MISS_UPCALL;
 843     } else if (userdata_len == MAX(8, sizeof cookie.flow_sample)
 844                && cookie.type == USER_ACTION_COOKIE_FLOW_SAMPLE) {
 845         return FLOW_SAMPLE_UPCALL;
 846     } else if (userdata_len == MAX(8, sizeof cookie.ipfix)
 847                && cookie.type == USER_ACTION_COOKIE_IPFIX) {
 848         return IPFIX_UPCALL;
 849     } else {
 850         VLOG_WARN_RL(&rl, "invalid user cookie of type %"PRIu16
 851                      " and size %"PRIuSIZE, cookie.type, userdata_len);
 852         return BAD_UPCALL;
 853     }
 854 }
 855
 856 /* Calculates slow path actions for 'xout'.  'buf' must statically be
 857  * initialized with at least 128 bytes of space. */
 858 static void
 859 compose_slow_path(struct udpif *udpif, struct xlate_out *xout,
 860                   const struct flow *flow, odp_port_t odp_in_port,
 861                   struct ofpbuf *buf)
 862 {
 863     union user_action_cookie cookie;
 864     odp_port_t port;
 865     uint32_t pid;
 866
 867     cookie.type = USER_ACTION_COOKIE_SLOW_PATH;
 868     cookie.slow_path.unused = 0;
 869     cookie.slow_path.reason = xout->slow;
 870
 871     port = xout->slow & (SLOW_CFM | SLOW_BFD | SLOW_LACP | SLOW_STP)
 872         ? ODPP_NONE
 873         : odp_in_port;
 874     pid = dpif_port_get_pid(udpif->dpif, port, flow_hash_5tuple(flow, 0));
 875     odp_put_userspace_action(pid, &cookie, sizeof cookie.slow_path, ODPP_NONE,
 876                              buf);
 877 }
 878
 879 /* If there is no error, the upcall must be destroyed with upcall_uninit()
 880  * before quiescing, as the referred objects are guaranteed to exist only
 881  * until the calling thread quiesces.  Otherwise, do not call upcall_uninit()
 882  * since the 'upcall->put_actions' remains uninitialized. */
 883 static int
 884 upcall_receive(struct upcall *upcall, const struct dpif_backer *backer,
 885                const struct ofpbuf *packet, enum dpif_upcall_type type,
 886                const struct nlattr *userdata, const struct flow *flow,
 887                const ovs_u128 *ufid, const int pmd_id)
 888 {
 889     int error;
 890
 891     error = xlate_lookup(backer, flow, &upcall->ofproto, &upcall->ipfix,
 892                          &upcall->sflow, NULL, &upcall->in_port);
 893     if (error) {
 894         return error;
 895     }
 896
 897     upcall->flow = flow;
 898     upcall->packet = packet;
 899     upcall->ufid = ufid;
 900     upcall->pmd_id = pmd_id;
 901     upcall->type = type;
 902     upcall->userdata = userdata;
 903     ofpbuf_init(&upcall->put_actions, 0);
 904
 905     upcall->xout_initialized = false;
 906     upcall->vsp_adjusted = false;
 907     upcall->ukey_persists = false;
 908
 909     upcall->ukey = NULL;
 910     upcall->key = NULL;
 911     upcall->key_len = 0;
 912
 913     upcall->out_tun_key = NULL;
 914
 915     return 0;
 916 }
 917
 918 static void
 919 upcall_xlate(struct udpif *udpif, struct upcall *upcall,
 920              struct ofpbuf *odp_actions)
 921 {
 922     struct dpif_flow_stats stats;
 923     struct xlate_in xin;
 924
 925     stats.n_packets = 1;
 926     stats.n_bytes = ofpbuf_size(upcall->packet);
 927     stats.used = time_msec();
 928     stats.tcp_flags = ntohs(upcall->flow->tcp_flags);
 929
 930     xlate_in_init(&xin, upcall->ofproto, upcall->flow, upcall->in_port, NULL,
 931                   stats.tcp_flags, upcall->packet);
 932     xin.odp_actions = odp_actions;
 933
 934     if (upcall->type == DPIF_UC_MISS) {
 935         xin.resubmit_stats = &stats;
 936     } else {
 937         /* For non-miss upcalls, there's a flow in the datapath which this
 938          * packet was accounted to.  Presumably the revalidators will deal
 939          * with pushing its stats eventually. */
 940     }
 941
 942     upcall->dump_seq = seq_read(udpif->dump_seq);
 943     upcall->reval_seq = seq_read(udpif->reval_seq);
 944     xlate_actions(&xin, &upcall->xout);
 945     upcall->xout_initialized = true;
 946
 947     /* Special case for fail-open mode.
 948      *
 949      * If we are in fail-open mode, but we are connected to a controller too,
 950      * then we should send the packet up to the controller in the hope that it
 951      * will try to set up a flow and thereby allow us to exit fail-open.
 952      *
 953      * See the top-level comment in fail-open.c for more information.
 954      *
 955      * Copy packets before they are modified by execution. */
 956     if (upcall->xout.fail_open) {
 957         const struct ofpbuf *packet = upcall->packet;
 958         struct ofproto_packet_in *pin;
 959
 960         pin = xmalloc(sizeof *pin);
 961         pin->up.packet = xmemdup(ofpbuf_data(packet), ofpbuf_size(packet));
 962         pin->up.packet_len = ofpbuf_size(packet);
 963         pin->up.reason = OFPR_NO_MATCH;
 964         pin->up.table_id = 0;
 965         pin->up.cookie = OVS_BE64_MAX;
 966         flow_get_metadata(upcall->flow, &pin->up.fmd);
 967         pin->send_len = 0; /* Not used for flow table misses. */
 968         pin->miss_type = OFPROTO_PACKET_IN_NO_MISS;
 969         ofproto_dpif_send_packet_in(upcall->ofproto, pin);
 970     }
 971
 972     if (!upcall->xout.slow) {
 973         ofpbuf_use_const(&upcall->put_actions,
 974                          ofpbuf_data(upcall->xout.odp_actions),
 975                          ofpbuf_size(upcall->xout.odp_actions));
 976     } else {
 977         ofpbuf_init(&upcall->put_actions, 0);
 978         compose_slow_path(udpif, &upcall->xout, upcall->flow,
 979                           upcall->flow->in_port.odp_port,
 980                           &upcall->put_actions);
 981     }
 982
 983     upcall->ukey = ukey_create_from_upcall(upcall);
 984 }
 985
 986 static void
 987 upcall_uninit(struct upcall *upcall)
 988 {
 989     if (upcall) {
 990         if (upcall->xout_initialized) {
 991             xlate_out_uninit(&upcall->xout);
 992         }
 993         ofpbuf_uninit(&upcall->put_actions);
 994         if (!upcall->ukey_persists) {
 995             ukey_delete__(upcall->ukey);
 996         }
 997     }
 998 }
 999
1000 static int
1001 upcall_cb(const struct ofpbuf *packet, const struct flow *flow, ovs_u128 *ufid,
1002           int pmd_id, enum dpif_upcall_type type,
1003           const struct nlattr *userdata, struct ofpbuf *actions,
1004           struct flow_wildcards *wc, struct ofpbuf *put_actions, void *aux)
1005 {
1006     struct udpif *udpif = aux;
1007     unsigned int flow_limit;
1008     struct upcall upcall;
1009     bool megaflow;
1010     int error;
1011
1012     atomic_read_relaxed(&enable_megaflows, &megaflow);
1013     atomic_read_relaxed(&udpif->flow_limit, &flow_limit);
1014
1015     error = upcall_receive(&upcall, udpif->backer, packet, type, userdata,
1016                            flow, ufid, pmd_id);
1017     if (error) {
1018         return error;
1019     }
1020
1021     error = process_upcall(udpif, &upcall, actions);
1022     if (error) {
1023         goto out;
1024     }
1025
1026     if (upcall.xout.slow && put_actions) {
1027         ofpbuf_put(put_actions, ofpbuf_data(&upcall.put_actions),
1028                    ofpbuf_size(&upcall.put_actions));
1029     }
1030
1031     if (OVS_LIKELY(wc)) {
1032         if (megaflow) {
1033             /* XXX: This could be avoided with sufficient API changes. */
1034             *wc = upcall.xout.wc;
1035         } else {
1036             flow_wildcards_init_for_packet(wc, flow);
1037         }
1038     }
1039
1040     if (udpif_get_n_flows(udpif) >= flow_limit) {
1041         error = ENOSPC;
1042         goto out;
1043     }
1044
1045     if (upcall.ukey && !ukey_install(udpif, upcall.ukey)) {
1046         error = ENOSPC;
1047     }
1048
1049 out:
1050     if (!error) {
1051         upcall.ukey_persists = true;
1052     }
1053     upcall_uninit(&upcall);
1054     return error;
1055 }
1056
1057 static int
1058 process_upcall(struct udpif *udpif, struct upcall *upcall,
1059                struct ofpbuf *odp_actions)
1060 {
1061     const struct nlattr *userdata = upcall->userdata;
1062     const struct ofpbuf *packet = upcall->packet;
1063     const struct flow *flow = upcall->flow;
1064
1065     switch (classify_upcall(upcall->type, userdata)) {
1066     case MISS_UPCALL:
1067         upcall_xlate(udpif, upcall, odp_actions);
1068         return 0;
1069
1070     case SFLOW_UPCALL:
1071         if (upcall->sflow) {
1072             union user_action_cookie cookie;
1073
1074             memset(&cookie, 0, sizeof cookie);
1075             memcpy(&cookie, nl_attr_get(userdata), sizeof cookie.sflow);
1076             dpif_sflow_received(upcall->sflow, packet, flow,
1077                                 flow->in_port.odp_port, &cookie);
1078         }
1079         break;
1080
1081     case IPFIX_UPCALL:
1082         if (upcall->ipfix) {
1083             union user_action_cookie cookie;
1084             struct flow_tnl output_tunnel_key;
1085
1086             memset(&cookie, 0, sizeof cookie);
1087             memcpy(&cookie, nl_attr_get(userdata), sizeof cookie.ipfix);
1088
1089             if (upcall->out_tun_key) {
1090                 memset(&output_tunnel_key, 0, sizeof output_tunnel_key);
1091                 odp_tun_key_from_attr(upcall->out_tun_key,
1092                                       &output_tunnel_key);
1093             }
1094             dpif_ipfix_bridge_sample(upcall->ipfix, packet, flow,
1095                                      flow->in_port.odp_port,
1096                                      cookie.ipfix.output_odp_port,
1097                                      upcall->out_tun_key ?
1098                                          &output_tunnel_key : NULL);
1099         }
1100         break;
1101
1102     case FLOW_SAMPLE_UPCALL:
1103         if (upcall->ipfix) {
1104             union user_action_cookie cookie;
1105
1106             memset(&cookie, 0, sizeof cookie);
1107             memcpy(&cookie, nl_attr_get(userdata), sizeof cookie.flow_sample);
1108
1109             /* The flow reflects exactly the contents of the packet.
1110              * Sample the packet using it. */
1111             dpif_ipfix_flow_sample(upcall->ipfix, packet, flow,
1112                                    cookie.flow_sample.collector_set_id,
1113                                    cookie.flow_sample.probability,
1114                                    cookie.flow_sample.obs_domain_id,
1115                                    cookie.flow_sample.obs_point_id);
1116         }
1117         break;
1118
1119     case BAD_UPCALL:
1120         break;
1121     }
1122
1123     return EAGAIN;
1124 }
1125
1126 static void
1127 handle_upcalls(struct udpif *udpif, struct upcall *upcalls,
1128                size_t n_upcalls)
1129 {
1130     struct dpif_op *opsp[UPCALL_MAX_BATCH * 2];
1131     struct ukey_op ops[UPCALL_MAX_BATCH * 2];
1132     unsigned int flow_limit;
1133     size_t n_ops, n_opsp, i;
1134     bool may_put;
1135     bool megaflow;
1136
1137     atomic_read_relaxed(&udpif->flow_limit, &flow_limit);
1138     atomic_read_relaxed(&enable_megaflows, &megaflow);
1139
1140     may_put = udpif_get_n_flows(udpif) < flow_limit;
1141
1142     /* Handle the packets individually in order of arrival.
1143      *
1144      *   - For SLOW_CFM, SLOW_LACP, SLOW_STP, and SLOW_BFD, translation is what
1145      *     processes received packets for these protocols.
1146      *
1147      *   - For SLOW_CONTROLLER, translation sends the packet to the OpenFlow
1148      *     controller.
1149      *
1150      * The loop fills 'ops' with an array of operations to execute in the
1151      * datapath. */
1152     n_ops = 0;
1153     for (i = 0; i < n_upcalls; i++) {
1154         struct upcall *upcall = &upcalls[i];
1155         const struct ofpbuf *packet = upcall->packet;
1156         struct ukey_op *op;
1157
1158         if (upcall->vsp_adjusted) {
1159             /* This packet was received on a VLAN splinter port.  We added a
1160              * VLAN to the packet to make the packet resemble the flow, but the
1161              * actions were composed assuming that the packet contained no
1162              * VLAN.  So, we must remove the VLAN header from the packet before
1163              * trying to execute the actions. */
1164             if (ofpbuf_size(upcall->xout.odp_actions)) {
1165                 eth_pop_vlan(CONST_CAST(struct ofpbuf *, upcall->packet));
1166             }
1167
1168             /* Remove the flow vlan tags inserted by vlan splinter logic
1169              * to ensure megaflow masks generated match the data path flow. */
1170             CONST_CAST(struct flow *, upcall->flow)->vlan_tci = 0;
1171         }
1172
1173         /* Do not install a flow into the datapath if:
1174          *
1175          *    - The datapath already has too many flows.
1176          *
1177          *    - We received this packet via some flow installed in the kernel
1178          *      already. */
1179         if (may_put && upcall->type == DPIF_UC_MISS) {
1180             struct udpif_key *ukey = upcall->ukey;
1181
1182             upcall->ukey_persists = true;
1183             op = &ops[n_ops++];
1184
1185             op->ukey = ukey;
1186             op->dop.type = DPIF_OP_FLOW_PUT;
1187             op->dop.u.flow_put.flags = DPIF_FP_CREATE;
1188             op->dop.u.flow_put.key = ukey->key;
1189             op->dop.u.flow_put.key_len = ukey->key_len;
1190             op->dop.u.flow_put.mask = ukey->mask;
1191             op->dop.u.flow_put.mask_len = ukey->mask_len;
1192             op->dop.u.flow_put.ufid = upcall->ufid;
1193             op->dop.u.flow_put.stats = NULL;
1194             op->dop.u.flow_put.actions = ofpbuf_data(ukey->actions);
1195             op->dop.u.flow_put.actions_len = ofpbuf_size(ukey->actions);
1196         }
1197
1198         if (ofpbuf_size(upcall->xout.odp_actions)) {
1199             op = &ops[n_ops++];
1200             op->ukey = NULL;
1201             op->dop.type = DPIF_OP_EXECUTE;
1202             op->dop.u.execute.packet = CONST_CAST(struct ofpbuf *, packet);
1203             odp_key_to_pkt_metadata(upcall->key, upcall->key_len,
1204                                     &op->dop.u.execute.md);
1205             op->dop.u.execute.actions = ofpbuf_data(upcall->xout.odp_actions);
1206             op->dop.u.execute.actions_len = ofpbuf_size(upcall->xout.odp_actions);
1207             op->dop.u.execute.needs_help = (upcall->xout.slow & SLOW_ACTION) != 0;
1208             op->dop.u.execute.probe = false;
1209         }
1210     }
1211
1212     /* Execute batch.
1213      *
1214      * We install ukeys before installing the flows, locking them for exclusive
1215      * access by this thread for the period of installation. This ensures that
1216      * other threads won't attempt to delete the flows as we are creating them.
1217      */
1218     n_opsp = 0;
1219     for (i = 0; i < n_ops; i++) {
1220         struct udpif_key *ukey = ops[i].ukey;
1221
1222         if (ukey) {
1223             /* If we can't install the ukey, don't install the flow. */
1224             if (!ukey_install_start(udpif, ukey)) {
1225                 ukey_delete__(ukey);
1226                 ops[i].ukey = NULL;
1227                 continue;
1228             }
1229         }
1230         opsp[n_opsp++] = &ops[i].dop;
1231     }
1232     dpif_operate(udpif->dpif, opsp, n_opsp);
1233     for (i = 0; i < n_ops; i++) {
1234         if (ops[i].ukey) {
1235             ukey_install_finish(ops[i].ukey, ops[i].dop.error);
1236         }
1237     }
1238 }
1239
1240 static uint32_t
1241 get_ufid_hash(const ovs_u128 *ufid)
1242 {
1243     return ufid->u32[0];
1244 }
1245
1246 static struct udpif_key *
1247 ukey_lookup(struct udpif *udpif, const ovs_u128 *ufid)
1248 {
1249     struct udpif_key *ukey;
1250     int idx = get_ufid_hash(ufid) % N_UMAPS;
1251     struct cmap *cmap = &udpif->ukeys[idx].cmap;
1252
1253     CMAP_FOR_EACH_WITH_HASH (ukey, cmap_node, get_ufid_hash(ufid), cmap) {
1254         if (ovs_u128_equal(&ukey->ufid, ufid)) {
1255             return ukey;
1256         }
1257     }
1258     return NULL;
1259 }
1260
1261 static struct udpif_key *
1262 ukey_create__(const struct nlattr *key, size_t key_len,
1263               const struct nlattr *mask, size_t mask_len,
1264               bool ufid_present, const ovs_u128 *ufid,
1265               const int pmd_id, const struct ofpbuf *actions,
1266               uint64_t dump_seq, uint64_t reval_seq, long long int used)
1267     OVS_NO_THREAD_SAFETY_ANALYSIS
1268 {
1269     struct udpif_key *ukey = xmalloc(sizeof *ukey);
1270
1271     memcpy(&ukey->keybuf, key, key_len);
1272     ukey->key = &ukey->keybuf.nla;
1273     ukey->key_len = key_len;
1274     memcpy(&ukey->maskbuf, mask, mask_len);
1275     ukey->mask = &ukey->maskbuf.nla;
1276     ukey->mask_len = mask_len;
1277     ukey->ufid_present = ufid_present;
1278     ukey->ufid = *ufid;
1279     ukey->pmd_id = pmd_id;
1280     ukey->hash = get_ufid_hash(&ukey->ufid);
1281     ukey->actions = ofpbuf_clone(actions);
1282
1283     ovs_mutex_init(&ukey->mutex);
1284     ukey->dump_seq = dump_seq;
1285     ukey->reval_seq = reval_seq;
1286     ukey->flow_exists = false;
1287     ukey->created = time_msec();
1288     memset(&ukey->stats, 0, sizeof ukey->stats);
1289     ukey->stats.used = used;
1290     ukey->xcache = NULL;
1291
1292     return ukey;
1293 }
1294
1295 static struct udpif_key *
1296 ukey_create_from_upcall(const struct upcall *upcall)
1297 {
1298     struct odputil_keybuf keystub, maskstub;
1299     struct ofpbuf keybuf, maskbuf;
1300     bool recirc, megaflow;
1301
1302     if (upcall->key_len) {
1303         ofpbuf_use_const(&keybuf, upcall->key, upcall->key_len);
1304     } else {
1305         /* dpif-netdev doesn't provide a netlink-formatted flow key in the
1306          * upcall, so convert the upcall's flow here. */
1307         ofpbuf_use_stack(&keybuf, &keystub, sizeof keystub);
1308         odp_flow_key_from_flow(&keybuf, upcall->flow, &upcall->xout.wc.masks,
1309                                upcall->flow->in_port.odp_port, true);
1310     }
1311
1312     atomic_read_relaxed(&enable_megaflows, &megaflow);
1313     recirc = ofproto_dpif_get_enable_recirc(upcall->ofproto);
1314     ofpbuf_use_stack(&maskbuf, &maskstub, sizeof maskstub);
1315     if (megaflow) {
1316         size_t max_mpls;
1317
1318         max_mpls = ofproto_dpif_get_max_mpls_depth(upcall->ofproto);
1319         odp_flow_key_from_mask(&maskbuf, &upcall->xout.wc.masks, upcall->flow,
1320                                UINT32_MAX, max_mpls, recirc);
1321     }
1322
1323     return ukey_create__(ofpbuf_data(&keybuf), ofpbuf_size(&keybuf),
1324                          ofpbuf_data(&maskbuf), ofpbuf_size(&maskbuf),
1325                          true, upcall->ufid, upcall->pmd_id,
1326                          &upcall->put_actions, upcall->dump_seq,
1327                          upcall->reval_seq, 0);
1328 }
1329
1330 static int
1331 ukey_create_from_dpif_flow(const struct udpif *udpif,
1332                            const struct dpif_flow *flow,
1333                            struct udpif_key **ukey)
1334 {
1335     struct dpif_flow full_flow;
1336     struct ofpbuf actions;
1337     uint64_t dump_seq, reval_seq;
1338     uint64_t stub[DPIF_FLOW_BUFSIZE / 8];
1339
1340     if (!flow->key_len) {
1341         struct ofpbuf buf;
1342         int err;
1343
1344         /* If the key was not provided by the datapath, fetch the full flow. */
1345         ofpbuf_use_stack(&buf, &stub, sizeof stub);
1346         err = dpif_flow_get(udpif->dpif, NULL, 0, &flow->ufid,
1347                             flow->pmd_id, &buf, &full_flow);
1348         if (err) {
1349             return err;
1350         }
1351         flow = &full_flow;
1352     }
1353     dump_seq = seq_read(udpif->dump_seq);
1354     reval_seq = seq_read(udpif->reval_seq);
1355     ofpbuf_use_const(&actions, &flow->actions, flow->actions_len);
1356     *ukey = ukey_create__(flow->key, flow->key_len,
1357                           flow->mask, flow->mask_len, flow->ufid_present,
1358                           &flow->ufid, flow->pmd_id, &actions, dump_seq,
1359                           reval_seq, flow->stats.used);
1360
1361     return 0;
1362 }
1363
1364 /* Attempts to insert a ukey into the shared ukey maps.
1365  *
1366  * On success, returns true, installs the ukey and returns it in a locked
1367  * state. Otherwise, returns false. */
1368 static bool
1369 ukey_install_start(struct udpif *udpif, struct udpif_key *new_ukey)
1370     OVS_TRY_LOCK(true, new_ukey->mutex)
1371 {
1372     struct umap *umap;
1373     struct udpif_key *old_ukey;
1374     uint32_t idx;
1375     bool locked = false;
1376
1377     idx = new_ukey->hash % N_UMAPS;
1378     umap = &udpif->ukeys[idx];
1379     ovs_mutex_lock(&umap->mutex);
1380     old_ukey = ukey_lookup(udpif, &new_ukey->ufid);
1381     if (old_ukey) {
1382         /* Uncommon case: A ukey is already installed with the same UFID. */
1383         if (old_ukey->key_len == new_ukey->key_len
1384             && !memcmp(old_ukey->key, new_ukey->key, new_ukey->key_len)) {
1385             COVERAGE_INC(handler_duplicate_upcall);
1386         } else {
1387             struct ds ds = DS_EMPTY_INITIALIZER;
1388
1389             odp_format_ufid(&old_ukey->ufid, &ds);
1390             ds_put_cstr(&ds, " ");
1391             odp_flow_key_format(old_ukey->key, old_ukey->key_len, &ds);
1392             ds_put_cstr(&ds, "\n");
1393             odp_format_ufid(&new_ukey->ufid, &ds);
1394             ds_put_cstr(&ds, " ");
1395             odp_flow_key_format(new_ukey->key, new_ukey->key_len, &ds);
1396
1397             VLOG_WARN_RL(&rl, "Conflicting ukey for flows:\n%s", ds_cstr(&ds));
1398             ds_destroy(&ds);
1399         }
1400     } else {
1401         ovs_mutex_lock(&new_ukey->mutex);
1402         cmap_insert(&umap->cmap, &new_ukey->cmap_node, new_ukey->hash);
1403         locked = true;
1404     }
1405     ovs_mutex_unlock(&umap->mutex);
1406
1407     return locked;
1408 }
1409
1410 static void
1411 ukey_install_finish__(struct udpif_key *ukey) OVS_REQUIRES(ukey->mutex)
1412 {
1413     ukey->flow_exists = true;
1414 }
1415
1416 static bool
1417 ukey_install_finish(struct udpif_key *ukey, int error)
1418     OVS_RELEASES(ukey->mutex)
1419 {
1420     if (!error) {
1421         ukey_install_finish__(ukey);
1422     }
1423     ovs_mutex_unlock(&ukey->mutex);
1424
1425     return !error;
1426 }
1427
1428 static bool
1429 ukey_install(struct udpif *udpif, struct udpif_key *ukey)
1430 {
1431     /* The usual way to keep 'ukey->flow_exists' in sync with the datapath is
1432      * to call ukey_install_start(), install the corresponding datapath flow,
1433      * then call ukey_install_finish(). The netdev interface using upcall_cb()
1434      * doesn't provide a function to separately finish the flow installation,
1435      * so we perform the operations together here.
1436      *
1437      * This is fine currently, as revalidator threads will only delete this
1438      * ukey during revalidator_sweep() and only if the dump_seq is mismatched.
1439      * It is unlikely for a revalidator thread to advance dump_seq and reach
1440      * the next GC phase between ukey creation and flow installation. */
1441     return ukey_install_start(udpif, ukey) && ukey_install_finish(ukey, 0);
1442 }
1443
1444 /* Searches for a ukey in 'udpif->ukeys' that matches 'flow' and attempts to
1445  * lock the ukey. If the ukey does not exist, create it.
1446  *
1447  * Returns 0 on success, setting *result to the matching ukey and returning it
1448  * in a locked state. Otherwise, returns an errno and clears *result. EBUSY
1449  * indicates that another thread is handling this flow. Other errors indicate
1450  * an unexpected condition creating a new ukey.
1451  *
1452  * *error is an output parameter provided to appease the threadsafety analyser,
1453  * and its value matches the return value. */
1454 static int
1455 ukey_acquire(struct udpif *udpif, const struct dpif_flow *flow,
1456              struct udpif_key **result, int *error)
1457     OVS_TRY_LOCK(0, (*result)->mutex)
1458 {
1459     struct udpif_key *ukey;
1460     int retval;
1461
1462     ukey = ukey_lookup(udpif, &flow->ufid);
1463     if (ukey) {
1464         retval = ovs_mutex_trylock(&ukey->mutex);
1465     } else {
1466         /* Usually we try to avoid installing flows from revalidator threads,
1467          * because locking on a umap may cause handler threads to block.
1468          * However there are certain cases, like when ovs-vswitchd is
1469          * restarted, where it is desirable to handle flows that exist in the
1470          * datapath gracefully (ie, don't just clear the datapath). */
1471         bool install;
1472
1473         retval = ukey_create_from_dpif_flow(udpif, flow, &ukey);
1474         if (retval) {
1475             goto done;
1476         }
1477         install = ukey_install_start(udpif, ukey);
1478         if (install) {
1479             ukey_install_finish__(ukey);
1480             retval = 0;
1481         } else {
1482             ukey_delete__(ukey);
1483             retval = EBUSY;
1484         }
1485     }
1486
1487 done:
1488     *error = retval;
1489     if (retval) {
1490         *result = NULL;
1491     } else {
1492         *result = ukey;
1493     }
1494     return retval;
1495 }
1496
1497 static void
1498 ukey_delete__(struct udpif_key *ukey)
1499     OVS_NO_THREAD_SAFETY_ANALYSIS
1500 {
1501     if (ukey) {
1502         xlate_cache_delete(ukey->xcache);
1503         ofpbuf_delete(ukey->actions);
1504         ovs_mutex_destroy(&ukey->mutex);
1505         free(ukey);
1506     }
1507 }
1508
1509 static void
1510 ukey_delete(struct umap *umap, struct udpif_key *ukey)
1511     OVS_REQUIRES(umap->mutex)
1512 {
1513     cmap_remove(&umap->cmap, &ukey->cmap_node, ukey->hash);
1514     ovsrcu_postpone(ukey_delete__, ukey);
1515 }
1516
1517 static bool
1518 should_revalidate(const struct udpif *udpif, uint64_t packets,
1519                   long long int used)
1520 {
1521     long long int metric, now, duration;
1522
1523     if (udpif->dump_duration < 200) {
1524         /* We are likely to handle full revalidation for the flows. */
1525         return true;
1526     }
1527
1528     /* Calculate the mean time between seeing these packets. If this
1529      * exceeds the threshold, then delete the flow rather than performing
1530      * costly revalidation for flows that aren't being hit frequently.
1531      *
1532      * This is targeted at situations where the dump_duration is high (~1s),
1533      * and revalidation is triggered by a call to udpif_revalidate(). In
1534      * these situations, revalidation of all flows causes fluctuations in the
1535      * flow_limit due to the interaction with the dump_duration and max_idle.
1536      * This tends to result in deletion of low-throughput flows anyway, so
1537      * skip the revalidation and just delete those flows. */
1538     packets = MAX(packets, 1);
1539     now = MAX(used, time_msec());
1540     duration = now - used;
1541     metric = duration / packets;
1542
1543     if (metric < 200) {
1544         /* The flow is receiving more than ~5pps, so keep it. */
1545         return true;
1546     }
1547     return false;
1548 }
1549
1550 static bool
1551 revalidate_ukey(struct udpif *udpif, struct udpif_key *ukey,
1552                 const struct dpif_flow_stats *stats, uint64_t reval_seq)
1553     OVS_REQUIRES(ukey->mutex)
1554 {
1555     uint64_t slow_path_buf[128 / 8];
1556     struct xlate_out xout, *xoutp;
1557     struct netflow *netflow;
1558     struct ofproto_dpif *ofproto;
1559     struct dpif_flow_stats push;
1560     struct ofpbuf xout_actions;
1561     struct flow flow, dp_mask;
1562     uint32_t *dp32, *xout32;
1563     ofp_port_t ofp_in_port;
1564     struct xlate_in xin;
1565     long long int last_used;
1566     int error;
1567     size_t i;
1568     bool ok;
1569     bool need_revalidate;
1570
1571     ok = false;
1572     xoutp = NULL;
1573     netflow = NULL;
1574
1575     need_revalidate = (ukey->reval_seq != reval_seq);
1576     last_used = ukey->stats.used;
1577     push.used = stats->used;
1578     push.tcp_flags = stats->tcp_flags;
1579     push.n_packets = (stats->n_packets > ukey->stats.n_packets
1580                       ? stats->n_packets - ukey->stats.n_packets
1581                       : 0);
1582     push.n_bytes = (stats->n_bytes > ukey->stats.n_bytes
1583                     ? stats->n_bytes - ukey->stats.n_bytes
1584                     : 0);
1585
1586     if (need_revalidate && last_used
1587         && !should_revalidate(udpif, push.n_packets, last_used)) {
1588         ok = false;
1589         goto exit;
1590     }
1591
1592     /* We will push the stats, so update the ukey stats cache. */
1593     ukey->stats = *stats;
1594     if (!push.n_packets && !need_revalidate) {
1595         ok = true;
1596         goto exit;
1597     }
1598
1599     if (ukey->xcache && !need_revalidate) {
1600         xlate_push_stats(ukey->xcache, &push);
1601         ok = true;
1602         goto exit;
1603     }
1604
1605     if (odp_flow_key_to_flow(ukey->key, ukey->key_len, &flow)
1606         == ODP_FIT_ERROR) {
1607         goto exit;
1608     }
1609
1610     error = xlate_lookup(udpif->backer, &flow, &ofproto, NULL, NULL, &netflow,
1611                          &ofp_in_port);
1612     if (error) {
1613         goto exit;
1614     }
1615
1616     if (need_revalidate) {
1617         xlate_cache_clear(ukey->xcache);
1618     }
1619     if (!ukey->xcache) {
1620         ukey->xcache = xlate_cache_new();
1621     }
1622
1623     xlate_in_init(&xin, ofproto, &flow, ofp_in_port, NULL, push.tcp_flags,
1624                   NULL);
1625     if (push.n_packets) {
1626         xin.resubmit_stats = &push;
1627         xin.may_learn = true;
1628     }
1629     xin.xcache = ukey->xcache;
1630     xin.skip_wildcards = !need_revalidate;
1631     xlate_actions(&xin, &xout);
1632     xoutp = &xout;
1633
1634     if (!need_revalidate) {
1635         ok = true;
1636         goto exit;
1637     }
1638
1639     if (!xout.slow) {
1640         ofpbuf_use_const(&xout_actions, ofpbuf_data(xout.odp_actions),
1641                          ofpbuf_size(xout.odp_actions));
1642     } else {
1643         ofpbuf_use_stack(&xout_actions, slow_path_buf, sizeof slow_path_buf);
1644         compose_slow_path(udpif, &xout, &flow, flow.in_port.odp_port,
1645                           &xout_actions);
1646     }
1647
1648     if (!ofpbuf_equal(&xout_actions, ukey->actions)) {
1649         goto exit;
1650     }
1651
1652     if (odp_flow_key_to_mask(ukey->mask, ukey->mask_len, &dp_mask, &flow)
1653         == ODP_FIT_ERROR) {
1654         goto exit;
1655     }
1656
1657     /* Since the kernel is free to ignore wildcarded bits in the mask, we can't
1658      * directly check that the masks are the same.  Instead we check that the
1659      * mask in the kernel is more specific i.e. less wildcarded, than what
1660      * we've calculated here.  This guarantees we don't catch any packets we
1661      * shouldn't with the megaflow. */
1662     dp32 = (uint32_t *) &dp_mask;
1663     xout32 = (uint32_t *) &xout.wc.masks;
1664     for (i = 0; i < FLOW_U32S; i++) {
1665         if ((dp32[i] | xout32[i]) != dp32[i]) {
1666             goto exit;
1667         }
1668     }
1669
1670     ok = true;
1671
1672 exit:
1673     if (ok) {
1674         ukey->reval_seq = reval_seq;
1675     }
1676     if (netflow && !ok) {
1677         netflow_flow_clear(netflow, &flow);
1678     }
1679     xlate_out_uninit(xoutp);
1680     return ok;
1681 }
1682
1683 static void
1684 delete_op_init__(struct udpif *udpif, struct ukey_op *op,
1685                  const struct dpif_flow *flow)
1686 {
1687     op->ukey = NULL;
1688     op->dop.type = DPIF_OP_FLOW_DEL;
1689     op->dop.u.flow_del.key = flow->key;
1690     op->dop.u.flow_del.key_len = flow->key_len;
1691     op->dop.u.flow_del.ufid = flow->ufid_present ? &flow->ufid : NULL;
1692     op->dop.u.flow_del.pmd_id = flow->pmd_id;
1693     op->dop.u.flow_del.stats = &op->stats;
1694     atomic_read_relaxed(&udpif->enable_ufid, &op->dop.u.flow_del.terse);
1695 }
1696
1697 static void
1698 delete_op_init(struct udpif *udpif, struct ukey_op *op, struct udpif_key *ukey)
1699 {
1700     op->ukey = ukey;
1701     op->dop.type = DPIF_OP_FLOW_DEL;
1702     op->dop.u.flow_del.key = ukey->key;
1703     op->dop.u.flow_del.key_len = ukey->key_len;
1704     op->dop.u.flow_del.ufid = ukey->ufid_present ? &ukey->ufid : NULL;
1705     op->dop.u.flow_del.pmd_id = ukey->pmd_id;
1706     op->dop.u.flow_del.stats = &op->stats;
1707     atomic_read_relaxed(&udpif->enable_ufid, &op->dop.u.flow_del.terse);
1708 }
1709
1710 static void
1711 push_ukey_ops__(struct udpif *udpif, struct ukey_op *ops, size_t n_ops)
1712 {
1713     struct dpif_op *opsp[REVALIDATE_MAX_BATCH];
1714     size_t i;
1715
1716     ovs_assert(n_ops <= REVALIDATE_MAX_BATCH);
1717     for (i = 0; i < n_ops; i++) {
1718         opsp[i] = &ops[i].dop;
1719     }
1720     dpif_operate(udpif->dpif, opsp, n_ops);
1721
1722     for (i = 0; i < n_ops; i++) {
1723         struct ukey_op *op = &ops[i];
1724         struct dpif_flow_stats *push, *stats, push_buf;
1725
1726         stats = op->dop.u.flow_del.stats;
1727         push = &push_buf;
1728
1729         if (op->ukey) {
1730             ovs_mutex_lock(&op->ukey->mutex);
1731             push->used = MAX(stats->used, op->ukey->stats.used);
1732             push->tcp_flags = stats->tcp_flags | op->ukey->stats.tcp_flags;
1733             push->n_packets = stats->n_packets - op->ukey->stats.n_packets;
1734             push->n_bytes = stats->n_bytes - op->ukey->stats.n_bytes;
1735             ovs_mutex_unlock(&op->ukey->mutex);
1736         } else {
1737             push = stats;
1738         }
1739
1740         if (push->n_packets || netflow_exists()) {
1741             const struct nlattr *key = op->dop.u.flow_del.key;
1742             size_t key_len = op->dop.u.flow_del.key_len;
1743             struct ofproto_dpif *ofproto;
1744             struct netflow *netflow;
1745             ofp_port_t ofp_in_port;
1746             struct flow flow;
1747             int error;
1748
1749             if (op->ukey) {
1750                 ovs_mutex_lock(&op->ukey->mutex);
1751                 if (op->ukey->xcache) {
1752                     xlate_push_stats(op->ukey->xcache, push);
1753                     ovs_mutex_unlock(&op->ukey->mutex);
1754                     continue;
1755                 }
1756                 ovs_mutex_unlock(&op->ukey->mutex);
1757                 key = op->ukey->key;
1758                 key_len = op->ukey->key_len;
1759             }
1760
1761             if (odp_flow_key_to_flow(key, key_len, &flow)
1762                 == ODP_FIT_ERROR) {
1763                 continue;
1764             }
1765
1766             error = xlate_lookup(udpif->backer, &flow, &ofproto,
1767                                  NULL, NULL, &netflow, &ofp_in_port);
1768             if (!error) {
1769                 struct xlate_in xin;
1770
1771                 xlate_in_init(&xin, ofproto, &flow, ofp_in_port, NULL,
1772                               push->tcp_flags, NULL);
1773                 xin.resubmit_stats = push->n_packets ? push : NULL;
1774                 xin.may_learn = push->n_packets > 0;
1775                 xin.skip_wildcards = true;
1776                 xlate_actions_for_side_effects(&xin);
1777
1778                 if (netflow) {
1779                     netflow_flow_clear(netflow, &flow);
1780                 }
1781             }
1782         }
1783     }
1784 }
1785
1786 static void
1787 push_ukey_ops(struct udpif *udpif, struct umap *umap,
1788               struct ukey_op *ops, size_t n_ops)
1789 {
1790     int i;
1791
1792     push_ukey_ops__(udpif, ops, n_ops);
1793     ovs_mutex_lock(&umap->mutex);
1794     for (i = 0; i < n_ops; i++) {
1795         ukey_delete(umap, ops[i].ukey);
1796     }
1797     ovs_mutex_unlock(&umap->mutex);
1798 }
1799
1800 static void
1801 log_unexpected_flow(const struct dpif_flow *flow, int error)
1802 {
1803     static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(10, 60);
1804     struct ds ds = DS_EMPTY_INITIALIZER;
1805
1806     ds_put_format(&ds, "Failed to acquire udpif_key corresponding to "
1807                   "unexpected flow (%s): ", ovs_strerror(error));
1808     odp_format_ufid(&flow->ufid, &ds);
1809     VLOG_WARN_RL(&rl, "%s", ds_cstr(&ds));
1810 }
1811
1812 static void
1813 revalidate(struct revalidator *revalidator)
1814 {
1815     struct udpif *udpif = revalidator->udpif;
1816     struct dpif_flow_dump_thread *dump_thread;
1817     uint64_t dump_seq, reval_seq;
1818     unsigned int flow_limit;
1819
1820     dump_seq = seq_read(udpif->dump_seq);
1821     reval_seq = seq_read(udpif->reval_seq);
1822     atomic_read_relaxed(&udpif->flow_limit, &flow_limit);
1823     dump_thread = dpif_flow_dump_thread_create(udpif->dump);
1824     for (;;) {
1825         struct ukey_op ops[REVALIDATE_MAX_BATCH];
1826         int n_ops = 0;
1827
1828         struct dpif_flow flows[REVALIDATE_MAX_BATCH];
1829         const struct dpif_flow *f;
1830         int n_dumped;
1831
1832         long long int max_idle;
1833         long long int now;
1834         size_t n_dp_flows;
1835         bool kill_them_all;
1836
1837         n_dumped = dpif_flow_dump_next(dump_thread, flows, ARRAY_SIZE(flows));
1838         if (!n_dumped) {
1839             break;
1840         }
1841
1842         now = time_msec();
1843
1844         /* In normal operation we want to keep flows around until they have
1845          * been idle for 'ofproto_max_idle' milliseconds.  However:
1846          *
1847          *     - If the number of datapath flows climbs above 'flow_limit',
1848          *       drop that down to 100 ms to try to bring the flows down to
1849          *       the limit.
1850          *
1851          *     - If the number of datapath flows climbs above twice
1852          *       'flow_limit', delete all the datapath flows as an emergency
1853          *       measure.  (We reassess this condition for the next batch of
1854          *       datapath flows, so we will recover before all the flows are
1855          *       gone.) */
1856         n_dp_flows = udpif_get_n_flows(udpif);
1857         kill_them_all = n_dp_flows > flow_limit * 2;
1858         max_idle = n_dp_flows > flow_limit ? 100 : ofproto_max_idle;
1859
1860         for (f = flows; f < &flows[n_dumped]; f++) {
1861             long long int used = f->stats.used;
1862             struct udpif_key *ukey;
1863             bool already_dumped, keep;
1864             int error;
1865
1866             if (ukey_acquire(udpif, f, &ukey, &error)) {
1867                 if (error == EBUSY) {
1868                     /* Another thread is processing this flow, so don't bother
1869                      * processing it.*/
1870                     COVERAGE_INC(upcall_ukey_contention);
1871                 } else {
1872                     log_unexpected_flow(f, error);
1873                     if (error != ENOENT) {
1874                         delete_op_init__(udpif, &ops[n_ops++], f);
1875                     }
1876                 }
1877                 continue;
1878             }
1879
1880             already_dumped = ukey->dump_seq == dump_seq;
1881             if (already_dumped) {
1882                 /* The flow has already been handled during this flow dump
1883                  * operation. Skip it. */
1884                 if (ukey->xcache) {
1885                     COVERAGE_INC(dumped_duplicate_flow);
1886                 } else {
1887                     COVERAGE_INC(dumped_new_flow);
1888                 }
1889                 ovs_mutex_unlock(&ukey->mutex);
1890                 continue;
1891             }
1892
1893             if (!used) {
1894                 used = ukey->created;
1895             }
1896             if (kill_them_all || (used && used < now - max_idle)) {
1897                 keep = false;
1898             } else {
1899                 keep = revalidate_ukey(udpif, ukey, &f->stats, reval_seq);
1900             }
1901             ukey->dump_seq = dump_seq;
1902             ukey->flow_exists = keep;
1903
1904             if (!keep) {
1905                 delete_op_init(udpif, &ops[n_ops++], ukey);
1906             }
1907             ovs_mutex_unlock(&ukey->mutex);
1908         }
1909
1910         if (n_ops) {
1911             push_ukey_ops__(udpif, ops, n_ops);
1912         }
1913         ovsrcu_quiesce();
1914     }
1915     dpif_flow_dump_thread_destroy(dump_thread);
1916 }
1917
1918 static bool
1919 handle_missed_revalidation(struct udpif *udpif, uint64_t reval_seq,
1920                            struct udpif_key *ukey)
1921 {
1922     struct dpif_flow_stats stats;
1923     bool keep;
1924
1925     COVERAGE_INC(revalidate_missed_dp_flow);
1926
1927     memset(&stats, 0, sizeof stats);
1928     ovs_mutex_lock(&ukey->mutex);
1929     keep = revalidate_ukey(udpif, ukey, &stats, reval_seq);
1930     ovs_mutex_unlock(&ukey->mutex);
1931
1932     return keep;
1933 }
1934
1935 static void
1936 revalidator_sweep__(struct revalidator *revalidator, bool purge)
1937 {
1938     struct udpif *udpif;
1939     uint64_t dump_seq, reval_seq;
1940     int slice;
1941
1942     udpif = revalidator->udpif;
1943     dump_seq = seq_read(udpif->dump_seq);
1944     reval_seq = seq_read(udpif->reval_seq);
1945     slice = revalidator - udpif->revalidators;
1946     ovs_assert(slice < udpif->n_revalidators);
1947
1948     for (int i = slice; i < N_UMAPS; i += udpif->n_revalidators) {
1949         struct ukey_op ops[REVALIDATE_MAX_BATCH];
1950         struct udpif_key *ukey;
1951         struct umap *umap = &udpif->ukeys[i];
1952         size_t n_ops = 0;
1953
1954         CMAP_FOR_EACH(ukey, cmap_node, &umap->cmap) {
1955             bool flow_exists, seq_mismatch;
1956
1957             /* Handler threads could be holding a ukey lock while it installs a
1958              * new flow, so don't hang around waiting for access to it. */
1959             if (ovs_mutex_trylock(&ukey->mutex)) {
1960                 continue;
1961             }
1962             flow_exists = ukey->flow_exists;
1963             seq_mismatch = (ukey->dump_seq != dump_seq
1964                             && ukey->reval_seq != reval_seq);
1965             ovs_mutex_unlock(&ukey->mutex);
1966
1967             if (flow_exists
1968                 && (purge
1969                     || (seq_mismatch
1970                         && !handle_missed_revalidation(udpif, reval_seq,
1971                                                        ukey)))) {
1972                 struct ukey_op *op = &ops[n_ops++];
1973
1974                 delete_op_init(udpif, op, ukey);
1975                 if (n_ops == REVALIDATE_MAX_BATCH) {
1976                     push_ukey_ops(udpif, umap, ops, n_ops);
1977                     n_ops = 0;
1978                 }
1979             } else if (!flow_exists) {
1980                 ovs_mutex_lock(&umap->mutex);
1981                 ukey_delete(umap, ukey);
1982                 ovs_mutex_unlock(&umap->mutex);
1983             }
1984         }
1985
1986         if (n_ops) {
1987             push_ukey_ops(udpif, umap, ops, n_ops);
1988         }
1989         ovsrcu_quiesce();
1990     }
1991 }
1992
1993 static void
1994 revalidator_sweep(struct revalidator *revalidator)
1995 {
1996     revalidator_sweep__(revalidator, false);
1997 }
1998
1999 static void
2000 revalidator_purge(struct revalidator *revalidator)
2001 {
2002     revalidator_sweep__(revalidator, true);
2003 }
2004 \f
2005 static void
2006 upcall_unixctl_show(struct unixctl_conn *conn, int argc OVS_UNUSED,
2007                     const char *argv[] OVS_UNUSED, void *aux OVS_UNUSED)
2008 {
2009     struct ds ds = DS_EMPTY_INITIALIZER;
2010     struct udpif *udpif;
2011
2012     LIST_FOR_EACH (udpif, list_node, &all_udpifs) {
2013         unsigned int flow_limit;
2014         bool ufid_enabled;
2015         size_t i;
2016
2017         atomic_read_relaxed(&udpif->flow_limit, &flow_limit);
2018         atomic_read_relaxed(&udpif->enable_ufid, &ufid_enabled);
2019
2020         ds_put_format(&ds, "%s:\n", dpif_name(udpif->dpif));
2021         ds_put_format(&ds, "\tflows         : (current %lu)"
2022             " (avg %u) (max %u) (limit %u)\n", udpif_get_n_flows(udpif),
2023             udpif->avg_n_flows, udpif->max_n_flows, flow_limit);
2024         ds_put_format(&ds, "\tdump duration : %lldms\n", udpif->dump_duration);
2025         ds_put_format(&ds, "\tufid enabled : ");
2026         if (ufid_enabled) {
2027             ds_put_format(&ds, "true\n");
2028         } else {
2029             ds_put_format(&ds, "false\n");
2030         }
2031         ds_put_char(&ds, '\n');
2032
2033         for (i = 0; i < n_revalidators; i++) {
2034             struct revalidator *revalidator = &udpif->revalidators[i];
2035             int j, elements = 0;
2036
2037             for (j = i; j < N_UMAPS; j += n_revalidators) {
2038                 elements += cmap_count(&udpif->ukeys[j].cmap);
2039             }
2040             ds_put_format(&ds, "\t%u: (keys %d)\n", revalidator->id, elements);
2041         }
2042     }
2043
2044     unixctl_command_reply(conn, ds_cstr(&ds));
2045     ds_destroy(&ds);
2046 }
2047
2048 /* Disable using the megaflows.
2049  *
2050  * This command is only needed for advanced debugging, so it's not
2051  * documented in the man page. */
2052 static void
2053 upcall_unixctl_disable_megaflows(struct unixctl_conn *conn,
2054                                  int argc OVS_UNUSED,
2055                                  const char *argv[] OVS_UNUSED,
2056                                  void *aux OVS_UNUSED)
2057 {
2058     atomic_store_relaxed(&enable_megaflows, false);
2059     udpif_flush_all_datapaths();
2060     unixctl_command_reply(conn, "megaflows disabled");
2061 }
2062
2063 /* Re-enable using megaflows.
2064  *
2065  * This command is only needed for advanced debugging, so it's not
2066  * documented in the man page. */
2067 static void
2068 upcall_unixctl_enable_megaflows(struct unixctl_conn *conn,
2069                                 int argc OVS_UNUSED,
2070                                 const char *argv[] OVS_UNUSED,
2071                                 void *aux OVS_UNUSED)
2072 {
2073     atomic_store_relaxed(&enable_megaflows, true);
2074     udpif_flush_all_datapaths();
2075     unixctl_command_reply(conn, "megaflows enabled");
2076 }
2077
2078 /* Disable skipping flow attributes during flow dump.
2079  *
2080  * This command is only needed for advanced debugging, so it's not
2081  * documented in the man page. */
2082 static void
2083 upcall_unixctl_disable_ufid(struct unixctl_conn *conn, int argc OVS_UNUSED,
2084                            const char *argv[] OVS_UNUSED, void *aux OVS_UNUSED)
2085 {
2086     struct udpif *udpif;
2087
2088     LIST_FOR_EACH (udpif, list_node, &all_udpifs) {
2089         atomic_store(&udpif->enable_ufid, false);
2090     }
2091     unixctl_command_reply(conn, "Datapath dumping tersely using UFID disabled");
2092 }
2093
2094 /* Re-enable skipping flow attributes during flow dump.
2095  *
2096  * This command is only needed for advanced debugging, so it's not documented
2097  * in the man page. */
2098 static void
2099 upcall_unixctl_enable_ufid(struct unixctl_conn *conn, int argc OVS_UNUSED,
2100                           const char *argv[] OVS_UNUSED, void *aux OVS_UNUSED)
2101 {
2102     struct udpif *udpif;
2103
2104     LIST_FOR_EACH (udpif, list_node, &all_udpifs) {
2105         atomic_store(&udpif->enable_ufid, true);
2106     }
2107     unixctl_command_reply(conn, "Datapath dumping tersely using UFID enabled");
2108 }
2109
2110 /* Set the flow limit.
2111  *
2112  * This command is only needed for advanced debugging, so it's not
2113  * documented in the man page. */
2114 static void
2115 upcall_unixctl_set_flow_limit(struct unixctl_conn *conn,
2116                               int argc OVS_UNUSED,
2117                               const char *argv[] OVS_UNUSED,
2118                               void *aux OVS_UNUSED)
2119 {
2120     struct ds ds = DS_EMPTY_INITIALIZER;
2121     struct udpif *udpif;
2122     unsigned int flow_limit = atoi(argv[1]);
2123
2124     LIST_FOR_EACH (udpif, list_node, &all_udpifs) {
2125         atomic_store_relaxed(&udpif->flow_limit, flow_limit);
2126     }
2127     ds_put_format(&ds, "set flow_limit to %u\n", flow_limit);
2128     unixctl_command_reply(conn, ds_cstr(&ds));
2129     ds_destroy(&ds);
2130 }
2131
2132 static void
2133 upcall_unixctl_dump_wait(struct unixctl_conn *conn,
2134                          int argc OVS_UNUSED,
2135                          const char *argv[] OVS_UNUSED,
2136                          void *aux OVS_UNUSED)
2137 {
2138     if (list_is_singleton(&all_udpifs)) {
2139         struct udpif *udpif = NULL;
2140         size_t len;
2141
2142         udpif = OBJECT_CONTAINING(list_front(&all_udpifs), udpif, list_node);
2143         len = (udpif->n_conns + 1) * sizeof *udpif->conns;
2144         udpif->conn_seq = seq_read(udpif->dump_seq);
2145         udpif->conns = xrealloc(udpif->conns, len);
2146         udpif->conns[udpif->n_conns++] = conn;
2147     } else {
2148         unixctl_command_reply_error(conn, "can't wait on multiple udpifs.");
2149     }
2150 }
2151
2152 static void
2153 upcall_unixctl_purge(struct unixctl_conn *conn, int argc OVS_UNUSED,
2154                      const char *argv[] OVS_UNUSED, void *aux OVS_UNUSED)
2155 {
2156     struct udpif *udpif;
2157
2158     LIST_FOR_EACH (udpif, list_node, &all_udpifs) {
2159         int n;
2160
2161         for (n = 0; n < udpif->n_revalidators; n++) {
2162             revalidator_purge(&udpif->revalidators[n]);
2163         }
2164     }
2165     unixctl_command_reply(conn, "");
2166 }