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