ovn-controller: Add incremental processing to lflow_run and physical_run

This code changes to allow incremental processing of the
logical flow and physical binding tables whenver possible.

Note: flows created by physical_run for multicast_groups are
*NOT* handled incrementally due to to be solved issues
with GWs and local routers.

Signed-off-by: Ryan Moats <rmoats@us.ibm.com>
Signed-off-by: Ben Pfaff <blp@ovn.org>

ovn/controller/binding.c

@@ -15,6 +15,8 @@
 
 #include <config.h>
 #include "binding.h"
+#include "lflow.h"
+#include "lport.h"
 
 #include "lib/bitmap.h"
 #include "lib/hmap.h"
@@ -129,6 +131,7 @@ remove_local_datapath(struct hmap *local_datapaths, struct local_datapath *ld)
     }
     hmap_remove(local_datapaths, &ld->hmap_node);
     free(ld);
+    lflow_reset_processing();
 }
 
 static void
@@ -145,6 +148,9 @@ add_local_datapath(struct hmap *local_datapaths,
     memcpy(&ld->uuid, &binding_rec->header_.uuid, sizeof ld->uuid);
     hmap_insert(local_datapaths, &ld->hmap_node,
                 binding_rec->datapath->tunnel_key);
+    lport_index_reset();
+    mcgroup_index_reset();
+    lflow_reset_processing();
 }
 
 static void

ovn/controller/encaps.c

@@ -17,6 +17,7 @@
 #include "encaps.h"
 #include "binding.h"
 #include "lflow.h"
+#include "lport.h"
 
 #include "lib/hash.h"
 #include "lib/sset.h"
@@ -237,6 +238,9 @@ tunnel_add(const struct sbrec_chassis *chassis_rec,
     free(port_name);
     free(ports);
     binding_reset_processing();
+    lport_index_reset();
+    mcgroup_index_reset();
+    lflow_reset_processing();
     process_full_encaps = true;
 }
 
@@ -421,6 +425,7 @@ encaps_run(struct controller_ctx *ctx, const struct ovsrec_bridge *br_int,
                                 &port_hash->uuid_node);
                     free(port_hash);
                     binding_reset_processing();
