event/dlb2: improve enqueue efficiency

If user tried to send 64 events, it will take 64 credits. Enqueue API
only gets 32 credits each time. If it does not have sufficient credits,
it just fails and returns. Enqueue API does not retry. On next enqueue,
it will get 32 more and send will work. This results in alternate
enqueues failing.

Add check to make sure DLB2_MAX_ENQUEUE_DEPTH <=
both DLB2_SW_CREDIT_QUANTA_DEFAULT and DLB2_SW_CREDIT_BATCH_SZ.

Add enough retires in the driver to satisfy max enqueue depth credits
based on set quanta. Note the credit quanta size is different for each
port. Retry count = Max enqueue depth / credit quanta in driver before
returning no credit.

Fixes: 3a6d0c04e7fb ("event/dlb2: add port setup")
Cc: stable@dpdk.org
Signed-off-by: Timothy McDaniel <timothy.mcdaniel@intel.com>

diff --git a/drivers/event/dlb2/dlb2.c b/drivers/event/dlb2/dlb2.c
index 36f07d0061bef6d99516522d23fa1b3337b942dc..3641ed29421e1e07fd88bb3922e1a5d4c85ac317 100644 (file)
--- a/drivers/event/dlb2/dlb2.c
+++ b/drivers/event/dlb2/dlb2.c
@@ -387,6 +387,11 @@ set_sw_credit_quanta(const char *key __rte_unused,
        if (ret < 0)
                return ret;
 
+       if (*sw_credit_quanta <= 0) {
+               DLB2_LOG_ERR("sw_credit_quanta must be > 0\n");
+               return -EINVAL;
+       }
+
        return 0;
 }
 
@@ -1773,9 +1778,48 @@ dlb2_eventdev_port_setup(struct rte_eventdev *dev,
                return -EINVAL;
        }
 
+       /* Default for worker ports */
+       sw_credit_quanta = dlb2->sw_credit_quanta;
+       hw_credit_quanta = dlb2->hw_credit_quanta;
+
        ev_port->qm_port.is_directed = port_conf->event_port_cfg &
                RTE_EVENT_PORT_CFG_SINGLE_LINK;
 
