[dpdk-dev] [PATCH 0/2] Improve service stop support

[dpdk-dev] [PATCH 0/2] Improve service stop support

[Gage Eads]

Existing service functions allow us to stop a service, but doing so doesn't
guarantee that the service has finished running on a service core. This
patch set introduces a function, rte_service_may_be_active(), to check
whether a stopped service is truly stopped.

This is needed for flows that modify a resource that the service is
using; for example when stopping an eventdev, any event adapters and/or
scheduler service need to be quiesced first.

This patch set also adds support for the event sw PMD's device stop flush
callback, which relies on this new mechanism to ensure that the
scheduler service is no longer active.

Gage Eads (2):
  service: add mechanism for quiescing a service
  event/sw: support device stop flush callback

 drivers/event/sw/sw_evdev.c                 | 114 +++++++++++++++++++++++++++-
 drivers/event/sw/sw_evdev_selftest.c        |  81 +++++++++++++++++++-
 lib/librte_eal/common/include/rte_service.h |  16 ++++
 lib/librte_eal/common/rte_service.c         |  31 +++++++-
 lib/librte_eal/rte_eal_version.map          |   1 +
 test/test/test_service_cores.c              |  43 +++++++++++
 6 files changed, 279 insertions(+), 7 deletions(-)

-- 
2.13.6

[dpdk-dev] [PATCH 1/2] service: add mechanism for quiescing a service

[Gage Eads]

Existing service functions allow us to stop a service, but doing so doesn't
guarantee that the service has finished running on a service core. This
commit introduces rte_service_may_be_active(), which returns whether the
service may be executing on one or more lcores currently, or definitely is
not.

The service core layer supports this function by setting a flag when
a service core is going to execute a service, and unsetting the flag when
the core is no longer able to run the service (its runstate becomes stopped
or the lcore is no longer mapped).

With this new function, applications can set a service's runstate to
stopped, then poll rte_service_may_be_active() until it returns false. At
that point, the service is quiesced.

Signed-off-by: Gage Eads <gage.eads@intel.com>
---
 lib/librte_eal/common/include/rte_service.h | 16 +++++++++++
 lib/librte_eal/common/rte_service.c         | 31 ++++++++++++++++++---
 lib/librte_eal/rte_eal_version.map          |  1 +
 test/test/test_service_cores.c              | 43 +++++++++++++++++++++++++++++
 4 files changed, 87 insertions(+), 4 deletions(-)

diff --git a/lib/librte_eal/common/include/rte_service.h b/lib/librte_eal/common/include/rte_service.h
index aea4d91b9..27b2dab7c 100644
--- a/lib/librte_eal/common/include/rte_service.h
+++ b/lib/librte_eal/common/include/rte_service.h
@@ -162,6 +162,22 @@ int32_t rte_service_runstate_set(uint32_t id, uint32_t runstate);
 int32_t rte_service_runstate_get(uint32_t id);
 
 /**
+ * This function returns whether the service may be currently executing on
+ * at least one lcore, or definitely is not. This function can be used to
+ * determine if, after setting the service runstate to stopped, the service
+ * is still executing an a service lcore.
+ *
+ * Care must be taken if calling this function when the service runstate is
+ * running, since the result of this function may be incorrect by the time the
+ * function returns due to service cores running in parallel.
+ *
+ * @retval 1 Service may be running on one or more lcores
+ * @retval 0 Service is not running on any lcore
+ * @retval -EINVAL Invalid service id
+ */
+int32_t rte_service_may_be_active(uint32_t id);
+
+/**
  * Enable or disable the check for a service-core being mapped to the service.
  * An application can disable the check when takes the responsibility to run a
  * service itself using *rte_service_run_iter_on_app_lcore*.
diff --git a/lib/librte_eal/common/rte_service.c b/lib/librte_eal/common/rte_service.c
index 73507aacb..d6c4c6039 100644
--- a/lib/librte_eal/common/rte_service.c
+++ b/lib/librte_eal/common/rte_service.c
@@ -52,6 +52,7 @@ struct rte_service_spec_impl {
 	rte_atomic32_t num_mapped_cores;
 	uint64_t calls;
 	uint64_t cycles_spent;
+	uint8_t active_on_lcore[RTE_MAX_LCORE];
 } __rte_cache_aligned;
 
 /* the internal values of a service core */
@@ -347,15 +348,19 @@ rte_service_runner_do_callback(struct rte_service_spec_impl *s,
 
 
 static inline int32_t
-service_run(uint32_t i, struct core_state *cs, uint64_t service_mask)
+service_run(uint32_t i, int lcore, struct core_state *cs, uint64_t service_mask)
 {
 	if (!service_valid(i))
 		return -EINVAL;
 	struct rte_service_spec_impl *s = &rte_services[i];
 	if (s->comp_runstate != RUNSTATE_RUNNING ||
 			s->app_runstate != RUNSTATE_RUNNING ||
-			!(service_mask & (UINT64_C(1) << i)))
+			!(service_mask & (UINT64_C(1) << i))) {
+		s->active_on_lcore[lcore] = 0;
 		return -ENOEXEC;
+	}
+
+	s->active_on_lcore[lcore] = 1;
 
 	/* check do we need cmpset, if MT safe or <= 1 core
 	 * mapped, atomic ops are not required.
@@ -374,6 +379,24 @@ service_run(uint32_t i, struct core_state *cs, uint64_t service_mask)
 	return 0;
 }
 
+int32_t rte_service_may_be_active(uint32_t id)
+{
+	uint32_t ids[RTE_MAX_LCORE] = {0};
+	struct rte_service_spec_impl *s = &rte_services[id];
+	int32_t lcore_count = rte_service_lcore_list(ids, RTE_MAX_LCORE);
+	int i;
+
+	if (!service_valid(id))
+		return -EINVAL;
+
+	for (i = 0; i < lcore_count; i++) {
+		if (s->active_on_lcore[ids[i]])
+			return 1;
+	}
+
+	return 0;
+}
+
 int32_t rte_service_run_iter_on_app_lcore(uint32_t id,
 		uint32_t serialize_mt_unsafe)
 {
@@ -398,7 +421,7 @@ int32_t rte_service_run_iter_on_app_lcore(uint32_t id,
 		return -EBUSY;
 	}
 
-	int ret = service_run(id, cs, UINT64_MAX);
+	int ret = service_run(id, rte_lcore_id(), cs, UINT64_MAX);
 
 	if (serialize_mt_unsafe)
 		rte_atomic32_dec(&s->num_mapped_cores);
@@ -419,7 +442,7 @@ rte_service_runner_func(void *arg)
 
 		for (i = 0; i < RTE_SERVICE_NUM_MAX; i++) {
 			/* return value ignored as no change to code flow */
-			service_run(i, cs, service_mask);
+			service_run(i, lcore, cs, service_mask);
 		}
 
 		rte_smp_rmb();
