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