distributor: add new burst-capable library
[dpdk.git] / lib / librte_distributor / rte_distributor.c
1 /*-
2  *   BSD LICENSE
3  *
4  *   Copyright(c) 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 <stdio.h>
34 #include <sys/queue.h>
35 #include <string.h>
36 #include <rte_mbuf.h>
37 #include <rte_memory.h>
38 #include <rte_cycles.h>
39 #include <rte_memzone.h>
40 #include <rte_errno.h>
41 #include <rte_string_fns.h>
42 #include <rte_eal_memconfig.h>
43 #include <rte_compat.h>
44 #include "rte_distributor_private.h"
45 #include "rte_distributor_next.h"
46 #include "rte_distributor_v20.h"
47
48 TAILQ_HEAD(rte_dist_burst_list, rte_distributor_v1705);
49
50 static struct rte_tailq_elem rte_dist_burst_tailq = {
51         .name = "RTE_DIST_BURST",
52 };
53 EAL_REGISTER_TAILQ(rte_dist_burst_tailq)
54
55 /**** APIs called by workers ****/
56
57 /**** Burst Packet APIs called by workers ****/
58
59 void
60 rte_distributor_request_pkt_v1705(struct rte_distributor_v1705 *d,
61                 unsigned int worker_id, struct rte_mbuf **oldpkt,
62                 unsigned int count)
63 {
64         struct rte_distributor_buffer_v1705 *buf = &(d->bufs[worker_id]);
65         unsigned int i;
66
67         volatile int64_t *retptr64;
68
69         if (unlikely(d->alg_type == RTE_DIST_ALG_SINGLE)) {
70                 rte_distributor_request_pkt(d->d_v20,
71                         worker_id, oldpkt[0]);
72                 return;
73         }
74
75         retptr64 = &(buf->retptr64[0]);
76         /* Spin while handshake bits are set (scheduler clears it) */
77         while (unlikely(*retptr64 & RTE_DISTRIB_GET_BUF)) {
78                 rte_pause();
79                 uint64_t t = rte_rdtsc()+100;
80
81                 while (rte_rdtsc() < t)
82                         rte_pause();
83         }
84
85         /*
86          * OK, if we've got here, then the scheduler has just cleared the
87          * handshake bits. Populate the retptrs with returning packets.
88          */
89
90         for (i = count; i < RTE_DIST_BURST_SIZE; i++)
91                 buf->retptr64[i] = 0;
92
93         /* Set Return bit for each packet returned */
94         for (i = count; i-- > 0; )
95                 buf->retptr64[i] =
96                         (((int64_t)(uintptr_t)(oldpkt[i])) <<
97                         RTE_DISTRIB_FLAG_BITS) | RTE_DISTRIB_RETURN_BUF;
98
99         /*
100          * Finally, set the GET_BUF  to signal to distributor that cache
101          * line is ready for processing
102          */
103         *retptr64 |= RTE_DISTRIB_GET_BUF;
104 }
105
106 int
107 rte_distributor_poll_pkt_v1705(struct rte_distributor_v1705 *d,
108                 unsigned int worker_id, struct rte_mbuf **pkts)
109 {
110         struct rte_distributor_buffer_v1705 *buf = &d->bufs[worker_id];
111         uint64_t ret;
112         int count = 0;
113         unsigned int i;
114
115         if (unlikely(d->alg_type == RTE_DIST_ALG_SINGLE)) {
116                 pkts[0] = rte_distributor_poll_pkt(d->d_v20, worker_id);
117                 return (pkts[0]) ? 1 : 0;
118         }
119
120         /* If bit is set, return */
121         if (buf->bufptr64[0] & RTE_DISTRIB_GET_BUF)
122                 return -1;
123
124         /* since bufptr64 is signed, this should be an arithmetic shift */
125         for (i = 0; i < RTE_DIST_BURST_SIZE; i++) {
126                 if (likely(buf->bufptr64[i] & RTE_DISTRIB_VALID_BUF)) {
127                         ret = buf->bufptr64[i] >> RTE_DISTRIB_FLAG_BITS;
128                         pkts[count++] = (struct rte_mbuf *)((uintptr_t)(ret));
129                 }
130         }
131
132         /*
133          * so now we've got the contents of the cacheline into an  array of
134          * mbuf pointers, so toggle the bit so scheduler can start working
135          * on the next cacheline while we're working.
136          */
137         buf->bufptr64[0] |= RTE_DISTRIB_GET_BUF;
138
139         return count;
140 }
141
142 int
143 rte_distributor_get_pkt_v1705(struct rte_distributor_v1705 *d,
144                 unsigned int worker_id, struct rte_mbuf **pkts,
145                 struct rte_mbuf **oldpkt, unsigned int return_count)
146 {
147         int count;
148
149         if (unlikely(d->alg_type == RTE_DIST_ALG_SINGLE)) {
150                 if (return_count <= 1) {
151                         pkts[0] = rte_distributor_get_pkt(d->d_v20,
152                                 worker_id, oldpkt[0]);
153                         return (pkts[0]) ? 1 : 0;
154                 } else
155                         return -EINVAL;
156         }
157
158         rte_distributor_request_pkt_v1705(d, worker_id, oldpkt, return_count);
159
160         count = rte_distributor_poll_pkt_v1705(d, worker_id, pkts);
161         while (count == -1) {
162                 uint64_t t = rte_rdtsc() + 100;
163
164                 while (rte_rdtsc() < t)
165                         rte_pause();
166
167                 count = rte_distributor_poll_pkt_v1705(d, worker_id, pkts);
168         }
169         return count;
170 }
171
172 int
173 rte_distributor_return_pkt_v1705(struct rte_distributor_v1705 *d,
174                 unsigned int worker_id, struct rte_mbuf **oldpkt, int num)
175 {
176         struct rte_distributor_buffer_v1705 *buf = &d->bufs[worker_id];
177         unsigned int i;
178
179         if (unlikely(d->alg_type == RTE_DIST_ALG_SINGLE)) {
180                 if (num == 1)
181                         return rte_distributor_return_pkt(d->d_v20,
182                                 worker_id, oldpkt[0]);
183                 else
184                         return -EINVAL;
185         }
186
187         for (i = 0; i < RTE_DIST_BURST_SIZE; i++)
188                 /* Switch off the return bit first */
189                 buf->retptr64[i] &= ~RTE_DISTRIB_RETURN_BUF;
190
191         for (i = num; i-- > 0; )
192                 buf->retptr64[i] = (((int64_t)(uintptr_t)oldpkt[i]) <<
193                         RTE_DISTRIB_FLAG_BITS) | RTE_DISTRIB_RETURN_BUF;
194
195         /* set the GET_BUF but even if we got no returns */
196         buf->retptr64[0] |= RTE_DISTRIB_GET_BUF;
197
198         return 0;
199 }
200
201 /**** APIs called on distributor core ***/
202
203 /* stores a packet returned from a worker inside the returns array */
204 static inline void
205 store_return(uintptr_t oldbuf, struct rte_distributor_v1705 *d,
206                 unsigned int *ret_start, unsigned int *ret_count)
207 {
208         if (!oldbuf)
209                 return;
210         /* store returns in a circular buffer */
211         d->returns.mbufs[(*ret_start + *ret_count) & RTE_DISTRIB_RETURNS_MASK]
212                         = (void *)oldbuf;
213         *ret_start += (*ret_count == RTE_DISTRIB_RETURNS_MASK);
214         *ret_count += (*ret_count != RTE_DISTRIB_RETURNS_MASK);
215 }
216
217 /*
218  * Match then flow_ids (tags) of the incoming packets to the flow_ids
219  * of the inflight packets (both inflight on the workers and in each worker
220  * backlog). This will then allow us to pin those packets to the relevant
221  * workers to give us our atomic flow pinning.
222  */
223 void
224 find_match_scalar(struct rte_distributor_v1705 *d,
225                         uint16_t *data_ptr,
226                         uint16_t *output_ptr)
227 {
228         struct rte_distributor_backlog *bl;
229         uint16_t i, j, w;
230
231         /*
232          * Function overview:
233          * 1. Loop through all worker ID's
234          * 2. Compare the current inflights to the incoming tags
235          * 3. Compare the current backlog to the incoming tags
236          * 4. Add any matches to the output
237          */
238
239         for (j = 0 ; j < RTE_DIST_BURST_SIZE; j++)
240                 output_ptr[j] = 0;
241
242         for (i = 0; i < d->num_workers; i++) {
243                 bl = &d->backlog[i];
244
245                 for (j = 0; j < RTE_DIST_BURST_SIZE ; j++)
246                         for (w = 0; w < RTE_DIST_BURST_SIZE; w++)
247                                 if (d->in_flight_tags[i][j] == data_ptr[w]) {
248                                         output_ptr[j] = i+1;
249                                         break;
250                                 }
251                 for (j = 0; j < RTE_DIST_BURST_SIZE; j++)
252                         for (w = 0; w < RTE_DIST_BURST_SIZE; w++)
253                                 if (bl->tags[j] == data_ptr[w]) {
254                                         output_ptr[j] = i+1;
255                                         break;
256                                 }
257         }
258
259         /*
260          * At this stage, the output contains 8 16-bit values, with
261          * each non-zero value containing the worker ID on which the
262          * corresponding flow is pinned to.
263          */
264 }
265
266
267 /*
268  * When the handshake bits indicate that there are packets coming
269  * back from the worker, this function is called to copy and store
270  * the valid returned pointers (store_return).
271  */
272 static unsigned int
273 handle_returns(struct rte_distributor_v1705 *d, unsigned int wkr)
274 {
275         struct rte_distributor_buffer_v1705 *buf = &(d->bufs[wkr]);
276         uintptr_t oldbuf;
277         unsigned int ret_start = d->returns.start,
278                         ret_count = d->returns.count;
279         unsigned int count = 0;
280         unsigned int i;
281
282         if (buf->retptr64[0] & RTE_DISTRIB_GET_BUF) {
283                 for (i = 0; i < RTE_DIST_BURST_SIZE; i++) {
284                         if (buf->retptr64[i] & RTE_DISTRIB_RETURN_BUF) {
285                                 oldbuf = ((uintptr_t)(buf->retptr64[i] >>
286                                         RTE_DISTRIB_FLAG_BITS));
287                                 /* store returns in a circular buffer */
288                                 store_return(oldbuf, d, &ret_start, &ret_count);
289                                 count++;
290                                 buf->retptr64[i] &= ~RTE_DISTRIB_RETURN_BUF;
291                         }
292                 }
293                 d->returns.start = ret_start;
294                 d->returns.count = ret_count;
295                 /* Clear for the worker to populate with more returns */
296                 buf->retptr64[0] = 0;
297         }
298         return count;
299 }
300
301 /*
302  * This function releases a burst (cache line) to a worker.
303  * It is called from the process function when a cacheline is
304  * full to make room for more packets for that worker, or when
305  * all packets have been assigned to bursts and need to be flushed
306  * to the workers.
307  * It also needs to wait for any outstanding packets from the worker
308  * before sending out new packets.
309  */
310 static unsigned int
311 release(struct rte_distributor_v1705 *d, unsigned int wkr)
312 {
313         struct rte_distributor_buffer_v1705 *buf = &(d->bufs[wkr]);
314         unsigned int i;
315
316         while (!(d->bufs[wkr].bufptr64[0] & RTE_DISTRIB_GET_BUF))
317                 rte_pause();
318
319         handle_returns(d, wkr);
320
321         buf->count = 0;
322
323         for (i = 0; i < d->backlog[wkr].count; i++) {
324                 d->bufs[wkr].bufptr64[i] = d->backlog[wkr].pkts[i] |
325                                 RTE_DISTRIB_GET_BUF | RTE_DISTRIB_VALID_BUF;
326                 d->in_flight_tags[wkr][i] = d->backlog[wkr].tags[i];
327         }
328         buf->count = i;
329         for ( ; i < RTE_DIST_BURST_SIZE ; i++) {
330                 buf->bufptr64[i] = RTE_DISTRIB_GET_BUF;
331                 d->in_flight_tags[wkr][i] = 0;
332         }
333
334         d->backlog[wkr].count = 0;
335
336         /* Clear the GET bit */
337         buf->bufptr64[0] &= ~RTE_DISTRIB_GET_BUF;
338         return  buf->count;
339
340 }
341
342
343 /* process a set of packets to distribute them to workers */
344 int
345 rte_distributor_process_v1705(struct rte_distributor_v1705 *d,
346                 struct rte_mbuf **mbufs, unsigned int num_mbufs)
347 {
348         unsigned int next_idx = 0;
349         static unsigned int wkr;
350         struct rte_mbuf *next_mb = NULL;
351         int64_t next_value = 0;
352         uint16_t new_tag = 0;
353         uint16_t flows[RTE_DIST_BURST_SIZE] __rte_cache_aligned;
354         unsigned int i, j, w, wid;
355
356         if (d->alg_type == RTE_DIST_ALG_SINGLE) {
357                 /* Call the old API */
358                 return rte_distributor_process(d->d_v20, mbufs, num_mbufs);
359         }
360
361         if (unlikely(num_mbufs == 0)) {
362                 /* Flush out all non-full cache-lines to workers. */
363                 for (wid = 0 ; wid < d->num_workers; wid++) {
364                         if ((d->bufs[wid].bufptr64[0] & RTE_DISTRIB_GET_BUF)) {
365                                 release(d, wid);
366                                 handle_returns(d, wid);
367                         }
368                 }
369                 return 0;
370         }
371
372         while (next_idx < num_mbufs) {
373                 uint16_t matches[RTE_DIST_BURST_SIZE];
374                 unsigned int pkts;
375
376                 if (d->bufs[wkr].bufptr64[0] & RTE_DISTRIB_GET_BUF)
377                         d->bufs[wkr].count = 0;
378
379                 if ((num_mbufs - next_idx) < RTE_DIST_BURST_SIZE)
380                         pkts = num_mbufs - next_idx;
381                 else
382                         pkts = RTE_DIST_BURST_SIZE;
383
384                 for (i = 0; i < pkts; i++) {
385                         if (mbufs[next_idx + i]) {
386                                 /* flows have to be non-zero */
387                                 flows[i] = mbufs[next_idx + i]->hash.usr | 1;
388                         } else
389                                 flows[i] = 0;
390                 }
391                 for (; i < RTE_DIST_BURST_SIZE; i++)
392                         flows[i] = 0;
393
394                 find_match_scalar(d, &flows[0], &matches[0]);
395
396                 /*
397                  * Matches array now contain the intended worker ID (+1) of
398                  * the incoming packets. Any zeroes need to be assigned
399                  * workers.
400                  */
401
402                 for (j = 0; j < pkts; j++) {
403
404                         next_mb = mbufs[next_idx++];
405                         next_value = (((int64_t)(uintptr_t)next_mb) <<
406                                         RTE_DISTRIB_FLAG_BITS);
407                         /*
408                          * User is advocated to set tag vaue for each
409                          * mbuf before calling rte_distributor_process.
410                          * User defined tags are used to identify flows,
411                          * or sessions.
412                          */
413                         /* flows MUST be non-zero */
414                         new_tag = (uint16_t)(next_mb->hash.usr) | 1;
415
416                         /*
417                          * Uncommenting the next line will cause the find_match
418                          * function to be optimised out, making this function
419                          * do parallel (non-atomic) distribution
420                          */
421                         /* matches[j] = 0; */
422
423                         if (matches[j]) {
424                                 struct rte_distributor_backlog *bl =
425                                                 &d->backlog[matches[j]-1];
426                                 if (unlikely(bl->count ==
427                                                 RTE_DIST_BURST_SIZE)) {
428                                         release(d, matches[j]-1);
429                                 }
430
431                                 /* Add to worker that already has flow */
432                                 unsigned int idx = bl->count++;
433
434                                 bl->tags[idx] = new_tag;
435                                 bl->pkts[idx] = next_value;
436
437                         } else {
438                                 struct rte_distributor_backlog *bl =
439                                                 &d->backlog[wkr];
440                                 if (unlikely(bl->count ==
441                                                 RTE_DIST_BURST_SIZE)) {
442                                         release(d, wkr);
443                                 }
444
445                                 /* Add to current worker worker */
446                                 unsigned int idx = bl->count++;
447
448                                 bl->tags[idx] = new_tag;
449                                 bl->pkts[idx] = next_value;
450                                 /*
451                                  * Now that we've just added an unpinned flow
452                                  * to a worker, we need to ensure that all
453                                  * other packets with that same flow will go
454                                  * to the same worker in this burst.
455                                  */
456                                 for (w = j; w < pkts; w++)
457                                         if (flows[w] == new_tag)
458                                                 matches[w] = wkr+1;
459                         }
460                 }
461                 wkr++;
462                 if (wkr >= d->num_workers)
463                         wkr = 0;
464         }
465
466         /* Flush out all non-full cache-lines to workers. */
467         for (wid = 0 ; wid < d->num_workers; wid++)
468                 if ((d->bufs[wid].bufptr64[0] & RTE_DISTRIB_GET_BUF))
469                         release(d, wid);
470
471         return num_mbufs;
472 }
473
474 /* return to the caller, packets returned from workers */
475 int
476 rte_distributor_returned_pkts_v1705(struct rte_distributor_v1705 *d,
477                 struct rte_mbuf **mbufs, unsigned int max_mbufs)
478 {
479         struct rte_distributor_returned_pkts *returns = &d->returns;
480         unsigned int retval = (max_mbufs < returns->count) ?
481                         max_mbufs : returns->count;
482         unsigned int i;
483
484         if (d->alg_type == RTE_DIST_ALG_SINGLE) {
485                 /* Call the old API */
486                 return rte_distributor_returned_pkts(d->d_v20,
487                                 mbufs, max_mbufs);
488         }
489
490         for (i = 0; i < retval; i++) {
491                 unsigned int idx = (returns->start + i) &
492                                 RTE_DISTRIB_RETURNS_MASK;
493
494                 mbufs[i] = returns->mbufs[idx];
495         }
496         returns->start += i;
497         returns->count -= i;
498
499         return retval;
500 }
501
502 /*
503  * Return the number of packets in-flight in a distributor, i.e. packets
504  * being workered on or queued up in a backlog.
505  */
506 static inline unsigned int
507 total_outstanding(const struct rte_distributor_v1705 *d)
508 {
509         unsigned int wkr, total_outstanding = 0;
510
511         for (wkr = 0; wkr < d->num_workers; wkr++)
512                 total_outstanding += d->backlog[wkr].count;
513
514         return total_outstanding;
515 }
516
517 /*
518  * Flush the distributor, so that there are no outstanding packets in flight or
519  * queued up.
520  */
521 int
522 rte_distributor_flush_v1705(struct rte_distributor_v1705 *d)
523 {
524         const unsigned int flushed = total_outstanding(d);
525         unsigned int wkr;
526
527         if (d->alg_type == RTE_DIST_ALG_SINGLE) {
528                 /* Call the old API */
529                 return rte_distributor_flush(d->d_v20);
530         }
531
532         while (total_outstanding(d) > 0)
533                 rte_distributor_process_v1705(d, NULL, 0);
534
535         /*
536          * Send empty burst to all workers to allow them to exit
537          * gracefully, should they need to.
538          */
539         rte_distributor_process_v1705(d, NULL, 0);
540
541         for (wkr = 0; wkr < d->num_workers; wkr++)
542                 handle_returns(d, wkr);
543
544         return flushed;
545 }
546
547 /* clears the internal returns array in the distributor */
548 void
549 rte_distributor_clear_returns_v1705(struct rte_distributor_v1705 *d)
550 {
551         unsigned int wkr;
552
553         if (d->alg_type == RTE_DIST_ALG_SINGLE) {
554                 /* Call the old API */
555                 rte_distributor_clear_returns(d->d_v20);
556         }
557
558         /* throw away returns, so workers can exit */
559         for (wkr = 0; wkr < d->num_workers; wkr++)
560                 d->bufs[wkr].retptr64[0] = 0;
561 }
562
563 /* creates a distributor instance */
564 struct rte_distributor_v1705 *
565 rte_distributor_create_v1705(const char *name,
566                 unsigned int socket_id,
567                 unsigned int num_workers,
568                 unsigned int alg_type)
569 {
570         struct rte_distributor_v1705 *d;
571         struct rte_dist_burst_list *dist_burst_list;
572         char mz_name[RTE_MEMZONE_NAMESIZE];
573         const struct rte_memzone *mz;
574         unsigned int i;
575
576         /* TODO Reorganise function properly around RTE_DIST_ALG_SINGLE/BURST */
577
578         /* compilation-time checks */
579         RTE_BUILD_BUG_ON((sizeof(*d) & RTE_CACHE_LINE_MASK) != 0);
580         RTE_BUILD_BUG_ON((RTE_DISTRIB_MAX_WORKERS & 7) != 0);
581
582         if (alg_type == RTE_DIST_ALG_SINGLE) {
583                 d = malloc(sizeof(struct rte_distributor_v1705));
584                 d->d_v20 = rte_distributor_create(name,
585                                 socket_id, num_workers);
586                 if (d->d_v20 == NULL) {
587                         /* rte_errno will have been set */
588                         return NULL;
589                 }
590                 d->alg_type = alg_type;
591                 return d;
592         }
593
594         if (name == NULL || num_workers >= RTE_DISTRIB_MAX_WORKERS) {
595                 rte_errno = EINVAL;
596                 return NULL;
597         }
598
599         snprintf(mz_name, sizeof(mz_name), RTE_DISTRIB_PREFIX"%s", name);
600         mz = rte_memzone_reserve(mz_name, sizeof(*d), socket_id, NO_FLAGS);
601         if (mz == NULL) {
602                 rte_errno = ENOMEM;
603                 return NULL;
604         }
605
606         d = mz->addr;
607         snprintf(d->name, sizeof(d->name), "%s", name);
608         d->num_workers = num_workers;
609         d->alg_type = alg_type;
610         d->dist_match_fn = RTE_DIST_MATCH_SCALAR;
611
612         /*
613          * Set up the backog tags so they're pointing at the second cache
614          * line for performance during flow matching
615          */
616         for (i = 0 ; i < num_workers ; i++)
617                 d->backlog[i].tags = &d->in_flight_tags[i][RTE_DIST_BURST_SIZE];
618
619         dist_burst_list = RTE_TAILQ_CAST(rte_dist_burst_tailq.head,
620                                           rte_dist_burst_list);
621
622
623         rte_rwlock_write_lock(RTE_EAL_TAILQ_RWLOCK);
624         TAILQ_INSERT_TAIL(dist_burst_list, d, next);
625         rte_rwlock_write_unlock(RTE_EAL_TAILQ_RWLOCK);
626
627         return d;
628 }