+       /*
+        * Validate credit config before creating port
+        */
+
+       /* Default for worker ports */
+       sw_credit_quanta = dlb2->sw_credit_quanta;
+       hw_credit_quanta = dlb2->hw_credit_quanta;
+
+       if (port_conf->event_port_cfg & RTE_EVENT_PORT_CFG_HINT_PRODUCER) {
+               /* Producer type ports. Mostly enqueue */
+               sw_credit_quanta = DLB2_SW_CREDIT_P_QUANTA_DEFAULT;
+               hw_credit_quanta = DLB2_SW_CREDIT_P_BATCH_SZ;
+       }
+       if (port_conf->event_port_cfg & RTE_EVENT_PORT_CFG_HINT_CONSUMER) {
+               /* Consumer type ports. Mostly dequeue */
+               sw_credit_quanta = DLB2_SW_CREDIT_C_QUANTA_DEFAULT;
+               hw_credit_quanta = DLB2_SW_CREDIT_C_BATCH_SZ;
+       }
+       ev_port->credit_update_quanta = sw_credit_quanta;
+       ev_port->qm_port.hw_credit_quanta = hw_credit_quanta;
+
+       if (port_conf->enqueue_depth > sw_credit_quanta ||
+           port_conf->enqueue_depth > hw_credit_quanta) {
+               DLB2_LOG_ERR("Invalid port config. Enqueue depth %d must be <= credit quanta %d and batch size %d\n",
+                            port_conf->enqueue_depth,
+                            sw_credit_quanta,
+                            hw_credit_quanta);
+               return -EINVAL;
+       }
+       ev_port->enq_retries = port_conf->enqueue_depth / sw_credit_quanta;
+
+       /*
+        * Create port
+        */
+
        if (!ev_port->qm_port.is_directed) {
                ret = dlb2_hw_create_ldb_port(dlb2,
                                              ev_port,
@@ -1811,23 +1855,6 @@ dlb2_eventdev_port_setup(struct rte_eventdev *dev,
        ev_port->inflight_credits = 0;
        ev_port->dlb2 = dlb2; /* reverse link */
 
-       /* Default for worker ports */
-       sw_credit_quanta = dlb2->sw_credit_quanta;
-       hw_credit_quanta = dlb2->hw_credit_quanta;
-
-       if (port_conf->event_port_cfg & RTE_EVENT_PORT_CFG_HINT_PRODUCER) {
-               /* Producer type ports. Mostly enqueue */
-               sw_credit_quanta = DLB2_SW_CREDIT_P_QUANTA_DEFAULT;
-               hw_credit_quanta = DLB2_SW_CREDIT_P_BATCH_SZ;
-       }
-       if (port_conf->event_port_cfg & RTE_EVENT_PORT_CFG_HINT_CONSUMER) {
-               /* Consumer type ports. Mostly dequeue */
-               sw_credit_quanta = DLB2_SW_CREDIT_C_QUANTA_DEFAULT;
-               hw_credit_quanta = DLB2_SW_CREDIT_C_BATCH_SZ;
-       }
-       ev_port->credit_update_quanta = sw_credit_quanta;
-       ev_port->qm_port.hw_credit_quanta = hw_credit_quanta;
-
        /* Tear down pre-existing port->queue links */
        if (dlb2->run_state == DLB2_RUN_STATE_STOPPED)
                dlb2_port_link_teardown(dlb2, &dlb2->ev_ports[ev_port_id]);
@@ -2970,6 +2997,7 @@ __dlb2_event_enqueue_burst(void *event_port,
        struct dlb2_eventdev_port *ev_port = event_port;
        struct dlb2_port *qm_port = &ev_port->qm_port;
        struct process_local_port_data *port_data;
+       int retries = ev_port->enq_retries;
        int i;
 
        RTE_ASSERT(ev_port->enq_configured);
@@ -2993,6 +3021,7 @@ __dlb2_event_enqueue_burst(void *event_port,
                for (; j < DLB2_NUM_QES_PER_CACHE_LINE && (i + j) < num; j++) {
                        const struct rte_event *ev = &events[i + j];
                        int16_t thresh = qm_port->token_pop_thresh;
+                       int ret;
 
                        if (use_delayed &&
                            qm_port->token_pop_mode == DELAYED_POP &&
@@ -3014,9 +3043,18 @@ __dlb2_event_enqueue_burst(void *event_port,
                                break;
                        }
 
-                       if (dlb2_event_enqueue_prep(ev_port, qm_port, ev,
-                                                   &sched_types[j],
-                                                   &queue_ids[j]))
+                       /*
+                        * Retry if insufficient credits
+                        */
+                       do {
+                               ret = dlb2_event_enqueue_prep(ev_port,
+                                                             qm_port,
+                                                             ev,
+                                                             &sched_types[j],
+                                                             &queue_ids[j]);
+                       } while ((ret == -ENOSPC) && (retries-- > 0));
+
+                       if (ret != 0)
                                break;
                }
 

diff --git a/drivers/event/dlb2/dlb2_priv.h b/drivers/event/dlb2/dlb2_priv.h
index 3e47e4776bc539181e5a7537e6ec9b8e1538b976..4a06d649abccdb266281940f46d4bb708b825d58 100644 (file)
--- a/drivers/event/dlb2/dlb2_priv.h
+++ b/drivers/event/dlb2/dlb2_priv.h
@@ -114,7 +114,7 @@
 
 #define DLB2_NUM_QES_PER_CACHE_LINE 4
 
-#define DLB2_MAX_ENQUEUE_DEPTH 64
+#define DLB2_MAX_ENQUEUE_DEPTH 32
 #define DLB2_MIN_ENQUEUE_DEPTH 4
 
 #define DLB2_NAME_SIZE 64
@@ -519,6 +519,7 @@ struct dlb2_eventdev_port {
         */
        uint16_t outstanding_releases;
        uint16_t inflight_max; /* app requested max inflights for this port */
+       int enq_retries; /* Number of attempts before ret ENOSPC */
        /* setup_done is set when the event port is setup */
        bool setup_done;
        /* enq_configured is set when the qm port is created */