diff --git a/lib/librte_eal/rte_eal_version.map b/lib/librte_eal/rte_eal_version.map
index f7dd0e7bc..6f38b02dc 100644
--- a/lib/librte_eal/rte_eal_version.map
+++ b/lib/librte_eal/rte_eal_version.map
@@ -238,6 +238,7 @@ DPDK_18.05 {
 	rte_service_set_runstate_mapped_check;
 	rte_service_set_stats_enable;
 	rte_service_start_with_defaults;
+	rte_service_may_be_active;
 
 } DPDK_18.02;
 
diff --git a/test/test/test_service_cores.c b/test/test/test_service_cores.c
index 86b4073d2..e6973d642 100644
--- a/test/test/test_service_cores.c
+++ b/test/test/test_service_cores.c
@@ -771,6 +771,48 @@ service_lcore_start_stop(void)
 	return unregister_all();
 }
 
+/* stop a service and wait for it to become inactive */
+static int
+service_may_be_active(void)
+{
+	const uint32_t sid = 0;
+	int i;
+
+	/* expected failure cases */
+	TEST_ASSERT_EQUAL(-EINVAL, rte_service_may_be_active(10000),
+			"Invalid service may be active check did not fail");
+
+	/* start the service */
+	TEST_ASSERT_EQUAL(0, rte_service_runstate_set(sid, 1),
+			"Starting valid service failed");
+	TEST_ASSERT_EQUAL(0, rte_service_lcore_add(slcore_id),
+			"Add service core failed when not in use before");
+	TEST_ASSERT_EQUAL(0, rte_service_map_lcore_set(sid, slcore_id, 1),
+			"Enabling valid service on valid core failed");
+	TEST_ASSERT_EQUAL(0, rte_service_lcore_start(slcore_id),
+			"Service core start after add failed");
+
+	/* ensures core really is running the service function */
+	TEST_ASSERT_EQUAL(1, service_lcore_running_check(),
+			"Service core expected to poll service but it didn't");
+
+	/* stop the service */
+	TEST_ASSERT_EQUAL(0, rte_service_runstate_set(sid, 0),
+			"Error: Service stop returned non-zero");
+
+	/* give the service 100ms to stop running */
+	for (i = 0; i < 100; i++) {
+		if (!rte_service_may_be_active(sid))
+			break;
+		rte_delay_ms(SERVICE_DELAY);
+	}
+
+	TEST_ASSERT_EQUAL(0, rte_service_may_be_active(sid),
+			  "Error: Service not stopped after 100ms");
+
+	return unregister_all();
+}
+
 static struct unit_test_suite service_tests  = {
 	.suite_name = "service core test suite",
 	.setup = testsuite_setup,
@@ -790,6 +832,7 @@ static struct unit_test_suite service_tests  = {
 		TEST_CASE_ST(dummy_register, NULL, service_mt_safe_poll),
 		TEST_CASE_ST(dummy_register, NULL, service_app_lcore_mt_safe),
 		TEST_CASE_ST(dummy_register, NULL, service_app_lcore_mt_unsafe),
+		TEST_CASE_ST(dummy_register, NULL, service_may_be_active),
 		TEST_CASES_END() /**< NULL terminate unit test array */
 	}
 };
-- 
2.13.6

Re: [dpdk-dev] [PATCH 1/2] service: add mechanism for quiescing a service

[Van Haaren, Harry]

> From: Eads, Gage
> Sent: Thursday, May 31, 2018 2:56 PM
> To: dev@dpdk.org
> Cc: jerin.jacob@caviumnetworks.com; Van Haaren, Harry
> <harry.van.haaren@intel.com>; Richardson, Bruce <bruce.richardson@intel.com>;
> Rao, Nikhil <nikhil.rao@intel.com>; Carrillo, Erik G
> <erik.g.carrillo@intel.com>; Gujjar, Abhinandan S
> <abhinandan.gujjar@intel.com>
> Subject: [PATCH 1/2] service: add mechanism for quiescing a service
> 
> Existing service functions allow us to stop a service, but doing so doesn't
> guarantee that the service has finished running on a service core. This
> commit introduces rte_service_may_be_active(), which returns whether the
> service may be executing on one or more lcores currently, or definitely is
> not.
> 
> The service core layer supports this function by setting a flag when
> a service core is going to execute a service, and unsetting the flag when
> the core is no longer able to run the service (its runstate becomes stopped
> or the lcore is no longer mapped).
> 
> With this new function, applications can set a service's runstate to
> stopped, then poll rte_service_may_be_active() until it returns false. At
> that point, the service is quiesced.
>
> Signed-off-by: Gage Eads <gage.eads@intel.com>


Good approach to the problem.

One nit below, version map should be alphabetical;

Acked-by: Harry van Haaren <harry.van.haaren@intel.com>


<snip>

> diff --git a/lib/librte_eal/rte_eal_version.map
> b/lib/librte_eal/rte_eal_version.map
> index f7dd0e7bc..6f38b02dc 100644
> --- a/lib/librte_eal/rte_eal_version.map
> +++ b/lib/librte_eal/rte_eal_version.map
> @@ -238,6 +238,7 @@ DPDK_18.05 {
>  	rte_service_set_runstate_mapped_check;
>  	rte_service_set_stats_enable;
>  	rte_service_start_with_defaults;
> +	rte_service_may_be_active;
> 
>  } DPDK_18.02;

Version map to be alphabetically ordered.

[dpdk-dev] [PATCH 2/2] event/sw: support device stop flush callback

[Gage Eads]

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

Signed-off-by: Gage Eads <gage.eads@intel.com>
---
 drivers/event/sw/sw_evdev.c          | 114 ++++++++++++++++++++++++++++++++++-
 drivers/event/sw/sw_evdev_selftest.c |  81 ++++++++++++++++++++++++-
 2 files changed, 192 insertions(+), 3 deletions(-)

diff --git a/drivers/event/sw/sw_evdev.c b/drivers/event/sw/sw_evdev.c
index 10f0e1ad4..95a6f1fda 100644
--- 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);
+	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 78d30e07a..c40912db5 100644
--- 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 @@ holb(struct test *t) /* test to check we avoid basic head-of-line blocking */
 	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);
-- 
2.13.6

Re: [dpdk-dev] [PATCH 2/2] event/sw: support device stop flush callback

