1/*-
2 *   BSD LICENSE
3 *
4 *   Copyright(c) 2010-2014 Intel Corporation. All rights reserved.
5 *   All rights reserved.
6 *
7 *   Redistribution and use in source and binary forms, with or without
8 *   modification, are permitted provided that the following conditions
9 *   are met:
10 *
11 *     * Redistributions of source code must retain the above copyright
12 *       notice, this list of conditions and the following disclaimer.
13 *     * Redistributions in binary form must reproduce the above copyright
14 *       notice, this list of conditions and the following disclaimer in
15 *       the documentation and/or other materials provided with the
16 *       distribution.
17 *     * Neither the name of Intel Corporation nor the names of its
18 *       contributors may be used to endorse or promote products derived
19 *       from this software without specific prior written permission.
20 *
21 *   THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
22 *   "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
23 *   LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
24 *   A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
25 *   OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
26 *   SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
27 *   LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
28 *   DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
29 *   THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
30 *   (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
31 *   OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
32 */
33
34#include <rte_eal.h>
35
36#include <rte_common.h>
37#include <rte_debug.h>
38#include <rte_errno.h>
39#include <rte_ethdev.h>
40#include <rte_launch.h>
41#include <rte_lcore.h>
42#include <rte_log.h>
43#include <rte_mbuf.h>
44#include <rte_ring.h>
45
46#include <rte_byteorder.h>
47
48#include "args.h"
49#include "main.h"
50#include "init.h"
51#include "../include/conf.h"
52
53
54#ifdef QW_SOFTWARE_FC
55#define SEND_PAUSE_FRAME(port_id, duration) send_pause_frame(port_id, duration)
56#else
57#define SEND_PAUSE_FRAME(port_id, duration) do { } while(0)
58#endif
59
60#define ETHER_TYPE_FLOW_CONTROL 0x8808
61
62struct ether_fc_frame {
63    uint16_t opcode;
64    uint16_t param;
65} __attribute__((__packed__));
66
67
68int *quota;
69unsigned int *low_watermark;
70
71uint8_t port_pairs[RTE_MAX_ETHPORTS];
72
73struct rte_ring *rings[RTE_MAX_LCORE][RTE_MAX_ETHPORTS];
74struct rte_mempool *mbuf_pool;
75
76
77static void send_pause_frame(uint8_t port_id, uint16_t duration)
78{
79    struct rte_mbuf *mbuf;
80    struct ether_fc_frame *pause_frame;
81    struct ether_hdr *hdr;
82    struct ether_addr mac_addr;
83
84    RTE_LOG(DEBUG, USER1, "Sending PAUSE frame (duration=%d) on port %d\n",
85            duration, port_id);
86
87    /* Get a mbuf from the pool */
88    mbuf = rte_pktmbuf_alloc(mbuf_pool);
89    if (unlikely(mbuf == NULL))
90        return;
91
92    /* Prepare a PAUSE frame */
93    hdr = rte_pktmbuf_mtod(mbuf, struct ether_hdr *);
94    pause_frame = (struct ether_fc_frame *) &hdr[1];
95
96    rte_eth_macaddr_get(port_id, &mac_addr);
97    ether_addr_copy(&mac_addr, &hdr->s_addr);
98
99    void *tmp = &hdr->d_addr.addr_bytes[0];
100    *((uint64_t *)tmp) = 0x010000C28001ULL;
101
102    hdr->ether_type = rte_cpu_to_be_16(ETHER_TYPE_FLOW_CONTROL);
103
104    pause_frame->opcode = rte_cpu_to_be_16(0x0001);
105    pause_frame->param  = rte_cpu_to_be_16(duration);
106
107    mbuf->pkt_len  = 60;
108    mbuf->data_len = 60;
109
110    rte_eth_tx_burst(port_id, 0, &mbuf, 1);
111}
112
113/**
114 * Get the previous enabled lcore ID
115 *
116 * @param lcore_id
117 *   The current lcore ID.
118 * @return
119 *   The previous enabled lcore_id or -1 if not found.
120 */
121static unsigned int
122get_previous_lcore_id(unsigned int lcore_id)
123{
124    int i;
125
126    for (i = lcore_id - 1; i >= 0; i--)
127        if (rte_lcore_is_enabled(i))
128            return i;
129
130    return -1;
131}
132
133/**
134 * Get the last enabled lcore ID
135 *
136 * @return
137 *   The last enabled lcore_id.
138 */
139static unsigned int
140get_last_lcore_id(void)
141{
142    int i;
143
144    for (i = RTE_MAX_LCORE; i >= 0; i--)
145        if (rte_lcore_is_enabled(i))
146            return i;
147
148    return 0;
149}
150
151static void
152receive_stage(__attribute__((unused)) void *args)
153{
154    int i, ret;
155
156    uint8_t port_id;
157    uint16_t nb_rx_pkts;
158
159    unsigned int lcore_id;
160
161    struct rte_mbuf *pkts[MAX_PKT_QUOTA];
162    struct rte_ring *ring;
163    enum ring_state ring_state[RTE_MAX_ETHPORTS] = { RING_READY };
164
165    lcore_id = rte_lcore_id();
166
167    RTE_LOG(INFO, USER1,
168            "%s() started on core %u\n", __func__, lcore_id);
169
170    while (1) {
171
172        /* Process each port round robin style */
173        for (port_id = 0; port_id < RTE_MAX_ETHPORTS; port_id++) {
174
175            if (!is_bit_set(port_id, portmask))
176                continue;
177
178            ring = rings[lcore_id][port_id];
179
180            if (ring_state[port_id] != RING_READY) {
181                if (rte_ring_count(ring) > *low_watermark)
182                    continue;
183                else
184                    ring_state[port_id] = RING_READY;
185            }
186
187            /* Enqueue received packets on the RX ring */
188            nb_rx_pkts = rte_eth_rx_burst(port_id, 0, pkts, (uint16_t) *quota);
189            ret = rte_ring_enqueue_bulk(ring, (void *) pkts, nb_rx_pkts);
190            if (ret == -EDQUOT) {
191                ring_state[port_id] = RING_OVERLOADED;
192                send_pause_frame(port_id, 1337);
193            }
194
195            else if (ret == -ENOBUFS) {
196
197                /* Return  mbufs to the pool, effectively dropping packets */
198                for (i = 0; i < nb_rx_pkts; i++)
199                    rte_pktmbuf_free(pkts[i]);
200            }
201        }
202    }
203}
204
205static void
206pipeline_stage(__attribute__((unused)) void *args)
207{
208    int i, ret;
209    int nb_dq_pkts;
210
211    uint8_t port_id;
212
213    unsigned int lcore_id, previous_lcore_id;
214
215    void *pkts[MAX_PKT_QUOTA];
216    struct rte_ring *rx, *tx;
217    enum ring_state ring_state[RTE_MAX_ETHPORTS] = { RING_READY };
218
219    lcore_id = rte_lcore_id();
220    previous_lcore_id = get_previous_lcore_id(lcore_id);
221
222    RTE_LOG(INFO, USER1,
223            "%s() started on core %u - processing packets from core %u\n",
224            __func__, lcore_id, previous_lcore_id);
225
226    while (1) {
227
228        for (port_id = 0; port_id < RTE_MAX_ETHPORTS; port_id++) {
229
230            if (!is_bit_set(port_id, portmask))
231                continue;
232
233            tx = rings[lcore_id][port_id];
234            rx = rings[previous_lcore_id][port_id];
235
236            if (ring_state[port_id] != RING_READY) {
237                if (rte_ring_count(tx) > *low_watermark)
238                    continue;
239                else
240                    ring_state[port_id] = RING_READY;
241            }
242
243            /* Dequeue up to quota mbuf from rx */
244            nb_dq_pkts = rte_ring_dequeue_burst(rx, pkts, *quota);
245            if (unlikely(nb_dq_pkts < 0))
246                continue;
247
248            /* Enqueue them on tx */
249            ret = rte_ring_enqueue_bulk(tx, pkts, nb_dq_pkts);
250            if (ret == -EDQUOT)
251                ring_state[port_id] = RING_OVERLOADED;
252
253            else if (ret == -ENOBUFS) {
254
255                /* Return  mbufs to the pool, effectively dropping packets */
256                for (i = 0; i < nb_dq_pkts; i++)
257                    rte_pktmbuf_free(pkts[i]);
258            }
259        }
260    }
261}
262
263static void
264send_stage(__attribute__((unused)) void *args)
265{
266	uint16_t nb_dq_pkts;
267
268    uint8_t port_id;
269    uint8_t dest_port_id;
270
271    unsigned int lcore_id, previous_lcore_id;
272
273    struct rte_ring *tx;
274    struct rte_mbuf *tx_pkts[MAX_PKT_QUOTA];
275
276    lcore_id = rte_lcore_id();
277    previous_lcore_id = get_previous_lcore_id(lcore_id);
278
279    RTE_LOG(INFO, USER1,
280            "%s() started on core %u - processing packets from core %u\n",
281            __func__, lcore_id, previous_lcore_id);
282
283    while (1) {
284
285        /* Process each ring round robin style */
286        for (port_id = 0; port_id < RTE_MAX_ETHPORTS; port_id++) {
287
288            if (!is_bit_set(port_id, portmask))
289                continue;
290
291            dest_port_id = port_pairs[port_id];
292            tx = rings[previous_lcore_id][port_id];
293
294            if (rte_ring_empty(tx))
295                continue;
296
297            /* Dequeue packets from tx and send them */
298            nb_dq_pkts = (uint16_t) rte_ring_dequeue_burst(tx, (void *) tx_pkts, *quota);
299            rte_eth_tx_burst(dest_port_id, 0, tx_pkts, nb_dq_pkts);
300
301            /* TODO: Check if nb_dq_pkts == nb_tx_pkts? */
302        }
303    }
304}
305
306int
307main(int argc, char **argv)
308{
309    int ret;
310    unsigned int lcore_id, master_lcore_id, last_lcore_id;
311
312    uint8_t port_id;
313
314    rte_set_log_level(RTE_LOG_INFO);
315
316    ret = rte_eal_init(argc, argv);
317    if (ret < 0)
318        rte_exit(EXIT_FAILURE, "Cannot initialize EAL\n");
319
320    argc -= ret;
321    argv += ret;
322
323    init_dpdk();
324    setup_shared_variables();
325
326    *quota = 32;
327    *low_watermark = 60 * RING_SIZE / 100;
328
329    last_lcore_id   = get_last_lcore_id();
330    master_lcore_id = rte_get_master_lcore();
331
332    /* Parse the application's arguments */
333    ret = parse_qw_args(argc, argv);
334    if (ret < 0)
335        rte_exit(EXIT_FAILURE, "Invalid quota/watermark argument(s)\n");
336
337    /* Create a pool of mbuf to store packets */
338    mbuf_pool = rte_pktmbuf_pool_create("mbuf_pool", MBUF_PER_POOL, 32, 0,
339	    MBUF_DATA_SIZE, rte_socket_id());
340    if (mbuf_pool == NULL)
341        rte_panic("%s\n", rte_strerror(rte_errno));
342
343    for (port_id = 0; port_id < RTE_MAX_ETHPORTS; port_id++)
344        if (is_bit_set(port_id, portmask)) {
345            configure_eth_port(port_id);
346            init_ring(master_lcore_id, port_id);
347        }
348
349    pair_ports();
350
351    /* Start pipeline_connect() on all the available slave lcore but the last */
352    for (lcore_id = 0 ; lcore_id < last_lcore_id; lcore_id++) {
353        if (rte_lcore_is_enabled(lcore_id) && lcore_id != master_lcore_id) {
354
355            for (port_id = 0; port_id < RTE_MAX_ETHPORTS; port_id++)
356                if (is_bit_set(port_id, portmask))
357                    init_ring(lcore_id, port_id);
358
359            /* typecast is a workaround for GCC 4.3 bug */
360            rte_eal_remote_launch((int (*)(void *))pipeline_stage, NULL, lcore_id);
361        }
362    }
363
364    /* Start send_stage() on the last slave core */
365    /* typecast is a workaround for GCC 4.3 bug */
366    rte_eal_remote_launch((int (*)(void *))send_stage, NULL, last_lcore_id);
367
368    /* Start receive_stage() on the master core */
369    receive_stage(NULL);
370
371    return 0;
372}
373