[v2,2/2] event/dpaa: add select based event support

diff --git a/doc/guides/eventdevs/dpaa.rst b/doc/guides/eventdevs/dpaa.rst
index 7383295..2f356d3 100644
--- a/doc/guides/eventdevs/dpaa.rst
+++ b/doc/guides/eventdevs/dpaa.rst
@@ -122,6 +122,8 @@  Example:
 
     ./your_eventdev_application --vdev="event_dpaa1"
 
+* Use dev arg option ``disable_intr=1`` to disable the interrupt mode
+
 Limitations
 -----------
 
diff --git a/drivers/event/dpaa/dpaa_eventdev.c b/drivers/event/dpaa/dpaa_eventdev.c
index 9ddaf30..1e247e4 100644
--- a/drivers/event/dpaa/dpaa_eventdev.c
+++ b/drivers/event/dpaa/dpaa_eventdev.c
@@ -30,6 +30,7 @@ 
 #include <rte_dpaa_bus.h>
 #include <rte_dpaa_logs.h>
 #include <rte_cycles.h>
+#include <rte_kvargs.h>
 
 #include <dpaa_ethdev.h>
 #include "dpaa_eventdev.h"
@@ -43,22 +44,34 @@ 
  * 1 Eventdev can have N Eventqueue
  */
 
+#define DISABLE_INTR_MODE "disable_intr"
+
 static int
 dpaa_event_dequeue_timeout_ticks(struct rte_eventdev *dev, uint64_t ns,
 				 uint64_t *timeout_ticks)
 {
-	uint64_t cycles_per_second;
-
 	EVENTDEV_INIT_FUNC_TRACE();
 
 	RTE_SET_USED(dev);
 
+	uint64_t cycles_per_second;
+
 	cycles_per_second = rte_get_timer_hz();
-	*timeout_ticks = ns * (cycles_per_second / NS_PER_S);
+	*timeout_ticks = (ns * cycles_per_second) / NS_PER_S;
 
 	return 0;
 }
 
+static int
+dpaa_event_dequeue_timeout_ticks_intr(struct rte_eventdev *dev, uint64_t ns,
+				 uint64_t *timeout_ticks)
+{
+	RTE_SET_USED(dev);
+
+	*timeout_ticks = ns/1000;
+	return 0;
+}
+
 static void
 dpaa_eventq_portal_add(u16 ch_id)
 {
@@ -100,6 +113,56 @@  dpaa_event_enqueue(void *port, const struct rte_event *ev)
 	return dpaa_event_enqueue_burst(port, ev, 1);
 }
 
+static void drain_4_bytes(int fd, fd_set *fdset)
+{
+	if (FD_ISSET(fd, fdset)) {
+		/* drain 4 bytes */
+		uint32_t junk;
+		ssize_t sjunk = read(qman_thread_fd(), &junk, sizeof(junk));
+		if (sjunk != sizeof(junk))
+			DPAA_EVENTDEV_ERR("UIO irq read error");
+	}
+}
+
+static inline int
+dpaa_event_dequeue_wait(uint64_t timeout_ticks)
+{
+	int fd_qman, nfds;
+	int ret;
+	fd_set readset;
+
+	/* Go into (and back out of) IRQ mode for each select,
+	 * it simplifies exit-path considerations and other
+	 * potential nastiness.
+	 */
+	struct timeval tv = {
+		.tv_sec = timeout_ticks / 1000000,
+		.tv_usec = timeout_ticks % 1000000
+	};
+
+	fd_qman = qman_thread_fd();
+	nfds = fd_qman + 1;
+	FD_ZERO(&readset);
+	FD_SET(fd_qman, &readset);
+
+	qman_irqsource_add(QM_PIRQ_DQRI);
+
+	ret = select(nfds, &readset, NULL, NULL, &tv);
+	if (ret < 0)
+		return ret;
+	/* Calling irqsource_remove() prior to thread_irq()
+	 * means thread_irq() will not process whatever caused
+	 * the interrupts, however it does ensure that, once
+	 * thread_irq() re-enables interrupts, they won't fire
+	 * again immediately.
+	 */
+	qman_irqsource_remove(~0);
+	drain_4_bytes(fd_qman, &readset);
+	qman_thread_irq();
+
+	return ret;
+}
+
 static uint16_t
 dpaa_event_dequeue_burst(void *port, struct rte_event ev[],
 			 uint16_t nb_events, uint64_t timeout_ticks)