[Van Haaren, Harry]

> From: Eads, Gage
> Sent: Thursday, May 31, 2018 2:56 PM
> To: dev@dpdk.org
> Cc: jerin.jacob@caviumnetworks.com; Van Haaren, Harry
> <harry.van.haaren@intel.com>; Richardson, Bruce <bruce.richardson@intel.com>;
> Rao, Nikhil <nikhil.rao@intel.com>; Carrillo, Erik G
> <erik.g.carrillo@intel.com>; Gujjar, Abhinandan S
> <abhinandan.gujjar@intel.com>
> Subject: [PATCH 2/2] 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>


One compile warning below, with that fixed;

Acked-by: Harry van Haaren <harry.van.haaren@intel.com>


<snip>

> +static void
> +sw_drain_queues(struct rte_eventdev *dev)
> +{
> +	struct sw_evdev *sw = sw_pmd_priv(dev);
> +	int i, j;
> +
> +	for (i = 0; i < sw->qid_count; i++) {

I'm seeing an int vs unsigned comparison between i and sw->qid_count.

> +		for (j = 0; j < SW_IQS_MAX; j++)
> +			sw_drain_queue(dev, &sw->qids[i].iq[j]);
> +	}
> +}

<snip>

Re: [dpdk-dev] [PATCH 2/2] event/sw: support device stop flush callback

[Jerin Jacob]

-----Original Message-----
> Date: Thu, 14 Jun 2018 10:20:08 +0000
> From: "Van Haaren, Harry" <harry.van.haaren@intel.com>
> To: "Eads, Gage" <gage.eads@intel.com>, "dev@dpdk.org" <dev@dpdk.org>
> CC: "jerin.jacob@caviumnetworks.com" <jerin.jacob@caviumnetworks.com>,
>  "Richardson, Bruce" <bruce.richardson@intel.com>, "Rao, Nikhil"
>  <nikhil.rao@intel.com>, "Carrillo, Erik G" <erik.g.carrillo@intel.com>,
>  "Gujjar, Abhinandan S" <abhinandan.gujjar@intel.com>
> Subject: RE: [PATCH 2/2] event/sw: support device stop flush callback
> 
> 
> > From: Eads, Gage
> > Sent: Thursday, May 31, 2018 2:56 PM
> > To: dev@dpdk.org
> > Cc: jerin.jacob@caviumnetworks.com; Van Haaren, Harry
> > <harry.van.haaren@intel.com>; Richardson, Bruce <bruce.richardson@intel.com>;
> > Rao, Nikhil <nikhil.rao@intel.com>; Carrillo, Erik G
> > <erik.g.carrillo@intel.com>; Gujjar, Abhinandan S
> > <abhinandan.gujjar@intel.com>
> > Subject: [PATCH 2/2] 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>
> 
> 
> One compile warning below, with that fixed;
> 
> Acked-by: Harry van Haaren <harry.van.haaren@intel.com>
> 

Applied v2 series to dpdk-next-eventdev/master. Thanks.

[dpdk-dev] [PATCH v2 0/2] Improve service stop support

[Gage Eads]

Existing service functions allow us to stop a service, but doing so doesn't
guarantee that the service has finished running on a service core. This
patch set introduces a function, rte_service_may_be_active(), to check
whether a stopped service is truly stopped.

This is needed for flows that modify a resource that the service is
using; for example when stopping an eventdev, any event adapters and/or
scheduler service need to be quiesced first.

This patch set also adds support for the event sw PMD's device stop flush
callback, which relies on this new mechanism to ensure that the
scheduler service is no longer active.

v2:
- Move function to DPDK_18.08 block in rte_eal_version.map
- Fix signed vs. unsigned comparison compiler warning

Gage Eads (2):
  service: add mechanism for quiescing a service
  event/sw: support device stop flush callback

 drivers/event/sw/sw_evdev.c                 | 114 +++++++++++++++++++++++++++-
 drivers/event/sw/sw_evdev_selftest.c        |  81 +++++++++++++++++++-
 lib/librte_eal/common/include/rte_service.h |  16 ++++
 lib/librte_eal/common/rte_service.c         |  31 +++++++-
 lib/librte_eal/rte_eal_version.map          |   7 ++
 test/test/test_service_cores.c              |  43 +++++++++++
 6 files changed, 285 insertions(+), 7 deletions(-)

-- 
2.13.6

[dpdk-dev] [PATCH v2 1/2] service: add mechanism for quiescing a service

[Gage Eads]

Existing service functions allow us to stop a service, but doing so doesn't
guarantee that the service has finished running on a service core. This
commit introduces rte_service_may_be_active(), which returns whether the
service may be executing on one or more lcores currently, or definitely is
not.

The service core layer supports this function by setting a flag when
a service core is going to execute a service, and unsetting the flag when
the core is no longer able to run the service (its runstate becomes stopped
or the lcore is no longer mapped).

With this new function, applications can set a service's runstate to
stopped, then poll rte_service_may_be_active() until it returns false. At
that point, the service is quiesced.

Signed-off-by: Gage Eads <gage.eads@intel.com>
Acked-by: Harry van Haaren <harry.van.haaren@intel.com>
---
 lib/librte_eal/common/include/rte_service.h | 16 +++++++++++
 lib/librte_eal/common/rte_service.c         | 31 ++++++++++++++++++---
 lib/librte_eal/rte_eal_version.map          |  7 +++++
 test/test/test_service_cores.c              | 43 +++++++++++++++++++++++++++++
 4 files changed, 93 insertions(+), 4 deletions(-)

diff --git a/lib/librte_eal/common/include/rte_service.h b/lib/librte_eal/common/include/rte_service.h
index aea4d91b9..27b2dab7c 100644
--- a/lib/librte_eal/common/include/rte_service.h
+++ b/lib/librte_eal/common/include/rte_service.h
@@ -162,6 +162,22 @@ int32_t rte_service_runstate_set(uint32_t id, uint32_t runstate);
 int32_t rte_service_runstate_get(uint32_t id);
 
 /**
+ * This function returns whether the service may be currently executing on
+ * at least one lcore, or definitely is not. This function can be used to
+ * determine if, after setting the service runstate to stopped, the service
+ * is still executing an a service lcore.
+ *
+ * Care must be taken if calling this function when the service runstate is
+ * running, since the result of this function may be incorrect by the time the
+ * function returns due to service cores running in parallel.
+ *
+ * @retval 1 Service may be running on one or more lcores
+ * @retval 0 Service is not running on any lcore
+ * @retval -EINVAL Invalid service id
+ */
+int32_t rte_service_may_be_active(uint32_t id);
+
+/**
  * Enable or disable the check for a service-core being mapped to the service.
  * An application can disable the check when takes the responsibility to run a
  * service itself using *rte_service_run_iter_on_app_lcore*.
diff --git a/lib/librte_eal/common/rte_service.c b/lib/librte_eal/common/rte_service.c
index 73507aacb..d6c4c6039 100644
--- a/lib/librte_eal/common/rte_service.c
+++ b/lib/librte_eal/common/rte_service.c
@@ -52,6 +52,7 @@ struct rte_service_spec_impl {
 	rte_atomic32_t num_mapped_cores;
 	uint64_t calls;
 	uint64_t cycles_spent;
+	uint8_t active_on_lcore[RTE_MAX_LCORE];
 } __rte_cache_aligned;
 
 /* the internal values of a service core */
