northd: Handle lb changes in lflow engine.

Since northd tracked data has the changed lb information,
the lflow change handler for northd inputs can now handle
lb updates incrementally.  All the lflows generated for
each lb is stored in the ovn_lb_datapaths->lflow_ref and
this is used similar to how we handle ovn_port changes.

Acked-by: Dumitru Ceara <dceara@redhat.com>
Acked-by: Han Zhou <hzhou@ovn.org>
Signed-off-by: Numan Siddique <numans@ovn.org>
(cherry picked from commit 1a7df26)

northd/en-lflow.c

@@ -123,11 +123,6 @@ lflow_northd_handler(struct engine_node *node,
         return false;
     }
 
-    /* Fall back to recompute if load balancers have changed. */
-    if (northd_has_lbs_in_tracked_data(&northd_data->trk_data)) {
-        return false;
-    }
-
     const struct engine_context *eng_ctx = engine_get_context();
     struct lflow_data *lflow_data = data;
 
@@ -141,6 +136,12 @@ lflow_northd_handler(struct engine_node *node,
         return false;
     }
 
+    if (!lflow_handle_northd_lb_changes(
+            eng_ctx->ovnsb_idl_txn, &northd_data->trk_data.trk_lbs,
+            &lflow_input, lflow_data->lflow_table)) {
+        return false;
+    }
+
     engine_set_node_state(node, EN_UPDATED);
     return true;
 }

northd/lb.c

@@ -23,7 +23,8 @@
 
 /* OVN includes */
 #include "lb.h"
-#include "lib/ovn-nb-idl.h"
+#include "lflow-mgr.h"
+#include "lib/lb.h"
 #include "northd.h"
 #include "ovn/lex.h"
 
@@ -563,6 +564,7 @@ ovn_lb_datapaths_create(const struct ovn_northd_lb *lb, size_t n_ls_datapaths,
     lb_dps->lb = lb;
     lb_dps->nb_ls_map = bitmap_allocate(n_ls_datapaths);
     lb_dps->nb_lr_map = bitmap_allocate(n_lr_datapaths);
+    lb_dps->lflow_ref = lflow_ref_create();
 
     return lb_dps;
 }
@@ -572,6 +574,7 @@ ovn_lb_datapaths_destroy(struct ovn_lb_datapaths *lb_dps)
 {
     bitmap_free(lb_dps->nb_lr_map);
     bitmap_free(lb_dps->nb_ls_map);
+    lflow_ref_destroy(lb_dps->lflow_ref);
     free(lb_dps);
 }
 

northd/lb.h

@@ -128,6 +128,7 @@ void ovn_lb_group_reinit(
     const struct nbrec_load_balancer_group *,
     const struct hmap *lbs);
 
