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