drivers/event/sw/sw_evdev_worker.c

   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 }