dpif-netdev: Purge all ukeys when reconfigure pmd.

When dpdk configuration changes, all pmd threads are recreated
and rx queues of each port are reloaded.  After this process,
rx queue could be mapped to a different pmd thread other than
the one before reconfiguration.  However, this is totally
transparent to ofproto layer modules.  So, if the ofproto-dpif-upcall
module still holds ukeys generated before pmd thread recreation,
this old ukey will collide with the ukey for the new upcalls
from same traffic flow, causing flow installation failure.

To fix the bug, this commit adds a new call-back function
in dpif layer for notifying upper layer the purging of datapath
(e.g. pmd thread deletion in dpif-netdev).  So, the
ofproto-dpif-upcall module can react properly with deleting
the ukeys and with collecting flows' last stats.

Reported-by: Ilya Maximets <[email protected]>
Signed-off-by: Alex Wang <[email protected]>
Acked-by: Daniele Di Proietto <[email protected]>
Tested-by: Daniele Di Proietto <[email protected]>
Acked-by: Joe Stringer <[email protected]>

lib/dpif-netdev.c

@@ -204,6 +204,11 @@ struct dp_netdev {
     upcall_callback *upcall_cb;  /* Callback function for executing upcalls. */
     void *upcall_aux;
 
+    /* Callback function for notifying the purging of dp flows (during
+     * reseting pmd deletion). */
+    dp_purge_callback *dp_purge_cb;
+    void *dp_purge_aux;
+
     /* Stores all 'struct dp_netdev_pmd_thread's. */
     struct cmap poll_threads;
 
@@ -2851,7 +2856,7 @@ dp_netdev_destroy_pmd(struct dp_netdev_pmd_thread *pmd)
 /* Stops the pmd thread, removes it from the 'dp->poll_threads',
  * and unrefs the struct. */
 static void
-dp_netdev_del_pmd(struct dp_netdev_pmd_thread *pmd)
+dp_netdev_del_pmd(struct dp_netdev *dp, struct dp_netdev_pmd_thread *pmd)
 {
     /* Uninit the 'flow_cache' since there is
      * no actual thread uninit it for NON_PMD_CORE_ID. */
@@ -2863,6 +2868,11 @@ dp_netdev_del_pmd(struct dp_netdev_pmd_thread *pmd)
         ovs_numa_unpin_core(pmd->core_id);
         xpthread_join(pmd->thread, NULL);
     }
+    /* Purges the 'pmd''s flows after stopping the thread, but before
+     * destroying the flows, so that the flow stats can be collected. */
+    if (dp->dp_purge_cb) {
+        dp->dp_purge_cb(dp->dp_purge_aux, pmd->core_id);
+    }
     cmap_remove(&pmd->dp->poll_threads, &pmd->node, hash_int(pmd->core_id, 0));
     dp_netdev_pmd_unref(pmd);
 }
@@ -2874,7 +2884,7 @@ dp_netdev_destroy_all_pmds(struct dp_netdev *dp)
     struct dp_netdev_pmd_thread *pmd;
 
     CMAP_FOR_EACH (pmd, node, &dp->poll_threads) {
-        dp_netdev_del_pmd(pmd);
+        dp_netdev_del_pmd(dp, pmd);
     }
 }
 
@@ -2886,7 +2896,7 @@ dp_netdev_del_pmds_on_numa(struct dp_netdev *dp, int numa_id)
 
     CMAP_FOR_EACH (pmd, node, &dp->poll_threads) {
         if (pmd->numa_id == numa_id) {
-            dp_netdev_del_pmd(pmd);
+            dp_netdev_del_pmd(dp, pmd);
         }
     }
 }
@@ -3390,6 +3400,15 @@ struct dp_netdev_execute_aux {
     struct dp_netdev_pmd_thread *pmd;
 };
 
+static void
+dpif_netdev_register_dp_purge_cb(struct dpif *dpif, dp_purge_callback *cb,
+                                 void *aux)
+{
+    struct dp_netdev *dp = get_dp_netdev(dpif);
+    dp->dp_purge_aux = aux;
+    dp->dp_purge_cb = cb;
+}
+
 static void
 dpif_netdev_register_upcall_cb(struct dpif *dpif, upcall_callback *cb,
                                void *aux)
