From mboxrd@z Thu Jan 1 00:00:00 1970 Return-Path: Received: from mga02.intel.com (mga02.intel.com [134.134.136.20]) by dpdk.org (Postfix) with ESMTP id 83BF51B1B6 for ; Thu, 11 Jan 2018 05:05:48 +0100 (CET) X-Amp-Result: SKIPPED(no attachment in message) X-Amp-File-Uploaded: False Received: from fmsmga006.fm.intel.com ([10.253.24.20]) by orsmga101.jf.intel.com with ESMTP/TLS/DHE-RSA-AES256-GCM-SHA384; 10 Jan 2018 20:05:47 -0800 X-ExtLoop1: 1 X-IronPort-AV: E=Sophos;i="5.46,343,1511856000"; d="scan'208";a="194041983" Received: from dpdk06.sh.intel.com ([10.67.110.196]) by fmsmga006.fm.intel.com with ESMTP; 10 Jan 2018 20:05:46 -0800 From: Jianfeng Tan To: dev@dpdk.org Cc: anatoly.burakov@intel.com, bruce.richardson@intel.com, konstantin.ananyev@intel.com, thomas@monjalon.net, Jianfeng Tan Date: Thu, 11 Jan 2018 04:07:31 +0000 Message-Id: <1515643654-129489-2-git-send-email-jianfeng.tan@intel.com> X-Mailer: git-send-email 2.7.4 In-Reply-To: <1515643654-129489-1-git-send-email-jianfeng.tan@intel.com> References: <1512067450-59203-1-git-send-email-jianfeng.tan@intel.com> <1515643654-129489-1-git-send-email-jianfeng.tan@intel.com> Subject: [dpdk-dev] [PATCH v2 1/4] eal: add channel for multi-process communication X-BeenThere: dev@dpdk.org X-Mailman-Version: 2.1.15 Precedence: list List-Id: DPDK patches and discussions List-Unsubscribe: , List-Archive: List-Post: List-Help: List-Subscribe: , X-List-Received-Date: Thu, 11 Jan 2018 04:05:49 -0000 Previouly, there are three channels for multi-process (i.e., primary/secondary) communication. 1. Config-file based channel, in which, the primary process writes info into a pre-defined config file, and the secondary process reads the info out. 2. vfio submodule has its own channel based on unix socket for the secondary process to get container fd and group fd from the primary process. 3. pdump submodule also has its own channel based on unix socket for packet dump. It'd be good to have a generic communication channel for multi-process communication to accomodate the requirements including: a. Secondary wants to send info to primary, for example, secondary would like to send request (about some specific vdev to primary). b. Sending info at any time, instead of just initialization time. c. Share FDs with the other side, for vdev like vhost, related FDs (memory region, kick) should be shared. d. A send message request needs the other side to response immediately. This patch proposes to create a communication channel, based on datagram unix socket, for above requirements. Each process will block on a unix socket waiting for messages from the peers. Three new APIs are added: 1. rte_eal_mp_action_register() is used to register an action, indexed by a string, when a component at receiver side would like to response the messages from the peer processe. 2. rte_eal_mp_action_unregister() is used to unregister the action if the calling component does not want to response the messages. 3. rte_eal_mp_sendmsg() is used to send a message, and returns immediately. If there are 1:n primary:secondary processes, the primary process will send n messages. Suggested-by: Konstantin Ananyev Signed-off-by: Jianfeng Tan --- lib/librte_eal/common/eal_common_proc.c | 388 ++++++++++++++++++++++++++++++++ lib/librte_eal/common/eal_filesystem.h | 17 ++ lib/librte_eal/common/eal_private.h | 10 + lib/librte_eal/common/include/rte_eal.h | 69 ++++++ lib/librte_eal/linuxapp/eal/eal.c | 8 + lib/librte_eal/rte_eal_version.map | 9 + 6 files changed, 501 insertions(+) diff --git a/lib/librte_eal/common/eal_common_proc.c b/lib/librte_eal/common/eal_common_proc.c index 40fa982..d700e9e 100644 --- a/lib/librte_eal/common/eal_common_proc.c +++ b/lib/librte_eal/common/eal_common_proc.c @@ -5,11 +5,55 @@ #include #include #include +#include +#include +#include +#include +#include +#include +#include + +#include #include +#include +#include +#include +#include "eal_private.h" #include "eal_filesystem.h" #include "eal_internal_cfg.h" +#define MAX_SECONDARY_PROCS 8 +#define MAX_ACTION_NAME_LEN 64 +#define MAX_UNIX_PATH_LEN 104 +#define MAX_MSG_LENGTH 1024 +#define SCM_MAX_FD 253 /* The max amount of fds */ + +static int mp_fd = -1; +static char *mp_sec_sockets[MAX_SECONDARY_PROCS]; +static pthread_mutex_t mp_mutex_action = PTHREAD_MUTEX_INITIALIZER; + +struct action_entry { + TAILQ_ENTRY(action_entry) next; /**< Next attached action entry */ + +#define MAX_ACTION_NAME_LEN 64 + char action_name[MAX_ACTION_NAME_LEN]; + rte_eal_mp_t action; +}; + +/** Double linked list of actions. */ +TAILQ_HEAD(action_entry_list, action_entry); + +static struct action_entry_list action_entry_list = + TAILQ_HEAD_INITIALIZER(action_entry_list); + +struct mp_msghdr { + char action_name[MAX_ACTION_NAME_LEN]; + int fds_num; + int len_params; + char params[0]; +} __rte_packed; + int rte_eal_primary_proc_alive(const char *config_file_path) { @@ -31,3 +75,347 @@ rte_eal_primary_proc_alive(const char *config_file_path) return !!ret; } + +static struct action_entry * +find_action_entry_by_name(const char *name) +{ + int len = strlen(name); + struct action_entry *entry; + + TAILQ_FOREACH(entry, &action_entry_list, next) { + if (strncmp(entry->action_name, name, len) == 0) + break; + } + + return entry; +} + +int +rte_eal_mp_action_register(const char *action_name, rte_eal_mp_t action) +{ + struct action_entry *entry = malloc(sizeof(struct action_entry)); + + if (entry == NULL) { + rte_errno = -ENOMEM; + return -1; + } + + if (strlen(action_name) > MAX_ACTION_NAME_LEN) { + rte_errno = -E2BIG; + return -1; + } + + pthread_mutex_lock(&mp_mutex_action); + if (find_action_entry_by_name(action_name) != NULL) { + free(entry); + rte_errno = -EEXIST; + return -1; + } + strncpy(entry->action_name, action_name, MAX_ACTION_NAME_LEN); + entry->action = action; + TAILQ_INSERT_TAIL(&action_entry_list, entry, next); + pthread_mutex_unlock(&mp_mutex_action); + return 0; +} + +void +rte_eal_mp_action_unregister(const char *name) +{ + struct action_entry *entry; + + pthread_mutex_lock(&mp_mutex_action); + entry = find_action_entry_by_name(name); + TAILQ_REMOVE(&action_entry_list, entry, next); + free(entry); + pthread_mutex_unlock(&mp_mutex_action); +} + +static int +read_msg(int fd, char *buf, int buflen, int *fds, int fds_num) +{ + int ret; + struct iovec iov; + struct msghdr msgh; + size_t fdsize = fds_num * sizeof(int); + char control[CMSG_SPACE(fdsize)]; + struct cmsghdr *cmsg; + + memset(&msgh, 0, sizeof(msgh)); + iov.iov_base = buf; + iov.iov_len = buflen; + + msgh.msg_iov = &iov; + msgh.msg_iovlen = 1; + msgh.msg_control = control; + msgh.msg_controllen = sizeof(control); + + ret = recvmsg(fd, &msgh, 0); + if (ret < 0) { + RTE_LOG(ERR, EAL, "recvmsg failed, %s\n", strerror(errno)); + return -1; + } + + if (msgh.msg_flags & (MSG_TRUNC | MSG_CTRUNC)) { + RTE_LOG(ERR, EAL, "truncted msg\n"); + return -1; + } + + /* read auxiliary FDs if any */ + for (cmsg = CMSG_FIRSTHDR(&msgh); cmsg != NULL; + cmsg = CMSG_NXTHDR(&msgh, cmsg)) { + if ((cmsg->cmsg_level == SOL_SOCKET) && + (cmsg->cmsg_type == SCM_RIGHTS)) { + memcpy(fds, CMSG_DATA(cmsg), fdsize); + break; + } + } + + return ret; +} + +static int +process_msg(struct mp_msghdr *hdr, int len, int fds[]) +{ + int ret; + int params_len; + struct action_entry *entry; + + RTE_LOG(DEBUG, EAL, "msg: %s\n", hdr->action_name); + + pthread_mutex_lock(&mp_mutex_action); + entry = find_action_entry_by_name(hdr->action_name); + if (entry == NULL) { + RTE_LOG(ERR, EAL, "cannot find action by: %s\n", + hdr->action_name); + pthread_mutex_unlock(&mp_mutex_action); + return -1; + } + + params_len = len - sizeof(struct mp_msghdr); + ret = entry->action(hdr->params, params_len, fds, hdr->fds_num); + pthread_mutex_unlock(&mp_mutex_action); + return ret; + +} + +static void * +mp_handle(void *arg __rte_unused) +{ + int len; + int fds[SCM_MAX_FD]; + char buf[MAX_MSG_LENGTH]; + + while (1) { + len = read_msg(mp_fd, buf, MAX_MSG_LENGTH, fds, SCM_MAX_FD); + if (len > 0) + process_msg((struct mp_msghdr *)buf, len, fds); + } + + return NULL; +} + +static inline const char * +get_unix_path(int is_server) +{ + static char unix_path[MAX_UNIX_PATH_LEN]; + const char *prefix = eal_mp_unix_path(); + const char *suffix = (is_server) ? "" : "_c"; + + if (rte_eal_process_type() == RTE_PROC_PRIMARY) + snprintf(unix_path, MAX_UNIX_PATH_LEN, "%s%s", prefix, suffix); + else + snprintf(unix_path, MAX_UNIX_PATH_LEN, "%s%s_%d", + prefix, suffix, getpid()); + return unix_path; +} + +static int +open_unix_fd(int is_server) +{ + int fd; + struct sockaddr_un un; + + fd = socket(AF_UNIX, SOCK_DGRAM, 0); + if (fd < 0) { + RTE_LOG(ERR, EAL, "failed to create unix socket\n"); + return -1; + } + + memset(&un, 0, sizeof(un)); + un.sun_family = AF_UNIX; + snprintf(un.sun_path, MAX_UNIX_PATH_LEN, "%s", + get_unix_path(is_server)); + unlink(un.sun_path); /* May still exist since last run */ + if (bind(fd, (struct sockaddr *)&un, sizeof(un)) < 0) { + RTE_LOG(ERR, EAL, "failed to bind %s: %s\n", + un.sun_path, strerror(errno)); + close(fd); + return -1; + } + + RTE_LOG(INFO, EAL, "bind to %s\n", un.sun_path); + return fd; +} + +int +rte_eal_mp_channel_init(void) +{ + pthread_t tid; + char thread_name[RTE_MAX_THREAD_NAME_LEN]; + + mp_fd = open_unix_fd(1); + if (mp_fd < 0) + return -1; + + if (pthread_create(&tid, NULL, mp_handle, NULL) < 0) { + RTE_LOG(ERR, EAL, "failed to create mp handle thead: %s\n", + strerror(errno)); + goto error; + } + + snprintf(thread_name, RTE_MAX_THREAD_NAME_LEN, "rte_mp_handle"); + if (rte_thread_setname(tid, thread_name) < 0) { + RTE_LOG(ERR, EAL, "failed to set thead name\n"); + goto error; + } + + return 0; +error: + close(mp_fd); + mp_fd = -1; + return -1; +} + +static inline struct mp_msghdr * +format_msg(const char *act_name, const void *p, int len_params, int fds_num) +{ + int len_msg; + struct mp_msghdr *msg; + + len_msg = sizeof(struct mp_msghdr) + len_params; + if (len_msg > MAX_MSG_LENGTH) { + RTE_LOG(ERR, EAL, "Message is too long\n"); + rte_errno = -EINVAL; + return NULL; + } + + msg = malloc(len_msg); + if (!msg) { + RTE_LOG(ERR, EAL, "Cannot alloc memory for msg\n"); + rte_errno = -ENOMEM; + return NULL; + } + memset(msg, 0, len_msg); + strcpy(msg->action_name, act_name); + msg->fds_num = fds_num; + msg->len_params = len_params; + memcpy(msg->params, p, len_params); + return msg; +} + +static int +send_msg(int fd, const char *dst_path, struct mp_msghdr *msg, int fds[]) +{ + int ret; + struct msghdr msgh; + struct iovec iov; + size_t fd_size = msg->fds_num * sizeof(int); + char control[CMSG_SPACE(fd_size)]; + struct cmsghdr *cmsg; + struct sockaddr_un dst; + + memset(&dst, 0, sizeof(dst)); + dst.sun_family = AF_UNIX; + snprintf(dst.sun_path, MAX_UNIX_PATH_LEN, "%s", dst_path); + + memset(&msgh, 0, sizeof(msgh)); + memset(control, 0, sizeof(control)); + + iov.iov_base = (uint8_t *)msg; + iov.iov_len = sizeof(struct mp_msghdr) + msg->len_params; + + msgh.msg_name = &dst; + msgh.msg_namelen = sizeof(dst); + msgh.msg_iov = &iov; + msgh.msg_iovlen = 1; + msgh.msg_control = control; + msgh.msg_controllen = sizeof(control); + + cmsg = CMSG_FIRSTHDR(&msgh); + cmsg->cmsg_len = CMSG_LEN(fd_size); + cmsg->cmsg_level = SOL_SOCKET; + cmsg->cmsg_type = SCM_RIGHTS; + memcpy(CMSG_DATA(cmsg), fds, fd_size); + + do { + ret = sendmsg(fd, &msgh, 0); + } while (ret < 0 && errno == EINTR); + + if (ret < 0) { + RTE_LOG(ERR, EAL, "failed to send msg: %s\n", strerror(errno)); + + if (rte_eal_process_type() == RTE_PROC_PRIMARY) + RTE_LOG(ERR, EAL, "secondary process (%s) exited\n", + dst_path); + else if (!rte_eal_primary_proc_alive(NULL)) + RTE_LOG(ERR, EAL, "primary process exited\n"); + + return 0; + } + + return 1; +} + +static int +mp_send(const char *action_name, + const void *params, + int len_params, + int fds[], + int fds_num) +{ + int i; + int n = 0; + int sockfd; + struct mp_msghdr *msg; + + if (fds_num > SCM_MAX_FD) { + RTE_LOG(ERR, EAL, "Cannot send more than %d FDs\n", SCM_MAX_FD); + rte_errno = -E2BIG; + return 0; + } + + msg = format_msg(action_name, params, len_params, fds_num); + if (msg == NULL) + return 0; + + if ((sockfd = open_unix_fd(0)) < 0) { + free(msg); + return 0; + } + + if (rte_eal_process_type() == RTE_PROC_PRIMARY) { + /* broadcast to all secondaries */ + for (i = 0; i < MAX_SECONDARY_PROCS; ++i) { + if (mp_sec_sockets[i] == NULL) + continue; + + n += send_msg(sockfd, mp_sec_sockets[i], msg, fds); + } + } else + n += send_msg(sockfd, eal_mp_unix_path(), msg, fds); + + free(msg); + close(sockfd); + return n; +} + +int +rte_eal_mp_sendmsg(const char *action_name, + const void *params, + int len_params, + int fds[], + int fds_num) +{ + RTE_LOG(DEBUG, EAL, "sendmsg: %s\n", action_name); + return mp_send(action_name, params, len_params, fds, fds_num); +} diff --git a/lib/librte_eal/common/eal_filesystem.h b/lib/librte_eal/common/eal_filesystem.h index e8959eb..e95399b 100644 --- a/lib/librte_eal/common/eal_filesystem.h +++ b/lib/librte_eal/common/eal_filesystem.h @@ -38,6 +38,23 @@ eal_runtime_config_path(void) return buffer; } +/** Path of primary/secondary communication unix socket file. */ +#define MP_UNIX_PATH_FMT "%s/.%s_unix" +static inline const char * +eal_mp_unix_path(void) +{ + static char buffer[PATH_MAX]; /* static so auto-zeroed */ + const char *directory = default_config_dir; + const char *home_dir = getenv("HOME"); + + if (getuid() != 0 && home_dir != NULL) + directory = home_dir; + snprintf(buffer, sizeof(buffer) - 1, MP_UNIX_PATH_FMT, + directory, internal_config.hugefile_prefix); + + return buffer; +} + /** Path of hugepage info file. */ #define HUGEPAGE_INFO_FMT "%s/.%s_hugepage_info" diff --git a/lib/librte_eal/common/eal_private.h b/lib/librte_eal/common/eal_private.h index c46dd8f..e36e3b5 100644 --- a/lib/librte_eal/common/eal_private.h +++ b/lib/librte_eal/common/eal_private.h @@ -195,4 +195,14 @@ int rte_eal_hugepage_attach(void); */ struct rte_bus *rte_bus_find_by_device_name(const char *str); +/** + * Create the unix channel for primary/secondary communication. + * + * @return + * 0 on success; + * (<0) on failure. + */ + +int rte_eal_mp_channel_init(void); + #endif /* _EAL_PRIVATE_H_ */ diff --git a/lib/librte_eal/common/include/rte_eal.h b/lib/librte_eal/common/include/rte_eal.h index 02fa109..9884c0b 100644 --- a/lib/librte_eal/common/include/rte_eal.h +++ b/lib/librte_eal/common/include/rte_eal.h @@ -186,6 +186,75 @@ int rte_eal_init(int argc, char **argv); int rte_eal_primary_proc_alive(const char *config_file_path); /** + * Action function typedef used by other components. + * + * As we create unix socket channel for primary/secondary communication, use + * this function typedef to register action for coming messages. + */ +typedef int (*rte_eal_mp_t)(const void *params, int len, + int fds[], int fds_num); + +/** + * Register an action function for primary/secondary communication. + * + * Call this function to register an action, if the calling component wants + * to response the messages from the corresponding component in its primary + * process or secondary processes. + * + * @param action_name + * The action_name argument plays as the nonredundant key to find the action. + * + * @param action + * The action argument is the function pointer to the action function. + * + * @return + * - 0 on success. + * - (<0) on failure. + */ +int rte_eal_mp_action_register(const char *action_name, rte_eal_mp_t action); + +/** + * Unregister an action function for primary/secondary communication. + * + * Call this function to unregister an action if the calling component does + * not want to response the messages from the corresponding component in its + * primary process or secondary processes. + * + * @param action_name + * The action_name argument plays as the nonredundant key to find the action. + * + */ +void rte_eal_mp_action_unregister(const char *name); + +/** + * Send a message to the peer process. + * + * This function will send a message which will be responsed by the action + * identified by action_name of the process on the other side. + * + * @param action_name + * The action_name argument is used to identify which action will be used. + * + * @param params + * The params argument contains the customized message. + * + * @param len_params + * The len_params argument is the length of the customized message. + * + * @param fds + * The fds argument is an array of fds sent with sendmsg. + * + * @param fds_num + * The fds_num argument is number of fds to be sent with sendmsg. + * + * @return + * - Returns the number of messages being sent successfully. + */ +int +rte_eal_mp_sendmsg(const char *action_name, const void *params, + int len_params, int fds[], int fds_num); + +/** * Usage function typedef used by the application usage function. * * Use this function typedef to define and call rte_set_application_usage_hook() diff --git a/lib/librte_eal/linuxapp/eal/eal.c b/lib/librte_eal/linuxapp/eal/eal.c index 229eec9..f231724 100644 --- a/lib/librte_eal/linuxapp/eal/eal.c +++ b/lib/librte_eal/linuxapp/eal/eal.c @@ -896,6 +896,14 @@ rte_eal_init(int argc, char **argv) eal_check_mem_on_local_socket(); + if (rte_eal_mp_channel_init() < 0) { + rte_eal_init_alert("failed to init mp channel\n"); + if (rte_eal_process_type() == RTE_PROC_PRIMARY) { + rte_errno = EFAULT; + return -1; + } + } + eal_thread_init_master(rte_config.master_lcore); ret = eal_thread_dump_affinity(cpuset, RTE_CPU_AFFINITY_STR_LEN); diff --git a/lib/librte_eal/rte_eal_version.map b/lib/librte_eal/rte_eal_version.map index f4f46c1..5dacde5 100644 --- a/lib/librte_eal/rte_eal_version.map +++ b/lib/librte_eal/rte_eal_version.map @@ -235,4 +235,13 @@ EXPERIMENTAL { rte_service_set_stats_enable; rte_service_start_with_defaults; +} DPDK_17.08; + +DPDK_18.02 { + global: + + rte_eal_mp_action_register; + rte_eal_mp_action_unregister; + rte_eal_mp_sendmsg; + } DPDK_17.11; -- 2.7.4