[RFC 0/8] remove IPC threads

[RFC 0/8] remove IPC threads

[Jianfeng Tan]

As discussed here, http://dpdk.org/dev/patchwork/patch/36579/, we will
try best to avoid thread creation in the library. Now we have two threads
for IPC, rte_mp_handle and rte_mp_handle_async. This patchset is to
finish the job.

This patchset:
- is rebased on commit 34345a9b69bb ("eventdev: fix build with icc").
- also needs http://dpdk.org/dev/patchwork/patch/37335/

Patch 1~2: code cleanup and bug fix. (Target for v18.05)
Patch 3~4: Remove IPC async thread. (Target for v18.08)
Patch 5~8: Remove IPC thread. (Target for v18.08)

TODO:

  After this patch, IPC depends on interrupt handling thread and alarm.
  It's OK for Linux-specific subsystem/feature, like memory hotplug and vfio;
  while pdump and vdev auto discovery could be on FreeBSD.
  So it's like that we also need to implement interrupt handling thread and
  alarm for FreeBSD.

Jianfeng Tan (8):
  ipc: clearn up code
  ipc: fix timeout not properly handled in async
  eal/linux: use glibc malloc in alarm
  ipc: remove IPC thread for async request
  eal/linux: use glibc malloc in interrupt handling
  eal: bring forward init of interrupt handling
  eal: add IPC type for interrupt thread
  ipc: remove IPC thread for message read

 lib/librte_eal/common/eal_common_proc.c            | 289 ++++++++-------------
 lib/librte_eal/common/include/rte_eal_interrupts.h |   1 +
 lib/librte_eal/linuxapp/eal/eal.c                  |  10 +-
 lib/librte_eal/linuxapp/eal/eal_alarm.c            |   9 +-
 lib/librte_eal/linuxapp/eal/eal_interrupts.c       |  18 +-
 test/test/test_interrupts.c                        |  29 ++-
 6 files changed, 151 insertions(+), 205 deletions(-)

-- 
2.7.4

[RFC 1/8] ipc: clearn up code

[Jianfeng Tan]

Following below commit, we change some internal function and variable
names:
  commit ce3a7312357b ("eal: rename IPC request as synchronous one")

Also use calloc to supersede malloc + memset for code clean up.

Signed-off-by: Jianfeng Tan <jianfeng.tan@intel.com>
---
 lib/librte_eal/common/eal_common_proc.c | 82 +++++++++++++++------------------
 1 file changed, 38 insertions(+), 44 deletions(-)

diff --git a/lib/librte_eal/common/eal_common_proc.c b/lib/librte_eal/common/eal_common_proc.c
index 2179f3d..070a075 100644
--- a/lib/librte_eal/common/eal_common_proc.c
+++ b/lib/librte_eal/common/eal_common_proc.c
@@ -108,7 +108,7 @@ mp_send(struct rte_mp_msg *msg, const char *peer, int type);
 
 
 static struct pending_request *
