forwarder.cc

#include "xellico.h"
#include "config.h"
#include "forwarder.h"
#include "force_quit.h"
#include "delay.h"
#include "config.h"

uint32_t l2fwd_dst_ports[RTE_MAX_ETHPORTS];

static inline void
l2fwd_simple_forward (struct rte_mbuf *m, unsigned portid)
{
  uint32_t lcore_id = rte_lcore_id();
  unsigned dst_port = l2fwd_dst_ports[portid];
  unsigned dst_queue = lcore_id;
  struct rte_eth_dev_tx_buffer* buffer = xeconf->lcore_conf[lcore_id].tx_buffer[dst_port];
  rte_eth_tx_buffer(dst_port, dst_queue, buffer, m);
}

void
forwarder (void)
{
  unsigned lcore_id = rte_lcore_id ();
  const size_t n_burst = xeconf->rx_burst_size;
  if (xeconf->lcore_conf[lcore_id].qconf.size() == 0)
    {
      RTE_LOG (INFO, XELLICO, " -- lcore=%u(socket%u) has nothing to do\n",
          lcore_id, rte_socket_id());
      return;
    }

  for (size_t i = 0; i < xeconf->lcore_conf[lcore_id].qconf.size(); i++)
    {
      uint32_t portid = xeconf->lcore_conf[lcore_id].qconf[i].port_id;
      uint32_t queueid = xeconf->lcore_conf[lcore_id].qconf[i].queue_id;
      RTE_LOG (INFO, XELLICO,
          " -- lcore=%u(socket%u) portid=%u queueid=%u\n",
          lcore_id, rte_socket_id(), portid, queueid);
    }

  uint64_t prev_tsc = 0;
  const uint64_t drain_tsc =
      (rte_get_tsc_hz () + US_PER_S - 1)
      / US_PER_S * BURST_TX_DRAIN_US;

  while (!force_quit)
    {
      /*
       * TX burst queue drain
       */
      uint64_t cur_tsc = rte_rdtsc ();
      uint64_t diff_tsc = cur_tsc - prev_tsc;
      if (unlikely (diff_tsc > drain_tsc))
        {
          for (size_t i = 0; i < xeconf->lcore_conf[lcore_id].qconf.size(); i++)
            {
              uint32_t dst_portid =
                l2fwd_dst_ports[xeconf->lcore_conf[lcore_id].qconf[i].port_id];
              uint32_t dst_queueid = rte_lcore_id();
              struct rte_eth_dev_tx_buffer *buffer =
                xeconf->lcore_conf[lcore_id].tx_buffer[dst_portid];

#if 0
              static size_t sum = 0, cnt = 0;
              size_t before = rte_rdtsc ();
#endif
              rte_eth_tx_buffer_flush (dst_portid, dst_queueid, buffer);
#if 0
              sum += rte_rdtsc () - before;
              cnt ++;
              if (cnt > 1000*10)
                {
                  size_t lat_clk = sum / cnt;
                  printf ("lat: %zd [clock] on lcore%u\n", lat_clk, rte_lcore_id ());
                  sum = 0;
                  cnt = 0;
                }
#endif
            }
          prev_tsc = cur_tsc;
        }

      /*
       * Read packet from RX queues
       */
      for (size_t i = 0; i < xeconf->lcore_conf[lcore_id].qconf.size(); i++)
        {
          struct rte_mbuf *pkts_burst[n_burst];
          uint32_t in_portid = xeconf->lcore_conf[lcore_id].qconf[i].port_id;
          uint32_t in_queueid = xeconf->lcore_conf[lcore_id].qconf[i].queue_id;
          uint32_t nb_rx = rte_eth_rx_burst (in_portid,
              in_queueid, pkts_burst, n_burst);
          for (size_t j = 0; j < nb_rx; j++)
            {
              struct rte_mbuf *m = pkts_burst[j];
              rte_prefetch0 (rte_pktmbuf_mtod (m, void *));
              // dirty_looped_delay (150);
              rte_delay_us_block (1);
              l2fwd_simple_forward (m, in_portid);
            }
        }
    }
}

void
init_fib (void)
{
  /*
   * Each logical core is assigned a dedicated TX queue on each port.
   */
  const size_t nb_ports = rte_eth_dev_count ();
  uint8_t last_port = 0;
  unsigned nb_ports_in_mask = 0;
  memset (l2fwd_dst_ports, 0x0, sizeof (l2fwd_dst_ports));
  for (uint8_t portid = 0; portid < nb_ports; portid++)
    {
      if (nb_ports_in_mask % 2)
        {
          l2fwd_dst_ports[portid] = last_port;
          l2fwd_dst_ports[last_port] = portid;
        }
      else
        last_port = portid;

      nb_ports_in_mask++;
    }
  if (nb_ports_in_mask % 2)
    {
      printf("Notice: odd number of ports in portmask.\n");
      l2fwd_dst_ports[last_port] = last_port;
    }
}