dpctl: Avoid making assumptions on pmd threads.

Currently dpctl depends on ovs-numa module to delete and create flows on
different pmd threads for pmd devices.

The next commits will move away the pmd threads state from ovs-numa to
dpif-netdev, so the ovs-numa interface will not be supported.

Also, the assignment between ports and thread is an implementation
detail of dpif-netdev, dpctl shouldn't know anything about it.

This commit changes the dpif_flow_put() and dpif_flow_del() calls to
iterate over all the pmd threads, if pmd_id is PMD_ID_NULL.

A simple test is added.

Signed-off-by: Daniele Di Proietto <diproiettod@vmware.com>
Acked-by: Ilya Maximets <i.maximets@samsung.com>

4 files changed, 150 insertions, 155 deletions

diff --git a/lib/dpctl.c b/lib/dpctl.c
index 7011b183d..23837ce74 100644
--- a/lib/dpctl.c
+++ b/lib/dpctl.c
@@ -40,7 +40,6 @@
 #include "netlink.h"
 #include "odp-util.h"
 #include "openvswitch/ofpbuf.h"
-#include "ovs-numa.h"
 #include "packets.h"
 #include "openvswitch/shash.h"
 #include "simap.h"
@@ -876,43 +875,12 @@ out_freefilter:
     return error;
 }
 
