<commit_before>/*
 * (C) 2018 - ntop
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in
 * all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 * THE SOFTWARE.
 */

#include <sched.h>
#include <stdlib.h>
#include <unistd.h>
#include <stdio.h>
#include <assert.h>
#include <inttypes.h>
#include <getopt.h>
#include <arpa/inet.h>
#include <sys/time.h>
#include <signal.h>

#include <rte_eal.h>
#include <rte_ether.h>
#include <rte_ethdev.h>
#include <rte_cycles.h>
#include <rte_lcore.h>
#include <rte_mbuf.h>

#define ALARM_SLEEP        1

/* NOTE: ether_hdr is defined in rte_ether.h */
#define ether_header ether_hdr
#include "ftutils.c"

#include "pfring_ft.h"

#define RX_RING_SIZE     8192
#define TX_RING_SIZE     1024
#define MBUF_CACHE_SIZE   256
#define BURST_SIZE         32
#define PREFETCH_OFFSET     3
#define TX_TEST_PKT_LEN    60

//#define SCATTERED_RX_TEST
#ifdef SCATTERED_RX_TEST
#define MBUF_BUF_SIZE     512
#else
#define MBUF_BUF_SIZE    RTE_MBUF_DEFAULT_BUF_SIZE
#endif

#define print_mac_addr(addr) printf("%02"PRIx8":%02"PRIx8":%02"PRIx8":%02"PRIx8":%02"PRIx8":%02"PRIx8, \
  addr.addr_bytes[0], addr.addr_bytes[1], addr.addr_bytes[2], addr.addr_bytes[3], addr.addr_bytes[4], addr.addr_bytes[5])

static struct rte_mempool *mbuf_pool[RTE_MAX_LCORE] = { NULL };
static pfring_ft_table *fts[RTE_MAX_LCORE] = { NULL };
static u_int32_t ft_flags = 0;
static u_int8_t port = 0, twin_port = 0xFF;
static u_int8_t num_queues = 1;
static u_int8_t compute_flows = 1;
static u_int8_t do_loop = 1;
static u_int8_t verbose = 0;
static u_int8_t fwd = 0;
static u_int8_t test_tx = 0;
static u_int16_t tx_test_pkt_len = TX_TEST_PKT_LEN;
static u_int32_t num_mbufs_per_lcore = 0;
static u_int32_t pps = 0;

static struct lcore_stats {
  u_int64_t num_pkts;
  u_int64_t num_bytes;
  u_int64_t last_pkts;
  u_int64_t last_bytes;
  u_int64_t tx_num_pkts;
  u_int64_t tx_drops;
  u_int64_t padding[2];
} stats[RTE_MAX_LCORE];

static const struct rte_eth_conf port_conf_default = {
  .rxmode = { .max_rx_pkt_len = ETHER_MAX_LEN }
};

/* ************************************ */

static int port_init(void) {
  struct rte_eth_conf port_conf = port_conf_default;
  int retval, i;
  u_int16_t q;
  char name[64];
  int num_ports;

  if (twin_port == 0xFF || twin_port == port) num_ports = 1;
  else num_ports = 2;

  num_mbufs_per_lcore = 2 * (
    RX_RING_SIZE + 
    TX_RING_SIZE + 
    BURST_SIZE * 2) 
#ifdef SCATTERED_RX_TEST
    * 4
#endif
  ;

  for (q = 0; q < num_queues; q++) {
    snprintf(name, sizeof(name), "MBUF_POOL_%u", q);

    mbuf_pool[q] = rte_pktmbuf_pool_create(name, num_mbufs_per_lcore, MBUF_CACHE_SIZE, 0, 
      MBUF_BUF_SIZE, rte_socket_id());

    if (mbuf_pool[q] == NULL)
      rte_exit(EXIT_FAILURE, "Cannot create mbuf pool: %s\n", rte_strerror(rte_errno));
  }

  for (i = 0; i < num_ports; i++) {
    u_int8_t port_id = (i == 0) ? port : twin_port;
    unsigned int numa_socket_id;
    
    printf("Configuring port %u...\n", port_id);

    retval = rte_eth_dev_configure(port_id, num_queues /* RX */, num_queues /* TX */, &port_conf);
    
    if (retval != 0)
      return retval;    

    numa_socket_id = rte_eth_dev_socket_id(port_id);
    
    for (q = 0; q < num_queues; q++) {

      printf("Configuring queue %u...\n", q);

      retval = rte_eth_rx_queue_setup(port_id, q, RX_RING_SIZE, numa_socket_id, NULL, mbuf_pool[q]);

      if (retval < 0)
	return retval;

      retval = rte_eth_tx_queue_setup(port_id, q, TX_RING_SIZE, numa_socket_id, NULL);

      if (retval < 0)
	return retval;
    }

    retval = rte_eth_dev_start(port_id);

    if (retval < 0)
      return retval;

    rte_eth_promiscuous_enable(port_id);
  }
  
  return 0;
}

