dpif: Make dpif_flow_dump_next() thread-safe.

This patch makes it the caller's responsibility to initialize a
per-thread 'state' object and pass it down to the dpif_flow_dump_next()
implementation. The implementation can expect to be called from multiple
threads with the same 'iter' and different 'state' objects.

When flow_dump_next() returns non-zero, the implementation must ensure
that subsequent calls with the same arguments also return non-zero.
Subsequent calls with the same 'iter' and different 'state' may return
zero, but should make progress towards returning non-zero.

Signed-off-by: Joe Stringer <[email protected]>
Signed-off-by: Ben Pfaff <[email protected]>

lib/dpif-linux.c

@@ -1000,7 +1000,7 @@ struct dpif_linux_flow_state {
 
 struct dpif_linux_flow_iter {
     struct nl_dump dump;
-    void *state;
+    atomic_int status;
 };
 
 static void
@@ -1041,21 +1041,20 @@ dpif_linux_flow_dump_start(const struct dpif *dpif_, void **iterp)
     dpif_linux_flow_to_ofpbuf(&request, buf);
     nl_dump_start(&iter->dump, NETLINK_GENERIC, buf);
     ofpbuf_delete(buf);
-
-    dpif_linux_flow_dump_state_init(&iter->state);
+    atomic_init(&iter->status, 0);
 
     return 0;
 }
 
 static int