-/* Extracts the in_port from the parsed keys, and returns the reference
- * to the 'struct netdev *' of the dpif port.  On error, returns NULL.
- * Users must call 'netdev_close()' after finish using the returned
- * reference. */
-static struct netdev *
-get_in_port_netdev_from_key(struct dpif *dpif, const struct ofpbuf *key)
-{
-    const struct nlattr *in_port_nla;
-    struct netdev *dev = NULL;
-
-    in_port_nla = nl_attr_find(key, 0, OVS_KEY_ATTR_IN_PORT);
-    if (in_port_nla) {
-        struct dpif_port dpif_port;
-        odp_port_t port_no;
-        int error;
-
-        port_no = ODP_PORT_C(nl_attr_get_u32(in_port_nla));
-        error = dpif_port_query_by_number(dpif, port_no, &dpif_port);
-        if (error) {
-            goto out;
-        }
-
-        netdev_open(dpif_port.name, dpif_port.type, &dev);
-        dpif_port_destroy(&dpif_port);
-    }
-
-out:
-    return dev;
-}
-
 static int
 dpctl_put_flow(int argc, const char *argv[], enum dpif_flow_put_flags flags,
                struct dpctl_params *dpctl_p)
 {
     const char *key_s = argv[argc - 2];
     const char *actions_s = argv[argc - 1];
-    struct netdev *in_port_netdev = NULL;
     struct dpif_flow_stats stats;
     struct dpif_port dpif_port;
     struct dpif_port_dump port_dump;
@@ -968,39 +936,15 @@ dpctl_put_flow(int argc, const char *argv[], enum dpif_flow_put_flags flags,
         goto out_freeactions;
     }
 
-    /* For DPDK interface, applies the operation to all pmd threads
-     * on the same numa node. */
-    in_port_netdev = get_in_port_netdev_from_key(dpif, &key);
-    if (in_port_netdev && netdev_is_pmd(in_port_netdev)) {
-        int numa_id;
-
-        numa_id = netdev_get_numa_id(in_port_netdev);
-        if (ovs_numa_numa_id_is_valid(numa_id)) {
-            struct ovs_numa_dump *dump = ovs_numa_dump_cores_on_numa(numa_id);
-            struct ovs_numa_info *iter;
-
-            FOR_EACH_CORE_ON_NUMA (iter, dump) {
-                if (ovs_numa_core_is_pinned(iter->core_id)) {
-                    error = dpif_flow_put(dpif, flags,
-                                          key.data, key.size,
-                                          mask.size == 0 ? NULL : mask.data,
-                                          mask.size, actions.data,
-                                          actions.size, ufid_present ? &ufid : NULL,
-                                          iter->core_id, dpctl_p->print_statistics ? &stats : NULL);
-                }
-            }
-            ovs_numa_dump_destroy(dump);
-        } else {
-            error = EINVAL;
-        }
-    } else {
-        error = dpif_flow_put(dpif, flags,
-                              key.data, key.size,
-                              mask.size == 0 ? NULL : mask.data,
-                              mask.size, actions.data,
-                              actions.size, ufid_present ? &ufid : NULL,
-                              PMD_ID_NULL, dpctl_p->print_statistics ? &stats : NULL);
-    }
+    /* The flow will be added on all pmds currently in the datapath. */
+    error = dpif_flow_put(dpif, flags,
+                          key.data, key.size,
+                          mask.size == 0 ? NULL : mask.data,
+                          mask.size, actions.data,
+                          actions.size, ufid_present ? &ufid : NULL,
+                          PMD_ID_NULL,
+                          dpctl_p->print_statistics ? &stats : NULL);
+
     if (error) {
         dpctl_error(dpctl_p, error, "updating flow table");
         goto out_freeactions;
@@ -1021,7 +965,6 @@ out_freekeymask:
     ofpbuf_uninit(&mask);
     ofpbuf_uninit(&key);
     dpif_close(dpif);
-    netdev_close(in_port_netdev);
     return error;
 }
 
@@ -1110,7 +1053,6 @@ static int
 dpctl_del_flow(int argc, const char *argv[], struct dpctl_params *dpctl_p)
 {
     const char *key_s = argv[argc - 1];
-    struct netdev *in_port_netdev = NULL;
     struct dpif_flow_stats stats;
     struct dpif_port dpif_port;
     struct dpif_port_dump port_dump;
@@ -1158,33 +1100,11 @@ dpctl_del_flow(int argc, const char *argv[], struct dpctl_params *dpctl_p)
         goto out;
     }
 
-    /* For DPDK interface, applies the operation to all pmd threads
-     * on the same numa node. */
-    in_port_netdev = get_in_port_netdev_from_key(dpif, &key);
-    if (in_port_netdev && netdev_is_pmd(in_port_netdev)) {
-        int numa_id;
-
-        numa_id = netdev_get_numa_id(in_port_netdev);
-        if (ovs_numa_numa_id_is_valid(numa_id)) {
-            struct ovs_numa_dump *dump = ovs_numa_dump_cores_on_numa(numa_id);
-            struct ovs_numa_info *iter;
+    /* The flow will be deleted from all pmds currently in the datapath. */
+    error = dpif_flow_del(dpif, key.data, key.size,
+                          ufid_present ? &ufid : NULL, PMD_ID_NULL,
+                          dpctl_p->print_statistics ? &stats : NULL);
 
-            FOR_EACH_CORE_ON_NUMA (iter, dump) {
-                if (ovs_numa_core_is_pinned(iter->core_id)) {
-                    error = dpif_flow_del(dpif, key.data,
-                                          key.size, ufid_present ? &ufid : NULL,
-                                          iter->core_id, dpctl_p->print_statistics ? &stats : NULL);
-                }
-            }
-            ovs_numa_dump_destroy(dump);
-        } else {
-            error = EINVAL;
-        }
-    } else {
-        error = dpif_flow_del(dpif, key.data, key.size,
-                              ufid_present ? &ufid : NULL, PMD_ID_NULL,
-                              dpctl_p->print_statistics ? &stats : NULL);
-    }
     if (error) {
         dpctl_error(dpctl_p, error, "deleting flow");
         if (error == ENOENT && !ufid_present) {
@@ -1212,7 +1132,6 @@ out:
     ofpbuf_uninit(&key);
     simap_destroy(&port_names);
     dpif_close(dpif);
-    netdev_close(in_port_netdev);
     return error;
 }
 
diff --git a/lib/dpif-netdev.c b/lib/dpif-netdev.c
index 22f7b1bc5..a9857ad2b 100644
--- a/lib/dpif-netdev.c
+++ b/lib/dpif-netdev.c
@@ -2308,54 +2308,26 @@ dp_netdev_flow_add(struct dp_netdev_pmd_thread *pmd,
 }
 
 static int
-dpif_netdev_flow_put(struct dpif *dpif, const struct dpif_flow_put *put)
+flow_put_on_pmd(struct dp_netdev_pmd_thread *pmd,
+                struct netdev_flow_key *key,
+                struct match *match,
+                ovs_u128 *ufid,
+                const struct dpif_flow_put *put,
+                struct dpif_flow_stats *stats)
 {
-    struct dp_netdev *dp = get_dp_netdev(dpif);
     struct dp_netdev_flow *netdev_flow;
-    struct netdev_flow_key key;
-    struct dp_netdev_pmd_thread *pmd;
-    struct match match;
-    ovs_u128 ufid;
-    unsigned pmd_id = put->pmd_id == PMD_ID_NULL
-                      ? NON_PMD_CORE_ID : put->pmd_id;
-    int error;
-
-    error = dpif_netdev_flow_from_nlattrs(put->key, put->key_len, &match.flow);
-    if (error) {
-        return error;
-    }
-    error = dpif_netdev_mask_from_nlattrs(put->key, put->key_len,
-                                          put->mask, put->mask_len,
-                                          &match.flow, &match.wc);
-    if (error) {
-        return error;
-    }
-
-    pmd = dp_netdev_get_pmd(dp, pmd_id);
-    if (!pmd) {
-        return EINVAL;
-    }
-
-    /* Must produce a netdev_flow_key for lookup.
-     * This interface is no longer performance critical, since it is not used
-     * for upcall processing any more. */
-    netdev_flow_key_from_flow(&key, &match.flow);
+    int error = 0;
 
-    if (put->ufid) {
-        ufid = *put->ufid;
-    } else {
-        dpif_flow_hash(dpif, &match.flow, sizeof match.flow, &ufid);
+    if (stats) {
+        memset(stats, 0, sizeof *stats);
     }
 
     ovs_mutex_lock(&pmd->flow_mutex);
-    netdev_flow = dp_netdev_pmd_lookup_flow(pmd, &key, NULL);
+    netdev_flow = dp_netdev_pmd_lookup_flow(pmd, key, NULL);
     if (!netdev_flow) {
         if (put->flags & DPIF_FP_CREATE) {
             if (cmap_count(&pmd->flow_table) < MAX_FLOWS) {
-                if (put->stats) {
-                    memset(put->stats, 0, sizeof *put->stats);
-                }
-                dp_netdev_flow_add(pmd, &match, &ufid, put->actions,
+                dp_netdev_flow_add(pmd, match, ufid, put->actions,
                                    put->actions_len);
                 error = 0;
             } else {
@@ -2366,7 +2338,7 @@ dpif_netdev_flow_put(struct dpif *dpif, const struct dpif_flow_put *put)
         }
     } else {
         if (put->flags & DPIF_FP_MODIFY
-            && flow_equal(&match.flow, &netdev_flow->flow)) {
+            && flow_equal(&match->flow, &netdev_flow->flow)) {
             struct dp_netdev_actions *new_actions;
             struct dp_netdev_actions *old_actions;
 
@@ -2376,8 +2348,8 @@ dpif_netdev_flow_put(struct dpif *dpif, const struct dpif_flow_put *put)
             old_actions = dp_netdev_flow_get_actions(netdev_flow);
             ovsrcu_set(&netdev_flow->actions, new_actions);
 
-            if (put->stats) {
-                get_dpif_flow_stats(netdev_flow, put->stats);
+            if (stats) {
+                get_dpif_flow_stats(netdev_flow, stats);
             }
             if (put->flags & DPIF_FP_ZERO_STATS) {
                 /* XXX: The userspace datapath uses thread local statistics
@@ -2401,39 +2373,137 @@ dpif_netdev_flow_put(struct dpif *dpif, const struct dpif_flow_put *put)
         }
     }
     ovs_mutex_unlock(&pmd->flow_mutex);
-    dp_netdev_pmd_unref(pmd);
-
     return error;
 }
 
 static int
-dpif_netdev_flow_del(struct dpif *dpif, const struct dpif_flow_del *del)
+dpif_netdev_flow_put(struct dpif *dpif, const struct dpif_flow_put *put)
 {
     struct dp_netdev *dp = get_dp_netdev(dpif);
-    struct dp_netdev_flow *netdev_flow;
+    struct netdev_flow_key key;
     struct dp_netdev_pmd_thread *pmd;
-    unsigned pmd_id = del->pmd_id == PMD_ID_NULL
-                      ? NON_PMD_CORE_ID : del->pmd_id;
-    int error = 0;
+    struct match match;
+    ovs_u128 ufid;
+    int error;
 
-    pmd = dp_netdev_get_pmd(dp, pmd_id);
-    if (!pmd) {
-        return EINVAL;
+    if (put->stats) {
+        memset(put->stats, 0, sizeof *put->stats);
+    }
+    error = dpif_netdev_flow_from_nlattrs(put->key, put->key_len, &match.flow);
+    if (error) {
+        return error;
+    }
+    error = dpif_netdev_mask_from_nlattrs(put->key, put->key_len,
+                                          put->mask, put->mask_len,
+                                          &match.flow, &match.wc);
+    if (error) {
+        return error;
+    }
+
+    if (put->ufid) {
+        ufid = *put->ufid;
+    } else {
+        dpif_flow_hash(dpif, &match.flow, sizeof match.flow, &ufid);
     }
 
+    /* Must produce a netdev_flow_key for lookup.
+     * This interface is no longer performance critical, since it is not used
+     * for upcall processing any more. */
+    netdev_flow_key_from_flow(&key, &match.flow);
+
+    if (put->pmd_id == PMD_ID_NULL) {
+        if (cmap_count(&dp->poll_threads) == 0) {
+            return EINVAL;
+        }
+        CMAP_FOR_EACH (pmd, node, &dp->poll_threads) {
+            struct dpif_flow_stats pmd_stats;
+            int pmd_error;
+
+            pmd_error = flow_put_on_pmd(pmd, &key, &match, &ufid, put,
+                                        &pmd_stats);
+            if (pmd_error) {
+                error = pmd_error;
+            } else if (put->stats) {
+                put->stats->n_packets += pmd_stats.n_packets;
+                put->stats->n_bytes += pmd_stats.n_bytes;
+                put->stats->used = MAX(put->stats->used, pmd_stats.used);
+                put->stats->tcp_flags |= pmd_stats.tcp_flags;
+            }
+        }
+    } else {
+        pmd = dp_netdev_get_pmd(dp, put->pmd_id);
+        if (!pmd) {
+            return EINVAL;
+        }
+        error = flow_put_on_pmd(pmd, &key, &match, &ufid, put, put->stats);
+        dp_netdev_pmd_unref(pmd);
+    }
+
+    return error;
+}
+
+static int
+flow_del_on_pmd(struct dp_netdev_pmd_thread *pmd,
+                struct dpif_flow_stats *stats,
+                const struct dpif_flow_del *del)
+{
+    struct dp_netdev_flow *netdev_flow;
+    int error = 0;
+
     ovs_mutex_lock(&pmd->flow_mutex);
     netdev_flow = dp_netdev_pmd_find_flow(pmd, del->ufid, del->key,
                                           del->key_len);
     if (netdev_flow) {
-        if (del->stats) {
-            get_dpif_flow_stats(netdev_flow, del->stats);
+        if (stats) {
+            get_dpif_flow_stats(netdev_flow, stats);
         }
         dp_netdev_pmd_remove_flow(pmd, netdev_flow);
     } else {
         error = ENOENT;
     }
     ovs_mutex_unlock(&pmd->flow_mutex);
-    dp_netdev_pmd_unref(pmd);
+
+    return error;
+}
+
+static int
+dpif_netdev_flow_del(struct dpif *dpif, const struct dpif_flow_del *del)
+{
+    struct dp_netdev *dp = get_dp_netdev(dpif);
+    struct dp_netdev_pmd_thread *pmd;
+    int error = 0;
+
+    if (del->stats) {
+        memset(del->stats, 0, sizeof *del->stats);
+    }
+
+    if (del->pmd_id == PMD_ID_NULL) {
+        if (cmap_count(&dp->poll_threads) == 0) {
+            return EINVAL;
+        }
+        CMAP_FOR_EACH (pmd, node, &dp->poll_threads) {
+            struct dpif_flow_stats pmd_stats;
+            int pmd_error;
+
+            pmd_error = flow_del_on_pmd(pmd, &pmd_stats, del);
+            if (pmd_error) {
+                error = pmd_error;
+            } else if (del->stats) {
+                del->stats->n_packets += pmd_stats.n_packets;
+                del->stats->n_bytes += pmd_stats.n_bytes;
+                del->stats->used = MAX(del->stats->used, pmd_stats.used);
+                del->stats->tcp_flags |= pmd_stats.tcp_flags;
+            }
+        }
+    } else {
+        pmd = dp_netdev_get_pmd(dp, del->pmd_id);
+        if (!pmd) {
+            return EINVAL;
+        }
+        error = flow_del_on_pmd(pmd, del->stats, del);
+        dp_netdev_pmd_unref(pmd);
+    }
+
 
     return error;
 }
diff --git a/lib/dpif.c b/lib/dpif.c
index cea2448f7..50c338287 100644
--- a/lib/dpif.c
+++ b/lib/dpif.c
@@ -907,7 +907,7 @@ dpif_probe_feature(struct dpif *dpif, const char *name,
      * previous run are still present in the datapath. */
     error = dpif_flow_put(dpif, DPIF_FP_CREATE | DPIF_FP_MODIFY | DPIF_FP_PROBE,
                           key->data, key->size, NULL, 0, NULL, 0,
-                          ufid, PMD_ID_NULL, NULL);
+                          ufid, NON_PMD_CORE_ID, NULL);
     if (error) {
         if (error != EINVAL) {
             VLOG_WARN("%s: %s flow probe failed (%s)",
@@ -918,7 +918,7 @@ dpif_probe_feature(struct dpif *dpif, const char *name,
 
     ofpbuf_use_stack(&reply, &stub, sizeof stub);
     error = dpif_flow_get(dpif, key->data, key->size, ufid,
-                          PMD_ID_NULL, &reply, &flow);
+                          NON_PMD_CORE_ID, &reply, &flow);
     if (!error
         && (!ufid || (flow.ufid_present
                       && ovs_u128_equals(*ufid, flow.ufid)))) {
@@ -926,7 +926,7 @@ dpif_probe_feature(struct dpif *dpif, const char *name,
     }
 
     error = dpif_flow_del(dpif, key->data, key->size, ufid,
-                          PMD_ID_NULL, NULL);
+                          NON_PMD_CORE_ID, NULL);
     if (error) {
         VLOG_WARN("%s: failed to delete %s feature probe flow",
                   dpif_name(dpif), name);
diff --git a/lib/dpif.h b/lib/dpif.h
index 40ffe29e7..aa4fb8b84 100644
--- a/lib/dpif.h
+++ b/lib/dpif.h
@@ -629,7 +629,8 @@ enum dpif_op_type {
  *
  *   - If the datapath implements multiple pmd thread with its own flow
  *     table, 'pmd_id' should be used to specify the particular polling
- *     thread for the operation.
+ *     thread for the operation. PMD_ID_NULL means that the flow should
+ *     be put on all the polling threads.
  */
 struct dpif_flow_put {
     /* Input. */
@@ -661,7 +662,8 @@ struct dpif_flow_put {
  *
  * If the datapath implements multiple polling thread with its own flow table,
  * 'pmd_id' should be used to specify the particular polling thread for the
- * operation.
+ * operation. PMD_ID_NULL means that the flow should be deleted from all the
+ * polling threads.
  *
  * If the operation succeeds, then 'stats', if nonnull, will be set to the
  * flow's statistics before its deletion. */
@@ -726,7 +728,8 @@ struct dpif_execute {
  *
  * If the datapath implements multiple polling thread with its own flow table,
  * 'pmd_id' should be used to specify the particular polling thread for the
- * operation.
+ * operation. PMD_ID_NULL means that the datapath will return the first
+ * matching flow from any poll thread.
  *
  * Succeeds with status 0 if the flow is fetched, or fails with ENOENT if no
  * such flow exists. Other failures are indicated with a positive errno value.
@@ -861,6 +864,9 @@ void dpif_get_netflow_ids(const struct dpif *,
 int dpif_queue_to_priority(const struct dpif *, uint32_t queue_id,
                            uint32_t *priority);
 
+int dpif_get_pmds_for_port(const struct dpif * dpif, odp_port_t port_no,
+                           unsigned int **pmds, size_t *n);
+
 char *dpif_get_dp_version(const struct dpif *);
 bool dpif_supports_tnl_push_pop(const struct dpif *);
 #ifdef  __cplusplus