/* ************************************ */

static void port_close(void) {
  int i;

  for (i = 0; i < 2; i++) {
    u_int8_t port_id = (i == 0) ? port : twin_port;
    
    if (port_id == 0xFF || (i == 1 && twin_port == port)) break;

    printf("Releasing port %u...\n", port_id);

    rte_eth_dev_stop(port_id);
    rte_eth_dev_close(port_id);
  }
}

/* ************************************ */

void processFlow(pfring_ft_flow *flow, void *user) {
  pfring_ft_table *ft = (pfring_ft_table *) user;
  pfring_ft_flow_key *k;
  pfring_ft_flow_value *v;
  char buf1[32], buf2[32], buf3[32];
  const char *ip1, *ip2;

  k = pfring_ft_flow_get_key(flow);
  v = pfring_ft_flow_get_value(flow);

  if (k->ip_version == 4){
    ip1 = _intoa(k->saddr.v4, buf1, sizeof(buf1));
    ip2 = _intoa(k->daddr.v4, buf2, sizeof(buf2));
  } else {
    ip1 = inet_ntop(AF_INET6, &k->saddr.v6, buf1, sizeof(buf1));
    ip2 = inet_ntop(AF_INET6, &k->daddr.v6, buf2, sizeof(buf2));
  }

  printf("[Flow] "
         "srcIp: %s, dstIp: %s, srcPort: %u, dstPort: %u, protocol: %u, tcpFlags: 0x%02X, "
         "l7: %s, "
         "c2s: { Packets: %ju, Bytes: %ju, First: %u.%u, Last: %u.%u }, "
         "s2c: { Packets: %ju, Bytes: %ju, First: %u.%u, Last: %u.%u }\n",
         ip1, ip2, k->sport, k->dport, k->protocol, v->direction[s2d_direction].tcp_flags | v->direction[d2s_direction].tcp_flags,
         pfring_ft_l7_protocol_name(ft, &v->l7_protocol, buf3, sizeof(buf3)),
         v->direction[s2d_direction].pkts, v->direction[s2d_direction].bytes, 
         (u_int) v->direction[s2d_direction].first.tv_sec, (u_int) v->direction[s2d_direction].first.tv_usec, 
         (u_int) v->direction[s2d_direction].last.tv_sec,  (u_int) v->direction[s2d_direction].last.tv_usec,
         v->direction[d2s_direction].pkts, v->direction[d2s_direction].bytes, 
         (u_int) v->direction[d2s_direction].first.tv_sec, (u_int) v->direction[d2s_direction].first.tv_usec, 
         (u_int) v->direction[d2s_direction].last.tv_sec,  (u_int) v->direction[d2s_direction].last.tv_usec);

  pfring_ft_flow_free(flow);
}

/* ************************************ */

