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 5
  48 #define DPDK_MAX_BUFFER_SIZE (4096 * 4)
  49 #define DPDK_CDM_DEFAULT_TIMEOUT 10000
  50
  51 enum DPDK_HELPER_STATUS {
  52   DPDK_HELPER_NOT_INITIALIZED = 0,
  53   DPDK_HELPER_INITIALIZING,
  54   DPDK_HELPER_WAITING_ON_PRIMARY,
  55   DPDK_HELPER_INITIALIZING_EAL,
  56   DPDK_HELPER_ALIVE_SENDING_EVENTS,
  57   DPDK_HELPER_GRACEFUL_QUIT,
  58 };
  59
  60 #define DPDK_HELPER_TRACE(_name)                                               \
  61   DEBUG("%s:%s:%d pid=%ld", _name, __FUNCTION__, __LINE__, (long)getpid())
  62
  63 struct dpdk_helper_ctx_s {
  64
  65   dpdk_eal_config_t eal_config;
  66   int eal_initialized;
  67
  68   size_t shm_size;
  69   const char *shm_name;
  70
  71   sem_t sema_cmd_start;
  72   sem_t sema_cmd_complete;
  73   cdtime_t cmd_wait_time;
  74
  75   pid_t pid;
  76   int pipes[2];
  77   int status;
  78
  79   int cmd;
  80   int cmd_result;
  81
  82   char priv_data[];
  83 };
  84
  85 static int dpdk_shm_init(const char *name, size_t size, void **map);
  86 static void dpdk_shm_cleanup(const char *name, size_t size, void *map);
  87
  88 static int dpdk_helper_spawn(dpdk_helper_ctx_t *phc);
  89 static int dpdk_helper_worker(dpdk_helper_ctx_t *phc);
  90 static int dpdk_helper_eal_init(dpdk_helper_ctx_t *phc);
  91 static int dpdk_helper_cmd_wait(dpdk_helper_ctx_t *phc, pid_t ppid);
  92 static int dpdk_helper_exit_command(dpdk_helper_ctx_t *phc,
  93                                     enum DPDK_HELPER_STATUS status);
  94 static int dpdk_helper_exit(dpdk_helper_ctx_t *phc,
  95                             enum DPDK_HELPER_STATUS status);
  96 static int dpdk_helper_status_check(dpdk_helper_ctx_t *phc);
  97 static void dpdk_helper_config_default(dpdk_helper_ctx_t *phc);
  98 static const char *dpdk_helper_status_str(enum DPDK_HELPER_STATUS status);
  99
 100 static void dpdk_helper_config_default(dpdk_helper_ctx_t *phc) {
 101   if (phc == NULL)
 102     return;
 103
 104   DPDK_HELPER_TRACE(phc->shm_name);
 105
 106   ssnprintf(phc->eal_config.coremask, DATA_MAX_NAME_LEN, "%s", "0xf");
 107   ssnprintf(phc->eal_config.memory_channels, DATA_MAX_NAME_LEN, "%s", "1");
 108   ssnprintf(phc->eal_config.process_type, DATA_MAX_NAME_LEN, "%s", "secondary");
 109   ssnprintf(phc->eal_config.file_prefix, DATA_MAX_NAME_LEN, "%s",
 110             DPDK_DEFAULT_RTE_CONFIG);
 111 }
 112
 113 int dpdk_helper_eal_config_set(dpdk_helper_ctx_t *phc, dpdk_eal_config_t *ec) {
 114   if (phc == NULL) {
 115     ERROR("Invalid argument (phc)");
 116     return -EINVAL;
 117   }
 118
 119   DPDK_HELPER_TRACE(phc->shm_name);
 120
 121   if (ec == NULL) {
 122     ERROR("Invalid argument (ec)");
 123     return -EINVAL;
 124   }
 125
 126   memcpy(&phc->eal_config, ec, sizeof(phc->eal_config));
 127
 128   return 0;
 129 }
 130
 131 int dpdk_helper_eal_config_get(dpdk_helper_ctx_t *phc, dpdk_eal_config_t *ec) {
 132   if (phc == NULL) {
 133     ERROR("Invalid argument (phc)");
 134     return -EINVAL;
 135   }
 136
 137   DPDK_HELPER_TRACE(phc->shm_name);
 138
 139   if (ec == NULL) {
 140     ERROR("Invalid argument (ec)");
 141     return -EINVAL;
 142   }
 143
 144   memcpy(ec, &phc->eal_config, sizeof(*ec));
 145
 146   return 0;
 147 }
 148
 149 int dpdk_helper_eal_config_parse(dpdk_helper_ctx_t *phc, oconfig_item_t *ci) {
 150   DPDK_HELPER_TRACE(phc->shm_name);
 151
 152   if (phc == NULL) {
 153     ERROR("Invalid argument (phc)");
 154     return -EINVAL;
 155   }
 156
 157   if (ci == NULL) {
 158     ERROR("Invalid argument (ci)");
 159     return -EINVAL;
 160   }
 161
 162   int status = 0;
 163   for (int i = 0; i < ci->children_num; i++) {
 164     oconfig_item_t *child = ci->children + i;
 165     if (strcasecmp("Coremask", child->key) == 0) {
 166       status = cf_util_get_string_buffer(child, phc->eal_config.coremask,
 167                                          sizeof(phc->eal_config.coremask));
 168       DEBUG("dpdk_common: EAL:Coremask %s", phc->eal_config.coremask);
 169     } else if (strcasecmp("MemoryChannels", child->key) == 0) {
 170       status =
 171           cf_util_get_string_buffer(child, phc->eal_config.memory_channels,
 172                                     sizeof(phc->eal_config.memory_channels));
 173       DEBUG("dpdk_common: EAL:Memory Channels %s",
 174             phc->eal_config.memory_channels);
 175     } else if (strcasecmp("SocketMemory", child->key) == 0) {
 176       status = cf_util_get_string_buffer(child, phc->eal_config.socket_memory,
 177                                          sizeof(phc->eal_config.socket_memory));
 178       DEBUG("dpdk_common: EAL:Socket memory %s", phc->eal_config.socket_memory);
 179     } else if (strcasecmp("ProcessType", child->key) == 0) {
 180       status = cf_util_get_string_buffer(child, phc->eal_config.process_type,
 181                                          sizeof(phc->eal_config.process_type));
 182       DEBUG("dpdk_common: EAL:Process type %s", phc->eal_config.process_type);
 183     } else if ((strcasecmp("FilePrefix", child->key) == 0) &&
 184                (child->values[0].type == OCONFIG_TYPE_STRING)) {
 185       ssnprintf(phc->eal_config.file_prefix, DATA_MAX_NAME_LEN,
 186                 "/var/run/.%s_config", child->values[0].value.string);
 187       DEBUG("dpdk_common: EAL:File prefix %s", phc->eal_config.file_prefix);
 188     } else {
 189       ERROR("dpdk_common: Invalid '%s' configuration option", child->key);
 190       status = -EINVAL;
 191     }
 192
 193     if (status != 0) {
 194       ERROR("dpdk_common: Parsing EAL configuration failed");
 195       break;
 196     }
 197   }
 198
 199   return status;
 200 }
 201
 202 static int dpdk_shm_init(const char *name, size_t size, void **map) {
 203   DPDK_HELPER_TRACE(name);
 204
 205   char errbuf[ERR_BUF_SIZE];
 206
 207   int fd = shm_open(name, O_CREAT | O_TRUNC | O_RDWR, 0666);
 208   if (fd < 0) {
 209     WARNING("dpdk_shm_init: Failed to open %s as SHM:%s", name,
 210             sstrerror(errno, errbuf, sizeof(errbuf)));
 211     *map = NULL;
 212     return -1;
 213   }
 214
 215   int ret = ftruncate(fd, size);
 216   if (ret != 0) {
 217     WARNING("dpdk_shm_init: Failed to resize SHM:%s",
 218             sstrerror(errno, errbuf, sizeof(errbuf)));
 219     close(fd);
 220     *map = NULL;
 221     dpdk_shm_cleanup(name, size, NULL);
 222     return -1;
 223   }
 224
 225   *map = mmap(0, size, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
 226   if (*map == MAP_FAILED) {
 227     WARNING("dpdk_shm_init:Failed to mmap SHM:%s",
 228             sstrerror(errno, errbuf, sizeof(errbuf)));
 229     close(fd);
 230     *map = NULL;
 231     dpdk_shm_cleanup(name, size, NULL);
 232     return -1;
 233   }
 234   /* File descriptor no longer needed for shared memory operations */
 235   close(fd);
 236   memset(*map, 0, size);
 237
 238   return 0;
 239 }
 240
 241 static void dpdk_shm_cleanup(const char *name, size_t size, void *map) {
 242   DPDK_HELPER_TRACE(name);
 243   char errbuf[ERR_BUF_SIZE];
 244
 245   if (map != NULL) {
 246     if (munmap(map, size))
 247       ERROR("munmap failure %s", sstrerror(errno, errbuf, sizeof(errbuf)));
 248   }
 249
 250   if (shm_unlink(name))
 251     ERROR("shm_unlink failure %s", sstrerror(errno, errbuf, sizeof(errbuf)));
 252 }
 253
 254 void *dpdk_helper_priv_get(dpdk_helper_ctx_t *phc) {
 255   if (phc)
 256     return phc->priv_data;
 257
 258   return NULL;
 259 }
 260
 261 int dpdk_helper_data_size_get(dpdk_helper_ctx_t *phc) {
 262   if (phc == NULL) {
 263     DPDK_CHILD_LOG("Invalid argument(phc)\n");
 264     return -EINVAL;
 265   }
 266
 267   return (phc->shm_size - sizeof(dpdk_helper_ctx_t));
 268 }
 269
 270 int dpdk_helper_init(const char *name, size_t data_size,
 271                      dpdk_helper_ctx_t **pphc) {
 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   int 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",
 299           sstrerror(errno, errbuf, sizeof(errbuf)));
 300     int errno_m = errno;
 301     dpdk_shm_cleanup(name, shm_size, (void *)phc);
 302     return -errno_m;
 303   }
 304
 305   err = sem_init(&phc->sema_cmd_complete, 1, 0);
 306   if (err != 0) {
 307     ERROR("sema_cmd_complete semaphore init failed: %s",
 308           sstrerror(errno, errbuf, sizeof(errbuf)));
 309     sem_destroy(&phc->sema_cmd_start);
 310     int errno_m = errno;
 311     dpdk_shm_cleanup(name, shm_size, (void *)phc);
 312     return -errno_m;
 313   }
 314
 315   phc->shm_size = shm_size;
 316   phc->shm_name = name;
 317
 318   dpdk_helper_config_default(phc);
 319
 320   *pphc = phc;
 321
 322   return 0;
 323 }
 324
 325 int dpdk_helper_shutdown(dpdk_helper_ctx_t *phc) {
 326   if (phc == NULL) {
 327     ERROR("%s:Invalid argument(phc)", __FUNCTION__);
 328     return -EINVAL;
 329   }
 330
 331   DPDK_HELPER_TRACE(phc->shm_name);
 332
 333   close(phc->pipes[1]);
 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   /*
 359    * Create a pipe for helper stdout back to collectd. This is necessary for
 360    * logging EAL failures, as rte_eal_init() calls rte_panic().
 361    */
 362   if (phc->pipes[1]) {
 363     DEBUG("dpdk_helper_spawn: collectd closing helper pipe %d", phc->pipes[1]);
 364   } else {
 365     DEBUG("dpdk_helper_spawn: collectd helper pipe %d, not closing",
 366           phc->pipes[1]);
 367   }
 368
 369   if (pipe(phc->pipes) != 0) {
 370     DEBUG("dpdk_helper_spawn: Could not create helper pipe: %s",
 371           sstrerror(errno, errbuf, sizeof(errbuf)));
 372     return -1;
 373   }
 374
 375   int pipe0_flags = fcntl(phc->pipes[0], F_GETFL, 0);
 376   int pipe1_flags = fcntl(phc->pipes[1], F_GETFL, 0);
 377   if (pipe0_flags == -1 || pipe1_flags == -1) {
 378     WARNING("dpdk_helper_spawn: error setting up pipe flags: %s",
 379             sstrerror(errno, errbuf, sizeof(errbuf)));
 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",
 385             sstrerror(errno, errbuf, sizeof(errbuf)));
 386   }
 387
 388   pid_t pid = fork();
 389   if (pid > 0) {
 390     phc->pid = pid;
 391     close(phc->pipes[1]);
 392     DEBUG("%s:dpdk_helper_spawn: helper pid %lu", phc->shm_name,
 393           (long)phc->pid);
 394   } else if (pid == 0) {
 395     /* Replace stdout with a pipe to collectd. */
 396     close(phc->pipes[0]);
 397     close(STDOUT_FILENO);
 398     dup2(phc->pipes[1], STDOUT_FILENO);
 399     DPDK_CHILD_TRACE(phc->shm_name);
 400     dpdk_helper_worker(phc);
 401     exit(0);
 402   } else {
 403     ERROR("dpdk_helper_start: Failed to fork helper process: %s",
 404           sstrerror(errno, errbuf, sizeof(errbuf)));
 405     return -1;
 406   }
 407
 408   return 0;
 409 }
 410
 411 static int dpdk_helper_exit(dpdk_helper_ctx_t *phc,
 412                             enum DPDK_HELPER_STATUS status) {
 413   DPDK_CHILD_LOG("%s:%s:%d %s\n", phc->shm_name, __FUNCTION__, __LINE__,
 414                  dpdk_helper_status_str(status));
 415
 416   close(phc->pipes[1]);
 417
 418   phc->status = status;
 419
 420   exit(0);
 421
 422   return 0;
 423 }
 424
 425 static int dpdk_helper_exit_command(dpdk_helper_ctx_t *phc,
 426                                     enum DPDK_HELPER_STATUS status) {
 427   char errbuf[ERR_BUF_SIZE];
 428   DPDK_HELPER_TRACE(phc->shm_name);
 429
 430   close(phc->pipes[1]);
 431
 432   if (phc->status == DPDK_HELPER_ALIVE_SENDING_EVENTS) {
 433     phc->status = status;
 434     DEBUG("%s:%s:%d %s", phc->shm_name, __FUNCTION__, __LINE__,
 435           dpdk_helper_status_str(status));
 436
 437     int ret = dpdk_helper_command(phc, DPDK_CMD_QUIT, NULL, 0);
 438     if (ret != 0) {
 439       DEBUG("%s:%s:%d kill helper (pid=%lu)", phc->shm_name, __FUNCTION__,
 440             __LINE__, (long)phc->pid);
 441
 442       int err = kill(phc->pid, SIGKILL);
 443       if (err) {
 444         ERROR("%s error sending kill to helper: %s", __FUNCTION__,
 445               sstrerror(errno, errbuf, sizeof(errbuf)));
 446       }
 447     }
 448   } else {
 449
 450     DEBUG("%s:%s:%d kill helper (pid=%lu)", phc->shm_name, __FUNCTION__,
 451           __LINE__, (long)phc->pid);
 452
 453     int err = kill(phc->pid, SIGKILL);
 454     if (err) {
 455       ERROR("%s error sending kill to helper: %s", __FUNCTION__,
 456             sstrerror(errno, errbuf, sizeof(errbuf)));
 457     }
 458   }
 459
 460   return 0;
 461 }
 462
 463 static int dpdk_helper_eal_init(dpdk_helper_ctx_t *phc) {
 464   phc->status = DPDK_HELPER_INITIALIZING_EAL;
 465   DPDK_CHILD_LOG("%s:%s:%d DPDK_HELPER_INITIALIZING_EAL (start)\n",
 466                  phc->shm_name, __FUNCTION__, __LINE__);
 467
 468   char *argp[DPDK_EAL_ARGC * 2 + 1];
 469   int argc = 0;
 470
 471   /* EAL config must be initialized */
 472   assert(phc->eal_config.coremask[0] != 0);
 473   assert(phc->eal_config.memory_channels[0] != 0);
 474   assert(phc->eal_config.process_type[0] != 0);
 475   assert(phc->eal_config.file_prefix[0] != 0);
 476
 477   argp[argc++] = "collectd-dpdk";
 478
 479   argp[argc++] = "-c";
 480   argp[argc++] = phc->eal_config.coremask;
 481
 482   argp[argc++] = "-n";
 483   argp[argc++] = phc->eal_config.memory_channels;
 484
 485   if (strcasecmp(phc->eal_config.socket_memory, "") != 0) {
 486     argp[argc++] = "--socket-mem";
 487     argp[argc++] = phc->eal_config.socket_memory;
 488   }
 489
 490   if (strcasecmp(phc->eal_config.file_prefix, DPDK_DEFAULT_RTE_CONFIG) != 0) {
 491     argp[argc++] = "--file-prefix";
 492     argp[argc++] = phc->eal_config.file_prefix;
 493   }
 494
 495   argp[argc++] = "--proc-type";
 496   argp[argc++] = phc->eal_config.process_type;
 497
 498   assert(argc <= (DPDK_EAL_ARGC * 2 + 1));
 499
 500   int ret = rte_eal_init(argc, argp);
 501
 502   if (ret < 0) {
 503
 504     phc->eal_initialized = 0;
 505
 506     DPDK_CHILD_LOG("dpdk_helper_eal_init: ERROR initializing EAL ret=%d\n",
 507                    ret);
 508
 509     printf("dpdk_helper_eal_init: EAL arguments: ");
 510     for (int i = 0; i < argc; i++) {
 511       printf("%s ", argp[i]);
 512     }
 513     printf("\n");
 514
 515     return ret;
 516   }
 517
 518   phc->eal_initialized = 1;
 519
 520   DPDK_CHILD_LOG("%s:%s:%d DPDK_HELPER_INITIALIZING_EAL (done)\n",
 521                  phc->shm_name, __FUNCTION__, __LINE__);
 522
 523   return 0;
 524 }
 525
 526 static int dpdk_helper_cmd_wait(dpdk_helper_ctx_t *phc, pid_t ppid) {
 527   DPDK_CHILD_TRACE(phc->shm_name);
 528
 529   struct timespec ts;
 530   cdtime_t now = cdtime();
 531   cdtime_t cmd_wait_time = MS_TO_CDTIME_T(1500) + phc->cmd_wait_time * 2;
 532   ts = CDTIME_T_TO_TIMESPEC(now + cmd_wait_time);
 533
 534   int ret = sem_timedwait(&phc->sema_cmd_start, &ts);
 535   DPDK_CHILD_LOG("%s:%s:%d pid=%lu got sema_cmd_start (ret=%d, errno=%d)\n",
 536                  phc->shm_name, __FUNCTION__, __LINE__, (long)getpid(), ret,
 537                  errno);
 538
 539   if (phc->cmd == DPDK_CMD_QUIT) {
 540     DPDK_CHILD_LOG("%s:%s:%d pid=%lu exiting\n", phc->shm_name, __FUNCTION__,
 541                    __LINE__, (long)getpid());
 542     exit(0);
 543   } else if (ret == -1 && errno == ETIMEDOUT) {
 544     if (phc->status == DPDK_HELPER_ALIVE_SENDING_EVENTS) {
 545       DPDK_CHILD_LOG("%s:dpdk_helper_cmd_wait: sem timedwait()"
 546                      " timeout, did collectd terminate?\n",
 547                      phc->shm_name);
 548       dpdk_helper_exit(phc, DPDK_HELPER_GRACEFUL_QUIT);
 549     }
 550   }
 551 #if COLLECT_DEBUG
 552   int val = 0;
 553   if (sem_getvalue(&phc->sema_cmd_start, &val) == 0)
 554     DPDK_CHILD_LOG("%s:%s:%d pid=%lu wait sema_cmd_start (value=%d)\n",
 555                    phc->shm_name, __FUNCTION__, __LINE__, (long)getpid(), val);
 556 #endif
 557
 558   /* Parent PID change means collectd died so quit the helper process. */
 559   if (ppid != getppid()) {
 560     DPDK_CHILD_LOG("dpdk_helper_cmd_wait: parent PID changed, quitting.\n");
 561     dpdk_helper_exit(phc, DPDK_HELPER_GRACEFUL_QUIT);
 562   }
 563
 564   /* Checking for DPDK primary process. */
 565   if (!rte_eal_primary_proc_alive(phc->eal_config.file_prefix)) {
 566     if (phc->eal_initialized) {
 567       DPDK_CHILD_LOG(
 568           "%s:dpdk_helper_cmd_wait: no primary alive but EAL initialized:"
 569           " quitting.\n",
 570           phc->shm_name);
 571       dpdk_helper_exit(phc, DPDK_HELPER_NOT_INITIALIZED);
 572     }
 573
 574     phc->status = DPDK_HELPER_WAITING_ON_PRIMARY;
 575     DPDK_CHILD_LOG("%s:%s:%d DPDK_HELPER_WAITING_ON_PRIMARY\n", phc->shm_name,
 576                    __FUNCTION__, __LINE__);
 577
 578     return -1;
 579   }
 580
 581   if (!phc->eal_initialized) {
 582     int ret = dpdk_helper_eal_init(phc);
 583     if (ret != 0) {
 584       DPDK_CHILD_LOG("Error initializing EAL\n");
 585       dpdk_helper_exit(phc, DPDK_HELPER_NOT_INITIALIZED);
 586     }
 587     phc->status = DPDK_HELPER_ALIVE_SENDING_EVENTS;
 588     DPDK_CHILD_LOG("%s:%s:%d DPDK_HELPER_ALIVE_SENDING_EVENTS\n", phc->shm_name,
 589                    __FUNCTION__, __LINE__);
 590     return -1;
 591   }
 592
 593   return 0;
 594 }
 595
 596 static int dpdk_helper_worker(dpdk_helper_ctx_t *phc) {
 597   DPDK_CHILD_TRACE(phc->shm_name);
 598
 599   pid_t ppid = getppid();
 600
 601   while (1) {
 602     if (dpdk_helper_cmd_wait(phc, ppid) == 0) {
 603       DPDK_CHILD_LOG("%s:%s:%d DPDK command handle (cmd=%d, pid=%lu)\n",
 604                      phc->shm_name, __FUNCTION__, __LINE__, phc->cmd,
 605                      (long)getpid());
 606       phc->cmd_result = dpdk_helper_command_handler(phc, phc->cmd);
 607     } else {
 608       phc->cmd_result = -1;
 609     }
 610
 611     /* now kick collectd to get results */
 612     int err = sem_post(&phc->sema_cmd_complete);
 613     DPDK_CHILD_LOG("%s:%s:%d post sema_cmd_complete (pid=%lu)\n", phc->shm_name,
 614                    __FUNCTION__, __LINE__, (long)getpid());
 615     if (err) {
 616       char errbuf[ERR_BUF_SIZE];
 617       DPDK_CHILD_LOG("dpdk_helper_worker: error posting sema_cmd_complete "
 618                      "semaphore (%s)\n",
 619                      sstrerror(errno, errbuf, sizeof(errbuf)));
 620     }
 621
 622 #if COLLECT_DEBUG
 623     int val = 0;
 624     if (sem_getvalue(&phc->sema_cmd_complete, &val) == 0)
 625       DPDK_CHILD_LOG("%s:%s:%d pid=%lu sema_cmd_complete (value=%d)\n",
 626                      phc->shm_name, __FUNCTION__, __LINE__, (long)getpid(),
 627                      val);
 628 #endif
 629
 630   } /* while(1) */
 631
 632   return 0;
 633 }
 634
 635 static const char *dpdk_helper_status_str(enum DPDK_HELPER_STATUS status) {
 636   switch (status) {
 637   case DPDK_HELPER_ALIVE_SENDING_EVENTS:
 638     return "DPDK_HELPER_ALIVE_SENDING_EVENTS";
 639   case DPDK_HELPER_WAITING_ON_PRIMARY:
 640     return "DPDK_HELPER_WAITING_ON_PRIMARY";
 641   case DPDK_HELPER_INITIALIZING:
 642     return "DPDK_HELPER_INITIALIZING";
 643   case DPDK_HELPER_INITIALIZING_EAL:
 644     return "DPDK_HELPER_INITIALIZING_EAL";
 645   case DPDK_HELPER_GRACEFUL_QUIT:
 646     return "DPDK_HELPER_GRACEFUL_QUIT";
 647   case DPDK_HELPER_NOT_INITIALIZED:
 648     return "DPDK_HELPER_NOT_INITIALIZED";
 649   default:
 650     return "UNKNOWN";
 651   }
 652 }
 653
 654 static int dpdk_helper_status_check(dpdk_helper_ctx_t *phc) {
 655   DEBUG("%s:%s:%d pid=%u %s", phc->shm_name, __FUNCTION__, __LINE__, getpid(),
 656         dpdk_helper_status_str(phc->status));
 657   char errbuf[ERR_BUF_SIZE];
 658
 659   if (phc->status == DPDK_HELPER_GRACEFUL_QUIT) {
 660     return 0;
 661   } else if (phc->status == DPDK_HELPER_NOT_INITIALIZED) {
 662     phc->status = DPDK_HELPER_INITIALIZING;
 663     DEBUG("%s:%s:%d DPDK_HELPER_INITIALIZING", phc->shm_name, __FUNCTION__,
 664           __LINE__);
 665     int err = dpdk_helper_spawn(phc);
 666     if (err) {
 667       ERROR("dpdkstat: error spawning helper %s",
 668             sstrerror(errno, errbuf, sizeof(errbuf)));
 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",
 681             sstrerror(errno, errbuf, sizeof(errbuf)));
 682     }
 683     return -1;
 684   }
 685
 686   if (phc->status == DPDK_HELPER_INITIALIZING_EAL) {
 687     return -1;
 688   }
 689
 690   return 0;
 691 }
 692
 693 static void dpdk_helper_check_pipe(dpdk_helper_ctx_t *phc) {
 694   char buf[DPDK_MAX_BUFFER_SIZE];
 695   char out[DPDK_MAX_BUFFER_SIZE];
 696
 697   /* non blocking check on helper logging pipe */
 698   struct pollfd fds = {
 699       .fd = phc->pipes[0], .events = POLLIN,
 700   };
 701   int data_avail = poll(&fds, 1, 0);
 702   if (data_avail < 0) {
 703     if (errno != EINTR || errno != EAGAIN) {
 704       char errbuf[ERR_BUF_SIZE];
 705       ERROR("%s: poll(2) failed: %s", phc->shm_name,
 706             sstrerror(errno, errbuf, sizeof(errbuf)));
 707     }
 708   }
 709   while (data_avail) {
 710     int nbytes = read(phc->pipes[0], buf, sizeof(buf));
 711     if (nbytes <= 0)
 712       break;
 713     sstrncpy(out, buf, nbytes);
 714     DEBUG("%s: helper process:\n%s", phc->shm_name, out);
 715   }
 716 }
 717
 718 int dpdk_helper_command(dpdk_helper_ctx_t *phc, enum DPDK_CMD cmd, int *result,
 719                         cdtime_t cmd_wait_time) {
 720   if (phc == NULL) {
 721     ERROR("Invalid argument(phc)");
 722     return -EINVAL;
 723   }
 724
 725   DEBUG("%s:%s:%d pid=%lu, cmd=%d", phc->shm_name, __FUNCTION__, __LINE__,
 726         (long)getpid(), cmd);
 727
 728   phc->cmd_wait_time = cmd_wait_time;
 729
 730   int ret = dpdk_helper_status_check(phc);
 731
 732   dpdk_helper_check_pipe(phc);
 733
 734   if (ret != 0) {
 735     return ret;
 736   }
 737
 738   DEBUG("%s: DPDK command execute (cmd=%d)", phc->shm_name, cmd);
 739
 740   phc->cmd_result = 0;
 741   phc->cmd = cmd;
 742
 743   /* kick helper to process command */
 744   int err = sem_post(&phc->sema_cmd_start);
 745   if (err) {
 746     char errbuf[ERR_BUF_SIZE];
 747     ERROR("dpdk_helper_worker: error posting sema_cmd_start semaphore (%s)",
 748           sstrerror(errno, errbuf, sizeof(errbuf)));
 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];
 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() {
 856   uint8_t ports = rte_eth_dev_count();
 857   if (ports == 0) {
 858     ERROR(
 859         "%s:%d: No DPDK ports available. Check bound devices to DPDK driver.\n",
 860         __FUNCTION__, __LINE__);
 861     return ports;
 862   }
 863
 864   if (ports > RTE_MAX_ETHPORTS) {
 865     ERROR("%s:%d: Number of DPDK ports (%u) is greater than "
 866           "RTE_MAX_ETHPORTS=%d. Ignoring extra ports\n",
 867           __FUNCTION__, __LINE__, ports, RTE_MAX_ETHPORTS);
 868     ports = RTE_MAX_ETHPORTS;
 869   }
 870
 871   return ports;
 872 }