X-Git-Url: http://git.droids-corp.org/?a=blobdiff_plain;ds=sidebyside;f=drivers%2Fevent%2Fsw%2Fsw_evdev_worker.c;h=d76d3d5c8fb7516b37d348123236341e69a1f6ed;hb=17d207d5eb9be9432bca0a322f6242a54b72b850;hp=ed08778387aa25095db2b4021ed87374396fe66f;hpb=656af9180014ee095deacf106161c873d3063bc8;p=dpdk.git

diff --git a/drivers/event/sw/sw_evdev_worker.c b/drivers/event/sw/sw_evdev_worker.c
index ed08778387..d76d3d5c8f 100644
--- a/drivers/event/sw/sw_evdev_worker.c
+++ b/drivers/event/sw/sw_evdev_worker.c
@@ -32,9 +32,9 @@
 
 #include <rte_atomic.h>
 #include <rte_cycles.h>
+#include <rte_event_ring.h>
 
 #include "sw_evdev.h"
-#include "event_ring.h"
 
 #define PORT_ENQUEUE_MAX_BURST_SIZE 64
 
@@ -48,18 +48,35 @@ sw_event_release(struct sw_port *p, uint8_t index)
 	RTE_SET_USED(index);
 
 	/* create drop message */
-	struct rte_event ev = {
-		.op = sw_qe_flag_map[RTE_EVENT_OP_RELEASE],
-	};
+	struct rte_event ev;
+	ev.op = sw_qe_flag_map[RTE_EVENT_OP_RELEASE];
 
 	uint16_t free_count;
-	qe_ring_enqueue_burst(p->rx_worker_ring, &ev, 1, &free_count);
+	rte_event_ring_enqueue_burst(p->rx_worker_ring, &ev, 1, &free_count);
 
 	/* each release returns one credit */
 	p->outstanding_releases--;
 	p->inflight_credits++;
 }
 
+/*
+ * special-case of rte_event_ring enqueue, with overriding the ops member on
+ * the events that get written to the ring.
+ */
+static inline unsigned int
+enqueue_burst_with_ops(struct rte_event_ring *r, const struct rte_event *events,
+		unsigned int n, uint8_t *ops)
+{
+	struct rte_event tmp_evs[PORT_ENQUEUE_MAX_BURST_SIZE];
+	unsigned int i;
+
+	memcpy(tmp_evs, events, n * sizeof(events[0]));
+	for (i = 0; i < n; i++)
+		tmp_evs[i].op = ops[i];
+
+	return rte_event_ring_enqueue_burst(r, tmp_evs, n, NULL);
+}
+
 uint16_t
 sw_event_enqueue_burst(void *port, const struct rte_event ev[], uint16_t num)
 {
@@ -88,6 +105,7 @@ sw_event_enqueue_burst(void *port, const struct rte_event ev[], uint16_t num)
 			return 0;
 	}
 
+	uint32_t forwards = 0;
 	for (i = 0; i < num; i++) {
 		int op = ev[i].op;
 		int outstanding = p->outstanding_releases > 0;
@@ -96,6 +114,7 @@ sw_event_enqueue_burst(void *port, const struct rte_event ev[], uint16_t num)
 		p->inflight_credits -= (op == RTE_EVENT_OP_NEW);
 		p->inflight_credits += (op == RTE_EVENT_OP_RELEASE) *
 					outstanding;
+		forwards += (op == RTE_EVENT_OP_FORWARD);
 
 		new_ops[i] = sw_qe_flag_map[op];
 		new_ops[i] &= ~(invalid_qid << QE_FLAG_VALID_SHIFT);
@@ -106,13 +125,19 @@ sw_event_enqueue_burst(void *port, const struct rte_event ev[], uint16_t num)
 		 */
 		if ((new_ops[i] & QE_FLAG_COMPLETE) && outstanding)
 			p->outstanding_releases--;
-		/* Branch to avoid touching p->stats except error case */
-		if (unlikely(invalid_qid))
+
+		/* error case: branch to avoid touching p->stats */
+		if (unlikely(invalid_qid)) {
 			p->stats.rx_dropped++;
+			p->inflight_credits++;
+		}
 	}
 
+	/* handle directed port forward credits */
+	p->inflight_credits -= forwards * p->is_directed;
+
 	/* returns number of events actually enqueued */
-	uint32_t enq = qe_ring_enqueue_burst_with_ops(p->rx_worker_ring, ev, i,
+	uint32_t enq = enqueue_burst_with_ops(p->rx_worker_ring, ev, i,
 					     new_ops);
 	if (p->outstanding_releases == 0 && p->last_dequeue_burst_sz != 0) {
 		uint64_t burst_ticks = rte_get_timer_cycles() -
@@ -139,7 +164,7 @@ sw_event_dequeue_burst(void *port, struct rte_event *ev, uint16_t num,
 	RTE_SET_USED(wait);
 	struct sw_port *p = (void *)port;
 	struct sw_evdev *sw = (void *)p->sw;
-	struct qe_ring *ring = p->cq_worker_ring;
+	struct rte_event_ring *ring = p->cq_worker_ring;
 	uint32_t credit_update_quanta = sw->credit_update_quanta;
 
 	/* check that all previous dequeues have been released */
@@ -151,7 +176,7 @@ sw_event_dequeue_burst(void *port, struct rte_event *ev, uint16_t num,
 	}
 
 	/* returns number of events actually dequeued */
-	uint16_t ndeq = qe_ring_dequeue_burst(ring, ev, num);
+	uint16_t ndeq = rte_event_ring_dequeue_burst(ring, ev, num, NULL);
 	if (unlikely(ndeq == 0)) {
 		p->outstanding_releases = 0;
 		p->zero_polls++;