1 /* SPDX-License-Identifier: BSD-3-Clause
2 * Copyright(c) 2016-2018 Intel Corporation
17 #include <sys/types.h>
18 #include <sys/socket.h>
22 #include <rte_common.h>
23 #include <rte_cycles.h>
25 #include <rte_errno.h>
26 #include <rte_lcore.h>
29 #include "eal_private.h"
30 #include "eal_filesystem.h"
31 #include "eal_internal_cfg.h"
33 static int mp_fd = -1;
34 static char mp_filter[PATH_MAX]; /* Filter for secondary process sockets */
35 static char mp_dir_path[PATH_MAX]; /* The directory path for all mp sockets */
36 static pthread_mutex_t mp_mutex_action = PTHREAD_MUTEX_INITIALIZER;
39 TAILQ_ENTRY(action_entry) next;
40 char action_name[RTE_MP_MAX_NAME_LEN];
44 /** Double linked list of actions. */
45 TAILQ_HEAD(action_entry_list, action_entry);
47 static struct action_entry_list action_entry_list =
48 TAILQ_HEAD_INITIALIZER(action_entry_list);
51 MP_MSG, /* Share message with peers, will not block */
52 MP_REQ, /* Request for information, Will block for a reply */
53 MP_REP, /* Response to previously-received request */
56 struct mp_msg_internal {
58 struct rte_mp_msg msg;
62 TAILQ_ENTRY(sync_request) next;
65 struct rte_mp_msg *request;
66 struct rte_mp_msg *reply;
70 TAILQ_HEAD(sync_request_list, sync_request);
73 struct sync_request_list requests;
76 .requests = TAILQ_HEAD_INITIALIZER(sync_requests.requests),
77 .lock = PTHREAD_MUTEX_INITIALIZER
80 static struct sync_request *
81 find_sync_request(const char *dst, const char *act_name)
83 struct sync_request *r;
85 TAILQ_FOREACH(r, &sync_requests.requests, next) {
86 if (!strcmp(r->dst, dst) &&
87 !strcmp(r->request->name, act_name))
95 rte_eal_primary_proc_alive(const char *config_file_path)
100 config_fd = open(config_file_path, O_RDONLY);
104 path = eal_runtime_config_path();
105 config_fd = open(path, O_RDONLY);
110 int ret = lockf(config_fd, F_TEST, 0);
116 static struct action_entry *
117 find_action_entry_by_name(const char *name)
119 struct action_entry *entry;
121 TAILQ_FOREACH(entry, &action_entry_list, next) {
122 if (strncmp(entry->action_name, name, RTE_MP_MAX_NAME_LEN) == 0)
130 validate_action_name(const char *name)
133 RTE_LOG(ERR, EAL, "Action name cannot be NULL\n");
137 if (strnlen(name, RTE_MP_MAX_NAME_LEN) == 0) {
138 RTE_LOG(ERR, EAL, "Length of action name is zero\n");
142 if (strnlen(name, RTE_MP_MAX_NAME_LEN) == RTE_MP_MAX_NAME_LEN) {
149 int __rte_experimental
150 rte_mp_action_register(const char *name, rte_mp_t action)
152 struct action_entry *entry;
154 if (validate_action_name(name))
157 entry = malloc(sizeof(struct action_entry));
162 strcpy(entry->action_name, name);
163 entry->action = action;
165 pthread_mutex_lock(&mp_mutex_action);
166 if (find_action_entry_by_name(name) != NULL) {
167 pthread_mutex_unlock(&mp_mutex_action);
172 TAILQ_INSERT_TAIL(&action_entry_list, entry, next);
173 pthread_mutex_unlock(&mp_mutex_action);
177 void __rte_experimental
178 rte_mp_action_unregister(const char *name)
180 struct action_entry *entry;
182 if (validate_action_name(name))
185 pthread_mutex_lock(&mp_mutex_action);
186 entry = find_action_entry_by_name(name);
188 pthread_mutex_unlock(&mp_mutex_action);
191 TAILQ_REMOVE(&action_entry_list, entry, next);
192 pthread_mutex_unlock(&mp_mutex_action);
197 read_msg(struct mp_msg_internal *m, struct sockaddr_un *s)
202 char control[CMSG_SPACE(sizeof(m->msg.fds))];
203 struct cmsghdr *cmsg;
204 int buflen = sizeof(*m) - sizeof(m->msg.fds);
206 memset(&msgh, 0, sizeof(msgh));
208 iov.iov_len = buflen;
211 msgh.msg_namelen = sizeof(*s);
214 msgh.msg_control = control;
215 msgh.msg_controllen = sizeof(control);
217 msglen = recvmsg(mp_fd, &msgh, 0);
219 RTE_LOG(ERR, EAL, "recvmsg failed, %s\n", strerror(errno));
223 if (msglen != buflen || (msgh.msg_flags & (MSG_TRUNC | MSG_CTRUNC))) {
224 RTE_LOG(ERR, EAL, "truncted msg\n");
228 /* read auxiliary FDs if any */
229 for (cmsg = CMSG_FIRSTHDR(&msgh); cmsg != NULL;
230 cmsg = CMSG_NXTHDR(&msgh, cmsg)) {
231 if ((cmsg->cmsg_level == SOL_SOCKET) &&
232 (cmsg->cmsg_type == SCM_RIGHTS)) {
233 memcpy(m->msg.fds, CMSG_DATA(cmsg), sizeof(m->msg.fds));
242 process_msg(struct mp_msg_internal *m, struct sockaddr_un *s)
244 struct sync_request *sync_req;
245 struct action_entry *entry;
246 struct rte_mp_msg *msg = &m->msg;
247 rte_mp_t action = NULL;
249 RTE_LOG(DEBUG, EAL, "msg: %s\n", msg->name);
251 if (m->type == MP_REP) {
252 pthread_mutex_lock(&sync_requests.lock);
253 sync_req = find_sync_request(s->sun_path, msg->name);
255 memcpy(sync_req->reply, msg, sizeof(*msg));
256 sync_req->reply_received = 1;
257 pthread_cond_signal(&sync_req->cond);
259 RTE_LOG(ERR, EAL, "Drop mp reply: %s\n", msg->name);
260 pthread_mutex_unlock(&sync_requests.lock);
264 pthread_mutex_lock(&mp_mutex_action);
265 entry = find_action_entry_by_name(msg->name);
267 action = entry->action;
268 pthread_mutex_unlock(&mp_mutex_action);
271 RTE_LOG(ERR, EAL, "Cannot find action: %s\n", msg->name);
272 else if (action(msg, s->sun_path) < 0)
273 RTE_LOG(ERR, EAL, "Fail to handle message: %s\n", msg->name);
277 mp_handle(void *arg __rte_unused)
279 struct mp_msg_internal msg;
280 struct sockaddr_un sa;
283 if (read_msg(&msg, &sa) == 0)
284 process_msg(&msg, &sa);
293 struct sockaddr_un un;
294 const char *prefix = eal_mp_socket_path();
296 mp_fd = socket(AF_UNIX, SOCK_DGRAM, 0);
298 RTE_LOG(ERR, EAL, "failed to create unix socket\n");
302 memset(&un, 0, sizeof(un));
303 un.sun_family = AF_UNIX;
304 if (rte_eal_process_type() == RTE_PROC_PRIMARY)
305 snprintf(un.sun_path, sizeof(un.sun_path), "%s", prefix);
307 snprintf(un.sun_path, sizeof(un.sun_path), "%s_%d_%"PRIx64,
308 prefix, getpid(), rte_rdtsc());
310 unlink(un.sun_path); /* May still exist since last run */
311 if (bind(mp_fd, (struct sockaddr *)&un, sizeof(un)) < 0) {
312 RTE_LOG(ERR, EAL, "failed to bind %s: %s\n",
313 un.sun_path, strerror(errno));
318 RTE_LOG(INFO, EAL, "Multi-process socket %s\n", un.sun_path);
323 unlink_sockets(const char *filter)
329 mp_dir = opendir(mp_dir_path);
331 RTE_LOG(ERR, EAL, "Unable to open directory %s\n", mp_dir_path);
334 dir_fd = dirfd(mp_dir);
336 while ((ent = readdir(mp_dir))) {
337 if (fnmatch(filter, ent->d_name, 0) == 0)
338 unlinkat(dir_fd, ent->d_name, 0);
346 unlink_socket_by_path(const char *path)
349 char *fullpath = strdup(path);
353 filename = basename(fullpath);
354 unlink_sockets(filename);
356 RTE_LOG(INFO, EAL, "Remove socket %s\n", path);
360 rte_mp_channel_init(void)
362 char thread_name[RTE_MAX_THREAD_NAME_LEN];
366 snprintf(mp_filter, PATH_MAX, ".%s_unix_*",
367 internal_config.hugefile_prefix);
369 path = strdup(eal_mp_socket_path());
370 snprintf(mp_dir_path, PATH_MAX, "%s", dirname(path));
373 if (rte_eal_process_type() == RTE_PROC_PRIMARY &&
374 unlink_sockets(mp_filter)) {
375 RTE_LOG(ERR, EAL, "failed to unlink mp sockets\n");
379 if (open_socket_fd() < 0)
382 if (pthread_create(&tid, NULL, mp_handle, NULL) < 0) {
383 RTE_LOG(ERR, EAL, "failed to create mp thead: %s\n",
390 /* try best to set thread name */
391 snprintf(thread_name, RTE_MAX_THREAD_NAME_LEN, "rte_mp_handle");
392 rte_thread_setname(tid, thread_name);
397 * Return -1, as fail to send message and it's caused by the local side.
398 * Return 0, as fail to send message and it's caused by the remote side.
399 * Return 1, as succeed to send message.
403 send_msg(const char *dst_path, struct rte_mp_msg *msg, int type)
408 struct cmsghdr *cmsg;
409 struct sockaddr_un dst;
410 struct mp_msg_internal m;
411 int fd_size = msg->num_fds * sizeof(int);
412 char control[CMSG_SPACE(fd_size)];
415 memcpy(&m.msg, msg, sizeof(*msg));
417 memset(&dst, 0, sizeof(dst));
418 dst.sun_family = AF_UNIX;
419 snprintf(dst.sun_path, sizeof(dst.sun_path), "%s", dst_path);
421 memset(&msgh, 0, sizeof(msgh));
422 memset(control, 0, sizeof(control));
425 iov.iov_len = sizeof(m) - sizeof(msg->fds);
427 msgh.msg_name = &dst;
428 msgh.msg_namelen = sizeof(dst);
431 msgh.msg_control = control;
432 msgh.msg_controllen = sizeof(control);
434 cmsg = CMSG_FIRSTHDR(&msgh);
435 cmsg->cmsg_len = CMSG_LEN(fd_size);
436 cmsg->cmsg_level = SOL_SOCKET;
437 cmsg->cmsg_type = SCM_RIGHTS;
438 memcpy(CMSG_DATA(cmsg), msg->fds, fd_size);
441 snd = sendmsg(mp_fd, &msgh, 0);
442 } while (snd < 0 && errno == EINTR);
446 /* Check if it caused by peer process exits */
447 if (errno == -ECONNREFUSED) {
448 /* We don't unlink the primary's socket here */
449 if (rte_eal_process_type() == RTE_PROC_PRIMARY)
450 unlink_socket_by_path(dst_path);
453 if (errno == -ENOBUFS) {
454 RTE_LOG(ERR, EAL, "Peer cannot receive message %s\n",
458 RTE_LOG(ERR, EAL, "failed to send to (%s) due to %s\n",
459 dst_path, strerror(errno));
467 mp_send(struct rte_mp_msg *msg, const char *peer, int type)
473 if (!peer && (rte_eal_process_type() == RTE_PROC_SECONDARY))
474 peer = eal_mp_socket_path();
477 if (send_msg(peer, msg, type) < 0)
483 /* broadcast to all secondary processes */
484 mp_dir = opendir(mp_dir_path);
486 RTE_LOG(ERR, EAL, "Unable to open directory %s\n",
491 while ((ent = readdir(mp_dir))) {
492 if (fnmatch(mp_filter, ent->d_name, 0) != 0)
495 if (send_msg(ent->d_name, msg, type) < 0)
504 check_input(const struct rte_mp_msg *msg)
507 RTE_LOG(ERR, EAL, "Msg cannot be NULL\n");
512 if (validate_action_name(msg->name))
515 if (msg->len_param > RTE_MP_MAX_PARAM_LEN) {
516 RTE_LOG(ERR, EAL, "Message data is too long\n");
521 if (msg->num_fds > RTE_MP_MAX_FD_NUM) {
522 RTE_LOG(ERR, EAL, "Cannot send more than %d FDs\n",
531 int __rte_experimental
532 rte_mp_sendmsg(struct rte_mp_msg *msg)
534 if (!check_input(msg))
537 RTE_LOG(DEBUG, EAL, "sendmsg: %s\n", msg->name);
538 return mp_send(msg, NULL, MP_MSG);
542 mp_request_one(const char *dst, struct rte_mp_msg *req,
543 struct rte_mp_reply *reply, const struct timespec *ts)
547 struct rte_mp_msg msg, *tmp;
548 struct sync_request sync_req, *exist;
550 sync_req.reply_received = 0;
551 strcpy(sync_req.dst, dst);
552 sync_req.request = req;
553 sync_req.reply = &msg;
554 pthread_cond_init(&sync_req.cond, NULL);
556 pthread_mutex_lock(&sync_requests.lock);
557 exist = find_sync_request(dst, req->name);
559 TAILQ_INSERT_TAIL(&sync_requests.requests, &sync_req, next);
560 pthread_mutex_unlock(&sync_requests.lock);
562 RTE_LOG(ERR, EAL, "A pending request %s:%s\n", dst, req->name);
567 ret = send_msg(dst, req, MP_REQ);
569 RTE_LOG(ERR, EAL, "Fail to send request %s:%s\n",
577 pthread_mutex_lock(&sync_requests.lock);
579 pthread_cond_timedwait(&sync_req.cond, &sync_requests.lock, ts);
580 /* Check spurious wakeups */
581 if (sync_req.reply_received == 1)
583 /* Check if time is out */
584 if (gettimeofday(&now, NULL) < 0)
586 if (now.tv_sec < ts->tv_sec)
588 else if (now.tv_sec == ts->tv_sec &&
589 now.tv_usec * 1000 < ts->tv_nsec)
592 /* We got the lock now */
593 TAILQ_REMOVE(&sync_requests.requests, &sync_req, next);
594 pthread_mutex_unlock(&sync_requests.lock);
596 if (sync_req.reply_received == 0) {
597 RTE_LOG(ERR, EAL, "Fail to recv reply for request %s:%s\n",
599 rte_errno = -ETIMEDOUT;
603 tmp = realloc(reply->msgs, sizeof(msg) * (reply->nb_received + 1));
605 RTE_LOG(ERR, EAL, "Fail to alloc reply for request %s:%s\n",
610 memcpy(&tmp[reply->nb_received], &msg, sizeof(msg));
612 reply->nb_received++;
616 int __rte_experimental
617 rte_mp_request(struct rte_mp_msg *req, struct rte_mp_reply *reply,
618 const struct timespec *ts)
626 RTE_LOG(DEBUG, EAL, "request: %s\n", req->name);
628 if (check_input(req) == false)
630 if (gettimeofday(&now, NULL) < 0) {
631 RTE_LOG(ERR, EAL, "Faile to get current time\n");
636 end.tv_nsec = (now.tv_usec * 1000 + ts->tv_nsec) % 1000000000;
637 end.tv_sec = now.tv_sec + ts->tv_sec +
638 (now.tv_usec * 1000 + ts->tv_nsec) / 1000000000;
641 reply->nb_received = 0;
644 /* for secondary process, send request to the primary process only */
645 if (rte_eal_process_type() == RTE_PROC_SECONDARY)
646 return mp_request_one(eal_mp_socket_path(), req, reply, &end);
648 /* for primary process, broadcast request, and collect reply 1 by 1 */
649 mp_dir = opendir(mp_dir_path);
651 RTE_LOG(ERR, EAL, "Unable to open directory %s\n", mp_dir_path);
656 while ((ent = readdir(mp_dir))) {
657 if (fnmatch(mp_filter, ent->d_name, 0) != 0)
660 if (mp_request_one(ent->d_name, req, reply, &end))
668 int __rte_experimental
669 rte_mp_reply(struct rte_mp_msg *msg, const char *peer)
672 RTE_LOG(DEBUG, EAL, "reply: %s\n", msg->name);
674 if (check_input(msg) == false)
678 RTE_LOG(ERR, EAL, "peer is not specified\n");
683 return mp_send(msg, peer, MP_REP);