ofproto/ofproto-dpif-upcall.c

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