@@ -107,8 +170,8 @@  dpaa_event_dequeue_burst(void *port, struct rte_event ev[],
 	int ret;
 	u16 ch_id;
 	void *buffers[8];
-	u32 num_frames, i;
-	uint64_t wait_time, cur_ticks, start_ticks;
+	u32 num_frames, i, irq = 0;
+	uint64_t cur_ticks = 0, wait_time_ticks = 0;
 	struct dpaa_port *portal = (struct dpaa_port *)port;
 	struct rte_mbuf *mbuf;
 
@@ -147,20 +210,21 @@  dpaa_event_dequeue_burst(void *port, struct rte_event ev[],
 	}
 	DPAA_PER_LCORE_DQRR_HELD = 0;
 
-	if (portal->timeout == DPAA_EVENT_PORT_DEQUEUE_TIMEOUT_INVALID)
-		wait_time = timeout_ticks;
+	if (timeout_ticks)
+		wait_time_ticks = timeout_ticks;
 	else
-		wait_time = portal->timeout;
+		wait_time_ticks = portal->timeout_us;
 
-	/* Lets dequeue the frames */
-	start_ticks = rte_get_timer_cycles();
-	wait_time += start_ticks;
+	wait_time_ticks += rte_get_timer_cycles();
 	do {
+		/* Lets dequeue the frames */
 		num_frames = qman_portal_dequeue(ev, nb_events, buffers);
-		if (num_frames != 0)
+		if (irq)
+			irq = 0;
+		if (num_frames)
 			break;
 		cur_ticks = rte_get_timer_cycles();
-	} while (cur_ticks < wait_time);
+	} while (cur_ticks < wait_time_ticks);
 
 	return num_frames;
 }
@@ -171,6 +235,86 @@  dpaa_event_dequeue(void *port, struct rte_event *ev, uint64_t timeout_ticks)
 	return dpaa_event_dequeue_burst(port, ev, 1, timeout_ticks);
 }
 
+static uint16_t
+dpaa_event_dequeue_burst_intr(void *port, struct rte_event ev[],
+			      uint16_t nb_events, uint64_t timeout_ticks)
+{
+	int ret;
+	u16 ch_id;
+	void *buffers[8];
+	u32 num_frames, i, irq = 0;
+	uint64_t cur_ticks = 0, wait_time_ticks = 0;
+	struct dpaa_port *portal = (struct dpaa_port *)port;
+	struct rte_mbuf *mbuf;
+
+	if (unlikely(!RTE_PER_LCORE(dpaa_io))) {
+		/* Affine current thread context to a qman portal */
+		ret = rte_dpaa_portal_init((void *)0);
+		if (ret) {
+			DPAA_EVENTDEV_ERR("Unable to initialize portal");
+			return ret;
+		}
+	}
+
+	if (unlikely(!portal->is_port_linked)) {
+		/*
+		 * Affine event queue for current thread context
+		 * to a qman portal.
+		 */
+		for (i = 0; i < portal->num_linked_evq; i++) {
+			ch_id = portal->evq_info[i].ch_id;
+			dpaa_eventq_portal_add(ch_id);
+		}
+		portal->is_port_linked = true;
+	}
+
+	/* Check if there are atomic contexts to be released */
+	i = 0;
+	while (DPAA_PER_LCORE_DQRR_SIZE) {
+		if (DPAA_PER_LCORE_DQRR_HELD & (1 << i)) {
+			qman_dca_index(i, 0);
+			mbuf = DPAA_PER_LCORE_DQRR_MBUF(i);
+			mbuf->seqn = DPAA_INVALID_MBUF_SEQN;
+			DPAA_PER_LCORE_DQRR_HELD &= ~(1 << i);
+			DPAA_PER_LCORE_DQRR_SIZE--;
+		}
+		i++;
+	}
+	DPAA_PER_LCORE_DQRR_HELD = 0;
+
+	if (timeout_ticks)
+		wait_time_ticks = timeout_ticks;
+	else
+		wait_time_ticks = portal->timeout_us;
+
+	do {
+		/* Lets dequeue the frames */
+		num_frames = qman_portal_dequeue(ev, nb_events, buffers);
+		if (irq)
+			irq = 0;
+		if (num_frames)
+			break;
+		if (wait_time_ticks) { /* wait for time */
+			if (dpaa_event_dequeue_wait(wait_time_ticks) > 0) {
+				irq = 1;
+				continue;
+			}
+			break; /* no event after waiting */
+		}
+		cur_ticks = rte_get_timer_cycles();
+	} while (cur_ticks < wait_time_ticks);
+
+	return num_frames;
+}
+
+static uint16_t
+dpaa_event_dequeue_intr(void *port,
+			struct rte_event *ev,
+			uint64_t timeout_ticks)
+{
+	return dpaa_event_dequeue_burst_intr(port, ev, 1, timeout_ticks);
+}
+
 static void
 dpaa_event_dev_info_get(struct rte_eventdev *dev,
 			struct rte_event_dev_info *dev_info)
