event/sw: support device stop flush callback

This commit also adds a flush callback test to the sw eventdev's selftest
suite.

Signed-off-by: Gage Eads <gage.eads@intel.com>
Acked-by: Harry van Haaren <harry.van.haaren@intel.com>

diff --git a/drivers/event/sw/sw_evdev.c b/drivers/event/sw/sw_evdev.c
index e122d84..a6bb913 100644 (file)
--- a/drivers/event/sw/sw_evdev.c
+++ b/drivers/event/sw/sw_evdev.c
@@ -361,9 +361,99 @@ sw_init_qid_iqs(struct sw_evdev *sw)
        }
 }
 
+static int
+sw_qids_empty(struct sw_evdev *sw)
+{
+       unsigned int i, j;
+
+       for (i = 0; i < sw->qid_count; i++) {
+               for (j = 0; j < SW_IQS_MAX; j++) {
+                       if (iq_count(&sw->qids[i].iq[j]))
+                               return 0;
+               }
+       }
+
+       return 1;
+}
+
+static int
+sw_ports_empty(struct sw_evdev *sw)
+{
+       unsigned int i;
+
+       for (i = 0; i < sw->port_count; i++) {
+               if ((rte_event_ring_count(sw->ports[i].rx_worker_ring)) ||
+                    rte_event_ring_count(sw->ports[i].cq_worker_ring))
+                       return 0;
+       }
+
+       return 1;
+}
+
+static void
+sw_drain_ports(struct rte_eventdev *dev)
+{
+       struct sw_evdev *sw = sw_pmd_priv(dev);
+       eventdev_stop_flush_t flush;
+       unsigned int i;
+       uint8_t dev_id;
+       void *arg;
+
+       flush = dev->dev_ops->dev_stop_flush;
+       dev_id = dev->data->dev_id;
+       arg = dev->data->dev_stop_flush_arg;
+
+       for (i = 0; i < sw->port_count; i++) {
+               struct rte_event ev;
+
+               while (rte_event_dequeue_burst(dev_id, i, &ev, 1, 0)) {
+                       if (flush)
+                               flush(dev_id, ev, arg);
+
+                       ev.op = RTE_EVENT_OP_RELEASE;
+                       rte_event_enqueue_burst(dev_id, i, &ev, 1);
+               }
+       }
+}
+
+static void
+sw_drain_queue(struct rte_eventdev *dev, struct sw_iq *iq)
+{
+       struct sw_evdev *sw = sw_pmd_priv(dev);
+       eventdev_stop_flush_t flush;
+       uint8_t dev_id;
+       void *arg;
+
+       flush = dev->dev_ops->dev_stop_flush;
+       dev_id = dev->data->dev_id;
+       arg = dev->data->dev_stop_flush_arg;
+
+       while (iq_count(iq) > 0) {
+               struct rte_event ev;
+
+               iq_dequeue_burst(sw, iq, &ev, 1);
+
+               if (flush)
+                       flush(dev_id, ev, arg);
+       }
+}
+
+static void
+sw_drain_queues(struct rte_eventdev *dev)
+{
+       struct sw_evdev *sw = sw_pmd_priv(dev);
+       unsigned int i, j;
+
+       for (i = 0; i < sw->qid_count; i++) {
+               for (j = 0; j < SW_IQS_MAX; j++)
+                       sw_drain_queue(dev, &sw->qids[i].iq[j]);
+       }
+}
+
 static void
