ofproto: Simplify thread creation API.

There's no particular reason for the function controlling the number
of threads to be bound up with dpif_recv_set().  This patch breaks
them up, but as a side effect means threads will run doing nothing
when datapath upcall receiving is disabled.  By doing this, the udpif
thread creation API becomes a bit easier to reason about once there
are multiple types of thread introduced in future patches.

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

7 files changed, 20 insertions, 35 deletions

diff --git a/ofproto/ofproto-dpif-upcall.c b/ofproto/ofproto-dpif-upcall.c
index 20cb9e087..be5656363 100644
--- a/ofproto/ofproto-dpif-upcall.c
+++ b/ofproto/ofproto-dpif-upcall.c
@@ -158,7 +158,7 @@ udpif_destroy(struct udpif *udpif)
     struct flow_miss_batch *fmb;
     struct drop_key *drop_key;
 
-    udpif_recv_set(udpif, 0, false);
+    udpif_set_threads(udpif, 0);
     list_remove(&udpif->list_node);
 
     while ((drop_key = drop_key_next(udpif))) {
@@ -177,15 +177,12 @@ udpif_destroy(struct udpif *udpif)
     free(udpif);
 }
 
-/* Tells 'udpif' to begin or stop handling flow misses depending on the value
- * of 'enable'.  'n_handlers' is the number of upcall_handler threads to
- * create.  Passing 'n_handlers' as zero is equivalent to passing 'enable' as
- * false. */
+/* Tells 'udpif' how many threads it should use to handle upcalls.  Disables
+ * all threads if 'n_handlers' is zero.  'udpif''s datapath handle must have
+ * packet reception enabled before starting threads. */
 void
