[PATCH v7 2/2] eal: add synchronous multi-process communication

[Jianfeng Tan <jianfeng.tan@intel.com>]

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 process,
   it will send request to peer1, collect response from peer1; then
   send request to peer2, collect response 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 <action_name, peer>, we guarantee that only one such request
   is on the fly.

Suggested-by: Anatoly Burakov <anatoly.burakov@intel.com>
Suggested-by: Konstantin Ananyev <konstantin.ananyev@intel.com>
Signed-off-by: Jianfeng Tan <jianfeng.tan@intel.com>
Reviewed-by: Anatoly Burakov <anatoly.burakov@intel.com>
Acked-by: Konstantin Ananyev <konstantin.ananyev@intel.com>
---
 doc/guides/rel_notes/release_18_02.rst  |   2 +
 lib/librte_eal/common/eal_common_proc.c | 254 +++++++++++++++++++++++++++++---
 lib/librte_eal/common/include/rte_eal.h |  60 +++++++-
 lib/librte_eal/rte_eal_version.map      |   2 +
 4 files changed, 298 insertions(+), 20 deletions(-)

diff --git a/doc/guides/rel_notes/release_18_02.rst b/doc/guides/rel_notes/release_18_02.rst
index 0531f59..bb8559b 100644
--- a/doc/guides/rel_notes/release_18_02.rst
+++ b/doc/guides/rel_notes/release_18_02.rst
@@ -169,6 +169,8 @@ New Features
 
   * ``rte_mp_register`` and ``rte_mp_unregister`` are for action (un)registration.
   * ``rte_mp_sendmsg`` is for sending a message without blocking for a response.
+  * ``rte_mp_request`` is for sending a request message and will block until
+    it gets a reply message which is sent from the peer by ``rte_mp_reply``.
 
 API Changes
 -----------
diff --git a/lib/librte_eal/common/eal_common_proc.c b/lib/librte_eal/common/eal_common_proc.c
index f63c9c2..b974837 100644
--- a/lib/librte_eal/common/eal_common_proc.c
+++ b/lib/librte_eal/common/eal_common_proc.c
@@ -13,6 +13,7 @@
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
+#include <sys/time.h>
 #include <sys/types.h>
 #include <sys/socket.h>
 #include <sys/un.h>
@@ -46,6 +47,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 {
+	TAILQ_ENTRY(sync_request) next;
+	int reply_received;
+	char dst[PATH_MAX];
+	struct rte_mp_msg *request;
+	struct rte_mp_msg *reply;
+	pthread_cond_t cond;
+};
+
+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)
 {
@@ -149,19 +194,21 @@ rte_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;
@@ -183,7 +230,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;
 		}
 	}
@@ -192,12 +239,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_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)
@@ -206,18 +269,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;
@@ -336,16 +400,20 @@ rte_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);
@@ -353,8 +421,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);
@@ -396,14 +464,17 @@ 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 ret = 0;
 	DIR *mp_dir;
 	struct dirent *ent;
 
-	if (rte_eal_process_type() == RTE_PROC_SECONDARY) {
-		if (send_msg(eal_mp_socket_path(), msg) < 0)
+	if (!peer && (rte_eal_process_type() == RTE_PROC_SECONDARY))
+		peer = eal_mp_socket_path();
+
+	if (peer) {
+		if (send_msg(peer, msg, type) < 0)
 			return -1;
 		else
 			return 0;
@@ -421,11 +492,11 @@ mp_send(struct rte_mp_msg *msg)
 		if (fnmatch(mp_filter, ent->d_name, 0) != 0)
 			continue;
 
-		if (send_msg(ent->d_name, msg) < 0)
+		if (send_msg(ent->d_name, msg, type) < 0)
 			ret = -1;
 	}
-	closedir(mp_dir);
 
+	closedir(mp_dir);
 	return ret;
 }
 
@@ -464,5 +535,150 @@ rte_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)
+{
+	int ret;
+	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);
+		rte_errno = -EEXIST;
+		return -1;
+	}
+
+	ret = send_msg(dst, req, MP_REQ);
+	if (ret < 0) {
+		RTE_LOG(ERR, EAL, "Fail to send request %s:%s\n",
+			dst, req->name);
+		return -1;
+	} else if (ret == 0)
+		return 0;
+
+	reply->nb_sent++;
+
+	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);
+		rte_errno = -ETIMEDOUT;
+		return -1;
+	}
+
+	tmp = realloc(reply->msgs, sizeof(msg) * (reply->nb_received + 1));
+	if (!tmp) {
+		RTE_LOG(ERR, EAL, "Fail to alloc reply for request %s:%s\n",
+			dst, req->name);
+		rte_errno = -ENOMEM;
+		return -1;
+	}
+	memcpy(&tmp[reply->nb_received], &msg, sizeof(msg));
+	reply->msgs = tmp;
+	reply->nb_received++;
+	return 0;
+}
+
+int __rte_experimental
+rte_mp_request(struct rte_mp_msg *req, struct rte_mp_reply *reply,
+		const struct timespec *ts)
+{
+	int ret = 0;
+	DIR *mp_dir;
+	struct dirent *ent;
+	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");
+		rte_errno = errno;
+		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_sent = 0;
+	reply->nb_received = 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);
+		rte_errno = errno;
+		return -1;
+	}
+
+	while ((ent = readdir(mp_dir))) {
+		if (fnmatch(mp_filter, ent->d_name, 0) != 0)
+			continue;
+
+		if (mp_request_one(ent->d_name, req, reply, &end))
+			ret = -1;
+	}
+
+	closedir(mp_dir);
+	return ret;
+}
+
+int __rte_experimental
+rte_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");
+		rte_errno = -EINVAL;
+		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 2d022c0..08c6637 100644
--- a/lib/librte_eal/common/include/rte_eal.h
+++ b/lib/librte_eal/common/include/rte_eal.h
@@ -13,6 +13,7 @@
 
 #include <stdint.h>
 #include <sched.h>
+#include <time.h>
 
 #include <rte_config.h>
 #include <rte_compat.h>
@@ -214,13 +215,19 @@ struct rte_mp_msg {
 	int fds[RTE_MP_MAX_FD_NUM];
 };
 
+struct rte_mp_reply {
+	int nb_sent;
+	int nb_received;
+	struct rte_mp_msg *msgs; /* caller to free */
+};
+
 /**
  * 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_mp_t)(const struct rte_mp_msg *msg);
+typedef int (*rte_mp_t)(const struct rte_mp_msg *msg, const void *peer);
 
 /**
  * @warning
@@ -282,6 +289,57 @@ int __rte_experimental
 rte_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
+ *  - On success, return 0.
+ *  - On failure, return -1, and the reason will be stored in rte_errno.
+ */
+int __rte_experimental
+rte_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
+ *  - On success, return 0.
+ *  - On failure, return -1, and the reason will be stored in rte_errno.
+ */
+int __rte_experimental
+rte_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 24deaef..4146907 100644
--- a/lib/librte_eal/rte_eal_version.map
+++ b/lib/librte_eal/rte_eal_version.map
@@ -223,6 +223,8 @@ EXPERIMENTAL {
 	rte_mp_action_register;
 	rte_mp_action_unregister;
 	rte_mp_sendmsg;
+	rte_mp_request;
+	rte_mp_reply;
 	rte_service_attr_get;
 	rte_service_attr_reset_all;
 	rte_service_component_register;
-- 
2.7.4