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