@@ -347,15 +348,19 @@ rte_service_runner_do_callback(struct rte_service_spec_impl *s,
 
 
 static inline int32_t
-service_run(uint32_t i, struct core_state *cs, uint64_t service_mask)
+service_run(uint32_t i, int lcore, struct core_state *cs, uint64_t service_mask)
 {
 	if (!service_valid(i))
 		return -EINVAL;
 	struct rte_service_spec_impl *s = &rte_services[i];
 	if (s->comp_runstate != RUNSTATE_RUNNING ||
 			s->app_runstate != RUNSTATE_RUNNING ||
-			!(service_mask & (UINT64_C(1) << i)))
+			!(service_mask & (UINT64_C(1) << i))) {
+		s->active_on_lcore[lcore] = 0;
 		return -ENOEXEC;
+	}
+
+	s->active_on_lcore[lcore] = 1;
 
 	/* check do we need cmpset, if MT safe or <= 1 core
 	 * mapped, atomic ops are not required.
@@ -374,6 +379,24 @@ service_run(uint32_t i, struct core_state *cs, uint64_t service_mask)
 	return 0;
 }
 
+int32_t rte_service_may_be_active(uint32_t id)
+{
+	uint32_t ids[RTE_MAX_LCORE] = {0};
+	struct rte_service_spec_impl *s = &rte_services[id];
+	int32_t lcore_count = rte_service_lcore_list(ids, RTE_MAX_LCORE);
+	int i;
+
+	if (!service_valid(id))
+		return -EINVAL;
+
+	for (i = 0; i < lcore_count; i++) {
+		if (s->active_on_lcore[ids[i]])
+			return 1;
+	}
+
+	return 0;
+}
+
 int32_t rte_service_run_iter_on_app_lcore(uint32_t id,
 		uint32_t serialize_mt_unsafe)
 {
@@ -398,7 +421,7 @@ int32_t rte_service_run_iter_on_app_lcore(uint32_t id,
 		return -EBUSY;
 	}
 
-	int ret = service_run(id, cs, UINT64_MAX);
+	int ret = service_run(id, rte_lcore_id(), cs, UINT64_MAX);
 
 	if (serialize_mt_unsafe)
 		rte_atomic32_dec(&s->num_mapped_cores);
@@ -419,7 +442,7 @@ rte_service_runner_func(void *arg)
 
 		for (i = 0; i < RTE_SERVICE_NUM_MAX; i++) {
 			/* return value ignored as no change to code flow */
-			service_run(i, cs, service_mask);
+			service_run(i, lcore, cs, service_mask);
 		}
 
 		rte_smp_rmb();
diff --git a/lib/librte_eal/rte_eal_version.map b/lib/librte_eal/rte_eal_version.map
index f7dd0e7bc..fb113b446 100644
--- a/lib/librte_eal/rte_eal_version.map
+++ b/lib/librte_eal/rte_eal_version.map
@@ -241,6 +241,13 @@ DPDK_18.05 {
 
 } DPDK_18.02;
 
+DPDK_18.08 {
+	global:
+
+	rte_service_may_be_active;
+
+} DPDK_18.05;
+
 EXPERIMENTAL {
 	global:
 
diff --git a/test/test/test_service_cores.c b/test/test/test_service_cores.c
index 86b4073d2..e6973d642 100644
--- a/test/test/test_service_cores.c
+++ b/test/test/test_service_cores.c
@@ -771,6 +771,48 @@ service_lcore_start_stop(void)
 	return unregister_all();
 }
 
+/* stop a service and wait for it to become inactive */
+static int
+service_may_be_active(void)
+{
+	const uint32_t sid = 0;
+	int i;
+
+	/* expected failure cases */
+	TEST_ASSERT_EQUAL(-EINVAL, rte_service_may_be_active(10000),
+			"Invalid service may be active check did not fail");
+
+	/* start the service */
+	TEST_ASSERT_EQUAL(0, rte_service_runstate_set(sid, 1),
+			"Starting valid service failed");
+	TEST_ASSERT_EQUAL(0, rte_service_lcore_add(slcore_id),
+			"Add service core failed when not in use before");
+	TEST_ASSERT_EQUAL(0, rte_service_map_lcore_set(sid, slcore_id, 1),
+			"Enabling valid service on valid core failed");
+	TEST_ASSERT_EQUAL(0, rte_service_lcore_start(slcore_id),
+			"Service core start after add failed");
+
+	/* ensures core really is running the service function */
+	TEST_ASSERT_EQUAL(1, service_lcore_running_check(),
+			"Service core expected to poll service but it didn't");
+
+	/* stop the service */
+	TEST_ASSERT_EQUAL(0, rte_service_runstate_set(sid, 0),
+			"Error: Service stop returned non-zero");
+
+	/* give the service 100ms to stop running */
+	for (i = 0; i < 100; i++) {
+		if (!rte_service_may_be_active(sid))
+			break;
+		rte_delay_ms(SERVICE_DELAY);
+	}
+
+	TEST_ASSERT_EQUAL(0, rte_service_may_be_active(sid),
+			  "Error: Service not stopped after 100ms");
+
+	return unregister_all();
+}
+
 static struct unit_test_suite service_tests  = {
 	.suite_name = "service core test suite",
 	.setup = testsuite_setup,
@@ -790,6 +832,7 @@ static struct unit_test_suite service_tests  = {
 		TEST_CASE_ST(dummy_register, NULL, service_mt_safe_poll),
 		TEST_CASE_ST(dummy_register, NULL, service_app_lcore_mt_safe),
 		TEST_CASE_ST(dummy_register, NULL, service_app_lcore_mt_unsafe),
+		TEST_CASE_ST(dummy_register, NULL, service_may_be_active),
 		TEST_CASES_END() /**< NULL terminate unit test array */
 	}
 };