-sw_clean_qid_iqs(struct sw_evdev *sw)
+sw_clean_qid_iqs(struct rte_eventdev *dev)
 {
+       struct sw_evdev *sw = sw_pmd_priv(dev);
        int i, j;
 
        /* Release the IQ memory of all configured qids */
@@ -729,10 +819,30 @@ static void
 sw_stop(struct rte_eventdev *dev)
 {
        struct sw_evdev *sw = sw_pmd_priv(dev);
-       sw_clean_qid_iqs(sw);
+       int32_t runstate;
+
+       /* Stop the scheduler if it's running */
+       runstate = rte_service_runstate_get(sw->service_id);
+       if (runstate == 1)
+               rte_service_runstate_set(sw->service_id, 0);
+
+       while (rte_service_may_be_active(sw->service_id))
+               rte_pause();
+
+       /* Flush all events out of the device */
+       while (!(sw_qids_empty(sw) && sw_ports_empty(sw))) {
+               sw_event_schedule(dev);
+               sw_drain_ports(dev);
+               sw_drain_queues(dev);
+       }
+
+       sw_clean_qid_iqs(dev);
        sw_xstats_uninit(sw);
        sw->started = 0;
        rte_smp_wmb();
+
+       if (runstate == 1)
+               rte_service_runstate_set(sw->service_id, 1);
 }
 
 static int

diff --git a/drivers/event/sw/sw_evdev_selftest.c b/drivers/event/sw/sw_evdev_selftest.c
index 78d30e0..c40912d 100644 (file)
--- a/drivers/event/sw/sw_evdev_selftest.c
+++ b/drivers/event/sw/sw_evdev_selftest.c
@@ -28,6 +28,7 @@
 #define MAX_PORTS 16
 #define MAX_QIDS 16
 #define NUM_PACKETS (1<<18)
+#define DEQUEUE_DEPTH 128
 
 static int evdev;
 
@@ -147,7 +148,7 @@ init(struct test *t, int nb_queues, int nb_ports)
                        .nb_event_ports = nb_ports,
                        .nb_event_queue_flows = 1024,
                        .nb_events_limit = 4096,
-                       .nb_event_port_dequeue_depth = 128,
+                       .nb_event_port_dequeue_depth = DEQUEUE_DEPTH,
                        .nb_event_port_enqueue_depth = 128,
        };
        int ret;
@@ -2807,6 +2808,78 @@ err:
        return -1;
 }
 
+static void
+flush(uint8_t dev_id __rte_unused, struct rte_event event, void *arg)
+{
+       *((uint8_t *) arg) += (event.u64 == 0xCA11BACC) ? 1 : 0;
+}
+
+static int
+dev_stop_flush(struct test *t) /* test to check we can properly flush events */
+{
+       const struct rte_event new_ev = {
+               .op = RTE_EVENT_OP_NEW,
+               .u64 = 0xCA11BACC,
+               .queue_id = 0
+       };
+       struct rte_event ev = new_ev;
+       uint8_t count = 0;
+       int i;
+
+       if (init(t, 1, 1) < 0 ||
+           create_ports(t, 1) < 0 ||
+           create_atomic_qids(t, 1) < 0) {
+               printf("%d: Error initializing device\n", __LINE__);
+               return -1;
+       }
+
+       /* Link the queue so *_start() doesn't error out */
+       if (rte_event_port_link(evdev, t->port[0], NULL, NULL, 0) != 1) {
+               printf("%d: Error linking queue to port\n", __LINE__);
+               goto err;
+       }
+
+       if (rte_event_dev_start(evdev) < 0) {
+               printf("%d: Error with start call\n", __LINE__);
+               goto err;
+       }
+
+       for (i = 0; i < DEQUEUE_DEPTH + 1; i++) {
+               if (rte_event_enqueue_burst(evdev, t->port[0], &ev, 1) != 1) {
+                       printf("%d: Error enqueuing events\n", __LINE__);
+                       goto err;
+               }
+       }
+
+       /* Schedule the events from the port to the IQ. At least one event
+        * should be remaining in the queue.
+        */
+       rte_service_run_iter_on_app_lcore(t->service_id, 1);
+
+       if (rte_event_dev_stop_flush_callback_register(evdev, flush, &count)) {
+               printf("%d: Error installing the flush callback\n", __LINE__);
+               goto err;
+       }
+
+       cleanup(t);
+
+       if (count == 0) {
+               printf("%d: Error executing the flush callback\n", __LINE__);
+               goto err;
+       }
+
+       if (rte_event_dev_stop_flush_callback_register(evdev, NULL, NULL)) {
+               printf("%d: Error uninstalling the flush callback\n", __LINE__);
+               goto err;
+       }
+
+       return 0;
+err:
+       rte_event_dev_dump(evdev, stdout);
+       cleanup(t);
+       return -1;
+}
+
 static int
 worker_loopback_worker_fn(void *arg)
 {
@@ -3211,6 +3284,12 @@ test_sw_eventdev(void)
                printf("ERROR - Head-of-line-blocking test FAILED.\n");
                goto test_fail;
        }
+       printf("*** Running Stop Flush test...\n");
+       ret = dev_stop_flush(t);
+       if (ret != 0) {
+               printf("ERROR - Stop Flush test FAILED.\n");
+               goto test_fail;
+       }
        if (rte_lcore_count() >= 3) {
                printf("*** Running Worker loopback test...\n");
                ret = worker_loopback(t, 0);