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