src/dpdkstat.c

   1 /*-
   2  * collectd - src/dpdkstat.c
   3  * MIT License
   4  *
   5  * Copyright(c) 2016 Intel Corporation. All rights reserved.
   6  *
   7  * Permission is hereby granted, free of charge, to any person obtaining a copy of
   8  * this software and associated documentation files (the "Software"), to deal in
   9  * the Software without restriction, including without limitation the rights to
  10  * use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies
  11  * of the Software, and to permit persons to whom the Software is furnished to do
  12  * so, subject to the following conditions:
  13  *
  14  * The above copyright notice and this permission notice shall be included in all
  15  * copies or substantial portions of the Software.
  16  *
  17  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  18  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  19  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  20  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  21  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  22  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  23  * SOFTWARE.
  24  *
  25  * Authors:
  26  *   Maryam Tahhan <maryam.tahhan@intel.com>
  27  *   Harry van Haaren <harry.van.haaren@intel.com>
  28  */
  29
  30 #include "collectd.h"
  31 #include "common.h" /* auxiliary functions */
  32 #include "plugin.h" /* plugin_register_*, plugin_dispatch_values */
  33 #include "utils_time.h"
  34
  35 #include <stdio.h>
  36 #include <string.h>
  37 #include <stdint.h>
  38 #include <signal.h>
  39 #include <stdarg.h>
  40 #include <stdlib.h>
  41 #include <errno.h>
  42 #include <getopt.h>
  43 #include <inttypes.h>
  44 #include <fcntl.h>
  45 #include <semaphore.h>
  46 #include <sys/mman.h>
  47 #include <sys/queue.h>
  48 #include <sys/stat.h>
  49 #include <sys/types.h>
  50 #include <sys/time.h>
  51 #include <sys/wait.h>
  52 #include <poll.h>
  53 #include <unistd.h>
  54 #include <string.h>
  55
  56 #include <rte_config.h>
  57 #include <rte_eal.h>
  58 #include <rte_ethdev.h>
  59 #include <rte_common.h>
  60 #include <rte_debug.h>
  61 #include <rte_malloc.h>
  62 #include <rte_memory.h>
  63 #include <rte_memzone.h>
  64 #include <rte_launch.h>
  65 #include <rte_tailq.h>
  66 #include <rte_lcore.h>
  67 #include <rte_per_lcore.h>
  68 #include <rte_debug.h>
  69 #include <rte_log.h>
  70 #include <rte_atomic.h>
  71 #include <rte_branch_prediction.h>
  72 #include <rte_string_fns.h>
  73
  74
  75 #define DATA_MAX_NAME_LEN        64
  76 #define DPDKSTAT_MAX_BUFFER_SIZE (4096*4)
  77 #define DPDK_SHM_NAME "dpdk_collectd_stats_shm"
  78 #define REINIT_SHM 1
  79 #define RESET 1
  80 #define NO_RESET 0
  81
  82 enum DPDK_HELPER_ACTION {
  83   DPDK_HELPER_ACTION_COUNT_STATS,
  84   DPDK_HELPER_ACTION_SEND_STATS,
  85 };
  86
  87 enum DPDK_HELPER_STATUS {
  88   DPDK_HELPER_NOT_INITIALIZED = 0,
  89   DPDK_HELPER_WAITING_ON_PRIMARY,
  90   DPDK_HELPER_INITIALIZING_EAL,
  91   DPDK_HELPER_ALIVE_SENDING_STATS,
  92   DPDK_HELPER_GRACEFUL_QUIT,
  93 };
  94
  95 struct dpdk_config_s {
  96   /* General DPDK params */
  97   char coremask[DATA_MAX_NAME_LEN];
  98   char memory_channels[DATA_MAX_NAME_LEN];
  99   char socket_memory[DATA_MAX_NAME_LEN];
 100   char process_type[DATA_MAX_NAME_LEN];
 101   char file_prefix[DATA_MAX_NAME_LEN];
 102   cdtime_t interval;
 103   uint32_t eal_initialized;
 104   uint32_t enabled_port_mask;
 105   uint32_t eal_argc;
 106   /* Helper info */
 107   int   collectd_reinit_shm;
 108   pid_t helper_pid;
 109   sem_t sema_helper_get_stats;
 110   sem_t sema_stats_in_shm;
 111   int   helper_pipes[2];
 112   enum DPDK_HELPER_STATUS helper_status;
 113   enum DPDK_HELPER_ACTION helper_action;
 114   /* xstats info */
 115   uint32_t num_ports;
 116   uint32_t num_xstats;
 117   cdtime_t port_read_time[RTE_MAX_ETHPORTS];
 118   uint32_t num_stats_in_port[RTE_MAX_ETHPORTS];
 119   struct rte_eth_link link_status[RTE_MAX_ETHPORTS];
 120   struct rte_eth_xstats xstats;
 121   /* rte_eth_xstats from here on until the end of the SHM */
 122 };
 123 typedef struct dpdk_config_s dpdk_config_t;
 124
 125 static int g_configured = 0;
 126 static dpdk_config_t *g_configuration = 0;
 127
 128 static int dpdk_config_init_default(void);
 129 static int dpdk_config(oconfig_item_t *ci);
 130 static int dpdk_helper_init_eal(void);
 131 static int dpdk_helper_run(void);
 132 static int dpdk_helper_spawn(enum DPDK_HELPER_ACTION action);
 133 static int dpdk_init (void);
 134 static int dpdk_read(user_data_t *ud);
 135 static int dpdk_shm_cleanup(void);
 136 static int dpdk_shm_init(size_t size);
 137 void module_register(void);
 138
 139 /* Write the default configuration to the g_configuration instances */
 140 static int dpdk_config_init_default(void)
 141 {
 142     g_configuration->interval = plugin_get_interval();
 143     WARNING("dpdkstat: No time interval was configured, default value %lu ms is set\n",
 144              CDTIME_T_TO_MS(g_configuration->interval));
 145     g_configuration->enabled_port_mask = 0;
 146     g_configuration->eal_argc = 2;
 147     g_configuration->eal_initialized = 0;
 148     snprintf(g_configuration->coremask, DATA_MAX_NAME_LEN, "%s", "0xf");
 149     snprintf(g_configuration->memory_channels, DATA_MAX_NAME_LEN, "%s", "1");
 150     snprintf(g_configuration->process_type, DATA_MAX_NAME_LEN, "%s", "secondary");
 151     snprintf(g_configuration->file_prefix, DATA_MAX_NAME_LEN, "%s",
 152              "/var/run/.rte_config");
 153   return 0;
 154 }
 155
 156 static int dpdk_config(oconfig_item_t *ci)
 157 {
 158   int i = 0, ret = 0;
 159
 160   /* Initialize a POSIX SHared Memory (SHM) object. */
 161   dpdk_shm_init(sizeof(dpdk_config_t));
 162
 163   /* Set defaults for config, overwritten by loop if config item exists */
 164   ret = dpdk_config_init_default();
 165   if(ret != 0) {
 166     return -1;
 167   }
 168
 169   for (i = 0; i < ci->children_num; i++) {
 170     oconfig_item_t *child = ci->children + i;
 171
 172     if (strcasecmp("Interval", child->key) == 0) {
 173       g_configuration->interval =
 174             DOUBLE_TO_CDTIME_T (child->values[0].value.number);
 175       DEBUG("dpdkstat: Plugin Read Interval %lu milliseconds\n",
 176             CDTIME_T_TO_MS(g_configuration->interval));
 177     } else if (strcasecmp("Coremask", child->key) == 0) {
 178       snprintf(g_configuration->coremask, DATA_MAX_NAME_LEN, "%s",
 179                child->values[0].value.string);
 180       DEBUG("dpdkstat:COREMASK %s \n", g_configuration->coremask);
 181       g_configuration->eal_argc+=1;
 182     } else if (strcasecmp("MemoryChannels", child->key) == 0) {
 183       snprintf(g_configuration->memory_channels, DATA_MAX_NAME_LEN, "%s",
 184                child->values[0].value.string);
 185       DEBUG("dpdkstat:Memory Channels %s \n", g_configuration->memory_channels);
 186       g_configuration->eal_argc+=1;
 187     } else if (strcasecmp("SocketMemory", child->key) == 0) {
 188       snprintf(g_configuration->socket_memory, DATA_MAX_NAME_LEN, "%s",
 189                child->values[0].value.string);
 190       DEBUG("dpdkstat: socket mem %s \n", g_configuration->socket_memory);
 191       g_configuration->eal_argc+=1;
 192     } else if (strcasecmp("ProcessType", child->key) == 0) {
 193       snprintf(g_configuration->process_type, DATA_MAX_NAME_LEN, "%s",
 194                child->values[0].value.string);
 195       DEBUG("dpdkstat: proc type %s \n", g_configuration->process_type);
 196       g_configuration->eal_argc+=1;
 197     } else if (strcasecmp("FilePrefix", child->key) == 0) {
 198       snprintf(g_configuration->file_prefix, DATA_MAX_NAME_LEN, "/var/run/.%s_config",
 199                child->values[0].value.string);
 200       DEBUG("dpdkstat: file prefix %s \n", g_configuration->file_prefix);
 201       if (strcasecmp(g_configuration->file_prefix, "/var/run/.rte_config") != 0) {
 202         g_configuration->eal_argc+=1;
 203       }
 204     } else {
 205       WARNING ("dpdkstat: The config option \"%s\" is unknown.",
 206                child->key);
 207     }
 208   } /* End for (i = 0; i < ci->children_num; i++)*/
 209   g_configured = 1; /* Bypass configuration in dpdk_shm_init(). */
 210
 211   return 0;
 212 }
 213
 214 /* Initialize SHared Memory (SHM) for config and helper process */
 215 static int dpdk_shm_init(size_t size)
 216 {
 217   /*
 218    * Check if SHM is already configured: when config items are provided, the
 219    * config function initializes SHM. If there is no config, then init() will
 220    * just return.
 221    */
 222   if(g_configuration)
 223     return 0;
 224
 225   /* Create and open a new object, or open an existing object. */
 226   int fd = shm_open(DPDK_SHM_NAME, O_CREAT | O_TRUNC | O_RDWR, 0666);
 227   if (fd < 0) {
 228     WARNING("dpdkstat:Failed to open %s as SHM:%s\n", DPDK_SHM_NAME,
 229             strerror(errno));
 230     goto fail;
 231   }
 232   /* Set the size of the shared memory object. */
 233   int ret = ftruncate(fd, size);
 234   if (ret != 0) {
 235     WARNING("dpdkstat:Failed to resize SHM:%s\n", strerror(errno));
 236     goto fail_close;
 237   }
 238   /* Map the shared memory object into this process' virtual address space. */
 239   g_configuration = (dpdk_config_t *)
 240                     mmap(0, size, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
 241   if (g_configuration == MAP_FAILED) {
 242     WARNING("dpdkstat:Failed to mmap SHM:%s\n", strerror(errno));
 243     goto fail_close;
 244   }
 245   /*
 246    * Close the file descriptor, the shared memory object still exists
 247    * and can only be removed by calling shm_unlink().
 248    */
 249   close(fd);
 250
 251   /* Initialize g_configuration. */
 252   memset(g_configuration, 0, size);
 253
 254   /* Initialize the semaphores for SHM use */
 255   sem_init(&g_configuration->sema_helper_get_stats, 1, 0);
 256   sem_init(&g_configuration->sema_stats_in_shm, 1, 0);
 257   return 0;
 258
 259 fail_close:
 260   close(fd);
 261 fail:
 262   /* Reset to zero, as it was set to MAP_FAILED aka: (void *)-1. Avoid
 263    * an issue if collectd attempts to run this plugin failure.
 264    */
 265   g_configuration = 0;
 266   return -1;
 267 }
 268
 269 static int dpdk_re_init_shm()
 270 {
 271   dpdk_config_t temp_config;
 272   memcpy(&temp_config,g_configuration, sizeof(dpdk_config_t));
 273   DEBUG("dpdkstat: %s: ports %d, xstats %d\n", __func__, temp_config.num_ports,
 274         temp_config.num_xstats);
 275
 276   int shm_xstats_size = sizeof(dpdk_config_t) + (sizeof(struct rte_eth_xstats) *
 277                         g_configuration->num_xstats);
 278   DEBUG("=== SHM new size for %d xstats\n", g_configuration->num_xstats);
 279
 280   int err = dpdk_shm_cleanup();
 281   if (err)
 282     ERROR("dpdkstat: Error in shm_cleanup in %s\n", __func__);
 283
 284   err = dpdk_shm_init(shm_xstats_size);
 285   if (err)
 286     ERROR("dpdkstat: Error in shm_init in %s\n", __func__);
 287
 288   /* If the XML config() function has been run, dont re-initialize defaults */
 289   if(!g_configured)
 290     dpdk_config_init_default();
 291
 292   memcpy(g_configuration,&temp_config, sizeof(dpdk_config_t));
 293   g_configuration->collectd_reinit_shm = 0;
 294
 295   return 0;
 296 }
 297
 298 static int dpdk_init (void)
 299 {
 300   int ret = 0;
 301   int err = dpdk_shm_init(sizeof(dpdk_config_t));
 302   if (err)
 303     ERROR("dpdkstat: %s : error %d in shm_init()", __func__, err);
 304
 305   /* If the XML config() function has been run, dont re-initialize defaults */
 306   if(!g_configured) {
 307     ret = dpdk_config_init_default();
 308     if (ret != 0) {
 309       return -1;
 310     }
 311   }
 312
 313   plugin_register_complex_read (NULL, "dpdkstat", dpdk_read,
 314                                 g_configuration->interval, NULL);
 315   return 0;
 316 }
 317
 318 static int dpdk_helper_exit(int reset)
 319 {
 320   g_configuration->helper_status = DPDK_HELPER_GRACEFUL_QUIT;
 321   if(reset) {
 322     g_configuration->eal_initialized = 0;
 323     g_configuration->num_ports = 0;
 324     memset(&g_configuration->xstats, 0, g_configuration->num_xstats* sizeof(struct rte_eth_xstats));
 325     g_configuration->num_xstats = 0;
 326     int i =0;
 327     for(;i < RTE_MAX_ETHPORTS; i++)
 328       g_configuration->num_stats_in_port[i] = 0;
 329   }
 330   close(g_configuration->helper_pipes[1]);
 331   kill(g_configuration->helper_pid, SIGKILL);
 332   return 0;
 333 }
 334
 335 static int dpdk_helper_spawn(enum DPDK_HELPER_ACTION action)
 336 {
 337   g_configuration->eal_initialized = 0;
 338   g_configuration->helper_action = action;
 339   /*
 340    * Create a pipe for helper stdout back to collectd. This is necessary for
 341    * logging EAL failures, as rte_eal_init() calls rte_panic().
 342    */
 343   if(g_configuration->helper_pipes[1]) {
 344     DEBUG("dpdkstat: collectd closing helper pipe %d\n",
 345           g_configuration->helper_pipes[1]);
 346   } else {
 347     DEBUG("dpdkstat: collectd helper pipe %d, not closing\n",
 348           g_configuration->helper_pipes[1]);
 349   }
 350   if(pipe(g_configuration->helper_pipes) != 0) {
 351     DEBUG("dpdkstat: Could not create helper pipe: %s\n", strerror(errno));
 352     return -1;
 353   }
 354
 355   int pipe0_flags = fcntl(g_configuration->helper_pipes[1], F_GETFL, 0);
 356   int pipe1_flags = fcntl(g_configuration->helper_pipes[0], F_GETFL, 0);
 357   fcntl(g_configuration->helper_pipes[1], F_SETFL, pipe1_flags | O_NONBLOCK);
 358   fcntl(g_configuration->helper_pipes[0], F_SETFL, pipe0_flags | O_NONBLOCK);
 359
 360   pid_t pid = fork();
 361   if (pid > 0) {
 362     close(g_configuration->helper_pipes[1]);
 363     g_configuration->helper_pid = pid;
 364     DEBUG("dpdkstat: helper pid %u\n", g_configuration->helper_pid);
 365     /* Kick helper once its alive to have it start processing */
 366     sem_post(&g_configuration->sema_helper_get_stats);
 367   } else if (pid == 0) {
 368     /* Replace stdout with a pipe to collectd. */
 369     close(g_configuration->helper_pipes[0]);
 370     close(STDOUT_FILENO);
 371     dup2(g_configuration->helper_pipes[1], STDOUT_FILENO);
 372     dpdk_helper_run();
 373   } else {
 374     ERROR("dpdkstat: Failed to fork helper process: %s\n", strerror(errno));
 375     return -1;
 376   }
 377   return 0;
 378 }
 379
 380 /*
 381  * Initialize the DPDK EAL, if this returns, EAL is successfully initialized.
 382  * On failure, the EAL prints an error message, and the helper process exits.
 383  */
 384 static int dpdk_helper_init_eal(void)
 385 {
 386   g_configuration->helper_status = DPDK_HELPER_INITIALIZING_EAL;
 387   char *argp[(g_configuration->eal_argc) + 1];
 388   int i = 0;
 389
 390   argp[i++] = "collectd-dpdk";
 391   if(strcasecmp(g_configuration->coremask, "") != 0) {
 392     argp[i++] = "-c";
 393     argp[i++] = g_configuration->coremask;
 394   }
 395   if(strcasecmp(g_configuration->memory_channels, "") != 0) {
 396     argp[i++] = "-n";
 397     argp[i++] = g_configuration->memory_channels;
 398   }
 399   if(strcasecmp(g_configuration->socket_memory, "") != 0) {
 400     argp[i++] = "--socket-mem";
 401     argp[i++] = g_configuration->socket_memory;
 402   }
 403   if(strcasecmp(g_configuration->file_prefix, "") != 0 &&
 404      strcasecmp(g_configuration->file_prefix, "/var/run/.rte_config") != 0) {
 405     argp[i++] = "--file-prefix";
 406     argp[i++] = g_configuration->file_prefix;
 407   }
 408   if(strcasecmp(g_configuration->process_type, "") != 0) {
 409     argp[i++] = "--proc-type";
 410     argp[i++] = g_configuration->process_type;
 411   }
 412   g_configuration->eal_argc = i;
 413
 414   g_configuration->eal_initialized = 1;
 415   int ret = rte_eal_init(g_configuration->eal_argc, argp);
 416   if (ret < 0) {
 417     g_configuration->eal_initialized = 0;
 418     printf("dpdkstat: ERROR initializing EAL ret = %d\n", ret);
 419     printf("dpdkstat: EAL arguments: ");
 420     for (i=0; i< g_configuration->eal_argc; i++) {
 421       printf("%s ", argp[i]);
 422     }
 423     printf("\n");
 424     return -1;
 425   }
 426   return 0;
 427 }
 428
 429 static int dpdk_helper_run (void)
 430 {
 431   pid_t ppid = getppid();
 432   g_configuration->helper_status = DPDK_HELPER_WAITING_ON_PRIMARY;
 433
 434    while(1) {
 435     /* sem_timedwait() to avoid blocking forever */
 436     struct timespec ts;
 437     cdtime_t now = cdtime();
 438     cdtime_t half_sec = MS_TO_CDTIME_T(1500);
 439     CDTIME_T_TO_TIMESPEC(now + half_sec + g_configuration->interval *2, &ts);
 440     int ret = sem_timedwait(&g_configuration->sema_helper_get_stats, &ts);
 441
 442     if(ret == -1 && errno == ETIMEDOUT) {
 443       ERROR("dpdkstat-helper: sem timedwait()"
 444              " timeout, did collectd terminate?\n");
 445       dpdk_helper_exit(RESET);
 446     }
 447
 448     /* Parent PID change means collectd died so quit the helper process. */
 449     if (ppid != getppid()) {
 450       WARNING("dpdkstat-helper: parent PID changed, quitting.\n");
 451       dpdk_helper_exit(RESET);
 452     }
 453
 454     /* Checking for DPDK primary process. */
 455     if (!rte_eal_primary_proc_alive(g_configuration->file_prefix)) {
 456       if(g_configuration->eal_initialized) {
 457         WARNING("dpdkstat-helper: no primary alive but EAL initialized:"
 458               " quitting.\n");
 459         dpdk_helper_exit(RESET);
 460       }
 461       g_configuration->helper_status = DPDK_HELPER_WAITING_ON_PRIMARY;
 462       /* Back to start of while() - waiting for primary process */
 463       continue;
 464     }
 465
 466     if(!g_configuration->eal_initialized) {
 467       /* Initialize EAL. */
 468       int ret = dpdk_helper_init_eal();
 469       if(ret != 0)
 470         dpdk_helper_exit(RESET);
 471     }
 472
 473     g_configuration->helper_status = DPDK_HELPER_ALIVE_SENDING_STATS;
 474
 475     uint8_t nb_ports;
 476     nb_ports = rte_eth_dev_count();
 477     if (nb_ports == 0) {
 478       DEBUG("dpdkstat-helper: No DPDK ports available. "
 479               "Check bound devices to DPDK driver.\n");
 480       return 0;
 481     }
 482
 483     if (nb_ports > RTE_MAX_ETHPORTS)
 484       nb_ports = RTE_MAX_ETHPORTS;
 485     /* If no port mask was specified enable all ports*/
 486     if (g_configuration->enabled_port_mask == 0)
 487       g_configuration->enabled_port_mask = 0xffff;
 488
 489     int i, len = 0, enabled_port_count = 0, num_xstats = 0;
 490     for (i = 0; i < nb_ports; i++) {
 491       if (g_configuration->enabled_port_mask & (1 << i)) {
 492         if(g_configuration->helper_action == DPDK_HELPER_ACTION_COUNT_STATS) {
 493           len = rte_eth_xstats_get(i, NULL, 0);
 494           if (len < 0) {
 495             ERROR("dpdkstat-helper: Cannot get xstats count\n");
 496             return -1;
 497           }
 498           num_xstats += len;
 499           g_configuration->num_stats_in_port[enabled_port_count] = len;
 500           enabled_port_count++;
 501           continue;
 502         } else {
 503           len = g_configuration->num_stats_in_port[enabled_port_count];
 504           g_configuration->port_read_time[enabled_port_count] = cdtime();
 505           ret = rte_eth_xstats_get(i, &g_configuration->xstats + num_xstats,
 506                                    g_configuration->num_stats_in_port[i]);
 507           if (ret < 0 || ret != len) {
 508             DEBUG("dpdkstat-helper: Error reading xstats on port %d len = %d\n",
 509                   i, len);
 510             return -1;
 511           }
 512           num_xstats += g_configuration->num_stats_in_port[i];
 513         }
 514       } /* if (enabled_port_mask) */
 515     } /* for (nb_ports) */
 516
 517     if(g_configuration->helper_action == DPDK_HELPER_ACTION_COUNT_STATS) {
 518       g_configuration->num_ports  = enabled_port_count;
 519       g_configuration->num_xstats = num_xstats;
 520       DEBUG("dpdkstat-helper ports: %d, num stats: %d\n",
 521             g_configuration->num_ports, g_configuration->num_xstats);
 522       /* Exit, allowing collectd to re-init SHM to the right size */
 523       g_configuration->collectd_reinit_shm = REINIT_SHM;
 524       dpdk_helper_exit(NO_RESET);
 525     }
 526     /* Now kick collectd send thread to send the stats */
 527     sem_post(&g_configuration->sema_stats_in_shm);
 528   } /* while(1) */
 529
 530   return 0;
 531 }
 532
 533 static int dpdk_read (user_data_t *ud)
 534 {
 535   int ret = 0;
 536
 537   /*
 538    * Check if SHM flag is set to be re-initialized. AKA DPDK ports have been
 539    * counted, so re-init SHM to be large enough to fit all the statistics.
 540    */
 541   if(g_configuration->collectd_reinit_shm) {
 542     DEBUG("dpdkstat: read() now reinit SHM then launching send-thread\n");
 543     dpdk_re_init_shm();
 544   }
 545
 546   /*
 547    * Check if DPDK proc is alive, and has already counted port / stats. This
 548    * must be done in dpdk_read(), because the DPDK primary process may not be
 549    * alive at dpdk_init() time.
 550    */
 551   if(g_configuration->helper_status == DPDK_HELPER_NOT_INITIALIZED ||
 552      g_configuration->helper_status == DPDK_HELPER_GRACEFUL_QUIT) {
 553       int action = DPDK_HELPER_ACTION_SEND_STATS;
 554       if(g_configuration->num_xstats == 0)
 555         action = DPDK_HELPER_ACTION_COUNT_STATS;
 556       /* Spawn the helper thread to count stats or to read stats. */
 557       dpdk_helper_spawn(action);
 558     }
 559
 560   int exit_status;
 561   pid_t ws = waitpid(g_configuration->helper_pid, &exit_status, WNOHANG);
 562   /*
 563    * Conditions under which to respawn helper:
 564    *  waitpid() fails, helper process died (or quit), so respawn
 565    */
 566   int respawn_helper = 0;
 567   if(ws != 0) {
 568     respawn_helper = 1;
 569   }
 570
 571   char buf[DPDKSTAT_MAX_BUFFER_SIZE];
 572   char out[DPDKSTAT_MAX_BUFFER_SIZE];
 573
 574   /* non blocking check on helper logging pipe */
 575   struct pollfd fds;
 576   fds.fd = g_configuration->helper_pipes[0];
 577   fds.events = POLLIN;
 578   int data_avail = poll(&fds, 1, 0);
 579   while(data_avail) {
 580     int nbytes = read(g_configuration->helper_pipes[0], buf, sizeof(buf));
 581     if(nbytes <= 0)
 582       break;
 583     snprintf( out, nbytes, "%s", buf);
 584     DEBUG("dpdkstat: helper-proc: %s\n", out);
 585   }
 586
 587   if(respawn_helper) {
 588     if (g_configuration->helper_pid)
 589       dpdk_helper_exit(RESET);
 590     dpdk_helper_spawn(DPDK_HELPER_ACTION_COUNT_STATS);
 591   }
 592
 593   struct timespec helper_kick_time;
 594   clock_gettime(CLOCK_REALTIME, &helper_kick_time);
 595   /* Kick helper process through SHM */
 596   sem_post(&g_configuration->sema_helper_get_stats);
 597
 598   struct timespec ts;
 599   cdtime_t now = cdtime();
 600   CDTIME_T_TO_TIMESPEC(now + g_configuration->interval, &ts);
 601   ret = sem_timedwait(&g_configuration->sema_stats_in_shm, &ts);
 602   if(ret == -1 && errno == ETIMEDOUT) {
 603     DEBUG("dpdkstat: timeout in collectd thread: is a DPDK Primary running? \n");
 604     return 0;
 605   }
 606
 607   /* Dispatch the stats.*/
 608     int i, j, count = 0;
 609
 610     for (i = 0; i < g_configuration->num_ports; i++) {
 611       cdtime_t time = g_configuration->port_read_time[i];
 612       char dev_name[64];
 613       int len = g_configuration->num_stats_in_port[i];
 614       snprintf(dev_name, sizeof(dev_name), "port.%d", i);
 615       struct rte_eth_xstats *xstats = (&g_configuration->xstats);
 616       xstats += count; /* pointer arithmetic to jump to each stats struct */
 617       for (j = 0; j < len; j++) {
 618         value_t dpdkstat_values[1];
 619         value_list_t dpdkstat_vl = VALUE_LIST_INIT;
 620
 621         dpdkstat_values[0].counter = xstats[j].value;
 622         dpdkstat_vl.values = dpdkstat_values;
 623         dpdkstat_vl.values_len = 1; /* Submit stats one at a time */
 624         dpdkstat_vl.time = time;
 625         sstrncpy (dpdkstat_vl.host, hostname_g, sizeof (dpdkstat_vl.host));
 626         sstrncpy (dpdkstat_vl.plugin, "dpdkstat", sizeof (dpdkstat_vl.plugin));
 627         sstrncpy (dpdkstat_vl.plugin_instance, dev_name,
 628                   sizeof (dpdkstat_vl.plugin_instance));
 629         sstrncpy (dpdkstat_vl.type, "counter",
 630                   sizeof (dpdkstat_vl.type));
 631         sstrncpy (dpdkstat_vl.type_instance, xstats[j].name,
 632                   sizeof (dpdkstat_vl.type_instance));
 633         plugin_dispatch_values (&dpdkstat_vl);
 634       }
 635       count += len;
 636     } /* for each port */
 637   return 0;
 638 }
 639
 640 static int dpdk_shm_cleanup(void)
 641 {
 642   int ret = munmap(g_configuration, sizeof(dpdk_config_t));
 643   g_configuration = 0;
 644   if(ret) {
 645     WARNING("dpdkstat: munmap returned %d\n", ret);
 646     return ret;
 647   }
 648   ret = shm_unlink(DPDK_SHM_NAME);
 649   if(ret) {
 650     WARNING("dpdkstat: shm_unlink returned %d\n", ret);
 651     return ret;
 652   }
 653   return 0;
 654 }
 655
 656 static int dpdk_shutdown (void)
 657 {
 658   close(g_configuration->helper_pipes[1]);
 659   kill(g_configuration->helper_pid, SIGKILL);
 660   int ret = dpdk_shm_cleanup();
 661
 662   return ret;
 663 }
 664
 665 void module_register (void)
 666 {
 667   plugin_register_complex_config ("dpdkstat", dpdk_config);
 668   plugin_register_init ("dpdkstat", dpdk_init);
 669   plugin_register_shutdown ("dpdkstat", dpdk_shutdown);
 670 }