-- 
2.13.6

Re: [dpdk-dev] [PATCH v2 1/2] service: add mechanism for quiescing a service

[Thomas Monjalon]

14/06/2018 15:51, Gage Eads:
> --- a/lib/librte_eal/common/include/rte_service.h
> +++ b/lib/librte_eal/common/include/rte_service.h
> @@ -162,6 +162,22 @@ int32_t rte_service_runstate_set(uint32_t id, uint32_t runstate);
>  int32_t rte_service_runstate_get(uint32_t id);
>  
>  /**
> + * This function returns whether the service may be currently executing on
> + * at least one lcore, or definitely is not. This function can be used to
> + * determine if, after setting the service runstate to stopped, the service
> + * is still executing an a service lcore.

Typo: "an a"

> --- a/lib/librte_eal/rte_eal_version.map
> +++ b/lib/librte_eal/rte_eal_version.map
> +DPDK_18.08 {
> +	global:
> +
> +	rte_service_may_be_active;
> +
> +} DPDK_18.05;

Why it is not experimental?

Re: [dpdk-dev] [PATCH v2 1/2] service: add mechanism for quiescing a service

[Eads, Gage]

> -----Original Message-----
> From: Thomas Monjalon [mailto:thomas@monjalon.net]
> Sent: Monday, June 18, 2018 5:14 PM
> To: Eads, Gage <gage.eads@intel.com>
> Cc: dev@dpdk.org; jerin.jacob@caviumnetworks.com; Van Haaren, Harry
> <harry.van.haaren@intel.com>; Richardson, Bruce
> <bruce.richardson@intel.com>; Rao, Nikhil <nikhil.rao@intel.com>; Carrillo, Erik
> G <erik.g.carrillo@intel.com>; Gujjar, Abhinandan S
> <abhinandan.gujjar@intel.com>
> Subject: Re: [dpdk-dev] [PATCH v2 1/2] service: add mechanism for quiescing a
> service
> 
> 14/06/2018 15:51, Gage Eads:
> > --- a/lib/librte_eal/common/include/rte_service.h
> > +++ b/lib/librte_eal/common/include/rte_service.h
> > @@ -162,6 +162,22 @@ int32_t rte_service_runstate_set(uint32_t id,
> > uint32_t runstate);  int32_t rte_service_runstate_get(uint32_t id);
> >
> >  /**
> > + * This function returns whether the service may be currently
> > + executing on
> > + * at least one lcore, or definitely is not. This function can be
> > + used to
> > + * determine if, after setting the service runstate to stopped, the
> > + service
> > + * is still executing an a service lcore.
> 
> Typo: "an a"

Will fix.

> 
> > --- a/lib/librte_eal/rte_eal_version.map
> > +++ b/lib/librte_eal/rte_eal_version.map
> > +DPDK_18.08 {
> > +	global:
> > +
> > +	rte_service_may_be_active;
> > +
> > +} DPDK_18.05;
> 
> Why it is not experimental?
> 

It should be -- my mistake.

Thanks,
Gage

[dpdk-dev] [PATCH v2 2/2] event/sw: support device stop flush callback

[Gage Eads]

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>
---
 drivers/event/sw/sw_evdev.c          | 114 ++++++++++++++++++++++++++++++++++-
 drivers/event/sw/sw_evdev_selftest.c |  81 ++++++++++++++++++++++++-
 2 files changed, 192 insertions(+), 3 deletions(-)

diff --git a/drivers/event/sw/sw_evdev.c b/drivers/event/sw/sw_evdev.c
index 10f0e1ad4..331c518ff 100644
--- 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 78d30e07a..c40912db5 100644
--- 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 @@ holb(struct test *t) /* test to check we avoid basic head-of-line blocking */
 	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);
-- 
2.13.6

[dpdk-dev] [PATCH v3 0/2] Improve service stop support

[Gage Eads]

Existing service functions allow us to stop a service, but doing so doesn't
guarantee that the service has finished running on a service core. This
patch set introduces a function, rte_service_may_be_active(), to check
whether a stopped service is truly stopped.

This is needed for flows that modify a resource that the service is
using; for example when stopping an eventdev, any event adapters and/or
scheduler service need to be quiesced first.

This patch set also adds support for the event sw PMD's device stop flush
callback, which relies on this new mechanism to ensure that the
scheduler service is no longer active.

v2:
- Move function to DPDK_18.08 block in rte_eal_version.map
- Fix signed vs. unsigned comparison compiler warning

v3:
- Move function to EXPERIMENTAL block and add experimental tags
- Fix typo in function documentation

Gage Eads (2):
  service: add mechanism for quiescing a service
  event/sw: support device stop flush callback

 drivers/event/sw/sw_evdev.c                 | 114 +++++++++++++++++++++++++++-
 drivers/event/sw/sw_evdev_selftest.c        |  81 +++++++++++++++++++-
 lib/librte_eal/common/include/rte_service.h |  20 +++++
 lib/librte_eal/common/rte_service.c         |  32 +++++++-
 lib/librte_eal/rte_eal_version.map          |   1 +
 test/test/test_service_cores.c              |  43 +++++++++++
 6 files changed, 284 insertions(+), 7 deletions(-)

-- 
2.13.6

[dpdk-dev] [PATCH v3 1/2] service: add mechanism for quiescing a service

[Gage Eads]

Existing service functions allow us to stop a service, but doing so doesn't
guarantee that the service has finished running on a service core. This
commit introduces rte_service_may_be_active(), which returns whether the
service may be executing on one or more lcores currently, or definitely is
not.

The service core layer supports this function by setting a flag when
a service core is going to execute a service, and unsetting the flag when
the core is no longer able to run the service (its runstate becomes stopped
or the lcore is no longer mapped).

With this new function, applications can set a service's runstate to
stopped, then poll rte_service_may_be_active() until it returns false. At
that point, the service is quiesced.

Signed-off-by: Gage Eads <gage.eads@intel.com>
Acked-by: Harry van Haaren <harry.van.haaren@intel.com>
---
 lib/librte_eal/common/include/rte_service.h | 20 ++++++++++++++
 lib/librte_eal/common/rte_service.c         | 32 ++++++++++++++++++---
 lib/librte_eal/rte_eal_version.map          |  1 +
 test/test/test_service_cores.c              | 43 +++++++++++++++++++++++++++++
 4 files changed, 92 insertions(+), 4 deletions(-)

