src/utils_dpdk.c

   1 /*
   2  * collectd - src/utils_dpdk.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
   8  * of this software and associated documentation files (the "Software"), to deal
   9  * in the Software without restriction, including without limitation the rights
  10  * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  11  * copies of the Software, and to permit persons to whom the Software is
  12  * furnished to do so, subject to the following conditions:
  13  *
  14  * The above copyright notice and this permission notice shall be included in
  15  * all 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  *   Taras Chornyi <tarasx.chornyi@intel.com>
  29  *   Serhiy Pshyk <serhiyx.pshyk@intel.com>
  30  *   Krzysztof Matczak <krzysztofx.matczak@intel.com>
  31  */
  32
  33 #include "collectd.h"
  34
  35 #include <poll.h>
  36 #include <semaphore.h>
  37 #include <sys/mman.h>
  38
  39 #include <rte_config.h>
  40 #include <rte_eal.h>
  41 #include <rte_ethdev.h>
  42
  43 #include "common.h"
  44 #include "utils_dpdk.h"
  45
  46 #define DPDK_DEFAULT_RTE_CONFIG "/var/run/.rte_config"
  47 #define DPDK_EAL_ARGC 10
  48 // Complete trace should fit into 1024 chars. Trace contain some headers
  49 // and text together with traced data from pipe. This is the reason why
  50 // we need to limit DPDK_MAX_BUFFER_SIZE value.
  51 #define DPDK_MAX_BUFFER_SIZE 896
  52 #define DPDK_CDM_DEFAULT_TIMEOUT 10000
  53
  54 enum DPDK_HELPER_STATUS {
  55   DPDK_HELPER_NOT_INITIALIZED = 0,
  56   DPDK_HELPER_INITIALIZING,
  57   DPDK_HELPER_WAITING_ON_PRIMARY,
  58   DPDK_HELPER_INITIALIZING_EAL,
  59   DPDK_HELPER_ALIVE_SENDING_EVENTS,
  60   DPDK_HELPER_GRACEFUL_QUIT,
  61 };
  62
  63 #define DPDK_HELPER_TRACE(_name)                                               \
  64   DEBUG("%s:%s:%d pid=%ld", _name, __FUNCTION__, __LINE__, (long)getpid())
  65
  66 struct dpdk_helper_ctx_s {
  67
  68   dpdk_eal_config_t eal_config;
  69   int eal_initialized;
  70
  71   size_t shm_size;
  72   char shm_name[DATA_MAX_NAME_LEN];
  73
  74   sem_t sema_cmd_start;
  75   sem_t sema_cmd_complete;
  76   cdtime_t cmd_wait_time;
  77
  78   pid_t pid;
  79   int pipes[2];
  80   int status;
  81
  82   int cmd;
  83   int cmd_result;
  84
  85   char priv_data[];
  86 };
  87
  88 static int dpdk_shm_init(const char *name, size_t size, void **map);
  89 static void dpdk_shm_cleanup(const char *name, size_t size, void *map);
  90
  91 static int dpdk_helper_spawn(dpdk_helper_ctx_t *phc);
  92 static int dpdk_helper_worker(dpdk_helper_ctx_t *phc);
  93 static int dpdk_helper_eal_init(dpdk_helper_ctx_t *phc);
  94 static int dpdk_helper_cmd_wait(dpdk_helper_ctx_t *phc, pid_t ppid);
  95 static int dpdk_helper_exit_command(dpdk_helper_ctx_t *phc,
  96                                     enum DPDK_HELPER_STATUS status);
  97 static int dpdk_helper_exit(dpdk_helper_ctx_t *phc,
  98                             enum DPDK_HELPER_STATUS status);
  99 static int dpdk_helper_status_check(dpdk_helper_ctx_t *phc);
 100 static void dpdk_helper_config_default(dpdk_helper_ctx_t *phc);
 101 static const char *dpdk_helper_status_str(enum DPDK_HELPER_STATUS status);
 102
 103 static void dpdk_helper_config_default(dpdk_helper_ctx_t *phc) {
 104   if (phc == NULL)
 105     return;
 106
 107   DPDK_HELPER_TRACE(phc->shm_name);
 108
 109   snprintf(phc->eal_config.coremask, DATA_MAX_NAME_LEN, "%s", "0xf");
 110   snprintf(phc->eal_config.memory_channels, DATA_MAX_NAME_LEN, "%s", "1");
 111   snprintf(phc->eal_config.file_prefix, DATA_MAX_NAME_LEN, "%s",
 112            DPDK_DEFAULT_RTE_CONFIG);
 113 }
 114
 115 int dpdk_helper_eal_config_set(dpdk_helper_ctx_t *phc, dpdk_eal_config_t *ec) {
 116   if (phc == NULL) {
 117     ERROR("Invalid argument (phc)");
 118     return -EINVAL;
 119   }
 120
 121   DPDK_HELPER_TRACE(phc->shm_name);
 122
 123   if (ec == NULL) {
 124     ERROR("Invalid argument (ec)");
 125     return -EINVAL;
 126   }
 127
 128   memcpy(&phc->eal_config, ec, sizeof(phc->eal_config));
 129
 130   return 0;
 131 }
 132
 133 int dpdk_helper_eal_config_get(dpdk_helper_ctx_t *phc, dpdk_eal_config_t *ec) {
 134   if (phc == NULL) {
 135     ERROR("Invalid argument (phc)");
 136     return -EINVAL;
 137   }
 138
 139   DPDK_HELPER_TRACE(phc->shm_name);
 140
 141   if (ec == NULL) {
 142     ERROR("Invalid argument (ec)");
 143     return -EINVAL;
 144   }
 145
 146   memcpy(ec, &phc->eal_config, sizeof(*ec));
 147
 148   return 0;
 149 }
 150
 151 int dpdk_helper_eal_config_parse(dpdk_helper_ctx_t *phc, oconfig_item_t *ci) {
 152   DPDK_HELPER_TRACE(phc->shm_name);
 153
 154   if (phc == NULL) {
 155     ERROR("Invalid argument (phc)");
 156     return -EINVAL;
 157   }
 158
 159   if (ci == NULL) {
 160     ERROR("Invalid argument (ci)");
 161     return -EINVAL;
 162   }
 163
 164   int status = 0;
 165   for (int i = 0; i < ci->children_num; i++) {
 166     oconfig_item_t *child = ci->children + i;
 167
 168     if (strcasecmp("Coremask", child->key) == 0) {
 169       status = cf_util_get_string_buffer(child, phc->eal_config.coremask,
 170                                          sizeof(phc->eal_config.coremask));
 171       DEBUG("dpdk_common: EAL:Coremask %s", phc->eal_config.coremask);
 172     } else if (strcasecmp("MemoryChannels", child->key) == 0) {
 173       status =
 174           cf_util_get_string_buffer(child, phc->eal_config.memory_channels,
 175                                     sizeof(phc->eal_config.memory_channels));
 176       DEBUG("dpdk_common: EAL:Memory Channels %s",
 177             phc->eal_config.memory_channels);
 178     } else if (strcasecmp("SocketMemory", child->key) == 0) {
 179       status = cf_util_get_string_buffer(child, phc->eal_config.socket_memory,
 180                                          sizeof(phc->eal_config.socket_memory));
 181       DEBUG("dpdk_common: EAL:Socket memory %s", phc->eal_config.socket_memory);
 182     } else if (strcasecmp("FilePrefix", child->key) == 0) {
 183       char prefix[DATA_MAX_NAME_LEN];
 184
 185       status = cf_util_get_string_buffer(child, prefix, sizeof(prefix));
 186       if (status == 0) {
 187         snprintf(phc->eal_config.file_prefix, DATA_MAX_NAME_LEN,
 188                  "/var/run/.%s_config", prefix);
 189         DEBUG("dpdk_common: EAL:File prefix %s", phc->eal_config.file_prefix);
 190       }
 191     } else if (strcasecmp("LogLevel", child->key) == 0) {
 192       status = cf_util_get_string_buffer(child, phc->eal_config.log_level,
 193                                          sizeof(phc->eal_config.log_level));
 194       DEBUG("dpdk_common: EAL:LogLevel %s", phc->eal_config.log_level);
 195     } else if (strcasecmp("RteDriverLibPath", child->key) == 0) {
 196       status = cf_util_get_string_buffer(
 197           child, phc->eal_config.rte_driver_lib_path,
 198           sizeof(phc->eal_config.rte_driver_lib_path));
 199       DEBUG("dpdk_common: EAL:RteDriverLibPath %s",
 200             phc->eal_config.rte_driver_lib_path);
 201     } else {
 202       ERROR("dpdk_common: Invalid '%s' configuration option", child->key);
 203       status = -EINVAL;
 204     }
 205
 206     if (status != 0) {
 207       ERROR("dpdk_common: Parsing EAL configuration failed");
 208       break;
 209     }
 210   }
 211
 212   return status;
 213 }
 214
 215 static int dpdk_shm_init(const char *name, size_t size, void **map) {
 216   DPDK_HELPER_TRACE(name);
 217
 218   char errbuf[ERR_BUF_SIZE];
 219
 220   int fd = shm_open(name, O_CREAT | O_TRUNC | O_RDWR, 0666);
 221   if (fd < 0) {
 222     WARNING("dpdk_shm_init: Failed to open %s as SHM:%s", name,
 223             sstrerror(errno, errbuf, sizeof(errbuf)));
 224     *map = NULL;
 225     return -1;
 226   }
 227
 228   int ret = ftruncate(fd, size);
 229   if (ret != 0) {
 230     WARNING("dpdk_shm_init: Failed to resize SHM:%s",
 231             sstrerror(errno, errbuf, sizeof(errbuf)));
 232     close(fd);
 233     *map = NULL;
 234     dpdk_shm_cleanup(name, size, NULL);
 235     return -1;
 236   }
 237
 238   *map = mmap(0, size, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
 239   if (*map == MAP_FAILED) {
 240     WARNING("dpdk_shm_init:Failed to mmap SHM:%s",
 241             sstrerror(errno, errbuf, sizeof(errbuf)));
 242     close(fd);
 243     *map = NULL;
 244     dpdk_shm_cleanup(name, size, NULL);
 245     return -1;
 246   }
 247   /* File descriptor no longer needed for shared memory operations */
 248   close(fd);
 249   memset(*map, 0, size);
 250
 251   return 0;
 252 }
 253
 254 static void dpdk_shm_cleanup(const char *name, size_t size, void *map) {
 255   DPDK_HELPER_TRACE(name);
 256   char errbuf[ERR_BUF_SIZE];
 257
 258   /*
 259    * Call shm_unlink first, as 'name' might be no longer accessible after munmap
 260    */
 261   if (shm_unlink(name))
 262     ERROR("shm_unlink failure %s", sstrerror(errno, errbuf, sizeof(errbuf)));
 263
 264   if (map != NULL) {
 265     if (munmap(map, size))
 266       ERROR("munmap failure %s", sstrerror(errno, errbuf, sizeof(errbuf)));
 267   }
 268 }
 269
 270 void *dpdk_helper_priv_get(dpdk_helper_ctx_t *phc) {
 271   if (phc)
 272     return phc->priv_data;
 273
 274   return NULL;
 275 }
 276
 277 int dpdk_helper_data_size_get(dpdk_helper_ctx_t *phc) {
 278   if (phc == NULL) {
 279     DPDK_CHILD_LOG("Invalid argument(phc)\n");
 280     return -EINVAL;
 281   }
 282
 283   return phc->shm_size - sizeof(dpdk_helper_ctx_t);
 284 }
 285
 286 int dpdk_helper_init(const char *name, size_t data_size,
 287                      dpdk_helper_ctx_t **pphc) {
 288   dpdk_helper_ctx_t *phc = NULL;
 289   size_t shm_size = sizeof(dpdk_helper_ctx_t) + data_size;
 290   char errbuf[ERR_BUF_SIZE];
 291
 292   if (pphc == NULL) {
 293     ERROR("%s:Invalid argument(pphc)", __FUNCTION__);
 294     return -EINVAL;
 295   }
 296
 297   if (name == NULL) {
 298     ERROR("%s:Invalid argument(name)", __FUNCTION__);
 299     return -EINVAL;
 300   }
 301
 302   DPDK_HELPER_TRACE(name);
 303
 304   /* Allocate dpdk_helper_ctx_t and
 305   * initialize a POSIX SHared Memory (SHM) object.
 306   */
 307   int err = dpdk_shm_init(name, shm_size, (void **)&phc);
 308   if (err != 0) {
 309     return -errno;
 310   }
 311
 312   err = sem_init(&phc->sema_cmd_start, 1, 0);
 313   if (err != 0) {
 314     ERROR("sema_cmd_start semaphore init failed: %s",
 315           sstrerror(errno, errbuf, sizeof(errbuf)));
 316     int errno_m = errno;
 317     dpdk_shm_cleanup(name, shm_size, (void *)phc);
 318     return -errno_m;
 319   }
 320
 321   err = sem_init(&phc->sema_cmd_complete, 1, 0);
 322   if (err != 0) {
 323     ERROR("sema_cmd_complete semaphore init failed: %s",
 324           sstrerror(errno, errbuf, sizeof(errbuf)));
 325     sem_destroy(&phc->sema_cmd_start);
 326     int errno_m = errno;
 327     dpdk_shm_cleanup(name, shm_size, (void *)phc);
 328     return -errno_m;
 329   }
 330
 331   phc->shm_size = shm_size;
 332   sstrncpy(phc->shm_name, name, sizeof(phc->shm_name));
 333
 334   dpdk_helper_config_default(phc);
 335
 336   *pphc = phc;
 337
 338   return 0;
 339 }
 340
 341 void dpdk_helper_shutdown(dpdk_helper_ctx_t *phc) {
 342   if (phc == NULL)
 343     return;
 344
 345   DPDK_HELPER_TRACE(phc->shm_name);
 346
 347   close(phc->pipes[1]);
 348
 349   if (phc->status != DPDK_HELPER_NOT_INITIALIZED) {
 350     dpdk_helper_exit_command(phc, DPDK_HELPER_GRACEFUL_QUIT);
 351   }
 352
 353   sem_destroy(&phc->sema_cmd_start);
 354   sem_destroy(&phc->sema_cmd_complete);
 355   dpdk_shm_cleanup(phc->shm_name, phc->shm_size, (void *)phc);
 356 }
 357
 358 static int dpdk_helper_spawn(dpdk_helper_ctx_t *phc) {
 359   char errbuf[ERR_BUF_SIZE];
 360   if (phc == NULL) {
 361     ERROR("Invalid argument(phc)");
 362     return -EINVAL;
 363   }
 364
 365   DPDK_HELPER_TRACE(phc->shm_name);
 366
 367   phc->eal_initialized = 0;
 368   phc->cmd_wait_time = MS_TO_CDTIME_T(DPDK_CDM_DEFAULT_TIMEOUT);
 369
 370   /*
 371    * Create a pipe for helper stdout back to collectd. This is necessary for
 372    * logging EAL failures, as rte_eal_init() calls rte_panic().
 373    */
 374   if (phc->pipes[1]) {
 375     DEBUG("dpdk_helper_spawn: collectd closing helper pipe %d", phc->pipes[1]);
 376   } else {
 377     DEBUG("dpdk_helper_spawn: collectd helper pipe %d, not closing",
 378           phc->pipes[1]);
 379   }
 380
 381   if (pipe(phc->pipes) != 0) {
 382     DEBUG("dpdk_helper_spawn: Could not create helper pipe: %s",
 383           sstrerror(errno, errbuf, sizeof(errbuf)));
 384     return -1;
 385   }
 386
 387   int pipe0_flags = fcntl(phc->pipes[0], F_GETFL, 0);
 388   int pipe1_flags = fcntl(phc->pipes[1], F_GETFL, 0);
 389   if (pipe0_flags == -1 || pipe1_flags == -1) {
 390     WARNING("dpdk_helper_spawn: error setting up pipe flags: %s",
 391             sstrerror(errno, errbuf, sizeof(errbuf)));
 392   }
 393   int pipe0_err = fcntl(phc->pipes[0], F_SETFL, pipe1_flags | O_NONBLOCK);
 394   int pipe1_err = fcntl(phc->pipes[1], F_SETFL, pipe0_flags | O_NONBLOCK);
 395   if (pipe0_err == -1 || pipe1_err == -1) {
 396     WARNING("dpdk_helper_spawn: error setting up pipes: %s",
 397             sstrerror(errno, errbuf, sizeof(errbuf)));
 398   }
 399
 400   pid_t pid = fork();
 401   if (pid > 0) {
 402     phc->pid = pid;
 403     close(phc->pipes[1]);
 404     DEBUG("%s:dpdk_helper_spawn: helper pid %lu", phc->shm_name,
 405           (long)phc->pid);
 406   } else if (pid == 0) {
 407     /* Replace stdout with a pipe to collectd. */
 408     close(phc->pipes[0]);
 409     close(STDOUT_FILENO);
 410     dup2(phc->pipes[1], STDOUT_FILENO);
 411     DPDK_CHILD_TRACE(phc->shm_name);
 412     dpdk_helper_worker(phc);
 413     exit(0);
 414   } else {
 415     ERROR("dpdk_helper_start: Failed to fork helper process: %s",
 416           sstrerror(errno, errbuf, sizeof(errbuf)));
 417     return -1;
 418   }
 419
 420   return 0;
 421 }
 422
 423 static int dpdk_helper_exit(dpdk_helper_ctx_t *phc,
 424                             enum DPDK_HELPER_STATUS status) {
 425   DPDK_CHILD_LOG("%s:%s:%d %s\n", phc->shm_name, __FUNCTION__, __LINE__,
 426                  dpdk_helper_status_str(status));
 427
 428   close(phc->pipes[1]);
 429
 430   phc->status = status;
 431
 432   exit(0);
 433
 434   return 0;
 435 }
 436
 437 static int dpdk_helper_exit_command(dpdk_helper_ctx_t *phc,
 438                                     enum DPDK_HELPER_STATUS status) {
 439   char errbuf[ERR_BUF_SIZE];
 440   DPDK_HELPER_TRACE(phc->shm_name);
 441
 442   close(phc->pipes[1]);
 443
 444   if (phc->status == DPDK_HELPER_ALIVE_SENDING_EVENTS) {
 445     phc->status = status;
 446     DEBUG("%s:%s:%d %s", phc->shm_name, __FUNCTION__, __LINE__,
 447           dpdk_helper_status_str(status));
 448
 449     int ret = dpdk_helper_command(phc, DPDK_CMD_QUIT, NULL, 0);
 450     if (ret != 0) {
 451       DEBUG("%s:%s:%d kill helper (pid=%lu)", phc->shm_name, __FUNCTION__,
 452             __LINE__, (long)phc->pid);
 453
 454       int err = kill(phc->pid, SIGKILL);
 455       if (err) {
 456         ERROR("%s error sending kill to helper: %s", __FUNCTION__,
 457               sstrerror(errno, errbuf, sizeof(errbuf)));
 458       }
 459     }
 460   } else {
 461
 462     DEBUG("%s:%s:%d kill helper (pid=%lu)", phc->shm_name, __FUNCTION__,
 463           __LINE__, (long)phc->pid);
 464
 465     int err = kill(phc->pid, SIGKILL);
 466     if (err) {
 467       ERROR("%s error sending kill to helper: %s", __FUNCTION__,
 468             sstrerror(errno, errbuf, sizeof(errbuf)));
 469     }
 470   }
 471
 472   return 0;
 473 }
 474
 475 static int dpdk_helper_eal_init(dpdk_helper_ctx_t *phc) {
 476   phc->status = DPDK_HELPER_INITIALIZING_EAL;
 477   DPDK_CHILD_LOG("%s:%s:%d DPDK_HELPER_INITIALIZING_EAL (start)\n",
 478                  phc->shm_name, __FUNCTION__, __LINE__);
 479
 480   char *argp[DPDK_EAL_ARGC * 2 + 1];
 481   int argc = 0;
 482
 483   /* EAL config must be initialized */
 484   assert(phc->eal_config.coremask[0] != 0);
 485   assert(phc->eal_config.memory_channels[0] != 0);
 486   assert(phc->eal_config.file_prefix[0] != 0);
 487
 488   argp[argc++] = "collectd-dpdk";
 489
 490   argp[argc++] = "-c";
 491   argp[argc++] = phc->eal_config.coremask;
 492
 493   argp[argc++] = "-n";
 494   argp[argc++] = phc->eal_config.memory_channels;
 495
 496   if (strcasecmp(phc->eal_config.socket_memory, "") != 0) {
 497     argp[argc++] = "--socket-mem";
 498     argp[argc++] = phc->eal_config.socket_memory;
 499   }
 500
 501   if (strcasecmp(phc->eal_config.file_prefix, DPDK_DEFAULT_RTE_CONFIG) != 0) {
 502     argp[argc++] = "--file-prefix";
 503     argp[argc++] = phc->eal_config.file_prefix;
 504   }
 505
 506   argp[argc++] = "--proc-type";
 507   argp[argc++] = "secondary";
 508
 509   if (strcasecmp(phc->eal_config.log_level, "") != 0) {
 510     argp[argc++] = "--log-level";
 511     argp[argc++] = phc->eal_config.log_level;
 512   }
 513   if (strcasecmp(phc->eal_config.rte_driver_lib_path, "") != 0) {
 514     argp[argc++] = "-d";
 515     argp[argc++] = phc->eal_config.rte_driver_lib_path;
 516   }
 517
 518   assert(argc <= (DPDK_EAL_ARGC * 2 + 1));
 519
 520   int ret = rte_eal_init(argc, argp);
 521
 522   if (ret < 0) {
 523
 524     phc->eal_initialized = 0;
 525
 526     DPDK_CHILD_LOG("dpdk_helper_eal_init: ERROR initializing EAL ret=%d\n",
 527                    ret);
 528
 529     printf("dpdk_helper_eal_init: EAL arguments: ");
 530     for (int i = 0; i < argc; i++) {
 531       printf("%s ", argp[i]);
 532     }
 533     printf("\n");
 534
 535     return ret;
 536   }
 537
 538   phc->eal_initialized = 1;
 539
 540   DPDK_CHILD_LOG("%s:%s:%d DPDK_HELPER_INITIALIZING_EAL (done)\n",
 541                  phc->shm_name, __FUNCTION__, __LINE__);
 542
 543   return 0;
 544 }
 545
 546 static int dpdk_helper_cmd_wait(dpdk_helper_ctx_t *phc, pid_t ppid) {
 547   DPDK_CHILD_TRACE(phc->shm_name);
 548
 549   struct timespec ts;
 550   cdtime_t now = cdtime();
 551   cdtime_t cmd_wait_time = MS_TO_CDTIME_T(1500) + phc->cmd_wait_time * 2;
 552   ts = CDTIME_T_TO_TIMESPEC(now + cmd_wait_time);
 553
 554   int ret = sem_timedwait(&phc->sema_cmd_start, &ts);
 555   DPDK_CHILD_LOG("%s:%s:%d pid=%lu got sema_cmd_start (ret=%d, errno=%d)\n",
 556                  phc->shm_name, __FUNCTION__, __LINE__, (long)getpid(), ret,
 557                  errno);
 558
 559   if (phc->cmd == DPDK_CMD_QUIT) {
 560     DPDK_CHILD_LOG("%s:%s:%d pid=%lu exiting\n", phc->shm_name, __FUNCTION__,
 561                    __LINE__, (long)getpid());
 562     exit(0);
 563   } else if (ret == -1 && errno == ETIMEDOUT) {
 564     if (phc->status == DPDK_HELPER_ALIVE_SENDING_EVENTS) {
 565       DPDK_CHILD_LOG("%s:dpdk_helper_cmd_wait: sem timedwait()"
 566                      " timeout, did collectd terminate?\n",
 567                      phc->shm_name);
 568       dpdk_helper_exit(phc, DPDK_HELPER_GRACEFUL_QUIT);
 569     }
 570   }
 571 #if COLLECT_DEBUG
 572   int val = 0;
 573   if (sem_getvalue(&phc->sema_cmd_start, &val) == 0)
 574     DPDK_CHILD_LOG("%s:%s:%d pid=%lu wait sema_cmd_start (value=%d)\n",
 575                    phc->shm_name, __FUNCTION__, __LINE__, (long)getpid(), val);
 576 #endif
 577
 578   /* Parent PID change means collectd died so quit the helper process. */
 579   if (ppid != getppid()) {
 580     DPDK_CHILD_LOG("dpdk_helper_cmd_wait: parent PID changed, quitting.\n");
 581     dpdk_helper_exit(phc, DPDK_HELPER_GRACEFUL_QUIT);
 582   }
 583
 584   /* Checking for DPDK primary process. */
 585   if (!rte_eal_primary_proc_alive(phc->eal_config.file_prefix)) {
 586     if (phc->eal_initialized) {
 587       DPDK_CHILD_LOG(
 588           "%s:dpdk_helper_cmd_wait: no primary alive but EAL initialized:"
 589           " quitting.\n",
 590           phc->shm_name);
 591       dpdk_helper_exit(phc, DPDK_HELPER_NOT_INITIALIZED);
 592     }
 593
 594     phc->status = DPDK_HELPER_WAITING_ON_PRIMARY;
 595     DPDK_CHILD_LOG("%s:%s:%d DPDK_HELPER_WAITING_ON_PRIMARY\n", phc->shm_name,
 596                    __FUNCTION__, __LINE__);
 597
 598     return -1;
 599   }
 600
 601   if (!phc->eal_initialized) {
 602     int ret = dpdk_helper_eal_init(phc);
 603     if (ret != 0) {
 604       DPDK_CHILD_LOG("Error initializing EAL\n");
 605       dpdk_helper_exit(phc, DPDK_HELPER_NOT_INITIALIZED);
 606     }
 607     phc->status = DPDK_HELPER_ALIVE_SENDING_EVENTS;
 608     DPDK_CHILD_LOG("%s:%s:%d DPDK_HELPER_ALIVE_SENDING_EVENTS\n", phc->shm_name,
 609                    __FUNCTION__, __LINE__);
 610     return -1;
 611   }
 612
 613   return 0;
 614 }
 615
 616 static int dpdk_helper_worker(dpdk_helper_ctx_t *phc) {
 617   DPDK_CHILD_TRACE(phc->shm_name);
 618
 619   pid_t ppid = getppid();
 620
 621   while (1) {
 622     if (dpdk_helper_cmd_wait(phc, ppid) == 0) {
 623       DPDK_CHILD_LOG("%s:%s:%d DPDK command handle (cmd=%d, pid=%lu)\n",
 624                      phc->shm_name, __FUNCTION__, __LINE__, phc->cmd,
 625                      (long)getpid());
 626       phc->cmd_result = dpdk_helper_command_handler(phc, phc->cmd);
 627     } else {
 628       phc->cmd_result = -1;
 629     }
 630
 631     /* now kick collectd to get results */
 632     int err = sem_post(&phc->sema_cmd_complete);
 633     DPDK_CHILD_LOG("%s:%s:%d post sema_cmd_complete (pid=%lu)\n", phc->shm_name,
 634                    __FUNCTION__, __LINE__, (long)getpid());
 635     if (err) {
 636       char errbuf[ERR_BUF_SIZE];
 637       DPDK_CHILD_LOG("dpdk_helper_worker: error posting sema_cmd_complete "
 638                      "semaphore (%s)\n",
 639                      sstrerror(errno, errbuf, sizeof(errbuf)));
 640     }
 641
 642 #if COLLECT_DEBUG
 643     int val = 0;
 644     if (sem_getvalue(&phc->sema_cmd_complete, &val) == 0)
 645       DPDK_CHILD_LOG("%s:%s:%d pid=%lu sema_cmd_complete (value=%d)\n",
 646                      phc->shm_name, __FUNCTION__, __LINE__, (long)getpid(),
 647                      val);
 648 #endif
 649
 650   } /* while(1) */
 651
 652   return 0;
 653 }
 654
 655 static const char *dpdk_helper_status_str(enum DPDK_HELPER_STATUS status) {
 656   switch (status) {
 657   case DPDK_HELPER_ALIVE_SENDING_EVENTS:
 658     return "DPDK_HELPER_ALIVE_SENDING_EVENTS";
 659   case DPDK_HELPER_WAITING_ON_PRIMARY:
 660     return "DPDK_HELPER_WAITING_ON_PRIMARY";
 661   case DPDK_HELPER_INITIALIZING:
 662     return "DPDK_HELPER_INITIALIZING";
 663   case DPDK_HELPER_INITIALIZING_EAL:
 664     return "DPDK_HELPER_INITIALIZING_EAL";
 665   case DPDK_HELPER_GRACEFUL_QUIT:
 666     return "DPDK_HELPER_GRACEFUL_QUIT";
 667   case DPDK_HELPER_NOT_INITIALIZED:
 668     return "DPDK_HELPER_NOT_INITIALIZED";
 669   default:
 670     return "UNKNOWN";
 671   }
 672 }
 673
 674 static int dpdk_helper_status_check(dpdk_helper_ctx_t *phc) {
 675   DEBUG("%s:%s:%d pid=%u %s", phc->shm_name, __FUNCTION__, __LINE__, getpid(),
 676         dpdk_helper_status_str(phc->status));
 677   char errbuf[ERR_BUF_SIZE];
 678
 679   if (phc->status == DPDK_HELPER_GRACEFUL_QUIT) {
 680     return 0;
 681   } else if (phc->status == DPDK_HELPER_NOT_INITIALIZED) {
 682     phc->status = DPDK_HELPER_INITIALIZING;
 683     DEBUG("%s:%s:%d DPDK_HELPER_INITIALIZING", phc->shm_name, __FUNCTION__,
 684           __LINE__);
 685     int err = dpdk_helper_spawn(phc);
 686     if (err) {
 687       ERROR("dpdkstat: error spawning helper %s",
 688             sstrerror(errno, errbuf, sizeof(errbuf)));
 689     }
 690     return -1;
 691   }
 692
 693   pid_t ws = waitpid(phc->pid, NULL, WNOHANG);
 694   if (ws != 0) {
 695     phc->status = DPDK_HELPER_INITIALIZING;
 696     DEBUG("%s:%s:%d DPDK_HELPER_INITIALIZING", phc->shm_name, __FUNCTION__,
 697           __LINE__);
 698     int err = dpdk_helper_spawn(phc);
 699     if (err) {
 700       ERROR("dpdkstat: error spawning helper %s",
 701             sstrerror(errno, errbuf, sizeof(errbuf)));
 702     }
 703     return -1;
 704   }
 705
 706   if (phc->status == DPDK_HELPER_INITIALIZING_EAL) {
 707     return -1;
 708   }
 709
 710   return 0;
 711 }
 712
 713 static void dpdk_helper_check_pipe(dpdk_helper_ctx_t *phc) {
 714   char buf[DPDK_MAX_BUFFER_SIZE];
 715   char out[DPDK_MAX_BUFFER_SIZE];
 716
 717   /* non blocking check on helper logging pipe */
 718   struct pollfd fds = {
 719       .fd = phc->pipes[0], .events = POLLIN,
 720   };
 721   int data_avail = poll(&fds, 1, 0);
 722   DEBUG("%s:dpdk_helper_check_pipe: poll data_avail=%d", phc->shm_name,
 723         data_avail);
 724   if (data_avail < 0) {
 725     if (errno != EINTR || errno != EAGAIN) {
 726       char errbuf[ERR_BUF_SIZE];
 727       ERROR("%s: poll(2) failed: %s", phc->shm_name,
 728             sstrerror(errno, errbuf, sizeof(errbuf)));
 729     }
 730   }
 731   while (data_avail) {
 732     int nbytes = read(phc->pipes[0], buf, (sizeof(buf) - 1));
 733     DEBUG("%s:dpdk_helper_check_pipe: read nbytes=%d", phc->shm_name, nbytes);
 734     if (nbytes <= 0)
 735       break;
 736     buf[nbytes] = '\0';
 737     sstrncpy(out, buf, sizeof(out));
 738     DEBUG("%s: helper process:\n%s", phc->shm_name, out);
 739   }
 740 }
 741
 742 int dpdk_helper_command(dpdk_helper_ctx_t *phc, enum DPDK_CMD cmd, int *result,
 743                         cdtime_t cmd_wait_time) {
 744   if (phc == NULL) {
 745     ERROR("Invalid argument(phc)");
 746     return -EINVAL;
 747   }
 748
 749   DEBUG("%s:%s:%d pid=%lu, cmd=%d", phc->shm_name, __FUNCTION__, __LINE__,
 750         (long)getpid(), cmd);
 751
 752   phc->cmd_wait_time = cmd_wait_time;
 753
 754   int ret = dpdk_helper_status_check(phc);
 755
 756   dpdk_helper_check_pipe(phc);
 757
 758   if (ret != 0) {
 759     return ret;
 760   }
 761
 762   DEBUG("%s: DPDK command execute (cmd=%d)", phc->shm_name, cmd);
 763
 764   phc->cmd_result = 0;
 765   phc->cmd = cmd;
 766
 767   /* kick helper to process command */
 768   int err = sem_post(&phc->sema_cmd_start);
 769   if (err) {
 770     char errbuf[ERR_BUF_SIZE];
 771     ERROR("dpdk_helper_worker: error posting sema_cmd_start semaphore (%s)",
 772           sstrerror(errno, errbuf, sizeof(errbuf)));
 773   }
 774
 775 #if COLLECT_DEBUG
 776   int val = 0;
 777   if (sem_getvalue(&phc->sema_cmd_start, &val) == 0)
 778     DEBUG("%s:dpdk_helper_command: post sema_cmd_start (value=%d)",
 779           phc->shm_name, val);
 780 #endif
 781
 782   if (phc->cmd != DPDK_CMD_QUIT) {
 783
 784     /* wait for helper to complete processing */
 785     struct timespec ts;
 786     cdtime_t now = cdtime();
 787
 788     if (phc->status != DPDK_HELPER_ALIVE_SENDING_EVENTS) {
 789       cmd_wait_time = MS_TO_CDTIME_T(DPDK_CDM_DEFAULT_TIMEOUT);
 790     }
 791
 792     ts = CDTIME_T_TO_TIMESPEC(now + cmd_wait_time);
 793     ret = sem_timedwait(&phc->sema_cmd_complete, &ts);
 794     if (ret == -1 && errno == ETIMEDOUT) {
 795       DPDK_HELPER_TRACE(phc->shm_name);
 796       DEBUG("%s:sema_cmd_start: timeout in collectd thread: is a DPDK Primary "
 797             "running?",
 798             phc->shm_name);
 799       return -ETIMEDOUT;
 800     }
 801
 802 #if COLLECT_DEBUG
 803     val = 0;
 804     if (sem_getvalue(&phc->sema_cmd_complete, &val) == 0)
 805       DEBUG("%s:dpdk_helper_command: wait sema_cmd_complete (value=%d)",
 806             phc->shm_name, val);
 807 #endif
 808
 809     if (result) {
 810       *result = phc->cmd_result;
 811     }
 812   }
 813
 814   dpdk_helper_check_pipe(phc);
 815
 816   DEBUG("%s: DPDK command complete (cmd=%d, result=%d)", phc->shm_name,
 817         phc->cmd, phc->cmd_result);
 818
 819   return 0;
 820 }
 821
 822 uint64_t strtoull_safe(const char *str, int *err) {
 823   uint64_t val = 0;
 824   char *endptr;
 825   int res = 0;
 826
 827   val = strtoull(str, &endptr, 16);
 828   if (*endptr) {
 829     ERROR("%s Failed to parse the value %s, endptr=%c", __FUNCTION__, str,
 830           *endptr);
 831     res = -EINVAL;
 832   }
 833   if (err != NULL)
 834     *err = res;
 835   return val;
 836 }
 837
 838 uint128_t str_to_uint128(const char *str, int len) {
 839   uint128_t lcore_mask;
 840   int err = 0;
 841
 842   memset(&lcore_mask, 0, sizeof(lcore_mask));
 843
 844   if (len <= 2 || strncmp(str, "0x", 2) != 0) {
 845     ERROR("%s Value %s should be represened in hexadecimal format",
 846           __FUNCTION__, str);
 847     return lcore_mask;
 848   }
 849   /* If str is <= 64 bit long ('0x' + 16 chars = 18 chars) then
 850    * conversion is straightforward. Otherwise str is splitted into 64b long
 851    * blocks */
 852   if (len <= 18) {
 853     lcore_mask.low = strtoull_safe(str, &err);
 854     if (err)
 855       return lcore_mask;
 856   } else {
 857     char low_str[DATA_MAX_NAME_LEN];
 858     char high_str[DATA_MAX_NAME_LEN];
 859
 860     memset(high_str, 0, sizeof(high_str));
 861     memset(low_str, 0, sizeof(low_str));
 862
 863     strncpy(high_str, str, len - 16);
 864     strncpy(low_str, str + len - 16, 16);
 865
 866     lcore_mask.low = strtoull_safe(low_str, &err);
 867     if (err)
 868       return lcore_mask;
 869
 870     lcore_mask.high = strtoull_safe(high_str, &err);
 871     if (err) {
 872       lcore_mask.low = 0;
 873       return lcore_mask;
 874     }
 875   }
 876   return lcore_mask;
 877 }
 878
 879 uint8_t dpdk_helper_eth_dev_count() {
 880   uint8_t ports = rte_eth_dev_count();
 881   if (ports == 0) {
 882     ERROR(
 883         "%s:%d: No DPDK ports available. Check bound devices to DPDK driver.\n",
 884         __FUNCTION__, __LINE__);
 885     return ports;
 886   }
 887
 888   if (ports > RTE_MAX_ETHPORTS) {
 889     ERROR("%s:%d: Number of DPDK ports (%u) is greater than "
 890           "RTE_MAX_ETHPORTS=%d. Ignoring extra ports\n",
 891           __FUNCTION__, __LINE__, ports, RTE_MAX_ETHPORTS);
 892     ports = RTE_MAX_ETHPORTS;
 893   }
 894
 895   return ports;
 896 }