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   int fd = shm_open(name, O_CREAT | O_TRUNC | O_RDWR, 0666);
 219   if (fd < 0) {
 220     WARNING("dpdk_shm_init: Failed to open %s as SHM:%s", name, STRERRNO);
 221     *map = NULL;
 222     return -1;
 223   }
 224
 225   int ret = ftruncate(fd, size);
 226   if (ret != 0) {
 227     WARNING("dpdk_shm_init: Failed to resize SHM:%s", STRERRNO);
 228     close(fd);
 229     *map = NULL;
 230     dpdk_shm_cleanup(name, size, NULL);
 231     return -1;
 232   }
 233
 234   *map = mmap(0, size, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
 235   if (*map == MAP_FAILED) {
 236     WARNING("dpdk_shm_init:Failed to mmap SHM:%s", STRERRNO);
 237     close(fd);
 238     *map = NULL;
 239     dpdk_shm_cleanup(name, size, NULL);
 240     return -1;
 241   }
 242   /* File descriptor no longer needed for shared memory operations */
 243   close(fd);
 244   memset(*map, 0, size);
 245
 246   return 0;
 247 }
 248
 249 static void dpdk_shm_cleanup(const char *name, size_t size, void *map) {
 250   DPDK_HELPER_TRACE(name);
 251
 252   /*
 253    * Call shm_unlink first, as 'name' might be no longer accessible after munmap
 254    */
 255   if (shm_unlink(name))
 256     ERROR("shm_unlink failure %s", STRERRNO);
 257
 258   if (map != NULL) {
 259     if (munmap(map, size))
 260       ERROR("munmap failure %s", STRERRNO);
 261   }
 262 }
 263
 264 void *dpdk_helper_priv_get(dpdk_helper_ctx_t *phc) {
 265   if (phc)
 266     return phc->priv_data;
 267
 268   return NULL;
 269 }
 270
 271 int dpdk_helper_data_size_get(dpdk_helper_ctx_t *phc) {
 272   if (phc == NULL) {
 273     DPDK_CHILD_LOG("Invalid argument(phc)\n");
 274     return -EINVAL;
 275   }
 276
 277   return phc->shm_size - sizeof(dpdk_helper_ctx_t);
 278 }
 279
 280 int dpdk_helper_init(const char *name, size_t data_size,
 281                      dpdk_helper_ctx_t **pphc) {
 282   dpdk_helper_ctx_t *phc = NULL;
 283   size_t shm_size = sizeof(dpdk_helper_ctx_t) + data_size;
 284
 285   if (pphc == NULL) {
 286     ERROR("%s:Invalid argument(pphc)", __FUNCTION__);
 287     return -EINVAL;
 288   }
 289
 290   if (name == NULL) {
 291     ERROR("%s:Invalid argument(name)", __FUNCTION__);
 292     return -EINVAL;
 293   }
 294
 295   DPDK_HELPER_TRACE(name);
 296
 297   /* Allocate dpdk_helper_ctx_t and
 298   * initialize a POSIX SHared Memory (SHM) object.
 299   */
 300   int err = dpdk_shm_init(name, shm_size, (void **)&phc);
 301   if (err != 0) {
 302     return -errno;
 303   }
 304
 305   err = sem_init(&phc->sema_cmd_start, 1, 0);
 306   if (err != 0) {
 307     ERROR("sema_cmd_start semaphore init failed: %s", STRERRNO);
 308     int errno_m = errno;
 309     dpdk_shm_cleanup(name, shm_size, (void *)phc);
 310     return -errno_m;
 311   }
 312
 313   err = sem_init(&phc->sema_cmd_complete, 1, 0);
 314   if (err != 0) {
 315     ERROR("sema_cmd_complete semaphore init failed: %s", STRERRNO);
 316     sem_destroy(&phc->sema_cmd_start);
 317     int errno_m = errno;
 318     dpdk_shm_cleanup(name, shm_size, (void *)phc);
 319     return -errno_m;
 320   }
 321
 322   phc->shm_size = shm_size;
 323   sstrncpy(phc->shm_name, name, sizeof(phc->shm_name));
 324
 325   dpdk_helper_config_default(phc);
 326
 327   *pphc = phc;
 328
 329   return 0;
 330 }
 331
 332 void dpdk_helper_shutdown(dpdk_helper_ctx_t *phc) {
 333   if (phc == NULL)
 334     return;
 335
 336   DPDK_HELPER_TRACE(phc->shm_name);
 337
 338   close(phc->pipes[1]);
 339
 340   if (phc->status != DPDK_HELPER_NOT_INITIALIZED) {
 341     dpdk_helper_exit_command(phc, DPDK_HELPER_GRACEFUL_QUIT);
 342   }
 343
 344   sem_destroy(&phc->sema_cmd_start);
 345   sem_destroy(&phc->sema_cmd_complete);
 346   dpdk_shm_cleanup(phc->shm_name, phc->shm_size, (void *)phc);
 347 }
 348
 349 static int dpdk_helper_spawn(dpdk_helper_ctx_t *phc) {
 350   if (phc == NULL) {
 351     ERROR("Invalid argument(phc)");
 352     return -EINVAL;
 353   }
 354
 355   DPDK_HELPER_TRACE(phc->shm_name);
 356
 357   phc->eal_initialized = 0;
 358   phc->cmd_wait_time = MS_TO_CDTIME_T(DPDK_CDM_DEFAULT_TIMEOUT);
 359
 360   /*
 361    * Create a pipe for helper stdout back to collectd. This is necessary for
 362    * logging EAL failures, as rte_eal_init() calls rte_panic().
 363    */
 364   if (phc->pipes[1]) {
 365     DEBUG("dpdk_helper_spawn: collectd closing helper pipe %d", phc->pipes[1]);
 366   } else {
 367     DEBUG("dpdk_helper_spawn: collectd helper pipe %d, not closing",
 368           phc->pipes[1]);
 369   }
 370
 371   if (pipe(phc->pipes) != 0) {
 372     DEBUG("dpdk_helper_spawn: Could not create helper pipe: %s", STRERRNO);
 373     return -1;
 374   }
 375
 376   int pipe0_flags = fcntl(phc->pipes[0], F_GETFL, 0);
 377   int pipe1_flags = fcntl(phc->pipes[1], F_GETFL, 0);
 378   if (pipe0_flags == -1 || pipe1_flags == -1) {
 379     WARNING("dpdk_helper_spawn: error setting up pipe flags: %s", STRERRNO);
 380   }
 381   int pipe0_err = fcntl(phc->pipes[0], F_SETFL, pipe1_flags | O_NONBLOCK);
 382   int pipe1_err = fcntl(phc->pipes[1], F_SETFL, pipe0_flags | O_NONBLOCK);
 383   if (pipe0_err == -1 || pipe1_err == -1) {
 384     WARNING("dpdk_helper_spawn: error setting up pipes: %s", STRERRNO);
 385   }
 386
 387   pid_t pid = fork();
 388   if (pid > 0) {
 389     phc->pid = pid;
 390     close(phc->pipes[1]);
 391     DEBUG("%s:dpdk_helper_spawn: helper pid %lu", phc->shm_name,
 392           (long)phc->pid);
 393   } else if (pid == 0) {
 394     /* Replace stdout with a pipe to collectd. */
 395     close(phc->pipes[0]);
 396     close(STDOUT_FILENO);
 397     dup2(phc->pipes[1], STDOUT_FILENO);
 398     DPDK_CHILD_TRACE(phc->shm_name);
 399     dpdk_helper_worker(phc);
 400     exit(0);
 401   } else {
 402     ERROR("dpdk_helper_start: Failed to fork helper process: %s", STRERRNO);
 403     return -1;
 404   }
 405
 406   return 0;
 407 }
 408
 409 static int dpdk_helper_exit(dpdk_helper_ctx_t *phc,
 410                             enum DPDK_HELPER_STATUS status) {
 411   DPDK_CHILD_LOG("%s:%s:%d %s\n", phc->shm_name, __FUNCTION__, __LINE__,
 412                  dpdk_helper_status_str(status));
 413
 414   close(phc->pipes[1]);
 415
 416   phc->status = status;
 417
 418   exit(0);
 419
 420   return 0;
 421 }
 422
 423 static int dpdk_helper_exit_command(dpdk_helper_ctx_t *phc,
 424                                     enum DPDK_HELPER_STATUS status) {
 425   DPDK_HELPER_TRACE(phc->shm_name);
 426
 427   close(phc->pipes[1]);
 428
 429   if (phc->status == DPDK_HELPER_ALIVE_SENDING_EVENTS) {
 430     phc->status = status;
 431     DEBUG("%s:%s:%d %s", phc->shm_name, __FUNCTION__, __LINE__,
 432           dpdk_helper_status_str(status));
 433
 434     int ret = dpdk_helper_command(phc, DPDK_CMD_QUIT, NULL, 0);
 435     if (ret != 0) {
 436       DEBUG("%s:%s:%d kill helper (pid=%lu)", phc->shm_name, __FUNCTION__,
 437             __LINE__, (long)phc->pid);
 438
 439       int err = kill(phc->pid, SIGKILL);
 440       if (err) {
 441         ERROR("%s error sending kill to helper: %s", __FUNCTION__, STRERRNO);
 442       }
 443     }
 444   } else {
 445
 446     DEBUG("%s:%s:%d kill helper (pid=%lu)", phc->shm_name, __FUNCTION__,
 447           __LINE__, (long)phc->pid);
 448
 449     int err = kill(phc->pid, SIGKILL);
 450     if (err) {
 451       ERROR("%s error sending kill to helper: %s", __FUNCTION__, STRERRNO);
 452     }
 453   }
 454
 455   return 0;
 456 }
 457
 458 static int dpdk_helper_eal_init(dpdk_helper_ctx_t *phc) {
 459   phc->status = DPDK_HELPER_INITIALIZING_EAL;
 460   DPDK_CHILD_LOG("%s:%s:%d DPDK_HELPER_INITIALIZING_EAL (start)\n",
 461                  phc->shm_name, __FUNCTION__, __LINE__);
 462
 463   char *argp[DPDK_EAL_ARGC * 2 + 1];
 464   int argc = 0;
 465
 466   /* EAL config must be initialized */
 467   assert(phc->eal_config.coremask[0] != 0);
 468   assert(phc->eal_config.memory_channels[0] != 0);
 469   assert(phc->eal_config.file_prefix[0] != 0);
 470
 471   argp[argc++] = "collectd-dpdk";
 472
 473   argp[argc++] = "-c";
 474   argp[argc++] = phc->eal_config.coremask;
 475
 476   argp[argc++] = "-n";
 477   argp[argc++] = phc->eal_config.memory_channels;
 478
 479   if (strcasecmp(phc->eal_config.socket_memory, "") != 0) {
 480     argp[argc++] = "--socket-mem";
 481     argp[argc++] = phc->eal_config.socket_memory;
 482   }
 483
 484   if (strcasecmp(phc->eal_config.file_prefix, DPDK_DEFAULT_RTE_CONFIG) != 0) {
 485     argp[argc++] = "--file-prefix";
 486     argp[argc++] = phc->eal_config.file_prefix;
 487   }
 488
 489   argp[argc++] = "--proc-type";
 490   argp[argc++] = "secondary";
 491
 492   if (strcasecmp(phc->eal_config.log_level, "") != 0) {
 493     argp[argc++] = "--log-level";
 494     argp[argc++] = phc->eal_config.log_level;
 495   }
 496   if (strcasecmp(phc->eal_config.rte_driver_lib_path, "") != 0) {
 497     argp[argc++] = "-d";
 498     argp[argc++] = phc->eal_config.rte_driver_lib_path;
 499   }
 500
 501   assert(argc <= (DPDK_EAL_ARGC * 2 + 1));
 502
 503   int ret = rte_eal_init(argc, argp);
 504
 505   if (ret < 0) {
 506
 507     phc->eal_initialized = 0;
 508
 509     DPDK_CHILD_LOG("dpdk_helper_eal_init: ERROR initializing EAL ret=%d\n",
 510                    ret);
 511
 512     printf("dpdk_helper_eal_init: EAL arguments: ");
 513     for (int i = 0; i < argc; i++) {
 514       printf("%s ", argp[i]);
 515     }
 516     printf("\n");
 517
 518     return ret;
 519   }
 520
 521   phc->eal_initialized = 1;
 522
 523   DPDK_CHILD_LOG("%s:%s:%d DPDK_HELPER_INITIALIZING_EAL (done)\n",
 524                  phc->shm_name, __FUNCTION__, __LINE__);
 525
 526   return 0;
 527 }
 528
 529 static int dpdk_helper_cmd_wait(dpdk_helper_ctx_t *phc, pid_t ppid) {
 530   DPDK_CHILD_TRACE(phc->shm_name);
 531
 532   struct timespec ts;
 533   cdtime_t now = cdtime();
 534   cdtime_t cmd_wait_time = MS_TO_CDTIME_T(1500) + phc->cmd_wait_time * 2;
 535   ts = CDTIME_T_TO_TIMESPEC(now + cmd_wait_time);
 536
 537   int ret = sem_timedwait(&phc->sema_cmd_start, &ts);
 538   DPDK_CHILD_LOG("%s:%s:%d pid=%lu got sema_cmd_start (ret=%d, errno=%d)\n",
 539                  phc->shm_name, __FUNCTION__, __LINE__, (long)getpid(), ret,
 540                  errno);
 541
 542   if (phc->cmd == DPDK_CMD_QUIT) {
 543     DPDK_CHILD_LOG("%s:%s:%d pid=%lu exiting\n", phc->shm_name, __FUNCTION__,
 544                    __LINE__, (long)getpid());
 545     exit(0);
 546   } else if (ret == -1 && errno == ETIMEDOUT) {
 547     if (phc->status == DPDK_HELPER_ALIVE_SENDING_EVENTS) {
 548       DPDK_CHILD_LOG("%s:dpdk_helper_cmd_wait: sem timedwait()"
 549                      " timeout, did collectd terminate?\n",
 550                      phc->shm_name);
 551       dpdk_helper_exit(phc, DPDK_HELPER_GRACEFUL_QUIT);
 552     }
 553   }
 554 #if COLLECT_DEBUG
 555   int val = 0;
 556   if (sem_getvalue(&phc->sema_cmd_start, &val) == 0)
 557     DPDK_CHILD_LOG("%s:%s:%d pid=%lu wait sema_cmd_start (value=%d)\n",
 558                    phc->shm_name, __FUNCTION__, __LINE__, (long)getpid(), val);
 559 #endif
 560
 561   /* Parent PID change means collectd died so quit the helper process. */
 562   if (ppid != getppid()) {
 563     DPDK_CHILD_LOG("dpdk_helper_cmd_wait: parent PID changed, quitting.\n");
 564     dpdk_helper_exit(phc, DPDK_HELPER_GRACEFUL_QUIT);
 565   }
 566
 567   /* Checking for DPDK primary process. */
 568   if (!rte_eal_primary_proc_alive(phc->eal_config.file_prefix)) {
 569     if (phc->eal_initialized) {
 570       DPDK_CHILD_LOG(
 571           "%s:dpdk_helper_cmd_wait: no primary alive but EAL initialized:"
 572           " quitting.\n",
 573           phc->shm_name);
 574       dpdk_helper_exit(phc, DPDK_HELPER_NOT_INITIALIZED);
 575     }
 576
 577     phc->status = DPDK_HELPER_WAITING_ON_PRIMARY;
 578     DPDK_CHILD_LOG("%s:%s:%d DPDK_HELPER_WAITING_ON_PRIMARY\n", phc->shm_name,
 579                    __FUNCTION__, __LINE__);
 580
 581     return -1;
 582   }
 583
 584   if (!phc->eal_initialized) {
 585     int ret = dpdk_helper_eal_init(phc);
 586     if (ret != 0) {
 587       DPDK_CHILD_LOG("Error initializing EAL\n");
 588       dpdk_helper_exit(phc, DPDK_HELPER_NOT_INITIALIZED);
 589     }
 590     phc->status = DPDK_HELPER_ALIVE_SENDING_EVENTS;
 591     DPDK_CHILD_LOG("%s:%s:%d DPDK_HELPER_ALIVE_SENDING_EVENTS\n", phc->shm_name,
 592                    __FUNCTION__, __LINE__);
 593     return -1;
 594   }
 595
 596   return 0;
 597 }
 598
 599 static int dpdk_helper_worker(dpdk_helper_ctx_t *phc) {
 600   DPDK_CHILD_TRACE(phc->shm_name);
 601
 602   pid_t ppid = getppid();
 603
 604   while (1) {
 605     if (dpdk_helper_cmd_wait(phc, ppid) == 0) {
 606       DPDK_CHILD_LOG("%s:%s:%d DPDK command handle (cmd=%d, pid=%lu)\n",
 607                      phc->shm_name, __FUNCTION__, __LINE__, phc->cmd,
 608                      (long)getpid());
 609       phc->cmd_result = dpdk_helper_command_handler(phc, phc->cmd);
 610     } else {
 611       phc->cmd_result = -1;
 612     }
 613
 614     /* now kick collectd to get results */
 615     int err = sem_post(&phc->sema_cmd_complete);
 616     DPDK_CHILD_LOG("%s:%s:%d post sema_cmd_complete (pid=%lu)\n", phc->shm_name,
 617                    __FUNCTION__, __LINE__, (long)getpid());
 618     if (err) {
 619       DPDK_CHILD_LOG("dpdk_helper_worker: error posting sema_cmd_complete "
 620                      "semaphore (%s)\n",
 621                      STRERRNO);
 622     }
 623
 624 #if COLLECT_DEBUG
 625     int val = 0;
 626     if (sem_getvalue(&phc->sema_cmd_complete, &val) == 0)
 627       DPDK_CHILD_LOG("%s:%s:%d pid=%lu sema_cmd_complete (value=%d)\n",
 628                      phc->shm_name, __FUNCTION__, __LINE__, (long)getpid(),
 629                      val);
 630 #endif
 631
 632   } /* while(1) */
 633
 634   return 0;
 635 }
 636
 637 static const char *dpdk_helper_status_str(enum DPDK_HELPER_STATUS status) {
 638   switch (status) {
 639   case DPDK_HELPER_ALIVE_SENDING_EVENTS:
 640     return "DPDK_HELPER_ALIVE_SENDING_EVENTS";
 641   case DPDK_HELPER_WAITING_ON_PRIMARY:
 642     return "DPDK_HELPER_WAITING_ON_PRIMARY";
 643   case DPDK_HELPER_INITIALIZING:
 644     return "DPDK_HELPER_INITIALIZING";
 645   case DPDK_HELPER_INITIALIZING_EAL:
 646     return "DPDK_HELPER_INITIALIZING_EAL";
 647   case DPDK_HELPER_GRACEFUL_QUIT:
 648     return "DPDK_HELPER_GRACEFUL_QUIT";
 649   case DPDK_HELPER_NOT_INITIALIZED:
 650     return "DPDK_HELPER_NOT_INITIALIZED";
 651   default:
 652     return "UNKNOWN";
 653   }
 654 }
 655
 656 static int dpdk_helper_status_check(dpdk_helper_ctx_t *phc) {
 657   DEBUG("%s:%s:%d pid=%u %s", phc->shm_name, __FUNCTION__, __LINE__, getpid(),
 658         dpdk_helper_status_str(phc->status));
 659
 660   if (phc->status == DPDK_HELPER_GRACEFUL_QUIT) {
 661     return 0;
 662   } else if (phc->status == DPDK_HELPER_NOT_INITIALIZED) {
 663     phc->status = DPDK_HELPER_INITIALIZING;
 664     DEBUG("%s:%s:%d DPDK_HELPER_INITIALIZING", phc->shm_name, __FUNCTION__,
 665           __LINE__);
 666     int err = dpdk_helper_spawn(phc);
 667     if (err) {
 668       ERROR("dpdkstat: error spawning helper %s", STRERRNO);
 669     }
 670     return -1;
 671   }
 672
 673   pid_t ws = waitpid(phc->pid, NULL, WNOHANG);
 674   if (ws != 0) {
 675     phc->status = DPDK_HELPER_INITIALIZING;
 676     DEBUG("%s:%s:%d DPDK_HELPER_INITIALIZING", phc->shm_name, __FUNCTION__,
 677           __LINE__);
 678     int err = dpdk_helper_spawn(phc);
 679     if (err) {
 680       ERROR("dpdkstat: error spawning helper %s", STRERRNO);
 681     }
 682     return -1;
 683   }
 684
 685   if (phc->status == DPDK_HELPER_INITIALIZING_EAL) {
 686     return -1;
 687   }
 688
 689   return 0;
 690 }
 691
 692 static void dpdk_helper_check_pipe(dpdk_helper_ctx_t *phc) {
 693   char buf[DPDK_MAX_BUFFER_SIZE];
 694   char out[DPDK_MAX_BUFFER_SIZE];
 695
 696   /* non blocking check on helper logging pipe */
 697   struct pollfd fds = {
 698       .fd = phc->pipes[0], .events = POLLIN,
 699   };
 700   int data_avail = poll(&fds, 1, 0);
 701   DEBUG("%s:dpdk_helper_check_pipe: poll data_avail=%d", phc->shm_name,
 702         data_avail);
 703   if (data_avail < 0) {
 704     if (errno != EINTR || errno != EAGAIN) {
 705       ERROR("%s: poll(2) failed: %s", phc->shm_name, STRERRNO);
 706     }
 707   }
 708   while (data_avail) {
 709     int nbytes = read(phc->pipes[0], buf, (sizeof(buf) - 1));
 710     DEBUG("%s:dpdk_helper_check_pipe: read nbytes=%d", phc->shm_name, nbytes);
 711     if (nbytes <= 0)
 712       break;
 713     buf[nbytes] = '\0';
 714     sstrncpy(out, buf, sizeof(out));
 715     DEBUG("%s: helper process:\n%s", phc->shm_name, out);
 716   }
 717 }
 718
 719 int dpdk_helper_command(dpdk_helper_ctx_t *phc, enum DPDK_CMD cmd, int *result,
 720                         cdtime_t cmd_wait_time) {
 721   if (phc == NULL) {
 722     ERROR("Invalid argument(phc)");
 723     return -EINVAL;
 724   }
 725
 726   DEBUG("%s:%s:%d pid=%lu, cmd=%d", phc->shm_name, __FUNCTION__, __LINE__,
 727         (long)getpid(), cmd);
 728
 729   phc->cmd_wait_time = cmd_wait_time;
 730
 731   int ret = dpdk_helper_status_check(phc);
 732
 733   dpdk_helper_check_pipe(phc);
 734
 735   if (ret != 0) {
 736     return ret;
 737   }
 738
 739   DEBUG("%s: DPDK command execute (cmd=%d)", phc->shm_name, cmd);
 740
 741   phc->cmd_result = 0;
 742   phc->cmd = cmd;
 743
 744   /* kick helper to process command */
 745   int err = sem_post(&phc->sema_cmd_start);
 746   if (err) {
 747     ERROR("dpdk_helper_worker: error posting sema_cmd_start semaphore (%s)",
 748           STRERRNO);
 749   }
 750
 751 #if COLLECT_DEBUG
 752   int val = 0;
 753   if (sem_getvalue(&phc->sema_cmd_start, &val) == 0)
 754     DEBUG("%s:dpdk_helper_command: post sema_cmd_start (value=%d)",
 755           phc->shm_name, val);
 756 #endif
 757
 758   if (phc->cmd != DPDK_CMD_QUIT) {
 759
 760     /* wait for helper to complete processing */
 761     struct timespec ts;
 762     cdtime_t now = cdtime();
 763
 764     if (phc->status != DPDK_HELPER_ALIVE_SENDING_EVENTS) {
 765       cmd_wait_time = MS_TO_CDTIME_T(DPDK_CDM_DEFAULT_TIMEOUT);
 766     }
 767
 768     ts = CDTIME_T_TO_TIMESPEC(now + cmd_wait_time);
 769     ret = sem_timedwait(&phc->sema_cmd_complete, &ts);
 770     if (ret == -1 && errno == ETIMEDOUT) {
 771       DPDK_HELPER_TRACE(phc->shm_name);
 772       DEBUG("%s:sema_cmd_start: timeout in collectd thread: is a DPDK Primary "
 773             "running?",
 774             phc->shm_name);
 775       return -ETIMEDOUT;
 776     }
 777
 778 #if COLLECT_DEBUG
 779     val = 0;
 780     if (sem_getvalue(&phc->sema_cmd_complete, &val) == 0)
 781       DEBUG("%s:dpdk_helper_command: wait sema_cmd_complete (value=%d)",
 782             phc->shm_name, val);
 783 #endif
 784
 785     if (result) {
 786       *result = phc->cmd_result;
 787     }
 788   }
 789
 790   dpdk_helper_check_pipe(phc);
 791
 792   DEBUG("%s: DPDK command complete (cmd=%d, result=%d)", phc->shm_name,
 793         phc->cmd, phc->cmd_result);
 794
 795   return 0;
 796 }
 797
 798 uint64_t strtoull_safe(const char *str, int *err) {
 799   uint64_t val = 0;
 800   char *endptr;
 801   int res = 0;
 802
 803   val = strtoull(str, &endptr, 16);
 804   if (*endptr) {
 805     ERROR("%s Failed to parse the value %s, endptr=%c", __FUNCTION__, str,
 806           *endptr);
 807     res = -EINVAL;
 808   }
 809   if (err != NULL)
 810     *err = res;
 811   return val;
 812 }
 813
 814 uint128_t str_to_uint128(const char *str, int len) {
 815   uint128_t lcore_mask;
 816   int err = 0;
 817
 818   memset(&lcore_mask, 0, sizeof(lcore_mask));
 819
 820   if (len <= 2 || strncmp(str, "0x", 2) != 0) {
 821     ERROR("%s Value %s should be represened in hexadecimal format",
 822           __FUNCTION__, str);
 823     return lcore_mask;
 824   }
 825   /* If str is <= 64 bit long ('0x' + 16 chars = 18 chars) then
 826    * conversion is straightforward. Otherwise str is splitted into 64b long
 827    * blocks */
 828   if (len <= 18) {
 829     lcore_mask.low = strtoull_safe(str, &err);
 830     if (err)
 831       return lcore_mask;
 832   } else {
 833     char low_str[DATA_MAX_NAME_LEN];
 834     char high_str[DATA_MAX_NAME_LEN * 2];
 835
 836     memset(high_str, 0, sizeof(high_str));
 837     memset(low_str, 0, sizeof(low_str));
 838
 839     strncpy(high_str, str, len - 16);
 840     strncpy(low_str, str + len - 16, 16);
 841
 842     lcore_mask.low = strtoull_safe(low_str, &err);
 843     if (err)
 844       return lcore_mask;
 845
 846     lcore_mask.high = strtoull_safe(high_str, &err);
 847     if (err) {
 848       lcore_mask.low = 0;
 849       return lcore_mask;
 850     }
 851   }
 852   return lcore_mask;
 853 }
 854
 855 uint8_t dpdk_helper_eth_dev_count(void) {
 856 #if RTE_VERSION < RTE_VERSION_NUM(18, 05, 0, 0)
 857   uint8_t ports = rte_eth_dev_count();
 858 #else
 859   uint8_t ports = rte_eth_dev_count_avail();
 860 #endif
 861   if (ports == 0) {
 862     ERROR(
 863         "%s:%d: No DPDK ports available. Check bound devices to DPDK driver.\n",
 864         __FUNCTION__, __LINE__);
 865     return ports;
 866   }
 867
 868   if (ports > RTE_MAX_ETHPORTS) {
 869     ERROR("%s:%d: Number of DPDK ports (%u) is greater than "
 870           "RTE_MAX_ETHPORTS=%d. Ignoring extra ports\n",
 871           __FUNCTION__, __LINE__, ports, RTE_MAX_ETHPORTS);
 872     ports = RTE_MAX_ETHPORTS;
 873   }
 874
 875   return ports;
 876 }