diff --git a/lib/librte_eal/common/include/rte_service.h b/lib/librte_eal/common/include/rte_service.h
index aea4d91b9..20f713b19 100644
--- a/lib/librte_eal/common/include/rte_service.h
+++ b/lib/librte_eal/common/include/rte_service.h
@@ -162,6 +162,26 @@ int32_t rte_service_runstate_set(uint32_t id, uint32_t runstate);
 int32_t rte_service_runstate_get(uint32_t id);
 
 /**
+ * @warning
+ * @b EXPERIMENTAL: this API may change, or be removed, without prior notice
+ *
+ * This function returns whether the service may be currently executing on
+ * at least one lcore, or definitely is not. This function can be used to
+ * determine if, after setting the service runstate to stopped, the service
+ * is still executing a service lcore.
+ *
+ * Care must be taken if calling this function when the service runstate is
+ * running, since the result of this function may be incorrect by the time the
+ * function returns due to service cores running in parallel.
+ *
+ * @retval 1 Service may be running on one or more lcores
+ * @retval 0 Service is not running on any lcore
+ * @retval -EINVAL Invalid service id
+ */
+int32_t __rte_experimental
+rte_service_may_be_active(uint32_t id);
+
+/**
  * Enable or disable the check for a service-core being mapped to the service.
  * An application can disable the check when takes the responsibility to run a
  * service itself using *rte_service_run_iter_on_app_lcore*.
diff --git a/lib/librte_eal/common/rte_service.c b/lib/librte_eal/common/rte_service.c
index 73507aacb..af9a660d4 100644
--- a/lib/librte_eal/common/rte_service.c
+++ b/lib/librte_eal/common/rte_service.c
@@ -52,6 +52,7 @@ struct rte_service_spec_impl {
 	rte_atomic32_t num_mapped_cores;
 	uint64_t calls;
 	uint64_t cycles_spent;
+	uint8_t active_on_lcore[RTE_MAX_LCORE];
 } __rte_cache_aligned;
 
 /* the internal values of a service core */
@@ -347,15 +348,19 @@ rte_service_runner_do_callback(struct rte_service_spec_impl *s,
 
 
 static inline int32_t
-service_run(uint32_t i, struct core_state *cs, uint64_t service_mask)
+service_run(uint32_t i, int lcore, struct core_state *cs, uint64_t service_mask)
 {
 	if (!service_valid(i))
 		return -EINVAL;
 	struct rte_service_spec_impl *s = &rte_services[i];
 	if (s->comp_runstate != RUNSTATE_RUNNING ||
 			s->app_runstate != RUNSTATE_RUNNING ||
-			!(service_mask & (UINT64_C(1) << i)))
+			!(service_mask & (UINT64_C(1) << i))) {
+		s->active_on_lcore[lcore] = 0;
 		return -ENOEXEC;
+	}
+
+	s->active_on_lcore[lcore] = 1;
 
 	/* check do we need cmpset, if MT safe or <= 1 core
 	 * mapped, atomic ops are not required.
@@ -374,6 +379,25 @@ service_run(uint32_t i, struct core_state *cs, uint64_t service_mask)
 	return 0;
 }
 
+int32_t __rte_experimental
+rte_service_may_be_active(uint32_t id)
+{
+	uint32_t ids[RTE_MAX_LCORE] = {0};
+	struct rte_service_spec_impl *s = &rte_services[id];
+	int32_t lcore_count = rte_service_lcore_list(ids, RTE_MAX_LCORE);
+	int i;
+
+	if (!service_valid(id))
+		return -EINVAL;
+
+	for (i = 0; i < lcore_count; i++) {
+		if (s->active_on_lcore[ids[i]])
+			return 1;
+	}
+
+	return 0;
+}
+
 int32_t rte_service_run_iter_on_app_lcore(uint32_t id,
 		uint32_t serialize_mt_unsafe)
 {
@@ -398,7 +422,7 @@ int32_t rte_service_run_iter_on_app_lcore(uint32_t id,
 		return -EBUSY;
 	}
 
-	int ret = service_run(id, cs, UINT64_MAX);
+	int ret = service_run(id, rte_lcore_id(), cs, UINT64_MAX);
 
 	if (serialize_mt_unsafe)
 		rte_atomic32_dec(&s->num_mapped_cores);
@@ -419,7 +443,7 @@ rte_service_runner_func(void *arg)
 
 		for (i = 0; i < RTE_SERVICE_NUM_MAX; i++) {
 			/* return value ignored as no change to code flow */
-			service_run(i, cs, service_mask);
+			service_run(i, lcore, cs, service_mask);
 		}
 
 		rte_smp_rmb();
diff --git a/lib/librte_eal/rte_eal_version.map b/lib/librte_eal/rte_eal_version.map
index f7dd0e7bc..7a9ccaa0a 100644
--- a/lib/librte_eal/rte_eal_version.map
+++ b/lib/librte_eal/rte_eal_version.map
@@ -294,6 +294,7 @@ EXPERIMENTAL {
 	rte_mp_request_sync;
 	rte_mp_request_async;
 	rte_mp_sendmsg;
+	rte_service_may_be_active;
 	rte_socket_count;
 	rte_socket_id_by_idx;
 	rte_vfio_dma_map;
diff --git a/test/test/test_service_cores.c b/test/test/test_service_cores.c
index 86b4073d2..e6973d642 100644
--- a/test/test/test_service_cores.c
+++ b/test/test/test_service_cores.c
@@ -771,6 +771,48 @@ service_lcore_start_stop(void)
 	return unregister_all();
 }
 
