-/*-
+/*
* collectd - src/dpdkstat.c
* MIT License
*
* Maryam Tahhan <maryam.tahhan@intel.com>
* Harry van Haaren <harry.van.haaren@intel.com>
* Taras Chornyi <tarasx.chornyi@intel.com>
+ * Serhiy Pshyk <serhiyx.pshyk@intel.com>
+ * Krzysztof Matczak <krzysztofx.matczak@intel.com>
*/
#include "collectd.h"
-#include "common.h" /* auxiliary functions */
-#include "plugin.h" /* plugin_register_*, plugin_dispatch_values */
-#include "utils_time.h"
-
-#include <getopt.h>
-#include <poll.h>
-#include <semaphore.h>
-#include <sys/mman.h>
-#include <sys/queue.h>
+#include "common.h"
+#include "utils_dpdk.h"
-#include <rte_atomic.h>
-#include <rte_branch_prediction.h>
-#include <rte_common.h>
#include <rte_config.h>
-#include <rte_debug.h>
-#include <rte_debug.h>
-#include <rte_eal.h>
#include <rte_ethdev.h>
-#include <rte_launch.h>
-#include <rte_lcore.h>
-#include <rte_log.h>
-#include <rte_malloc.h>
-#include <rte_memory.h>
-#include <rte_memzone.h>
-#include <rte_per_lcore.h>
-#include <rte_string_fns.h>
-#include <rte_tailq.h>
-#include <rte_version.h>
-
-#define DPDK_DEFAULT_RTE_CONFIG "/var/run/.rte_config"
-#define DPDK_MAX_ARGC 8
-#define DPDKSTAT_MAX_BUFFER_SIZE (4096 * 4)
-#define DPDK_SHM_NAME "dpdk_collectd_stats_shm"
-#define ERR_BUF_SIZE 1024
-#define REINIT_SHM 1
-#define RESET 1
-#define NO_RESET 0
+
+#define DPDK_STATS_PLUGIN "dpdkstat"
+#define DPDK_STATS_NAME "dpdk_collectd_stats"
+
+#define DPDK_STATS_TRACE() \
+ DEBUG("%s:%s:%d pid=%u", DPDK_STATS_PLUGIN, __FUNCTION__, __LINE__, getpid())
+
+struct dpdk_stats_config_s {
+ cdtime_t interval;
+ uint32_t enabled_port_mask;
+ char port_name[RTE_MAX_ETHPORTS][DATA_MAX_NAME_LEN];
+};
+typedef struct dpdk_stats_config_s dpdk_stats_config_t;
#define RTE_VERSION_16_07 RTE_VERSION_NUM(16, 7, 0, 16)
ctx->xstats = (struct rte_eth_xstat *)&ctx->raw_data[0]; \
ctx->xnames = \
(struct rte_eth_xstat_name *)&ctx \
- ->raw_data[ctx->num_xstats * sizeof(struct rte_eth_xstat)]; \
+ ->raw_data[ctx->stats_count * sizeof(struct rte_eth_xstat)]; \
} while (0)
#endif
-enum DPDK_HELPER_ACTION {
- DPDK_HELPER_ACTION_COUNT_STATS,
- DPDK_HELPER_ACTION_SEND_STATS,
-};
-
-enum DPDK_HELPER_STATUS {
- DPDK_HELPER_NOT_INITIALIZED = 0,
- DPDK_HELPER_WAITING_ON_PRIMARY,
- DPDK_HELPER_INITIALIZING_EAL,
- DPDK_HELPER_ALIVE_SENDING_STATS,
- DPDK_HELPER_GRACEFUL_QUIT,
-};
-
-struct dpdk_config_s {
- /* General DPDK params */
- char coremask[DATA_MAX_NAME_LEN];
- char memory_channels[DATA_MAX_NAME_LEN];
- char socket_memory[DATA_MAX_NAME_LEN];
- char process_type[DATA_MAX_NAME_LEN];
- char file_prefix[DATA_MAX_NAME_LEN];
- cdtime_t interval;
- uint32_t eal_initialized;
- uint32_t enabled_port_mask;
- char port_name[RTE_MAX_ETHPORTS][DATA_MAX_NAME_LEN];
- uint32_t eal_argc;
- /* Helper info */
- int collectd_reinit_shm;
- pid_t helper_pid;
- sem_t sema_helper_get_stats;
- sem_t sema_stats_in_shm;
- int helper_pipes[2];
- enum DPDK_HELPER_STATUS helper_status;
- enum DPDK_HELPER_ACTION helper_action;
- /* xstats info */
- uint32_t num_ports;
- uint32_t num_xstats;
+struct dpdk_stats_ctx_s {
+ dpdk_stats_config_t config;
+ uint32_t stats_count;
+ uint32_t ports_count;
cdtime_t port_read_time[RTE_MAX_ETHPORTS];
- uint32_t num_stats_in_port[RTE_MAX_ETHPORTS];
- struct rte_eth_link link_status[RTE_MAX_ETHPORTS];
+ uint32_t port_stats_count[RTE_MAX_ETHPORTS];
#if RTE_VERSION < RTE_VERSION_16_07
struct rte_eth_xstats *xstats;
#else
struct rte_eth_xstat *xstats;
struct rte_eth_xstat_name *xnames;
#endif
- char *raw_data;
- /* rte_eth_xstats from here on until the end of the SHM */
+ char raw_data[];
};
-typedef struct dpdk_config_s dpdk_config_t;
-
-static int g_configured;
-static dpdk_config_t *g_configuration;
-
-static void dpdk_config_init_default(void);
-static int dpdk_config(oconfig_item_t *ci);
-static int dpdk_helper_init_eal(void);
-static int dpdk_helper_run(void);
-static int dpdk_helper_spawn(enum DPDK_HELPER_ACTION action);
-static int dpdk_init(void);
-static int dpdk_read(user_data_t *ud);
-static int dpdk_shm_cleanup(void);
-static int dpdk_shm_init(size_t size);
-
-/* Write the default configuration to the g_configuration instances */
-static void dpdk_config_init_default(void) {
- g_configuration->interval = plugin_get_interval();
- if (g_configuration->interval == cf_get_default_interval())
- WARNING("dpdkstat: No time interval was configured, default value %" PRIu64
- " ms is set",
- CDTIME_T_TO_MS(g_configuration->interval));
- /* Default is all ports enabled */
- g_configuration->enabled_port_mask = ~0;
- g_configuration->eal_argc = DPDK_MAX_ARGC;
- g_configuration->eal_initialized = 0;
- ssnprintf(g_configuration->coremask, DATA_MAX_NAME_LEN, "%s", "0xf");
- ssnprintf(g_configuration->memory_channels, DATA_MAX_NAME_LEN, "%s", "1");
- ssnprintf(g_configuration->process_type, DATA_MAX_NAME_LEN, "%s",
- "secondary");
- ssnprintf(g_configuration->file_prefix, DATA_MAX_NAME_LEN, "%s",
- DPDK_DEFAULT_RTE_CONFIG);
-
- for (int i = 0; i < RTE_MAX_ETHPORTS; i++)
- g_configuration->port_name[i][0] = 0;
+typedef struct dpdk_stats_ctx_s dpdk_stats_ctx_t;
+
+#define DPDK_STATS_CTX_GET(a) ((dpdk_stats_ctx_t *)dpdk_helper_priv_get(a))
+
+dpdk_helper_ctx_t *g_hc = NULL;
+
+static void dpdk_stats_default_config(void) {
+ dpdk_stats_ctx_t *ec = DPDK_STATS_CTX_GET(g_hc);
+
+ ec->config.interval = plugin_get_interval();
+
+ for (int i = 0; i < RTE_MAX_ETHPORTS; i++) {
+ ec->config.port_name[i][0] = 0;
+ }
}
-static int dpdk_config(oconfig_item_t *ci) {
- int port_counter = 0;
- /* Allocate g_configuration and
- * initialize a POSIX SHared Memory (SHM) object.
- */
- int err = dpdk_shm_init(sizeof(dpdk_config_t));
- if (err) {
+static int dpdk_stats_preinit(void) {
+ DPDK_STATS_TRACE();
+
+ if (g_hc != NULL) {
+ /* already initialized if config callback was called before init callback */
+ DEBUG("dpdk_stats_preinit: helper already initialized");
+ return 0;
+ }
+
+ int ret = dpdk_helper_init(DPDK_STATS_NAME, sizeof(dpdk_stats_ctx_t), &g_hc);
+ if (ret != 0) {
char errbuf[ERR_BUF_SIZE];
- ERROR("dpdkstat: error in shm_init, %s",
- sstrerror(errno, errbuf, sizeof(errbuf)));
- return -1;
+ ERROR("%s: failed to initialize %s helper(error: %s)", DPDK_STATS_PLUGIN,
+ DPDK_STATS_NAME, sstrerror(errno, errbuf, sizeof(errbuf)));
+ return ret;
}
- /* Set defaults for config, overwritten by loop if config item exists */
- dpdk_config_init_default();
+ dpdk_stats_default_config();
+
+ return ret;
+}
+
+static int dpdk_stats_config(oconfig_item_t *ci) {
+ DPDK_STATS_TRACE();
+
+ int ret = dpdk_stats_preinit();
+ if (ret)
+ return ret;
+
+ dpdk_stats_ctx_t *ctx = DPDK_STATS_CTX_GET(g_hc);
for (int i = 0; i < ci->children_num; i++) {
oconfig_item_t *child = ci->children + i;
- if (strcasecmp("Coremask", child->key) == 0) {
- cf_util_get_string_buffer(child, g_configuration->coremask,
- sizeof(g_configuration->coremask));
- DEBUG("dpdkstat:COREMASK %s ", g_configuration->coremask);
- } else if (strcasecmp("MemoryChannels", child->key) == 0) {
- cf_util_get_string_buffer(child, g_configuration->memory_channels,
- sizeof(g_configuration->memory_channels));
- DEBUG("dpdkstat:Memory Channels %s ", g_configuration->memory_channels);
- } else if (strcasecmp("SocketMemory", child->key) == 0) {
- cf_util_get_string_buffer(child, g_configuration->socket_memory,
- sizeof(g_configuration->memory_channels));
- DEBUG("dpdkstat: socket mem %s ", g_configuration->socket_memory);
- } else if (strcasecmp("ProcessType", child->key) == 0) {
- cf_util_get_string_buffer(child, g_configuration->process_type,
- sizeof(g_configuration->process_type));
- DEBUG("dpdkstat: proc type %s ", g_configuration->process_type);
- } else if ((strcasecmp("FilePrefix", child->key) == 0) &&
- (child->values[0].type == OCONFIG_TYPE_STRING)) {
- ssnprintf(g_configuration->file_prefix, DATA_MAX_NAME_LEN,
- "/var/run/.%s_config", child->values[0].value.string);
- DEBUG("dpdkstat: file prefix %s ", g_configuration->file_prefix);
- } else if ((strcasecmp("EnabledPortMask", child->key) == 0) &&
- (child->values[0].type == OCONFIG_TYPE_NUMBER)) {
- g_configuration->enabled_port_mask =
- (uint32_t)child->values[0].value.number;
- DEBUG("dpdkstat: Enabled Port Mask %u",
- g_configuration->enabled_port_mask);
- } else if (strcasecmp("PortName", child->key) == 0) {
- cf_util_get_string_buffer(
- child, g_configuration->port_name[port_counter],
- sizeof(g_configuration->port_name[port_counter]));
- DEBUG("dpdkstat: Port %d Name: %s ", port_counter,
- g_configuration->port_name[port_counter]);
- port_counter++;
- } else {
- WARNING("dpdkstat: The config option \"%s\" is unknown.", child->key);
+ if ((strcasecmp("EnabledPortMask", child->key) == 0) &&
+ (child->values[0].type == OCONFIG_TYPE_NUMBER)) {
+ ctx->config.enabled_port_mask = child->values[0].value.number;
+ DEBUG("%s: Enabled Port Mask 0x%X", DPDK_STATS_PLUGIN,
+ ctx->config.enabled_port_mask);
+ } else if (strcasecmp("EAL", child->key) == 0) {
+ ret = dpdk_helper_eal_config_parse(g_hc, child);
+ if (ret)
+ return ret;
}
- } /* End for (int i = 0; i < ci->children_num; i++)*/
- g_configured = 1; /* Bypass configuration in dpdk_shm_init(). */
+ }
- return 0;
-}
+ int port_num = 0;
-/*
- * Allocate g_configuration and initialize SHared Memory (SHM)
- * for config and helper process
- */
-static int dpdk_shm_init(size_t size) {
- /*
- * Check if SHM is already configured: when config items are provided, the
- * config function initializes SHM. If there is no config, then init() will
- * just return.
- */
- if (g_configuration)
- return 0;
+ /* parse port names after EnabledPortMask was parsed */
+ for (int i = 0; i < ci->children_num; i++) {
+ oconfig_item_t *child = ci->children + i;
- char errbuf[ERR_BUF_SIZE];
+ if (strcasecmp("PortName", child->key) == 0) {
- /* Create and open a new object, or open an existing object. */
- int fd = shm_open(DPDK_SHM_NAME, O_CREAT | O_TRUNC | O_RDWR, 0666);
- if (fd < 0) {
- WARNING("dpdkstat:Failed to open %s as SHM:%s", DPDK_SHM_NAME,
- sstrerror(errno, errbuf, sizeof(errbuf)));
- goto fail;
- }
- /* Set the size of the shared memory object. */
- int ret = ftruncate(fd, size);
- if (ret != 0) {
- WARNING("dpdkstat:Failed to resize SHM:%s",
- sstrerror(errno, errbuf, sizeof(errbuf)));
- goto fail_close;
- }
- /* Map the shared memory object into this process' virtual address space. */
- g_configuration = mmap(0, size, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
- if (g_configuration == MAP_FAILED) {
- WARNING("dpdkstat:Failed to mmap SHM:%s",
- sstrerror(errno, errbuf, sizeof(errbuf)));
- goto fail_close;
- }
- /*
- * Close the file descriptor, the shared memory object still exists
- * and can only be removed by calling shm_unlink().
- */
- close(fd);
-
- /* Initialize g_configuration. */
- memset(g_configuration, 0, size);
-
- /* Initialize the semaphores for SHM use */
- int err = sem_init(&g_configuration->sema_helper_get_stats, 1, 0);
- if (err) {
- ERROR("dpdkstat semaphore init failed: %s",
- sstrerror(errno, errbuf, sizeof(errbuf)));
- goto fail;
- }
- err = sem_init(&g_configuration->sema_stats_in_shm, 1, 0);
- if (err) {
- ERROR("dpdkstat semaphore init failed: %s",
- sstrerror(errno, errbuf, sizeof(errbuf)));
- goto fail;
- }
+ while (!(ctx->config.enabled_port_mask & (1 << port_num)))
+ port_num++;
- g_configuration->xstats = NULL;
+ cf_util_get_string_buffer(child, ctx->config.port_name[port_num],
+ sizeof(ctx->config.port_name[port_num]));
+ DEBUG("%s: Port %d Name: %s", DPDK_STATS_PLUGIN, port_num,
+ ctx->config.port_name[port_num]);
- return 0;
+ port_num++;
+ }
+ }
-fail_close:
- close(fd);
-fail:
- /* Reset to zero, as it was set to MAP_FAILED aka: (void *)-1. Avoid
- * an issue if collectd attempts to run this plugin failure.
- */
- g_configuration = 0;
- return -1;
+ return ret;
}
-static int dpdk_re_init_shm() {
- dpdk_config_t temp_config;
- memcpy(&temp_config, g_configuration, sizeof(dpdk_config_t));
- DEBUG("dpdkstat: %s: ports %" PRIu32 ", xstats %" PRIu32, __func__,
- temp_config.num_ports, temp_config.num_xstats);
-
- size_t shm_xstats_size =
- sizeof(dpdk_config_t) +
- (DPDK_STATS_CTX_GET_XSTAT_SIZE * g_configuration->num_xstats);
- DEBUG("=== SHM new size for %" PRIu32 " xstats", g_configuration->num_xstats);
-
- int err = dpdk_shm_cleanup();
- if (err) {
- ERROR("dpdkstat: Error in shm_cleanup in %s", __func__);
- return err;
- }
- err = dpdk_shm_init(shm_xstats_size);
- if (err) {
- WARNING("dpdkstat: Error in shm_init in %s", __func__);
- return err;
- }
- /* If the XML config() function has been run, don't re-initialize defaults */
- if (!g_configured)
- dpdk_config_init_default();
-
- memcpy(g_configuration, &temp_config, sizeof(dpdk_config_t));
- g_configuration->collectd_reinit_shm = 0;
- g_configuration->raw_data = (char *)(g_configuration + 1);
- DPDK_STATS_CTX_INIT(g_configuration);
- return 0;
-}
+static int dpdk_helper_stats_get(dpdk_helper_ctx_t *phc) {
+ dpdk_stats_ctx_t *ctx = DPDK_STATS_CTX_GET(phc);
-static int dpdk_init(void) {
- int err = dpdk_shm_init(sizeof(dpdk_config_t));
- if (err) {
- ERROR("dpdkstat: %s : error %d in shm_init()", __func__, err);
- return err;
- }
+ /* get stats from DPDK */
- /* If the XML config() function has been run, dont re-initialize defaults */
- if (!g_configured) {
- dpdk_config_init_default();
+ uint8_t ports_count = rte_eth_dev_count();
+ if (ports_count == 0) {
+ DPDK_CHILD_LOG("%s: No DPDK ports available. "
+ "Check bound devices to DPDK driver.\n",
+ DPDK_STATS_PLUGIN);
+ return -ENODEV;
}
- return 0;
-}
+ if (ports_count > RTE_MAX_ETHPORTS)
+ ports_count = RTE_MAX_ETHPORTS;
-static int dpdk_helper_stop(int reset) {
- g_configuration->helper_status = DPDK_HELPER_GRACEFUL_QUIT;
- if (reset) {
- g_configuration->eal_initialized = 0;
- g_configuration->num_ports = 0;
- g_configuration->xstats = NULL;
- g_configuration->num_xstats = 0;
- for (int i = 0; i < RTE_MAX_ETHPORTS; i++)
- g_configuration->num_stats_in_port[i] = 0;
- }
- close(g_configuration->helper_pipes[1]);
- int err = kill(g_configuration->helper_pid, SIGKILL);
- if (err) {
- char errbuf[ERR_BUF_SIZE];
- WARNING("dpdkstat: error sending kill to helper: %s",
- sstrerror(errno, errbuf, sizeof(errbuf)));
+ ctx->ports_count = ports_count;
+
+ int len = 0;
+ int ret = 0;
+ int stats = 0;
+
+ for (uint8_t i = 0; i < ports_count; i++) {
+ if (!(ctx->config.enabled_port_mask & (1 << i)))
+ continue;
+ ctx->port_read_time[i] = cdtime();
+ len = ctx->port_stats_count[i];
+ ret = rte_eth_xstats_get(i, &ctx->xstats[stats], len);
+ if (ret < 0 || ret != len) {
+ DPDK_CHILD_LOG("%s: Error reading stats (port=%d; len=%d)\n",
+ DPDK_STATS_PLUGIN, i, len);
+ return -1;
+ }
+#if RTE_VERSION >= RTE_VERSION_16_07
+ ret = rte_eth_xstats_get_names(i, &ctx->xnames[stats], len);
+ if (ret < 0 || ret != len) {
+ DPDK_CHILD_LOG("%s: Error reading stat names (port=%d; len=%d)\n",
+ DPDK_STATS_PLUGIN, i, len);
+ return -1;
+ }
+#endif
+ stats += len;
}
+ assert(stats == ctx->stats_count);
+
return 0;
}
-static int dpdk_helper_spawn(enum DPDK_HELPER_ACTION action) {
- char errbuf[ERR_BUF_SIZE];
- g_configuration->eal_initialized = 0;
- g_configuration->helper_action = action;
- /*
- * Create a pipe for helper stdout back to collectd. This is necessary for
- * logging EAL failures, as rte_eal_init() calls rte_panic().
- */
- if (pipe(g_configuration->helper_pipes) != 0) {
- DEBUG("dpdkstat: Could not create helper pipe: %s",
- sstrerror(errno, errbuf, sizeof(errbuf)));
- return -1;
- }
+static int dpdk_helper_stats_count_get(dpdk_helper_ctx_t *phc) {
+ dpdk_stats_ctx_t *ctx = DPDK_STATS_CTX_GET(phc);
- int pipe0_flags = fcntl(g_configuration->helper_pipes[0], F_GETFL, 0);
- int pipe1_flags = fcntl(g_configuration->helper_pipes[1], F_GETFL, 0);
- if (pipe0_flags == -1 || pipe1_flags == -1) {
- WARNING("dpdkstat: Failed setting up pipe flags: %s",
- sstrerror(errno, errbuf, sizeof(errbuf)));
- }
- int pipe0_err = fcntl(g_configuration->helper_pipes[0], F_SETFL,
- pipe1_flags | O_NONBLOCK);
- int pipe1_err = fcntl(g_configuration->helper_pipes[1], F_SETFL,
- pipe0_flags | O_NONBLOCK);
- if (pipe0_err == -1 || pipe1_err == -1) {
- WARNING("dpdkstat: Failed setting up pipes: %s",
- sstrerror(errno, errbuf, sizeof(errbuf)));
+ uint8_t ports = rte_eth_dev_count();
+ if (ports == 0) {
+ DPDK_CHILD_LOG("%s: No DPDK ports available. "
+ "Check bound devices to DPDK driver.\n",
+ DPDK_STATS_PLUGIN);
+ return -ENODEV;
}
- pid_t pid = fork();
- if (pid > 0) {
- close(g_configuration->helper_pipes[1]);
- g_configuration->helper_pid = pid;
- DEBUG("dpdkstat: helper pid %li", (long)g_configuration->helper_pid);
- /* Kick helper once its alive to have it start processing */
- sem_post(&g_configuration->sema_helper_get_stats);
- } else if (pid == 0) {
- /* Replace stdout with a pipe to collectd. */
- close(g_configuration->helper_pipes[0]);
- close(STDOUT_FILENO);
- dup2(g_configuration->helper_pipes[1], STDOUT_FILENO);
- dpdk_helper_run();
- exit(0);
- } else {
- ERROR("dpdkstat: Failed to fork helper process: %s",
- sstrerror(errno, errbuf, sizeof(errbuf)));
- return -1;
+ if (ports > RTE_MAX_ETHPORTS)
+ ports = RTE_MAX_ETHPORTS;
+
+ ctx->ports_count = ports;
+
+ int len = 0;
+ int stats_count = 0;
+
+ for (int i = 0; i < ports; i++) {
+ if (!(ctx->config.enabled_port_mask & (1 << i)))
+ continue;
+#if RTE_VERSION >= RTE_VERSION_16_07
+ len = rte_eth_xstats_get_names(i, NULL, 0);
+#else
+ len = rte_eth_xstats_get(i, NULL, 0);
+#endif
+ if (len < 0) {
+ DPDK_CHILD_LOG("%s: Cannot get stats count\n", DPDK_STATS_PLUGIN);
+ return -1;
+ }
+ ctx->port_stats_count[i] = len;
+ stats_count += len;
}
- return 0;
+
+ DPDK_CHILD_LOG("%s:%s:%d stats_count=%d\n", DPDK_STATS_PLUGIN, __FUNCTION__,
+ __LINE__, stats_count);
+
+ return stats_count;
}
-/*
- * Initialize the DPDK EAL, if this returns, EAL is successfully initialized.
- * On failure, the EAL prints an error message, and the helper process exits.
- */
-static int dpdk_helper_init_eal(void) {
- g_configuration->helper_status = DPDK_HELPER_INITIALIZING_EAL;
- char *argp[(g_configuration->eal_argc) + 1];
- int i = 0;
-
- argp[i++] = "collectd-dpdk";
- if (strcasecmp(g_configuration->coremask, "") != 0) {
- argp[i++] = "-c";
- argp[i++] = g_configuration->coremask;
- }
- if (strcasecmp(g_configuration->memory_channels, "") != 0) {
- argp[i++] = "-n";
- argp[i++] = g_configuration->memory_channels;
- }
- if (strcasecmp(g_configuration->socket_memory, "") != 0) {
- argp[i++] = "--socket-mem";
- argp[i++] = g_configuration->socket_memory;
- }
- if (strcasecmp(g_configuration->file_prefix, "") != 0 &&
- strcasecmp(g_configuration->file_prefix, DPDK_DEFAULT_RTE_CONFIG) != 0) {
- argp[i++] = "--file-prefix";
- argp[i++] = g_configuration->file_prefix;
- }
- if (strcasecmp(g_configuration->process_type, "") != 0) {
- argp[i++] = "--proc-type";
- argp[i++] = g_configuration->process_type;
+int dpdk_helper_command_handler(dpdk_helper_ctx_t *phc, enum DPDK_CMD cmd) {
+ /* this function is called from helper context */
+ int ret = 0;
+
+ if (phc == NULL) {
+ DPDK_CHILD_LOG("%s: Invalid argument(phc)\n", DPDK_STATS_PLUGIN);
+ return -EINVAL;
}
- g_configuration->eal_argc = i;
- g_configuration->eal_initialized = 1;
- int ret = rte_eal_init(g_configuration->eal_argc, argp);
- if (ret < 0) {
- g_configuration->eal_initialized = 0;
- return ret;
+ if (cmd != DPDK_CMD_GET_STATS) {
+ DPDK_CHILD_LOG("%s: Unknown command (cmd=%d)\n", DPDK_STATS_PLUGIN, cmd);
+ return -EINVAL;
}
- return 0;
-}
-static int dpdk_helper_run(void) {
- char errbuf[ERR_BUF_SIZE];
- pid_t ppid = getppid();
- g_configuration->helper_status = DPDK_HELPER_WAITING_ON_PRIMARY;
-
- while (1) {
- /* sem_timedwait() to avoid blocking forever */
- cdtime_t now = cdtime();
- cdtime_t safety_period = MS_TO_CDTIME_T(1500);
- int ret =
- sem_timedwait(&g_configuration->sema_helper_get_stats,
- &CDTIME_T_TO_TIMESPEC(now + safety_period +
- g_configuration->interval * 2));
-
- if (ret == -1 && errno == ETIMEDOUT) {
- ERROR("dpdkstat-helper: sem timedwait()"
- " timeout, did collectd terminate?");
- dpdk_helper_stop(RESET);
- }
- /* Parent PID change means collectd died so quit the helper process. */
- if (ppid != getppid()) {
- WARNING("dpdkstat-helper: parent PID changed, quitting.");
- dpdk_helper_stop(RESET);
- }
+ dpdk_stats_ctx_t *ctx = DPDK_STATS_CTX_GET(phc);
- /* Checking for DPDK primary process. */
- if (!rte_eal_primary_proc_alive(g_configuration->file_prefix)) {
- if (g_configuration->eal_initialized) {
- WARNING("dpdkstat-helper: no primary alive but EAL initialized:"
- " quitting.");
- dpdk_helper_stop(RESET);
- }
- g_configuration->helper_status = DPDK_HELPER_WAITING_ON_PRIMARY;
- /* Back to start of while() - waiting for primary process */
- continue;
- }
+ if (ctx->stats_count == 0) {
- if (!g_configuration->eal_initialized) {
- /* Initialize EAL. */
- int ret = dpdk_helper_init_eal();
- if (ret != 0) {
- WARNING("ERROR INITIALIZING EAL");
- dpdk_helper_stop(RESET);
- }
- }
+ int stats_count = dpdk_helper_stats_count_get(phc);
- g_configuration->helper_status = DPDK_HELPER_ALIVE_SENDING_STATS;
+ if (stats_count < 0) {
+ return stats_count;
+ }
- uint8_t nb_ports = rte_eth_dev_count();
- if (nb_ports == 0) {
- DEBUG("dpdkstat-helper: No DPDK ports available. "
- "Check bound devices to DPDK driver.");
- dpdk_helper_stop(RESET);
+ int stats_size = stats_count * DPDK_STATS_CTX_GET_XSTAT_SIZE;
+ ctx->stats_count = stats_count;
+
+ if ((dpdk_helper_data_size_get(phc) - sizeof(dpdk_stats_ctx_t)) <
+ stats_size) {
+ DPDK_CHILD_LOG(
+ "%s:%s:%d not enough space for stats (available=%d, "
+ "needed=%d)\n",
+ DPDK_STATS_PLUGIN, __FUNCTION__, __LINE__,
+ (int)(dpdk_helper_data_size_get(phc) - sizeof(dpdk_stats_ctx_t)),
+ stats_size);
+ return -ENOBUFS;
}
+ }
- if (nb_ports > RTE_MAX_ETHPORTS)
- nb_ports = RTE_MAX_ETHPORTS;
+ ret = dpdk_helper_stats_get(phc);
- int len = 0, enabled_port_count = 0, num_xstats = 0;
- for (uint8_t i = 0; i < nb_ports; i++) {
- if (!(g_configuration->enabled_port_mask & (1 << i)))
- continue;
+ return ret;
+}
- if (g_configuration->helper_action == DPDK_HELPER_ACTION_COUNT_STATS) {
-#if RTE_VERSION >= RTE_VERSION_16_07
- len = rte_eth_xstats_get_names(i, NULL, 0);
-#else
- len = rte_eth_xstats_get(i, NULL, 0);
-#endif
- if (len < 0) {
- ERROR("dpdkstat-helper: Cannot get xstats count on port %" PRIu8, i);
- break;
- }
- num_xstats += len;
- g_configuration->num_stats_in_port[enabled_port_count] = len;
- enabled_port_count++;
- continue;
- } else {
- len = g_configuration->num_stats_in_port[enabled_port_count];
- g_configuration->port_read_time[enabled_port_count] = cdtime();
- ret = rte_eth_xstats_get(
- i, g_configuration->xstats + num_xstats,
- g_configuration->num_stats_in_port[enabled_port_count]);
- if (ret < 0 || ret != len) {
- DEBUG("dpdkstat-helper: Error reading xstats on port %" PRIu8
- " len = %d",
- i, len);
- break;
- }
-#if RTE_VERSION >= RTE_VERSION_16_07
- ret = rte_eth_xstats_get_names(i, g_configuration->xnames + num_xstats,
- len);
- if (ret < 0 || ret != len) {
- ERROR("dpdkstat-helper: Error reading xstat names (port=%d; len=%d)",
- i, len);
- break;
- }
-#endif
- num_xstats += g_configuration->num_stats_in_port[enabled_port_count];
- enabled_port_count++;
- }
- } /* for (nb_ports) */
-
- if (g_configuration->helper_action == DPDK_HELPER_ACTION_COUNT_STATS) {
- g_configuration->num_ports = enabled_port_count;
- g_configuration->num_xstats = num_xstats;
- DEBUG("dpdkstat-helper ports: %" PRIu32 ", num stats: %" PRIu32,
- g_configuration->num_ports, g_configuration->num_xstats);
- /* Exit, allowing collectd to re-init SHM to the right size */
- g_configuration->collectd_reinit_shm = REINIT_SHM;
- dpdk_helper_stop(NO_RESET);
+static void dpdk_stats_resolve_cnt_type(char *cnt_type, size_t cnt_type_len,
+ const char *cnt_name) {
+ char *type_end;
+ type_end = strrchr(cnt_name, '_');
+
+ if ((type_end != NULL) && (strncmp(cnt_name, "rx_", strlen("rx_")) == 0)) {
+ if (strncmp(type_end, "_errors", strlen("_errors")) == 0) {
+ sstrncpy(cnt_type, "if_rx_errors", cnt_type_len);
+ } else if (strncmp(type_end, "_dropped", strlen("_dropped")) == 0) {
+ sstrncpy(cnt_type, "if_rx_dropped", cnt_type_len);
+ } else if (strncmp(type_end, "_bytes", strlen("_bytes")) == 0) {
+ sstrncpy(cnt_type, "if_rx_octets", cnt_type_len);
+ } else if (strncmp(type_end, "_packets", strlen("_packets")) == 0) {
+ sstrncpy(cnt_type, "if_rx_packets", cnt_type_len);
+ } else if (strncmp(type_end, "_placement", strlen("_placement")) == 0) {
+ sstrncpy(cnt_type, "if_rx_errors", cnt_type_len);
+ } else if (strncmp(type_end, "_buff", strlen("_buff")) == 0) {
+ sstrncpy(cnt_type, "if_rx_errors", cnt_type_len);
+ } else {
+ /* Does not fit obvious type: use a more generic one */
+ sstrncpy(cnt_type, "derive", cnt_type_len);
+ }
+
+ } else if ((type_end != NULL) &&
+ (strncmp(cnt_name, "tx_", strlen("tx_"))) == 0) {
+ if (strncmp(type_end, "_errors", strlen("_errors")) == 0) {
+ sstrncpy(cnt_type, "if_tx_errors", cnt_type_len);
+ } else if (strncmp(type_end, "_dropped", strlen("_dropped")) == 0) {
+ sstrncpy(cnt_type, "if_tx_dropped", cnt_type_len);
+ } else if (strncmp(type_end, "_bytes", strlen("_bytes")) == 0) {
+ sstrncpy(cnt_type, "if_tx_octets", cnt_type_len);
+ } else if (strncmp(type_end, "_packets", strlen("_packets")) == 0) {
+ sstrncpy(cnt_type, "if_tx_packets", cnt_type_len);
+ } else {
+ /* Does not fit obvious type: use a more generic one */
+ sstrncpy(cnt_type, "derive", cnt_type_len);
}
- /* Now kick collectd send thread to send the stats */
- int err = sem_post(&g_configuration->sema_stats_in_shm);
- if (err) {
- WARNING("dpdkstat: error posting semaphore to helper %s",
- sstrerror(errno, errbuf, sizeof(errbuf)));
- dpdk_helper_stop(RESET);
+ } else if ((type_end != NULL) &&
+ (strncmp(cnt_name, "flow_", strlen("flow_"))) == 0) {
+
+ if (strncmp(type_end, "_filters", strlen("_filters")) == 0) {
+ sstrncpy(cnt_type, "operations", cnt_type_len);
+ } else if (strncmp(type_end, "_errors", strlen("_errors")) == 0) {
+ sstrncpy(cnt_type, "errors", cnt_type_len);
+ } else if (strncmp(type_end, "_filters", strlen("_filters")) == 0) {
+ sstrncpy(cnt_type, "filter_result", cnt_type_len);
}
- } /* while(1) */
+ } else if ((type_end != NULL) &&
+ (strncmp(cnt_name, "mac_", strlen("mac_"))) == 0) {
+ if (strncmp(type_end, "_errors", strlen("_errors")) == 0) {
+ sstrncpy(cnt_type, "errors", cnt_type_len);
+ }
+ } else {
+ /* Does not fit obvious type, or strrchr error:
+ * use a more generic type */
+ sstrncpy(cnt_type, "derive", cnt_type_len);
+ }
+}
- return 0;
+static void dpdk_stats_counter_submit(const char *plugin_instance,
+ const char *cnt_name, derive_t value,
+ cdtime_t port_read_time) {
+ value_list_t vl = VALUE_LIST_INIT;
+ vl.values = &(value_t){.derive = value};
+ vl.values_len = 1;
+ vl.time = port_read_time;
+ sstrncpy(vl.host, hostname_g, sizeof(vl.host));
+ sstrncpy(vl.plugin, DPDK_STATS_PLUGIN, sizeof(vl.plugin));
+ sstrncpy(vl.plugin_instance, plugin_instance, sizeof(vl.plugin_instance));
+ dpdk_stats_resolve_cnt_type(vl.type, sizeof(vl.type), cnt_name);
+ sstrncpy(vl.type_instance, cnt_name, sizeof(vl.type_instance));
+ plugin_dispatch_values(&vl);
}
-static void dpdk_submit_xstats(const char *dev_name, int count,
- uint32_t counters, cdtime_t port_read_time) {
- for (uint32_t j = 0; j < counters; j++) {
- value_list_t vl = VALUE_LIST_INIT;
- char *counter_name;
- char *type_end;
-
- vl.values = &(value_t){.derive = (derive_t)DPDK_STATS_XSTAT_GET_VALUE(
- g_configuration, count + j)};
- vl.values_len = 1; /* Submit stats one at a time */
- vl.time = port_read_time;
- sstrncpy(vl.plugin, "dpdkstat", sizeof(vl.plugin));
- sstrncpy(vl.plugin_instance, dev_name, sizeof(vl.plugin_instance));
- counter_name = DPDK_STATS_XSTAT_GET_NAME(g_configuration, count + j);
- if (counter_name == NULL) {
- WARNING("dpdkstat: Failed to get counter name.");
- return;
- }
+static int dpdk_stats_counters_dispatch(dpdk_helper_ctx_t *phc) {
+ dpdk_stats_ctx_t *ctx = DPDK_STATS_CTX_GET(phc);
+
+ /* dispatch stats values to collectd */
+
+ DEBUG("%s:%s:%d ports=%u", DPDK_STATS_PLUGIN, __FUNCTION__, __LINE__,
+ ctx->ports_count);
- type_end = strrchr(counter_name, '_');
-
- if ((type_end != NULL) &&
- (strncmp(counter_name, "rx_", strlen("rx_")) == 0)) {
- if (strncmp(type_end, "_errors", strlen("_errors")) == 0) {
- sstrncpy(vl.type, "if_rx_errors", sizeof(vl.type));
- } else if (strncmp(type_end, "_dropped", strlen("_dropped")) == 0) {
- sstrncpy(vl.type, "if_rx_dropped", sizeof(vl.type));
- } else if (strncmp(type_end, "_bytes", strlen("_bytes")) == 0) {
- sstrncpy(vl.type, "if_rx_octets", sizeof(vl.type));
- } else if (strncmp(type_end, "_packets", strlen("_packets")) == 0) {
- sstrncpy(vl.type, "if_rx_packets", sizeof(vl.type));
- } else if (strncmp(type_end, "_placement", strlen("_placement")) == 0) {
- sstrncpy(vl.type, "if_rx_errors", sizeof(vl.type));
- } else if (strncmp(type_end, "_buff", strlen("_buff")) == 0) {
- sstrncpy(vl.type, "if_rx_errors", sizeof(vl.type));
- } else {
- /* Does not fit obvious type: use a more generic one */
- sstrncpy(vl.type, "derive", sizeof(vl.type));
- }
-
- } else if ((type_end != NULL) &&
- (strncmp(counter_name, "tx_", strlen("tx_"))) == 0) {
- if (strncmp(type_end, "_errors", strlen("_errors")) == 0) {
- sstrncpy(vl.type, "if_tx_errors", sizeof(vl.type));
- } else if (strncmp(type_end, "_dropped", strlen("_dropped")) == 0) {
- sstrncpy(vl.type, "if_tx_dropped", sizeof(vl.type));
- } else if (strncmp(type_end, "_bytes", strlen("_bytes")) == 0) {
- sstrncpy(vl.type, "if_tx_octets", sizeof(vl.type));
- } else if (strncmp(type_end, "_packets", strlen("_packets")) == 0) {
- sstrncpy(vl.type, "if_tx_packets", sizeof(vl.type));
- } else {
- /* Does not fit obvious type: use a more generic one */
- sstrncpy(vl.type, "derive", sizeof(vl.type));
- }
- } else if ((type_end != NULL) &&
- (strncmp(counter_name, "flow_", strlen("flow_"))) == 0) {
-
- if (strncmp(type_end, "_filters", strlen("_filters")) == 0) {
- sstrncpy(vl.type, "operations", sizeof(vl.type));
- } else if (strncmp(type_end, "_errors", strlen("_errors")) == 0) {
- sstrncpy(vl.type, "errors", sizeof(vl.type));
- } else if (strncmp(type_end, "_filters", strlen("_filters")) == 0) {
- sstrncpy(vl.type, "filter_result", sizeof(vl.type));
- }
- } else if ((type_end != NULL) &&
- (strncmp(counter_name, "mac_", strlen("mac_"))) == 0) {
- if (strncmp(type_end, "_errors", strlen("_errors")) == 0) {
- sstrncpy(vl.type, "errors", sizeof(vl.type));
- }
+ int stats_count = 0;
+
+ for (int i = 0; i < ctx->ports_count; i++) {
+ if (!(ctx->config.enabled_port_mask & (1 << i)))
+ continue;
+
+ char dev_name[64];
+ if (ctx->config.port_name[i][0] != 0) {
+ ssnprintf(dev_name, sizeof(dev_name), "%s", ctx->config.port_name[i]);
} else {
- /* Does not fit obvious type, or strrchr error:
- * use a more generic type */
- sstrncpy(vl.type, "derive", sizeof(vl.type));
+ ssnprintf(dev_name, sizeof(dev_name), "port.%d", i);
}
- sstrncpy(vl.type_instance, counter_name, sizeof(vl.type_instance));
- plugin_dispatch_values(&vl);
+ DEBUG(" === Dispatch stats for port %d (name=%s; stats_count=%d)", i,
+ dev_name, ctx->port_stats_count[i]);
+
+ for (int j = 0; j < ctx->port_stats_count[i]; j++) {
+ const char *cnt_name = DPDK_STATS_XSTAT_GET_NAME(ctx, stats_count);
+ if (cnt_name == NULL)
+ WARNING("dpdkstat: Invalid counter name");
+ else
+ dpdk_stats_counter_submit(
+ dev_name, cnt_name,
+ (derive_t)DPDK_STATS_XSTAT_GET_VALUE(ctx, stats_count),
+ ctx->port_read_time[i]);
+ stats_count++;
+
+ assert(stats_count <= ctx->stats_count);
+ }
}
+
+ return 0;
}
-static int dpdk_read(user_data_t *ud) {
- int ret = 0;
+static int dpdk_stats_reinit_helper() {
+ DPDK_STATS_TRACE();
- /*
- * Check if SHM flag is set to be re-initialized. AKA DPDK ports have been
- * counted, so re-init SHM to be large enough to fit all the statistics.
- */
- if (g_configuration->collectd_reinit_shm) {
- DEBUG("dpdkstat: read() now reinit SHM then launching send-thread");
- dpdk_re_init_shm();
- }
+ dpdk_stats_ctx_t *ctx = DPDK_STATS_CTX_GET(g_hc);
- /*
- * Check if DPDK proc is alive, and has already counted port / stats. This
- * must be done in dpdk_read(), because the DPDK primary process may not be
- * alive at dpdk_init() time.
- */
- if (g_configuration->helper_status == DPDK_HELPER_NOT_INITIALIZED ||
- g_configuration->helper_status == DPDK_HELPER_GRACEFUL_QUIT) {
- int action = DPDK_HELPER_ACTION_SEND_STATS;
- if (g_configuration->num_xstats == 0)
- action = DPDK_HELPER_ACTION_COUNT_STATS;
- /* Spawn the helper thread to count stats or to read stats. */
- int err = dpdk_helper_spawn(action);
- if (err) {
- char errbuf[ERR_BUF_SIZE];
- ERROR("dpdkstat: error spawning helper %s",
- sstrerror(errno, errbuf, sizeof(errbuf)));
- return -1;
- }
- }
+ size_t data_size = sizeof(dpdk_stats_ctx_t) +
+ (ctx->stats_count * DPDK_STATS_CTX_GET_XSTAT_SIZE);
- pid_t ws = waitpid(g_configuration->helper_pid, NULL, WNOHANG);
- /*
- * Conditions under which to respawn helper:
- * waitpid() fails, helper process died (or quit), so respawn
- */
- _Bool respawn_helper = 0;
- if (ws != 0) {
- respawn_helper = 1;
- }
+ DEBUG("%s:%d helper reinit (new_size=%zu)", __FUNCTION__, __LINE__,
+ data_size);
- char buf[DPDKSTAT_MAX_BUFFER_SIZE];
- char out[DPDKSTAT_MAX_BUFFER_SIZE];
+ dpdk_stats_ctx_t tmp_ctx;
+ dpdk_eal_config_t tmp_eal;
- /* non blocking check on helper logging pipe */
- struct pollfd fds = {
- .fd = g_configuration->helper_pipes[0], .events = POLLIN,
- };
- int data_avail = poll(&fds, 1, 0);
- if (data_avail < 0) {
+ memcpy(&tmp_ctx, ctx, sizeof(dpdk_stats_ctx_t));
+ dpdk_helper_eal_config_get(g_hc, &tmp_eal);
+
+ dpdk_helper_shutdown(g_hc);
+
+ g_hc = NULL;
+
+ int ret;
+ ret = dpdk_helper_init(DPDK_STATS_NAME, data_size, &g_hc);
+ if (ret != 0) {
char errbuf[ERR_BUF_SIZE];
- if (errno != EINTR || errno != EAGAIN)
- ERROR("dpdkstats: poll(2) failed: %s",
- sstrerror(errno, errbuf, sizeof(errbuf)));
- }
- while (data_avail) {
- int nbytes = read(g_configuration->helper_pipes[0], buf, sizeof(buf));
- if (nbytes <= 0)
- break;
- ssnprintf(out, nbytes, "%s", buf);
- DEBUG("dpdkstat: helper-proc: %s", out);
+ ERROR("%s: failed to initialize %s helper(error: %s)", DPDK_STATS_PLUGIN,
+ DPDK_STATS_NAME, sstrerror(errno, errbuf, sizeof(errbuf)));
+ return ret;
}
- if (respawn_helper) {
- if (g_configuration->helper_pid)
- dpdk_helper_stop(RESET);
- dpdk_helper_spawn(DPDK_HELPER_ACTION_COUNT_STATS);
+ ctx = DPDK_STATS_CTX_GET(g_hc);
+ memcpy(ctx, &tmp_ctx, sizeof(dpdk_stats_ctx_t));
+ DPDK_STATS_CTX_INIT(ctx);
+ dpdk_helper_eal_config_set(g_hc, &tmp_eal);
+
+ return ret;
+}
+
+static int dpdk_stats_read(user_data_t *ud) {
+ DPDK_STATS_TRACE();
+
+ int ret = 0;
+
+ if (g_hc == NULL) {
+ ERROR("dpdk stats plugin not initialized");
+ return -EINVAL;
}
- /* Kick helper process through SHM */
- sem_post(&g_configuration->sema_helper_get_stats);
+ dpdk_stats_ctx_t *ctx = DPDK_STATS_CTX_GET(g_hc);
- cdtime_t now = cdtime();
- ret = sem_timedwait(&g_configuration->sema_stats_in_shm,
- &CDTIME_T_TO_TIMESPEC(now + g_configuration->interval));
- if (ret == -1) {
- if (errno == ETIMEDOUT)
- DEBUG(
- "dpdkstat: timeout in collectd thread: is a DPDK Primary running? ");
+ int result = 0;
+ ret = dpdk_helper_command(g_hc, DPDK_CMD_GET_STATS, &result,
+ ctx->config.interval);
+ if (ret != 0) {
return 0;
}
- /* Dispatch the stats.*/
- uint32_t count = 0, port_num = 0;
-
- for (uint32_t i = 0; i < g_configuration->num_ports; i++) {
- char dev_name[64];
- cdtime_t port_read_time = g_configuration->port_read_time[i];
- uint32_t counters_num = g_configuration->num_stats_in_port[i];
- size_t ports_max = CHAR_BIT * sizeof(g_configuration->enabled_port_mask);
- for (size_t j = port_num; j < ports_max; j++) {
- if ((g_configuration->enabled_port_mask & (1 << j)) != 0)
- break;
- port_num++;
- }
+ if (result == -ENOBUFS) {
+ dpdk_stats_reinit_helper();
+ } else if (result == -ENODEV) {
+ dpdk_helper_shutdown(g_hc);
+ } else if (result == 0) {
+ dpdk_stats_counters_dispatch(g_hc);
+ }
- if (g_configuration->port_name[i][0] != 0)
- ssnprintf(dev_name, sizeof(dev_name), "%s",
- g_configuration->port_name[i]);
- else
- ssnprintf(dev_name, sizeof(dev_name), "port.%" PRIu32, port_num);
- dpdk_submit_xstats(dev_name, count, counters_num, port_read_time);
- count += counters_num;
- port_num++;
- } /* for each port */
return 0;
}
-static int dpdk_shm_cleanup(void) {
- int ret = munmap(g_configuration, sizeof(dpdk_config_t));
- g_configuration = 0;
- if (ret) {
- ERROR("dpdkstat: munmap returned %d", ret);
- return ret;
- }
- ret = shm_unlink(DPDK_SHM_NAME);
- if (ret) {
- ERROR("dpdkstat: shm_unlink returned %d", ret);
+static int dpdk_stats_init(void) {
+ DPDK_STATS_TRACE();
+
+ int ret = 0;
+
+ ret = dpdk_stats_preinit();
+ if (ret != 0) {
return ret;
}
+
return 0;
}
-static int dpdk_shutdown(void) {
+static int dpdk_stats_shutdown(void) {
+ DPDK_STATS_TRACE();
+
int ret = 0;
- char errbuf[ERR_BUF_SIZE];
- close(g_configuration->helper_pipes[1]);
- int err = kill(g_configuration->helper_pid, SIGKILL);
- if (err) {
- ERROR("dpdkstat: error sending sigkill to helper %s",
- sstrerror(errno, errbuf, sizeof(errbuf)));
- ret = -1;
- }
- err = dpdk_shm_cleanup();
- if (err) {
- ERROR("dpdkstat: error cleaning up SHM: %s",
- sstrerror(errno, errbuf, sizeof(errbuf)));
- ret = -1;
+
+ ret = dpdk_helper_shutdown(g_hc);
+ g_hc = NULL;
+ if (ret != 0) {
+ ERROR("%s: failed to cleanup %s helper", DPDK_STATS_PLUGIN,
+ DPDK_STATS_NAME);
+ return ret;
}
return ret;
}
void module_register(void) {
- plugin_register_complex_config("dpdkstat", dpdk_config);
- plugin_register_init("dpdkstat", dpdk_init);
- plugin_register_complex_read(NULL, "dpdkstat", dpdk_read, 0, NULL);
- plugin_register_shutdown("dpdkstat", dpdk_shutdown);
+ plugin_register_init(DPDK_STATS_PLUGIN, dpdk_stats_init);
+ plugin_register_complex_config(DPDK_STATS_PLUGIN, dpdk_stats_config);
+ plugin_register_complex_read(NULL, DPDK_STATS_PLUGIN, dpdk_stats_read, 0,
+ NULL);
+ plugin_register_shutdown(DPDK_STATS_PLUGIN, dpdk_stats_shutdown);
}
--- /dev/null
+/**
+ * collectd - src/utils_dpdk.c
+ * MIT License
+ *
+ * Copyright(c) 2016 Intel Corporation. All rights reserved.
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ * and/or sell copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ *
+ * Authors:
+ * Maryam Tahhan <maryam.tahhan@intel.com>
+ * Harry van Haaren <harry.van.haaren@intel.com>
+ * Serhiy Pshyk <serhiyx.pshyk@intel.com>
+ **/
+
+#include "collectd.h"
+
+#include <poll.h>
+#include <semaphore.h>
+#include <sys/mman.h>
+
+#include <rte_config.h>
+#include <rte_eal.h>
+
+#include "common.h"
+#include "utils_dpdk.h"
+
+#define DPDK_DEFAULT_RTE_CONFIG "/var/run/.rte_config"
+#define DPDK_EAL_ARGC 5
+#define DPDK_MAX_BUFFER_SIZE (4096 * 4)
+#define DPDK_CDM_DEFAULT_TIMEOUT 10000
+
+enum DPDK_HELPER_STATUS {
+ DPDK_HELPER_NOT_INITIALIZED = 0,
+ DPDK_HELPER_INITIALIZING,
+ DPDK_HELPER_WAITING_ON_PRIMARY,
+ DPDK_HELPER_INITIALIZING_EAL,
+ DPDK_HELPER_ALIVE_SENDING_EVENTS,
+ DPDK_HELPER_GRACEFUL_QUIT,
+};
+
+#define DPDK_HELPER_TRACE(_name) \
+ DEBUG("%s:%s:%d pid=%lu", _name, __FUNCTION__, __LINE__, (long)getpid())
+
+#define DPDK_HELPER_USE_PIPES
+
+struct dpdk_helper_ctx_s {
+
+ dpdk_eal_config_t eal_config;
+ int eal_initialized;
+
+ size_t shm_size;
+ const char *shm_name;
+
+ sem_t sema_cmd_start;
+ sem_t sema_cmd_complete;
+ cdtime_t cmd_wait_time;
+
+ pid_t pid;
+#ifdef DPDK_HELPER_USE_PIPES
+ int pipes[2];
+#endif /* DPDK_HELPER_USE_PIPES */
+ int status;
+
+ int cmd;
+ int cmd_result;
+
+ char priv_data[];
+};
+
+static int dpdk_shm_init(const char *name, size_t size, void **map);
+static int dpdk_shm_cleanup(const char *name, size_t size, void *map);
+
+static int dpdk_helper_spawn(dpdk_helper_ctx_t *phc);
+static int dpdk_helper_worker(dpdk_helper_ctx_t *phc);
+static int dpdk_helper_eal_init(dpdk_helper_ctx_t *phc);
+static int dpdk_helper_cmd_wait(dpdk_helper_ctx_t *phc, pid_t ppid);
+static int dpdk_helper_exit_command(dpdk_helper_ctx_t *phc,
+ enum DPDK_HELPER_STATUS status);
+static int dpdk_helper_exit(dpdk_helper_ctx_t *phc,
+ enum DPDK_HELPER_STATUS status);
+static int dpdk_helper_status_check(dpdk_helper_ctx_t *phc);
+static void dpdk_helper_config_default(dpdk_helper_ctx_t *phc);
+static const char *dpdk_helper_status_str(enum DPDK_HELPER_STATUS status);
+
+static void dpdk_helper_config_default(dpdk_helper_ctx_t *phc) {
+ if (phc == NULL)
+ return;
+
+ DPDK_HELPER_TRACE(phc->shm_name);
+
+ ssnprintf(phc->eal_config.coremask, DATA_MAX_NAME_LEN, "%s", "0xf");
+ ssnprintf(phc->eal_config.memory_channels, DATA_MAX_NAME_LEN, "%s", "1");
+ ssnprintf(phc->eal_config.process_type, DATA_MAX_NAME_LEN, "%s", "secondary");
+ ssnprintf(phc->eal_config.file_prefix, DATA_MAX_NAME_LEN, "%s",
+ DPDK_DEFAULT_RTE_CONFIG);
+}
+
+int dpdk_helper_eal_config_set(dpdk_helper_ctx_t *phc, dpdk_eal_config_t *ec) {
+ if (phc == NULL) {
+ ERROR("Invalid argument (phc)");
+ return -EINVAL;
+ }
+
+ DPDK_HELPER_TRACE(phc->shm_name);
+
+ if (ec == NULL) {
+ ERROR("Invalid argument (ec)");
+ return -EINVAL;
+ }
+
+ memcpy(&phc->eal_config, ec, sizeof(dpdk_eal_config_t));
+
+ return 0;
+}
+
+int dpdk_helper_eal_config_get(dpdk_helper_ctx_t *phc, dpdk_eal_config_t *ec) {
+ if (phc == NULL) {
+ ERROR("Invalid argument (phc)");
+ return -EINVAL;
+ }
+
+ DPDK_HELPER_TRACE(phc->shm_name);
+
+ if (ec == NULL) {
+ ERROR("Invalid argument (ec)");
+ return -EINVAL;
+ }
+
+ memcpy(ec, &phc->eal_config, sizeof(dpdk_eal_config_t));
+
+ return 0;
+}
+
+int dpdk_helper_eal_config_parse(dpdk_helper_ctx_t *phc, oconfig_item_t *ci) {
+ DPDK_HELPER_TRACE(phc->shm_name);
+
+ if (phc == NULL) {
+ ERROR("Invalid argument (phc)");
+ return -EINVAL;
+ }
+
+ if (ci == NULL) {
+ ERROR("Invalid argument (ci)");
+ return -EINVAL;
+ }
+
+ for (int i = 0; i < ci->children_num; i++) {
+ oconfig_item_t *child = ci->children + i;
+ if (strcasecmp("Coremask", child->key) == 0) {
+ cf_util_get_string_buffer(child, phc->eal_config.coremask,
+ sizeof(phc->eal_config.coremask));
+ DEBUG("dpdk_common: EAL:Coremask %s", phc->eal_config.coremask);
+ } else if (strcasecmp("MemoryChannels", child->key) == 0) {
+ cf_util_get_string_buffer(child, phc->eal_config.memory_channels,
+ sizeof(phc->eal_config.memory_channels));
+ DEBUG("dpdk_common: EAL:Memory Channels %s",
+ phc->eal_config.memory_channels);
+ } else if (strcasecmp("SocketMemory", child->key) == 0) {
+ cf_util_get_string_buffer(child, phc->eal_config.socket_memory,
+ sizeof(phc->eal_config.socket_memory));
+ DEBUG("dpdk_common: EAL:Socket memory %s", phc->eal_config.socket_memory);
+ } else if (strcasecmp("ProcessType", child->key) == 0) {
+ cf_util_get_string_buffer(child, phc->eal_config.process_type,
+ sizeof(phc->eal_config.process_type));
+ DEBUG("dpdk_common: EAL:Process type %s", phc->eal_config.process_type);
+ } else if ((strcasecmp("FilePrefix", child->key) == 0) &&
+ (child->values[0].type == OCONFIG_TYPE_STRING)) {
+ ssnprintf(phc->eal_config.file_prefix, DATA_MAX_NAME_LEN,
+ "/var/run/.%s_config", child->values[0].value.string);
+ DEBUG("dpdk_common: EAL:File prefix %s", phc->eal_config.file_prefix);
+ }
+ }
+
+ return 0;
+}
+
+static int dpdk_shm_init(const char *name, size_t size, void **map) {
+ DPDK_HELPER_TRACE(name);
+
+ char errbuf[ERR_BUF_SIZE];
+
+ int fd = shm_open(name, O_CREAT | O_TRUNC | O_RDWR, 0666);
+ if (fd < 0) {
+ WARNING("dpdk_shm_init: Failed to open %s as SHM:%s\n", name,
+ sstrerror(errno, errbuf, sizeof(errbuf)));
+ goto fail;
+ }
+
+ int ret = ftruncate(fd, size);
+ if (ret != 0) {
+ WARNING("dpdk_shm_init: Failed to resize SHM:%s\n",
+ sstrerror(errno, errbuf, sizeof(errbuf)));
+ goto fail_close;
+ }
+
+ *map = mmap(0, size, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
+ if (*map == MAP_FAILED) {
+ WARNING("dpdk_shm_init:Failed to mmap SHM:%s\n",
+ sstrerror(errno, errbuf, sizeof(errbuf)));
+ goto fail_close;
+ }
+ /*
+ * Close the file descriptor, the shared memory object still exists
+ * and can only be removed by calling shm_unlink().
+ */
+ close(fd);
+
+ memset(*map, 0, size);
+
+ return 0;
+
+fail_close:
+ close(fd);
+fail:
+ *map = NULL;
+ return -1;
+}
+
+static int dpdk_shm_cleanup(const char *name, size_t size, void *map) {
+ DPDK_HELPER_TRACE(name);
+
+ int ret = munmap(map, size);
+ if (ret) {
+ ERROR("munmap returned %d\n", ret);
+ }
+
+ ret = shm_unlink(name);
+ if (ret) {
+ ERROR("shm_unlink returned %d\n", ret);
+ }
+
+ return 0;
+}
+
+inline void *dpdk_helper_priv_get(dpdk_helper_ctx_t *phc) {
+ if (phc)
+ return (void *)phc->priv_data;
+
+ return NULL;
+}
+
+int dpdk_helper_data_size_get(dpdk_helper_ctx_t *phc) {
+ if (phc == NULL) {
+ DPDK_CHILD_LOG("Invalid argument(phc)\n");
+ return -EINVAL;
+ }
+
+ return (phc->shm_size - sizeof(dpdk_helper_ctx_t));
+}
+
+int dpdk_helper_init(const char *name, size_t data_size,
+ dpdk_helper_ctx_t **pphc) {
+ int err = 0;
+ dpdk_helper_ctx_t *phc = NULL;
+ size_t shm_size = sizeof(dpdk_helper_ctx_t) + data_size;
+ char errbuf[ERR_BUF_SIZE];
+
+ if (pphc == NULL) {
+ ERROR("%s:Invalid argument(pphc)", __FUNCTION__);
+ return -EINVAL;
+ }
+
+ if (name == NULL) {
+ ERROR("%s:Invalid argument(name)", __FUNCTION__);
+ return -EINVAL;
+ }
+
+ DPDK_HELPER_TRACE(name);
+
+ /* Allocate dpdk_helper_ctx_t and
+ * initialize a POSIX SHared Memory (SHM) object.
+ */
+ err = dpdk_shm_init(name, shm_size, (void **)&phc);
+ if (err != 0) {
+ return -errno;
+ }
+
+ err = sem_init(&phc->sema_cmd_start, 1, 0);
+ if (err != 0) {
+ ERROR("sema_cmd_start semaphore init failed: %s\n",
+ sstrerror(errno, errbuf, sizeof(errbuf)));
+ dpdk_shm_cleanup(name, shm_size, (void *)phc);
+ return -errno;
+ }
+
+ err = sem_init(&phc->sema_cmd_complete, 1, 0);
+ if (err != 0) {
+ ERROR("sema_cmd_complete semaphore init failed: %s\n",
+ sstrerror(errno, errbuf, sizeof(errbuf)));
+ sem_destroy(&phc->sema_cmd_start);
+ dpdk_shm_cleanup(name, shm_size, (void *)phc);
+ return -errno;
+ }
+
+ phc->shm_size = shm_size;
+ phc->shm_name = name;
+
+ dpdk_helper_config_default(phc);
+
+ *pphc = phc;
+
+ return 0;
+}
+
+int dpdk_helper_shutdown(dpdk_helper_ctx_t *phc) {
+ if (phc == NULL) {
+ ERROR("%s:Invalid argument(phc)", __FUNCTION__);
+ return -EINVAL;
+ }
+
+ DPDK_HELPER_TRACE(phc->shm_name);
+
+#ifdef DPDK_HELPER_USE_PIPES
+ close(phc->pipes[1]);
+#endif
+
+ if (phc->status != DPDK_HELPER_NOT_INITIALIZED) {
+ dpdk_helper_exit_command(phc, DPDK_HELPER_GRACEFUL_QUIT);
+ }
+
+ sem_destroy(&phc->sema_cmd_start);
+ sem_destroy(&phc->sema_cmd_complete);
+ dpdk_shm_cleanup(phc->shm_name, phc->shm_size, (void *)phc);
+
+ return 0;
+}
+
+static int dpdk_helper_spawn(dpdk_helper_ctx_t *phc) {
+ char errbuf[ERR_BUF_SIZE];
+ if (phc == NULL) {
+ ERROR("Invalid argument(phc)");
+ return -EINVAL;
+ }
+
+ DPDK_HELPER_TRACE(phc->shm_name);
+
+ phc->eal_initialized = 0;
+ phc->cmd_wait_time = MS_TO_CDTIME_T(DPDK_CDM_DEFAULT_TIMEOUT);
+
+#ifdef DPDK_HELPER_USE_PIPES
+ /*
+ * Create a pipe for helper stdout back to collectd. This is necessary for
+ * logging EAL failures, as rte_eal_init() calls rte_panic().
+ */
+ if (phc->pipes[1]) {
+ DEBUG("dpdk_helper_spawn: collectd closing helper pipe %d", phc->pipes[1]);
+ } else {
+ DEBUG("dpdk_helper_spawn: collectd helper pipe %d, not closing",
+ phc->pipes[1]);
+ }
+
+ if (pipe(phc->pipes) != 0) {
+ DEBUG("dpdk_helper_spawn: Could not create helper pipe: %s\n",
+ sstrerror(errno, errbuf, sizeof(errbuf)));
+ return -1;
+ }
+
+ int pipe0_flags = fcntl(phc->pipes[0], F_GETFL, 0);
+ int pipe1_flags = fcntl(phc->pipes[1], F_GETFL, 0);
+ if (pipe0_flags == -1 || pipe1_flags == -1) {
+ WARNING("dpdk_helper_spawn: error setting up pipe flags: %s",
+ sstrerror(errno, errbuf, sizeof(errbuf)));
+ }
+ int pipe0_err = fcntl(phc->pipes[0], F_SETFL, pipe1_flags | O_NONBLOCK);
+ int pipe1_err = fcntl(phc->pipes[1], F_SETFL, pipe0_flags | O_NONBLOCK);
+ if (pipe0_err == -1 || pipe1_err == -1) {
+ WARNING("dpdk_helper_spawn: error setting up pipes: %s",
+ sstrerror(errno, errbuf, sizeof(errbuf)));
+ }
+#endif /* DPDK_HELPER_USE_PIPES */
+
+ pid_t pid = fork();
+ if (pid > 0) {
+ phc->pid = pid;
+#ifdef DPDK_HELPER_USE_PIPES
+ close(phc->pipes[1]);
+#endif /* DPDK_HELPER_USE_PIPES */
+ DEBUG("%s:dpdk_helper_spawn: helper pid %lu", phc->shm_name,
+ (long)phc->pid);
+ } else if (pid == 0) {
+#ifdef DPDK_HELPER_USE_PIPES
+ /* Replace stdout with a pipe to collectd. */
+ close(phc->pipes[0]);
+ close(STDOUT_FILENO);
+ dup2(phc->pipes[1], STDOUT_FILENO);
+#endif /* DPDK_HELPER_USE_PIPES */
+ DPDK_CHILD_TRACE(phc->shm_name);
+ dpdk_helper_worker(phc);
+ exit(0);
+ } else {
+ ERROR("dpdk_helper_start: Failed to fork helper process: %s\n",
+ sstrerror(errno, errbuf, sizeof(errbuf)));
+ return -1;
+ }
+
+ return 0;
+}
+
+static int dpdk_helper_exit(dpdk_helper_ctx_t *phc,
+ enum DPDK_HELPER_STATUS status) {
+ DPDK_CHILD_LOG("%s:%s:%d %s\n", phc->shm_name, __FUNCTION__, __LINE__,
+ dpdk_helper_status_str(status));
+
+#ifdef DPDK_HELPER_USE_PIPES
+ close(phc->pipes[1]);
+#endif /* DPDK_HELPER_USE_PIPES */
+
+ phc->status = status;
+
+ exit(0);
+
+ return 0;
+}
+
+static int dpdk_helper_exit_command(dpdk_helper_ctx_t *phc,
+ enum DPDK_HELPER_STATUS status) {
+ char errbuf[ERR_BUF_SIZE];
+ DPDK_HELPER_TRACE(phc->shm_name);
+
+#ifdef DPDK_HELPER_USE_PIPES
+ close(phc->pipes[1]);
+#endif /* DPDK_HELPER_USE_PIPES */
+
+ if (phc->status == DPDK_HELPER_ALIVE_SENDING_EVENTS) {
+ phc->status = status;
+ DEBUG("%s:%s:%d %s", phc->shm_name, __FUNCTION__, __LINE__,
+ dpdk_helper_status_str(status));
+
+ int ret = dpdk_helper_command(phc, DPDK_CMD_QUIT, NULL, 0);
+ if (ret != 0) {
+ DEBUG("%s:%s:%d kill helper (pid=%lu)", phc->shm_name, __FUNCTION__,
+ __LINE__, (long)phc->pid);
+
+ int err = kill(phc->pid, SIGKILL);
+ if (err) {
+ ERROR("%s error sending kill to helper: %s\n", __FUNCTION__,
+ sstrerror(errno, errbuf, sizeof(errbuf)));
+ }
+ }
+ } else {
+
+ DEBUG("%s:%s:%d kill helper (pid=%lu)", phc->shm_name, __FUNCTION__,
+ __LINE__, (long)phc->pid);
+
+ int err = kill(phc->pid, SIGKILL);
+ if (err) {
+ ERROR("%s error sending kill to helper: %s\n", __FUNCTION__,
+ sstrerror(errno, errbuf, sizeof(errbuf)));
+ }
+ }
+
+ return 0;
+}
+
+static int dpdk_helper_eal_init(dpdk_helper_ctx_t *phc) {
+ phc->status = DPDK_HELPER_INITIALIZING_EAL;
+ DPDK_CHILD_LOG("%s:%s:%d DPDK_HELPER_INITIALIZING_EAL (start)\n",
+ phc->shm_name, __FUNCTION__, __LINE__);
+
+ char *argp[DPDK_EAL_ARGC * 2 + 1];
+ int argc = 0;
+
+ /* EAL config must be initialized */
+ assert(phc->eal_config.coremask[0] != 0);
+ assert(phc->eal_config.memory_channels[0] != 0);
+ assert(phc->eal_config.process_type[0] != 0);
+ assert(phc->eal_config.file_prefix[0] != 0);
+
+ argp[argc++] = "collectd-dpdk";
+
+ argp[argc++] = "-c";
+ argp[argc++] = phc->eal_config.coremask;
+
+ argp[argc++] = "-n";
+ argp[argc++] = phc->eal_config.memory_channels;
+
+ if (strcasecmp(phc->eal_config.socket_memory, "") != 0) {
+ argp[argc++] = "--socket-mem";
+ argp[argc++] = phc->eal_config.socket_memory;
+ }
+
+ if (strcasecmp(phc->eal_config.file_prefix, DPDK_DEFAULT_RTE_CONFIG) != 0) {
+ argp[argc++] = "--file-prefix";
+ argp[argc++] = phc->eal_config.file_prefix;
+ }
+
+ argp[argc++] = "--proc-type";
+ argp[argc++] = phc->eal_config.process_type;
+
+ assert(argc <= (DPDK_EAL_ARGC * 2 + 1));
+
+ int ret = rte_eal_init(argc, argp);
+
+ if (ret < 0) {
+
+ phc->eal_initialized = 0;
+
+ DPDK_CHILD_LOG("dpdk_helper_eal_init: ERROR initializing EAL ret=%d\n",
+ ret);
+
+ printf("dpdk_helper_eal_init: EAL arguments: ");
+ for (int i = 0; i < argc; i++) {
+ printf("%s ", argp[i]);
+ }
+ printf("\n");
+
+ return ret;
+ }
+
+ phc->eal_initialized = 1;
+
+ DPDK_CHILD_LOG("%s:%s:%d DPDK_HELPER_INITIALIZING_EAL (done)\n",
+ phc->shm_name, __FUNCTION__, __LINE__);
+
+ return 0;
+}
+
+static int dpdk_helper_cmd_wait(dpdk_helper_ctx_t *phc, pid_t ppid) {
+ DPDK_CHILD_TRACE(phc->shm_name);
+
+ struct timespec ts;
+ cdtime_t now = cdtime();
+ cdtime_t cmd_wait_time = MS_TO_CDTIME_T(1500) + phc->cmd_wait_time * 2;
+ ts = CDTIME_T_TO_TIMESPEC(now + cmd_wait_time);
+
+ int ret = sem_timedwait(&phc->sema_cmd_start, &ts);
+ DPDK_CHILD_LOG("%s:%s:%d pid=%lu got sema_cmd_start (ret=%d, errno=%d)\n",
+ phc->shm_name, __FUNCTION__, __LINE__, (long)getpid(), ret,
+ errno);
+
+ if (phc->cmd == DPDK_CMD_QUIT) {
+ DPDK_CHILD_LOG("%s:%s:%d pid=%lu exiting\n", phc->shm_name, __FUNCTION__,
+ __LINE__, (long)getpid());
+ exit(0);
+ } else if (ret == -1 && errno == ETIMEDOUT) {
+ if (phc->status == DPDK_HELPER_ALIVE_SENDING_EVENTS) {
+ DPDK_CHILD_LOG("%s:dpdk_helper_cmd_wait: sem timedwait()"
+ " timeout, did collectd terminate?\n",
+ phc->shm_name);
+ dpdk_helper_exit(phc, DPDK_HELPER_GRACEFUL_QUIT);
+ }
+ }
+#if COLLECT_DEBUG
+ int val = 0;
+ if (sem_getvalue(&phc->sema_cmd_start, &val) == 0)
+ DPDK_CHILD_LOG("%s:%s:%d pid=%lu wait sema_cmd_start (value=%d)\n",
+ phc->shm_name, __FUNCTION__, __LINE__, (long)getpid(), val);
+#endif
+
+ /* Parent PID change means collectd died so quit the helper process. */
+ if (ppid != getppid()) {
+ DPDK_CHILD_LOG("dpdk_helper_cmd_wait: parent PID changed, quitting.\n");
+ dpdk_helper_exit(phc, DPDK_HELPER_GRACEFUL_QUIT);
+ }
+
+ /* Checking for DPDK primary process. */
+ if (!rte_eal_primary_proc_alive(phc->eal_config.file_prefix)) {
+ if (phc->eal_initialized) {
+ DPDK_CHILD_LOG(
+ "%s:dpdk_helper_cmd_wait: no primary alive but EAL initialized:"
+ " quitting.\n",
+ phc->shm_name);
+ dpdk_helper_exit(phc, DPDK_HELPER_NOT_INITIALIZED);
+ }
+
+ phc->status = DPDK_HELPER_WAITING_ON_PRIMARY;
+ DPDK_CHILD_LOG("%s:%s:%d DPDK_HELPER_WAITING_ON_PRIMARY\n", phc->shm_name,
+ __FUNCTION__, __LINE__);
+
+ return -1;
+ }
+
+ if (!phc->eal_initialized) {
+ int ret = dpdk_helper_eal_init(phc);
+ if (ret != 0) {
+ DPDK_CHILD_LOG("Error initializing EAL\n");
+ dpdk_helper_exit(phc, DPDK_HELPER_NOT_INITIALIZED);
+ }
+ phc->status = DPDK_HELPER_ALIVE_SENDING_EVENTS;
+ DPDK_CHILD_LOG("%s:%s:%d DPDK_HELPER_ALIVE_SENDING_EVENTS\n", phc->shm_name,
+ __FUNCTION__, __LINE__);
+ return -1;
+ }
+
+ return 0;
+}
+
+static int dpdk_helper_worker(dpdk_helper_ctx_t *phc) {
+ DPDK_CHILD_TRACE(phc->shm_name);
+
+ pid_t ppid = getppid();
+
+ while (1) {
+ if (dpdk_helper_cmd_wait(phc, ppid) == 0) {
+ DPDK_CHILD_LOG("%s:%s:%d DPDK command handle (cmd=%d, pid=%lu)\n",
+ phc->shm_name, __FUNCTION__, __LINE__, phc->cmd,
+ (long)getpid());
+ phc->cmd_result = dpdk_helper_command_handler(phc, phc->cmd);
+ } else {
+ phc->cmd_result = -1;
+ }
+
+ /* now kick collectd to get results */
+ int err = sem_post(&phc->sema_cmd_complete);
+ DPDK_CHILD_LOG("%s:%s:%d post sema_cmd_complete (pid=%lu)\n", phc->shm_name,
+ __FUNCTION__, __LINE__, (long)getpid());
+ if (err) {
+ char errbuf[ERR_BUF_SIZE];
+ DPDK_CHILD_LOG("dpdk_helper_worker: error posting sema_cmd_complete "
+ "semaphore (%s)\n",
+ sstrerror(errno, errbuf, sizeof(errbuf)));
+ }
+
+#if COLLECT_DEBUG
+ int val = 0;
+ if (sem_getvalue(&phc->sema_cmd_complete, &val) == 0)
+ DPDK_CHILD_LOG("%s:%s:%d pid=%lu sema_cmd_complete (value=%d)\n",
+ phc->shm_name, __FUNCTION__, __LINE__, (long)getpid(),
+ val);
+#endif
+
+ } /* while(1) */
+
+ return 0;
+}
+
+static const char *dpdk_helper_status_str(enum DPDK_HELPER_STATUS status) {
+ switch (status) {
+ case DPDK_HELPER_ALIVE_SENDING_EVENTS:
+ return "DPDK_HELPER_ALIVE_SENDING_EVENTS";
+ case DPDK_HELPER_WAITING_ON_PRIMARY:
+ return "DPDK_HELPER_WAITING_ON_PRIMARY";
+ case DPDK_HELPER_INITIALIZING:
+ return "DPDK_HELPER_INITIALIZING";
+ case DPDK_HELPER_INITIALIZING_EAL:
+ return "DPDK_HELPER_INITIALIZING_EAL";
+ case DPDK_HELPER_GRACEFUL_QUIT:
+ return "DPDK_HELPER_GRACEFUL_QUIT";
+ case DPDK_HELPER_NOT_INITIALIZED:
+ return "DPDK_HELPER_NOT_INITIALIZED";
+ default:
+ return "UNKNOWN";
+ }
+}
+
+static int dpdk_helper_status_check(dpdk_helper_ctx_t *phc) {
+ DEBUG("%s:%s:%d pid=%u %s", phc->shm_name, __FUNCTION__, __LINE__, getpid(),
+ dpdk_helper_status_str(phc->status));
+ char errbuf[ERR_BUF_SIZE];
+
+ if (phc->status == DPDK_HELPER_GRACEFUL_QUIT) {
+ return 0;
+ } else if (phc->status == DPDK_HELPER_NOT_INITIALIZED) {
+ phc->status = DPDK_HELPER_INITIALIZING;
+ DEBUG("%s:%s:%d DPDK_HELPER_INITIALIZING", phc->shm_name, __FUNCTION__,
+ __LINE__);
+ int err = dpdk_helper_spawn(phc);
+ if (err) {
+ ERROR("dpdkstat: error spawning helper %s",
+ sstrerror(errno, errbuf, sizeof(errbuf)));
+ }
+ return -1;
+ }
+
+ pid_t ws = waitpid(phc->pid, NULL, WNOHANG);
+ if (ws != 0) {
+ phc->status = DPDK_HELPER_INITIALIZING;
+ DEBUG("%s:%s:%d DPDK_HELPER_INITIALIZING", phc->shm_name, __FUNCTION__,
+ __LINE__);
+ int err = dpdk_helper_spawn(phc);
+ if (err) {
+ ERROR("dpdkstat: error spawning helper %s",
+ sstrerror(errno, errbuf, sizeof(errbuf)));
+ }
+ return -1;
+ }
+
+ if (phc->status == DPDK_HELPER_INITIALIZING_EAL) {
+ return -1;
+ }
+
+ return 0;
+}
+
+#ifdef DPDK_HELPER_USE_PIPES
+static void dpdk_helper_check_pipe(dpdk_helper_ctx_t *phc) {
+ char buf[DPDK_MAX_BUFFER_SIZE];
+ char out[DPDK_MAX_BUFFER_SIZE];
+
+ /* non blocking check on helper logging pipe */
+ struct pollfd fds = {
+ .fd = phc->pipes[0], .events = POLLIN,
+ };
+ int data_avail = poll(&fds, 1, 0);
+ if (data_avail < 0) {
+ if (errno != EINTR || errno != EAGAIN) {
+ char errbuf[ERR_BUF_SIZE];
+ ERROR("%s: poll(2) failed: %s", phc->shm_name,
+ sstrerror(errno, errbuf, sizeof(errbuf)));
+ }
+ }
+ while (data_avail) {
+ int nbytes = read(phc->pipes[0], buf, sizeof(buf));
+ if (nbytes <= 0)
+ break;
+ sstrncpy(out, buf, nbytes);
+ DEBUG("%s: helper process:\n%s", phc->shm_name, out);
+ }
+}
+#endif /* DPDK_HELPER_USE_PIPES */
+
+int dpdk_helper_command(dpdk_helper_ctx_t *phc, enum DPDK_CMD cmd, int *result,
+ cdtime_t cmd_wait_time) {
+ if (phc == NULL) {
+ ERROR("Invalid argument(phc)");
+ return -EINVAL;
+ }
+
+ DEBUG("%s:%s:%d pid=%lu, cmd=%d", phc->shm_name, __FUNCTION__, __LINE__,
+ (long)getpid(), cmd);
+
+ phc->cmd_wait_time = cmd_wait_time;
+
+ int ret = 0;
+
+ ret = dpdk_helper_status_check(phc);
+
+#ifdef DPDK_HELPER_USE_PIPES
+ dpdk_helper_check_pipe(phc);
+#endif /* DPDK_HELPER_USE_PIPES */
+
+ if (ret != 0) {
+ return ret;
+ }
+
+ DEBUG("%s: DPDK command execute (cmd=%d)", phc->shm_name, cmd);
+
+ phc->cmd_result = 0;
+ phc->cmd = cmd;
+
+ /* kick helper to process command */
+ int err = sem_post(&phc->sema_cmd_start);
+ if (err) {
+ char errbuf[ERR_BUF_SIZE];
+ ERROR("dpdk_helper_worker: error posting sema_cmd_start semaphore (%s)\n",
+ sstrerror(errno, errbuf, sizeof(errbuf)));
+ }
+
+#if COLLECT_DEBUG
+ int val = 0;
+ if (sem_getvalue(&phc->sema_cmd_start, &val) == 0)
+ DEBUG("%s:dpdk_helper_command: post sema_cmd_start (value=%d)",
+ phc->shm_name, val);
+#endif
+
+ if (phc->cmd != DPDK_CMD_QUIT) {
+
+ /* wait for helper to complete processing */
+ struct timespec ts;
+ cdtime_t now = cdtime();
+
+ if (phc->status != DPDK_HELPER_ALIVE_SENDING_EVENTS) {
+ cmd_wait_time = MS_TO_CDTIME_T(DPDK_CDM_DEFAULT_TIMEOUT);
+ }
+
+ ts = CDTIME_T_TO_TIMESPEC(now + cmd_wait_time);
+ ret = sem_timedwait(&phc->sema_cmd_complete, &ts);
+ if (ret == -1 && errno == ETIMEDOUT) {
+ DPDK_HELPER_TRACE(phc->shm_name);
+ DEBUG("%s:sema_cmd_start: timeout in collectd thread: is a DPDK Primary "
+ "running?",
+ phc->shm_name);
+ return -ETIMEDOUT;
+ }
+
+#if COLLECT_DEBUG
+ val = 0;
+ if (sem_getvalue(&phc->sema_cmd_complete, &val) == 0)
+ DEBUG("%s:dpdk_helper_command: wait sema_cmd_complete (value=%d)",
+ phc->shm_name, val);
+#endif
+
+ if (result) {
+ *result = phc->cmd_result;
+ }
+ }
+
+#ifdef DPDK_HELPER_USE_PIPES
+ dpdk_helper_check_pipe(phc);
+#endif /* DPDK_HELPER_USE_PIPES */
+
+ DEBUG("%s: DPDK command complete (cmd=%d, result=%d)", phc->shm_name,
+ phc->cmd, phc->cmd_result);
+
+ return 0;
+}
+
+uint64_t strtoull_safe(const char *str, int *err) {
+ uint64_t val = 0;
+ char *endptr;
+ int res = 0;
+
+ val = strtoull(str, &endptr, 16);
+ if (*endptr) {
+ ERROR("%s Failed to parse the value %s, endptr=%c", __FUNCTION__, str,
+ *endptr);
+ res = -EINVAL;
+ }
+ if (err != NULL)
+ *err = res;
+ return val;
+}
+
+uint128_t str_to_uint128(const char *str, int len) {
+ uint128_t lcore_mask;
+ int err = 0;
+
+ memset(&lcore_mask, 0, sizeof(uint128_t));
+
+ if (len <= 2 || strncmp(str, "0x", 2) != 0) {
+ ERROR("%s Value %s should be represened in hexadecimal format",
+ __FUNCTION__, str);
+ return lcore_mask;
+ }
+
+ if (len <= 18) {
+ lcore_mask.low = strtoull_safe(str, &err);
+ if (err)
+ goto parse_out;
+ } else {
+ char low_str[DATA_MAX_NAME_LEN];
+ char high_str[DATA_MAX_NAME_LEN];
+
+ memset(high_str, 0, sizeof(high_str));
+ memset(low_str, 0, sizeof(low_str));
+
+ strncpy(high_str, str, len - 16);
+ strncpy(low_str, str + len - 16, 16);
+
+ lcore_mask.low = strtoull_safe(low_str, &err);
+ if (err)
+ goto parse_out;
+
+ lcore_mask.high = strtoull_safe(high_str, &err);
+ if (err) {
+ lcore_mask.low = 0;
+ goto parse_out;
+ }
+ }
+
+parse_out:
+ return lcore_mask;
+}
projects / collectd.git / commitdiff