+ 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);