-udpif_recv_set(struct udpif *udpif, size_t n_handlers, bool enable)
+udpif_set_threads(struct udpif *udpif, size_t n_handlers)
 {
-    n_handlers = enable ? n_handlers : 0;
-
     /* Stop the old threads (if any). */
     if (udpif->handlers && udpif->n_handlers != n_handlers) {
         size_t i;
diff --git a/ofproto/ofproto-dpif-upcall.h b/ofproto/ofproto-dpif-upcall.h
index 805530bff..a4d8228aa 100644
--- a/ofproto/ofproto-dpif-upcall.h
+++ b/ofproto/ofproto-dpif-upcall.h
@@ -33,7 +33,7 @@ struct dpif_backer;
  * module. */
 
 struct udpif *udpif_create(struct dpif_backer *, struct dpif *);
-void udpif_recv_set(struct udpif *, size_t n_workers, bool enable);
+void udpif_set_threads(struct udpif *, size_t n_handlers);
 void udpif_destroy(struct udpif *);
 
 void udpif_wait(struct udpif *);
diff --git a/ofproto/ofproto-dpif.c b/ofproto/ofproto-dpif.c
index d6afc8084..2b0715890 100644
--- a/ofproto/ofproto-dpif.c
+++ b/ofproto/ofproto-dpif.c
@@ -450,9 +450,6 @@ struct dpif_backer {
      * performance in new situations.  */
     unsigned max_n_subfacet;         /* Maximum number of flows */
     unsigned avg_n_subfacet;         /* Average number of flows. */
-
-    /* Number of upcall handling threads. */
-    unsigned int n_handler_threads;
 };
 
 /* All existing ofproto_backer instances, indexed by ofproto->up.type. */
@@ -690,22 +687,15 @@ type_run(const char *type)
 
         error = dpif_recv_set(backer->dpif, backer->recv_set_enable);
         if (error) {
-            udpif_recv_set(backer->udpif, 0, false);
             VLOG_ERR("Failed to enable receiving packets in dpif.");
             return error;
         }
-        udpif_recv_set(backer->udpif, n_handler_threads,
-                       backer->recv_set_enable);
         dpif_flow_flush(backer->dpif);
         backer->need_revalidate = REV_RECONFIGURE;
     }
 
-    /* If the n_handler_threads is reconfigured, call udpif_recv_set()
-     * to reset the handler threads. */
-    if (backer->n_handler_threads != n_handler_threads) {
-        udpif_recv_set(backer->udpif, n_handler_threads,
-                       backer->recv_set_enable);
-        backer->n_handler_threads = n_handler_threads;
+    if (backer->recv_set_enable) {
+        udpif_set_threads(backer->udpif, n_handlers);
     }
 
     if (backer->need_revalidate) {
@@ -1171,9 +1161,10 @@ open_dpif_backer(const char *type, struct dpif_backer **backerp)
         close_dpif_backer(backer);
         return error;
     }
-    udpif_recv_set(backer->udpif, n_handler_threads,
-                   backer->recv_set_enable);
-    backer->n_handler_threads = n_handler_threads;
+
+    if (backer->recv_set_enable) {
+        udpif_set_threads(backer->udpif, n_handlers);
+    }
 
     backer->max_n_subfacet = 0;
     backer->avg_n_subfacet = 0;
diff --git a/ofproto/ofproto-provider.h b/ofproto/ofproto-provider.h
index cc1967f5f..77553f61c 100644
--- a/ofproto/ofproto-provider.h
+++ b/ofproto/ofproto-provider.h
@@ -458,7 +458,7 @@ extern unsigned flow_eviction_threshold;
 
 /* Number of upcall handler threads. Only affects the ofproto-dpif
  * implementation. */
-extern unsigned n_handler_threads;
+extern size_t n_handlers;
 
 /* Determines which model to use for handling misses in the ofproto-dpif
  * implementation */
diff --git a/ofproto/ofproto.c b/ofproto/ofproto.c
index 6d25daba5..e9bbc4d33 100644
--- a/ofproto/ofproto.c
+++ b/ofproto/ofproto.c
@@ -307,9 +307,10 @@ static size_t allocated_ofproto_classes;
 struct ovs_mutex ofproto_mutex = OVS_MUTEX_INITIALIZER;
 
 unsigned flow_eviction_threshold = OFPROTO_FLOW_EVICTION_THRESHOLD_DEFAULT;
-unsigned n_handler_threads;
 enum ofproto_flow_miss_model flow_miss_model = OFPROTO_HANDLE_MISS_AUTO;
 
+size_t n_handlers;
+
 /* Map from datapath name to struct ofproto, for use by unixctl commands. */
 static struct hmap all_ofprotos = HMAP_INITIALIZER(&all_ofprotos);
 
@@ -736,14 +737,10 @@ ofproto_set_mac_table_config(struct ofproto *ofproto, unsigned idle_time,
 /* Sets number of upcall handler threads.  The default is
  * (number of online cores - 2). */
 void
-ofproto_set_n_handler_threads(unsigned limit)
+ofproto_set_threads(size_t n_handlers_)
 {
-    if (limit) {
-        n_handler_threads = limit;
-    } else {
-        int n_proc = count_cpu_cores();
-        n_handler_threads = n_proc > 2 ? n_proc - 2 : 1;
-    }
+    int threads = MAX(count_cpu_cores() - 2, 1);
+    n_handlers = n_handlers_ ? n_handlers_ : threads;
 }
 
 void
diff --git a/ofproto/ofproto.h b/ofproto/ofproto.h
index 219940aba..482212878 100644
--- a/ofproto/ofproto.h
+++ b/ofproto/ofproto.h
@@ -248,7 +248,7 @@ void ofproto_set_flow_miss_model(unsigned model);
 void ofproto_set_forward_bpdu(struct ofproto *, bool forward_bpdu);
 void ofproto_set_mac_table_config(struct ofproto *, unsigned idle_time,
                                   size_t max_entries);
-void ofproto_set_n_handler_threads(unsigned limit);
+void ofproto_set_threads(size_t n_handlers);
 void ofproto_set_dp_desc(struct ofproto *, const char *dp_desc);
 int ofproto_set_snoops(struct ofproto *, const struct sset *snoops);
 int ofproto_set_netflow(struct ofproto *,
diff --git a/vswitchd/bridge.c b/vswitchd/bridge.c
index 3e1613442..1d60326fe 100644
--- a/vswitchd/bridge.c
+++ b/vswitchd/bridge.c
@@ -496,7 +496,7 @@ bridge_reconfigure(const struct ovsrec_open_vswitch *ovs_cfg)
         smap_get_int(&ovs_cfg->other_config, "flow-eviction-threshold",
                      OFPROTO_FLOW_EVICTION_THRESHOLD_DEFAULT));
 
-    ofproto_set_n_handler_threads(
+    ofproto_set_threads(
         smap_get_int(&ovs_cfg->other_config, "n-handler-threads", 0));
 
     bridge_configure_flow_miss_model(smap_get(&ovs_cfg->other_config,