2 * collectd - src/utils_dpdk.c
5 * Copyright(c) 2016 Intel Corporation. All rights reserved.
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:
14 * The above copyright notice and this permission notice shall be included in
15 * all copies or substantial portions of the Software.
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
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>
36 #include <semaphore.h>
39 #include <rte_config.h>
41 #include <rte_ethdev.h>
43 #include "utils/common/common.h"
44 #include "utils/dpdk/dpdk.h"
46 #if RTE_VERSION <= RTE_VERSION_NUM(18, 5, 0, 0)
47 #define DPDK_DEFAULT_RTE_CONFIG "/var/run/.rte_config"
49 #define DPDK_DEFAULT_RTE_CONFIG "/var/run/dpdk/rte/config"
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
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,
67 #define DPDK_HELPER_TRACE(_name) \
68 DEBUG("%s:%s:%d pid=%ld", _name, __FUNCTION__, __LINE__, (long)getpid())
70 struct dpdk_helper_ctx_s {
72 dpdk_eal_config_t eal_config;
76 char shm_name[DATA_MAX_NAME_LEN];
79 sem_t sema_cmd_complete;
80 cdtime_t cmd_wait_time;
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);
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);
107 static void dpdk_helper_config_default(dpdk_helper_ctx_t *phc) {
111 DPDK_HELPER_TRACE(phc->shm_name);
113 ssnprintf(phc->eal_config.coremask, DATA_MAX_NAME_LEN, "%s", "0xf");
114 ssnprintf(phc->eal_config.memory_channels, DATA_MAX_NAME_LEN, "%s", "1");
115 ssnprintf(phc->eal_config.file_prefix, DATA_MAX_NAME_LEN, "%s",
116 DPDK_DEFAULT_RTE_CONFIG);
119 int dpdk_helper_eal_config_set(dpdk_helper_ctx_t *phc, dpdk_eal_config_t *ec) {
121 ERROR("Invalid argument (phc)");
125 DPDK_HELPER_TRACE(phc->shm_name);
128 ERROR("Invalid argument (ec)");
132 memcpy(&phc->eal_config, ec, sizeof(phc->eal_config));
137 int dpdk_helper_eal_config_get(dpdk_helper_ctx_t *phc, dpdk_eal_config_t *ec) {
139 ERROR("Invalid argument (phc)");
143 DPDK_HELPER_TRACE(phc->shm_name);
146 ERROR("Invalid argument (ec)");
150 memcpy(ec, &phc->eal_config, sizeof(*ec));
155 int dpdk_helper_eal_config_parse(dpdk_helper_ctx_t *phc, oconfig_item_t *ci) {
156 DPDK_HELPER_TRACE(phc->shm_name);
159 ERROR("Invalid argument (phc)");
164 ERROR("Invalid argument (ci)");
169 for (int i = 0; i < ci->children_num; i++) {
170 oconfig_item_t *child = ci->children + i;
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) {
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];
189 status = cf_util_get_string_buffer(child, prefix, sizeof(prefix));
191 #if RTE_VERSION <= RTE_VERSION_NUM(18, 5, 0, 0)
192 ssnprintf(phc->eal_config.file_prefix, DATA_MAX_NAME_LEN,
193 "/var/run/.%s_config", prefix);
195 ssnprintf(phc->eal_config.file_prefix, DATA_MAX_NAME_LEN,
196 "/var/run/dpdk/%s/config", prefix);
198 DEBUG("dpdk_common: EAL:File prefix %s", phc->eal_config.file_prefix);
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);
211 ERROR("dpdk_common: Invalid '%s' configuration option", child->key);
216 ERROR("dpdk_common: Parsing EAL configuration failed");
224 static int dpdk_shm_init(const char *name, size_t size, void **map) {
225 DPDK_HELPER_TRACE(name);
227 int fd = shm_open(name, O_CREAT | O_TRUNC | O_RDWR, 0666);
229 WARNING("dpdk_shm_init: Failed to open %s as SHM:%s", name, STRERRNO);
234 int ret = ftruncate(fd, size);
236 WARNING("dpdk_shm_init: Failed to resize SHM:%s", STRERRNO);
239 dpdk_shm_cleanup(name, size, NULL);
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);
248 dpdk_shm_cleanup(name, size, NULL);
251 /* File descriptor no longer needed for shared memory operations */
253 memset(*map, 0, size);
258 static void dpdk_shm_cleanup(const char *name, size_t size, void *map) {
259 DPDK_HELPER_TRACE(name);
262 * Call shm_unlink first, as 'name' might be no longer accessible after munmap
264 if (shm_unlink(name))
265 ERROR("shm_unlink failure %s", STRERRNO);
268 if (munmap(map, size))
269 ERROR("munmap failure %s", STRERRNO);
273 void *dpdk_helper_priv_get(dpdk_helper_ctx_t *phc) {
275 return phc->priv_data;
280 int dpdk_helper_data_size_get(dpdk_helper_ctx_t *phc) {
282 DPDK_CHILD_LOG("Invalid argument(phc)\n");
286 return phc->shm_size - sizeof(dpdk_helper_ctx_t);
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;
295 ERROR("%s:Invalid argument(pphc)", __FUNCTION__);
300 ERROR("%s:Invalid argument(name)", __FUNCTION__);
304 DPDK_HELPER_TRACE(name);
306 /* Allocate dpdk_helper_ctx_t and
307 * initialize a POSIX SHared Memory (SHM) object.
309 int err = dpdk_shm_init(name, shm_size, (void **)&phc);
314 err = sem_init(&phc->sema_cmd_start, 1, 0);
316 ERROR("sema_cmd_start semaphore init failed: %s", STRERRNO);
318 dpdk_shm_cleanup(name, shm_size, (void *)phc);
322 err = sem_init(&phc->sema_cmd_complete, 1, 0);
324 ERROR("sema_cmd_complete semaphore init failed: %s", STRERRNO);
325 sem_destroy(&phc->sema_cmd_start);
327 dpdk_shm_cleanup(name, shm_size, (void *)phc);
331 phc->shm_size = shm_size;
332 sstrncpy(phc->shm_name, name, sizeof(phc->shm_name));
334 dpdk_helper_config_default(phc);
341 void dpdk_helper_shutdown(dpdk_helper_ctx_t *phc) {
345 DPDK_HELPER_TRACE(phc->shm_name);
347 close(phc->pipes[1]);
349 if (phc->status != DPDK_HELPER_NOT_INITIALIZED) {
350 dpdk_helper_exit_command(phc, DPDK_HELPER_GRACEFUL_QUIT);
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);
358 static int dpdk_helper_spawn(dpdk_helper_ctx_t *phc) {
360 ERROR("Invalid argument(phc)");
364 DPDK_HELPER_TRACE(phc->shm_name);
366 phc->eal_initialized = 0;
367 phc->cmd_wait_time = MS_TO_CDTIME_T(DPDK_CDM_DEFAULT_TIMEOUT);
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().
374 DEBUG("dpdk_helper_spawn: collectd closing helper pipe %d", phc->pipes[1]);
376 DEBUG("dpdk_helper_spawn: collectd helper pipe %d, not closing",
380 if (pipe(phc->pipes) != 0) {
381 DEBUG("dpdk_helper_spawn: Could not create helper pipe: %s", STRERRNO);
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);
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);
399 close(phc->pipes[1]);
400 DEBUG("%s:dpdk_helper_spawn: helper pid %lu", phc->shm_name,
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);
411 ERROR("dpdk_helper_start: Failed to fork helper process: %s", STRERRNO);
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));
423 close(phc->pipes[1]);
425 phc->status = status;
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);
436 close(phc->pipes[1]);
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));
443 int ret = dpdk_helper_command(phc, DPDK_CMD_QUIT, NULL, 0);
445 DEBUG("%s:%s:%d kill helper (pid=%lu)", phc->shm_name, __FUNCTION__,
446 __LINE__, (long)phc->pid);
448 int err = kill(phc->pid, SIGKILL);
450 ERROR("%s error sending kill to helper: %s", __FUNCTION__, STRERRNO);
455 DEBUG("%s:%s:%d kill helper (pid=%lu)", phc->shm_name, __FUNCTION__,
456 __LINE__, (long)phc->pid);
458 int err = kill(phc->pid, SIGKILL);
460 ERROR("%s error sending kill to helper: %s", __FUNCTION__, STRERRNO);
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__);
472 char *argp[DPDK_EAL_ARGC * 2 + 1];
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);
480 argp[argc++] = "collectd-dpdk";
483 argp[argc++] = phc->eal_config.coremask;
486 argp[argc++] = phc->eal_config.memory_channels;
488 if (strcasecmp(phc->eal_config.socket_memory, "") != 0) {
489 argp[argc++] = "--socket-mem";
490 argp[argc++] = phc->eal_config.socket_memory;
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;
498 argp[argc++] = "--proc-type";
499 argp[argc++] = "secondary";
501 if (strcasecmp(phc->eal_config.log_level, "") != 0) {
502 argp[argc++] = "--log-level";
503 argp[argc++] = phc->eal_config.log_level;
505 if (strcasecmp(phc->eal_config.rte_driver_lib_path, "") != 0) {
507 argp[argc++] = phc->eal_config.rte_driver_lib_path;
510 assert(argc <= (DPDK_EAL_ARGC * 2 + 1));
512 int ret = rte_eal_init(argc, argp);
516 phc->eal_initialized = 0;
518 DPDK_CHILD_LOG("dpdk_helper_eal_init: ERROR initializing EAL ret=%d\n",
521 printf("dpdk_helper_eal_init: EAL arguments: ");
522 for (int i = 0; i < argc; i++) {
523 printf("%s ", argp[i]);
530 phc->eal_initialized = 1;
532 DPDK_CHILD_LOG("%s:%s:%d DPDK_HELPER_INITIALIZING_EAL (done)\n",
533 phc->shm_name, __FUNCTION__, __LINE__);
538 static int dpdk_helper_cmd_wait(dpdk_helper_ctx_t *phc, pid_t ppid) {
539 DPDK_CHILD_TRACE(phc->shm_name);
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);
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,
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());
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",
560 dpdk_helper_exit(phc, DPDK_HELPER_GRACEFUL_QUIT);
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);
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);
576 /* Checking for DPDK primary process. */
577 if (!rte_eal_primary_proc_alive(phc->eal_config.file_prefix)) {
578 if (phc->eal_initialized) {
580 "%s:dpdk_helper_cmd_wait: no primary alive but EAL initialized:"
583 dpdk_helper_exit(phc, DPDK_HELPER_NOT_INITIALIZED);
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__);
593 if (!phc->eal_initialized) {
594 int ret = dpdk_helper_eal_init(phc);
596 DPDK_CHILD_LOG("Error initializing EAL\n");
597 dpdk_helper_exit(phc, DPDK_HELPER_NOT_INITIALIZED);
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__);
608 static int dpdk_helper_worker(dpdk_helper_ctx_t *phc) {
609 DPDK_CHILD_TRACE(phc->shm_name);
611 pid_t ppid = getppid();
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,
618 phc->cmd_result = dpdk_helper_command_handler(phc, phc->cmd);
620 phc->cmd_result = -1;
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());
628 DPDK_CHILD_LOG("dpdk_helper_worker: error posting sema_cmd_complete "
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(),
646 static const char *dpdk_helper_status_str(enum DPDK_HELPER_STATUS 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";
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));
669 if (phc->status == DPDK_HELPER_GRACEFUL_QUIT) {
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__,
675 int err = dpdk_helper_spawn(phc);
677 ERROR("dpdkstat: error spawning helper %s", STRERRNO);
682 pid_t ws = waitpid(phc->pid, NULL, WNOHANG);
684 phc->status = DPDK_HELPER_INITIALIZING;
685 DEBUG("%s:%s:%d DPDK_HELPER_INITIALIZING", phc->shm_name, __FUNCTION__,
687 int err = dpdk_helper_spawn(phc);
689 ERROR("dpdkstat: error spawning helper %s", STRERRNO);
694 if (phc->status == DPDK_HELPER_INITIALIZING_EAL) {
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];
705 /* non blocking check on helper logging pipe */
706 struct pollfd fds = {
707 .fd = phc->pipes[0], .events = POLLIN,
709 int data_avail = poll(&fds, 1, 0);
710 DEBUG("%s:dpdk_helper_check_pipe: poll data_avail=%d", phc->shm_name,
712 if (data_avail < 0) {
713 if (errno != EINTR || errno != EAGAIN) {
714 ERROR("%s: poll(2) failed: %s", phc->shm_name, STRERRNO);
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);
723 sstrncpy(out, buf, sizeof(out));
724 DEBUG("%s: helper process:\n%s", phc->shm_name, out);
728 int dpdk_helper_command(dpdk_helper_ctx_t *phc, enum DPDK_CMD cmd, int *result,
729 cdtime_t cmd_wait_time) {
731 ERROR("Invalid argument(phc)");
735 DEBUG("%s:%s:%d pid=%lu, cmd=%d", phc->shm_name, __FUNCTION__, __LINE__,
736 (long)getpid(), cmd);
738 phc->cmd_wait_time = cmd_wait_time;
740 int ret = dpdk_helper_status_check(phc);
742 dpdk_helper_check_pipe(phc);
748 DEBUG("%s: DPDK command execute (cmd=%d)", phc->shm_name, cmd);
753 /* kick helper to process command */
754 int err = sem_post(&phc->sema_cmd_start);
756 ERROR("dpdk_helper_worker: error posting sema_cmd_start semaphore (%s)",
762 if (sem_getvalue(&phc->sema_cmd_start, &val) == 0)
763 DEBUG("%s:dpdk_helper_command: post sema_cmd_start (value=%d)",
767 if (phc->cmd != DPDK_CMD_QUIT) {
769 /* wait for helper to complete processing */
771 cdtime_t now = cdtime();
773 if (phc->status != DPDK_HELPER_ALIVE_SENDING_EVENTS) {
774 cmd_wait_time = MS_TO_CDTIME_T(DPDK_CDM_DEFAULT_TIMEOUT);
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 "
789 if (sem_getvalue(&phc->sema_cmd_complete, &val) == 0)
790 DEBUG("%s:dpdk_helper_command: wait sema_cmd_complete (value=%d)",
795 *result = phc->cmd_result;
799 dpdk_helper_check_pipe(phc);
801 DEBUG("%s: DPDK command complete (cmd=%d, result=%d)", phc->shm_name,
802 phc->cmd, phc->cmd_result);
807 uint64_t strtoull_safe(const char *str, int *err) {
812 val = strtoull(str, &endptr, 16);
814 ERROR("%s Failed to parse the value %s, endptr=%c", __FUNCTION__, str,
823 uint128_t str_to_uint128(const char *str, int len) {
824 uint128_t lcore_mask;
827 memset(&lcore_mask, 0, sizeof(lcore_mask));
829 if (len <= 2 || strncmp(str, "0x", 2) != 0) {
830 ERROR("%s Value %s should be represened in hexadecimal format",
834 /* If str is <= 64 bit long ('0x' + 16 chars = 18 chars) then
835 * conversion is straightforward. Otherwise str is splitted into 64b long
838 lcore_mask.low = strtoull_safe(str, &err);
842 char low_str[DATA_MAX_NAME_LEN];
843 char high_str[DATA_MAX_NAME_LEN * 2];
845 memset(high_str, 0, sizeof(high_str));
846 memset(low_str, 0, sizeof(low_str));
848 strncpy(high_str, str, len - 16);
849 strncpy(low_str, str + len - 16, 16);
851 lcore_mask.low = strtoull_safe(low_str, &err);
855 lcore_mask.high = strtoull_safe(high_str, &err);
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();
868 uint8_t ports = rte_eth_dev_count_avail();
872 "%s:%d: No DPDK ports available. Check bound devices to DPDK driver.\n",
873 __FUNCTION__, __LINE__);
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;