metron-sensors/fastcapa/src/args.c - metron - Git at Google

 /**
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */

 #include "args.h"

 /*
  * Print usage information to the user.
  */
 static void print_usage(void)
 {
     printf("fastcapa [EAL options] -- [APP options]\n"
            "  -p PORT_MASK        bitmask of ports to bind                     [%s]\n"
            "  -b RX_BURST_SIZE    burst size of receive worker                 [%d]\n"
            "  -w TX_BURST_SIZE    burst size of transmit worker                [%d]\n"
            "  -d NB_RX_DESC       num of descriptors for receive ring          [%d]\n"
            "  -x TX_RING_SIZE     size of tx rings (must be a power of 2)      [%d]\n"
            "  -q NB_RX_QUEUE      num of receive queues for each device        [%d]\n"
            "  -t KAFKA_TOPIC      name of the kafka topic                      [%s]\n"
            "  -c KAFKA_CONF       file containing configs for kafka client         \n"
            "  -s KAFKA_STATS      append kafka client stats to a file              \n"
            "  -h                  print this help message                          \n",
         STR(DEFAULT_PORT_MASK),
         DEFAULT_RX_BURST_SIZE,
         DEFAULT_TX_BURST_SIZE,
         DEFAULT_NB_RX_DESC,
         DEFAULT_TX_RING_SIZE,
         DEFAULT_NB_RX_QUEUE,
         STR(DEFAULT_KAFKA_TOPIC));
 }

 /*
  * Parse the 'portmask' command line argument.
  */
 static int parse_portmask(const char* portmask)
 {
     char* end = NULL;
     unsigned long pm;

     // parse hexadecimal string
     pm = strtoul(portmask, &end, 16);

     if ((portmask[0] == '\0') || (end == NULL) || (*end != '\0')) {
         return -1;
     }
     else if (pm == 0) {
         return -1;
     }
     else {
         return pm;
     }
 }

 /*
  * Check if a file exists
  */
 static bool file_exists(const char* filepath)
 {
     struct stat buf;
     return (stat(filepath, &buf) == 0);
 }

 /*
  * Parse the command line arguments passed to the application; the arguments
  * which no not go directly to DPDK's EAL.
  */
 static int parse_app_args(int argc, char** argv, app_params* p)
 {
     int opt;
     char** argvopt;
     int option_index;
     unsigned int nb_workers;
     static struct option lgopts[] = {
         { NULL, 0, 0, 0 }
     };

     // set default args
     p->enabled_port_mask = parse_portmask(STR(DEFAULT_PORT_MASK));
     p->kafka_topic = STR(DEFAULT_KAFKA_TOPIC);
     p->rx_burst_size = DEFAULT_RX_BURST_SIZE;
     p->tx_burst_size = DEFAULT_TX_BURST_SIZE;
     p->nb_rx_desc = DEFAULT_NB_RX_DESC;
     p->nb_rx_queue = DEFAULT_NB_RX_QUEUE;
     p->tx_ring_size = DEFAULT_TX_RING_SIZE;

     // parse arguments to this application
     argvopt = argv;
     while ((opt = getopt_long(argc, argvopt, "b:c:d:hp:q:s:t:w:x:", lgopts, &option_index)) != EOF) {
         switch (opt) {

             // help
             case 'h':
                 print_usage();
                 return -1;

             // rx burst size
             case 'b':
                 p->rx_burst_size = atoi(optarg);
                 if(p->rx_burst_size < 1 || p->rx_burst_size > MAX_RX_BURST_SIZE) {
                     fprintf(stderr, "Invalid burst size; burst=%u must be in [1, %u]. \n", p->rx_burst_size, MAX_RX_BURST_SIZE);
                     print_usage();
                     return -1;
                 }
                 break;

             // tx burst size
             case 'w':
                 p->tx_burst_size = atoi(optarg);
                 if(p->tx_burst_size < 1) {
                     fprintf(stderr, "Invalid burst size; burst=%u must be > 0. \n", p->tx_burst_size);
                     print_usage();
                     return -1;
                 }
                 break;

             // number of receive descriptors
             case 'd':
                 p->nb_rx_desc = atoi(optarg);
                 if (p->nb_rx_desc < 1) {
                     fprintf(stderr, "Invalid num of receive descriptors: '%s' \n", optarg);
                     print_usage();
                     return -1;
                 }
                 break;

             // size of each transmit ring
             case 'x':
                 p->tx_ring_size = atoi(optarg);

                 break;

             // number of receive queues for each device
             case 'q':
                 p->nb_rx_queue = atoi(optarg);
                 if (p->nb_rx_queue < 1) {
                     fprintf(stderr, "Invalid num of receive queues: '%s' \n", optarg);
                     print_usage();
                     return -1;
                 }
                 break;

           // port mask
           case 'p':
               p->enabled_port_mask = parse_portmask(optarg);
               if (p->enabled_port_mask == 0) {
                   fprintf(stderr, "Invalid portmask: '%s'\n", optarg);
                   print_usage();
                   return -1;
               }
               break;

           // kafka topic
           case 't':
               p->kafka_topic = strdup(optarg);
               if (!valid(p->kafka_topic)) {
                   printf("Invalid kafka topic: '%s'\n", optarg);
                   print_usage();
                   return -1;
               }
               break;

           // kafka config path
           case 'c':
               p->kafka_config_path = strdup(optarg);
               if (!valid(p->kafka_config_path) || !file_exists(p->kafka_config_path)) {
                   fprintf(stderr, "Invalid kafka config: '%s'\n", optarg);
                   print_usage();
                   return -1;
               }
               break;

           // kafka stats path
           case 's':
               p->kafka_stats_path = strdup(optarg);
               break;

           default:
               print_usage();
               return -1;
           }
     }

     // check number of ethernet devices
     if (rte_eth_dev_count() == 0) {
          rte_exit(EXIT_FAILURE, "No ethernet ports detected.\n");
      }

     // check number of workers
     nb_workers = rte_lcore_count() - 1;

     // need at least 1 worker for each receive queue
     if(nb_workers < p->nb_rx_queue) {
         rte_exit(EXIT_FAILURE, "Minimum 1 worker per receive queue; workers=%u rx_queues=%u. \n",
             nb_workers, p->nb_rx_queue);
     }

     p->nb_rx_workers = p->nb_rx_queue;
     p->nb_tx_workers = nb_workers - p->nb_rx_workers;

     printf("[ -p PORT_MASK ] defined as %d \n", p->enabled_port_mask);
     printf("[ -b RX_BURST_SIZE ] defined as %d \n", p->rx_burst_size);
     printf("[ -w TX_BURST_SIZE ] defined as %d \n", p->tx_burst_size);
     printf("[ -d NB_RX_DESC ] defined as %d \n", p->nb_rx_desc);
     printf("[ -x TX_RING_SIZE ] defined as %d \n", p->tx_ring_size);
     printf("[ -q NB_RX_QUEUE ] defined as %d \n", p->nb_rx_queue);
     printf("[ -t KAFKA_TOPIC ] defined as %s \n", p->kafka_topic);
     printf("[ -c KAFKA_CONFIG ] defined as %s \n", p->kafka_config_path);
     printf("[ -s KAFKA_STATS ] defined as %s \n", p->kafka_stats_path);
     printf("[ NUM_RX_WORKERS ] defined as %d \n", p->nb_rx_workers);
     printf("[ NUM_TX_WORKERS ] defined as %d \n", p->nb_tx_workers);

     // reset getopt lib
     optind = 0;
     return 0;
 }

 /*
  * Parse the command line arguments passed to the application.
  */
 int parse_args(int argc, char** argv, app_params* p)
 {
   // initialize the environment
   int ret = rte_eal_init(argc, argv);
   if (ret < 0) {
       rte_exit(EXIT_FAILURE, "Failed to initialize EAL: %i\n", ret);
   }

   // advance past the environmental settings
   argc -= ret;
   argv += ret;

   // parse arguments to the application
   ret = parse_app_args(argc, argv, p);
   if (ret < 0) {
       rte_exit(EXIT_FAILURE, "\n");
   }

   p->nb_ports = rte_eth_dev_count();
   p->nb_rx_workers = p->nb_rx_queue;
   p->nb_tx_workers = (rte_lcore_count() - 1) - p->nb_rx_workers;

   // validate the number of workers
   if(p->nb_tx_workers < p->nb_rx_workers) {
       rte_exit(EXIT_FAILURE, "Additional lcore(s) required; found=%u, required=%u \n",
           rte_lcore_count(), (p->nb_rx_queue*2) + 1);
   }

   return 0;
 }
	/**
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/

	#include "args.h"

	/*
	* Print usage information to the user.
	*/
	static void print_usage(void)
	{
	printf("fastcapa [EAL options] -- [APP options]\n"
	" -p PORT_MASK bitmask of ports to bind [%s]\n"
	" -b RX_BURST_SIZE burst size of receive worker [%d]\n"
	" -w TX_BURST_SIZE burst size of transmit worker [%d]\n"
	" -d NB_RX_DESC num of descriptors for receive ring [%d]\n"
	" -x TX_RING_SIZE size of tx rings (must be a power of 2) [%d]\n"
	" -q NB_RX_QUEUE num of receive queues for each device [%d]\n"
	" -t KAFKA_TOPIC name of the kafka topic [%s]\n"
	" -c KAFKA_CONF file containing configs for kafka client \n"
	" -s KAFKA_STATS append kafka client stats to a file \n"
	" -h print this help message \n",
	STR(DEFAULT_PORT_MASK),
	DEFAULT_RX_BURST_SIZE,
	DEFAULT_TX_BURST_SIZE,
	DEFAULT_NB_RX_DESC,
	DEFAULT_TX_RING_SIZE,
	DEFAULT_NB_RX_QUEUE,
	STR(DEFAULT_KAFKA_TOPIC));
	}

	/*
	* Parse the 'portmask' command line argument.
	*/
	static int parse_portmask(const char* portmask)
	{
	char* end = NULL;
	unsigned long pm;

	// parse hexadecimal string
	pm = strtoul(portmask, &end, 16);

	if ((portmask[0] == '\0') \|\| (end == NULL) \|\| (*end != '\0')) {
	return -1;
	}
	else if (pm == 0) {
	return -1;
	}
	else {
	return pm;
	}
	}

	/*
	* Check if a file exists
	*/
	static bool file_exists(const char* filepath)
	{
	struct stat buf;
	return (stat(filepath, &buf) == 0);
	}

	/*
	* Parse the command line arguments passed to the application; the arguments
	* which no not go directly to DPDK's EAL.
	*/
	static int parse_app_args(int argc, char** argv, app_params* p)
	{
	int opt;
	char** argvopt;
	int option_index;
	unsigned int nb_workers;
	static struct option lgopts[] = {
	{ NULL, 0, 0, 0 }
	};

	// set default args
	p->enabled_port_mask = parse_portmask(STR(DEFAULT_PORT_MASK));
	p->kafka_topic = STR(DEFAULT_KAFKA_TOPIC);
	p->rx_burst_size = DEFAULT_RX_BURST_SIZE;
	p->tx_burst_size = DEFAULT_TX_BURST_SIZE;
	p->nb_rx_desc = DEFAULT_NB_RX_DESC;
	p->nb_rx_queue = DEFAULT_NB_RX_QUEUE;
	p->tx_ring_size = DEFAULT_TX_RING_SIZE;

	// parse arguments to this application
	argvopt = argv;
	while ((opt = getopt_long(argc, argvopt, "b:c:d:hp:q:s:t:w:x:", lgopts, &option_index)) != EOF) {
	switch (opt) {

	// help
	case 'h':
	print_usage();
	return -1;

	// rx burst size
	case 'b':
	p->rx_burst_size = atoi(optarg);
	if(p->rx_burst_size < 1 \|\| p->rx_burst_size > MAX_RX_BURST_SIZE) {
	fprintf(stderr, "Invalid burst size; burst=%u must be in [1, %u]. \n", p->rx_burst_size, MAX_RX_BURST_SIZE);
	print_usage();
	return -1;
	}
	break;

	// tx burst size
	case 'w':
	p->tx_burst_size = atoi(optarg);
	if(p->tx_burst_size < 1) {
	fprintf(stderr, "Invalid burst size; burst=%u must be > 0. \n", p->tx_burst_size);
	print_usage();
	return -1;
	}
	break;

	// number of receive descriptors
	case 'd':
	p->nb_rx_desc = atoi(optarg);
	if (p->nb_rx_desc < 1) {
	fprintf(stderr, "Invalid num of receive descriptors: '%s' \n", optarg);
	print_usage();
	return -1;
	}
	break;

	// size of each transmit ring
	case 'x':
	p->tx_ring_size = atoi(optarg);

	break;

	// number of receive queues for each device
	case 'q':
	p->nb_rx_queue = atoi(optarg);
	if (p->nb_rx_queue < 1) {
	fprintf(stderr, "Invalid num of receive queues: '%s' \n", optarg);
	print_usage();
	return -1;
	}
	break;

	// port mask
	case 'p':
	p->enabled_port_mask = parse_portmask(optarg);
	if (p->enabled_port_mask == 0) {
	fprintf(stderr, "Invalid portmask: '%s'\n", optarg);
	print_usage();
	return -1;
	}
	break;

	// kafka topic
	case 't':
	p->kafka_topic = strdup(optarg);
	if (!valid(p->kafka_topic)) {
	printf("Invalid kafka topic: '%s'\n", optarg);
	print_usage();
	return -1;
	}
	break;

	// kafka config path
	case 'c':
	p->kafka_config_path = strdup(optarg);
	if (!valid(p->kafka_config_path) \|\| !file_exists(p->kafka_config_path)) {
	fprintf(stderr, "Invalid kafka config: '%s'\n", optarg);
	print_usage();
	return -1;
	}
	break;

	// kafka stats path
	case 's':
	p->kafka_stats_path = strdup(optarg);
	break;

	default:
	print_usage();
	return -1;
	}
	}

	// check number of ethernet devices
	if (rte_eth_dev_count() == 0) {
	rte_exit(EXIT_FAILURE, "No ethernet ports detected.\n");
	}

	// check number of workers
	nb_workers = rte_lcore_count() - 1;

	// need at least 1 worker for each receive queue
	if(nb_workers < p->nb_rx_queue) {
	rte_exit(EXIT_FAILURE, "Minimum 1 worker per receive queue; workers=%u rx_queues=%u. \n",
	nb_workers, p->nb_rx_queue);
	}

	p->nb_rx_workers = p->nb_rx_queue;
	p->nb_tx_workers = nb_workers - p->nb_rx_workers;

	printf("[ -p PORT_MASK ] defined as %d \n", p->enabled_port_mask);
	printf("[ -b RX_BURST_SIZE ] defined as %d \n", p->rx_burst_size);
	printf("[ -w TX_BURST_SIZE ] defined as %d \n", p->tx_burst_size);
	printf("[ -d NB_RX_DESC ] defined as %d \n", p->nb_rx_desc);
	printf("[ -x TX_RING_SIZE ] defined as %d \n", p->tx_ring_size);
	printf("[ -q NB_RX_QUEUE ] defined as %d \n", p->nb_rx_queue);
	printf("[ -t KAFKA_TOPIC ] defined as %s \n", p->kafka_topic);
	printf("[ -c KAFKA_CONFIG ] defined as %s \n", p->kafka_config_path);
	printf("[ -s KAFKA_STATS ] defined as %s \n", p->kafka_stats_path);
	printf("[ NUM_RX_WORKERS ] defined as %d \n", p->nb_rx_workers);
	printf("[ NUM_TX_WORKERS ] defined as %d \n", p->nb_tx_workers);

	// reset getopt lib
	optind = 0;
	return 0;
	}

	/*
	* Parse the command line arguments passed to the application.
	*/
	int parse_args(int argc, char** argv, app_params* p)
	{
	// initialize the environment
	int ret = rte_eal_init(argc, argv);
	if (ret < 0) {
	rte_exit(EXIT_FAILURE, "Failed to initialize EAL: %i\n", ret);
	}

	// advance past the environmental settings
	argc -= ret;
	argv += ret;

	// parse arguments to the application
	ret = parse_app_args(argc, argv, p);
	if (ret < 0) {
	rte_exit(EXIT_FAILURE, "\n");
	}

	p->nb_ports = rte_eth_dev_count();
	p->nb_rx_workers = p->nb_rx_queue;
	p->nb_tx_workers = (rte_lcore_count() - 1) - p->nb_rx_workers;

	// validate the number of workers
	if(p->nb_tx_workers < p->nb_rx_workers) {
	rte_exit(EXIT_FAILURE, "Additional lcore(s) required; found=%u, required=%u \n",
	rte_lcore_count(), (p->nb_rx_queue*2) + 1);
	}

	return 0;
	}