ofproto-dpif: Handle learn action flow mods asynchronously.

Once we have multiple threads running, having each execute flow mods
created by the learn action won't be tenable.  It essentially will
require us to make the core ofproto module thread safe, which is not
the direction we want to go.  This patch punts on the problem by
handing flow mods to ofproto-dpif to handle later.

Signed-off-by: Ethan Jackson <ethan@nicira.com>
Acked-by: Ben Pfaff <blp@nicira.com>

diff --git a/lib/ofp-util.h b/lib/ofp-util.h
index f94982d..21311f7 100644 (file)
--- a/lib/ofp-util.h
+++ b/lib/ofp-util.h
@@ -212,6 +212,8 @@ struct ofpbuf *ofputil_make_flow_mod_table_id(bool flow_mod_table_id);
  * The handling of cookies across multiple versions of OpenFlow is a bit
  * confusing.  See DESIGN for the details. */
 struct ofputil_flow_mod {
+    struct list list_node;    /* For queuing flow_mods. */
+
     struct match match;
     unsigned int priority;
 

diff --git a/ofproto/ofproto-dpif-xlate.c b/ofproto/ofproto-dpif-xlate.c
index 6ce30cb..bba4355 100644 (file)
--- a/ofproto/ofproto-dpif-xlate.c
+++ b/ofproto/ofproto-dpif-xlate.c
@@ -1897,11 +1897,8 @@ static void
 xlate_learn_action(struct xlate_ctx *ctx,
                    const struct ofpact_learn *learn)
 {
-    static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(5, 1);
-    struct ofputil_flow_mod fm;
-    uint64_t ofpacts_stub[1024 / 8];
+    struct ofputil_flow_mod *fm;
     struct ofpbuf ofpacts;
-    int error;
 
     ctx->xout->has_learn = true;
 
@@ -1911,16 +1908,11 @@ xlate_learn_action(struct xlate_ctx *ctx,
         return;
     }
 
-    ofpbuf_use_stack(&ofpacts, ofpacts_stub, sizeof ofpacts_stub);
-    learn_execute(learn, &ctx->xin->flow, &fm, &ofpacts);
-
-    error = ofproto_dpif_flow_mod(ctx->xbridge->ofproto, &fm);
-    if (error && !VLOG_DROP_WARN(&rl)) {
-        VLOG_WARN("learning action failed to modify flow table (%s)",
-                  ofperr_get_name(error));
-    }
+    fm = xmalloc(sizeof *fm);
+    ofpbuf_init(&ofpacts, 0);
+    learn_execute(learn, &ctx->xin->flow, fm, &ofpacts);
 
-    ofpbuf_uninit(&ofpacts);
+    ofproto_dpif_flow_mod(ctx->xbridge->ofproto, fm);
 }
 
 /* Reduces '*timeout' to no more than 'max'.  A value of zero in either case

diff --git a/ofproto/ofproto-dpif.c b/ofproto/ofproto-dpif.c
index ce783f0..3bf4fe3 100644 (file)
--- a/ofproto/ofproto-dpif.c
+++ b/ofproto/ofproto-dpif.c
@@ -71,6 +71,7 @@ COVERAGE_DEFINE(facet_revalidate);
 COVERAGE_DEFINE(facet_unexpected);
 COVERAGE_DEFINE(facet_suppress);
 COVERAGE_DEFINE(subfacet_install_fail);
+COVERAGE_DEFINE(flow_mod_overflow);
 
 /* Number of implemented OpenFlow tables. */
 enum { N_TABLES = 255 };
@@ -349,6 +350,7 @@ static int set_bfd(struct ofport *, const struct smap *);
 static int set_cfm(struct ofport *, const struct cfm_settings *);
 static void ofport_update_peer(struct ofport_dpif *);
 static void run_fast_rl(void);
+static int run_fast(struct ofproto *);
 
 struct dpif_completion {
     struct list list_node;
@@ -494,6 +496,11 @@ struct ofproto_dpif {
     /* Per ofproto's dpif stats. */
     uint64_t n_hit;
     uint64_t n_missed;
+
+    /* Work queues. */
+    struct ovs_mutex flow_mod_mutex;
+    struct list flow_mods OVS_GUARDED;
+    size_t n_flow_mods OVS_GUARDED;
 };
 
 /* Defer flow mod completion until "ovs-appctl ofproto/unclog"?  (Useful only
@@ -538,11 +545,23 @@ static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(1, 5);
 /* Initial mappings of port to bridge mappings. */
 static struct shash init_ofp_ports = SHASH_INITIALIZER(&init_ofp_ports);
 
-int
+/* Executes and takes ownership of 'fm'. */
+void
 ofproto_dpif_flow_mod(struct ofproto_dpif *ofproto,
                       struct ofputil_flow_mod *fm)
 {
-    return ofproto_flow_mod(&ofproto->up, fm);
+    ovs_mutex_lock(&ofproto->flow_mod_mutex);
+    if (ofproto->n_flow_mods > 1024) {
+        ovs_mutex_unlock(&ofproto->flow_mod_mutex);
+        COVERAGE_INC(flow_mod_overflow);
+        free(fm->ofpacts);
+        free(fm);
+        return;
+    }
+
+    list_push_back(&ofproto->flow_mods, &fm->list_node);
+    ofproto->n_flow_mods++;
+    ovs_mutex_unlock(&ofproto->flow_mod_mutex);
 }
 
 void
@@ -1010,13 +1029,9 @@ run_fast_rl(void)
 
     if (time_msec() >= port_rl) {
         struct ofproto_dpif *ofproto;
-        struct ofport_dpif *ofport;
 
         HMAP_FOR_EACH (ofproto, all_ofproto_dpifs_node, &all_ofproto_dpifs) {
-
-            HMAP_FOR_EACH (ofport, up.hmap_node, &ofproto->up.ports) {
-                port_run_fast(ofport);
-            }
+            run_fast(&ofproto->up);
         }
         port_rl = time_msec() + 200;
     }
@@ -1258,6 +1273,12 @@ construct(struct ofproto *ofproto_)
 
     list_init(&ofproto->completions);
 
+    ovs_mutex_init(&ofproto->flow_mod_mutex, PTHREAD_MUTEX_NORMAL);
+    ovs_mutex_lock(&ofproto->flow_mod_mutex);
+    list_init(&ofproto->flow_mods);
+    ofproto->n_flow_mods = 0;
+    ovs_mutex_unlock(&ofproto->flow_mod_mutex);
+
     ofproto_dpif_unixctl_init();
 
     hmap_init(&ofproto->vlandev_map);
@@ -1388,6 +1409,7 @@ destruct(struct ofproto *ofproto_)
 {
     struct ofproto_dpif *ofproto = ofproto_dpif_cast(ofproto_);
     struct rule_dpif *rule, *next_rule;
+    struct ofputil_flow_mod *fm, *next_fm;
     struct oftable *table;
 
     ofproto->backer->need_revalidate = REV_RECONFIGURE;
@@ -1405,6 +1427,16 @@ destruct(struct ofproto *ofproto_)
         }
     }
 
+    ovs_mutex_lock(&ofproto->flow_mod_mutex);
+    LIST_FOR_EACH_SAFE (fm, next_fm, list_node, &ofproto->flow_mods) {
+        list_remove(&fm->list_node);
+        ofproto->n_flow_mods--;
+        free(fm->ofpacts);
+        free(fm);
+    }
+    ovs_mutex_unlock(&ofproto->flow_mod_mutex);
+    ovs_mutex_destroy(&ofproto->flow_mod_mutex);
+
     mbridge_unref(ofproto->mbridge);
 
     netflow_destroy(ofproto->netflow);
@@ -1428,7 +1460,9 @@ static int
 run_fast(struct ofproto *ofproto_)
 {
     struct ofproto_dpif *ofproto = ofproto_dpif_cast(ofproto_);
+    struct ofputil_flow_mod *fm, *next;
     struct ofport_dpif *ofport;
+    struct list flow_mods;
 
     /* Do not perform any periodic activity required by 'ofproto' while
      * waiting for flow restore to complete. */
@@ -1436,6 +1470,29 @@ run_fast(struct ofproto *ofproto_)
         return 0;
     }
 
+    ovs_mutex_lock(&ofproto->flow_mod_mutex);
+    if (ofproto->n_flow_mods) {
+        flow_mods = ofproto->flow_mods;
+        list_moved(&flow_mods);
+        list_init(&ofproto->flow_mods);
+        ofproto->n_flow_mods = 0;
+    } else {
+        list_init(&flow_mods);
+    }
+    ovs_mutex_unlock(&ofproto->flow_mod_mutex);
+
+    LIST_FOR_EACH_SAFE (fm, next, list_node, &flow_mods) {
+        int error = ofproto_flow_mod(&ofproto->up, fm);
+        if (error && !VLOG_DROP_WARN(&rl)) {
+            VLOG_WARN("learning action failed to modify flow table (%s)",
+                      ofperr_get_name(error));
+        }
+
+        list_remove(&fm->list_node);
+        free(fm->ofpacts);
+        free(fm);
+    }
+
     HMAP_FOR_EACH (ofport, up.hmap_node, &ofproto->up.ports) {
         port_run_fast(ofport);
     }

diff --git a/ofproto/ofproto-dpif.h b/ofproto/ofproto-dpif.h
index f67b88d..b356c06 100644 (file)
--- a/ofproto/ofproto-dpif.h
+++ b/ofproto/ofproto-dpif.h
@@ -79,6 +79,6 @@ int ofproto_dpif_queue_to_priority(const struct ofproto_dpif *,
 
 void ofproto_dpif_send_packet_in(struct ofproto_dpif *,
                                  struct ofputil_packet_in *pin);
-int ofproto_dpif_flow_mod(struct ofproto_dpif *, struct ofputil_flow_mod *);
+void ofproto_dpif_flow_mod(struct ofproto_dpif *, struct ofputil_flow_mod *);
 
 #endif /* ofproto-dpif.h */