examples/quota_watermark/qw/main.c

   1 /*-
   2  *   BSD LICENSE
   3  *
   4  *   Copyright(c) 2010-2013 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
  35 #include <rte_eal.h>
  36
  37 #include <rte_common.h>
  38 #include <rte_debug.h>
  39 #include <rte_errno.h>
  40 #include <rte_ethdev.h>
  41 #include <rte_launch.h>
  42 #include <rte_lcore.h>
  43 #include <rte_log.h>
  44 #include <rte_mbuf.h>
  45 #include <rte_ring.h>
  46
  47 #include <rte_byteorder.h>
  48
  49 #include "args.h"
  50 #include "main.h"
  51 #include "init.h"
  52 #include "../include/conf.h"
  53
  54
  55 #ifdef QW_SOFTWARE_FC
  56 #define SEND_PAUSE_FRAME(port_id, duration) send_pause_frame(port_id, duration)
  57 #else
  58 #define SEND_PAUSE_FRAME(port_id, duration) do { } while(0)
  59 #endif
  60
  61 #define ETHER_TYPE_FLOW_CONTROL 0x8808
  62
  63 struct ether_fc_frame {
  64     uint16_t opcode;
  65     uint16_t param;
  66 } __attribute__((__packed__));
  67
  68
  69 int *quota;
  70 unsigned int *low_watermark;
  71
  72 uint8_t port_pairs[RTE_MAX_ETHPORTS];
  73
  74 struct rte_ring *rings[RTE_MAX_LCORE][RTE_MAX_ETHPORTS];
  75 struct rte_mempool *mbuf_pool;
  76
  77
  78 static void send_pause_frame(uint8_t port_id, uint16_t duration)
  79 {
  80     struct rte_mbuf *mbuf;
  81     struct ether_fc_frame *pause_frame;
  82     struct ether_hdr *hdr;
  83     struct ether_addr mac_addr;
  84
  85     RTE_LOG(DEBUG, USER1, "Sending PAUSE frame (duration=%d) on port %d\n",
  86             duration, port_id);
  87
  88     /* Get a mbuf from the pool */
  89     mbuf = rte_pktmbuf_alloc(mbuf_pool);
  90     if (unlikely(mbuf == NULL))
  91         return;
  92
  93     /* Prepare a PAUSE frame */
  94     hdr = rte_pktmbuf_mtod(mbuf, struct ether_hdr *);
  95     pause_frame = (struct ether_fc_frame *) &hdr[1];
  96
  97     rte_eth_macaddr_get(port_id, &mac_addr);
  98     ether_addr_copy(&mac_addr, &hdr->s_addr);
  99
 100     void *tmp = &hdr->d_addr.addr_bytes[0];
 101     *((uint64_t *)tmp) = 0x010000C28001;
 102
 103     hdr->ether_type = rte_cpu_to_be_16(ETHER_TYPE_FLOW_CONTROL);
 104
 105     pause_frame->opcode = rte_cpu_to_be_16(0x0001);
 106     pause_frame->param  = rte_cpu_to_be_16(duration);
 107
 108     mbuf->pkt.pkt_len  = 60;
 109     mbuf->pkt.data_len = 60;
 110
 111     rte_eth_tx_burst(port_id, 0, &mbuf, 1);
 112 }
 113
 114 /**
 115  * Get the previous enabled lcore ID
 116  *
 117  * @param lcore_id
 118  *   The current lcore ID.
 119  * @return
 120  *   The previous enabled lcore_id or -1 if not found.
 121  */
 122 static unsigned int
 123 get_previous_lcore_id(unsigned int lcore_id)
 124 {
 125     int i;
 126
 127     for (i = lcore_id - 1; i >= 0; i--)
 128         if (rte_lcore_is_enabled(i))
 129             return i;
 130
 131     return -1;
 132 }
 133
 134 /**
 135  * Get the last enabled lcore ID
 136  *
 137  * @return
 138  *   The last enabled lcore_id.
 139  */
 140 static unsigned int
 141 get_last_lcore_id(void)
 142 {
 143     int i;
 144
 145     for (i = RTE_MAX_LCORE; i >= 0; i--)
 146         if (rte_lcore_is_enabled(i))
 147             return i;
 148
 149     return 0;
 150 }
 151
 152 static void
 153 receive_stage(__attribute__((unused)) void *args)
 154 {
 155     int i, ret;
 156
 157     uint8_t port_id;
 158     uint16_t nb_rx_pkts;
 159
 160     unsigned int lcore_id;
 161
 162     struct rte_mbuf *pkts[MAX_PKT_QUOTA];
 163     struct rte_ring *ring;
 164     enum ring_state ring_state[RTE_MAX_ETHPORTS] = { RING_READY };
 165
 166     lcore_id = rte_lcore_id();
 167
 168     RTE_LOG(INFO, USER1,
 169             "%s() started on core %u\n", __func__, lcore_id);
 170
 171     while (1) {
 172
 173         /* Process each port round robin style */
 174         for (port_id = 0; port_id < RTE_MAX_ETHPORTS; port_id++) {
 175
 176             if (!is_bit_set(port_id, portmask))
 177                 continue;
 178
 179             ring = rings[lcore_id][port_id];
 180
 181             if (ring_state[port_id] != RING_READY) {
 182                 if (rte_ring_count(ring) > *low_watermark)
 183                     continue;
 184                 else
 185                     ring_state[port_id] = RING_READY;
 186             }
 187
 188             /* Enqueue received packets on the RX ring */
 189             nb_rx_pkts = rte_eth_rx_burst(port_id, 0, pkts, (uint16_t) *quota);
 190             ret = rte_ring_enqueue_bulk(ring, (void *) pkts, nb_rx_pkts);
 191             if (ret == -EDQUOT) {
 192                 ring_state[port_id] = RING_OVERLOADED;
 193                 send_pause_frame(port_id, 1337);
 194             }
 195
 196             else if (ret == -ENOBUFS) {
 197
 198                 /* Return  mbufs to the pool, effectively dropping packets */
 199                 for (i = 0; i < nb_rx_pkts; i++)
 200                     rte_pktmbuf_free(pkts[i]);
 201             }
 202         }
 203     }
 204 }
 205
 206 static void
 207 pipeline_stage(__attribute__((unused)) void *args)
 208 {
 209     int i, ret;
 210     int nb_dq_pkts;
 211
 212     uint8_t port_id;
 213
 214     unsigned int lcore_id, previous_lcore_id;
 215
 216     void *pkts[MAX_PKT_QUOTA];
 217     struct rte_ring *rx, *tx;
 218     enum ring_state ring_state[RTE_MAX_ETHPORTS] = { RING_READY };
 219
 220     lcore_id = rte_lcore_id();
 221     previous_lcore_id = get_previous_lcore_id(lcore_id);
 222
 223     RTE_LOG(INFO, USER1,
 224             "%s() started on core %u - processing packets from core %u\n",
 225             __func__, lcore_id, previous_lcore_id);
 226
 227     while (1) {
 228
 229         for (port_id = 0; port_id < RTE_MAX_ETHPORTS; port_id++) {
 230
 231             if (!is_bit_set(port_id, portmask))
 232                 continue;
 233
 234             tx = rings[lcore_id][port_id];
 235             rx = rings[previous_lcore_id][port_id];
 236
 237             if (ring_state[port_id] != RING_READY) {
 238                 if (rte_ring_count(tx) > *low_watermark)
 239                     continue;
 240                 else
 241                     ring_state[port_id] = RING_READY;
 242             }
 243
 244             /* Dequeue up to quota mbuf from rx */
 245             nb_dq_pkts = rte_ring_dequeue_burst(rx, pkts, *quota);
 246             if (unlikely(nb_dq_pkts < 0))
 247                 continue;
 248
 249             /* Enqueue them on tx */
 250             ret = rte_ring_enqueue_bulk(tx, pkts, nb_dq_pkts);
 251             if (ret == -EDQUOT)
 252                 ring_state[port_id] = RING_OVERLOADED;
 253
 254             else if (ret == -ENOBUFS) {
 255
 256                 /* Return  mbufs to the pool, effectively dropping packets */
 257                 for (i = 0; i < nb_dq_pkts; i++)
 258                     rte_pktmbuf_free(pkts[i]);
 259             }
 260         }
 261     }
 262 }
 263
 264 static void
 265 send_stage(__attribute__((unused)) void *args)
 266 {
 267         uint16_t nb_dq_pkts;
 268
 269     uint8_t port_id;
 270     uint8_t dest_port_id;
 271
 272     unsigned int lcore_id, previous_lcore_id;
 273
 274     struct rte_ring *tx;
 275     struct rte_mbuf *tx_pkts[MAX_PKT_QUOTA];
 276
 277     lcore_id = rte_lcore_id();
 278     previous_lcore_id = get_previous_lcore_id(lcore_id);
 279
 280     RTE_LOG(INFO, USER1,
 281             "%s() started on core %u - processing packets from core %u\n",
 282             __func__, lcore_id, previous_lcore_id);
 283
 284     while (1) {
 285
 286         /* Process each ring round robin style */
 287         for (port_id = 0; port_id < RTE_MAX_ETHPORTS; port_id++) {
 288
 289             if (!is_bit_set(port_id, portmask))
 290                 continue;
 291
 292             dest_port_id = port_pairs[port_id];
 293             tx = rings[previous_lcore_id][port_id];
 294
 295             if (rte_ring_empty(tx))
 296                 continue;
 297
 298             /* Dequeue packets from tx and send them */
 299             nb_dq_pkts = (uint16_t) rte_ring_dequeue_burst(tx, (void *) tx_pkts, *quota);
 300             rte_eth_tx_burst(dest_port_id, 0, tx_pkts, nb_dq_pkts);
 301
 302             /* TODO: Check if nb_dq_pkts == nb_tx_pkts? */
 303         }
 304     }
 305 }
 306
 307 int
 308 MAIN(int argc, char **argv)
 309 {
 310     int ret;
 311     unsigned int lcore_id, master_lcore_id, last_lcore_id;
 312
 313     uint8_t port_id;
 314
 315     rte_set_log_level(RTE_LOG_INFO);
 316
 317     ret = rte_eal_init(argc, argv);
 318     if (ret < 0)
 319         rte_exit(EXIT_FAILURE, "Cannot initialize EAL\n");
 320
 321     argc -= ret;
 322     argv += ret;
 323
 324     init_dpdk();
 325     setup_shared_variables();
 326
 327     *quota = 32;
 328     *low_watermark = 60 * RING_SIZE / 100;
 329
 330     last_lcore_id   = get_last_lcore_id();
 331     master_lcore_id = rte_get_master_lcore();
 332
 333     /* Parse the application's arguments */
 334     ret = parse_qw_args(argc, argv);
 335     if (ret < 0)
 336         rte_exit(EXIT_FAILURE, "Invalid quota/watermark argument(s)\n");
 337
 338     /* Create a pool of mbuf to store packets */
 339     mbuf_pool = rte_mempool_create("mbuf_pool", MBUF_PER_POOL, MBUF_SIZE, 32,
 340                                    sizeof(struct rte_pktmbuf_pool_private),
 341                                    rte_pktmbuf_pool_init, NULL,
 342                                    rte_pktmbuf_init, NULL,
 343                                    rte_socket_id(), 0);
 344     if (mbuf_pool == NULL)
 345         rte_panic("%s\n", rte_strerror(rte_errno));
 346
 347     for (port_id = 0; port_id < RTE_MAX_ETHPORTS; port_id++)
 348         if (is_bit_set(port_id, portmask)) {
 349             configure_eth_port(port_id);
 350             init_ring(master_lcore_id, port_id);
 351         }
 352
 353     pair_ports();
 354
 355     /* Start pipeline_connect() on all the available slave lcore but the last */
 356     for (lcore_id = 0 ; lcore_id < last_lcore_id; lcore_id++) {
 357         if (rte_lcore_is_enabled(lcore_id) && lcore_id != master_lcore_id) {
 358
 359             for (port_id = 0; port_id < RTE_MAX_ETHPORTS; port_id++)
 360                 if (is_bit_set(port_id, portmask))
 361                     init_ring(lcore_id, port_id);
 362
 363             /* typecast is a workaround for GCC 4.3 bug */
 364             rte_eal_remote_launch((int (*)(void *))pipeline_stage, NULL, lcore_id);
 365         }
 366     }
 367
 368     /* Start send_stage() on the last slave core */
 369     /* typecast is a workaround for GCC 4.3 bug */
 370     rte_eal_remote_launch((int (*)(void *))send_stage, NULL, last_lcore_id);
 371
 372     /* Start receive_stage() on the master core */
 373     receive_stage(NULL);
 374
 375     return 0;
 376 }