static void tx_test(u_int16_t queue_id) {
  struct rte_mbuf *tx_bufs[BURST_SIZE];
  u_int32_t i;
  u_int16_t sent;
  int rc;
#if !(defined(__arm__) || defined(__mips__))
  ticks tick_start = 0, tick_delta = 0;

  if (pps != 0) {
    double td;
    ticks hz = 0;

    /* computing usleep delay */
    tick_start = getticks();
    usleep(1);
    tick_delta = getticks() - tick_start;

    /* computing CPU freq */
    tick_start = getticks();
    usleep(1001);
    hz = (getticks() - tick_start - tick_delta) * 1000 /*kHz -> Hz*/;
    printf("Estimated CPU freq: %lu Hz\n", (long unsigned int)hz);

    td = (double) (hz / pps);
    tick_delta = (ticks) td;
    printf("Rate set to %u pps\n", pps);
  }
#endif

  printf("Generating traffic on port %u queue %u...\n", port, queue_id);

  num_ips = TX_RING_SIZE;

  while (do_loop) {

    rc = rte_mempool_get_bulk(mbuf_pool[queue_id], (void **) tx_bufs, BURST_SIZE);

    if (rc) {
      fprintf(stderr, "rte_mempool_get_bulk error (%d)\n", rc);
      return;
    }
  
    for (i = 0; i < BURST_SIZE; i++) {
      if (tx_bufs[i]->data_len != tx_test_pkt_len) {
        forge_udp_packet_fast((u_char *) rte_pktmbuf_mtod(tx_bufs[i], char *), 
          tx_test_pkt_len, stats[queue_id].tx_num_pkts + i);
        tx_bufs[i]->data_len = tx_bufs[i]->pkt_len = tx_test_pkt_len;
      }
    }

    sent = rte_eth_tx_burst(port, queue_id, tx_bufs, BURST_SIZE);

    stats[queue_id].tx_num_pkts += sent;
    stats[queue_id].tx_drops += (BURST_SIZE - sent);

    if (sent < BURST_SIZE)
      rte_mempool_put_bulk(mbuf_pool[queue_id], (void **) &tx_bufs[sent], BURST_SIZE - sent);

#if !(defined(__arm__) || defined(__mips__))
    if (pps > 0) {
      while ((getticks() - tick_start) < (stats[queue_id].tx_num_pkts * tick_delta))
        if (unlikely(!do_loop)) break;
    }
#endif

  }
}

/* ************************************ */

static int packet_consumer(__attribute__((unused)) void *arg) {
  unsigned lcore_id = rte_lcore_id();
  u_int16_t queue_id = lcore_id;
  pfring_ft_table *ft;
  pfring_ft_pcap_pkthdr h;
  pfring_ft_ext_pkthdr ext_hdr = { 0 };
  struct rte_mbuf *bufs[BURST_SIZE];
  struct rte_mbuf *tx_bufs[BURST_SIZE];
  struct rte_mbuf *mseg;
  u_int16_t num, tx_num, sent;
  u_int32_t num_ports, i;
  u_int8_t ports[2];

  if (queue_id >= num_queues)
    return 0;

  if (test_tx) {
    tx_test(queue_id);
    return 0;
  }

  ports[0] = port;
  num_ports = 1;

  if (twin_port != 0xFF) {
    ports[1] = twin_port;
    num_ports = 2;
  } else {
    fwd = 0;
  }

  ft = fts[queue_id];

  printf("Capturing from port %u queue %u...\n", port, queue_id);

  while (do_loop) {
    u_int32_t idx;
    
    for (idx = 0; idx < num_ports; idx++) {
      u_int8_t in_port      = ports[idx];
      u_int8_t out_port_id = ports[idx^1];
      
      num = rte_eth_rx_burst(in_port, queue_id, bufs, BURST_SIZE);

      if (unlikely(num == 0)) {
        if (likely(compute_flows))
	  pfring_ft_housekeeping(ft, time(NULL));
	continue;
      }

      for (i = 0; i < PREFETCH_OFFSET && i < num; i++)
	rte_prefetch0(rte_pktmbuf_mtod(bufs[i], void *));

      tx_num = 0;
      for (i = 0; i < num; i++) {
	char *data = rte_pktmbuf_mtod(bufs[i], char *);
	int len = rte_pktmbuf_pkt_len(bufs[i]);
	pfring_ft_action action = PFRING_FT_ACTION_DEFAULT;
	
	if (likely(compute_flows)) {
          h.len = h.caplen = len;
          gettimeofday(&h.ts, NULL);

	  action = pfring_ft_process(ft, (const u_char *) data, &h, &ext_hdr);
        }
 
        stats[queue_id].num_pkts++;
        stats[queue_id].num_bytes += len + 24;

	if (unlikely(verbose)) {
	  int j;

	  printf("[Q#%u][Packet] len: %u segs: %u", queue_id, len, bufs[i]->nb_segs);

 	  if (action == PFRING_FT_ACTION_DISCARD)
	    printf(" [discard]");
         
          mseg = bufs[i];
          while (mseg && mseg->data_len) {
            printf("\n[data] len: %u hex: ", mseg->data_len);
	    data = rte_pktmbuf_mtod(mseg, char *);
  	    for (j = 0; j < mseg->data_len; j++)
	      printf("%02X ", data[j] & 0xFF);
            mseg = mseg->next;
          }

	  printf("\n");
	}

	if (fwd && action != PFRING_FT_ACTION_DISCARD) {
          tx_bufs[tx_num++] = bufs[i];
        } else {
          rte_pktmbuf_free(bufs[i]);
        }
      }

      if (tx_num > 0) {        
        sent = rte_eth_tx_burst(out_port_id, queue_id, tx_bufs, tx_num);
        stats[queue_id].tx_num_pkts += sent;
        stats[queue_id].tx_drops += (tx_num - sent);
        for (i = sent; i < tx_num; i++)
          rte_pktmbuf_free(tx_bufs[i]);
      }

    } /* for */
  } /* while */

  return 0;
}