+/* stop a service and wait for it to become inactive */
+static int
+service_may_be_active(void)
+{
+	const uint32_t sid = 0;
+	int i;
+
+	/* expected failure cases */
+	TEST_ASSERT_EQUAL(-EINVAL, rte_service_may_be_active(10000),
+			"Invalid service may be active check did not fail");
+
+	/* start the service */
+	TEST_ASSERT_EQUAL(0, rte_service_runstate_set(sid, 1),
+			"Starting valid service failed");
+	TEST_ASSERT_EQUAL(0, rte_service_lcore_add(slcore_id),
+			"Add service core failed when not in use before");
+	TEST_ASSERT_EQUAL(0, rte_service_map_lcore_set(sid, slcore_id, 1),
+			"Enabling valid service on valid core failed");
+	TEST_ASSERT_EQUAL(0, rte_service_lcore_start(slcore_id),
+			"Service core start after add failed");
+
+	/* ensures core really is running the service function */
+	TEST_ASSERT_EQUAL(1, service_lcore_running_check(),
+			"Service core expected to poll service but it didn't");
+
+	/* stop the service */
+	TEST_ASSERT_EQUAL(0, rte_service_runstate_set(sid, 0),
+			"Error: Service stop returned non-zero");
+
+	/* give the service 100ms to stop running */
+	for (i = 0; i < 100; i++) {
+		if (!rte_service_may_be_active(sid))
+			break;
+		rte_delay_ms(SERVICE_DELAY);
+	}
+
+	TEST_ASSERT_EQUAL(0, rte_service_may_be_active(sid),
+			  "Error: Service not stopped after 100ms");
+
+	return unregister_all();
+}
+
 static struct unit_test_suite service_tests  = {
 	.suite_name = "service core test suite",
 	.setup = testsuite_setup,
@@ -790,6 +832,7 @@ static struct unit_test_suite service_tests  = {
 		TEST_CASE_ST(dummy_register, NULL, service_mt_safe_poll),
 		TEST_CASE_ST(dummy_register, NULL, service_app_lcore_mt_safe),
 		TEST_CASE_ST(dummy_register, NULL, service_app_lcore_mt_unsafe),
+		TEST_CASE_ST(dummy_register, NULL, service_may_be_active),
 		TEST_CASES_END() /**< NULL terminate unit test array */
 	}
 };
-- 
2.13.6

[dpdk-dev] [PATCH v3 2/2] event/sw: support device stop flush callback

[Gage Eads]

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>
---
 drivers/event/sw/sw_evdev.c          | 114 ++++++++++++++++++++++++++++++++++-
 drivers/event/sw/sw_evdev_selftest.c |  81 ++++++++++++++++++++++++-
 2 files changed, 192 insertions(+), 3 deletions(-)

diff --git a/drivers/event/sw/sw_evdev.c b/drivers/event/sw/sw_evdev.c
index 10f0e1ad4..331c518ff 100644
--- 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 78d30e07a..c40912db5 100644
--- 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 @@ holb(struct test *t) /* test to check we avoid basic head-of-line blocking */
 	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);
-- 
2.13.6

Re: [dpdk-dev] [PATCH v3 0/2] Improve service stop support

[Jerin Jacob]

-----Original Message-----
> Date: Thu, 21 Jun 2018 09:23:21 -0500
> From: Gage Eads <gage.eads@intel.com>
> To: dev@dpdk.org
> CC: jerin.jacob@caviumnetworks.com, harry.van.haaren@intel.com,
>  bruce.richardson@intel.com, nikhil.rao@intel.com,
>  erik.g.carrillo@intel.com, abhinandan.gujjar@intel.com,
>  thomas@monjalon.net
> Subject: [PATCH v3 0/2] Improve service stop support
> X-Mailer: git-send-email 2.13.6
> 
> 
> Existing service functions allow us to stop a service, but doing so doesn't
> guarantee that the service has finished running on a service core. This
> patch set introduces a function, rte_service_may_be_active(), to check
> whether a stopped service is truly stopped.
> 
> This is needed for flows that modify a resource that the service is
> using; for example when stopping an eventdev, any event adapters and/or
> scheduler service need to be quiesced first.
> 
> This patch set also adds support for the event sw PMD's device stop flush
> callback, which relies on this new mechanism to ensure that the
> scheduler service is no longer active.
> 
> v2:
> - Move function to DPDK_18.08 block in rte_eal_version.map
> - Fix signed vs. unsigned comparison compiler warning
> 
> v3:
> - Move function to EXPERIMENTAL block and add experimental tags
> - Fix typo in function documentation
> 

Applied series to dpdk-next-eventdev/master. Thanks.

> Gage Eads (2):
>   service: add mechanism for quiescing a service
>   event/sw: support device stop flush callback
> 
>  drivers/event/sw/sw_evdev.c                 | 114 +++++++++++++++++++++++++++-
>  drivers/event/sw/sw_evdev_selftest.c        |  81 +++++++++++++++++++-
>  lib/librte_eal/common/include/rte_service.h |  20 +++++
>  lib/librte_eal/common/rte_service.c         |  32 +++++++-
>  lib/librte_eal/rte_eal_version.map          |   1 +
>  test/test/test_service_cores.c              |  43 +++++++++++
>  6 files changed, 284 insertions(+), 7 deletions(-)
> 
> --
> 2.13.6
> 

[dpdk-dev] [PATCH 1/2] eventdev: add device stop flush callback

[Gage Eads]

When an event device is stopped, it drains all event queues. These events
may contain pointers, so to prevent memory leaks eventdev now supports a
user-provided flush callback that is called during the queue drain process.
This callback is stored in process memory, so the callback must be
registered by any process that may call rte_event_dev_stop().

This commit also clarifies the behavior of rte_event_dev_stop().

This follows this mailing list discussion:
http://dpdk.org/ml/archives/dev/2018-January/087484.html

Signed-off-by: Gage Eads <gage.eads@intel.com>
---
 lib/librte_eventdev/rte_eventdev.c           | 20 +++++++++++
 lib/librte_eventdev/rte_eventdev.h           | 53 ++++++++++++++++++++++++++--
 lib/librte_eventdev/rte_eventdev_version.map |  6 ++++
 3 files changed, 77 insertions(+), 2 deletions(-)

diff --git a/lib/librte_eventdev/rte_eventdev.c b/lib/librte_eventdev/rte_eventdev.c
index 851a119..7de44f9 100644
--- a/lib/librte_eventdev/rte_eventdev.c
+++ b/lib/librte_eventdev/rte_eventdev.c
@@ -1123,6 +1123,26 @@ rte_event_dev_start(uint8_t dev_id)
 	return 0;
 }
 