-find_sync_request(const char *dst, const char *act_name)
+find_pending_request(const char *dst, const char *act_name)
 {
 	struct pending_request *r;
 
@@ -282,7 +282,7 @@ read_msg(struct mp_msg_internal *m, struct sockaddr_un *s)
 static void
 process_msg(struct mp_msg_internal *m, struct sockaddr_un *s)
 {
-	struct pending_request *sync_req;
+	struct pending_request *pending_req;
 	struct action_entry *entry;
 	struct rte_mp_msg *msg = &m->msg;
 	rte_mp_t action = NULL;
@@ -291,15 +291,15 @@ process_msg(struct mp_msg_internal *m, struct sockaddr_un *s)
 
 	if (m->type == MP_REP || m->type == MP_IGN) {
 		pthread_mutex_lock(&pending_requests.lock);
-		sync_req = find_sync_request(s->sun_path, msg->name);
-		if (sync_req) {
-			memcpy(sync_req->reply, msg, sizeof(*msg));
+		pending_req = find_pending_request(s->sun_path, msg->name);
+		if (pending_req) {
+			memcpy(pending_req->reply, msg, sizeof(*msg));
 			/* -1 indicates that we've been asked to ignore */
-			sync_req->reply_received = m->type == MP_REP ? 1 : -1;
+			pending_req->reply_received = m->type == MP_REP ? 1 : -1;
 
-			if (sync_req->type == REQUEST_TYPE_SYNC)
-				pthread_cond_signal(&sync_req->sync.cond);
-			else if (sync_req->type == REQUEST_TYPE_ASYNC)
+			if (pending_req->type == REQUEST_TYPE_SYNC)
+				pthread_cond_signal(&pending_req->sync.cond);
+			else if (pending_req->type == REQUEST_TYPE_ASYNC)
 				pthread_cond_signal(
 					&pending_requests.async_cond);
 		} else
@@ -322,6 +322,7 @@ process_msg(struct mp_msg_internal *m, struct sockaddr_un *s)
 			 * yet ready to process this request.
 			 */
 			struct rte_mp_msg dummy;
+
 			memset(&dummy, 0, sizeof(dummy));
 			snprintf(dummy.name, sizeof(dummy.name),
 					"%s", msg->name);
@@ -854,30 +855,27 @@ mp_request_async(const char *dst, struct rte_mp_msg *req,
 		struct async_request_param *param)
 {
 	struct rte_mp_msg *reply_msg;
-	struct pending_request *sync_req, *exist;
+	struct pending_request *pending_req, *exist;
 	int ret;
 
-	sync_req = malloc(sizeof(*sync_req));
-	reply_msg = malloc(sizeof(*reply_msg));
-	if (sync_req == NULL || reply_msg == NULL) {
+	pending_req = calloc(1, sizeof(*pending_req));
+	reply_msg = calloc(1, sizeof(*reply_msg));
+	if (pending_req == NULL || reply_msg == NULL) {
 		RTE_LOG(ERR, EAL, "Could not allocate space for sync request\n");
 		rte_errno = ENOMEM;
 		ret = -1;
 		goto fail;
 	}
 
-	memset(sync_req, 0, sizeof(*sync_req));
-	memset(reply_msg, 0, sizeof(*reply_msg));
-
-	sync_req->type = REQUEST_TYPE_ASYNC;
-	strcpy(sync_req->dst, dst);
-	sync_req->request = req;
-	sync_req->reply = reply_msg;
-	sync_req->async.param = param;
+	pending_req->type = REQUEST_TYPE_ASYNC;
+	strcpy(pending_req->dst, dst);
+	pending_req->request = req;
+	pending_req->reply = reply_msg;
+	pending_req->async.param = param;
 
 	/* queue already locked by caller */
 
-	exist = find_sync_request(dst, req->name);
+	exist = find_pending_request(dst, req->name);
 	if (exist) {
 		RTE_LOG(ERR, EAL, "A pending request %s:%s\n", dst, req->name);
 		rte_errno = EEXIST;
@@ -895,13 +893,13 @@ mp_request_async(const char *dst, struct rte_mp_msg *req,
 		ret = 0;
 		goto fail;
 	}
-	TAILQ_INSERT_TAIL(&pending_requests.requests, sync_req, next);
+	TAILQ_INSERT_TAIL(&pending_requests.requests, pending_req, next);
 
 	param->user_reply.nb_sent++;
 
 	return 0;
 fail:
-	free(sync_req);
+	free(pending_req);
 	free(reply_msg);
 	return ret;
 }
@@ -912,16 +910,16 @@ mp_request_sync(const char *dst, struct rte_mp_msg *req,
 {
 	int ret;
 	struct rte_mp_msg msg, *tmp;
-	struct pending_request sync_req, *exist;
+	struct pending_request pending_req, *exist;
 
-	sync_req.type = REQUEST_TYPE_SYNC;
-	sync_req.reply_received = 0;
-	strcpy(sync_req.dst, dst);
-	sync_req.request = req;
-	sync_req.reply = &msg;
-	pthread_cond_init(&sync_req.sync.cond, NULL);
+	pending_req.type = REQUEST_TYPE_SYNC;
+	pending_req.reply_received = 0;
+	strcpy(pending_req.dst, dst);
+	pending_req.request = req;
+	pending_req.reply = &msg;
+	pthread_cond_init(&pending_req.sync.cond, NULL);
 
-	exist = find_sync_request(dst, req->name);
+	exist = find_pending_request(dst, req->name);
 	if (exist) {
 		RTE_LOG(ERR, EAL, "A pending request %s:%s\n", dst, req->name);
 		rte_errno = EEXIST;
@@ -936,24 +934,24 @@ mp_request_sync(const char *dst, struct rte_mp_msg *req,
 	} else if (ret == 0)
 		return 0;
 
-	TAILQ_INSERT_TAIL(&pending_requests.requests, &sync_req, next);
+	TAILQ_INSERT_TAIL(&pending_requests.requests, &pending_req, next);
 
 	reply->nb_sent++;
 
 	do {
-		ret = pthread_cond_timedwait(&sync_req.sync.cond,
+		ret = pthread_cond_timedwait(&pending_req.sync.cond,
 				&pending_requests.lock, ts);
 	} while (ret != 0 && ret != ETIMEDOUT);
 
-	TAILQ_REMOVE(&pending_requests.requests, &sync_req, next);
+	TAILQ_REMOVE(&pending_requests.requests, &pending_req, next);
 
-	if (sync_req.reply_received == 0) {
+	if (pending_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;
 	}
-	if (sync_req.reply_received == -1) {
+	if (pending_req.reply_received == -1) {
 		RTE_LOG(DEBUG, EAL, "Asked to ignore response\n");
 		/* not receiving this message is not an error, so decrement
 		 * number of sent messages
@@ -1078,19 +1076,15 @@ rte_mp_request_async(struct rte_mp_msg *req, const struct timespec *ts,
 		rte_errno = errno;
 		return -1;
 	}
-	copy = malloc(sizeof(*copy));
-	dummy = malloc(sizeof(*dummy));
-	param = malloc(sizeof(*param));
+	copy = calloc(1, sizeof(*copy));
+	dummy = calloc(1, sizeof(*dummy));
+	param = calloc(1, sizeof(*param));
 	if (copy == NULL || dummy == NULL || param == NULL) {
 		RTE_LOG(ERR, EAL, "Failed to allocate memory for async reply\n");
 		rte_errno = ENOMEM;
 		goto fail;
 	}
 
-	memset(copy, 0, sizeof(*copy));
-	memset(dummy, 0, sizeof(*dummy));
-	memset(param, 0, sizeof(*param));
-
 	/* copy message */
 	memcpy(copy, req, sizeof(*copy));
 
-- 
2.7.4

[RFC 2/8] ipc: fix timeout not properly handled in async

[Jianfeng Tan]

In original implementation, timeout event for an async request
will be ignored. As a result, an async request will never
trigger the action if it cannot receive any reply any more.

We fix this by counting timeout as a processed reply.

Fixes: f05e26051c15 ("eal: add IPC asynchronous request")

Signed-off-by: Jianfeng Tan <jianfeng.tan@intel.com>
Signed-off-by: Anatoly Burakov <anatoly.burakov@intel.com>
---
 lib/librte_eal/common/eal_common_proc.c | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/lib/librte_eal/common/eal_common_proc.c b/lib/librte_eal/common/eal_common_proc.c
index 070a075..27de16e 100644
--- a/lib/librte_eal/common/eal_common_proc.c
+++ b/lib/librte_eal/common/eal_common_proc.c
@@ -419,7 +419,13 @@ process_async_request(struct pending_request *sr, const struct timespec *now)
 	} else if (sr->reply_received == -1) {
 		/* we were asked to ignore this process */
 		reply->nb_sent--;
+	} else if (timeout) {
+		/* count it as processed reponse, but don't increment
+		 * nb_received.
+		 */
+		param->n_responses_processed++;
 	}
+
 	free(sr->reply);
 
 	last_msg = param->n_responses_processed == reply->nb_sent;
-- 
2.7.4

[RFC 3/8] eal/linux: use glibc malloc in alarm

[Jianfeng Tan]

We will reply on alarm API for async IPC request as following patch
indicates. rte_malloc could require async IPC request.

To avoid such chicken or the egg causality dilemma, we change to
use glibc malloc in alarm implimentation.

Signed-off-by: Jianfeng Tan <jianfeng.tan@intel.com>
---
 lib/librte_eal/linuxapp/eal/eal_alarm.c | 9 ++++-----
 1 file changed, 4 insertions(+), 5 deletions(-)

diff --git a/lib/librte_eal/linuxapp/eal/eal_alarm.c b/lib/librte_eal/linuxapp/eal/eal_alarm.c
index c115e82..391d2a6 100644
--- a/lib/librte_eal/linuxapp/eal/eal_alarm.c
+++ b/lib/librte_eal/linuxapp/eal/eal_alarm.c
@@ -19,7 +19,6 @@
 #include <rte_launch.h>
 #include <rte_lcore.h>
 #include <rte_errno.h>
-#include <rte_malloc.h>
 #include <rte_spinlock.h>
 #include <eal_private.h>
 
@@ -91,7 +90,7 @@ eal_alarm_callback(void *arg __rte_unused)
 		rte_spinlock_lock(&alarm_list_lk);
 
 		LIST_REMOVE(ap, next);
-		rte_free(ap);
+		free(ap);
 	}
 
 	if (!LIST_EMPTY(&alarm_list)) {
@@ -122,7 +121,7 @@ rte_eal_alarm_set(uint64_t us, rte_eal_alarm_callback cb_fn, void *cb_arg)
 	if (us < 1 || us > (UINT64_MAX - US_PER_S) || cb_fn == NULL)
 		return -EINVAL;
 
-	new_alarm = rte_zmalloc(NULL, sizeof(*new_alarm), 0);
+	new_alarm = calloc(1, sizeof(*new_alarm));
 	if (new_alarm == NULL)
 		return -ENOMEM;
 
@@ -196,7 +195,7 @@ rte_eal_alarm_cancel(rte_eal_alarm_callback cb_fn, void *cb_arg)
 
 			if (ap->executing == 0) {
 				LIST_REMOVE(ap, next);
-				rte_free(ap);
+				free(ap);
 				count++;
 			} else {
 				/* If calling from other context, mark that alarm is executing
@@ -220,7 +219,7 @@ rte_eal_alarm_cancel(rte_eal_alarm_callback cb_fn, void *cb_arg)
 
 				if (ap->executing == 0) {
 					LIST_REMOVE(ap, next);
-					rte_free(ap);
+					free(ap);
 					count++;
 					ap = ap_prev;
 				} else if (pthread_equal(ap->executing_id, pthread_self()) == 0)
-- 
2.7.4

[RFC 4/8] ipc: remove IPC thread for async request

[Jianfeng Tan]

As discussed here, http://dpdk.org/dev/patchwork/patch/36579/,
we remove IPC threads, rte_mp_handle and rte_mp_handle_async.
This patch targets to remove thread rte_mp_handle_async.

Previously, to handle replies for an async request, rte_mp_handle
wakes up the rte_mp_handle_async thread to process through
pending_requests.async_cond. Now, we change to handle that in
rte_mp_handle context directly.

To handle timeout events, for each async request which is sent,
we set an alarm for it. If its reply is received before timeout,
we will cancel the alarm when we handle the reply; otherwise,
alarm will invoke the async_reply_handle() as the alarm callback.

Cc: Anatoly Burakov <anatoly.burakov@intel.com>

Suggested-by: Thomas Monjalon <thomas@monjalon.net>
Signed-off-by: Jianfeng Tan <jianfeng.tan@intel.com>
---
 lib/librte_eal/common/eal_common_proc.c | 169 ++++++++------------------------
 1 file changed, 43 insertions(+), 126 deletions(-)

diff --git a/lib/librte_eal/common/eal_common_proc.c b/lib/librte_eal/common/eal_common_proc.c
index 27de16e..4cb460e 100644
--- a/lib/librte_eal/common/eal_common_proc.c
+++ b/lib/librte_eal/common/eal_common_proc.c
@@ -20,6 +20,7 @@
 #include <sys/un.h>
 #include <unistd.h>
 
+#include <rte_alarm.h>
 #include <rte_common.h>
 #include <rte_cycles.h>
 #include <rte_eal.h>
@@ -94,14 +95,14 @@ TAILQ_HEAD(pending_request_list, pending_request);
 static struct {
 	struct pending_request_list requests;
 	pthread_mutex_t lock;
-	pthread_cond_t async_cond;
 } pending_requests = {
 	.requests = TAILQ_HEAD_INITIALIZER(pending_requests.requests),
 	.lock = PTHREAD_MUTEX_INITIALIZER,
-	.async_cond = PTHREAD_COND_INITIALIZER
 	/**< used in async requests only */
 };
 
+static void async_reply_handle(void *arg);
+
 /* forward declarations */
 static int
 mp_send(struct rte_mp_msg *msg, const char *peer, int type);
@@ -299,9 +300,11 @@ process_msg(struct mp_msg_internal *m, struct sockaddr_un *s)
 
 			if (pending_req->type == REQUEST_TYPE_SYNC)
 				pthread_cond_signal(&pending_req->sync.cond);
-			else if (pending_req->type == REQUEST_TYPE_ASYNC)
-				pthread_cond_signal(
-					&pending_requests.async_cond);
+			else if (pending_req->type == REQUEST_TYPE_ASYNC) {
+				pthread_mutex_unlock(&pending_requests.lock);
+				async_reply_handle(pending_req);
+				pthread_mutex_lock(&pending_requests.lock);
+			}
 		} else
 			RTE_LOG(ERR, EAL, "Drop mp reply: %s\n", msg->name);
 		pthread_mutex_unlock(&pending_requests.lock);
@@ -450,115 +453,39 @@ trigger_async_action(struct pending_request *sr)
 	free(sr->request);
 }
 
-static struct pending_request *
-check_trigger(struct timespec *ts)
-{
-	struct pending_request *next, *cur, *trigger = NULL;
-
-	TAILQ_FOREACH_SAFE(cur, &pending_requests.requests, next, next) {
-		enum async_action action;
-		if (cur->type != REQUEST_TYPE_ASYNC)
-			continue;
-
-		action = process_async_request(cur, ts);
-		if (action == ACTION_FREE) {
-			TAILQ_REMOVE(&pending_requests.requests, cur, next);
-			free(cur);
-		} else if (action == ACTION_TRIGGER) {
-			TAILQ_REMOVE(&pending_requests.requests, cur, next);
-			trigger = cur;
-			break;
-		}
-	}
-	return trigger;
-}
-
 static void
-wait_for_async_messages(void)
-{
-	struct pending_request *sr;
-	struct timespec timeout;
-	bool timedwait = false;
-	bool nowait = false;
-	int ret;
-
-	/* scan through the list and see if there are any timeouts that
-	 * are earlier than our current timeout.
-	 */
-	TAILQ_FOREACH(sr, &pending_requests.requests, next) {
-		if (sr->type != REQUEST_TYPE_ASYNC)
-			continue;
-		if (!timedwait || timespec_cmp(&sr->async.param->end,
-				&timeout) < 0) {
-			memcpy(&timeout, &sr->async.param->end,
-				sizeof(timeout));
-			timedwait = true;
-		}
-
-		/* sometimes, we don't even wait */
-		if (sr->reply_received) {
-			nowait = true;
-			break;
-		}
-	}
-
-	if (nowait)
-		return;
-
-	do {
-		ret = timedwait ?
-			pthread_cond_timedwait(
-				&pending_requests.async_cond,
-				&pending_requests.lock,
-				&timeout) :
-			pthread_cond_wait(
-				&pending_requests.async_cond,
-				&pending_requests.lock);
-	} while (ret != 0 && ret != ETIMEDOUT);
-
-	/* we've been woken up or timed out */
-}
-
-static void *
-async_reply_handle(void *arg __rte_unused)
+async_reply_handle(void *arg)
 {
 	struct timeval now;
 	struct timespec ts_now;
-	while (1) {
-		struct pending_request *trigger = NULL;
+	enum async_action action;
+	struct pending_request *req = (struct pending_request *)arg;
 
-		pthread_mutex_lock(&pending_requests.lock);
-
-		/* we exit this function holding the lock */
-		wait_for_async_messages();
-
-		if (gettimeofday(&now, NULL) < 0) {
-			RTE_LOG(ERR, EAL, "Cannot get current time\n");
-			break;
-		}
-		ts_now.tv_nsec = now.tv_usec * 1000;
-		ts_now.tv_sec = now.tv_sec;
-
-		do {
-			trigger = check_trigger(&ts_now);
-			/* unlock request list */
-			pthread_mutex_unlock(&pending_requests.lock);
-
-			if (trigger) {
-				trigger_async_action(trigger);
-				free(trigger);
+	if (gettimeofday(&now, NULL) < 0) {
+		/* This could lead to disaster */
+		RTE_LOG(ERR, EAL, "Cannot get current time\n");
+		return;
+	}
+	ts_now.tv_nsec = now.tv_usec * 1000;
+	ts_now.tv_sec = now.tv_sec;
 
-				/* we've triggered a callback, but there may be
-				 * more, so lock the list and check again.
-				 */
-				pthread_mutex_lock(&pending_requests.lock);
-			}
-		} while (trigger);
+	pthread_mutex_lock(&pending_requests.lock);
+	action = process_async_request(req, &ts_now);
+	if (action == ACTION_NONE) {
+		pthread_mutex_unlock(&pending_requests.lock);
+		return;
 	}
+	TAILQ_REMOVE(&pending_requests.requests, req, next);
+	pthread_mutex_unlock(&pending_requests.lock);
 
-	RTE_LOG(ERR, EAL, "ERROR: asynchronous requests disabled\n");
+	if (action == ACTION_TRIGGER)
+		trigger_async_action(req);
 
-	return NULL;
+	if (rte_eal_alarm_cancel(async_reply_handle, req) < 0 &&
+	    rte_errno != EINPROGRESS)
+		RTE_LOG(ERR, EAL, "Failed to cancel alarm\n");
+
+	free(req);
 }
 
 static int
@@ -624,7 +551,7 @@ rte_mp_channel_init(void)
 	char thread_name[RTE_MAX_THREAD_NAME_LEN];
 	char path[PATH_MAX];
 	int dir_fd;
-	pthread_t mp_handle_tid, async_reply_handle_tid;
+	pthread_t mp_handle_tid;
 
 	/* create filter path */
 	create_socket_path("*", path, sizeof(path));
@@ -669,24 +596,10 @@ rte_mp_channel_init(void)
 		return -1;
 	}
 
-	if (pthread_create(&async_reply_handle_tid, NULL,
-			async_reply_handle, NULL) < 0) {
-		RTE_LOG(ERR, EAL, "failed to create mp thead: %s\n",
-			strerror(errno));
-		close(mp_fd);
-		close(dir_fd);
-		mp_fd = -1;
-		return -1;
-	}
-
 	/* try best to set thread name */
 	snprintf(thread_name, RTE_MAX_THREAD_NAME_LEN, "rte_mp_handle");
 	rte_thread_setname(mp_handle_tid, thread_name);
 
-	/* try best to set thread name */
-	snprintf(thread_name, RTE_MAX_THREAD_NAME_LEN, "rte_mp_async_handle");
-	rte_thread_setname(async_reply_handle_tid, thread_name);
-
 	/* unlock the directory */
 	flock(dir_fd, LOCK_UN);
 	close(dir_fd);
@@ -858,7 +771,7 @@ rte_mp_sendmsg(struct rte_mp_msg *msg)
 
 static int
 mp_request_async(const char *dst, struct rte_mp_msg *req,
-		struct async_request_param *param)
+		struct async_request_param *param, const struct timespec *ts)
 {
 	struct rte_mp_msg *reply_msg;
 	struct pending_request *pending_req, *exist;
@@ -903,6 +816,13 @@ mp_request_async(const char *dst, struct rte_mp_msg *req,
 
 	param->user_reply.nb_sent++;
 
+	if (rte_eal_alarm_set(ts->tv_sec * 1000000 + ts->tv_nsec / 10000,
+				async_reply_handle, pending_req) < 0) {
+		/* TODO: If error happends, turn to busy waiting */
+		RTE_LOG(ERR, EAL, "Fail to set alarm for request %s:%s\n",
+			dst, req->name);
+	}
+
 	return 0;
 fail:
 	free(pending_req);
@@ -1124,7 +1044,7 @@ rte_mp_request_async(struct rte_mp_msg *req, const struct timespec *ts,
 
 	/* for secondary process, send request to the primary process only */
 	if (rte_eal_process_type() == RTE_PROC_SECONDARY) {
-		ret = mp_request_async(eal_mp_socket_path(), copy, param);
+		ret = mp_request_async(eal_mp_socket_path(), copy, param, ts);
 
 		/* if we didn't send anything, put dummy request on the queue */
 		if (ret == 0 && reply->nb_sent == 0) {
@@ -1167,7 +1087,7 @@ rte_mp_request_async(struct rte_mp_msg *req, const struct timespec *ts,
 		snprintf(path, sizeof(path), "%s/%s", mp_dir_path,
 			 ent->d_name);
 
-		if (mp_request_async(path, copy, param))
+		if (mp_request_async(path, copy, param, ts))
 			ret = -1;
 	}
 	/* if we didn't send anything, put dummy request on the queue */
@@ -1176,9 +1096,6 @@ rte_mp_request_async(struct rte_mp_msg *req, const struct timespec *ts,
 		dummy_used = true;
 	}
 
-	/* trigger async request thread wake up */
-	pthread_cond_signal(&pending_requests.async_cond);
-
 	/* finally, unlock the queue */
 	pthread_mutex_unlock(&pending_requests.lock);
 
-- 
2.7.4

Re: [RFC 4/8] ipc: remove IPC thread for async request

[Burakov, Anatoly]

On 19-Apr-18 4:03 PM, Jianfeng Tan wrote:
> As discussed here, http://dpdk.org/dev/patchwork/patch/36579/,
> we remove IPC threads, rte_mp_handle and rte_mp_handle_async.
> This patch targets to remove thread rte_mp_handle_async.
> 
> Previously, to handle replies for an async request, rte_mp_handle
> wakes up the rte_mp_handle_async thread to process through
> pending_requests.async_cond. Now, we change to handle that in
> rte_mp_handle context directly.
> 
> To handle timeout events, for each async request which is sent,
> we set an alarm for it. If its reply is received before timeout,
> we will cancel the alarm when we handle the reply; otherwise,
> alarm will invoke the async_reply_handle() as the alarm callback.
> 
> Cc: Anatoly Burakov <anatoly.burakov@intel.com>
> 
> Suggested-by: Thomas Monjalon <thomas@monjalon.net>
> Signed-off-by: Jianfeng Tan <jianfeng.tan@intel.com>
> ---

<...>

> @@ -299,9 +300,11 @@ process_msg(struct mp_msg_internal *m, struct sockaddr_un *s)
>   
>   			if (pending_req->type == REQUEST_TYPE_SYNC)
>   				pthread_cond_signal(&pending_req->sync.cond);
> -			else if (pending_req->type == REQUEST_TYPE_ASYNC)
> -				pthread_cond_signal(
> -					&pending_requests.async_cond);
> +			else if (pending_req->type == REQUEST_TYPE_ASYNC) {
> +				pthread_mutex_unlock(&pending_requests.lock);
> +				async_reply_handle(pending_req);
> +				pthread_mutex_lock(&pending_requests.lock);

There must be a better way to do this than to unlock mutex before 
locking it again :) I haven't looked at implications of this suggestion 
yet, but how about alarm calling a wrapper function that locks the 
mutex, but leave async_reply_handle without any locking whatsoever?

> +			}
>   		} else
>   			RTE_LOG(ERR, EAL, "Drop mp reply: %s\n", msg->name);
>   		pthread_mutex_unlock(&pending_requests.lock);
> @@ -450,115 +453,39 @@ trigger_async_action(struct pending_request *sr)
>   	free(sr->request);
>   }
>   
> -static struct pending_request *
> -check_trigger(struct timespec *ts)

<...>

> +	ts_now.tv_nsec = now.tv_usec * 1000;
> +	ts_now.tv_sec = now.tv_sec;
>   
> -				/* we've triggered a callback, but there may be
> -				 * more, so lock the list and check again.
> -				 */
> -				pthread_mutex_lock(&pending_requests.lock);
> -			}
> -		} while (trigger);
> +	pthread_mutex_lock(&pending_requests.lock);
> +	action = process_async_request(req, &ts_now);
> +	if (action == ACTION_NONE) {
> +		pthread_mutex_unlock(&pending_requests.lock);
> +		return;
>   	}
> +	TAILQ_REMOVE(&pending_requests.requests, req, next);
> +	pthread_mutex_unlock(&pending_requests.lock);
>   
> -	RTE_LOG(ERR, EAL, "ERROR: asynchronous requests disabled\n");
> +	if (action == ACTION_TRIGGER)
> +		trigger_async_action(req);
>   
> -	return NULL;
> +	if (rte_eal_alarm_cancel(async_reply_handle, req) < 0 &&
> +	    rte_errno != EINPROGRESS)
> +		RTE_LOG(ERR, EAL, "Failed to cancel alarm\n");
> +

Perhaps cancel the alarm before triggering callback?

-- 
Thanks,
Anatoly

[RFC 5/8] eal/linux: use glibc malloc in interrupt handling

[Jianfeng Tan]

We will rely on interrupt thread to implement IPC; and IPC initialization
is in very early stage, when memory subsystem is not initialized yet.
So we change to use glibc malloc/free.

Signed-off-by: Jianfeng Tan <jianfeng.tan@intel.com>
---
 lib/librte_eal/linuxapp/eal/eal_interrupts.c | 13 +++++--------
 1 file changed, 5 insertions(+), 8 deletions(-)

diff --git a/lib/librte_eal/linuxapp/eal/eal_interrupts.c b/lib/librte_eal/linuxapp/eal/eal_interrupts.c
index 58e9328..5245372 100644
--- a/lib/librte_eal/linuxapp/eal/eal_interrupts.c
+++ b/lib/librte_eal/linuxapp/eal/eal_interrupts.c
@@ -30,7 +30,6 @@
 #include <rte_branch_prediction.h>
 #include <rte_debug.h>
 #include <rte_log.h>
-#include <rte_malloc.h>
 #include <rte_errno.h>
 #include <rte_spinlock.h>
 #include <rte_pause.h>
@@ -405,8 +404,7 @@ rte_intr_callback_register(const struct rte_intr_handle *intr_handle,
 	}
 
 	/* allocate a new interrupt callback entity */
-	callback = rte_zmalloc("interrupt callback list",
-				sizeof(*callback), 0);
+	callback = calloc(1, sizeof(*callback));
 	if (callback == NULL) {
 		RTE_LOG(ERR, EAL, "Can not allocate memory\n");
 		return -ENOMEM;
@@ -431,10 +429,9 @@ rte_intr_callback_register(const struct rte_intr_handle *intr_handle,
 
 	/* no existing callbacks for this - add new source */
 	if (src == NULL) {
-		if ((src = rte_zmalloc("interrupt source list",
-				sizeof(*src), 0)) == NULL) {
+		if ((src = calloc(1, sizeof(*src))) == NULL) {
 			RTE_LOG(ERR, EAL, "Can not allocate memory\n");
-			rte_free(callback);
+			free(callback);
 			ret = -ENOMEM;
 		} else {
 			src->intr_handle = *intr_handle;
@@ -501,7 +498,7 @@ rte_intr_callback_unregister(const struct rte_intr_handle *intr_handle,
 			if (cb->cb_fn == cb_fn && (cb_arg == (void *)-1 ||
 					cb->cb_arg == cb_arg)) {
 				TAILQ_REMOVE(&src->callbacks, cb, next);
-				rte_free(cb);
+				free(cb);
 				ret++;
 			}
 		}
@@ -509,7 +506,7 @@ rte_intr_callback_unregister(const struct rte_intr_handle *intr_handle,
 		/* all callbacks for that source are removed. */
 		if (TAILQ_EMPTY(&src->callbacks)) {
 			TAILQ_REMOVE(&intr_sources, src, next);
-			rte_free(src);
+			free(src);
 		}
 	}
 
-- 
2.7.4

[RFC 6/8] eal: bring forward init of interrupt handling

[Jianfeng Tan]

IPC will reply on interrupt handling, so we move forward the init
of interrupt handling.

Signed-off-by: Jianfeng Tan <jianfeng.tan@intel.com>
---
 lib/librte_eal/linuxapp/eal/eal.c | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/lib/librte_eal/linuxapp/eal/eal.c b/lib/librte_eal/linuxapp/eal/eal.c
index 5b23bf0..4f64c60 100644
--- a/lib/librte_eal/linuxapp/eal/eal.c
+++ b/lib/librte_eal/linuxapp/eal/eal.c
@@ -772,6 +772,11 @@ rte_eal_init(int argc, char **argv)
 
 	rte_config_init();
 
+	if (rte_eal_intr_init() < 0) {
+		rte_eal_init_alert("Cannot init interrupt-handling thread\n");
+		return -1;
+	}
+
 	/* Put mp channel init before bus scan so that we can init the vdev
 	 * bus through mp channel in the secondary process before the bus scan.
 	 */
@@ -901,11 +906,6 @@ rte_eal_init(int argc, char **argv)
 		rte_config.master_lcore, (int)thread_id, cpuset,
 		ret == 0 ? "" : "...");
 
-	if (rte_eal_intr_init() < 0) {
-		rte_eal_init_alert("Cannot init interrupt-handling thread\n");
-		return -1;
-	}
-
 	RTE_LCORE_FOREACH_SLAVE(i) {
 
 		/*
-- 
2.7.4

[RFC 7/8] eal: add IPC type for interrupt thread

[Jianfeng Tan]

We are going to merge IPC into interrupt thread. This patch adds
IPC type for interrupt thread.

Signed-off-by: Jianfeng Tan <jianfeng.tan@intel.com>
---
 lib/librte_eal/common/include/rte_eal_interrupts.h |  1 +
 lib/librte_eal/linuxapp/eal/eal_interrupts.c       |  5 ++++
 test/test/test_interrupts.c                        | 29 +++++++++++++++++++++-
 3 files changed, 34 insertions(+), 1 deletion(-)

diff --git a/lib/librte_eal/common/include/rte_eal_interrupts.h b/lib/librte_eal/common/include/rte_eal_interrupts.h
index 6eb4932..344db76 100644
--- a/lib/librte_eal/common/include/rte_eal_interrupts.h
+++ b/lib/librte_eal/common/include/rte_eal_interrupts.h
@@ -35,6 +35,7 @@ enum rte_intr_handle_type {
 	RTE_INTR_HANDLE_EXT,          /**< external handler */
 	RTE_INTR_HANDLE_VDEV,         /**< virtual device */
 	RTE_INTR_HANDLE_DEV_EVENT,    /**< device event handle */
+	RTE_INTR_HANDLE_IPC,          /**< IPC event handle */
 	RTE_INTR_HANDLE_MAX           /**< count of elements */
 };
 
diff --git a/lib/librte_eal/linuxapp/eal/eal_interrupts.c b/lib/librte_eal/linuxapp/eal/eal_interrupts.c
index 5245372..b23fd9e 100644
--- a/lib/librte_eal/linuxapp/eal/eal_interrupts.c
+++ b/lib/librte_eal/linuxapp/eal/eal_interrupts.c
@@ -559,6 +559,8 @@ rte_intr_enable(const struct rte_intr_handle *intr_handle)
 	/* not used at this moment */
 	case RTE_INTR_HANDLE_DEV_EVENT:
 		return -1;
+	case RTE_INTR_HANDLE_IPC:
+		return -1;
 	/* unknown handle type */
 	default:
 		RTE_LOG(ERR, EAL,
@@ -609,6 +611,8 @@ rte_intr_disable(const struct rte_intr_handle *intr_handle)
 	/* not used at this moment */
 	case RTE_INTR_HANDLE_DEV_EVENT:
 		return -1;
+	case RTE_INTR_HANDLE_IPC:
+		return -1;
 	/* unknown handle type */
 	default:
 		RTE_LOG(ERR, EAL,
@@ -678,6 +682,7 @@ eal_intr_process_interrupts(struct epoll_event *events, int nfds)
 			call = true;
 			break;
 		case RTE_INTR_HANDLE_DEV_EVENT:
+		case RTE_INTR_HANDLE_IPC:
 			bytes_read = 0;
 			call = true;
 			break;
diff --git a/test/test/test_interrupts.c b/test/test/test_interrupts.c
index dc19175..7c42d71 100644
--- a/test/test/test_interrupts.c
+++ b/test/test/test_interrupts.c
@@ -21,6 +21,7 @@ enum test_interrupt_handle_type {
 	TEST_INTERRUPT_HANDLE_VALID_UIO,
 	TEST_INTERRUPT_HANDLE_VALID_ALARM,
 	TEST_INTERRUPT_HANDLE_VALID_DEV_EVENT,
+	TEST_INTERRUPT_HANDLE_VALID_IPC,
 	TEST_INTERRUPT_HANDLE_CASE1,
 	TEST_INTERRUPT_HANDLE_MAX
 };
@@ -85,6 +86,10 @@ test_interrupt_init(void)
 	intr_handles[TEST_INTERRUPT_HANDLE_VALID_DEV_EVENT].type =
 					RTE_INTR_HANDLE_DEV_EVENT;
 
+	intr_handles[TEST_INTERRUPT_HANDLE_VALID_IPC].fd = pfds.readfd;
+	intr_handles[TEST_INTERRUPT_HANDLE_VALID_IPC].type =
+					RTE_INTR_HANDLE_IPC;
+
 	intr_handles[TEST_INTERRUPT_HANDLE_CASE1].fd = pfds.writefd;
 	intr_handles[TEST_INTERRUPT_HANDLE_CASE1].type = RTE_INTR_HANDLE_UIO;
 
@@ -263,6 +268,14 @@ test_interrupt_enable(void)
 		return -1;
 	}
 
+	/* check with specific valid intr_handle */
+	test_intr_handle = intr_handles[TEST_INTERRUPT_HANDLE_VALID_IPC];
+	if (rte_intr_enable(&test_intr_handle) == 0) {
+		printf("unexpectedly enable a specific intr_handle "
+			"successfully\n");
+		return -1;
+	}
+
 	/* check with valid handler and its type */
 	test_intr_handle = intr_handles[TEST_INTERRUPT_HANDLE_CASE1];
 	if (rte_intr_enable(&test_intr_handle) < 0) {
@@ -327,6 +340,14 @@ test_interrupt_disable(void)
 		return -1;
 	}
 
+	/* check with specific valid intr_handle */
+	test_intr_handle = intr_handles[TEST_INTERRUPT_HANDLE_VALID_IPC];
+	if (rte_intr_disable(&test_intr_handle) == 0) {
+		printf("unexpectedly disable a specific intr_handle "
+			"successfully\n");
+		return -1;
+	}
+
 	/* check with valid handler and its type */
 	test_intr_handle = intr_handles[TEST_INTERRUPT_HANDLE_CASE1];
 	if (rte_intr_disable(&test_intr_handle) < 0) {
@@ -424,7 +445,7 @@ test_interrupt(void)
 
 	printf("Check valid alarm interrupt full path\n");
 	if (test_interrupt_full_path_check(
-		TEST_INTERRUPT_HANDLE_VALID_DEV_EVENT) < 0) {
+		TEST_INTERRUPT_HANDLE_VALID_IPC) < 0) {
 		printf("failure occurred during checking valid alarm "
 						"interrupt full path\n");
 		goto out;
@@ -548,6 +569,12 @@ test_interrupt(void)
 	rte_intr_callback_unregister(&test_intr_handle,
 			test_interrupt_callback_1, (void *)-1);
 
+	test_intr_handle = intr_handles[TEST_INTERRUPT_HANDLE_VALID_VFIO];
+	rte_intr_callback_unregister(&test_intr_handle,
+			test_interrupt_callback, (void *)-1);
+	rte_intr_callback_unregister(&test_intr_handle,
+			test_interrupt_callback_1, (void *)-1);
+
 	rte_delay_ms(2 * TEST_INTERRUPT_CHECK_INTERVAL);
 	/* deinit */
 	test_interrupt_deinit();
-- 
2.7.4

[RFC 8/8] ipc: remove IPC thread for message read

[Jianfeng Tan]

As discussed here, http://dpdk.org/dev/patchwork/patch/36579/,
we remove IPC thread, rte_mp_handle, in this patch.

Previously, to handle requests from peer(s), or replies for a
request (sync or async) by itself, thread rte_mp_handle will
blockly readmsg on the unix socket for incoming messages.
Now, we change to make use of the IPC event callback mechanism
(added in previous patch); in other words, we merge this into
interrupt thread.

Cc: Anatoly Burakov <anatoly.burakov@intel.com>

Suggested-by: Thomas Monjalon <thomas@monjalon.net>
Signed-off-by: Jianfeng Tan <jianfeng.tan@intel.com>
---
 lib/librte_eal/common/eal_common_proc.c | 38 ++++++++++++++++-----------------
 1 file changed, 19 insertions(+), 19 deletions(-)

diff --git a/lib/librte_eal/common/eal_common_proc.c b/lib/librte_eal/common/eal_common_proc.c
index 4cb460e..1f102ce 100644
--- a/lib/librte_eal/common/eal_common_proc.c
+++ b/lib/librte_eal/common/eal_common_proc.c
@@ -25,6 +25,7 @@
 #include <rte_cycles.h>
 #include <rte_eal.h>
 #include <rte_errno.h>
+#include <rte_interrupts.h>
 #include <rte_lcore.h>
 #include <rte_log.h>
 #include <rte_tailq.h>
@@ -101,6 +102,8 @@ static struct {
 	/**< used in async requests only */
 };
 
+static struct rte_intr_handle ipc_intr_handle;
+
 static void async_reply_handle(void *arg);
 
 /* forward declarations */
@@ -339,18 +342,18 @@ process_msg(struct mp_msg_internal *m, struct sockaddr_un *s)
 	}
 }
 
-static void *
+static void
 mp_handle(void *arg __rte_unused)
 {
 	struct mp_msg_internal msg;
 	struct sockaddr_un sa;
 
 	while (1) {
-		if (read_msg(&msg, &sa) == 0)
-			process_msg(&msg, &sa);
-	}
+		if (read_msg(&msg, &sa) < 0)
+			break;
 
-	return NULL;
+		process_msg(&msg, &sa);
+	}
 }
 
 static int
@@ -548,10 +551,8 @@ unlink_sockets(const char *filter)
 int
 rte_mp_channel_init(void)
 {
-	char thread_name[RTE_MAX_THREAD_NAME_LEN];
 	char path[PATH_MAX];
 	int dir_fd;
-	pthread_t mp_handle_tid;
 
 	/* create filter path */
 	create_socket_path("*", path, sizeof(path));
@@ -579,32 +580,31 @@ rte_mp_channel_init(void)
 	if (rte_eal_process_type() == RTE_PROC_PRIMARY &&
 			unlink_sockets(mp_filter)) {
 		RTE_LOG(ERR, EAL, "failed to unlink mp sockets\n");
-		close(dir_fd);
-		return -1;
+		goto res;
 	}
 
 	if (open_socket_fd() < 0) {
-		close(dir_fd);
-		return -1;
+		mp_fd = -1;
+		goto res;
 	}
 
-	if (pthread_create(&mp_handle_tid, NULL, mp_handle, NULL) < 0) {
-		RTE_LOG(ERR, EAL, "failed to create mp thead: %s\n",
+	ipc_intr_handle.fd = mp_fd;
+	ipc_intr_handle.type = RTE_INTR_HANDLE_IPC;
+
+	if (rte_intr_callback_register(&ipc_intr_handle,
+					mp_handle, NULL) < 0) {
+                RTE_LOG(ERR, EAL, "failed to register IPC callback: %s\n",
 			strerror(errno));
 		close(mp_fd);
 		mp_fd = -1;
-		return -1;
 	}
 
-	/* try best to set thread name */
-	snprintf(thread_name, RTE_MAX_THREAD_NAME_LEN, "rte_mp_handle");
-	rte_thread_setname(mp_handle_tid, thread_name);
-
+res:
 	/* unlock the directory */
 	flock(dir_fd, LOCK_UN);
 	close(dir_fd);
 
-	return 0;
+	return (mp_fd < 0) ? -1 : 0;
 }
 
 /**
-- 
2.7.4

Re: [RFC 0/8] remove IPC threads

[Thomas Monjalon]

19/04/2018 17:03, Jianfeng Tan:
> As discussed here, http://dpdk.org/dev/patchwork/patch/36579/, we will
> try best to avoid thread creation in the library. Now we have two threads
> for IPC, rte_mp_handle and rte_mp_handle_async. This patchset is to
> finish the job.

Thanks for working on it.

> This patchset:
> - is rebased on commit 34345a9b69bb ("eventdev: fix build with icc").
> - also needs http://dpdk.org/dev/patchwork/patch/37335/
> 
> Patch 1~2: code cleanup and bug fix. (Target for v18.05)
> Patch 3~4: Remove IPC async thread. (Target for v18.08)
> Patch 5~8: Remove IPC thread. (Target for v18.08)

You should send fixes for 18.05 in a separate series.
It is preferred to merge full series or nothing.

[RFC v2 0/6] Remove IPC threads

[Anatoly Burakov]

As previously discussed [1], IPC threads need to be removed and their
workload moved to interrupt thread.

The transition is complete as far as Linux support is concerned, however
since there is no interrupt thread on FreeBSD, this patchset effectively
disables IPC on FreeBSD for now (hence it still being an RFC and not a v1).

Work on adding interrupt thread to FreeBSD is in progress.

[1] http://dpdk.org/dev/patchwork/patch/36579/

Anatoly Burakov (2):
  ipc: remove IPC thread for async requests
  ipc: remove main IPC thread

Jianfeng Tan (4):
  eal/linux: use glibc malloc in alarm
  eal/linux: use glibc malloc in interrupt handling
  eal: bring forward init of interrupt handling
  eal: add IPC type for interrupt thread

 lib/librte_eal/common/eal_common_proc.c       | 233 +++++++-----------
 .../common/include/rte_eal_interrupts.h       |   1 +
 lib/librte_eal/linuxapp/eal/eal.c             |  10 +-
 lib/librte_eal/linuxapp/eal/eal_alarm.c       |   9 +-
 lib/librte_eal/linuxapp/eal/eal_interrupts.c  |  19 +-
 test/test/test_interrupts.c                   |  29 ++-
 6 files changed, 137 insertions(+), 164 deletions(-)

-- 
2.17.0

[RFC v2 1/6] eal/linux: use glibc malloc in alarm

[Anatoly Burakov]

From: Jianfeng Tan <jianfeng.tan@intel.com>

We will reply on alarm API for async IPC request as following patch
indicates. rte_malloc could require async IPC request.

To avoid such chicken or the egg causality dilemma, we change to
use glibc malloc in alarm implimentation.

Signed-off-by: Jianfeng Tan <jianfeng.tan@intel.com>
Signed-off-by: Anatoly Burakov <anatoly.burakov@intel.com>
---
 lib/librte_eal/linuxapp/eal/eal_alarm.c | 9 ++++-----
 1 file changed, 4 insertions(+), 5 deletions(-)

diff --git a/lib/librte_eal/linuxapp/eal/eal_alarm.c b/lib/librte_eal/linuxapp/eal/eal_alarm.c
index c115e823a..391d2a65f 100644
--- a/lib/librte_eal/linuxapp/eal/eal_alarm.c
+++ b/lib/librte_eal/linuxapp/eal/eal_alarm.c
@@ -19,7 +19,6 @@
 #include <rte_launch.h>
 #include <rte_lcore.h>
 #include <rte_errno.h>
-#include <rte_malloc.h>
 #include <rte_spinlock.h>
 #include <eal_private.h>
 
@@ -91,7 +90,7 @@ eal_alarm_callback(void *arg __rte_unused)
 		rte_spinlock_lock(&alarm_list_lk);
 
 		LIST_REMOVE(ap, next);
-		rte_free(ap);
+		free(ap);
 	}
 
 	if (!LIST_EMPTY(&alarm_list)) {
@@ -122,7 +121,7 @@ rte_eal_alarm_set(uint64_t us, rte_eal_alarm_callback cb_fn, void *cb_arg)
 	if (us < 1 || us > (UINT64_MAX - US_PER_S) || cb_fn == NULL)
 		return -EINVAL;
 
-	new_alarm = rte_zmalloc(NULL, sizeof(*new_alarm), 0);
+	new_alarm = calloc(1, sizeof(*new_alarm));
 	if (new_alarm == NULL)
 		return -ENOMEM;
 
@@ -196,7 +195,7 @@ rte_eal_alarm_cancel(rte_eal_alarm_callback cb_fn, void *cb_arg)
 
 			if (ap->executing == 0) {
 				LIST_REMOVE(ap, next);
-				rte_free(ap);
+				free(ap);
 				count++;
 			} else {
 				/* If calling from other context, mark that alarm is executing
@@ -220,7 +219,7 @@ rte_eal_alarm_cancel(rte_eal_alarm_callback cb_fn, void *cb_arg)
 
 				if (ap->executing == 0) {
 					LIST_REMOVE(ap, next);
-					rte_free(ap);
+					free(ap);
 					count++;
 					ap = ap_prev;
 				} else if (pthread_equal(ap->executing_id, pthread_self()) == 0)
-- 
2.17.0

[RFC v2 2/6] ipc: remove IPC thread for async requests

[Anatoly Burakov]

Previously, we were using two IPC threads - one to handle messages
and synchronous requests, and another to handle asynchronous requests.
To handle replies for an async request, rte_mp_handle woke up the
rte_mp_handle_async thread to process through pthread_cond variable.

Change it to handle asynchronous messages within the main IPC thread.
To handle timeout events, for each async request which is sent,
we set an alarm for it. If its reply is received before timeout,
we will cancel the alarm when we handle the reply; otherwise,
alarm will invoke the async_reply_handle() as the alarm callback.

Signed-off-by: Jianfeng Tan <jianfeng.tan@intel.com>
Signed-off-by: Anatoly Burakov <anatoly.burakov@intel.com>
Suggested-by: Thomas Monjalon <thomas@monjalon.net>
---

Notes:
    RFC->RFCv2:
    - Rebased on latest code
    - Implemented comments to the original RFC

 lib/librte_eal/common/eal_common_proc.c | 191 ++++++++----------------
 1 file changed, 65 insertions(+), 126 deletions(-)

diff --git a/lib/librte_eal/common/eal_common_proc.c b/lib/librte_eal/common/eal_common_proc.c
index 707d8ab30..6f3366403 100644
--- a/lib/librte_eal/common/eal_common_proc.c
+++ b/lib/librte_eal/common/eal_common_proc.c
@@ -20,6 +20,7 @@
 #include <sys/un.h>
 #include <unistd.h>
 
+#include <rte_alarm.h>
 #include <rte_common.h>
 #include <rte_cycles.h>
 #include <rte_eal.h>
@@ -94,11 +95,9 @@ TAILQ_HEAD(pending_request_list, pending_request);
 static struct {
 	struct pending_request_list requests;
 	pthread_mutex_t lock;
-	pthread_cond_t async_cond;
 } pending_requests = {
 	.requests = TAILQ_HEAD_INITIALIZER(pending_requests.requests),
 	.lock = PTHREAD_MUTEX_INITIALIZER,
-	.async_cond = PTHREAD_COND_INITIALIZER
 	/**< used in async requests only */
 };
 
@@ -106,6 +105,16 @@ static struct {
 static int
 mp_send(struct rte_mp_msg *msg, const char *peer, int type);
 
+/* for use with alarm callback */
+static void
+async_reply_handle(void *arg);
+
+/* for use with process_msg */
+static struct pending_request *
+async_reply_handle_thread_unsafe(void *arg);
+
+static void
+trigger_async_action(struct pending_request *req);
 
 static struct pending_request *
 find_pending_request(const char *dst, const char *act_name)
@@ -290,6 +299,8 @@ process_msg(struct mp_msg_internal *m, struct sockaddr_un *s)
 	RTE_LOG(DEBUG, EAL, "msg: %s\n", msg->name);
 
 	if (m->type == MP_REP || m->type == MP_IGN) {
+		struct pending_request *req = NULL;
+
 		pthread_mutex_lock(&pending_requests.lock);
 		pending_req = find_pending_request(s->sun_path, msg->name);
 		if (pending_req) {
@@ -301,11 +312,14 @@ process_msg(struct mp_msg_internal *m, struct sockaddr_un *s)
 			if (pending_req->type == REQUEST_TYPE_SYNC)
 				pthread_cond_signal(&pending_req->sync.cond);
 			else if (pending_req->type == REQUEST_TYPE_ASYNC)
-				pthread_cond_signal(
-					&pending_requests.async_cond);
+				req = async_reply_handle_thread_unsafe(
+						pending_req);
 		} else
 			RTE_LOG(ERR, EAL, "Drop mp reply: %s\n", msg->name);
 		pthread_mutex_unlock(&pending_requests.lock);
+
+		if (req != NULL)
+			trigger_async_action(req);
 		return;
 	}
 
@@ -365,7 +379,6 @@ timespec_cmp(const struct timespec *a, const struct timespec *b)
 }
 
 enum async_action {
-	ACTION_NONE, /**< don't do anything */
 	ACTION_FREE, /**< free the action entry, but don't trigger callback */
 	ACTION_TRIGGER /**< trigger callback, then free action entry */
 };
@@ -375,7 +388,7 @@ process_async_request(struct pending_request *sr, const struct timespec *now)
 {
 	struct async_request_param *param;
 	struct rte_mp_reply *reply;
-	bool timeout, received, last_msg;
+	bool timeout, last_msg;
 
 	param = sr->async.param;
 	reply = &param->user_reply;
@@ -383,13 +396,6 @@ process_async_request(struct pending_request *sr, const struct timespec *now)
 	/* did we timeout? */
 	timeout = timespec_cmp(&param->end, now) <= 0;
 
-	/* did we receive a response? */
-	received = sr->reply_received != 0;
-
-	/* if we didn't time out, and we didn't receive a response, ignore */
-	if (!timeout && !received)
-		return ACTION_NONE;
-
 	/* if we received a response, adjust relevant data and copy mesasge. */
 	if (sr->reply_received == 1 && sr->reply) {
 		struct rte_mp_msg *msg, *user_msgs, *tmp;
@@ -448,118 +454,58 @@ trigger_async_action(struct pending_request *sr)
 	free(sr->async.param->user_reply.msgs);
 	free(sr->async.param);
 	free(sr->request);
+	free(sr);
 }
 
 static struct pending_request *
-check_trigger(struct timespec *ts)
+async_reply_handle_thread_unsafe(void *arg)
 {
-	struct pending_request *next, *cur, *trigger = NULL;
-
-	TAILQ_FOREACH_SAFE(cur, &pending_requests.requests, next, next) {
-		enum async_action action;
-		if (cur->type != REQUEST_TYPE_ASYNC)
-			continue;
+	struct pending_request *req = (struct pending_request *)arg;
+	enum async_action action;
+	struct timespec ts_now;
+	struct timeval now;
 
-		action = process_async_request(cur, ts);
-		if (action == ACTION_FREE) {
-			TAILQ_REMOVE(&pending_requests.requests, cur, next);
-			free(cur);
-		} else if (action == ACTION_TRIGGER) {
-			TAILQ_REMOVE(&pending_requests.requests, cur, next);
-			trigger = cur;
-			break;
-		}
+	if (gettimeofday(&now, NULL) < 0) {
+		RTE_LOG(ERR, EAL, "Cannot get current time\n");
+		goto no_trigger;
 	}
-	return trigger;
-}
+	ts_now.tv_nsec = now.tv_usec * 1000;
+	ts_now.tv_sec = now.tv_sec;
 
-static void
-wait_for_async_messages(void)
-{
-	struct pending_request *sr;
-	struct timespec timeout;
-	bool timedwait = false;
-	bool nowait = false;
-	int ret;
+	action = process_async_request(req, &ts_now);
 
-	/* scan through the list and see if there are any timeouts that
-	 * are earlier than our current timeout.
-	 */
-	TAILQ_FOREACH(sr, &pending_requests.requests, next) {
-		if (sr->type != REQUEST_TYPE_ASYNC)
-			continue;
-		if (!timedwait || timespec_cmp(&sr->async.param->end,
-				&timeout) < 0) {
-			memcpy(&timeout, &sr->async.param->end,
-				sizeof(timeout));
-			timedwait = true;
-		}
+	TAILQ_REMOVE(&pending_requests.requests, req, next);
 
-		/* sometimes, we don't even wait */
-		if (sr->reply_received) {
-			nowait = true;
-			break;
+	if (rte_eal_alarm_cancel(async_reply_handle, req) < 0) {
+		/* if we failed to cancel the alarm because it's already in
+		 * progress, don't proceed because otherwise we will end up
+		 * handling the same message twice.
+		 */
+		if (rte_errno == EINPROGRESS) {
+			RTE_LOG(DEBUG, EAL, "Request handling is already in progress\n");
+			goto no_trigger;
 		}
+		RTE_LOG(ERR, EAL, "Failed to cancel alarm\n");
 	}
 
-	if (nowait)
-		return;
-
-	do {
-		ret = timedwait ?
-			pthread_cond_timedwait(
-				&pending_requests.async_cond,
-				&pending_requests.lock,
-				&timeout) :
-			pthread_cond_wait(
-				&pending_requests.async_cond,
-				&pending_requests.lock);
-	} while (ret != 0 && ret != ETIMEDOUT);
-
-	/* we've been woken up or timed out */
+	if (action == ACTION_TRIGGER)
+		return req;
+no_trigger:
+	free(req);
+	return NULL;
 }
 
-static void *
-async_reply_handle(void *arg __rte_unused)
+static void
+async_reply_handle(void *arg)
 {
-	struct timeval now;
-	struct timespec ts_now;
-	while (1) {
-		struct pending_request *trigger = NULL;
+	struct pending_request *req;
 
-		pthread_mutex_lock(&pending_requests.lock);
-
-		/* we exit this function holding the lock */
-		wait_for_async_messages();
-
-		if (gettimeofday(&now, NULL) < 0) {
-			pthread_mutex_unlock(&pending_requests.lock);
-			RTE_LOG(ERR, EAL, "Cannot get current time\n");
-			break;
-		}
-		ts_now.tv_nsec = now.tv_usec * 1000;
-		ts_now.tv_sec = now.tv_sec;
-
-		do {
-			trigger = check_trigger(&ts_now);
-			/* unlock request list */
-			pthread_mutex_unlock(&pending_requests.lock);
-
-			if (trigger) {
-				trigger_async_action(trigger);
-				free(trigger);
-
-				/* we've triggered a callback, but there may be
-				 * more, so lock the list and check again.
-				 */
-				pthread_mutex_lock(&pending_requests.lock);
-			}
-		} while (trigger);
-	}
-
-	RTE_LOG(ERR, EAL, "ERROR: asynchronous requests disabled\n");
+	pthread_mutex_lock(&pending_requests.lock);
+	req = async_reply_handle_thread_unsafe(arg);
+	pthread_mutex_unlock(&pending_requests.lock);
 
-	return NULL;
+	if (req != NULL)
+		trigger_async_action(req);
 }
 
 static int
@@ -624,7 +570,7 @@ rte_mp_channel_init(void)
 {
 	char path[PATH_MAX];
 	int dir_fd;
-	pthread_t mp_handle_tid, async_reply_handle_tid;
+	pthread_t mp_handle_tid;
 
 	/* create filter path */
 	create_socket_path("*", path, sizeof(path));
@@ -671,17 +617,6 @@ rte_mp_channel_init(void)
 		return -1;
 	}
 
-	if (rte_ctrl_thread_create(&async_reply_handle_tid,
-			"rte_mp_async", NULL,
-			async_reply_handle, NULL) < 0) {
-		RTE_LOG(ERR, EAL, "failed to create mp thead: %s\n",
-			strerror(errno));
-		close(mp_fd);
-		close(dir_fd);
-		mp_fd = -1;
-		return -1;
-	}
-
 	/* unlock the directory */
 	flock(dir_fd, LOCK_UN);
 	close(dir_fd);
@@ -853,7 +788,7 @@ rte_mp_sendmsg(struct rte_mp_msg *msg)
 
 static int
 mp_request_async(const char *dst, struct rte_mp_msg *req,
-		struct async_request_param *param)
+		struct async_request_param *param, const struct timespec *ts)
 {
 	struct rte_mp_msg *reply_msg;
 	struct pending_request *pending_req, *exist;
@@ -898,6 +833,13 @@ mp_request_async(const char *dst, struct rte_mp_msg *req,
 
 	param->user_reply.nb_sent++;
 
+	if (rte_eal_alarm_set(ts->tv_sec * 1000000 + ts->tv_nsec / 1000,
+			      async_reply_handle, pending_req) < 0) {
+		RTE_LOG(ERR, EAL, "Fail to set alarm for request %s:%s\n",
+			dst, req->name);
+		rte_panic("Fix the above shit to properly free all memory\n");
+	}
+
 	return 0;
 fail:
 	free(pending_req);
@@ -1119,7 +1061,7 @@ rte_mp_request_async(struct rte_mp_msg *req, const struct timespec *ts,
 
 	/* for secondary process, send request to the primary process only */
 	if (rte_eal_process_type() == RTE_PROC_SECONDARY) {
-		ret = mp_request_async(eal_mp_socket_path(), copy, param);
+		ret = mp_request_async(eal_mp_socket_path(), copy, param, ts);
 
 		/* if we didn't send anything, put dummy request on the queue */
 		if (ret == 0 && reply->nb_sent == 0) {
@@ -1162,7 +1104,7 @@ rte_mp_request_async(struct rte_mp_msg *req, const struct timespec *ts,
 		snprintf(path, sizeof(path), "%s/%s", mp_dir_path,
 			 ent->d_name);
 
-		if (mp_request_async(path, copy, param))
+		if (mp_request_async(path, copy, param, ts))
 			ret = -1;
 	}
 	/* if we didn't send anything, put dummy request on the queue */
@@ -1171,9 +1113,6 @@ rte_mp_request_async(struct rte_mp_msg *req, const struct timespec *ts,
 		dummy_used = true;
 	}
 
-	/* trigger async request thread wake up */
-	pthread_cond_signal(&pending_requests.async_cond);
-
 	/* finally, unlock the queue */
 	pthread_mutex_unlock(&pending_requests.lock);
 
-- 
2.17.0

[RFC v2 3/6] eal/linux: use glibc malloc in interrupt handling

[Anatoly Burakov]

From: Jianfeng Tan <jianfeng.tan@intel.com>

We will rely on interrupt thread to implement IPC; and IPC initialization
is in very early stage, when memory subsystem is not initialized yet.
So we change to use glibc malloc/free.

Signed-off-by: Jianfeng Tan <jianfeng.tan@intel.com>
Signed-off-by: Anatoly Burakov <anatoly.burakov@intel.com>
---
 lib/librte_eal/linuxapp/eal/eal_interrupts.c | 14 ++++++--------
 1 file changed, 6 insertions(+), 8 deletions(-)

diff --git a/lib/librte_eal/linuxapp/eal/eal_interrupts.c b/lib/librte_eal/linuxapp/eal/eal_interrupts.c
index 056d41c12..180c0378a 100644
--- a/lib/librte_eal/linuxapp/eal/eal_interrupts.c
+++ b/lib/librte_eal/linuxapp/eal/eal_interrupts.c
@@ -30,7 +30,6 @@
 #include <rte_branch_prediction.h>
 #include <rte_debug.h>
 #include <rte_log.h>
-#include <rte_malloc.h>
 #include <rte_errno.h>
 #include <rte_spinlock.h>
 #include <rte_pause.h>
@@ -405,8 +404,7 @@ rte_intr_callback_register(const struct rte_intr_handle *intr_handle,
 	}
 
 	/* allocate a new interrupt callback entity */
-	callback = rte_zmalloc("interrupt callback list",
-				sizeof(*callback), 0);
+	callback = calloc(1, sizeof(*callback));
 	if (callback == NULL) {
 		RTE_LOG(ERR, EAL, "Can not allocate memory\n");
 		return -ENOMEM;
@@ -431,10 +429,10 @@ rte_intr_callback_register(const struct rte_intr_handle *intr_handle,
 
 	/* no existing callbacks for this - add new source */
 	if (src == NULL) {
-		if ((src = rte_zmalloc("interrupt source list",
-				sizeof(*src), 0)) == NULL) {
+		src = calloc(1, sizeof(*src));
+		if (src == NULL) {
 			RTE_LOG(ERR, EAL, "Can not allocate memory\n");
-			rte_free(callback);
+			free(callback);
 			ret = -ENOMEM;
 		} else {
 			src->intr_handle = *intr_handle;
@@ -501,7 +499,7 @@ rte_intr_callback_unregister(const struct rte_intr_handle *intr_handle,
 			if (cb->cb_fn == cb_fn && (cb_arg == (void *)-1 ||
 					cb->cb_arg == cb_arg)) {
 				TAILQ_REMOVE(&src->callbacks, cb, next);
-				rte_free(cb);
+				free(cb);
 				ret++;
 			}
 		}
@@ -509,7 +507,7 @@ rte_intr_callback_unregister(const struct rte_intr_handle *intr_handle,
 		/* all callbacks for that source are removed. */
 		if (TAILQ_EMPTY(&src->callbacks)) {
 			TAILQ_REMOVE(&intr_sources, src, next);
-			rte_free(src);
+			free(src);
 		}
 	}
 
-- 
2.17.0

[RFC v2 4/6] eal: bring forward init of interrupt handling

[Anatoly Burakov]

From: Jianfeng Tan <jianfeng.tan@intel.com>

IPC will reply on interrupt handling, so we move forward the init
of interrupt handling.

Signed-off-by: Jianfeng Tan <jianfeng.tan@intel.com>
Signed-off-by: Anatoly Burakov <anatoly.burakov@intel.com>
---
 lib/librte_eal/linuxapp/eal/eal.c | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/lib/librte_eal/linuxapp/eal/eal.c b/lib/librte_eal/linuxapp/eal/eal.c
index 8655b8691..f8a0c06d7 100644
--- a/lib/librte_eal/linuxapp/eal/eal.c
+++ b/lib/librte_eal/linuxapp/eal/eal.c
@@ -839,6 +839,11 @@ rte_eal_init(int argc, char **argv)
 
 	rte_config_init();
 
+	if (rte_eal_intr_init() < 0) {
+		rte_eal_init_alert("Cannot init interrupt-handling thread\n");
+		return -1;
+	}
+
 	/* Put mp channel init before bus scan so that we can init the vdev
 	 * bus through mp channel in the secondary process before the bus scan.
 	 */
@@ -968,11 +973,6 @@ rte_eal_init(int argc, char **argv)
 		rte_config.master_lcore, (int)thread_id, cpuset,
 		ret == 0 ? "" : "...");
 
-	if (rte_eal_intr_init() < 0) {
-		rte_eal_init_alert("Cannot init interrupt-handling thread\n");
-		return -1;
-	}
-
 	RTE_LCORE_FOREACH_SLAVE(i) {
 
 		/*
-- 
2.17.0

[RFC v2 5/6] eal: add IPC type for interrupt thread

[Anatoly Burakov]

From: Jianfeng Tan <jianfeng.tan@intel.com>

We are going to merge IPC into interrupt thread. This patch adds
IPC type for interrupt thread.

Signed-off-by: Jianfeng Tan <jianfeng.tan@intel.com>
Signed-off-by: Anatoly Burakov <anatoly.burakov@intel.com>
---

Notes:
    RFC->RFCv2:
    - Fixed typo in test app

 .../common/include/rte_eal_interrupts.h       |  1 +
 lib/librte_eal/linuxapp/eal/eal_interrupts.c  |  5 ++++
 test/test/test_interrupts.c                   | 29 ++++++++++++++++++-
 3 files changed, 34 insertions(+), 1 deletion(-)

diff --git a/lib/librte_eal/common/include/rte_eal_interrupts.h b/lib/librte_eal/common/include/rte_eal_interrupts.h
index 6eb493271..344db768d 100644
--- a/lib/librte_eal/common/include/rte_eal_interrupts.h
+++ b/lib/librte_eal/common/include/rte_eal_interrupts.h
@@ -35,6 +35,7 @@ enum rte_intr_handle_type {
 	RTE_INTR_HANDLE_EXT,          /**< external handler */
 	RTE_INTR_HANDLE_VDEV,         /**< virtual device */
 	RTE_INTR_HANDLE_DEV_EVENT,    /**< device event handle */
+	RTE_INTR_HANDLE_IPC,          /**< IPC event handle */
 	RTE_INTR_HANDLE_MAX           /**< count of elements */
 };
 
diff --git a/lib/librte_eal/linuxapp/eal/eal_interrupts.c b/lib/librte_eal/linuxapp/eal/eal_interrupts.c
index 180c0378a..390672739 100644
--- a/lib/librte_eal/linuxapp/eal/eal_interrupts.c
+++ b/lib/librte_eal/linuxapp/eal/eal_interrupts.c
@@ -560,6 +560,8 @@ rte_intr_enable(const struct rte_intr_handle *intr_handle)
 	/* not used at this moment */
 	case RTE_INTR_HANDLE_DEV_EVENT:
 		return -1;
+	case RTE_INTR_HANDLE_IPC:
+		return -1;
 	/* unknown handle type */
 	default:
 		RTE_LOG(ERR, EAL,
@@ -610,6 +612,8 @@ rte_intr_disable(const struct rte_intr_handle *intr_handle)
 	/* not used at this moment */
 	case RTE_INTR_HANDLE_DEV_EVENT:
 		return -1;
+	case RTE_INTR_HANDLE_IPC:
+		return -1;
 	/* unknown handle type */
 	default:
 		RTE_LOG(ERR, EAL,
@@ -679,6 +683,7 @@ eal_intr_process_interrupts(struct epoll_event *events, int nfds)
 			call = true;
 			break;
 		case RTE_INTR_HANDLE_DEV_EVENT:
+		case RTE_INTR_HANDLE_IPC:
 			bytes_read = 0;
 			call = true;
 			break;
diff --git a/test/test/test_interrupts.c b/test/test/test_interrupts.c
index dc19175d3..fa18ddf75 100644
--- a/test/test/test_interrupts.c
+++ b/test/test/test_interrupts.c
@@ -21,6 +21,7 @@ enum test_interrupt_handle_type {
 	TEST_INTERRUPT_HANDLE_VALID_UIO,
 	TEST_INTERRUPT_HANDLE_VALID_ALARM,
 	TEST_INTERRUPT_HANDLE_VALID_DEV_EVENT,
+	TEST_INTERRUPT_HANDLE_VALID_IPC,
 	TEST_INTERRUPT_HANDLE_CASE1,
 	TEST_INTERRUPT_HANDLE_MAX
 };
@@ -85,6 +86,10 @@ test_interrupt_init(void)
 	intr_handles[TEST_INTERRUPT_HANDLE_VALID_DEV_EVENT].type =
 					RTE_INTR_HANDLE_DEV_EVENT;
 
+	intr_handles[TEST_INTERRUPT_HANDLE_VALID_IPC].fd = pfds.readfd;
+	intr_handles[TEST_INTERRUPT_HANDLE_VALID_IPC].type =
+					RTE_INTR_HANDLE_IPC;
+
 	intr_handles[TEST_INTERRUPT_HANDLE_CASE1].fd = pfds.writefd;
 	intr_handles[TEST_INTERRUPT_HANDLE_CASE1].type = RTE_INTR_HANDLE_UIO;
 
@@ -263,6 +268,14 @@ test_interrupt_enable(void)
 		return -1;
 	}
 
+	/* check with specific valid intr_handle */
+	test_intr_handle = intr_handles[TEST_INTERRUPT_HANDLE_VALID_IPC];
+	if (rte_intr_enable(&test_intr_handle) == 0) {
+		printf("unexpectedly enable a specific intr_handle "
+			"successfully\n");
+		return -1;
+	}
+
 	/* check with valid handler and its type */
 	test_intr_handle = intr_handles[TEST_INTERRUPT_HANDLE_CASE1];
 	if (rte_intr_enable(&test_intr_handle) < 0) {
@@ -327,6 +340,14 @@ test_interrupt_disable(void)
 		return -1;
 	}
 
+	/* check with specific valid intr_handle */
+	test_intr_handle = intr_handles[TEST_INTERRUPT_HANDLE_VALID_IPC];
+	if (rte_intr_disable(&test_intr_handle) == 0) {
+		printf("unexpectedly disable a specific intr_handle "
+			"successfully\n");
+		return -1;
+	}
+
 	/* check with valid handler and its type */
 	test_intr_handle = intr_handles[TEST_INTERRUPT_HANDLE_CASE1];
 	if (rte_intr_disable(&test_intr_handle) < 0) {
@@ -424,7 +445,7 @@ test_interrupt(void)
 
 	printf("Check valid alarm interrupt full path\n");
 	if (test_interrupt_full_path_check(
-		TEST_INTERRUPT_HANDLE_VALID_DEV_EVENT) < 0) {
+		TEST_INTERRUPT_HANDLE_VALID_IPC) < 0) {
 		printf("failure occurred during checking valid alarm "
 						"interrupt full path\n");
 		goto out;
@@ -548,6 +569,12 @@ test_interrupt(void)
 	rte_intr_callback_unregister(&test_intr_handle,
 			test_interrupt_callback_1, (void *)-1);
 
+	test_intr_handle = intr_handles[TEST_INTERRUPT_HANDLE_VALID_IPC];
+	rte_intr_callback_unregister(&test_intr_handle,
+			test_interrupt_callback, (void *)-1);
+	rte_intr_callback_unregister(&test_intr_handle,
+			test_interrupt_callback_1, (void *)-1);
+
 	rte_delay_ms(2 * TEST_INTERRUPT_CHECK_INTERVAL);
 	/* deinit */
 	test_interrupt_deinit();
-- 
2.17.0

[RFC v2 6/6] ipc: remove main IPC thread

[Anatoly Burakov]

Previously, to handle requests from peer(s), or replies for a
request (sync or async) by itself, a dedicated IPC thread was set
up.

Now that every other piece of the puzzle is in place, we can get rid
of the IPC thread, and move waiting for IPC messages entirely into the
interrupt thread.

Signed-off-by: Jianfeng Tan <jianfeng.tan@intel.com>
Signed-off-by: Anatoly Burakov <anatoly.burakov@intel.com>
Suggested-by: Thomas Monjalon <thomas@monjalon.net>
---

Notes:
    RFC->RFCv2:
    - Fixed resource leaks
    - Improved readability

 lib/librte_eal/common/eal_common_proc.c | 46 ++++++++++++++-----------
 1 file changed, 25 insertions(+), 21 deletions(-)

diff --git a/lib/librte_eal/common/eal_common_proc.c b/lib/librte_eal/common/eal_common_proc.c
index 6f3366403..162d67ca5 100644
--- a/lib/librte_eal/common/eal_common_proc.c
+++ b/lib/librte_eal/common/eal_common_proc.c
@@ -25,6 +25,7 @@
 #include <rte_cycles.h>
 #include <rte_eal.h>
 #include <rte_errno.h>
+#include <rte_interrupts.h>
 #include <rte_lcore.h>
 #include <rte_log.h>
 #include <rte_tailq.h>
@@ -101,6 +102,8 @@ static struct {
 	/**< used in async requests only */
 };
 
+static struct rte_intr_handle ipc_intr_handle;
+
 /* forward declarations */
 static int
 mp_send(struct rte_mp_msg *msg, const char *peer, int type);
@@ -350,18 +353,17 @@ process_msg(struct mp_msg_internal *m, struct sockaddr_un *s)
 	}
 }
 
-static void *
+static void
 mp_handle(void *arg __rte_unused)
 {
 	struct mp_msg_internal msg;
 	struct sockaddr_un sa;
 
 	while (1) {
-		if (read_msg(&msg, &sa) == 0)
-			process_msg(&msg, &sa);
+		if (read_msg(&msg, &sa) < 0)
+			break;
+		process_msg(&msg, &sa);
 	}
-
-	return NULL;
 }
 
 static int
@@ -570,7 +572,6 @@ rte_mp_channel_init(void)
 {
 	char path[PATH_MAX];
 	int dir_fd;
-	pthread_t mp_handle_tid;
 
 	/* create filter path */
 	create_socket_path("*", path, sizeof(path));
@@ -585,36 +586,32 @@ rte_mp_channel_init(void)
 	if (dir_fd < 0) {
 		RTE_LOG(ERR, EAL, "failed to open %s: %s\n",
 			mp_dir_path, strerror(errno));
-		return -1;
+		goto fail;
 	}
 
 	if (flock(dir_fd, LOCK_EX)) {
 		RTE_LOG(ERR, EAL, "failed to lock %s: %s\n",
 			mp_dir_path, strerror(errno));
-		close(dir_fd);
-		return -1;
+		goto fail;
 	}
 
 	if (rte_eal_process_type() == RTE_PROC_PRIMARY &&
 			unlink_sockets(mp_filter)) {
 		RTE_LOG(ERR, EAL, "failed to unlink mp sockets\n");
-		close(dir_fd);
-		return -1;
+		goto fail;
 	}
 
 	if (open_socket_fd() < 0) {
-		close(dir_fd);
-		return -1;
+		goto fail;
 	}
 
-	if (rte_ctrl_thread_create(&mp_handle_tid, "rte_mp_handle",
-			NULL, mp_handle, NULL) < 0) {
-		RTE_LOG(ERR, EAL, "failed to create mp thead: %s\n",
-			strerror(errno));
-		close(mp_fd);
-		close(dir_fd);
-		mp_fd = -1;
-		return -1;
+	ipc_intr_handle.fd = mp_fd;
+	ipc_intr_handle.type = RTE_INTR_HANDLE_IPC;
+
+	if (rte_intr_callback_register(&ipc_intr_handle, mp_handle, NULL) < 0) {
+		RTE_LOG(ERR, EAL, "failed to register IPC interrupt callback: %s\n",
+				strerror(errno));
+		goto fail;
 	}
 
 	/* unlock the directory */
@@ -622,6 +619,13 @@ rte_mp_channel_init(void)
 	close(dir_fd);
 
 	return 0;
+fail:
+	if (dir_fd >= 0)
+		close(dir_fd);
+	if (mp_fd >= 0)
+		close(mp_fd);
+	mp_fd = -1;
+	return -1;
 }
 
 /**
-- 
2.17.0