1 /*- 2 * BSD LICENSE 3 * 4 * Copyright(c) 2016-2017 Intel Corporation. All rights reserved. 5 * 6 * Redistribution and use in source and binary forms, with or without 7 * modification, are permitted provided that the following conditions 8 * are met: 9 * 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 15 * distribution. 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. 19 * 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. 31 */ 32 33 #include <rte_atomic.h> 34 #include <rte_cycles.h> 35 #include <rte_event_ring.h> 36 37 #include "sw_evdev.h" 38 39 #define PORT_ENQUEUE_MAX_BURST_SIZE 64 40 41 static inline void 42 sw_event_release(struct sw_port *p, uint8_t index) 43 { 44 /* 45 * Drops the next outstanding event in our history. Used on dequeue 46 * to clear any history before dequeuing more events. 47 */ 48 RTE_SET_USED(index); 49 50 /* create drop message */ 51 struct rte_event ev; 52 ev.op = sw_qe_flag_map[RTE_EVENT_OP_RELEASE]; 53 54 uint16_t free_count; 55 rte_event_ring_enqueue_burst(p->rx_worker_ring, &ev, 1, &free_count); 56 57 /* each release returns one credit */ 58 p->outstanding_releases--; 59 p->inflight_credits++; 60 } 61 62 /* 63 * special-case of rte_event_ring enqueue, with overriding the ops member on 64 * the events that get written to the ring. 65 */ 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) 69 { 70 struct rte_event tmp_evs[PORT_ENQUEUE_MAX_BURST_SIZE]; 71 unsigned int i; 72 73 memcpy(tmp_evs, events, n * sizeof(events[0])); 74 for (i = 0; i < n; i++) 75 tmp_evs[i].op = ops[i]; 76 77 return rte_event_ring_enqueue_burst(r, tmp_evs, n, NULL); 78 } 79 80 uint16_t 81 sw_event_enqueue_burst(void *port, const struct rte_event ev[], uint16_t num) 82 { 83 int32_t i; 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); 88 int new = 0; 89 90 if (num > PORT_ENQUEUE_MAX_BURST_SIZE) 91 num = PORT_ENQUEUE_MAX_BURST_SIZE; 92 93 for (i = 0; i < num; i++) 94 new += (ev[i].op == RTE_EVENT_OP_NEW); 95 96 if (unlikely(new > 0 && p->inflight_max < sw_inflights)) 97 return 0; 98 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) 103 return 0; 104 105 rte_atomic32_add(&sw->inflights, credit_update_quanta); 106 p->inflight_credits += (credit_update_quanta); 107 108 if (p->inflight_credits < new) 109 return 0; 110 } 111 112 uint32_t forwards = 0; 113 for (i = 0; i < num; i++) { 114 int op = ev[i].op; 115 int outstanding = p->outstanding_releases > 0; 116 const uint8_t invalid_qid = (ev[i].queue_id >= sw->qid_count); 117 118 p->inflight_credits -= (op == RTE_EVENT_OP_NEW); 119 p->inflight_credits += (op == RTE_EVENT_OP_RELEASE) * 120 outstanding; 121 forwards += (op == RTE_EVENT_OP_FORWARD); 122 123 new_ops[i] = sw_qe_flag_map[op]; 124 new_ops[i] &= ~(invalid_qid << QE_FLAG_VALID_SHIFT); 125 126 /* FWD and RELEASE packets will both resolve to taken (assuming 127 * correct usage of the API), providing very high correct 128 * prediction rate. 129 */ 130 if ((new_ops[i] & QE_FLAG_COMPLETE) && outstanding) 131 p->outstanding_releases--; 132 133 /* error case: branch to avoid touching p->stats */ 134 if (unlikely(invalid_qid)) { 135 p->stats.rx_dropped++; 136 p->inflight_credits++; 137 } 138 } 139 140 /* handle directed port forward credits */ 141 p->inflight_credits -= forwards * p->is_directed; 142 143 /* returns number of events actually enqueued */ 144 uint32_t enq = enqueue_burst_with_ops(p->rx_worker_ring, ev, i, 145 new_ops); 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; 154 } 155 return enq; 156 } 157 158 uint16_t 159 sw_event_enqueue(void *port, const struct rte_event *ev) 160 { 161 return sw_event_enqueue_burst(port, ev, 1); 162 } 163 164 uint16_t 165 sw_event_dequeue_burst(void *port, struct rte_event *ev, uint16_t num, 166 uint64_t wait) 167 { 168 RTE_SET_USED(wait); 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; 173 174 /* check that all previous dequeues have been released */ 175 if (!p->is_directed) { 176 uint16_t out_rels = p->outstanding_releases; 177 uint16_t i; 178 for (i = 0; i < out_rels; i++) 179 sw_event_release(p, i); 180 } 181 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; 186 p->zero_polls++; 187 p->total_polls++; 188 goto end; 189 } 190 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]++; 197 p->total_polls++; 198 199 end: 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; 204 } 205 return ndeq; 206 } 207 208 uint16_t 209 sw_event_dequeue(void *port, struct rte_event *ev, uint64_t wait) 210 { 211 return sw_event_dequeue_burst(port, ev, 1, wait); 212 } 213