+                    lflow_reset_processing();
                 }
             } else if (sbrec_chassis_is_new(chassis_rec)) {
                 check_and_add_tunnel(chassis_rec, chassis_id);

ovn/controller/lflow.c

@@ -27,6 +27,7 @@
 #include "ovn/lib/ovn-dhcp.h"
 #include "ovn/lib/ovn-sb-idl.h"
 #include "packets.h"
+#include "physical.h"
 #include "simap.h"
 #include "sset.h"
 
@@ -40,6 +41,17 @@ static struct shash symtab;
 /* Contains an internal expr datastructure that represents an address set. */
 static struct shash expr_address_sets;
 
+static bool full_flow_processing = false;
+static bool full_logical_flow_processing = false;
+static bool full_neighbor_flow_processing = false;
+
+void
+lflow_reset_processing(void)
+{
+    full_flow_processing = true;
+    physical_reset_processing();
+}
+
 static void
 add_logical_register(struct shash *symtab, enum mf_field_id id)
 {
@@ -367,6 +379,15 @@ add_logical_flows(struct controller_ctx *ctx, const struct lport_index *lports,
                   const struct simap *ct_zones)
 {
     uint32_t conj_id_ofs = 1;
+    const struct sbrec_logical_flow *lflow;
+
+    if (full_flow_processing) {
+        ovn_flow_table_clear();
+        full_logical_flow_processing = true;
+        full_neighbor_flow_processing = true;
+        full_flow_processing = false;
+        physical_reset_processing();
+    }
 
     struct hmap dhcp_opts = HMAP_INITIALIZER(&dhcp_opts);
     const struct sbrec_dhcp_options *dhcp_opt_row;
@@ -375,11 +396,35 @@ add_logical_flows(struct controller_ctx *ctx, const struct lport_index *lports,
                      dhcp_opt_row->type);
     }
 
-    const struct sbrec_logical_flow *lflow;
-    SBREC_LOGICAL_FLOW_FOR_EACH (lflow, ctx->ovnsb_idl) {
-        consider_logical_flow(lports, mcgroups, lflow, local_datapaths,
-                              patched_datapaths, group_table, ct_zones,
-                              &dhcp_opts, &conj_id_ofs);
+    if (full_logical_flow_processing) {
+        SBREC_LOGICAL_FLOW_FOR_EACH (lflow, ctx->ovnsb_idl) {
+            consider_logical_flow(lports, mcgroups, lflow, local_datapaths,
+                                  patched_datapaths, group_table, ct_zones,
+                                  &dhcp_opts, &conj_id_ofs);
+        }
+        full_logical_flow_processing = false;
+    } else {
+        /* First, remove any flows that should be removed. */
+        SBREC_LOGICAL_FLOW_FOR_EACH_TRACKED (lflow, ctx->ovnsb_idl) {
+            if (sbrec_logical_flow_is_deleted(lflow)) {
+                ofctrl_remove_flows(&lflow->header_.uuid);
+            }
+        }
+
+        /* Now, add/modify existing flows. */
+        SBREC_LOGICAL_FLOW_FOR_EACH_TRACKED (lflow, ctx->ovnsb_idl) {
+            if (!sbrec_logical_flow_is_deleted(lflow)) {
+                /* if the logical flow is a modification, just remove
+                 * the flows for this row, and then add new flows. */
+                if (!sbrec_logical_flow_is_new(lflow)) {
+                    ofctrl_remove_flows(&lflow->header_.uuid);
+                }
+                consider_logical_flow(lports, mcgroups, lflow,
+                                      local_datapaths, patched_datapaths,
+                                      group_table, ct_zones,
+                                      &dhcp_opts, &conj_id_ofs);
+            }
+        }
     }
 
     dhcp_opts_destroy(&dhcp_opts);
@@ -616,9 +661,24 @@ add_neighbor_flows(struct controller_ctx *ctx,
     ofpbuf_init(&ofpacts, 0);
 
     const struct sbrec_mac_binding *b;
-    SBREC_MAC_BINDING_FOR_EACH (b, ctx->ovnsb_idl) {
-        consider_neighbor_flow(lports, b, &ofpacts, &match);
+    if (full_neighbor_flow_processing) {
+        SBREC_MAC_BINDING_FOR_EACH (b, ctx->ovnsb_idl) {
+            consider_neighbor_flow(lports, b, &ofpacts, &match);
+        }
+        full_neighbor_flow_processing = false;
+    } else {
+        SBREC_MAC_BINDING_FOR_EACH_TRACKED (b, ctx->ovnsb_idl) {
+            if (sbrec_mac_binding_is_deleted(b)) {
+                ofctrl_remove_flows(&b->header_.uuid);
+            } else {
+                if (!sbrec_mac_binding_is_new(b)) {
+                    ofctrl_remove_flows(&b->header_.uuid);
+                }
+                consider_neighbor_flow(lports, b, &ofpacts, &match);
+            }
+        }
     }
+
     ofpbuf_uninit(&ofpacts);
 }
 

ovn/controller/lflow.h

@@ -67,5 +67,6 @@ void lflow_run(struct controller_ctx *, const struct lport_index *,
                struct group_table *group_table,
                const struct simap *ct_zones);
 void lflow_destroy(void);
+void lflow_reset_processing(void);
 
 #endif /* ovn/lflow.h */

ovn/controller/lport.c

@@ -48,7 +48,7 @@ lport_index_init(struct lport_index *lports)
     hmap_init(&lports->by_uuid);
 }
 
-void
+bool
 lport_index_remove(struct lport_index *lports, const struct uuid *uuid)
 {
     const struct lport *port_ = lport_lookup_by_uuid(lports, uuid);
@@ -58,7 +58,9 @@ lport_index_remove(struct lport_index *lports, const struct uuid *uuid)
         hmap_remove(&lports->by_key, &port->key_node);
         hmap_remove(&lports->by_uuid, &port->uuid_node);
         free(port);
+        return true;
     }
+    return false;
 }
 
 void
@@ -74,6 +76,7 @@ lport_index_clear(struct lport_index *lports)
         hmap_remove(&lports->by_uuid, &port->uuid_node);
         free(port);
     }
+    lflow_reset_processing();
 }
 
 static void
