From mboxrd@z Thu Jan 1 00:00:00 1970 Return-Path: Received: from mga03.intel.com (mga03.intel.com [134.134.136.65]) by dpdk.org (Postfix) with ESMTP id 32BB7A48B for ; Thu, 25 Jan 2018 05:14:38 +0100 (CET) X-Amp-Result: SKIPPED(no attachment in message) X-Amp-File-Uploaded: False Received: from fmsmga007.fm.intel.com ([10.253.24.52]) by orsmga103.jf.intel.com with ESMTP/TLS/DHE-RSA-AES256-GCM-SHA384; 24 Jan 2018 20:14:37 -0800 X-ExtLoop1: 1 X-IronPort-AV: E=Sophos;i="5.46,409,1511856000"; d="scan'208";a="12847711" Received: from dpdk06.sh.intel.com ([10.67.110.196]) by fmsmga007.fm.intel.com with ESMTP; 24 Jan 2018 20:14:35 -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, 25 Jan 2018 04:16:22 +0000 Message-Id: <1516853783-108023-3-git-send-email-jianfeng.tan@intel.com> X-Mailer: git-send-email 2.7.4 In-Reply-To: <1516853783-108023-1-git-send-email-jianfeng.tan@intel.com> References: <1512067450-59203-1-git-send-email-jianfeng.tan@intel.com> <1516853783-108023-1-git-send-email-jianfeng.tan@intel.com> Subject: [dpdk-dev] [PATCH v3 2/3] eal: add synchronous 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, 25 Jan 2018 04:14:38 -0000 We need the synchronous way for multi-process communication, i.e., blockingly waiting for reply message when we send a request to the peer process. We add two APIs rte_eal_mp_request() and rte_eal_mp_reply() for such use case. By invoking rte_eal_mp_request(), a request message is sent out, and then it waits there for a reply message. The caller can specify the timeout. And the response messages will be collected and returned so that the caller can decide how to translate them. The API rte_eal_mp_reply() is always called by an mp action handler. Here we add another parameter for rte_eal_mp_t so that the action handler knows which peer address to reply. sender-process receiver-process ---------------------- ---------------- thread-n |_rte_eal_mp_request() ----------> mp-thread |_timedwait() |_process_msg() |_action() |_rte_eal_mp_reply() mp_thread <---------------------| |_process_msg() |_signal(send_thread) thread-m <----------| |_collect-reply * A secondary process is only allowed to talk to the primary process. * If there are multiple secondary processes for the primary proces, it will send request to peer1, collect response from peer1; then send request to peer2, collect reponse from peer2, and so on. * When thread-n is sending request, thread-m of that process can send request at the same time. * For pair , we guarantee that only one such request is on the fly. Suggested-by: Anatoly Burakov Suggested-by: Konstantin Ananyev Signed-off-by: Jianfeng Tan --- doc/guides/rel_notes/release_18_02.rst | 15 ++ lib/librte_eal/common/eal_common_proc.c | 237 +++++++++++++++++++++++++++++--- lib/librte_eal/common/include/rte_eal.h | 58 +++++++- lib/librte_eal/rte_eal_version.map | 3 + 4 files changed, 295 insertions(+), 18 deletions(-) diff --git a/doc/guides/rel_notes/release_18_02.rst b/doc/guides/rel_notes/release_18_02.rst index 00b3224..f6ed666 100644 --- a/doc/guides/rel_notes/release_18_02.rst +++ b/doc/guides/rel_notes/release_18_02.rst @@ -151,6 +151,21 @@ New Features renamed the application from SW PMD specific ``eventdev_pipeline_sw_pmd`` to PMD agnostic ``eventdev_pipeline``. +* **Added new multi-process communication channel** + + Added a generic channel in EAL for multi-process (primary/secondary) synchronous + and asynchronous communication. Each component who wants to reponse a message + shall register the action; and each process has a thread to receive the message + and invokes the registered action. The list of new APIs: + + * ``rte_eal_mp_register`` + * ``rte_eal_mp_unregister`` + * ``rte_eal_mp_sendmsg`` + * ``rte_eal_mp_request`` + * ``rte_eal_mp_reply`` + + Note as we changed to use the new channel for communication, applications cannot + talk with old version through the old (private) communication channel. API Changes ----------- diff --git a/lib/librte_eal/common/eal_common_proc.c b/lib/librte_eal/common/eal_common_proc.c index baeb7d1..69df943 100644 --- a/lib/librte_eal/common/eal_common_proc.c +++ b/lib/librte_eal/common/eal_common_proc.c @@ -12,6 +12,7 @@ #include #include #include +#include #include #include #include @@ -44,6 +45,50 @@ TAILQ_HEAD(action_entry_list, action_entry); static struct action_entry_list action_entry_list = TAILQ_HEAD_INITIALIZER(action_entry_list); +enum mp_type { + MP_MSG, /* Share message with peers, will not block */ + MP_REQ, /* Request for information, Will block for a reply */ + MP_REP, /* Response to previously-received request */ +}; + +struct mp_msg_internal { + int type; + struct rte_mp_msg msg; +}; + +struct sync_request { + int reply_received; + char dst[PATH_MAX]; + struct rte_mp_msg *request; + struct rte_mp_msg *reply; + pthread_cond_t cond; + TAILQ_ENTRY(sync_request) next; +}; + +TAILQ_HEAD(sync_request_list, sync_request); + +static struct { + struct sync_request_list requests; + pthread_mutex_t lock; +} sync_requests = { + .requests = TAILQ_HEAD_INITIALIZER(sync_requests.requests), + .lock = PTHREAD_MUTEX_INITIALIZER +}; + +static struct sync_request * +find_sync_request(const char *dst, const char *act_name) +{ + struct sync_request *r; + + TAILQ_FOREACH(r, &sync_requests.requests, next) { + if (!strcmp(r->dst, dst) && + !strcmp(r->request->name, act_name)) + break; + } + + return r; +} + int rte_eal_primary_proc_alive(const char *config_file_path) { @@ -147,19 +192,21 @@ rte_eal_mp_action_unregister(const char *name) } static int -read_msg(struct rte_mp_msg *msg) +read_msg(struct mp_msg_internal *m, struct sockaddr_un *s) { int msglen; struct iovec iov; struct msghdr msgh; - char control[CMSG_SPACE(sizeof(msg->fds))]; + char control[CMSG_SPACE(sizeof(m->msg.fds))]; struct cmsghdr *cmsg; - int buflen = sizeof(*msg) - sizeof(msg->fds); + int buflen = sizeof(*m) - sizeof(m->msg.fds); memset(&msgh, 0, sizeof(msgh)); - iov.iov_base = msg; + iov.iov_base = m; iov.iov_len = buflen; + msgh.msg_name = s; + msgh.msg_namelen = sizeof(*s); msgh.msg_iov = &iov; msgh.msg_iovlen = 1; msgh.msg_control = control; @@ -181,7 +228,7 @@ read_msg(struct rte_mp_msg *msg) cmsg = CMSG_NXTHDR(&msgh, cmsg)) { if ((cmsg->cmsg_level == SOL_SOCKET) && (cmsg->cmsg_type == SCM_RIGHTS)) { - memcpy(msg->fds, CMSG_DATA(cmsg), sizeof(msg->fds)); + memcpy(m->msg.fds, CMSG_DATA(cmsg), sizeof(m->msg.fds)); break; } } @@ -190,12 +237,28 @@ read_msg(struct rte_mp_msg *msg) } static void -process_msg(struct rte_mp_msg *msg) +process_msg(struct mp_msg_internal *m, struct sockaddr_un *s) { + struct sync_request *sync_req; struct action_entry *entry; + struct rte_mp_msg *msg = &m->msg; rte_eal_mp_t action = NULL; RTE_LOG(DEBUG, EAL, "msg: %s\n", msg->name); + + if (m->type == MP_REP) { + pthread_mutex_lock(&sync_requests.lock); + sync_req = find_sync_request(s->sun_path, msg->name); + if (sync_req) { + memcpy(sync_req->reply, msg, sizeof(*msg)); + sync_req->reply_received = 1; + pthread_cond_signal(&sync_req->cond); + } else + RTE_LOG(ERR, EAL, "Drop mp reply: %s\n", msg->name); + pthread_mutex_unlock(&sync_requests.lock); + return; + } + pthread_mutex_lock(&mp_mutex_action); entry = find_action_entry_by_name(msg->name); if (entry != NULL) @@ -204,18 +267,19 @@ process_msg(struct rte_mp_msg *msg) if (!action) RTE_LOG(ERR, EAL, "Cannot find action: %s\n", msg->name); - else if (action(msg) < 0) + else if (action(msg, s->sun_path) < 0) RTE_LOG(ERR, EAL, "Fail to handle message: %s\n", msg->name); } static void * mp_handle(void *arg __rte_unused) { - struct rte_mp_msg msg; + struct mp_msg_internal msg; + struct sockaddr_un sa; while (1) { - if (read_msg(&msg) == 0) - process_msg(&msg); + if (read_msg(&msg, &sa) == 0) + process_msg(&msg, &sa); } return NULL; @@ -309,16 +373,20 @@ rte_eal_mp_channel_init(void) } static int -send_msg(const char *dst_path, struct rte_mp_msg *msg) +send_msg(const char *dst_path, struct rte_mp_msg *msg, int type) { int snd; struct iovec iov; struct msghdr msgh; struct cmsghdr *cmsg; struct sockaddr_un dst; + struct mp_msg_internal m; int fd_size = msg->num_fds * sizeof(int); char control[CMSG_SPACE(fd_size)]; + m.type = type; + memcpy(&m.msg, msg, sizeof(*msg)); + memset(&dst, 0, sizeof(dst)); dst.sun_family = AF_UNIX; snprintf(dst.sun_path, sizeof(dst.sun_path), "%s", dst_path); @@ -326,8 +394,8 @@ send_msg(const char *dst_path, struct rte_mp_msg *msg) memset(&msgh, 0, sizeof(msgh)); memset(control, 0, sizeof(control)); - iov.iov_base = msg; - iov.iov_len = sizeof(*msg) - sizeof(msg->fds); + iov.iov_base = &m; + iov.iov_len = sizeof(m) - sizeof(msg->fds); msgh.msg_name = &dst; msgh.msg_namelen = sizeof(dst); @@ -355,12 +423,16 @@ send_msg(const char *dst_path, struct rte_mp_msg *msg) } static int -mp_send(struct rte_mp_msg *msg) +mp_send(struct rte_mp_msg *msg, const char *peer, int type) { int n = 0; DIR *mp_dir; struct dirent *ent; + + if (peer) + return send_msg(peer, msg, type); + if (rte_eal_process_type() == RTE_PROC_PRIMARY) { /* broadcast to all secondary processes */ mp_dir = opendir(mp_dir_path); @@ -373,11 +445,11 @@ mp_send(struct rte_mp_msg *msg) if (fnmatch(mp_filter, ent->d_name, 0) != 0) continue; - n += send_msg(ent->d_name, msg); + n += send_msg(ent->d_name, msg, type); } closedir(mp_dir); } else - n += send_msg(eal_mp_socket_path(), msg); + n += send_msg(eal_mp_socket_path(), msg, type); return n; } @@ -417,5 +489,136 @@ rte_eal_mp_sendmsg(struct rte_mp_msg *msg) return -1; RTE_LOG(DEBUG, EAL, "sendmsg: %s\n", msg->name); - return mp_send(msg); + return mp_send(msg, NULL, MP_MSG); +} + +static int +mp_request_one(const char *dst, struct rte_mp_msg *req, + struct rte_mp_reply *reply, const struct timespec *ts) +{ + struct timeval now; + struct rte_mp_msg msg, *tmp; + struct sync_request sync_req, *exist; + + sync_req.reply_received = 0; + strcpy(sync_req.dst, dst); + sync_req.request = req; + sync_req.reply = &msg; + pthread_cond_init(&sync_req.cond, NULL); + + pthread_mutex_lock(&sync_requests.lock); + exist = find_sync_request(dst, req->name); + if (!exist) + TAILQ_INSERT_TAIL(&sync_requests.requests, &sync_req, next); + pthread_mutex_unlock(&sync_requests.lock); + if (exist) { + RTE_LOG(ERR, EAL, "A pending request %s:%s\n", dst, req->name); + return 0; + } + + if (send_msg(dst, req, MP_REQ) != 1) { + RTE_LOG(ERR, EAL, "Fail to send request %s:%s\n", + dst, req->name); + return 0; + } + + pthread_mutex_lock(&sync_requests.lock); + do { + pthread_cond_timedwait(&sync_req.cond, &sync_requests.lock, ts); + /* Check spurious wakeups */ + if (sync_req.reply_received == 1) + break; + /* Check if time is out */ + if (gettimeofday(&now, NULL) < 0) + break; + if (now.tv_sec < ts->tv_sec) + break; + else if (now.tv_sec == ts->tv_sec && + now.tv_usec * 1000 < ts->tv_nsec) + break; + } while (1); + /* We got the lock now */ + TAILQ_REMOVE(&sync_requests.requests, &sync_req, next); + pthread_mutex_unlock(&sync_requests.lock); + + if (sync_req.reply_received == 0) { + RTE_LOG(ERR, EAL, "Fail to recv reply for request %s:%s\n", + dst, req->name); + return 1; + } + + tmp = realloc(reply->msgs, sizeof(msg) * (reply->nb_msgs + 1)); + if (!tmp) { + RTE_LOG(ERR, EAL, "Fail to alloc reply for request %s:%s\n", + dst, req->name); + return 1; + } + memcpy(&tmp[reply->nb_msgs], &msg, sizeof(msg)); + reply->msgs = tmp; + reply->nb_msgs++; + return 1; +} + +int +rte_eal_mp_request(struct rte_mp_msg *req, + struct rte_mp_reply *reply, + const struct timespec *ts) +{ + DIR *mp_dir; + struct dirent *ent; + int nb_snds = 0; + struct timeval now; + struct timespec end; + + RTE_LOG(DEBUG, EAL, "request: %s\n", req->name); + + if (check_input(req) == false) + return -1; + if (gettimeofday(&now, NULL) < 0) { + RTE_LOG(ERR, EAL, "Faile to get current time\n"); + return -1; + } + end.tv_nsec = (now.tv_usec * 1000 + ts->tv_nsec) % 1000000000; + end.tv_sec = now.tv_sec + ts->tv_sec + + (now.tv_usec * 1000 + ts->tv_nsec) / 1000000000; + + reply->nb_msgs = 0; + reply->msgs = NULL; + + /* for secondary process, send request to the primary process only */ + if (rte_eal_process_type() == RTE_PROC_SECONDARY) + return mp_request_one(eal_mp_socket_path(), req, reply, &end); + + /* for primary process, broadcast request, and collect reply 1 by 1 */ + mp_dir = opendir(mp_dir_path); + if (!mp_dir) { + RTE_LOG(ERR, EAL, "Unable to open directory %s\n", mp_dir_path); + return -1; + } + while ((ent = readdir(mp_dir))) { + if (fnmatch(mp_filter, ent->d_name, 0) != 0) + continue; + + nb_snds += mp_request_one(ent->d_name, req, reply, &end); + } + closedir(mp_dir); + + return nb_snds; +} + +int +rte_eal_mp_reply(struct rte_mp_msg *msg, const char *peer) +{ + + RTE_LOG(DEBUG, EAL, "reply: %s\n", msg->name); + + if (check_input(msg) == false) + return -1; + + if (peer == NULL) { + RTE_LOG(ERR, EAL, "peer is not specified\n"); + return -1; + } + + return mp_send(msg, peer, MP_REP); } diff --git a/lib/librte_eal/common/include/rte_eal.h b/lib/librte_eal/common/include/rte_eal.h index 9a1aac2..8e234e0 100644 --- a/lib/librte_eal/common/include/rte_eal.h +++ b/lib/librte_eal/common/include/rte_eal.h @@ -13,6 +13,7 @@ #include #include +#include #include #include @@ -197,13 +198,18 @@ struct rte_mp_msg { int fds[RTE_MP_MAX_FD_NUM]; }; +struct rte_mp_reply { + int nb_msgs; + struct rte_mp_msg *msgs; +}; + /** * Action function typedef used by other components. * * As we create socket channel for primary/secondary communication, use * this function typedef to register action for coming messages. */ -typedef int (*rte_eal_mp_t)(const struct rte_mp_msg *msg); +typedef int (*rte_eal_mp_t)(const struct rte_mp_msg *msg, const void *peer); /** * @warning @@ -262,6 +268,56 @@ void rte_eal_mp_action_unregister(const char *name); int rte_eal_mp_sendmsg(struct rte_mp_msg *msg); /** + * @warning + * @b EXPERIMENTAL: this API may change without prior notice + * + * Send a request to the peer process and expect a reply. + * + * This function sends a request message to the peer process, and will + * block until receiving reply message from the peer process. + * + * @note The caller is responsible to free reply->replies. + * + * @param req + * The req argument contains the customized request message. + * + * @param reply + * The reply argument will be for storing all the replied messages; + * the caller is responsible for free reply->replies. + * + * @param ts + * The ts argument specifies how long we can wait for the peer(s) to reply. + * + * @return + * - (<0) on invalid parameters; + * - (>=0) as the number of messages being sent successfully. + */ +int rte_eal_mp_request(struct rte_mp_msg *req, + struct rte_mp_reply *reply, const struct timespec *ts); + +/** + * @warning + * @b EXPERIMENTAL: this API may change without prior notice + * + * Send a reply to the peer process. + * + * This function will send a reply message in response to a request message + * received previously. + * + * @param msg + * The msg argument contains the customized message. + * + * @param peer + * The peer argument is the pointer to the peer socket path. + * + * @return + * - (1) on success; + * - (0) on failure; + * - (<0) on invalid parameters. + */ +int rte_eal_mp_reply(struct rte_mp_msg *msg, const char *peer); + +/** * 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/rte_eal_version.map b/lib/librte_eal/rte_eal_version.map index adeadfb..3015bc6 100644 --- a/lib/librte_eal/rte_eal_version.map +++ b/lib/librte_eal/rte_eal_version.map @@ -220,6 +220,9 @@ EXPERIMENTAL { rte_eal_mp_action_register; rte_eal_mp_action_unregister; rte_eal_mp_sendmsg; + rte_eal_mp_request; + rte_eal_mp_reply; + rte_eal_mp_sendmsg; rte_service_attr_get; rte_service_attr_reset_all; rte_service_component_register; -- 2.7.4