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