@@ -93,6 +96,7 @@ consider_lport_index(struct lport_index *lports,
                 uuid_hash(&pb->header_.uuid));
     memcpy(&p->uuid, &pb->header_.uuid, sizeof p->uuid);
     p->pb = pb;
+    lflow_reset_processing();
 }
 
 void
@@ -108,7 +112,10 @@ lport_index_fill(struct lport_index *lports, struct ovsdb_idl *ovnsb_idl)
     } else {
         SBREC_PORT_BINDING_FOR_EACH_TRACKED (pb, ovnsb_idl) {
             if (sbrec_port_binding_is_deleted(pb)) {
-                lport_index_remove(lports, &pb->header_.uuid);
+                while (lport_index_remove(lports, &pb->header_.uuid)) {
+                    ;
+                }
+                lflow_reset_processing();
             } else {
                 consider_lport_index(lports, pb);
             }
@@ -202,6 +209,7 @@ mcgroup_index_remove(struct mcgroup_index *mcgroups, const struct uuid *uuid)
         hmap_remove(&mcgroups->by_uuid, &mcgroup->uuid_node);
         free(mcgroup);
     }
+    lflow_reset_processing();
 }
 
 void
@@ -231,6 +239,7 @@ consider_mcgroup_index(struct mcgroup_index *mcgroups,
                 uuid_hash(&mg->header_.uuid));
     memcpy(&m->uuid, &mg->header_.uuid, sizeof m->uuid);
     m->mg = mg;
+    lflow_reset_processing();
 }
 
 void
