Branch data Line data Source code
1 : : /* SPDX-License-Identifier: BSD-3-Clause
2 : : * Copyright(c) 2016-2018 Intel Corporation
3 : : */
4 : :
5 : : #include <dirent.h>
6 : : #include <errno.h>
7 : : #include <fcntl.h>
8 : : #include <fnmatch.h>
9 : : #include <inttypes.h>
10 : : #include <libgen.h>
11 : : #include <limits.h>
12 : : #include <pthread.h>
13 : : #include <stdio.h>
14 : : #include <stdlib.h>
15 : : #include <string.h>
16 : : #include <sys/file.h>
17 : : #include <sys/time.h>
18 : : #include <sys/socket.h>
19 : : #include <sys/un.h>
20 : : #include <unistd.h>
21 : :
22 : : #include <rte_alarm.h>
23 : : #include <rte_common.h>
24 : : #include <rte_cycles.h>
25 : : #include <rte_eal.h>
26 : : #include <rte_errno.h>
27 : : #include <rte_lcore.h>
28 : : #include <rte_log.h>
29 : : #include <rte_thread.h>
30 : :
31 : : #include <eal_export.h>
32 : : #include "eal_memcfg.h"
33 : : #include "eal_private.h"
34 : : #include "eal_filesystem.h"
35 : : #include "eal_internal_cfg.h"
36 : :
37 : : static RTE_ATOMIC(int) mp_fd = -1;
38 : : static rte_thread_t mp_handle_tid;
39 : : static char mp_filter[UNIX_PATH_MAX]; /* Filter for secondary process sockets */
40 : : static char mp_dir_path[UNIX_PATH_MAX]; /* The directory path for all mp sockets */
41 : : static pthread_mutex_t mp_mutex_action = PTHREAD_MUTEX_INITIALIZER;
42 : : static char peer_name[UNIX_PATH_MAX];
43 : :
44 : : struct action_entry {
45 : : TAILQ_ENTRY(action_entry) next;
46 : : char action_name[RTE_MP_MAX_NAME_LEN];
47 : : rte_mp_t action;
48 : : };
49 : :
50 : : /** Double linked list of actions. */
51 : : TAILQ_HEAD(action_entry_list, action_entry);
52 : :
53 : : static struct action_entry_list action_entry_list =
54 : : TAILQ_HEAD_INITIALIZER(action_entry_list);
55 : :
56 : : enum mp_type {
57 : : MP_MSG, /* Share message with peers, will not block */
58 : : MP_REQ, /* Request for information, Will block for a reply */
59 : : MP_REP, /* Response to previously-received request */
60 : : MP_IGN, /* Response telling requester to ignore this response */
61 : : };
62 : :
63 : : struct mp_msg_internal {
64 : : int type;
65 : : struct rte_mp_msg msg;
66 : : };
67 : :
68 : : struct async_request_param {
69 : : rte_mp_async_reply_t clb;
70 : : struct rte_mp_reply user_reply;
71 : : struct timespec end;
72 : : int n_responses_processed;
73 : : };
74 : :
75 : : struct pending_request {
76 : : TAILQ_ENTRY(pending_request) next;
77 : : unsigned long id;
78 : : enum {
79 : : REQUEST_TYPE_SYNC,
80 : : REQUEST_TYPE_ASYNC
81 : : } type;
82 : : char dst[UNIX_PATH_MAX];
83 : : struct rte_mp_msg *request;
84 : : struct rte_mp_msg *reply;
85 : : int reply_received;
86 : : union {
87 : : struct {
88 : : struct async_request_param *param;
89 : : } async;
90 : : struct {
91 : : pthread_cond_t cond;
92 : : } sync;
93 : : };
94 : : };
95 : :
96 : : static unsigned long next_request_id;
97 : :
98 : : TAILQ_HEAD(pending_request_list, pending_request);
99 : :
100 : : static struct {
101 : : struct pending_request_list requests;
102 : : pthread_mutex_t lock;
103 : : } pending_requests = {
104 : : .requests = TAILQ_HEAD_INITIALIZER(pending_requests.requests),
105 : : .lock = PTHREAD_MUTEX_INITIALIZER,
106 : : /**< used in async requests only */
107 : : };
108 : :
109 : : /* forward declarations */
110 : : static int
111 : : mp_send(struct rte_mp_msg *msg, const char *peer, int type);
112 : :
113 : : /* for use with alarm callback */
114 : : static void
115 : : async_reply_handle(void *arg);
116 : :
117 : : /* for use with alarm callback and process_msg */
118 : : static struct pending_request *
119 : : async_reply_handle_thread_unsafe(struct pending_request *req);
120 : :
121 : : static void
122 : : trigger_async_action(struct pending_request *req);
123 : :
124 : : static struct pending_request *
125 : 65 : find_request_by_name(const char *dst, const char *act_name)
126 : : {
127 : : struct pending_request *r;
128 : :
129 [ + + ]: 65 : TAILQ_FOREACH(r, &pending_requests.requests, next) {
130 [ + - ]: 32 : if (!strcmp(r->dst, dst) &&
131 [ - + ]: 32 : !strcmp(r->request->name, act_name))
132 : : break;
133 : : }
134 : :
135 : 65 : return r;
136 : : }
137 : :
138 : : static struct pending_request *
139 : : find_async_request_by_id(unsigned long id)
140 : : {
141 : : struct pending_request *r;
142 : :
143 [ # # ]: 0 : TAILQ_FOREACH(r, &pending_requests.requests, next) {
144 [ # # # # ]: 0 : if (r->id == id && r->type == REQUEST_TYPE_ASYNC)
145 : : return r;
146 : : }
147 : :
148 : : return NULL;
149 : : }
150 : :
151 : : /*
152 : : * Combine prefix and name(optional) to return unix domain socket path
153 : : * return the number of characters that would have been put into buffer.
154 : : */
155 : : static int
156 : 651 : create_socket_path(const char *name, char *buf, size_t len)
157 : : {
158 : 651 : const char *prefix = eal_mp_socket_path();
159 : :
160 [ + + ]: 651 : if (strlen(name) > 0)
161 : 281 : return snprintf(buf, len, "%s_%s", prefix, name);
162 : : else
163 : 370 : return strlcpy(buf, prefix, len);
164 : : }
165 : :
166 : : RTE_EXPORT_SYMBOL(rte_eal_primary_proc_alive)
167 : : int
168 : 0 : rte_eal_primary_proc_alive(const char *config_file_path)
169 : : {
170 : : int config_fd;
171 : :
172 [ # # ]: 0 : if (config_file_path)
173 : : config_fd = open(config_file_path, O_RDONLY);
174 : : else {
175 : : const char *path;
176 : :
177 : 0 : path = eal_runtime_config_path();
178 : : config_fd = open(path, O_RDONLY);
179 : : }
180 [ # # ]: 0 : if (config_fd < 0)
181 : : return 0;
182 : :
183 : 0 : int ret = lockf(config_fd, F_TEST, 0);
184 : 0 : close(config_fd);
185 : :
186 : 0 : return !!ret;
187 : : }
188 : :
189 : : static struct action_entry *
190 : 1610 : find_action_entry_by_name(const char *name)
191 : : {
192 : : struct action_entry *entry;
193 : :
194 [ + + ]: 3792 : TAILQ_FOREACH(entry, &action_entry_list, next) {
195 [ + + ]: 2703 : if (strncmp(entry->action_name, name, RTE_MP_MAX_NAME_LEN) == 0)
196 : : break;
197 : : }
198 : :
199 : 1610 : return entry;
200 : : }
201 : :
202 : : static int
203 : 2707 : validate_action_name(const char *name)
204 : : {
205 [ - + ]: 2707 : if (name == NULL) {
206 : 0 : EAL_LOG(ERR, "Action name cannot be NULL");
207 : 0 : rte_errno = EINVAL;
208 : 0 : return -1;
209 : : }
210 [ - + ]: 2707 : if (strnlen(name, RTE_MP_MAX_NAME_LEN) == 0) {
211 : 0 : EAL_LOG(ERR, "Length of action name is zero");
212 : 0 : rte_errno = EINVAL;
213 : 0 : return -1;
214 : : }
215 [ - + ]: 2707 : if (strnlen(name, RTE_MP_MAX_NAME_LEN) == RTE_MP_MAX_NAME_LEN) {
216 : 0 : rte_errno = E2BIG;
217 : 0 : return -1;
218 : : }
219 : : return 0;
220 : : }
221 : :
222 : : RTE_EXPORT_SYMBOL(rte_mp_action_register)
223 : : int
224 : 733 : rte_mp_action_register(const char *name, rte_mp_t action)
225 : : {
226 : : struct action_entry *entry;
227 : : const struct internal_config *internal_conf =
228 : 733 : eal_get_internal_configuration();
229 : :
230 [ + - ]: 733 : if (validate_action_name(name) != 0)
231 : : return -1;
232 : :
233 [ + + ]: 733 : if (internal_conf->no_shconf) {
234 : 30 : EAL_LOG(DEBUG, "No shared files mode enabled, IPC is disabled");
235 : 30 : rte_errno = ENOTSUP;
236 : 30 : return -1;
237 : : }
238 : :
239 : 703 : entry = malloc(sizeof(struct action_entry));
240 [ - + ]: 703 : if (entry == NULL) {
241 : 0 : rte_errno = ENOMEM;
242 : 0 : return -1;
243 : : }
244 : 703 : strlcpy(entry->action_name, name, sizeof(entry->action_name));
245 : 703 : entry->action = action;
246 : :
247 : 703 : pthread_mutex_lock(&mp_mutex_action);
248 [ - + ]: 703 : if (find_action_entry_by_name(name) != NULL) {
249 : 0 : pthread_mutex_unlock(&mp_mutex_action);
250 : 0 : rte_errno = EEXIST;
251 : 0 : free(entry);
252 : 0 : return -1;
253 : : }
254 : 703 : TAILQ_INSERT_TAIL(&action_entry_list, entry, next);
255 : 703 : pthread_mutex_unlock(&mp_mutex_action);
256 : 703 : return 0;
257 : : }
258 : :
259 : : RTE_EXPORT_SYMBOL(rte_mp_action_unregister)
260 : : void
261 : 923 : rte_mp_action_unregister(const char *name)
262 : : {
263 : : struct action_entry *entry;
264 : : const struct internal_config *internal_conf =
265 : 923 : eal_get_internal_configuration();
266 : :
267 [ + - ]: 923 : if (validate_action_name(name) != 0)
268 : : return;
269 : :
270 [ + + ]: 923 : if (internal_conf->no_shconf) {
271 : 51 : EAL_LOG(DEBUG, "No shared files mode enabled, IPC is disabled");
272 : 51 : return;
273 : : }
274 : :
275 : 872 : pthread_mutex_lock(&mp_mutex_action);
276 : 872 : entry = find_action_entry_by_name(name);
277 [ + + ]: 872 : if (entry == NULL) {
278 : 386 : pthread_mutex_unlock(&mp_mutex_action);
279 : 386 : return;
280 : : }
281 [ + + ]: 486 : TAILQ_REMOVE(&action_entry_list, entry, next);
282 : 486 : pthread_mutex_unlock(&mp_mutex_action);
283 : 486 : free(entry);
284 : : }
285 : :
286 : : static int
287 : 283 : read_msg(int fd, struct mp_msg_internal *m, struct sockaddr_un *s)
288 : : {
289 : : int msglen;
290 : : struct iovec iov;
291 : : struct msghdr msgh;
292 : : char control[CMSG_SPACE(sizeof(m->msg.fds))];
293 : : struct cmsghdr *cmsg;
294 : : int buflen = sizeof(*m) - sizeof(m->msg.fds);
295 : :
296 : : memset(&msgh, 0, sizeof(msgh));
297 : 283 : iov.iov_base = m;
298 : 283 : iov.iov_len = buflen;
299 : :
300 : 283 : msgh.msg_name = s;
301 : 283 : msgh.msg_namelen = sizeof(*s);
302 : 283 : msgh.msg_iov = &iov;
303 : 283 : msgh.msg_iovlen = 1;
304 : 283 : msgh.msg_control = control;
305 : 283 : msgh.msg_controllen = sizeof(control);
306 : :
307 : 283 : retry:
308 : 283 : msglen = recvmsg(fd, &msgh, 0);
309 : :
310 : : /* zero length message means socket was closed */
311 [ + - ]: 67 : if (msglen == 0)
312 : : return 0;
313 : :
314 [ - + ]: 67 : if (msglen < 0) {
315 [ # # ]: 0 : if (errno == EINTR)
316 : 0 : goto retry;
317 : :
318 : 0 : EAL_LOG(ERR, "recvmsg failed, %s", strerror(errno));
319 : 0 : return -1;
320 : : }
321 : :
322 [ + - - + ]: 67 : if (msglen != buflen || (msgh.msg_flags & (MSG_TRUNC | MSG_CTRUNC))) {
323 : 0 : EAL_LOG(ERR, "truncated msg");
324 : 0 : return -1;
325 : : }
326 : :
327 : : /* read auxiliary FDs if any */
328 [ - + - + ]: 134 : for (cmsg = CMSG_FIRSTHDR(&msgh); cmsg != NULL;
329 : : cmsg = CMSG_NXTHDR(&msgh, cmsg)) {
330 [ # # ]: 0 : if ((cmsg->cmsg_level == SOL_SOCKET) &&
331 : : (cmsg->cmsg_type == SCM_RIGHTS)) {
332 : 0 : memcpy(m->msg.fds, CMSG_DATA(cmsg), sizeof(m->msg.fds));
333 : : break;
334 : : }
335 : : }
336 : : /* sanity-check the response */
337 [ - + ]: 67 : if (m->msg.num_fds < 0 || m->msg.num_fds > RTE_MP_MAX_FD_NUM) {
338 : 0 : EAL_LOG(ERR, "invalid number of fd's received");
339 : 0 : return -1;
340 : : }
341 [ - + ]: 67 : if (m->msg.len_param < 0 || m->msg.len_param > RTE_MP_MAX_PARAM_LEN) {
342 : 0 : EAL_LOG(ERR, "invalid received data length");
343 : 0 : return -1;
344 : : }
345 : : return msglen;
346 : : }
347 : :
348 : : static void
349 : : cleanup_msg_fds(const struct rte_mp_msg *msg)
350 : : {
351 : : int i;
352 : :
353 [ # # # # ]: 0 : for (i = 0; i < msg->num_fds; i++)
354 : 0 : close(msg->fds[i]);
355 : : }
356 : :
357 : : static void
358 : 67 : process_msg(struct mp_msg_internal *m, struct sockaddr_un *s)
359 : : {
360 : : struct pending_request *pending_req;
361 : : struct action_entry *entry;
362 : 67 : struct rte_mp_msg *msg = &m->msg;
363 : : rte_mp_t action = NULL;
364 : : const struct internal_config *internal_conf =
365 : 67 : eal_get_internal_configuration();
366 : :
367 : 67 : EAL_LOG(DEBUG, "msg: %s", msg->name);
368 : :
369 [ + + ]: 67 : if (m->type == MP_REP || m->type == MP_IGN) {
370 : : struct pending_request *req = NULL;
371 : :
372 : 32 : pthread_mutex_lock(&pending_requests.lock);
373 : 32 : pending_req = find_request_by_name(s->sun_path, msg->name);
374 [ + - ]: 32 : if (pending_req) {
375 [ - + ]: 32 : memcpy(pending_req->reply, msg, sizeof(*msg));
376 : : /* -1 indicates that we've been asked to ignore */
377 : 32 : pending_req->reply_received =
378 [ - + ]: 32 : m->type == MP_REP ? 1 : -1;
379 : :
380 [ + + ]: 32 : if (pending_req->type == REQUEST_TYPE_SYNC)
381 : 31 : pthread_cond_signal(&pending_req->sync.cond);
382 [ + - ]: 1 : else if (pending_req->type == REQUEST_TYPE_ASYNC)
383 : 1 : req = async_reply_handle_thread_unsafe(pending_req);
384 : : } else {
385 : 0 : EAL_LOG(ERR, "Drop mp reply: %s", msg->name);
386 : : cleanup_msg_fds(msg);
387 : : }
388 : 32 : pthread_mutex_unlock(&pending_requests.lock);
389 : :
390 [ + + ]: 32 : if (req != NULL)
391 : 1 : trigger_async_action(req);
392 : 32 : return;
393 : : }
394 : :
395 : 35 : pthread_mutex_lock(&mp_mutex_action);
396 : 35 : entry = find_action_entry_by_name(msg->name);
397 [ + - ]: 35 : if (entry != NULL)
398 : 35 : action = entry->action;
399 : 35 : pthread_mutex_unlock(&mp_mutex_action);
400 : :
401 [ - + ]: 35 : if (!action) {
402 [ # # # # ]: 0 : if (m->type == MP_REQ && !internal_conf->init_complete) {
403 : : /* if this is a request, and init is not yet complete,
404 : : * and callback wasn't registered, we should tell the
405 : : * requester to ignore our existence because we're not
406 : : * yet ready to process this request.
407 : : */
408 : : struct rte_mp_msg dummy;
409 : :
410 : : memset(&dummy, 0, sizeof(dummy));
411 : : strlcpy(dummy.name, msg->name, sizeof(dummy.name));
412 : 0 : mp_send(&dummy, s->sun_path, MP_IGN);
413 : : } else {
414 : 0 : EAL_LOG(ERR, "Cannot find action: %s",
415 : : msg->name);
416 : : }
417 : : cleanup_msg_fds(msg);
418 [ - + ]: 35 : } else if (action(msg, s->sun_path) < 0) {
419 : 0 : EAL_LOG(ERR, "Fail to handle message: %s", msg->name);
420 : : }
421 : : }
422 : :
423 : : static uint32_t
424 : 217 : mp_handle(void *arg __rte_unused)
425 : : {
426 : : struct mp_msg_internal msg;
427 : : struct sockaddr_un sa;
428 : : int fd;
429 : :
430 [ + + ]: 284 : while ((fd = rte_atomic_load_explicit(&mp_fd, rte_memory_order_relaxed)) >= 0) {
431 : : int ret;
432 : :
433 : 283 : ret = read_msg(fd, &msg, &sa);
434 [ + - ]: 67 : if (ret <= 0)
435 : : break;
436 : :
437 : 67 : process_msg(&msg, &sa);
438 : : }
439 : :
440 : 1 : return 0;
441 : : }
442 : :
443 : : static int
444 : : timespec_cmp(const struct timespec *a, const struct timespec *b)
445 : : {
446 : 1 : if (a->tv_sec < b->tv_sec)
447 : : return -1;
448 [ - + ]: 1 : if (a->tv_sec > b->tv_sec)
449 : : return 1;
450 [ # # ]: 0 : if (a->tv_nsec < b->tv_nsec)
451 : : return -1;
452 [ # # ]: 0 : if (a->tv_nsec > b->tv_nsec)
453 : 0 : return 1;
454 : : return 0;
455 : : }
456 : :
457 : : enum async_action {
458 : : ACTION_FREE, /**< free the action entry, but don't trigger callback */
459 : : ACTION_TRIGGER /**< trigger callback, then free action entry */
460 : : };
461 : :
462 : : static enum async_action
463 : 1 : process_async_request(struct pending_request *sr, const struct timespec *now)
464 : : {
465 : : struct async_request_param *param;
466 : : struct rte_mp_reply *reply;
467 : : bool timeout, last_msg;
468 : :
469 [ + - ]: 1 : param = sr->async.param;
470 : : reply = ¶m->user_reply;
471 : :
472 : : /* did we timeout? */
473 : : timeout = timespec_cmp(¶m->end, now) <= 0;
474 : :
475 : : /* if we received a response, adjust relevant data and copy message. */
476 [ + - + - ]: 1 : if (sr->reply_received == 1 && sr->reply) {
477 : : struct rte_mp_msg *msg, *user_msgs, *tmp;
478 : :
479 : : msg = sr->reply;
480 : 1 : user_msgs = reply->msgs;
481 : :
482 : 1 : tmp = realloc(user_msgs, sizeof(*msg) *
483 : 1 : (reply->nb_received + 1));
484 [ - + ]: 1 : if (!tmp) {
485 : 0 : EAL_LOG(ERR, "Fail to alloc reply for request %s:%s",
486 : : sr->dst, sr->request->name);
487 : : /* this entry is going to be removed and its message
488 : : * dropped, but we don't want to leak memory, so
489 : : * continue.
490 : : */
491 : : } else {
492 : : user_msgs = tmp;
493 : 1 : reply->msgs = user_msgs;
494 : 1 : memcpy(&user_msgs[reply->nb_received],
495 : : msg, sizeof(*msg));
496 : 1 : reply->nb_received++;
497 : : }
498 : :
499 : : /* mark this request as processed */
500 : 1 : param->n_responses_processed++;
501 [ # # ]: 0 : } else if (sr->reply_received == -1) {
502 : : /* we were asked to ignore this process */
503 : 0 : reply->nb_sent--;
504 [ # # ]: 0 : } else if (timeout) {
505 : : /* count it as processed response, but don't increment
506 : : * nb_received.
507 : : */
508 : 0 : param->n_responses_processed++;
509 : : }
510 : :
511 : 1 : free(sr->reply);
512 : :
513 : 1 : last_msg = param->n_responses_processed == reply->nb_sent;
514 : :
515 : 1 : return last_msg ? ACTION_TRIGGER : ACTION_FREE;
516 : : }
517 : :
518 : : static void
519 : 1 : trigger_async_action(struct pending_request *sr)
520 : : {
521 : : struct async_request_param *param;
522 : : struct rte_mp_reply *reply;
523 : :
524 : 1 : param = sr->async.param;
525 : 1 : reply = ¶m->user_reply;
526 : :
527 : 1 : param->clb(sr->request, reply);
528 : :
529 : : /* clean up */
530 : 1 : free(sr->async.param->user_reply.msgs);
531 : 1 : free(sr->async.param);
532 : 1 : free(sr->request);
533 : 1 : free(sr);
534 : 1 : }
535 : :
536 : : static struct pending_request *
537 : 1 : async_reply_handle_thread_unsafe(struct pending_request *req)
538 : : {
539 : : enum async_action action;
540 : : struct timespec ts_now;
541 : :
542 [ - + ]: 1 : if (clock_gettime(CLOCK_MONOTONIC, &ts_now) < 0) {
543 : 0 : EAL_LOG(ERR, "Cannot get current time");
544 : 0 : goto no_trigger;
545 : : }
546 : :
547 : 1 : action = process_async_request(req, &ts_now);
548 : :
549 [ - + ]: 1 : TAILQ_REMOVE(&pending_requests.requests, req, next);
550 : :
551 [ - + ]: 1 : if (action == ACTION_TRIGGER)
552 : : return req;
553 : 0 : no_trigger:
554 : 0 : free(req);
555 : 0 : return NULL;
556 : : }
557 : :
558 : : static void
559 : 0 : async_reply_handle(void *arg)
560 : : {
561 : : struct pending_request *req;
562 : : /* alarm arg carries the request ID packed into a void * via uintptr_t */
563 : 0 : unsigned long id = (uintptr_t)arg;
564 : :
565 : : /* ensure request ID matches pointer size */
566 : : RTE_BUILD_BUG_ON(sizeof(next_request_id) != sizeof(uintptr_t));
567 : :
568 : 0 : pthread_mutex_lock(&pending_requests.lock);
569 : : req = find_async_request_by_id(id);
570 [ # # ]: 0 : if (req != NULL)
571 : 0 : req = async_reply_handle_thread_unsafe(req);
572 : 0 : pthread_mutex_unlock(&pending_requests.lock);
573 : :
574 [ # # ]: 0 : if (req != NULL)
575 : 0 : trigger_async_action(req);
576 : 0 : }
577 : :
578 : : static int
579 : 217 : open_socket_fd(void)
580 : : {
581 : 217 : struct sockaddr_un un = { .sun_family = AF_UNIX };
582 : :
583 : 217 : peer_name[0] = '\0';
584 [ + + ]: 217 : if (rte_eal_process_type() == RTE_PROC_SECONDARY)
585 : 32 : snprintf(peer_name, sizeof(peer_name),
586 : : "%d_%"PRIx64, getpid(), rte_rdtsc());
587 : :
588 [ - + ]: 217 : if (create_socket_path(peer_name, un.sun_path, sizeof(un.sun_path))
589 : : >= (int)sizeof(un.sun_path)) {
590 : 0 : EAL_LOG(ERR, "peer '%s' socket path too long", peer_name);
591 : 0 : return -1;
592 : : }
593 : :
594 : 217 : mp_fd = socket(AF_UNIX, SOCK_DGRAM, 0);
595 [ - + ]: 217 : if (mp_fd < 0) {
596 : 0 : EAL_LOG(ERR, "failed to create unix socket");
597 : 0 : return -1;
598 : : }
599 : :
600 : 217 : unlink(un.sun_path); /* May still exist since last run */
601 : :
602 [ - + ]: 217 : if (bind(mp_fd, (struct sockaddr *)&un, sizeof(un)) < 0) {
603 : 0 : EAL_LOG(ERR, "failed to bind %s: %s",
604 : : un.sun_path, strerror(errno));
605 : 0 : close(mp_fd);
606 : 0 : return -1;
607 : : }
608 : :
609 : 217 : EAL_LOG(INFO, "Multi-process socket %s", un.sun_path);
610 : 217 : return mp_fd;
611 : : }
612 : :
613 : : static void
614 : 217 : close_socket_fd(int fd)
615 : : {
616 : : char path[UNIX_PATH_MAX];
617 : :
618 : 217 : close(fd);
619 : :
620 [ - + ]: 217 : if (create_socket_path(peer_name, path, sizeof(path)) < 0)
621 : 0 : EAL_LOG(ERR, "file prefix path for peer '%s' too long", peer_name);
622 : : else
623 : 217 : unlink(path);
624 : 217 : }
625 : :
626 : : int
627 : 227 : rte_mp_channel_init(void)
628 : : {
629 : : char path[UNIX_PATH_MAX];
630 : : int dir_fd;
631 : : const struct internal_config *internal_conf =
632 : 227 : eal_get_internal_configuration();
633 : :
634 : : /* in no shared files mode, we do not have secondary processes support,
635 : : * so no need to initialize IPC.
636 : : */
637 [ + + ]: 227 : if (internal_conf->no_shconf) {
638 : 10 : EAL_LOG(DEBUG, "No shared files mode enabled, IPC will be disabled");
639 : 10 : rte_errno = ENOTSUP;
640 : 10 : return -1;
641 : : }
642 : :
643 : : /* create filter path */
644 [ - + ]: 217 : if (create_socket_path("*", path, sizeof(path)) < 0) {
645 : 0 : EAL_LOG(ERR, "file prefix path too long");
646 : 0 : rte_errno = ENAMETOOLONG;
647 : 0 : return -1;
648 : : }
649 : :
650 : 217 : rte_basename(path, mp_filter, sizeof(mp_filter));
651 : 217 : strlcpy(mp_dir_path, dirname(path), sizeof(mp_dir_path));
652 : :
653 : : /* lock the directory */
654 : : dir_fd = open(mp_dir_path, O_RDONLY);
655 [ - + ]: 217 : if (dir_fd < 0) {
656 : 0 : EAL_LOG(ERR, "failed to open %s: %s",
657 : : mp_dir_path, strerror(errno));
658 : 0 : return -1;
659 : : }
660 : :
661 [ - + ]: 217 : if (flock(dir_fd, LOCK_EX)) {
662 : 0 : EAL_LOG(ERR, "failed to lock %s: %s",
663 : : mp_dir_path, strerror(errno));
664 : 0 : close(dir_fd);
665 : 0 : return -1;
666 : : }
667 : :
668 [ - + ]: 217 : if (open_socket_fd() < 0) {
669 : 0 : close(dir_fd);
670 : 0 : return -1;
671 : : }
672 : :
673 [ - + ]: 217 : if (rte_thread_create_internal_control(&mp_handle_tid, "mp-msg",
674 : : mp_handle, NULL) < 0) {
675 : 0 : EAL_LOG(ERR, "failed to create mp thread: %s",
676 : : strerror(errno));
677 : 0 : close(dir_fd);
678 : 0 : close(rte_atomic_exchange_explicit(&mp_fd, -1, rte_memory_order_relaxed));
679 : 0 : return -1;
680 : : }
681 : :
682 : : /* unlock the directory */
683 : 217 : flock(dir_fd, LOCK_UN);
684 : 217 : close(dir_fd);
685 : :
686 : 217 : return 0;
687 : : }
688 : :
689 : : void
690 : 296 : rte_mp_channel_cleanup(void)
691 : : {
692 : : int fd;
693 : :
694 : 296 : fd = rte_atomic_exchange_explicit(&mp_fd, -1, rte_memory_order_relaxed);
695 [ + + ]: 296 : if (fd < 0)
696 : : return;
697 : :
698 : 217 : pthread_cancel((pthread_t)mp_handle_tid.opaque_id);
699 : 217 : rte_thread_join(mp_handle_tid, NULL);
700 : 217 : close_socket_fd(fd);
701 : : }
702 : :
703 : : /**
704 : : * Return -1, as fail to send message and it's caused by the local side.
705 : : * Return 0, as fail to send message and it's caused by the remote side.
706 : : * Return 1, as succeed to send message.
707 : : */
708 : : static int
709 : 68 : send_msg(const char *dst_path, struct rte_mp_msg *msg, int type)
710 : : {
711 : : int snd;
712 : : struct iovec iov;
713 : : struct msghdr msgh;
714 : : struct cmsghdr *cmsg;
715 : : struct sockaddr_un dst;
716 : : struct mp_msg_internal m;
717 : 68 : int fd_size = msg->num_fds * sizeof(int);
718 : 68 : const int32_t control_sz = CMSG_SPACE(fd_size);
719 : : char control[CMSG_SPACE(sizeof(msg->fds))];
720 : :
721 [ + - ]: 68 : m.type = type;
722 : : memcpy(&m.msg, msg, sizeof(*msg));
723 : :
724 : : memset(&dst, 0, sizeof(dst));
725 [ + - ]: 68 : dst.sun_family = AF_UNIX;
726 : : strlcpy(dst.sun_path, dst_path, sizeof(dst.sun_path));
727 : :
728 : : memset(&msgh, 0, sizeof(msgh));
729 : : memset(control, 0, sizeof(control));
730 : :
731 : 68 : iov.iov_base = &m;
732 : 68 : iov.iov_len = sizeof(m) - sizeof(msg->fds);
733 : :
734 : 68 : msgh.msg_name = &dst;
735 : 68 : msgh.msg_namelen = sizeof(dst);
736 : 68 : msgh.msg_iov = &iov;
737 : 68 : msgh.msg_iovlen = 1;
738 : 68 : msgh.msg_control = control;
739 : 68 : msgh.msg_controllen = control_sz;
740 : :
741 [ + - ]: 68 : cmsg = CMSG_FIRSTHDR(&msgh);
742 : 68 : cmsg->cmsg_len = CMSG_LEN(fd_size);
743 : 68 : cmsg->cmsg_level = SOL_SOCKET;
744 : 68 : cmsg->cmsg_type = SCM_RIGHTS;
745 : 68 : memcpy(CMSG_DATA(cmsg), msg->fds, fd_size);
746 : :
747 : : do {
748 : 68 : snd = sendmsg(mp_fd, &msgh, 0);
749 [ + + - + ]: 68 : } while (snd < 0 && errno == EINTR);
750 : :
751 [ + + ]: 68 : if (snd < 0) {
752 : 1 : rte_errno = errno;
753 : : /* Check if it caused by peer process exits */
754 [ - + - - ]: 1 : if (errno == ECONNREFUSED &&
755 : 0 : rte_eal_process_type() == RTE_PROC_PRIMARY) {
756 : 0 : unlink(dst_path);
757 : 0 : return 0;
758 : : }
759 : 1 : EAL_LOG(ERR, "failed to send to (%s) due to %s",
760 : : dst_path, strerror(errno));
761 : 1 : return -1;
762 : : }
763 : :
764 : : return 1;
765 : : }
766 : :
767 : : static int
768 : 39 : mp_send(struct rte_mp_msg *msg, const char *peer, int type)
769 : : {
770 : : int dir_fd, ret = 0;
771 : : DIR *mp_dir;
772 : : struct dirent *ent;
773 : :
774 [ + + + + ]: 39 : if (!peer && (rte_eal_process_type() == RTE_PROC_SECONDARY))
775 : 1 : peer = eal_mp_socket_path();
776 : :
777 [ + + ]: 39 : if (peer) {
778 [ + - ]: 33 : if (send_msg(peer, msg, type) < 0)
779 : : return -1;
780 : : else
781 : 33 : return 0;
782 : : }
783 : :
784 : : /* broadcast to all secondary processes */
785 : 6 : mp_dir = opendir(mp_dir_path);
786 [ - + ]: 6 : if (!mp_dir) {
787 : 0 : EAL_LOG(ERR, "Unable to open directory %s",
788 : : mp_dir_path);
789 : 0 : rte_errno = errno;
790 : 0 : return -1;
791 : : }
792 : :
793 : 6 : dir_fd = dirfd(mp_dir);
794 : : /* lock the directory to prevent processes spinning up while we send */
795 [ - + ]: 6 : if (flock(dir_fd, LOCK_SH)) {
796 : 0 : EAL_LOG(ERR, "Unable to lock directory %s",
797 : : mp_dir_path);
798 : 0 : rte_errno = errno;
799 : 0 : closedir(mp_dir);
800 : 0 : return -1;
801 : : }
802 : :
803 [ + + ]: 60 : while ((ent = readdir(mp_dir))) {
804 : : char path[UNIX_PATH_MAX];
805 : :
806 [ + + ]: 54 : if (fnmatch(mp_filter, ent->d_name, 0) != 0)
807 : 52 : continue;
808 : :
809 [ - + ]: 2 : if (snprintf(path, sizeof(path), "%s/%s", mp_dir_path, ent->d_name)
810 : : >= (int)sizeof(path)) {
811 : 0 : EAL_LOG(ERR, "Unix domain path %s/%s too long", mp_dir_path, ent->d_name);
812 : : ret = -1;
813 [ - + ]: 2 : } else if (send_msg(path, msg, type) < 0)
814 : : ret = -1;
815 : : }
816 : : /* unlock the dir */
817 : 6 : flock(dir_fd, LOCK_UN);
818 : :
819 : : /* dir_fd automatically closed on closedir */
820 : 6 : closedir(mp_dir);
821 : 6 : return ret;
822 : : }
823 : :
824 : : static int
825 : 1051 : check_input(const struct rte_mp_msg *msg)
826 : : {
827 [ - + ]: 1051 : if (msg == NULL) {
828 : 0 : EAL_LOG(ERR, "Msg cannot be NULL");
829 : 0 : rte_errno = EINVAL;
830 : 0 : return -1;
831 : : }
832 : :
833 [ + - ]: 1051 : if (validate_action_name(msg->name) != 0)
834 : : return -1;
835 : :
836 [ - + ]: 1051 : if (msg->len_param < 0) {
837 : 0 : EAL_LOG(ERR, "Message data length is negative");
838 : 0 : rte_errno = EINVAL;
839 : 0 : return -1;
840 : : }
841 : :
842 [ - + ]: 1051 : if (msg->num_fds < 0) {
843 : 0 : EAL_LOG(ERR, "Number of fd's is negative");
844 : 0 : rte_errno = EINVAL;
845 : 0 : return -1;
846 : : }
847 : :
848 [ - + ]: 1051 : if (msg->len_param > RTE_MP_MAX_PARAM_LEN) {
849 : 0 : EAL_LOG(ERR, "Message data is too long");
850 : 0 : rte_errno = E2BIG;
851 : 0 : return -1;
852 : : }
853 : :
854 [ - + ]: 1051 : if (msg->num_fds > RTE_MP_MAX_FD_NUM) {
855 : 0 : EAL_LOG(ERR, "Cannot send more than %d FDs",
856 : : RTE_MP_MAX_FD_NUM);
857 : 0 : rte_errno = E2BIG;
858 : 0 : return -1;
859 : : }
860 : :
861 : : return 0;
862 : : }
863 : :
864 : : RTE_EXPORT_SYMBOL(rte_mp_sendmsg)
865 : : int
866 : 7 : rte_mp_sendmsg(struct rte_mp_msg *msg)
867 : : {
868 : : const struct internal_config *internal_conf =
869 : 7 : eal_get_internal_configuration();
870 : :
871 [ + - ]: 7 : if (check_input(msg) != 0)
872 : : return -1;
873 : :
874 [ - + ]: 7 : if (internal_conf->no_shconf) {
875 : 0 : EAL_LOG(DEBUG, "No shared files mode enabled, IPC is disabled");
876 : 0 : rte_errno = ENOTSUP;
877 : 0 : return -1;
878 : : }
879 : :
880 : 7 : EAL_LOG(DEBUG, "sendmsg: %s", msg->name);
881 : 7 : return mp_send(msg, NULL, MP_MSG);
882 : : }
883 : :
884 : : static int
885 : 1 : mp_request_async(const char *dst, struct rte_mp_msg *req,
886 : : struct async_request_param *param, const struct timespec *ts)
887 : : {
888 : : struct rte_mp_msg *reply_msg;
889 : : struct pending_request *pending_req, *exist;
890 : : unsigned long id;
891 : : int ret = -1;
892 : :
893 : : /* queue already locked by caller */
894 : :
895 : 1 : exist = find_request_by_name(dst, req->name);
896 [ - + ]: 1 : if (exist) {
897 : 0 : EAL_LOG(ERR, "A pending request %s:%s", dst, req->name);
898 : 0 : rte_errno = EEXIST;
899 : 0 : return -1;
900 : : }
901 : :
902 : 1 : pending_req = calloc(1, sizeof(*pending_req));
903 : 1 : reply_msg = calloc(1, sizeof(*reply_msg));
904 [ - + ]: 1 : if (pending_req == NULL || reply_msg == NULL) {
905 : 0 : EAL_LOG(ERR, "Could not allocate space for async request");
906 : 0 : rte_errno = ENOMEM;
907 : : ret = -1;
908 : 0 : goto fail;
909 : : }
910 : :
911 : 1 : id = ++next_request_id;
912 : 1 : pending_req->type = REQUEST_TYPE_ASYNC;
913 : 1 : pending_req->id = id;
914 : 1 : strlcpy(pending_req->dst, dst, sizeof(pending_req->dst));
915 : 1 : pending_req->request = req;
916 : 1 : pending_req->reply = reply_msg;
917 : 1 : pending_req->async.param = param;
918 : :
919 : 1 : ret = send_msg(dst, req, MP_REQ);
920 [ - + ]: 1 : if (ret < 0) {
921 : 0 : EAL_LOG(ERR, "Fail to send request %s:%s",
922 : : dst, req->name);
923 : : ret = -1;
924 : 0 : goto fail;
925 [ - + ]: 1 : } else if (ret == 0) {
926 : : ret = 0;
927 : 0 : goto fail;
928 : : }
929 : 1 : param->user_reply.nb_sent++;
930 : :
931 : : /* if alarm set fails, we simply ignore the reply */
932 [ - + ]: 1 : if (rte_eal_alarm_set(ts->tv_sec * 1000000 + ts->tv_nsec / 1000,
933 : : async_reply_handle, (void *)(uintptr_t)id) < 0) {
934 : 0 : EAL_LOG(ERR, "Fail to set alarm for request %s:%s",
935 : : dst, req->name);
936 : : ret = -1;
937 : 0 : goto fail;
938 : : }
939 : 1 : TAILQ_INSERT_TAIL(&pending_requests.requests, pending_req, next);
940 : :
941 : 1 : return 0;
942 : 0 : fail:
943 : 0 : free(pending_req);
944 : 0 : free(reply_msg);
945 : 0 : return ret;
946 : : }
947 : :
948 : : static int
949 : 32 : mp_request_sync(const char *dst, struct rte_mp_msg *req,
950 : : struct rte_mp_reply *reply, const struct timespec *ts)
951 : : {
952 : : int ret;
953 : : pthread_condattr_t attr;
954 : : struct rte_mp_msg msg, *tmp;
955 : : struct pending_request pending_req, *exist;
956 : :
957 : 32 : pending_req.type = REQUEST_TYPE_SYNC;
958 : 32 : pending_req.reply_received = 0;
959 : : strlcpy(pending_req.dst, dst, sizeof(pending_req.dst));
960 : 32 : pending_req.request = req;
961 : 32 : pending_req.reply = &msg;
962 : 32 : pthread_condattr_init(&attr);
963 : 32 : pthread_condattr_setclock(&attr, CLOCK_MONOTONIC);
964 : 32 : pthread_cond_init(&pending_req.sync.cond, &attr);
965 : :
966 : 32 : exist = find_request_by_name(dst, req->name);
967 [ - + ]: 32 : if (exist) {
968 : 0 : EAL_LOG(ERR, "A pending request %s:%s", dst, req->name);
969 : 0 : rte_errno = EEXIST;
970 : 0 : return -1;
971 : : }
972 : :
973 : 32 : ret = send_msg(dst, req, MP_REQ);
974 [ + + ]: 32 : if (ret < 0) {
975 : 1 : EAL_LOG(ERR, "Fail to send request %s:%s",
976 : : dst, req->name);
977 : 1 : return -1;
978 [ + - ]: 31 : } else if (ret == 0)
979 : : return 0;
980 : :
981 : 31 : TAILQ_INSERT_TAIL(&pending_requests.requests, &pending_req, next);
982 : :
983 : 31 : reply->nb_sent++;
984 : :
985 : : do {
986 : 31 : ret = pthread_cond_timedwait(&pending_req.sync.cond,
987 : : &pending_requests.lock, ts);
988 [ - + ]: 31 : } while (ret != 0 && ret != ETIMEDOUT);
989 : :
990 [ - + ]: 31 : TAILQ_REMOVE(&pending_requests.requests, &pending_req, next);
991 : :
992 [ - + ]: 31 : if (pending_req.reply_received == 0) {
993 : 0 : EAL_LOG(ERR, "Fail to recv reply for request %s:%s",
994 : : dst, req->name);
995 : 0 : rte_errno = ETIMEDOUT;
996 : 0 : return -1;
997 : : }
998 [ - + ]: 31 : if (pending_req.reply_received == -1) {
999 : 0 : EAL_LOG(DEBUG, "Asked to ignore response");
1000 : : /* not receiving this message is not an error, so decrement
1001 : : * number of sent messages
1002 : : */
1003 : 0 : reply->nb_sent--;
1004 : 0 : return 0;
1005 : : }
1006 : :
1007 : 31 : tmp = realloc(reply->msgs, sizeof(msg) * (reply->nb_received + 1));
1008 [ - + ]: 31 : if (!tmp) {
1009 : 0 : EAL_LOG(ERR, "Fail to alloc reply for request %s:%s",
1010 : : dst, req->name);
1011 : 0 : rte_errno = ENOMEM;
1012 : 0 : return -1;
1013 : : }
1014 : 31 : memcpy(&tmp[reply->nb_received], &msg, sizeof(msg));
1015 : 31 : reply->msgs = tmp;
1016 : 31 : reply->nb_received++;
1017 : 31 : return 0;
1018 : : }
1019 : :
1020 : : RTE_EXPORT_SYMBOL(rte_mp_request_sync)
1021 : : int
1022 : 1011 : rte_mp_request_sync(struct rte_mp_msg *req, struct rte_mp_reply *reply,
1023 : : const struct timespec *ts)
1024 : : {
1025 : : int dir_fd, ret = -1;
1026 : : DIR *mp_dir;
1027 : : struct dirent *ent;
1028 : : struct timespec now, end;
1029 : : const struct internal_config *internal_conf =
1030 : 1011 : eal_get_internal_configuration();
1031 : :
1032 : 1011 : EAL_LOG(DEBUG, "request: %s", req->name);
1033 : :
1034 : 1011 : reply->nb_sent = 0;
1035 : 1011 : reply->nb_received = 0;
1036 : 1011 : reply->msgs = NULL;
1037 : :
1038 [ - + ]: 1011 : if (check_input(req) != 0)
1039 : 0 : goto end;
1040 : :
1041 [ + + ]: 1011 : if (internal_conf->no_shconf) {
1042 : 9 : EAL_LOG(DEBUG, "No shared files mode enabled, IPC is disabled");
1043 : 9 : rte_errno = ENOTSUP;
1044 : 9 : return -1;
1045 : : }
1046 : :
1047 [ - + ]: 1002 : if (clock_gettime(CLOCK_MONOTONIC, &now) < 0) {
1048 : 0 : EAL_LOG(ERR, "Failed to get current time");
1049 : 0 : rte_errno = errno;
1050 : 0 : goto end;
1051 : : }
1052 : :
1053 : 1002 : end.tv_nsec = (now.tv_nsec + ts->tv_nsec) % 1000000000;
1054 : 1002 : end.tv_sec = now.tv_sec + ts->tv_sec +
1055 : 1002 : (now.tv_nsec + ts->tv_nsec) / 1000000000;
1056 : :
1057 : : /* for secondary process, send request to the primary process only */
1058 [ + + ]: 1002 : if (rte_eal_process_type() == RTE_PROC_SECONDARY) {
1059 : 32 : pthread_mutex_lock(&pending_requests.lock);
1060 : 32 : ret = mp_request_sync(eal_mp_socket_path(), req, reply, &end);
1061 : 32 : pthread_mutex_unlock(&pending_requests.lock);
1062 : 32 : goto end;
1063 : : }
1064 : :
1065 : : /* for primary process, broadcast request, and collect reply 1 by 1 */
1066 : 970 : mp_dir = opendir(mp_dir_path);
1067 [ - + ]: 970 : if (!mp_dir) {
1068 : 0 : EAL_LOG(ERR, "Unable to open directory %s", mp_dir_path);
1069 : 0 : rte_errno = errno;
1070 : 0 : goto end;
1071 : : }
1072 : :
1073 : 970 : dir_fd = dirfd(mp_dir);
1074 : : /* lock the directory to prevent processes spinning up while we send */
1075 [ - + ]: 970 : if (flock(dir_fd, LOCK_SH)) {
1076 : 0 : EAL_LOG(ERR, "Unable to lock directory %s",
1077 : : mp_dir_path);
1078 : 0 : rte_errno = errno;
1079 : 0 : goto close_end;
1080 : : }
1081 : :
1082 : 970 : pthread_mutex_lock(&pending_requests.lock);
1083 [ + + ]: 10583 : while ((ent = readdir(mp_dir))) {
1084 : : char path[UNIX_PATH_MAX];
1085 : :
1086 [ + - ]: 9613 : if (fnmatch(mp_filter, ent->d_name, 0) != 0)
1087 : 9613 : continue;
1088 : :
1089 [ # # ]: 0 : if (snprintf(path, sizeof(path), "%s/%s", mp_dir_path, ent->d_name)
1090 : : >= (int)sizeof(path)) {
1091 : 0 : EAL_LOG(ERR, "Unix domain socket path '%s/%s' too long", mp_dir_path,
1092 : : ent->d_name);
1093 : 0 : rte_errno = ENAMETOOLONG;
1094 : 0 : goto unlock_end;
1095 : : }
1096 : :
1097 : : /* unlocks the mutex while waiting for response,
1098 : : * locks on receive
1099 : : */
1100 [ # # ]: 0 : if (mp_request_sync(path, req, reply, &end))
1101 : 0 : goto unlock_end;
1102 : : }
1103 : : ret = 0;
1104 : :
1105 : 970 : unlock_end:
1106 : 970 : pthread_mutex_unlock(&pending_requests.lock);
1107 : : /* unlock the directory */
1108 : 970 : flock(dir_fd, LOCK_UN);
1109 : :
1110 : 970 : close_end:
1111 : : /* dir_fd automatically closed on closedir */
1112 : 970 : closedir(mp_dir);
1113 : :
1114 : 1002 : end:
1115 [ + + ]: 1002 : if (ret) {
1116 : 1 : free(reply->msgs);
1117 : 1 : reply->nb_received = 0;
1118 : 1 : reply->msgs = NULL;
1119 : : }
1120 : : return ret;
1121 : : }
1122 : :
1123 : : RTE_EXPORT_SYMBOL(rte_mp_request_async)
1124 : : int
1125 : 1 : rte_mp_request_async(struct rte_mp_msg *req, const struct timespec *ts,
1126 : : rte_mp_async_reply_t clb)
1127 : : {
1128 : : struct rte_mp_msg *copy;
1129 : : struct pending_request *dummy;
1130 : : struct async_request_param *param;
1131 : : struct rte_mp_reply *reply;
1132 : : int dir_fd, ret = 0;
1133 : : DIR *mp_dir;
1134 : : struct dirent *ent;
1135 : : struct timespec now;
1136 : : struct timespec *end;
1137 : : bool dummy_used = false;
1138 : : const struct internal_config *internal_conf =
1139 : 1 : eal_get_internal_configuration();
1140 : :
1141 : 1 : EAL_LOG(DEBUG, "request: %s", req->name);
1142 : :
1143 [ + - ]: 1 : if (check_input(req) != 0)
1144 : : return -1;
1145 : :
1146 [ - + ]: 1 : if (internal_conf->no_shconf) {
1147 : 0 : EAL_LOG(DEBUG, "No shared files mode enabled, IPC is disabled");
1148 : 0 : rte_errno = ENOTSUP;
1149 : 0 : return -1;
1150 : : }
1151 : :
1152 [ - + ]: 1 : if (clock_gettime(CLOCK_MONOTONIC, &now) < 0) {
1153 : 0 : EAL_LOG(ERR, "Failed to get current time");
1154 : 0 : rte_errno = errno;
1155 : 0 : return -1;
1156 : : }
1157 : 1 : copy = calloc(1, sizeof(*copy));
1158 : 1 : dummy = calloc(1, sizeof(*dummy));
1159 : 1 : param = calloc(1, sizeof(*param));
1160 [ + - - + ]: 1 : if (copy == NULL || dummy == NULL || param == NULL) {
1161 : 0 : EAL_LOG(ERR, "Failed to allocate memory for async reply");
1162 : 0 : rte_errno = ENOMEM;
1163 : 0 : goto fail;
1164 : : }
1165 : :
1166 : : /* copy message */
1167 : : memcpy(copy, req, sizeof(*copy));
1168 : :
1169 : 1 : param->n_responses_processed = 0;
1170 : 1 : param->clb = clb;
1171 : : end = ¶m->end;
1172 : : reply = ¶m->user_reply;
1173 : :
1174 : 1 : end->tv_nsec = (now.tv_nsec + ts->tv_nsec) % 1000000000;
1175 : 1 : end->tv_sec = now.tv_sec + ts->tv_sec +
1176 : 1 : (now.tv_nsec + ts->tv_nsec) / 1000000000;
1177 : 1 : reply->nb_sent = 0;
1178 : 1 : reply->nb_received = 0;
1179 : 1 : reply->msgs = NULL;
1180 : :
1181 : : /* we have to lock the request queue here, as we will be adding a bunch
1182 : : * of requests to the queue at once, and some of the replies may arrive
1183 : : * before we add all of the requests to the queue.
1184 : : */
1185 : 1 : pthread_mutex_lock(&pending_requests.lock);
1186 : :
1187 : : /* we have to ensure that callback gets triggered even if we don't send
1188 : : * anything, therefore earlier we have allocated a dummy request. fill
1189 : : * it, and put it on the queue if we don't send any requests.
1190 : : */
1191 : 1 : dummy->type = REQUEST_TYPE_ASYNC;
1192 : 1 : dummy->id = ++next_request_id;
1193 : 1 : dummy->request = copy;
1194 : 1 : dummy->reply = NULL;
1195 : 1 : dummy->async.param = param;
1196 : 1 : dummy->reply_received = 1; /* short-circuit the timeout */
1197 : :
1198 : : /* for secondary process, send request to the primary process only */
1199 [ - + ]: 1 : if (rte_eal_process_type() == RTE_PROC_SECONDARY) {
1200 : 0 : ret = mp_request_async(eal_mp_socket_path(), copy, param, ts);
1201 : :
1202 : : /* if we didn't send anything, put dummy request on the queue
1203 : : * and set a minimum-delay alarm so the callback fires immediately.
1204 : : */
1205 [ # # # # ]: 0 : if (ret == 0 && reply->nb_sent == 0) {
1206 : 0 : TAILQ_INSERT_TAIL(&pending_requests.requests, dummy, next);
1207 : : dummy_used = true;
1208 [ # # ]: 0 : if (rte_eal_alarm_set(1, async_reply_handle,
1209 : 0 : (void *)(uintptr_t)dummy->id) < 0) {
1210 : 0 : EAL_LOG(ERR, "Fail to set alarm for dummy request");
1211 : : /* roll back the changes */
1212 [ # # ]: 0 : TAILQ_REMOVE(&pending_requests.requests, dummy, next);
1213 : : dummy_used = false;
1214 : : ret = -1;
1215 : 0 : goto unlock_fail;
1216 : : }
1217 : : }
1218 : :
1219 : 0 : pthread_mutex_unlock(&pending_requests.lock);
1220 : :
1221 : : /* if we couldn't send anything, clean up */
1222 [ # # ]: 0 : if (ret != 0)
1223 : 0 : goto fail;
1224 [ # # ]: 0 : if (!dummy_used)
1225 : 0 : free(dummy);
1226 : 0 : return 0;
1227 : : }
1228 : :
1229 : : /* for primary process, broadcast request */
1230 : 1 : mp_dir = opendir(mp_dir_path);
1231 [ - + ]: 1 : if (!mp_dir) {
1232 : 0 : EAL_LOG(ERR, "Unable to open directory %s", mp_dir_path);
1233 : 0 : rte_errno = errno;
1234 : 0 : goto unlock_fail;
1235 : : }
1236 : 1 : dir_fd = dirfd(mp_dir);
1237 : :
1238 : : /* lock the directory to prevent processes spinning up while we send */
1239 [ - + ]: 1 : if (flock(dir_fd, LOCK_SH)) {
1240 : 0 : EAL_LOG(ERR, "Unable to lock directory %s",
1241 : : mp_dir_path);
1242 : 0 : rte_errno = errno;
1243 : 0 : goto closedir_fail;
1244 : : }
1245 : :
1246 [ + + ]: 14 : while ((ent = readdir(mp_dir))) {
1247 : : char path[UNIX_PATH_MAX];
1248 : :
1249 [ + + ]: 13 : if (fnmatch(mp_filter, ent->d_name, 0) != 0)
1250 : 12 : continue;
1251 : :
1252 [ - + ]: 1 : if (snprintf(path, sizeof(path), "%s/%s", mp_dir_path, ent->d_name)
1253 : : >= (int)sizeof(path)) {
1254 : 0 : EAL_LOG(ERR, "Unix domain path %s/%s too long", mp_dir_path, ent->d_name);
1255 : : ret = -1;
1256 [ - + ]: 1 : } else if (mp_request_async(path, copy, param, ts))
1257 : : ret = -1;
1258 : : }
1259 : :
1260 : : /*
1261 : : * On partial failure, roll back all queued requests. We hold the lock
1262 : : * so no one else touches the queue. All requests in this batch share
1263 : : * the same param pointer. Stale alarms will fire and harmlessly find
1264 : : * nothing via ID-based lookup.
1265 : : */
1266 [ - + - - ]: 1 : if (ret != 0 && reply->nb_sent > 0) {
1267 : : struct pending_request *r, *tmp;
1268 : :
1269 [ # # ]: 0 : RTE_TAILQ_FOREACH_SAFE(r, &pending_requests.requests, next, tmp) {
1270 [ # # # # ]: 0 : if (r->type == REQUEST_TYPE_ASYNC && r->async.param == param) {
1271 [ # # ]: 0 : TAILQ_REMOVE(&pending_requests.requests, r, next);
1272 : 0 : free(r->reply);
1273 : : /* r->request == copy, freed below after the loop */
1274 : 0 : free(r);
1275 : : }
1276 : : }
1277 : : /* requests on the queue were removed so keep things consistent */
1278 : 0 : reply->nb_sent = 0;
1279 : : }
1280 : :
1281 : : /* if we didn't send anything, put dummy request on the queue
1282 : : * and set a minimum-delay alarm so the callback fires immediately.
1283 : : */
1284 [ + - - + ]: 1 : if (ret == 0 && reply->nb_sent == 0) {
1285 [ # # ]: 0 : TAILQ_INSERT_HEAD(&pending_requests.requests, dummy, next);
1286 : : dummy_used = true;
1287 : :
1288 [ # # ]: 0 : if (rte_eal_alarm_set(1, async_reply_handle,
1289 : 0 : (void *)(uintptr_t)dummy->id) < 0) {
1290 : 0 : EAL_LOG(ERR, "Fail to set alarm for dummy request");
1291 : : /* roll back the changes */
1292 [ # # ]: 0 : TAILQ_REMOVE(&pending_requests.requests, dummy, next);
1293 : : dummy_used = false;
1294 : : ret = -1;
1295 : 0 : goto closedir_fail;
1296 : : }
1297 : : }
1298 : :
1299 : : /* finally, unlock the queue */
1300 : 1 : pthread_mutex_unlock(&pending_requests.lock);
1301 : :
1302 : : /* unlock the directory */
1303 : 1 : flock(dir_fd, LOCK_UN);
1304 : :
1305 : : /* dir_fd automatically closed on closedir */
1306 : 1 : closedir(mp_dir);
1307 : :
1308 : : /* if dummy was unused, free it */
1309 [ + - ]: 1 : if (!dummy_used)
1310 : 1 : free(dummy);
1311 : : /* if nothing was sent, nobody owns copy/param */
1312 [ - + ]: 1 : if (ret != 0) {
1313 : 0 : free(param);
1314 : 0 : free(copy);
1315 : : }
1316 : :
1317 : : return ret;
1318 : 0 : closedir_fail:
1319 : 0 : closedir(mp_dir);
1320 : 0 : unlock_fail:
1321 : 0 : pthread_mutex_unlock(&pending_requests.lock);
1322 : 0 : fail:
1323 : 0 : free(dummy);
1324 : 0 : free(param);
1325 : 0 : free(copy);
1326 : 0 : return -1;
1327 : : }
1328 : :
1329 : : RTE_EXPORT_SYMBOL(rte_mp_reply)
1330 : : int
1331 : 32 : rte_mp_reply(struct rte_mp_msg *msg, const char *peer)
1332 : : {
1333 : 32 : EAL_LOG(DEBUG, "reply: %s", msg->name);
1334 : : const struct internal_config *internal_conf =
1335 : 32 : eal_get_internal_configuration();
1336 : :
1337 [ + - ]: 32 : if (check_input(msg) != 0)
1338 : : return -1;
1339 : :
1340 [ - + ]: 32 : if (peer == NULL) {
1341 : 0 : EAL_LOG(ERR, "peer is not specified");
1342 : 0 : rte_errno = EINVAL;
1343 : 0 : return -1;
1344 : : }
1345 : :
1346 [ - + ]: 32 : if (internal_conf->no_shconf) {
1347 : 0 : EAL_LOG(DEBUG, "No shared files mode enabled, IPC is disabled");
1348 : 0 : return 0;
1349 : : }
1350 : :
1351 : 32 : return mp_send(msg, peer, MP_REP);
1352 : : }
1353 : :
1354 : : /* Internally, the status of the mp feature is represented as a three-state:
1355 : : * - "unknown" as long as no secondary process attached to a primary process
1356 : : * and there was no call to rte_mp_disable yet,
1357 : : * - "enabled" as soon as a secondary process attaches to a primary process,
1358 : : * - "disabled" when a primary process successfully called rte_mp_disable,
1359 : : */
1360 : : enum mp_status {
1361 : : MP_STATUS_UNKNOWN,
1362 : : MP_STATUS_DISABLED,
1363 : : MP_STATUS_ENABLED,
1364 : : };
1365 : :
1366 : : static bool
1367 : 162 : set_mp_status(enum mp_status status)
1368 : : {
1369 : 162 : struct rte_mem_config *mcfg = rte_eal_get_configuration()->mem_config;
1370 : : uint8_t expected;
1371 : : uint8_t desired;
1372 : :
1373 : : expected = MP_STATUS_UNKNOWN;
1374 : 162 : desired = status;
1375 [ + + ]: 162 : if (rte_atomic_compare_exchange_strong_explicit(&mcfg->mp_status, &expected, desired,
1376 : : rte_memory_order_relaxed, rte_memory_order_relaxed))
1377 : : return true;
1378 : :
1379 : 150 : return rte_atomic_load_explicit(&mcfg->mp_status, rte_memory_order_relaxed) == desired;
1380 : : }
1381 : :
1382 : : RTE_EXPORT_SYMBOL(rte_mp_disable)
1383 : : bool
1384 : 130 : rte_mp_disable(void)
1385 : : {
1386 : 130 : return set_mp_status(MP_STATUS_DISABLED);
1387 : : }
1388 : :
1389 : : bool
1390 : 32 : __rte_mp_enable(void)
1391 : : {
1392 : 32 : return set_mp_status(MP_STATUS_ENABLED);
1393 : : }
|
