[ovs-dev,RFC,2/2] controller: introduce meter incremental processing engine

diff --git a/controller/ofctrl.c b/controller/ofctrl.c
index bf715787e..14ca08e94 100644
--- a/controller/ofctrl.c
+++ b/controller/ofctrl.c
@@ -1810,40 +1810,26 @@  uint32_t ofctrl_get_meter_id(const char *name, bool new_id)
     return id;
 }
 
-static void
-add_meter(struct ovn_extend_table_info *m_desired,
-          const struct sbrec_meter_table *meter_table,
-          struct ovs_list *msgs)
+void
+set_meter(const struct sbrec_meter *meter, uint32_t id, int cmd)
 {
-    const struct sbrec_meter *sb_meter;
-    SBREC_METER_TABLE_FOR_EACH (sb_meter, meter_table) {
-        if (!strcmp(m_desired->name, sb_meter->name)) {
-            break;
-        }
-    }
-
-    if (!sb_meter) {
-        static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(5, 1);
-        VLOG_ERR_RL(&rl, "could not find meter named \"%s\"", m_desired->name);
-        return;
-    }
-
-    struct ofputil_meter_mod mm;
-    mm.command = OFPMC13_ADD;
-    mm.meter.meter_id = m_desired->table_id;
-    mm.meter.flags = OFPMF13_STATS;
+    struct ovs_list msgs = OVS_LIST_INITIALIZER(&msgs);
+    struct ofputil_meter_mod mm = {
+        .command = cmd,
+        .meter.meter_id = id,
+        .meter.flags = OFPMF13_STATS,
+    };
 
-    if (!strcmp(sb_meter->unit, "pktps")) {
+    if (!strcmp(meter->unit, "pktps")) {
         mm.meter.flags |= OFPMF13_PKTPS;
     } else {
         mm.meter.flags |= OFPMF13_KBPS;
     }
-
-    mm.meter.n_bands = sb_meter->n_bands;
+    mm.meter.n_bands = meter->n_bands;
     mm.meter.bands = xcalloc(mm.meter.n_bands, sizeof *mm.meter.bands);
 
-    for (size_t i = 0; i < sb_meter->n_bands; i++) {
-        struct sbrec_meter_band *sb_band = sb_meter->bands[i];
+    for (size_t i = 0; i < meter->n_bands; i++) {
+        struct sbrec_meter_band *sb_band = meter->bands[i];
         struct ofputil_meter_band *mm_band = &mm.meter.bands[i];
 
         if (!strcmp(sb_band->action, "drop")) {
@@ -1858,9 +1844,41 @@  add_meter(struct ovn_extend_table_info *m_desired,
             mm.meter.flags |= OFPMF13_BURST;
         }
     }
-
-    add_meter_mod(&mm, msgs);
+    add_meter_mod(&mm, &msgs);
     free(mm.meter.bands);
+
+    if (!ovs_list_is_empty(&msgs)) {
+        /* Add a barrier to the list of messages. */
+        struct ofpbuf *barrier = ofputil_encode_barrier_request(OFP15_VERSION);
+
+        ovs_list_push_back(&msgs, &barrier->list_node);
+        /* Queue the messages. */
+        struct ofpbuf *msg;
+        LIST_FOR_EACH_POP (msg, list_node, &msgs) {
+            queue_msg(msg);
+        }
+    }
+}
+
+void
+remove_meter(uint32_t id)
+{
+    struct ovs_list msgs = OVS_LIST_INITIALIZER(&msgs);
+    /* Delete the meter. */
+    struct ofputil_meter_mod mm = {
+        .command = OFPMC13_DELETE,
+        .meter = { .meter_id = id },
+    };
+    add_meter_mod(&mm, &msgs);
+    /* Add a barrier to the list of messages. */
+    struct ofpbuf *barrier = ofputil_encode_barrier_request(OFP15_VERSION);
+
+    ovs_list_push_back(&msgs, &barrier->list_node);
+    /* Queue the messages. */
+    struct ofpbuf *msg;
+    LIST_FOR_EACH_POP (msg, list_node, &msgs) {
+        queue_msg(msg);
+    }
 }
 
 static void
@@ -2161,7 +2179,6 @@  void
 ofctrl_put(struct ovn_desired_flow_table *lflow_table,
            struct ovn_desired_flow_table *pflow_table,
            struct shash *pending_ct_zones,
-           const struct sbrec_meter_table *meter_table,
            uint64_t req_cfg,
            bool lflows_changed,
            bool pflows_changed)
@@ -2246,8 +2263,6 @@  ofctrl_put(struct ovn_desired_flow_table *lflow_table,
             /* The "set-meter" action creates a meter entry name that
              * describes the meter itself. */
             add_meter_string(m_desired, &msgs);
-        } else {
-            add_meter(m_desired, meter_table, &msgs);
         }
     }
 
diff --git a/controller/ofctrl.h b/controller/ofctrl.h
index 0efcb2ef5..53af92cfb 100644
--- a/controller/ofctrl.h
+++ b/controller/ofctrl.h
@@ -29,6 +29,7 @@  struct match;
 struct ofpbuf;
 struct ovsrec_bridge;
 struct sbrec_meter_table;
+struct sbrec_meter;
 struct shash;
 
 struct ovn_desired_flow_table {
@@ -55,7 +56,6 @@  enum mf_field_id ofctrl_get_mf_field_id(void);
 void ofctrl_put(struct ovn_desired_flow_table *lflow_table,
                 struct ovn_desired_flow_table *pflow_table,
                 struct shash *pending_ct_zones,
-                const struct sbrec_meter_table *,
                 uint64_t nb_cfg,
                 bool lflow_changed,
                 bool pflow_changed);
@@ -130,5 +130,7 @@  bool ofctrl_is_connected(void);
 void ofctrl_set_probe_interval(int probe_interval);
 void ofctrl_get_memory_usage(struct simap *usage);
 uint32_t ofctrl_get_meter_id(const char *name, bool new_id);
+void set_meter(const struct sbrec_meter *meter, uint32_t id, int cmd);
+void remove_meter(uint32_t id);
 
 #endif /* controller/ofctrl.h */
diff --git a/controller/ovn-controller.c b/controller/ovn-controller.c
index 124877c23..a98ef70bc 100644
--- a/controller/ovn-controller.c
+++ b/controller/ovn-controller.c
@@ -76,6 +76,7 @@ 
 #include "stopwatch.h"
 #include "lib/inc-proc-eng.h"
 #include "hmapx.h"
+#include "openvswitch/ofp-util.h"
 
 VLOG_DEFINE_THIS_MODULE(main);
 
@@ -114,6 +115,7 @@  static unixctl_cb_func debug_delay_nb_cfg_report;
 #define OVS_STARTUP_TS_NAME "ovn-startup-ts"
 
 static struct engine *flow_engine;
+static struct engine *meter_engine;
 
 static char *parse_options(int argc, char *argv[]);
 OVS_NO_RETURN static void usage(void);
@@ -963,7 +965,8 @@  ctrl_register_ovs_idl(struct ovsdb_idl *ovs_idl)
     SB_NODE(dhcpv6_options, "dhcpv6_options") \
     SB_NODE(dns, "dns") \
     SB_NODE(load_balancer, "load_balancer") \
-    SB_NODE(fdb, "fdb")
+    SB_NODE(fdb, "fdb") \
+    SB_NODE(meter, "meter")
 
 enum sb_engine_node {
 #define SB_NODE(NAME, NAME_STR) SB_##NAME,
@@ -1508,6 +1511,65 @@  addr_sets_sb_address_set_handler(struct engine_node *node, void *data)
     return true;
 }
 
+struct ed_type_meter {
+    bool change_tracked;
+};
+
+static void *
+en_meter_init(struct engine_node *node OVS_UNUSED,
+              struct engine_arg *arg OVS_UNUSED)
+{
+    struct ed_type_meter *m = xzalloc(sizeof *m);
+
+    m->change_tracked = false;
+    return m;
+}
+
+static void
+en_meter_cleanup(void *data OVS_UNUSED)
+{
+}
+
+static void
+en_meter_run(struct engine_node *node, void *data)
+{
+    struct ed_type_meter *m = data;
+
+    engine_set_node_state(node, EN_UPDATED);
+    m->change_tracked = false;
+}
+
+static bool
+meter_sb_meter_handler(struct engine_node *node, void *data)
+{
+    struct ovs_list msgs = OVS_LIST_INITIALIZER(&msgs);
+    struct ed_type_meter *m = data;
+
+    struct sbrec_meter_table *m_table =
+        (struct sbrec_meter_table *)EN_OVSDB_GET(
+            engine_get_input("SB_meter", node));
+
+    uint32_t id;
+    const struct sbrec_meter *iter;
+    SBREC_METER_TABLE_FOR_EACH_TRACKED (iter, m_table) {
+        id = ofctrl_get_meter_id(iter->name, !sbrec_meter_is_deleted(iter));
+        if (id == EXT_TABLE_ID_INVALID) {
+            return false;
+        }
+
+        if (sbrec_meter_is_deleted(iter)) {
+            remove_meter(id);
+        } else {
+            int cmd = sbrec_meter_is_new(iter) ? OFPMC13_ADD : OFPMC13_MODIFY;
+
+            set_meter(iter, id, cmd);
+        }
+    }
+    m->change_tracked = true;
+
+    return true;
+}
+
 struct ed_type_port_groups{
     /* A copy of SB port_groups, each converted as a sset for efficient lport
      * lookup. */
@@ -3220,6 +3282,7 @@  main(int argc, char *argv[])
     ENGINE_NODE(lflow_output, "logical_flow_output");
     ENGINE_NODE(flow_output, "flow_output");
     ENGINE_NODE(addr_sets, "addr_sets");
+    ENGINE_NODE(meter, "meter");
     ENGINE_NODE_WITH_CLEAR_TRACK_DATA(port_groups, "port_groups");
 
 #define SB_NODE(NAME, NAME_STR) ENGINE_NODE_SB(NAME, NAME_STR);
@@ -3344,11 +3407,14 @@  main(int argc, char *argv[])
     engine_add_input(&en_flow_output, &en_pflow_output,
                      flow_output_pflow_output_handler);
 
+    engine_add_input(&en_meter, &en_sb_meter, meter_sb_meter_handler);
+
     struct engine_arg engine_arg = {
         .sb_idl = ovnsb_idl_loop.idl,
         .ovs_idl = ovs_idl_loop.idl,
     };
     engine_init(&flow_engine, &en_flow_output, &engine_arg);
+    engine_init(&meter_engine, &en_meter, &engine_arg);
 
     engine_ovsdb_node_add_index(&en_sb_chassis, "name", sbrec_chassis_by_name);
     engine_ovsdb_node_add_index(&en_sb_multicast_group, "name_datapath",
@@ -3481,6 +3547,7 @@  main(int argc, char *argv[])
         }
 
         engine_init_run(flow_engine);
+        engine_init_run(meter_engine);
 
         struct ovsdb_idl_txn *ovs_idl_txn = ovsdb_idl_loop_run(&ovs_idl_loop);
         unsigned int new_ovs_cond_seqno
@@ -3587,6 +3654,7 @@  main(int argc, char *argv[])
                 engine_set_force_recompute(flow_engine, true);
             }
 
+            engine_run(meter_engine, true);
             if (br_int) {
                 ct_zones_data = engine_get_data(&en_ct_zones);
                 if (ct_zones_data) {
@@ -3754,7 +3822,6 @@  main(int argc, char *argv[])
                         ofctrl_put(&lflow_output_data->flow_table,
                                    &pflow_output_data->flow_table,
                                    &ct_zones_data->pending,
-                                   sbrec_meter_table_get(ovnsb_idl_loop.idl),
                                    ofctrl_seqno_get_req_cfg(),
                                    engine_node_changed(&en_lflow_output),
                                    engine_node_changed(&en_pflow_output));
@@ -3898,6 +3965,7 @@  loop_done:
 
     engine_set_context(flow_engine, NULL);
     engine_cleanup(&flow_engine);
+    engine_cleanup(&meter_engine);
 
     /* It's time to exit.  Clean up the databases if we are not restarting */
     if (!restart) {
diff --git a/tests/ovn.at b/tests/ovn.at
index a4ed03041..2e5a31e7c 100644
--- a/tests/ovn.at
+++ b/tests/ovn.at
@@ -9181,7 +9181,7 @@  ovn-nbctl acl-del lsw0 to-lport 1000 'tcp.dst==80'
 ovn-nbctl --wait=hv sync
 AT_CHECK([ovs-ofctl -O OpenFlow13 dump-meters br-int | grep meter], [0], [ignore], [ignore])
 
-# Delete acl2, meter should be deleted in OVS
+# Delete acl2, meter should be kept in OVS
 ovn-nbctl acl-del lsw0 to-lport 1000 'tcp.dst==81'
 ovn-nbctl --wait=hv sync
 AT_CHECK([ovs-ofctl -O OpenFlow13 dump-meters br-int | grep meter], [1])
@@ -29690,3 +29690,54 @@  OVS_WAIT_UNTIL([
 OVN_CLEANUP([hv1],[hv2])
 AT_CLEANUP
 ])
+
+OVN_FOR_EACH_NORTHD([
+AT_SETUP([ovn-controller - check meters update])
+AT_KEYWORDS([meters-update])
+
+ovn_start
+
+net_add n1
+sim_add hv1
+as hv1
+ovs-vsctl add-br br-phys
+ovn_attach n1 br-phys 192.168.0.10
+
+check ovn-nbctl ls-add sw0
+check ovn-nbctl lsp-add sw0 lsp
+
+as hv1 ovs-vsctl \
+    -- add-port br-int vif1 \
+    -- set Interface vif1 external_ids:iface-id=lsp
+
+# Wait for port to be bound.
+wait_row_count Chassis 1 name=hv1
+ch=$(fetch_column Chassis _uuid name=hv1)
+wait_row_count Port_Binding 1 logical_port=lsp chassis=$ch
+
+# Add a new meter
+check ovn-nbctl meter-add meter0 drop 10 pktps
+check ovn-nbctl --log --severity=alert --meter=meter0 --name=http acl-add sw0 \
+                      to-lport 1000 'tcp.dst == 80' drop
+check ovn-nbctl --wait=hv sync
+
+AT_CHECK([as hv1 ovs-ofctl -OOpenFlow15 dump-meters br-int | grep -q rate=10], [0])
+
+# Update existing meter
+check ovn-nbctl --may-exist meter-add meter0 drop 20 pktps
+check ovn-nbctl --wait=hv sync
+
+AT_CHECK([as hv1 ovs-ofctl -OOpenFlow15 dump-meters br-int | grep -q rate=20], [0])
+
+# Update existing meter and do a full recompute
+as hv1 ovn-appctl -t ovn-controller debug/pause
+check ovn-nbctl --may-exist meter-add meter0 drop 30 pktps
+check ovn-sbctl chassis-add hv2 geneve 127.0.0.1
+check ovn-nbctl --wait=sb sync
+as hv1 ovn-appctl -t ovn-controller debug/resume
+
+AT_CHECK([as hv1 ovs-ofctl -OOpenFlow15 dump-meters br-int | grep -q rate=30], [0])
+
+OVN_CLEANUP([hv1])
+AT_CLEANUP
+])
diff --git a/tests/system-ovn.at b/tests/system-ovn.at
index 7f6cb32dc..80d5192ca 100644
--- a/tests/system-ovn.at
+++ b/tests/system-ovn.at
@@ -6667,8 +6667,25 @@  OVS_WAIT_UNTIL([
     test "${n_reject}" = "2"
 ])
 kill $(pidof tcpdump)
+rm -f reject.pcap
+
+# Let's update the meter
+NS_EXEC([sw01], [tcpdump -l -n -i sw01 icmp -Q in > reject.pcap &])
+check ovn-nbctl --may-exist meter-add acl-meter drop 10 pktps 0
+ip netns exec sw01 scapy -H <<-EOF
+p = IP(src="192.168.1.2", dst="192.168.1.1") / UDP(dport = 12345) / Raw(b"X"*64)
+send (p, iface='sw01', loop = 0, verbose = 0, count = 100)
+EOF
 
+# 10pps + 1 burst size
+OVS_WAIT_UNTIL([
+    n_reject=$(grep unreachable reject.pcap | wc -l)
+    test "${n_reject}" = "20"
+])
+
+kill $(pidof tcpdump)
 rm -f reject.pcap
+
 NS_EXEC([sw01], [tcpdump -l -n -i sw01 icmp -Q in > reject.pcap &])
 check ovn-nbctl --wait=hv ls-copp-del sw0 reject