ofproto/ofproto-dpif-upcall.c

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