@@ -247,6 +256,7 @@ mcgroup_index_fill(struct mcgroup_index *mcgroups, struct ovsdb_idl *ovnsb_idl)
         SBREC_MULTICAST_GROUP_FOR_EACH_TRACKED (mg, ovnsb_idl) {
             if (sbrec_multicast_group_is_deleted(mg)) {
                 mcgroup_index_remove(mcgroups, &mg->header_.uuid);
+                lflow_reset_processing();
             } else {
                 consider_mcgroup_index(mcgroups, mg);
             }

ovn/controller/lport.h

@@ -39,9 +39,10 @@ struct lport_index {
 void lport_index_reset(void);
 void lport_index_init(struct lport_index *);
 void lport_index_fill(struct lport_index *, struct ovsdb_idl *);
-void lport_index_remove(struct lport_index *, const struct uuid *);
+bool lport_index_remove(struct lport_index *, const struct uuid *);
 void lport_index_clear(struct lport_index *);
 void lport_index_destroy(struct lport_index *);
+void lport_index_rebuild(void);
 
 const struct sbrec_port_binding *lport_lookup_by_name(
     const struct lport_index *, const char *name);
@@ -73,6 +74,7 @@ void mcgroup_index_fill(struct mcgroup_index *, struct ovsdb_idl *);
 void mcgroup_index_remove(struct mcgroup_index *, const struct uuid *);
 void mcgroup_index_clear(struct mcgroup_index *);
 void mcgroup_index_destroy(struct mcgroup_index *);
+void mcgroup_index_rebuild(void);
 
 const struct sbrec_multicast_group *mcgroup_lookup_by_dp_name(
     const struct mcgroup_index *,

ovn/controller/ovn-controller.c

@@ -441,7 +441,6 @@ main(int argc, char *argv[])
             update_ct_zones(&all_lports, &patched_datapaths, &ct_zones,
                             ct_zone_bitmap);
 
-            ovn_flow_table_clear();
             lflow_run(&ctx, &lports, &mcgroups, &local_datapaths,
                       &patched_datapaths, &group_table, &ct_zones);
             if (chassis_id) {

ovn/controller/patch.c

@@ -18,8 +18,10 @@
 #include "patch.h"
 
 #include "hash.h"
+#include "lflow.h"
 #include "lib/hmap.h"
 #include "lib/vswitch-idl.h"
+#include "lport.h"
 #include "openvswitch/vlog.h"
 #include "ovn-controller.h"
 
@@ -93,6 +95,9 @@ create_patch_port(struct controller_ctx *ctx,
     ovsrec_bridge_verify_ports(src);
     ovsrec_bridge_set_ports(src, ports, src->n_ports + 1);
 
+    lport_index_reset();
+    mcgroup_index_reset();
+    lflow_reset_processing();
     free(ports);
 }
 
@@ -125,6 +130,9 @@ remove_port(struct controller_ctx *ctx,
             return;
         }
     }
+    lport_index_reset();
+    mcgroup_index_reset();
+    lflow_reset_processing();
 }
 
 /* Obtains external-ids:ovn-bridge-mappings from OVSDB and adds patch ports for

ovn/controller/physical.c

@@ -57,6 +57,13 @@ static struct hmap tunnels = HMAP_INITIALIZER(&tunnels);
 
 /* UUID to identify OF flows not associated with ovsdb rows. */
 static struct uuid *hc_uuid = NULL;
+static bool full_binding_processing = false;
+
+void
+physical_reset_processing(void)
+{
+    full_binding_processing = true;
+}
 
 /* Maps from a chassis to the OpenFlow port number of the tunnel that can be
  * used to reach that chassis. */
@@ -631,15 +638,51 @@ physical_run(struct controller_ctx *ctx, enum mf_field_id mff_ovn_geneve,
             bool is_patch = !strcmp(iface_rec->type, "patch");
             if (is_patch && localnet) {
                 /* localnet patch ports can be handled just like VIFs. */
-                simap_put(&localvif_to_ofport, localnet, ofport);
+                if (simap_find(&localvif_to_ofport, localnet)) {
+                    unsigned int old_port = simap_get(&localvif_to_ofport,
+                                                      localnet);
+                    if (old_port != ofport) {
+                        simap_put(&localvif_to_ofport, localnet, ofport);
+                        full_binding_processing = true;
+                        lflow_reset_processing();
+                    }
+                } else {
+                    simap_put(&localvif_to_ofport, localnet, ofport);
+                    full_binding_processing = true;
+                    lflow_reset_processing();
+                }
                 break;
             } else if (is_patch && l2gateway) {
                 /* L2 gateway patch ports can be handled just like VIFs. */
-                simap_put(&localvif_to_ofport, l2gateway, ofport);
+                if (simap_find(&localvif_to_ofport, l2gateway)) {
+                    unsigned int old_port = simap_get(&localvif_to_ofport,
+                                                      l2gateway);
+                    if (old_port != ofport) {
+                        simap_put(&localvif_to_ofport, l2gateway, ofport);
+                        full_binding_processing = true;
+                        lflow_reset_processing();
+                    }
+                } else {
+                    simap_put(&localvif_to_ofport, l2gateway, ofport);
+                    full_binding_processing = true;
+                    lflow_reset_processing();
+                }
                 break;
             } else if (is_patch && logpatch) {
                 /* Logical patch ports can be handled just like VIFs. */
-                simap_put(&localvif_to_ofport, logpatch, ofport);
+                if (simap_find(&localvif_to_ofport, logpatch)) {
+                    unsigned int old_port = simap_get(&localvif_to_ofport,
+                                                      logpatch);
+                    if (old_port != ofport) {
+                        simap_put(&localvif_to_ofport, logpatch, ofport);
+                        full_binding_processing = true;
+                        lflow_reset_processing();
+                    }
+                } else {
+                    simap_put(&localvif_to_ofport, logpatch, ofport);
+                    full_binding_processing = true;
+                    lflow_reset_processing();
+                }
                 break;
             } else if (chassis_id) {
                 enum chassis_tunnel_type tunnel_type;
@@ -662,12 +705,25 @@ physical_run(struct controller_ctx *ctx, enum mf_field_id mff_ovn_geneve,
                 tun->chassis_id = chassis_id;
                 tun->ofport = u16_to_ofp(ofport);
                 tun->type = tunnel_type;
+                full_binding_processing = true;
                 break;
             } else {
                 const char *iface_id = smap_get(&iface_rec->external_ids,
                                                 "iface-id");
                 if (iface_id) {
-                    simap_put(&localvif_to_ofport, iface_id, ofport);
+                    if (simap_find(&localvif_to_ofport, iface_id)) {
+                        unsigned int old_port = simap_get(&localvif_to_ofport,
+                                                          iface_id);
+                        if (old_port != ofport) {
+                            simap_put(&localvif_to_ofport, iface_id, ofport);
+                            full_binding_processing = true;
+                            lflow_reset_processing();
+                        }
+                    } else {
+                        simap_put(&localvif_to_ofport, iface_id, ofport);
+                        full_binding_processing = true;
+                        lflow_reset_processing();
+                    }
                 }
             }
         }
@@ -679,10 +735,24 @@ physical_run(struct controller_ctx *ctx, enum mf_field_id mff_ovn_geneve,
     /* Set up flows in table 0 for physical-to-logical translation and in table
      * 64 for logical-to-physical translation. */
     const struct sbrec_port_binding *binding;
-    SBREC_PORT_BINDING_FOR_EACH (binding, ctx->ovnsb_idl) {
-        consider_port_binding(mff_ovn_geneve, ct_zones,
-                              local_datapaths, patched_datapaths, binding,
-                              &ofpacts);
+    if (full_binding_processing) {
+        SBREC_PORT_BINDING_FOR_EACH (binding, ctx->ovnsb_idl) {
+            consider_port_binding(mff_ovn_geneve, ct_zones, local_datapaths,
+                                  patched_datapaths, binding, &ofpacts);
+        }
+        full_binding_processing = false;
+    } else {
+        SBREC_PORT_BINDING_FOR_EACH_TRACKED (binding, ctx->ovnsb_idl) {
+            if (sbrec_port_binding_is_deleted(binding)) {
+                ofctrl_remove_flows(&binding->header_.uuid);
+            } else {
+                if (!sbrec_port_binding_is_new(binding)) {
+                    ofctrl_remove_flows(&binding->header_.uuid);
+                }
+                consider_port_binding(mff_ovn_geneve, ct_zones, local_datapaths,
+                                      patched_datapaths, binding, &ofpacts);
+            }
+        }
     }
 
     /* Handle output to multicast groups, in tables 32 and 33. */
@@ -796,7 +866,6 @@ physical_run(struct controller_ctx *ctx, enum mf_field_id mff_ovn_geneve,
     ofctrl_add_flow(OFTABLE_DROP_LOOPBACK, 0, &match, &ofpacts, hc_uuid);
 
     ofpbuf_uninit(&ofpacts);
-    simap_clear(&localvif_to_ofport);
     HMAP_FOR_EACH_POP (tun, hmap_node, &tunnels) {
         free(tun);
     }

ovn/controller/physical.h

@@ -45,5 +45,6 @@ void physical_run(struct controller_ctx *, enum mf_field_id mff_ovn_geneve,
                   const struct ovsrec_bridge *br_int, const char *chassis_id,
                   const struct simap *ct_zones,
                   struct hmap *local_datapaths, struct hmap *patched_datapaths);
+void physical_reset_processing(void);
 
 #endif /* ovn/physical.h */