@@ -184,7 +328,7 @@  dpaa_event_dev_info_get(struct rte_eventdev *dev,
 	dev_info->max_dequeue_timeout_ns =
 		DPAA_EVENT_MAX_DEQUEUE_TIMEOUT;
 	dev_info->dequeue_timeout_ns =
-		DPAA_EVENT_MIN_DEQUEUE_TIMEOUT;
+		DPAA_EVENT_PORT_DEQUEUE_TIMEOUT_NS;
 	dev_info->max_event_queues =
 		DPAA_EVENT_MAX_QUEUES;
 	dev_info->max_event_queue_flows =
@@ -230,15 +374,6 @@  dpaa_event_dev_configure(const struct rte_eventdev *dev)
 	priv->nb_event_port_enqueue_depth = conf->nb_event_port_enqueue_depth;
 	priv->event_dev_cfg = conf->event_dev_cfg;
 
-	/* Check dequeue timeout method is per dequeue or global */
-	if (priv->event_dev_cfg & RTE_EVENT_DEV_CFG_PER_DEQUEUE_TIMEOUT) {
-		/*
-		 * Use timeout value as given in dequeue operation.
-		 * So invalidating this timetout value.
-		 */
-		priv->dequeue_timeout_ns = 0;
-	}
-
 	ch_id = rte_malloc("dpaa-channels",
 			  sizeof(uint32_t) * priv->nb_event_queues,
 			  RTE_CACHE_LINE_SIZE);
@@ -260,24 +395,35 @@  dpaa_event_dev_configure(const struct rte_eventdev *dev)
 	/* Lets prepare event ports */
 	memset(&priv->ports[0], 0,
 	      sizeof(struct dpaa_port) * priv->nb_event_ports);
+
+	/* Check dequeue timeout method is per dequeue or global */
 	if (priv->event_dev_cfg & RTE_EVENT_DEV_CFG_PER_DEQUEUE_TIMEOUT) {
-		for (i = 0; i < priv->nb_event_ports; i++) {
-			priv->ports[i].timeout =
-				DPAA_EVENT_PORT_DEQUEUE_TIMEOUT_INVALID;
-		}
-	} else if (priv->dequeue_timeout_ns == 0) {
-		for (i = 0; i < priv->nb_event_ports; i++) {
-			dpaa_event_dequeue_timeout_ticks(NULL,
-				DPAA_EVENT_PORT_DEQUEUE_TIMEOUT_NS,
-				&priv->ports[i].timeout);
-		}
+		/*
+		 * Use timeout value as given in dequeue operation.
+		 * So invalidating this timeout value.
+		 */
+		priv->dequeue_timeout_ns = 0;
+
+	} else if (conf->dequeue_timeout_ns == 0) {
+		priv->dequeue_timeout_ns = DPAA_EVENT_PORT_DEQUEUE_TIMEOUT_NS;
 	} else {
-		for (i = 0; i < priv->nb_event_ports; i++) {
-			dpaa_event_dequeue_timeout_ticks(NULL,
-				priv->dequeue_timeout_ns,
-				&priv->ports[i].timeout);
+		priv->dequeue_timeout_ns = conf->dequeue_timeout_ns;
+	}
+
+	for (i = 0; i < priv->nb_event_ports; i++) {
+		if (priv->intr_mode) {
+			priv->ports[i].timeout_us =
+				priv->dequeue_timeout_ns/1000;
+		} else {
+			uint64_t cycles_per_second;
+
+			cycles_per_second = rte_get_timer_hz();
+			priv->ports[i].timeout_us =
+				(priv->dequeue_timeout_ns * cycles_per_second)
+					/ NS_PER_S;
 		}
 	}
+
 	/*
 	 * TODO: Currently portals are affined with threads. Maximum threads
 	 * can be created equals to number of lcore.
@@ -454,7 +600,8 @@  dpaa_event_port_unlink(struct rte_eventdev *dev, void *port,
 		event_queue->event_port = NULL;
 	}
 
-	event_port->num_linked_evq = event_port->num_linked_evq - i;
+	if (event_port->num_linked_evq)
+		event_port->num_linked_evq = event_port->num_linked_evq - i;
 
 	return (int)i;
 }
@@ -593,8 +740,44 @@  static struct rte_eventdev_ops dpaa_eventdev_ops = {
 	.eth_rx_adapter_stop = dpaa_event_eth_rx_adapter_stop,
 };
 
+static int flag_check_handler(__rte_unused const char *key,
+		const char *value, __rte_unused void *opaque)
+{
+	if (strcmp(value, "1"))
+		return -1;
+
+	return 0;
+}
+
+static int
+dpaa_event_check_flags(const char *params)
+{
+	struct rte_kvargs *kvlist;
+
+	if (params == NULL || params[0] == '\0')
+		return 0;
+
+	kvlist = rte_kvargs_parse(params, NULL);
+	if (kvlist == NULL)
+		return 0;
+
+	if (!rte_kvargs_count(kvlist, DISABLE_INTR_MODE)) {
+		rte_kvargs_free(kvlist);
+		return 0;
+	}
+	/* INTR MODE is disabled when there's key-value pair: disable_intr = 1*/
+	if (rte_kvargs_process(kvlist, DISABLE_INTR_MODE,
+				flag_check_handler, NULL) < 0) {
+		rte_kvargs_free(kvlist);
+		return 0;
+	}
+	rte_kvargs_free(kvlist);
+
+	return 1;
+}
+
 static int
-dpaa_event_dev_create(const char *name)
+dpaa_event_dev_create(const char *name, const char *params)
 {
 	struct rte_eventdev *eventdev;
 	struct dpaa_eventdev *priv;
@@ -606,18 +789,27 @@  dpaa_event_dev_create(const char *name)
 		DPAA_EVENTDEV_ERR("Failed to create eventdev vdev %s", name);
 		goto fail;
 	}
+	priv = eventdev->data->dev_private;
 
 	eventdev->dev_ops       = &dpaa_eventdev_ops;
 	eventdev->enqueue       = dpaa_event_enqueue;
 	eventdev->enqueue_burst = dpaa_event_enqueue_burst;
-	eventdev->dequeue       = dpaa_event_dequeue;
-	eventdev->dequeue_burst = dpaa_event_dequeue_burst;
+
+	if (dpaa_event_check_flags(params)) {
+		eventdev->dequeue	= dpaa_event_dequeue;
+		eventdev->dequeue_burst = dpaa_event_dequeue_burst;
+	} else {
+		priv->intr_mode = 1;
+		eventdev->dev_ops->timeout_ticks =
+				dpaa_event_dequeue_timeout_ticks_intr;
+		eventdev->dequeue	= dpaa_event_dequeue_intr;
+		eventdev->dequeue_burst = dpaa_event_dequeue_burst_intr;
+	}
 
 	/* For secondary processes, the primary has done all the work */
 	if (rte_eal_process_type() != RTE_PROC_PRIMARY)
 		return 0;
 
-	priv = eventdev->data->dev_private;
 	priv->max_event_queues = DPAA_EVENT_MAX_QUEUES;
 
 	return 0;
@@ -629,11 +821,14 @@  static int
 dpaa_event_dev_probe(struct rte_vdev_device *vdev)
 {
 	const char *name;
+	const char *params;
 
 	name = rte_vdev_device_name(vdev);
 	DPAA_EVENTDEV_INFO("Initializing %s", name);
 
-	return dpaa_event_dev_create(name);
+	params = rte_vdev_device_args(vdev);
+
+	return dpaa_event_dev_create(name, params);
 }
 
 static int
@@ -653,3 +848,5 @@  static struct rte_vdev_driver vdev_eventdev_dpaa_pmd = {
 };
 
 RTE_PMD_REGISTER_VDEV(EVENTDEV_NAME_DPAA_PMD, vdev_eventdev_dpaa_pmd);
+RTE_PMD_REGISTER_PARAM_STRING(EVENTDEV_NAME_DPAA_PMD,
+		DISABLE_INTR_MODE "=<int>");
diff --git a/drivers/event/dpaa/dpaa_eventdev.h b/drivers/event/dpaa/dpaa_eventdev.h
index 3994bd6..8134e6b 100644
--- a/drivers/event/dpaa/dpaa_eventdev.h
+++ b/drivers/event/dpaa/dpaa_eventdev.h
@@ -12,8 +12,8 @@ 
 
 #define EVENTDEV_NAME_DPAA_PMD		event_dpaa1
 
-#define DPAA_EVENT_MAX_PORTS			8
-#define DPAA_EVENT_MAX_QUEUES			16
+#define DPAA_EVENT_MAX_PORTS			4
+#define DPAA_EVENT_MAX_QUEUES			8
 #define DPAA_EVENT_MIN_DEQUEUE_TIMEOUT	1
 #define DPAA_EVENT_MAX_DEQUEUE_TIMEOUT	(UINT32_MAX - 1)
 #define DPAA_EVENT_MAX_QUEUE_FLOWS		2048
@@ -21,7 +21,7 @@ 
 #define DPAA_EVENT_MAX_EVENT_PRIORITY_LEVELS	0
 #define DPAA_EVENT_MAX_EVENT_PORT		RTE_MIN(RTE_MAX_LCORE, INT8_MAX)
 #define DPAA_EVENT_MAX_PORT_DEQUEUE_DEPTH	8
-#define DPAA_EVENT_PORT_DEQUEUE_TIMEOUT_NS	100UL
+#define DPAA_EVENT_PORT_DEQUEUE_TIMEOUT_NS	100000UL
 #define DPAA_EVENT_PORT_DEQUEUE_TIMEOUT_INVALID	((uint64_t)-1)
 #define DPAA_EVENT_MAX_PORT_ENQUEUE_DEPTH	1
 #define DPAA_EVENT_MAX_NUM_EVENTS		(INT32_MAX - 1)
@@ -54,7 +54,7 @@  struct dpaa_port {
 	struct dpaa_eventq evq_info[DPAA_EVENT_MAX_QUEUES];
 	uint8_t num_linked_evq;
 	uint8_t is_port_linked;
-	uint64_t timeout;
+	uint64_t timeout_us;
 };
 
 struct dpaa_eventdev {
@@ -65,7 +65,7 @@  struct dpaa_eventdev {
 	uint8_t max_event_queues;
 	uint8_t nb_event_queues;
 	uint8_t nb_event_ports;
-	uint8_t resvd;
+	uint8_t intr_mode;
 	uint32_t nb_event_queue_flows;
 	uint32_t nb_event_port_dequeue_depth;
 	uint32_t nb_event_port_enqueue_depth;