+int
+rte_event_dev_stop_flush_callback_register(uint8_t dev_id,
+		eventdev_stop_flush_t callback, void *userdata)
+{
+	struct rte_eventdev *dev;
+
+	RTE_EDEV_LOG_DEBUG("Stop flush register dev_id=%" PRIu8, dev_id);
+
+	RTE_EVENTDEV_VALID_DEVID_OR_ERR_RET(dev_id, -EINVAL);
+	dev = &rte_eventdevs[dev_id];
+
+	if (callback == NULL)
+		return -EINVAL;
+
+	dev->dev_stop_flush = callback;
+	dev->dev_stop_flush_arg = userdata;
+
+	return 0;
+}
+
 void
 rte_event_dev_stop(uint8_t dev_id)
 {
diff --git a/lib/librte_eventdev/rte_eventdev.h b/lib/librte_eventdev/rte_eventdev.h
index b21c271..ec3497f 100644
--- a/lib/librte_eventdev/rte_eventdev.h
+++ b/lib/librte_eventdev/rte_eventdev.h
@@ -835,11 +835,23 @@ int
 rte_event_dev_start(uint8_t dev_id);
 
 /**
- * Stop an event device. The device can be restarted with a call to
- * rte_event_dev_start()
+ * Stop an event device.
+ *
+ * This function causes all queued events to be drained. While draining events
+ * out of the device, this function calls the user-provided flush callback
+ * (if one was registered) once per event.
+ *
+ * This function does not drain events from event ports; the application is
+ * responsible for flushing events from all ports before stopping the device.
+ *
+ * The device can be restarted with a call to rte_event_dev_start(). Threads
+ * that continue to enqueue/dequeue while the device is stopped, or being
+ * stopped, will result in undefined behavior.
  *
  * @param dev_id
  *   Event device identifier.
+ *
+ * @see rte_event_dev_stop_flush_callback_register()
  */
 void
 rte_event_dev_stop(uint8_t dev_id);
@@ -1115,6 +1127,12 @@ typedef uint16_t (*event_dequeue_burst_t)(void *port, struct rte_event ev[],
 		uint16_t nb_events, uint64_t timeout_ticks);
 /**< @internal Dequeue burst of events from port of a device */
 
+typedef void (*eventdev_stop_flush_t)(uint8_t dev_id, struct rte_event event,
+		void *arg);
+/**< Callback function called during rte_event_dev_stop(), invoked once per
+ * flushed event.
+ */
+
 #define RTE_EVENTDEV_NAME_MAX_LEN	(64)
 /**< @internal Max length of name of event PMD */
 
@@ -1176,6 +1194,11 @@ struct rte_eventdev {
 	event_dequeue_burst_t dequeue_burst;
 	/**< Pointer to PMD dequeue burst function. */
 
+	eventdev_stop_flush_t dev_stop_flush;
+	/**< Optional, user-provided event flush function */
+	void *dev_stop_flush_arg;
+	/**< User-provided argument for event flush function */
+
 	struct rte_eventdev_data *data;
 	/**< Pointer to device data */
 	const struct rte_eventdev_ops *dev_ops;
@@ -1822,6 +1845,32 @@ rte_event_dev_xstats_reset(uint8_t dev_id,
  */
 int rte_event_dev_selftest(uint8_t dev_id);
 
+/**
+ * Registers a callback function to be invoked during rte_event_dev_stop() for
+ * each flushed event. This function can be used to properly dispose of queued
+ * events, for example events containing memory pointers.
+ *
+ * The callback function is only registered for the calling process. The
+ * callback function must be registered in every process that can call
+ * rte_event_dev_stop().
+ *
+ * @param dev_id
+ *   The identifier of the device.
+ * @param callback
+ *   Callback function invoked once per flushed event.
+ * @param userdata
+ *   Argument supplied to callback.
+ *
+ * @return
+ *  - 0 on success.
+ *  - -EINVAL if *dev_id* is invalid or *callback* is NULL
+ *
+ * @see rte_event_dev_stop()
+ */
+int
+rte_event_dev_stop_flush_callback_register(uint8_t dev_id,
+		eventdev_stop_flush_t callback, void *userdata);
+
 #ifdef __cplusplus
 }
 #endif
diff --git a/lib/librte_eventdev/rte_eventdev_version.map b/lib/librte_eventdev/rte_eventdev_version.map
index 2aef470..4396536 100644
--- a/lib/librte_eventdev/rte_eventdev_version.map
+++ b/lib/librte_eventdev/rte_eventdev_version.map
@@ -74,3 +74,9 @@ DPDK_18.02 {
 
 	rte_event_dev_selftest;
 } DPDK_17.11;
+
+DPDK_18.05 {
+	global:
+
+	rte_event_dev_stop_flush_callback_register;
+} DPDK_18.02;
-- 
2.7.4

[dpdk-dev] [PATCH 2/2] event/sw: support device stop flush callback

[Gage Eads]

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

Signed-off-by: Gage Eads <gage.eads@intel.com>
---
 drivers/event/sw/sw_evdev.c          | 25 +++++++++++-
 drivers/event/sw/sw_evdev_selftest.c | 75 +++++++++++++++++++++++++++++++++++-
 2 files changed, 97 insertions(+), 3 deletions(-)

diff --git a/drivers/event/sw/sw_evdev.c b/drivers/event/sw/sw_evdev.c
index 6672fd8..4b57e5b 100644
--- a/drivers/event/sw/sw_evdev.c
+++ b/drivers/event/sw/sw_evdev.c
@@ -362,8 +362,25 @@ sw_init_qid_iqs(struct sw_evdev *sw)
 }
 
 static void
-sw_clean_qid_iqs(struct sw_evdev *sw)
+sw_flush_iq(struct rte_eventdev *dev, struct sw_iq *iq)
 {
+	struct sw_evdev *sw = sw_pmd_priv(dev);
+
+	while (iq_count(iq) > 0) {
+		struct rte_event event;
+
+		iq_dequeue_burst(sw, iq, &event, 1);
+
+		dev->dev_stop_flush(dev->data->dev_id,
+				    event,
+				    dev->dev_stop_flush_arg);
+	}
+}
+
+static void
+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 */
@@ -373,7 +390,11 @@ sw_clean_qid_iqs(struct sw_evdev *sw)
 		for (j = 0; j < SW_IQS_MAX; j++) {
 			if (!qid->iq[j].head)
 				continue;
+
+			if (dev->dev_stop_flush)
+				sw_flush_iq(dev, &qid->iq[j]);
 			iq_free_chunk_list(sw, qid->iq[j].head);
+
 			qid->iq[j].head = NULL;
 		}
 	}
@@ -702,7 +723,7 @@ static void
 sw_stop(struct rte_eventdev *dev)
 {
 	struct sw_evdev *sw = sw_pmd_priv(dev);
-	sw_clean_qid_iqs(sw);
+	sw_clean_qid_iqs(dev);
 	sw_xstats_uninit(sw);
 	sw->started = 0;
 	rte_smp_wmb();
diff --git a/drivers/event/sw/sw_evdev_selftest.c b/drivers/event/sw/sw_evdev_selftest.c
index 78d30e0..f59362e 100644
--- 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,72 @@ holb(struct test *t) /* test to check we avoid basic head-of-line blocking */
 	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
+			/* all other fields zero */
+	};
+	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;
+	}
+
+	return 0;
+err:
+	rte_event_dev_dump(evdev, stdout);
+	cleanup(t);
+	return -1;
+}
 static int
 worker_loopback_worker_fn(void *arg)
 {
@@ -3211,6 +3278,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);
-- 
2.7.4