ceph/src/dpdk/examples/quota_watermark/qw/main.c

   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
  62 struct ether_fc_frame {
  63     uint16_t opcode;
  64     uint16_t param;
  65 } __attribute__((__packed__));
  66
  67
  68 int *quota;
  69 unsigned int *low_watermark;
  70
  71 uint8_t port_pairs[RTE_MAX_ETHPORTS];
  72
  73 struct rte_ring *rings[RTE_MAX_LCORE][RTE_MAX_ETHPORTS];
  74 struct rte_mempool *mbuf_pool;
  75
  76
  77 static 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  */
 121 static unsigned int
 122 get_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  */
 139 static unsigned int
 140 get_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
 151 static void
 152 receive_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
 205 static void
 206 pipeline_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
 263 static void
 264 send_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
 306 int
 307 main(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 }