+struct lflow_ref;
 struct ovn_lb_datapaths {
     struct hmap_node hmap_node;
 
@@ -137,6 +138,33 @@ struct ovn_lb_datapaths {
 
     size_t n_nb_lr;
     unsigned long *nb_lr_map;
+
+    /* Reference of lflows generated for this load balancer.
+     *
+     * This data is initialized and destroyed by the en_northd node, but
+     * populated and used only by the en_lflow node. Ideally this data should
+     * be maintained as part of en_lflow's data (struct lflow_data): a hash
+     * index from ovn_port key to lflows.  However, it would be less efficient
+     * and more complex:
+     *
+     * 1. It would require an extra search (using the index) to find the
+     * lflows.
+     *
+     * 2. Building the index needs to be thread-safe, using either a global
+     * lock which is obviously less efficient, or hash-based lock array which
+     * is more complex.
+     *
+     * Maintaining the lflow_ref here is more straightforward. The drawback is
+     * that we need to keep in mind that this data belongs to en_lflow node,
+     * so never access it from any other nodes.
+     *
+     * 'lflow_ref' is used to reference logical flows generated for this
+     *  load balancer.
+     *
+     * Note: lflow_ref is not thread safe.  Only one thread should
+     * access ovn_lb_datapaths->lflow_ref at any given time.
+     */
+    struct lflow_ref *lflow_ref;
 };
 
 struct ovn_lb_datapaths *ovn_lb_datapaths_create(const struct ovn_northd_lb *,

northd/lflow-mgr.c

@@ -537,7 +537,15 @@ struct lflow_ref_node {
     /* The lflow. */
     struct ovn_lflow *lflow;
 
-    /* Index id of the datapath this lflow_ref_node belongs to. */
+    /* Indicates whether the lflow was added with a dp_group using the
+     * ovn_lflow_add_with_dp_group() macro. */
+    bool dpgrp_lflow;
+    /* dpgrp bitmap and bitmap length.  Valid only of dpgrp_lflow is true. */
+    unsigned long *dpgrp_bitmap;
+    size_t dpgrp_bitmap_len;
+
+    /* Index id of the datapath this lflow_ref_node belongs to.
+     * Valid only if dpgrp_lflow is false. */
     size_t dp_index;
 
     /* Indicates if the lflow_ref_node for an lflow - L(M, A) is linked
@@ -575,17 +583,29 @@ lflow_ref_destroy(struct lflow_ref *lflow_ref)
 
 /* Unlinks the lflows referenced by the 'lflow_ref'.
  * For each lflow_ref_node (lrn) in the lflow_ref, it basically clears
- * the datapath id (lrn->dp_index) from the lrn->lflow's dpg bitmap.
+ * the datapath id (lrn->dp_index) or all the datapath id bits in the
+ * dp group bitmap (set when ovn_lflow_add_with_dp_group macro was used)
+ * from the lrn->lflow's dpg bitmap
  */
 void
 lflow_ref_unlink_lflows(struct lflow_ref *lflow_ref)
 {
     struct lflow_ref_node *lrn;
 
     HMAP_FOR_EACH (lrn, ref_node, &lflow_ref->lflow_ref_nodes) {
-        if (dp_refcnt_release(&lrn->lflow->dp_refcnts_map,
-                              lrn->dp_index)) {
-            bitmap_set0(lrn->lflow->dpg_bitmap, lrn->dp_index);
+        if (lrn->dpgrp_lflow) {
+            size_t index;
+            BITMAP_FOR_EACH_1 (index, lrn->dpgrp_bitmap_len,
+                               lrn->dpgrp_bitmap) {
+                if (dp_refcnt_release(&lrn->lflow->dp_refcnts_map, index)) {
+                    bitmap_set0(lrn->lflow->dpg_bitmap, index);
+                }
+            }
+        } else {
+            if (dp_refcnt_release(&lrn->lflow->dp_refcnts_map,
+                                  lrn->dp_index)) {
+                bitmap_set0(lrn->lflow->dpg_bitmap, lrn->dp_index);
+            }
         }
 
         lrn->linked = false;
@@ -675,17 +695,25 @@ lflow_table_add_lflow(struct lflow_table *lflow_table,
                          io_port, ctrl_meter, stage_hint, where);
 
     if (lflow_ref) {
-        /* lflow referencing is only supported if 'od' is not NULL. */
-        ovs_assert(od);
-
         struct lflow_ref_node *lrn =
             lflow_ref_node_find(&lflow_ref->lflow_ref_nodes, lflow, hash);
         if (!lrn) {
             lrn = xzalloc(sizeof *lrn);
             lrn->lflow = lflow;
             lrn->lflow_ref = lflow_ref;
-            lrn->dp_index = od->index;
-            dp_refcnt_use(&lflow->dp_refcnts_map, lrn->dp_index);
+            lrn->dpgrp_lflow = !od;
+            if (lrn->dpgrp_lflow) {
+                lrn->dpgrp_bitmap = bitmap_clone(dp_bitmap, dp_bitmap_len);
+                lrn->dpgrp_bitmap_len = dp_bitmap_len;
+
+                size_t index;
+                BITMAP_FOR_EACH_1 (index, dp_bitmap_len, dp_bitmap) {
+                    dp_refcnt_use(&lflow->dp_refcnts_map, index);
+                }
+            } else {
+                lrn->dp_index = od->index;
+                dp_refcnt_use(&lflow->dp_refcnts_map, lrn->dp_index);
+            }
             ovs_list_insert(&lflow->referenced_by, &lrn->ref_list_node);
             hmap_insert(&lflow_ref->lflow_ref_nodes, &lrn->ref_node, hash);
         }
@@ -1418,5 +1446,8 @@ lflow_ref_node_destroy(struct lflow_ref_node *lrn)
 {
     hmap_remove(&lrn->lflow_ref->lflow_ref_nodes, &lrn->ref_node);
     ovs_list_remove(&lrn->ref_list_node);
+    if (lrn->dpgrp_lflow) {
+        bitmap_free(lrn->dpgrp_bitmap);
+    }
     free(lrn);
 }