event/sw: add worker core functions
[dpdk.git] / 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
36 #include "sw_evdev.h"
37 #include "event_ring.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                 .op = sw_qe_flag_map[RTE_EVENT_OP_RELEASE],
53         };
54
55         uint16_t free_count;
56         qe_ring_enqueue_burst(p->rx_worker_ring, &ev, 1, &free_count);
57
58         /* each release returns one credit */
59         p->outstanding_releases--;
60         p->inflight_credits++;
61 }
62
63 uint16_t
64 sw_event_enqueue_burst(void *port, const struct rte_event ev[], uint16_t num)
65 {
66         int32_t i;
67         uint8_t new_ops[PORT_ENQUEUE_MAX_BURST_SIZE];
68         struct sw_port *p = port;
69         struct sw_evdev *sw = (void *)p->sw;
70         uint32_t sw_inflights = rte_atomic32_read(&sw->inflights);
71
72         if (unlikely(p->inflight_max < sw_inflights))
73                 return 0;
74
75         if (num > PORT_ENQUEUE_MAX_BURST_SIZE)
76                 num = PORT_ENQUEUE_MAX_BURST_SIZE;
77
78         if (p->inflight_credits < num) {
79                 /* check if event enqueue brings port over max threshold */
80                 uint32_t credit_update_quanta = sw->credit_update_quanta;
81                 if (sw_inflights + credit_update_quanta > sw->nb_events_limit)
82                         return 0;
83
84                 rte_atomic32_add(&sw->inflights, credit_update_quanta);
85                 p->inflight_credits += (credit_update_quanta);
86
87                 if (p->inflight_credits < num)
88                         return 0;
89         }
90
91         for (i = 0; i < num; i++) {
92                 int op = ev[i].op;
93                 int outstanding = p->outstanding_releases > 0;
94                 const uint8_t invalid_qid = (ev[i].queue_id >= sw->qid_count);
95
96                 p->inflight_credits -= (op == RTE_EVENT_OP_NEW);
97                 p->inflight_credits += (op == RTE_EVENT_OP_RELEASE) *
98                                         outstanding;
99
100                 new_ops[i] = sw_qe_flag_map[op];
101                 new_ops[i] &= ~(invalid_qid << QE_FLAG_VALID_SHIFT);
102
103                 /* FWD and RELEASE packets will both resolve to taken (assuming
104                  * correct usage of the API), providing very high correct
105                  * prediction rate.
106                  */
107                 if ((new_ops[i] & QE_FLAG_COMPLETE) && outstanding)
108                         p->outstanding_releases--;
109                 /* Branch to avoid touching p->stats except error case */
110                 if (unlikely(invalid_qid))
111                         p->stats.rx_dropped++;
112         }
113
114         /* returns number of events actually enqueued */
115         uint32_t enq = qe_ring_enqueue_burst_with_ops(p->rx_worker_ring, ev, i,
116                                              new_ops);
117         if (p->outstanding_releases == 0 && p->last_dequeue_burst_sz != 0) {
118                 uint64_t burst_ticks = rte_get_timer_cycles() -
119                                 p->last_dequeue_ticks;
120                 uint64_t burst_pkt_ticks =
121                         burst_ticks / p->last_dequeue_burst_sz;
122                 p->avg_pkt_ticks -= p->avg_pkt_ticks / NUM_SAMPLES;
123                 p->avg_pkt_ticks += burst_pkt_ticks / NUM_SAMPLES;
124                 p->last_dequeue_ticks = 0;
125         }
126         return enq;
127 }
128
129 uint16_t
130 sw_event_enqueue(void *port, const struct rte_event *ev)
131 {
132         return sw_event_enqueue_burst(port, ev, 1);
133 }
134
135 uint16_t
136 sw_event_dequeue_burst(void *port, struct rte_event *ev, uint16_t num,
137                 uint64_t wait)
138 {
139         RTE_SET_USED(wait);
140         struct sw_port *p = (void *)port;
141         struct sw_evdev *sw = (void *)p->sw;
142         struct qe_ring *ring = p->cq_worker_ring;
143         uint32_t credit_update_quanta = sw->credit_update_quanta;
144
145         /* check that all previous dequeues have been released */
146         if (!p->is_directed) {
147                 uint16_t out_rels = p->outstanding_releases;
148                 uint16_t i;
149                 for (i = 0; i < out_rels; i++)
150                         sw_event_release(p, i);
151         }
152
153         /* returns number of events actually dequeued */
154         uint16_t ndeq = qe_ring_dequeue_burst(ring, ev, num);
155         if (unlikely(ndeq == 0)) {
156                 p->outstanding_releases = 0;
157                 p->zero_polls++;
158                 p->total_polls++;
159                 goto end;
160         }
161
162         /* only add credits for directed ports - LB ports send RELEASEs */
163         p->inflight_credits += ndeq * p->is_directed;
164         p->outstanding_releases = ndeq;
165         p->last_dequeue_burst_sz = ndeq;
166         p->last_dequeue_ticks = rte_get_timer_cycles();
167         p->poll_buckets[(ndeq - 1) >> SW_DEQ_STAT_BUCKET_SHIFT]++;
168         p->total_polls++;
169
170 end:
171         if (p->inflight_credits >= credit_update_quanta * 2 &&
172                         p->inflight_credits > credit_update_quanta + ndeq) {
173                 rte_atomic32_sub(&sw->inflights, credit_update_quanta);
174                 p->inflight_credits -= credit_update_quanta;
175         }
176         return ndeq;
177 }
178
179 uint16_t
180 sw_event_dequeue(void *port, struct rte_event *ev, uint64_t wait)
181 {
182         return sw_event_dequeue_burst(port, ev, 1, wait);
183 }