@@ -3637,6 +3656,7 @@ const struct dpif_class dpif_netdev_class = {
     NULL,                       /* recv */
     NULL,                       /* recv_wait */
     NULL,                       /* recv_purge */
+    dpif_netdev_register_dp_purge_cb,
     dpif_netdev_register_upcall_cb,
     dpif_netdev_enable_upcall,
     dpif_netdev_disable_upcall,

lib/dpif-netlink.c

@@ -2309,6 +2309,7 @@ const struct dpif_class dpif_netlink_class = {
     dpif_netlink_recv,
     dpif_netlink_recv_wait,
     dpif_netlink_recv_purge,
+    NULL,                       /* register_dp_purge_cb */
     NULL,                       /* register_upcall_cb */
     NULL,                       /* enable_upcall */
     NULL,                       /* disable_upcall */

lib/dpif-provider.h

@@ -361,13 +361,22 @@ struct dpif_class {
      * return. */
     void (*recv_purge)(struct dpif *dpif);
 
+    /* When 'dpif' is about to purge the datapath, the higher layer may want
+     * to be notified so that it could try reacting accordingly (e.g. grabbing
+     * all flow stats before they are gone).
+     *
+     * Registers an upcall callback function with 'dpif'.  This is only used
+     * if 'dpif' needs to notify the purging of datapath.  'aux' is passed to
+     * the callback on invocation. */
+    void (*register_dp_purge_cb)(struct dpif *, dp_purge_callback *, void *aux);
+
     /* For datapaths that run in userspace (i.e. dpif-netdev), threads polling
      * for incoming packets can directly call upcall functions instead of
      * offloading packet processing to separate handler threads. Datapaths
      * that directly call upcall functions should use the functions below to
      * to register an upcall function and enable / disable upcalls.
      *
-     * Registers an upcall callback function with 'dpif'. This is only used if
+     * Registers an upcall callback function with 'dpif'. This is only used
      * if 'dpif' directly executes upcall functions. 'aux' is passed to the
      * callback on invocation. */
     void (*register_upcall_cb)(struct dpif *, upcall_callback *, void *aux);

lib/dpif.c

@@ -1345,6 +1345,14 @@ dpif_handlers_set(struct dpif *dpif, uint32_t n_handlers)
     return error;
 }
 
+void
+dpif_register_dp_purge_cb(struct dpif *dpif, dp_purge_callback *cb, void *aux)
+{
+    if (dpif->dpif_class->register_dp_purge_cb) {
+        dpif->dpif_class->register_dp_purge_cb(dpif, cb, aux);
+    }
+}
+
 void
 dpif_register_upcall_cb(struct dpif *dpif, upcall_callback *cb, void *aux)
 {

lib/dpif.h

@@ -787,6 +787,19 @@ struct dpif_upcall {
     struct nlattr *actions;    /* Argument to OVS_ACTION_ATTR_USERSPACE. */
 };
 
+/* A callback to notify higher layer of dpif about to be purged, so that
+ * higher layer could try reacting to this (e.g. grabbing all flow stats
+ * before they are gone).  This function is currently implemented only by
+ * dpif-netdev.
+ *
+ * The caller needs to provide the 'aux' pointer passed down by higher
+ * layer from the dpif_register_notify_cb() function and the 'pmd_id' of
+ * the polling thread.
+ */
+    typedef void dp_purge_callback(void *aux, unsigned pmd_id);
+
+void dpif_register_dp_purge_cb(struct dpif *, dp_purge_callback *, void *aux);
+
 /* A callback to process an upcall, currently implemented only by dpif-netdev.
  *
  * The caller provides the 'packet' and 'flow' to process, the corresponding

ofproto/ofproto-dpif-upcall.c

@@ -333,6 +333,7 @@ static int upcall_receive(struct upcall *, const struct dpif_backer *,
 static void upcall_uninit(struct upcall *);
 
 static upcall_callback upcall_cb;
+static dp_purge_callback dp_purge_cb;
 
 static atomic_bool enable_megaflows = ATOMIC_VAR_INIT(true);
 static atomic_bool enable_ufid = ATOMIC_VAR_INIT(true);
@@ -386,6 +387,7 @@ udpif_create(struct dpif_backer *backer, struct dpif *dpif)
     }
 
     dpif_register_upcall_cb(dpif, upcall_cb, udpif);
+    dpif_register_dp_purge_cb(dpif, dp_purge_cb, udpif);
 
     return udpif;
 }
@@ -2258,6 +2260,39 @@ revalidator_purge(struct revalidator *revalidator)
 {
     revalidator_sweep__(revalidator, true);
 }
+
+/* In reaction to dpif purge, purges all 'ukey's with same 'pmd_id'. */
+static void
+dp_purge_cb(void *aux, unsigned pmd_id)
+{
+    struct udpif *udpif = aux;
+    size_t i;
+
+    udpif_pause_revalidators(udpif);
+    for (i = 0; i < N_UMAPS; i++) {
+        struct ukey_op ops[REVALIDATE_MAX_BATCH];
+        struct udpif_key *ukey;
+        struct umap *umap = &udpif->ukeys[i];
+        size_t n_ops = 0;
+
+        CMAP_FOR_EACH(ukey, cmap_node, &umap->cmap) {
+             if (ukey->pmd_id == pmd_id) {
+                delete_op_init(udpif, &ops[n_ops++], ukey);
+                if (n_ops == REVALIDATE_MAX_BATCH) {
+                    push_ukey_ops(udpif, umap, ops, n_ops);
+                    n_ops = 0;
+                }
+            }
+        }
+
+        if (n_ops) {
+            push_ukey_ops(udpif, umap, ops, n_ops);
+        }
+
+        ovsrcu_quiesce();
+    }
+    udpif_resume_revalidators(udpif);
+}
 
 static void
 upcall_unixctl_show(struct unixctl_conn *conn, int argc OVS_UNUSED,