-dpif_linux_flow_dump_next(const struct dpif *dpif_, void *iter_,
+dpif_linux_flow_dump_next(const struct dpif *dpif_, void *iter_, void *state_,
                           const struct nlattr **key, size_t *key_len,
                           const struct nlattr **mask, size_t *mask_len,
                           const struct nlattr **actions, size_t *actions_len,
                           const struct dpif_flow_stats **stats)
 {
     struct dpif_linux_flow_iter *iter = iter_;
-    struct dpif_linux_flow_state *state = iter->state;
+    struct dpif_linux_flow_state *state = state_;
     struct ofpbuf buf;
     int error;
 
@@ -1069,6 +1068,7 @@ dpif_linux_flow_dump_next(const struct dpif *dpif_, void *iter_,
 
         error = dpif_linux_flow_from_ofpbuf(&state->flow, &buf);
         if (error) {
+            atomic_store(&iter->status, error);
             return error;
         }
 
@@ -1108,10 +1108,13 @@ static int
 dpif_linux_flow_dump_done(const struct dpif *dpif OVS_UNUSED, void *iter_)
 {
     struct dpif_linux_flow_iter *iter = iter_;
-    int error = nl_dump_done(&iter->dump);
-    dpif_linux_flow_dump_state_uninit(iter->state);
+    int dump_status;
+    unsigned int nl_status = nl_dump_done(&iter->dump);
+
+    atomic_read(&iter->status, &dump_status);
+    atomic_destroy(&iter->status);
     free(iter);
-    return error;
+    return dump_status ? dump_status : nl_status;
 }
 
 static void

lib/dpif-netdev.c

@@ -1315,7 +1315,8 @@ struct dp_netdev_flow_state {
 struct dp_netdev_flow_iter {
     uint32_t bucket;
     uint32_t offset;
-    void *state;
+    int status;
+    struct ovs_mutex mutex;
 };
 
 static void
@@ -1344,32 +1345,43 @@ dpif_netdev_flow_dump_start(const struct dpif *dpif OVS_UNUSED, void **iterp)
     *iterp = iter = xmalloc(sizeof *iter);
     iter->bucket = 0;
     iter->offset = 0;
-    dpif_netdev_flow_dump_state_init(&iter->state);
+    iter->status = 0;
+    ovs_mutex_init(&iter->mutex);
     return 0;
 }
 
 static int
-dpif_netdev_flow_dump_next(const struct dpif *dpif, void *iter_,
+dpif_netdev_flow_dump_next(const struct dpif *dpif, void *iter_, void *state_,
                            const struct nlattr **key, size_t *key_len,
                            const struct nlattr **mask, size_t *mask_len,
                            const struct nlattr **actions, size_t *actions_len,
                            const struct dpif_flow_stats **stats)
 {
     struct dp_netdev_flow_iter *iter = iter_;
-    struct dp_netdev_flow_state *state = iter->state;
+    struct dp_netdev_flow_state *state = state_;
     struct dp_netdev *dp = get_dp_netdev(dpif);
     struct dp_netdev_flow *netdev_flow;
-    struct hmap_node *node;
+    int error;
 
-    fat_rwlock_rdlock(&dp->cls.rwlock);
-    node = hmap_at_position(&dp->flow_table, &iter->bucket, &iter->offset);
-    if (node) {
-        netdev_flow = CONTAINER_OF(node, struct dp_netdev_flow, node);
-        dp_netdev_flow_ref(netdev_flow);
+    ovs_mutex_lock(&iter->mutex);
+    error = iter->status;
+    if (!error) {
+        struct hmap_node *node;
+
+        fat_rwlock_rdlock(&dp->cls.rwlock);
+        node = hmap_at_position(&dp->flow_table, &iter->bucket, &iter->offset);
+        if (node) {
+            netdev_flow = CONTAINER_OF(node, struct dp_netdev_flow, node);
+            dp_netdev_flow_ref(netdev_flow);
+        }
+        fat_rwlock_unlock(&dp->cls.rwlock);
+        if (!node) {
+            iter->status = error = EOF;
+        }
     }
-    fat_rwlock_unlock(&dp->cls.rwlock);
-    if (!node) {
-        return EOF;
+    ovs_mutex_unlock(&iter->mutex);
+    if (error) {
+        return error;
     }
 
     if (key) {
@@ -1424,7 +1436,7 @@ dpif_netdev_flow_dump_done(const struct dpif *dpif OVS_UNUSED, void *iter_)
 {
     struct dp_netdev_flow_iter *iter = iter_;
 
-    dpif_netdev_flow_dump_state_uninit(iter->state);
+    ovs_mutex_destroy(&iter->mutex);
     free(iter);
     return 0;
 }

lib/dpif-provider.h

@@ -272,13 +272,18 @@ struct dpif_class {
      * On failure, returns a positive errno value. */
     int (*flow_dump_start)(const struct dpif *dpif, void **iterp);
 
-    /* Attempts to retrieve another flow from 'dpif' for 'iter', which was
-     * initialized by a successful call to the 'flow_dump_start' function for
-     * 'dpif'.  On success, updates the output parameters as described below
-     * and returns 0.  Returns EOF if the end of the flow table has been
-     * reached, or a positive errno value on error.  This function will not be
-     * called again once it returns nonzero within a given iteration (but the
-     * 'flow_dump_done' function will be called afterward).
+    /* Attempts to retrieve another flow from 'dpif' for 'iter', using
+     * 'state' for storage. 'iter' must have been initialized by a successful
+     * call to the 'flow_dump_start' function for 'dpif'. 'state' must have
+     * been initialised with a call to the 'flow_dump_state_init' function for
+     * 'dpif.
+     *
+     * On success, updates the output parameters as described below and returns
+     * 0. Returns EOF if the end of the flow table has been reached, or a
+     * positive errno value on error. Multiple threads may use the same 'dpif'
+     * and 'iter' with this function, but all other parameters must be
+     * different for each thread. If this function returns non-zero,
+     * subsequent calls with the same arguments will also return non-zero.
      *
      * On success:
      *
@@ -300,15 +305,17 @@ struct dpif_class {
      * All of the returned data is owned by 'dpif', not by the caller, and the
      * caller must not modify or free it.  'dpif' must guarantee that it
      * remains accessible and unchanging until at least the next call to
-     * 'flow_dump_next' or 'flow_dump_done' for 'iter'. */
-    int (*flow_dump_next)(const struct dpif *dpif, void *iter,
+     * 'flow_dump_next' or 'flow_dump_done' for 'iter' and 'state'. */
+    int (*flow_dump_next)(const struct dpif *dpif, void *iter, void *state,
                           const struct nlattr **key, size_t *key_len,
                           const struct nlattr **mask, size_t *mask_len,
                           const struct nlattr **actions, size_t *actions_len,
                           const struct dpif_flow_stats **stats);
 
     /* Releases resources from 'dpif' for 'iter', which was initialized by a
-     * successful call to the 'flow_dump_start' function for 'dpif'.  */
+     * successful call to the 'flow_dump_start' function for 'dpif'. Callers
+     * must ensure that this function is called once within a given iteration,
+     * as the final flow dump operation. */
     int (*flow_dump_done)(const struct dpif *dpif, void *iter);
 
     /* Releases 'state' which was initialized by a call to the

lib/dpif.c

@@ -992,12 +992,17 @@ dpif_flow_dump_start(struct dpif_flow_dump *dump, const struct dpif *dpif)
     log_operation(dpif, "flow_dump_start", dump->error);
 }
 
-/* Attempts to retrieve another flow from 'dump', which must have been
- * initialized with dpif_flow_dump_start().  On success, updates the output
- * parameters as described below and returns true.  Otherwise, returns false.
- * Failure might indicate an actual error or merely the end of the flow table.
- * An error status for the entire dump operation is provided when it is
- * completed by calling dpif_flow_dump_done().
+/* Attempts to retrieve another flow from 'dump', using 'state' for
+ * thread-local storage. 'dump' must have been initialized with
+ * dpif_flow_dump_start(), and 'state' must have been initialized with
+ * dpif_flow_state_init().
+ *
+ * On success, updates the output parameters as described below and returns
+ * true. Otherwise, returns false. Failure might indicate an actual error or
+ * merely the end of the flow table. An error status for the entire dump
+ * operation is provided when it is completed by calling dpif_flow_dump_done().
+ * Multiple threads may use the same 'dump' with this function, but all other
+ * parameters must not be shared.
  *
  * On success, if 'key' and 'key_len' are nonnull then '*key' and '*key_len'
  * will be set to Netlink attributes with types OVS_KEY_ATTR_* representing the
@@ -1009,9 +1014,9 @@ dpif_flow_dump_start(struct dpif_flow_dump *dump, const struct dpif *dpif)
  * All of the returned data is owned by 'dpif', not by the caller, and the
  * caller must not modify or free it.  'dpif' guarantees that it remains
  * accessible and unchanging until at least the next call to 'flow_dump_next'
- * or 'flow_dump_done' for 'dump'. */
+ * or 'flow_dump_done' for 'dump' and 'state'. */
 bool
-dpif_flow_dump_next(struct dpif_flow_dump *dump,
+dpif_flow_dump_next(struct dpif_flow_dump *dump, void *state,
                     const struct nlattr **key, size_t *key_len,
                     const struct nlattr **mask, size_t *mask_len,
                     const struct nlattr **actions, size_t *actions_len,
@@ -1021,7 +1026,7 @@ dpif_flow_dump_next(struct dpif_flow_dump *dump,
     int error = dump->error;
 
     if (!error) {
-        error = dpif->dpif_class->flow_dump_next(dpif, dump->iter,
+        error = dpif->dpif_class->flow_dump_next(dpif, dump->iter, state,
                                                  key, key_len,
                                                  mask, mask_len,
                                                  actions, actions_len,

lib/dpif.h

@@ -356,11 +356,19 @@
  *      thread-safe: they may be called from different threads only on
  *      different dpif objects.
  *
- *    - Functions that operate on struct dpif_port_dump or struct
- *      dpif_flow_dump are conditionally thread-safe with respect to those
- *      objects.  That is, one may dump ports or flows from any number of
- *      threads at once, but each thread must use its own struct dpif_port_dump
- *      or dpif_flow_dump.
+ *    - dpif_flow_dump_next() is conditionally thread-safe: It may be called
+ *      from different threads with the same 'struct dpif_flow_dump', but all
+ *      other parameters must be different for each thread.
+ *
+ *    - dpif_flow_dump_done() is conditionally thread-safe: All threads that
+ *      share the same 'struct dpif_flow_dump' must have finished using it.
+ *      This function must then be called exactly once for a particular
+ *      dpif_flow_dump to finish the corresponding flow dump operation.
+ *
+ *    - Functions that operate on 'struct dpif_port_dump' are conditionally
+ *      thread-safe with respect to those objects.  That is, one may dump ports
+ *      from any number of threads at once, but each thread must use its own
+ *      struct dpif_port_dump.
  */
 #ifndef DPIF_H
 #define DPIF_H 1
@@ -511,7 +519,7 @@ struct dpif_flow_dump {
 };
 void dpif_flow_dump_state_init(const struct dpif *, void **statep);
 void dpif_flow_dump_start(struct dpif_flow_dump *, const struct dpif *);
-bool dpif_flow_dump_next(struct dpif_flow_dump *,
+bool dpif_flow_dump_next(struct dpif_flow_dump *, void *state,
                          const struct nlattr **key, size_t *key_len,
                          const struct nlattr **mask, size_t *mask_len,
                          const struct nlattr **actions, size_t *actions_len,

ofproto/ofproto-dpif-upcall.c

@@ -552,6 +552,7 @@ udpif_flow_dumper(void *arg)
         bool need_revalidate;
         uint64_t reval_seq;
         size_t n_flows, i;
+        void *state = NULL;
 
         reval_seq = seq_read(udpif->reval_seq);
         need_revalidate = udpif->last_reval_seq != reval_seq;
@@ -563,8 +564,9 @@ udpif_flow_dumper(void *arg)
 
         start_time = time_msec();
         dpif_flow_dump_start(&dump, udpif->dpif);
-        while (dpif_flow_dump_next(&dump, &key, &key_len, &mask, &mask_len,
-                                   NULL, NULL, &stats)
+        dpif_flow_dump_state_init(udpif->dpif, &state);
+        while (dpif_flow_dump_next(&dump, state, &key, &key_len,
+                                   &mask, &mask_len, NULL, NULL, &stats)
                && !latch_is_set(&udpif->exit_latch)) {
             struct udpif_flow_dump *udump = xmalloc(sizeof *udump);
             struct revalidator *revalidator;
@@ -595,6 +597,7 @@ udpif_flow_dumper(void *arg)
             xpthread_cond_signal(&revalidator->wake_cond);
             ovs_mutex_unlock(&revalidator->mutex);
         }
+        dpif_flow_dump_state_uninit(udpif->dpif, state);
         dpif_flow_dump_done(&dump);
 
         /* Let all the revalidators finish and garbage collect. */

ofproto/ofproto-dpif.c

@@ -4165,6 +4165,7 @@ ofproto_unixctl_dpif_dump_flows(struct unixctl_conn *conn,
     struct dpif_port dpif_port;
     struct dpif_port_dump port_dump;
     struct hmap portno_names;
+    void *state = NULL;
 
     ofproto = ofproto_dpif_lookup(argv[argc - 1]);
     if (!ofproto) {
@@ -4183,8 +4184,10 @@ ofproto_unixctl_dpif_dump_flows(struct unixctl_conn *conn,
 
     ds_init(&ds);
     dpif_flow_dump_start(&flow_dump, ofproto->backer->dpif);
-    while (dpif_flow_dump_next(&flow_dump, &key, &key_len, &mask, &mask_len,
-                               &actions, &actions_len, &stats)) {
+    dpif_flow_dump_state_init(ofproto->backer->dpif, &state);
+    while (dpif_flow_dump_next(&flow_dump, state, &key, &key_len,
+                               &mask, &mask_len, &actions, &actions_len,
+                               &stats)) {
         if (!ofproto_dpif_contains_flow(ofproto, key, key_len)) {
             continue;
         }
@@ -4197,6 +4200,7 @@ ofproto_unixctl_dpif_dump_flows(struct unixctl_conn *conn,
         format_odp_actions(&ds, actions, actions_len);
         ds_put_char(&ds, '\n');
     }
+    dpif_flow_dump_state_uninit(ofproto->backer->dpif, state);
 
     if (dpif_flow_dump_done(&flow_dump)) {
         ds_clear(&ds);

utilities/ovs-dpctl.c

@@ -764,6 +764,7 @@ dpctl_dump_flows(int argc, char *argv[])
     char *name, *error, *filter = NULL;
     struct flow flow_filter;
     struct flow_wildcards wc_filter;
+    void *state = NULL;
 
     if (argc > 1 && !strncmp(argv[argc - 1], "filter=", 7)) {
         filter = xstrdup(argv[--argc] + 7);
@@ -791,9 +792,10 @@ dpctl_dump_flows(int argc, char *argv[])
 
     ds_init(&ds);
     dpif_flow_dump_start(&flow_dump, dpif);
-    while (dpif_flow_dump_next(&flow_dump, &key, &key_len,
-                               &mask, &mask_len,
-                               &actions, &actions_len, &stats)) {
+    dpif_flow_dump_state_init(dpif, &state);
+    while (dpif_flow_dump_next(&flow_dump, state, &key, &key_len,
+                               &mask, &mask_len, &actions, &actions_len,
+                               &stats)) {
         if (filter) {
             struct flow flow;
             struct flow_wildcards wc;
@@ -824,6 +826,7 @@ dpctl_dump_flows(int argc, char *argv[])
         format_odp_actions(&ds, actions, actions_len);
         printf("%s\n", ds_cstr(&ds));
     }
+    dpif_flow_dump_state_uninit(dpif, state);
     dpif_flow_dump_done(&flow_dump);
 
     free(filter);