/* ************************************ */

static void print_help(void) {
  printf("ftflow_dpdk - (C) 2018 ntop.org\n");
  printf("Usage: ftflow_dpdk [EAL options] -- [options]\n");
  printf("-p <id>[,<id>]  Port id (up to 2 ports are supported)\n");
  printf("-7              Enable L7 protocol detection (nDPI)\n");
  printf("-n <num cores>  Enable multiple cores/queues (default: 1)\n");
  printf("-0              Do not compute flows (packet capture only)\n");
  printf("-F              Enable forwarding when 2 ports are specified in -p\n");
  printf("-t              Test TX\n");
  printf("-T <size>       TX test packet size\n");
  printf("-P <pps>        TX test packet rate (pps)\n");
  printf("-v              Verbose (print also raw packets)\n");
  printf("-h              Print this help\n");
}

/* ************************************ */

static int parse_args(int argc, char **argv) {
  int opt, ret;
  char **argvopt;
  int option_index;
  char *prgname = argv[0];
  static struct option lgopts[] = {
    { NULL, 0, 0, 0 }
  };

  argvopt = argv;

  while ((opt = getopt_long(argc, argvopt, "Fhn:p:tvP:T:07", lgopts, &option_index)) != EOF) {
    switch (opt) {
    case 'F':
      fwd = 1;
      break;
    case 'n':
      if (optarg) {
        num_queues = atoi(optarg);
      }
      break;
    case 'h':
      print_help();
      exit(0);
      break;
    case 'p':
      if(optarg) {
	char *p = strtok(optarg, ",");

	if(p) {
	  port = atoi(p);
	  p = strtok(NULL, ",");
	  if(p)
	    twin_port = atoi(p);<commit_msg>Option to set the interface mac address (-M <addr>)<commit_after>/*
 * (C) 2018 - ntop
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in
 * all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 * THE SOFTWARE.
 */

#include <sched.h>
#include <stdlib.h>
#include <unistd.h>
#include <stdio.h>
#include <assert.h>
#include <inttypes.h>
#include <getopt.h>
#include <arpa/inet.h>
#include <sys/time.h>
#include <signal.h>

#include <rte_eal.h>
#include <rte_ether.h>
#include <rte_ethdev.h>
#include <rte_cycles.h>
#include <rte_lcore.h>
#include <rte_mbuf.h>

#define ALARM_SLEEP        1

/* NOTE: ether_hdr is defined in rte_ether.h */
#define ether_header ether_hdr
#include "ftutils.c"

#include "pfring_ft.h"

#define RX_RING_SIZE     8192
#define TX_RING_SIZE     1024
#define MBUF_CACHE_SIZE   256
#define BURST_SIZE         32
#define PREFETCH_OFFSET     3
#define TX_TEST_PKT_LEN    60

//#define SCATTERED_RX_TEST
#ifdef SCATTERED_RX_TEST
#define MBUF_BUF_SIZE     512
#else
#define MBUF_BUF_SIZE    RTE_MBUF_DEFAULT_BUF_SIZE
#endif

#define print_mac_addr(addr) printf("%02"PRIx8":%02"PRIx8":%02"PRIx8":%02"PRIx8":%02"PRIx8":%02"PRIx8, \
  addr.addr_bytes[0], addr.addr_bytes[1], addr.addr_bytes[2], addr.addr_bytes[3], addr.addr_bytes[4], addr.addr_bytes[5])

static struct rte_mempool *mbuf_pool[RTE_MAX_LCORE] = { NULL };
static pfring_ft_table *fts[RTE_MAX_LCORE] = { NULL };
static u_int32_t ft_flags = 0;
static u_int8_t port = 0, twin_port = 0xFF;
static u_int8_t num_queues = 1;
static u_int8_t compute_flows = 1;
static u_int8_t do_loop = 1;
static u_int8_t verbose = 0;
static u_int8_t fwd = 0;
static u_int8_t test_tx = 0;
static u_int8_t set_if_mac = 0;
static u_int16_t tx_test_pkt_len = TX_TEST_PKT_LEN;
static u_int32_t num_mbufs_per_lcore = 0;
static u_int32_t pps = 0;
static struct ether_addr if_mac = { 0 };

static struct lcore_stats {
  u_int64_t num_pkts;
  u_int64_t num_bytes;
  u_int64_t last_pkts;
  u_int64_t last_bytes;
  u_int64_t tx_num_pkts;
  u_int64_t tx_drops;
  u_int64_t padding[2];
} stats[RTE_MAX_LCORE];

static const struct rte_eth_conf port_conf_default = {
  .rxmode = { .max_rx_pkt_len = ETHER_MAX_LEN }
};

/* ************************************ */

static int port_init(void) {
  struct rte_eth_conf port_conf = port_conf_default;
  int retval, i;
  u_int16_t q;
  char name[64];
  int num_ports;

  if (twin_port == 0xFF || twin_port == port) num_ports = 1;
  else num_ports = 2;

  num_mbufs_per_lcore = 2 * (
    RX_RING_SIZE + 
    TX_RING_SIZE + 
    BURST_SIZE * 2) 
#ifdef SCATTERED_RX_TEST
    * 4
#endif
  ;

  for (q = 0; q < num_queues; q++) {
    snprintf(name, sizeof(name), "MBUF_POOL_%u", q);

    mbuf_pool[q] = rte_pktmbuf_pool_create(name, num_mbufs_per_lcore, MBUF_CACHE_SIZE, 0, 
      MBUF_BUF_SIZE, rte_socket_id());

    if (mbuf_pool[q] == NULL)
      rte_exit(EXIT_FAILURE, "Cannot create mbuf pool: %s\n", rte_strerror(rte_errno));
  }

  for (i = 0; i < num_ports; i++) {
    u_int8_t port_id = (i == 0) ? port : twin_port;
    unsigned int numa_socket_id;
    
    printf("Configuring port %u...\n", port_id);

    retval = rte_eth_dev_configure(port_id, num_queues /* RX */, num_queues /* TX */, &port_conf);
    
    if (retval != 0)
      return retval;    

    numa_socket_id = rte_eth_dev_socket_id(port_id);
    
    for (q = 0; q < num_queues; q++) {

      printf("Configuring queue %u...\n", q);

      retval = rte_eth_rx_queue_setup(port_id, q, RX_RING_SIZE, numa_socket_id, NULL, mbuf_pool[q]);

      if (retval < 0)
	return retval;

      retval = rte_eth_tx_queue_setup(port_id, q, TX_RING_SIZE, numa_socket_id, NULL);

      if (retval < 0)
	return retval;
    }

    retval = rte_eth_dev_start(port_id);

    if (retval < 0)
      return retval;

    rte_eth_promiscuous_enable(port_id);
  }
 
  if (set_if_mac) {
    retval = rte_eth_dev_default_mac_addr_set(port, &if_mac);
    if (retval != 0)
      printf("Unable to set the interface MAC address (%d)\n", retval);
  }
 
  return 0;
}

/* ************************************ */

static void port_close(void) {
  int i;

  for (i = 0; i < 2; i++) {
    u_int8_t port_id = (i == 0) ? port : twin_port;
    
    if (port_id == 0xFF || (i == 1 && twin_port == port)) break;

    printf("Releasing port %u...\n", port_id);

    rte_eth_dev_stop(port_id);
    rte_eth_dev_close(port_id);
  }
}

/* ************************************ */

void processFlow(pfring_ft_flow *flow, void *user) {
  pfring_ft_table *ft = (pfring_ft_table *) user;
  pfring_ft_flow_key *k;
  pfring_ft_flow_value *v;
  char buf1[32], buf2[32], buf3[32];
  const char *ip1, *ip2;

  k = pfring_ft_flow_get_key(flow);
  v = pfring_ft_flow_get_value(flow);

  if (k->ip_version == 4){
    ip1 = _intoa(k->saddr.v4, buf1, sizeof(buf1));
    ip2 = _intoa(k->daddr.v4, buf2, sizeof(buf2));
  } else {
    ip1 = inet_ntop(AF_INET6, &k->saddr.v6, buf1, sizeof(buf1));
    ip2 = inet_ntop(AF_INET6, &k->daddr.v6, buf2, sizeof(buf2));
  }

  printf("[Flow] "
         "srcIp: %s, dstIp: %s, srcPort: %u, dstPort: %u, protocol: %u, tcpFlags: 0x%02X, "
         "l7: %s, "
         "c2s: { Packets: %ju, Bytes: %ju, First: %u.%u, Last: %u.%u }, "
         "s2c: { Packets: %ju, Bytes: %ju, First: %u.%u, Last: %u.%u }\n",
         ip1, ip2, k->sport, k->dport, k->protocol, v->direction[s2d_direction].tcp_flags | v->direction[d2s_direction].tcp_flags,
         pfring_ft_l7_protocol_name(ft, &v->l7_protocol, buf3, sizeof(buf3)),
         v->direction[s2d_direction].pkts, v->direction[s2d_direction].bytes, 
         (u_int) v->direction[s2d_direction].first.tv_sec, (u_int) v->direction[s2d_direction].first.tv_usec, 
         (u_int) v->direction[s2d_direction].last.tv_sec,  (u_int) v->direction[s2d_direction].last.tv_usec,
         v->direction[d2s_direction].pkts, v->direction[d2s_direction].bytes, 
         (u_int) v->direction[d2s_direction].first.tv_sec, (u_int) v->direction[d2s_direction].first.tv_usec, 
         (u_int) v->direction[d2s_direction].last.tv_sec,  (u_int) v->direction[d2s_direction].last.tv_usec);

  pfring_ft_flow_free(flow);
}

/* ************************************ */

static void tx_test(u_int16_t queue_id) {
  struct rte_mbuf *tx_bufs[BURST_SIZE];
  u_int32_t i;
  u_int16_t sent;
  int rc;
#if !(defined(__arm__) || defined(__mips__))
  ticks tick_start = 0, tick_delta = 0;

  if (pps != 0) {
    double td;
    ticks hz = 0;

    /* computing usleep delay */
    tick_start = getticks();
    usleep(1);
    tick_delta = getticks() - tick_start;

    /* computing CPU freq */
    tick_start = getticks();
    usleep(1001);
    hz = (getticks() - tick_start - tick_delta) * 1000 /*kHz -> Hz*/;
    printf("Estimated CPU freq: %lu Hz\n", (long unsigned int)hz);

    td = (double) (hz / pps);
    tick_delta = (ticks) td;
    printf("Rate set to %u pps\n", pps);
  }
#endif

  printf("Generating traffic on port %u queue %u...\n", port, queue_id);

  num_ips = TX_RING_SIZE;

  while (do_loop) {

    rc = rte_mempool_get_bulk(mbuf_pool[queue_id], (void **) tx_bufs, BURST_SIZE);

    if (rc) {
      fprintf(stderr, "rte_mempool_get_bulk error (%d)\n", rc);
      return;
    }
  
    for (i = 0; i < BURST_SIZE; i++) {
      if (tx_bufs[i]->data_len != tx_test_pkt_len) {
        forge_udp_packet_fast((u_char *) rte_pktmbuf_mtod(tx_bufs[i], char *), 
          tx_test_pkt_len, stats[queue_id].tx_num_pkts + i);
        tx_bufs[i]->data_len = tx_bufs[i]->pkt_len = tx_test_pkt_len;
      }
    }

    sent = rte_eth_tx_burst(port, queue_id, tx_bufs, BURST_SIZE);

    stats[queue_id].tx_num_pkts += sent;
    stats[queue_id].tx_drops += (BURST_SIZE - sent);

    if (sent < BURST_SIZE)
      rte_mempool_put_bulk(mbuf_pool[queue_id], (void **) &tx_bufs[sent], BURST_SIZE - sent);

#if !(defined(__arm__) || defined(__mips__))
    if (pps > 0) {
      while ((getticks() - tick_start) < (stats[queue_id].tx_num_pkts * tick_delta))
        if (unlikely(!do_loop)) break;
    }
#endif

  }
}

/* ************************************ */

static int packet_consumer(__attribute__((unused)) void *arg) {
  unsigned lcore_id = rte_lcore_id();
  u_int16_t queue_id = lcore_id;
  pfring_ft_table *ft;
  pfring_ft_pcap_pkthdr h;
  pfring_ft_ext_pkthdr ext_hdr = { 0 };
  struct rte_mbuf *bufs[BURST_SIZE];
  struct rte_mbuf *tx_bufs[BURST_SIZE];
  struct rte_mbuf *mseg;
  u_int16_t num, tx_num, sent;
  u_int32_t num_ports, i;
  u_int8_t ports[2];

  if (queue_id >= num_queues)
    return 0;

  if (test_tx) {
    tx_test(queue_id);
    return 0;
  }

  ports[0] = port;
  num_ports = 1;

  if (twin_port != 0xFF) {
    ports[1] = twin_port;
    num_ports = 2;
  } else {
    fwd = 0;
  }

  ft = fts[queue_id];

  printf("Capturing from port %u queue %u...\n", port, queue_id);

  while (do_loop) {
    u_int32_t idx;
    
    for (idx = 0; idx < num_ports; idx++) {
      u_int8_t in_port      = ports[idx];
      u_int8_t out_port_id = ports[idx^1];
      
      num = rte_eth_rx_burst(in_port, queue_id, bufs, BURST_SIZE);

      if (unlikely(num == 0)) {
        if (likely(compute_flows))
	  pfring_ft_housekeeping(ft, time(NULL));
	continue;
      }

      for (i = 0; i < PREFETCH_OFFSET && i < num; i++)
	rte_prefetch0(rte_pktmbuf_mtod(bufs[i], void *));

      tx_num = 0;
      for (i = 0; i < num; i++) {
	char *data = rte_pktmbuf_mtod(bufs[i], char *);
	int len = rte_pktmbuf_pkt_len(bufs[i]);
	pfring_ft_action action = PFRING_FT_ACTION_DEFAULT;
	
	if (likely(compute_flows)) {
          h.len = h.caplen = len;
          gettimeofday(&h.ts, NULL);

	  action = pfring_ft_process(ft, (const u_char *) data, &h, &ext_hdr);
        }
 
        stats[queue_id].num_pkts++;
        stats[queue_id].num_bytes += len + 24;

	if (unlikely(verbose)) {
	  int j;

	  printf("[Q#%u][Packet] len: %u segs: %u", queue_id, len, bufs[i]->nb_segs);

 	  if (action == PFRING_FT_ACTION_DISCARD)
	    printf(" [discard]");
         
          mseg = bufs[i];
          while (mseg && mseg->data_len) {
            printf("\n[data] len: %u hex: ", mseg->data_len);
	    data = rte_pktmbuf_mtod(mseg, char *);
  	    for (j = 0; j < mseg->data_len; j++)
	      printf("%02X ", data[j] & 0xFF);
            mseg = mseg->next;
          }

	  printf("\n");
	}

	if (fwd && action != PFRING_FT_ACTION_DISCARD) {
          tx_bufs[tx_num++] = bufs[i];
        } else {
          rte_pktmbuf_free(bufs[i]);
        }
      }

      if (tx_num > 0) {        
        sent = rte_eth_tx_burst(out_port_id, queue_id, tx_bufs, tx_num);
        stats[queue_id].tx_num_pkts += sent;
        stats[queue_id].tx_drops += (tx_num - sent);
        for (i = sent; i < tx_num; i++)
          rte_pktmbuf_free(tx_bufs[i]);
      }

    } /* for */
  } /* while */

  return 0;
}

/* ************************************ */

static void print_help(void) {
  printf("ftflow_dpdk - (C) 2018 ntop.org\n");
  printf("Usage: ftflow_dpdk [EAL options] -- [options]\n");
  printf("-p <id>[,<id>]  Port id (up to 2 ports are supported)\n");
  printf("-7              Enable L7 protocol detection (nDPI)\n");
  printf("-n <num cores>  Enable multiple cores/queues (default: 1)\n");
  printf("-0              Do not compute flows (packet capture only)\n");
  printf("-F              Enable forwarding when 2 ports are specified in -p\n");
  printf("-M <addr>       Set the port MAC address\n");
  printf("-t              Test TX\n");
  printf("-T <size>       TX test packet size\n");
  printf("-P <pps>        TX test packet rate (pps)\n");
  printf("-v              Verbose (print also raw packets)\n");
  printf("-h              Print this help\n");
}

/* ************************************ */

static int parse_args(int argc, char **argv) {
  int opt, ret;
  char **argvopt;
  int option_index;
  char *prgname = argv[0];
  u_int mac_a, mac_b, mac_c, mac_d, mac_e, mac_f;
  static struct option lgopts[] = {
    { NULL, 0, 0, 0 }
  };

  argvopt = argv;

  while ((opt = getopt_long(argc, argvopt, "FhM:n:p:tvP:T:07", lgopts, &option_index)) != EOF) {
    switch (opt) {
    case 'F':
      fwd = 1;
      break;
    case 'h':
      print_help();
      exit(0);
      break;
    case 'M':
      if(sscanf(optarg, "%02X:%02X:%02X:%02X:%02X:%02X", &mac_a, &mac_b, &mac_c, &mac_d, &mac_e, &mac_f) != 6) {
	printf("Invalid MAC address format (XX:XX:XX:XX:XX:XX)\n");
	exit(0);
      }
      if_mac.addr_bytes[0] = mac_a, if_mac.addr_bytes[1] = mac_b, if_mac.addr_bytes[2] = mac_c,
      if_mac.addr_bytes[3] = mac_d, if_mac.addr_bytes[4] = mac_e, if_mac.addr_bytes[5] = mac_f;
      set_if_mac = 1;
      break;
    case 'n':
      if (optarg) {
        num_queues = atoi(optarg);
      }
      break;
    case 'p':
      if(optarg) {
	char *p = strtok(optarg, ",");

	if(p) {
	  port = atoi(p);
	  p = strtok(NULL, ",");
	  if(p)
	    twin_port = atoi(p);
{'fraction_non_alphanumeric': 0.1586282022387499, 'fraction_numerical': 0.019495154383592516, 'mean_word_length': 3.2346111022745347, 'pattern_counts': {'":': 10, '<': 79, '<?xml version=': 0, '>': 117, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '109369', 'n_tokens_mistral': 11052, 'n_tokens_neox': 10571, 'n_words': 2347}