4 * Copyright(c) 2016-2017 Intel Corporation. All rights reserved.
6 * Redistribution and use in source and binary forms, with or without
7 * modification, are permitted provided that the following conditions
10 * * Redistributions of source code must retain the above copyright
11 * notice, this list of conditions and the following disclaimer.
12 * * Redistributions in binary form must reproduce the above copyright
13 * notice, this list of conditions and the following disclaimer in
14 * the documentation and/or other materials provided with the
16 * * Neither the name of Intel Corporation nor the names of its
17 * contributors may be used to endorse or promote products derived
18 * from this software without specific prior written permission.
20 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
21 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
22 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
23 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
24 * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
25 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
26 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
27 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
28 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
29 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
30 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
33 #include <rte_atomic.h>
34 #include <rte_cycles.h>
35 #include <rte_event_ring.h>
39 #define PORT_ENQUEUE_MAX_BURST_SIZE 64
42 sw_event_release(struct sw_port *p, uint8_t index)
45 * Drops the next outstanding event in our history. Used on dequeue
46 * to clear any history before dequeuing more events.
50 /* create drop message */
52 ev.op = sw_qe_flag_map[RTE_EVENT_OP_RELEASE];
55 rte_event_ring_enqueue_burst(p->rx_worker_ring, &ev, 1, &free_count);
57 /* each release returns one credit */
58 p->outstanding_releases--;
59 p->inflight_credits++;
63 * special-case of rte_event_ring enqueue, with overriding the ops member on
64 * the events that get written to the ring.
66 static inline unsigned int
67 enqueue_burst_with_ops(struct rte_event_ring *r, const struct rte_event *events,
68 unsigned int n, uint8_t *ops)
70 struct rte_event tmp_evs[PORT_ENQUEUE_MAX_BURST_SIZE];
73 memcpy(tmp_evs, events, n * sizeof(events[0]));
74 for (i = 0; i < n; i++)
75 tmp_evs[i].op = ops[i];
77 return rte_event_ring_enqueue_burst(r, tmp_evs, n, NULL);
81 sw_event_enqueue_burst(void *port, const struct rte_event ev[], uint16_t num)
84 uint8_t new_ops[PORT_ENQUEUE_MAX_BURST_SIZE];
85 struct sw_port *p = port;
86 struct sw_evdev *sw = (void *)p->sw;
87 uint32_t sw_inflights = rte_atomic32_read(&sw->inflights);
90 if (num > PORT_ENQUEUE_MAX_BURST_SIZE)
91 num = PORT_ENQUEUE_MAX_BURST_SIZE;
93 for (i = 0; i < num; i++)
94 new += (ev[i].op == RTE_EVENT_OP_NEW);
96 if (unlikely(new > 0 && p->inflight_max < sw_inflights))
99 if (p->inflight_credits < new) {
100 /* check if event enqueue brings port over max threshold */
101 uint32_t credit_update_quanta = sw->credit_update_quanta;
102 if (sw_inflights + credit_update_quanta > sw->nb_events_limit)
105 rte_atomic32_add(&sw->inflights, credit_update_quanta);
106 p->inflight_credits += (credit_update_quanta);
108 if (p->inflight_credits < new)
112 uint32_t forwards = 0;
113 for (i = 0; i < num; i++) {
115 int outstanding = p->outstanding_releases > 0;
116 const uint8_t invalid_qid = (ev[i].queue_id >= sw->qid_count);
118 p->inflight_credits -= (op == RTE_EVENT_OP_NEW);
119 p->inflight_credits += (op == RTE_EVENT_OP_RELEASE) *
121 forwards += (op == RTE_EVENT_OP_FORWARD);
123 new_ops[i] = sw_qe_flag_map[op];
124 new_ops[i] &= ~(invalid_qid << QE_FLAG_VALID_SHIFT);
126 /* FWD and RELEASE packets will both resolve to taken (assuming
127 * correct usage of the API), providing very high correct
130 if ((new_ops[i] & QE_FLAG_COMPLETE) && outstanding)
131 p->outstanding_releases--;
133 /* error case: branch to avoid touching p->stats */
134 if (unlikely(invalid_qid)) {
135 p->stats.rx_dropped++;
136 p->inflight_credits++;
140 /* handle directed port forward credits */
141 p->inflight_credits -= forwards * p->is_directed;
143 /* returns number of events actually enqueued */
144 uint32_t enq = enqueue_burst_with_ops(p->rx_worker_ring, ev, i,
146 if (p->outstanding_releases == 0 && p->last_dequeue_burst_sz != 0) {
147 uint64_t burst_ticks = rte_get_timer_cycles() -
148 p->last_dequeue_ticks;
149 uint64_t burst_pkt_ticks =
150 burst_ticks / p->last_dequeue_burst_sz;
151 p->avg_pkt_ticks -= p->avg_pkt_ticks / NUM_SAMPLES;
152 p->avg_pkt_ticks += burst_pkt_ticks / NUM_SAMPLES;
153 p->last_dequeue_ticks = 0;
159 sw_event_enqueue(void *port, const struct rte_event *ev)
161 return sw_event_enqueue_burst(port, ev, 1);
165 sw_event_dequeue_burst(void *port, struct rte_event *ev, uint16_t num,
169 struct sw_port *p = (void *)port;
170 struct sw_evdev *sw = (void *)p->sw;
171 struct rte_event_ring *ring = p->cq_worker_ring;
172 uint32_t credit_update_quanta = sw->credit_update_quanta;
174 /* check that all previous dequeues have been released */
175 if (!p->is_directed) {
176 uint16_t out_rels = p->outstanding_releases;
178 for (i = 0; i < out_rels; i++)
179 sw_event_release(p, i);
182 /* returns number of events actually dequeued */
183 uint16_t ndeq = rte_event_ring_dequeue_burst(ring, ev, num, NULL);
184 if (unlikely(ndeq == 0)) {
185 p->outstanding_releases = 0;
191 /* only add credits for directed ports - LB ports send RELEASEs */
192 p->inflight_credits += ndeq * p->is_directed;
193 p->outstanding_releases = ndeq;
194 p->last_dequeue_burst_sz = ndeq;
195 p->last_dequeue_ticks = rte_get_timer_cycles();
196 p->poll_buckets[(ndeq - 1) >> SW_DEQ_STAT_BUCKET_SHIFT]++;
200 if (p->inflight_credits >= credit_update_quanta * 2 &&
201 p->inflight_credits > credit_update_quanta + ndeq) {
202 rte_atomic32_sub(&sw->inflights, credit_update_quanta);
203 p->inflight_credits -= credit_update_quanta;
209 sw_event_dequeue(void *port, struct rte_event *ev, uint64_t wait)
211 return sw_event_dequeue_burst(port, ev, 1, wait);