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