zebra: limit queued route updates

Impose a configurable limit on the number of route updates
that can be queued towards the dataplane subsystem.

Signed-off-by: Mark Stapp <mjs@voltanet.io>

diff --git a/zebra/zebra_dplane.c b/zebra/zebra_dplane.c
index 641e73b6525c9bd527732d3d302b9e2c39a9ba0d..9d75ae940a79c6558aacfef84a76ac23e13d2450 100644 (file)
--- a/zebra/zebra_dplane.c
+++ b/zebra/zebra_dplane.c
@@ -38,6 +38,10 @@ DEFINE_MTYPE(ZEBRA, DP_PROV, "Zebra DPlane Provider")
 #  define AOK 0
 #endif
 
+/* Default value for max queued incoming updates */
+const uint32_t DPLANE_DEFAULT_MAX_QUEUED = 200;
+
+
 /* Validation check macro for context blocks */
 /* #define DPLANE_DEBUG 1 */
 
@@ -164,6 +168,9 @@ static struct zebra_dplane_globals {
        /* Counter used to assign internal ids to providers */
        uint32_t dg_provider_id;
 
+       /* Limit number of pending, unprocessed updates */
+       _Atomic uint32_t dg_max_queued_updates;
+
        _Atomic uint64_t dg_routes_in;
        _Atomic uint32_t dg_routes_queued;
        _Atomic uint32_t dg_routes_queued_max;
@@ -507,6 +514,37 @@ const struct zebra_dplane_info *dplane_ctx_get_ns(
  * End of dplane context accessors
  */
 
+/*
+ * Retrieve the limit on the number of pending, unprocessed updates.
+ */
+uint32_t dplane_get_in_queue_limit(void)
+{
+       return atomic_load_explicit(&zdplane_info.dg_max_queued_updates,
+                                   memory_order_relaxed);
+}
+
+/*
+ * Configure limit on the number of pending, queued updates.
+ */
+void dplane_set_in_queue_limit(uint32_t limit, bool set)
+{
+       /* Reset to default on 'unset' */
+       if (!set)
+               limit = DPLANE_DEFAULT_MAX_QUEUED;
+
+       atomic_store_explicit(&zdplane_info.dg_max_queued_updates, limit,
+                             memory_order_relaxed);
+}
+
+/*
+ * Retrieve the current queue depth of incoming, unprocessed updates
+ */
+uint32_t dplane_get_in_queue_len(void)
+{
+       return atomic_load_explicit(&zdplane_info.dg_routes_queued,
+                                   memory_order_seq_cst);
+}
+
 /*
  * Initialize a context block for a route update from zebra data structs.
  */
@@ -603,6 +641,7 @@ done:
 static int dplane_route_enqueue(struct zebra_dplane_ctx *ctx)
 {
        int ret = EINVAL;
+       uint32_t high, curr;
 
        /* Enqueue for processing by the dataplane thread */
        DPLANE_LOCK();
@@ -612,6 +651,21 @@ static int dplane_route_enqueue(struct zebra_dplane_ctx *ctx)
        }
        DPLANE_UNLOCK();
 
+       curr = atomic_add_fetch_explicit(&zdplane_info.dg_routes_queued,
+                                        1, memory_order_seq_cst);
+
+       /* Maybe update high-water counter also */
+       high = atomic_load_explicit(&zdplane_info.dg_routes_queued_max,
+                                   memory_order_seq_cst);
+       while (high < curr) {
+               if (atomic_compare_exchange_weak_explicit(
+                           &zdplane_info.dg_routes_queued_max,
+                           &high, curr,
+                           memory_order_seq_cst,
+                           memory_order_seq_cst))
+                       break;
+       }
+
        /* Ensure that an event for the dataplane thread is active */
        thread_add_event(zdplane_info.dg_master, dplane_route_process, NULL, 0,
                         &zdplane_info.dg_t_update);
@@ -694,33 +748,13 @@ dplane_route_update_internal(struct route_node *rn,
        }
 
 done:
-       /* Update counters */
+       /* Update counter */
        atomic_fetch_add_explicit(&zdplane_info.dg_routes_in, 1,
                                  memory_order_relaxed);
 
-       if (ret == AOK) {
-               uint32_t high, curr;
-
-               curr = atomic_fetch_add_explicit(&zdplane_info.dg_routes_queued,
-                                                1, memory_order_seq_cst);
-
-               /* We don't have add_and_fetch - sigh */
-               curr++;
-
-               /* Maybe update high-water counter also */
-               high = atomic_load_explicit(&zdplane_info.dg_routes_queued_max,
-                                           memory_order_seq_cst);
-               while (high < curr) {
-                       if (atomic_compare_exchange_weak_explicit(
-                                   &zdplane_info.dg_routes_queued_max,
-                                   &high, curr,
-                                   memory_order_seq_cst,
-                                   memory_order_seq_cst))
-                               break;
-               }
-
+       if (ret == AOK)
                result = ZEBRA_DPLANE_REQUEST_QUEUED;
-       } else if (ctx) {
+       else if (ctx) {
                atomic_fetch_add_explicit(&zdplane_info.dg_route_errors, 1,
                                          memory_order_relaxed);
                dplane_ctx_free(&ctx);
@@ -828,7 +862,7 @@ static int dplane_route_process(struct thread *event)
                ctx->zd_status = res;
 
                /* Enqueue result to zebra main context */
-               (*zdplane_info.dg_results_cb)(ctx);
+               zdplane_info.dg_results_cb(ctx);
 
                ctx = NULL;
        }
@@ -841,13 +875,15 @@ static int dplane_route_process(struct thread *event)
  */
 int dplane_show_helper(struct vty *vty, bool detailed)
 {
-       uint64_t queued, queue_max, errs, incoming;
+       uint64_t queued, limit, queue_max, errs, incoming;
 
        /* Using atomics because counters are being changed in different
         * contexts.
         */
        incoming = atomic_load_explicit(&zdplane_info.dg_routes_in,
                                        memory_order_relaxed);
+       limit = atomic_load_explicit(&zdplane_info.dg_max_queued_updates,
+                                    memory_order_relaxed);
        queued = atomic_load_explicit(&zdplane_info.dg_routes_queued,
                                      memory_order_relaxed);
        queue_max = atomic_load_explicit(&zdplane_info.dg_routes_queued_max,
@@ -857,6 +893,7 @@ int dplane_show_helper(struct vty *vty, bool detailed)
 
        vty_out(vty, "Route updates:            %"PRIu64"\n", incoming);
        vty_out(vty, "Route update errors:      %"PRIu64"\n", errs);
+       vty_out(vty, "Route update queue limit: %"PRIu64"\n", limit);
        vty_out(vty, "Route update queue depth: %"PRIu64"\n", queued);
        vty_out(vty, "Route update queue max:   %"PRIu64"\n", queue_max);
 
@@ -956,6 +993,8 @@ static void zebra_dplane_init_internal(struct zebra_t *zebra)
        TAILQ_INIT(&zdplane_info.dg_route_ctx_q);
        TAILQ_INIT(&zdplane_info.dg_providers_q);
 
+       zdplane_info.dg_max_queued_updates = DPLANE_DEFAULT_MAX_QUEUED;
+
        /* TODO -- register default kernel 'provider' during init */
 
        zdplane_info.dg_run = true;

diff --git a/zebra/zebra_dplane.h b/zebra/zebra_dplane.h
index 65f5350b2ad018076e15133683af2f39f6b7cf82..999e0f39e4c3607f13272a2a3f8fcadad7a305e7 100644 (file)
--- a/zebra/zebra_dplane.h
+++ b/zebra/zebra_dplane.h
@@ -193,6 +193,17 @@ enum zebra_dplane_result dplane_route_update(struct route_node *rn,
 enum zebra_dplane_result dplane_route_delete(struct route_node *rn,
                                             struct route_entry *re);
 
+/* Retrieve the limit on the number of pending, unprocessed updates. */
+uint32_t dplane_get_in_queue_limit(void);
+
+/* Configure limit on the number of pending, queued updates. If 'unset', reset
+ * to default value.
+ */
+void dplane_set_in_queue_limit(uint32_t limit, bool set);
+
+/* Retrieve the current queue depth of incoming, unprocessed updates */
+uint32_t dplane_get_in_queue_len(void);
+
 /*
  * Vty/cli apis
  */

diff --git a/zebra/zebra_rib.c b/zebra/zebra_rib.c
index a080bba659f6961c2b5917d034b92f74fa437cdd..0d1a149694664221683ae02d8076d0812ad3d20b 100644 (file)
--- a/zebra/zebra_rib.c
+++ b/zebra/zebra_rib.c
@@ -1873,8 +1873,7 @@ done:
 }
 
 /*
- * TODO - WIP version of route-update processing after async dataplane
- * update.
+ * Route-update results processing after async dataplane update.
  */
 static void rib_process_after(struct zebra_dplane_ctx *ctx)
 {
@@ -2185,6 +2184,22 @@ static wq_item_status meta_queue_process(struct work_queue *dummy, void *data)
 {
        struct meta_queue *mq = data;
        unsigned i;
+       uint32_t queue_len, queue_limit;
+
+       /* Ensure there's room for more dataplane updates */
+       queue_limit = dplane_get_in_queue_limit();
+       queue_len = dplane_get_in_queue_len();
+       if (queue_len > queue_limit) {
+               if (IS_ZEBRA_DEBUG_RIB_DETAILED)
+                       zlog_debug("meta_queue_process: dplane queue len %u, "
+                                  "limit %u, retrying", queue_len, queue_limit);
+
+               /* Ensure that the meta-queue is actually enqueued */
+               if (work_queue_empty(zebrad.ribq))
+                       work_queue_add(zebrad.ribq, zebrad.mq);
+
+               return WQ_QUEUE_BLOCKED;
+       }
 
        for (i = 0; i < MQ_SIZE; i++)
                if (process_subq(mq->subq[i], i)) {

diff --git a/zebra/zebra_vty.c b/zebra/zebra_vty.c
index d9881b98c55de3420319b392bb3608773d29bd33..14ca8134367d0024ada603ff75f24dace2873320 100644 (file)
--- a/zebra/zebra_vty.c
+++ b/zebra/zebra_vty.c
@@ -2604,6 +2604,38 @@ DEFUN (show_dataplane_providers,
        return dplane_show_provs_helper(vty, detailed);
 }
 
+/* Configure dataplane incoming queue limit */
+DEFUN (zebra_dplane_queue_limit,
+       zebra_dplane_queue_limit_cmd,
+       "zebra dplane limit (0-10000)",
+       ZEBRA_STR
+       "Zebra dataplane\n"
+       "Limit incoming queued updates\n"
+       "Number of queued updates\n")
+{
+       uint32_t limit = 0;
+
+       limit = strtoul(argv[3]->arg, NULL, 10);
+
+       dplane_set_in_queue_limit(limit, true);
+
+       return CMD_SUCCESS;
+}
+
+/* Reset dataplane queue limit to default value */
+DEFUN (no_zebra_dplane_queue_limit,
+       no_zebra_dplane_queue_limit_cmd,
+       "no zebra dplane limit [(0-10000)]",
+       NO_STR
+       ZEBRA_STR
+       "Zebra dataplane\n"
+       "Limit incoming queued updates\n"
+       "Number of queued updates\n")
+{
+       dplane_set_in_queue_limit(0, false);
+
+       return CMD_SUCCESS;
+}
 
 /* Table configuration write function. */
 static int config_write_table(struct vty *vty)
@@ -2737,4 +2769,6 @@ void zebra_vty_init(void)
 
        install_element(VIEW_NODE, &show_dataplane_cmd);
        install_element(VIEW_NODE, &show_dataplane_providers_cmd);
+       install_element(CONFIG_NODE, &zebra_dplane_queue_limit_cmd);
+       install_element(CONFIG_NODE, &no_zebra_dplane_queue_limit_cmd);
 }