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