lib/librte_eal/common/eal_common_proc.c

   1 /* SPDX-License-Identifier: BSD-3-Clause
   2  * Copyright(c) 2016-2018 Intel Corporation
   3  */
   4
   5 #include <dirent.h>
   6 #include <errno.h>
   7 #include <fcntl.h>
   8 #include <fnmatch.h>
   9 #include <inttypes.h>
  10 #include <libgen.h>
  11 #include <limits.h>
  12 #include <pthread.h>
  13 #include <stdio.h>
  14 #include <stdlib.h>
  15 #include <string.h>
  16 #include <sys/time.h>
  17 #include <sys/types.h>
  18 #include <sys/socket.h>
  19 #include <sys/un.h>
  20 #include <unistd.h>
  21
  22 #include <rte_common.h>
  23 #include <rte_cycles.h>
  24 #include <rte_eal.h>
  25 #include <rte_errno.h>
  26 #include <rte_lcore.h>
  27 #include <rte_log.h>
  28
  29 #include "eal_private.h"
  30 #include "eal_filesystem.h"
  31 #include "eal_internal_cfg.h"
  32
  33 static int mp_fd = -1;
  34 static char mp_filter[PATH_MAX];   /* Filter for secondary process sockets */
  35 static char mp_dir_path[PATH_MAX]; /* The directory path for all mp sockets */
  36 static pthread_mutex_t mp_mutex_action = PTHREAD_MUTEX_INITIALIZER;
  37
  38 struct action_entry {
  39         TAILQ_ENTRY(action_entry) next;
  40         char action_name[RTE_MP_MAX_NAME_LEN];
  41         rte_mp_t action;
  42 };
  43
  44 /** Double linked list of actions. */
  45 TAILQ_HEAD(action_entry_list, action_entry);
  46
  47 static struct action_entry_list action_entry_list =
  48         TAILQ_HEAD_INITIALIZER(action_entry_list);
  49
  50 enum mp_type {
  51         MP_MSG, /* Share message with peers, will not block */
  52         MP_REQ, /* Request for information, Will block for a reply */
  53         MP_REP, /* Response to previously-received request */
  54 };
  55
  56 struct mp_msg_internal {
  57         int type;
  58         struct rte_mp_msg msg;
  59 };
  60
  61 struct sync_request {
  62         TAILQ_ENTRY(sync_request) next;
  63         int reply_received;
  64         char dst[PATH_MAX];
  65         struct rte_mp_msg *request;
  66         struct rte_mp_msg *reply;
  67         pthread_cond_t cond;
  68 };
  69
  70 TAILQ_HEAD(sync_request_list, sync_request);
  71
  72 static struct {
  73         struct sync_request_list requests;
  74         pthread_mutex_t lock;
  75 } sync_requests = {
  76         .requests = TAILQ_HEAD_INITIALIZER(sync_requests.requests),
  77         .lock = PTHREAD_MUTEX_INITIALIZER
  78 };
  79
  80 static struct sync_request *
  81 find_sync_request(const char *dst, const char *act_name)
  82 {
  83         struct sync_request *r;
  84
  85         TAILQ_FOREACH(r, &sync_requests.requests, next) {
  86                 if (!strcmp(r->dst, dst) &&
  87                     !strcmp(r->request->name, act_name))
  88                         break;
  89         }
  90
  91         return r;
  92 }
  93
  94 int
  95 rte_eal_primary_proc_alive(const char *config_file_path)
  96 {
  97         int config_fd;
  98
  99         if (config_file_path)
 100                 config_fd = open(config_file_path, O_RDONLY);
 101         else {
 102                 const char *path;
 103
 104                 path = eal_runtime_config_path();
 105                 config_fd = open(path, O_RDONLY);
 106         }
 107         if (config_fd < 0)
 108                 return 0;
 109
 110         int ret = lockf(config_fd, F_TEST, 0);
 111         close(config_fd);
 112
 113         return !!ret;
 114 }
 115
 116 static struct action_entry *
 117 find_action_entry_by_name(const char *name)
 118 {
 119         struct action_entry *entry;
 120
 121         TAILQ_FOREACH(entry, &action_entry_list, next) {
 122                 if (strncmp(entry->action_name, name, RTE_MP_MAX_NAME_LEN) == 0)
 123                         break;
 124         }
 125
 126         return entry;
 127 }
 128
 129 static int
 130 validate_action_name(const char *name)
 131 {
 132         if (name == NULL) {
 133                 RTE_LOG(ERR, EAL, "Action name cannot be NULL\n");
 134                 rte_errno = EINVAL;
 135                 return -1;
 136         }
 137         if (strnlen(name, RTE_MP_MAX_NAME_LEN) == 0) {
 138                 RTE_LOG(ERR, EAL, "Length of action name is zero\n");
 139                 rte_errno = EINVAL;
 140                 return -1;
 141         }
 142         if (strnlen(name, RTE_MP_MAX_NAME_LEN) == RTE_MP_MAX_NAME_LEN) {
 143                 rte_errno = E2BIG;
 144                 return -1;
 145         }
 146         return 0;
 147 }
 148
 149 int __rte_experimental
 150 rte_mp_action_register(const char *name, rte_mp_t action)
 151 {
 152         struct action_entry *entry;
 153
 154         if (validate_action_name(name))
 155                 return -1;
 156
 157         entry = malloc(sizeof(struct action_entry));
 158         if (entry == NULL) {
 159                 rte_errno = ENOMEM;
 160                 return -1;
 161         }
 162         strcpy(entry->action_name, name);
 163         entry->action = action;
 164
 165         pthread_mutex_lock(&mp_mutex_action);
 166         if (find_action_entry_by_name(name) != NULL) {
 167                 pthread_mutex_unlock(&mp_mutex_action);
 168                 rte_errno = EEXIST;
 169                 free(entry);
 170                 return -1;
 171         }
 172         TAILQ_INSERT_TAIL(&action_entry_list, entry, next);
 173         pthread_mutex_unlock(&mp_mutex_action);
 174         return 0;
 175 }
 176
 177 void __rte_experimental
 178 rte_mp_action_unregister(const char *name)
 179 {
 180         struct action_entry *entry;
 181
 182         if (validate_action_name(name))
 183                 return;
 184
 185         pthread_mutex_lock(&mp_mutex_action);
 186         entry = find_action_entry_by_name(name);
 187         if (entry == NULL) {
 188                 pthread_mutex_unlock(&mp_mutex_action);
 189                 return;
 190         }
 191         TAILQ_REMOVE(&action_entry_list, entry, next);
 192         pthread_mutex_unlock(&mp_mutex_action);
 193         free(entry);
 194 }
 195
 196 static int
 197 read_msg(struct mp_msg_internal *m, struct sockaddr_un *s)
 198 {
 199         int msglen;
 200         struct iovec iov;
 201         struct msghdr msgh;
 202         char control[CMSG_SPACE(sizeof(m->msg.fds))];
 203         struct cmsghdr *cmsg;
 204         int buflen = sizeof(*m) - sizeof(m->msg.fds);
 205
 206         memset(&msgh, 0, sizeof(msgh));
 207         iov.iov_base = m;
 208         iov.iov_len  = buflen;
 209
 210         msgh.msg_name = s;
 211         msgh.msg_namelen = sizeof(*s);
 212         msgh.msg_iov = &iov;
 213         msgh.msg_iovlen = 1;
 214         msgh.msg_control = control;
 215         msgh.msg_controllen = sizeof(control);
 216
 217         msglen = recvmsg(mp_fd, &msgh, 0);
 218         if (msglen < 0) {
 219                 RTE_LOG(ERR, EAL, "recvmsg failed, %s\n", strerror(errno));
 220                 return -1;
 221         }
 222
 223         if (msglen != buflen || (msgh.msg_flags & (MSG_TRUNC | MSG_CTRUNC))) {
 224                 RTE_LOG(ERR, EAL, "truncted msg\n");
 225                 return -1;
 226         }
 227
 228         /* read auxiliary FDs if any */
 229         for (cmsg = CMSG_FIRSTHDR(&msgh); cmsg != NULL;
 230                 cmsg = CMSG_NXTHDR(&msgh, cmsg)) {
 231                 if ((cmsg->cmsg_level == SOL_SOCKET) &&
 232                         (cmsg->cmsg_type == SCM_RIGHTS)) {
 233                         memcpy(m->msg.fds, CMSG_DATA(cmsg), sizeof(m->msg.fds));
 234                         break;
 235                 }
 236         }
 237
 238         return 0;
 239 }
 240
 241 static void
 242 process_msg(struct mp_msg_internal *m, struct sockaddr_un *s)
 243 {
 244         struct sync_request *sync_req;
 245         struct action_entry *entry;
 246         struct rte_mp_msg *msg = &m->msg;
 247         rte_mp_t action = NULL;
 248
 249         RTE_LOG(DEBUG, EAL, "msg: %s\n", msg->name);
 250
 251         if (m->type == MP_REP) {
 252                 pthread_mutex_lock(&sync_requests.lock);
 253                 sync_req = find_sync_request(s->sun_path, msg->name);
 254                 if (sync_req) {
 255                         memcpy(sync_req->reply, msg, sizeof(*msg));
 256                         sync_req->reply_received = 1;
 257                         pthread_cond_signal(&sync_req->cond);
 258                 } else
 259                         RTE_LOG(ERR, EAL, "Drop mp reply: %s\n", msg->name);
 260                 pthread_mutex_unlock(&sync_requests.lock);
 261                 return;
 262         }
 263
 264         pthread_mutex_lock(&mp_mutex_action);
 265         entry = find_action_entry_by_name(msg->name);
 266         if (entry != NULL)
 267                 action = entry->action;
 268         pthread_mutex_unlock(&mp_mutex_action);
 269
 270         if (!action)
 271                 RTE_LOG(ERR, EAL, "Cannot find action: %s\n", msg->name);
 272         else if (action(msg, s->sun_path) < 0)
 273                 RTE_LOG(ERR, EAL, "Fail to handle message: %s\n", msg->name);
 274 }
 275
 276 static void *
 277 mp_handle(void *arg __rte_unused)
 278 {
 279         struct mp_msg_internal msg;
 280         struct sockaddr_un sa;
 281
 282         while (1) {
 283                 if (read_msg(&msg, &sa) == 0)
 284                         process_msg(&msg, &sa);
 285         }
 286
 287         return NULL;
 288 }
 289
 290 static int
 291 open_socket_fd(void)
 292 {
 293         struct sockaddr_un un;
 294         const char *prefix = eal_mp_socket_path();
 295
 296         mp_fd = socket(AF_UNIX, SOCK_DGRAM, 0);
 297         if (mp_fd < 0) {
 298                 RTE_LOG(ERR, EAL, "failed to create unix socket\n");
 299                 return -1;
 300         }
 301
 302         memset(&un, 0, sizeof(un));
 303         un.sun_family = AF_UNIX;
 304         if (rte_eal_process_type() == RTE_PROC_PRIMARY)
 305                 snprintf(un.sun_path, sizeof(un.sun_path), "%s", prefix);
 306         else {
 307                 snprintf(un.sun_path, sizeof(un.sun_path), "%s_%d_%"PRIx64,
 308                          prefix, getpid(), rte_rdtsc());
 309         }
 310         unlink(un.sun_path); /* May still exist since last run */
 311         if (bind(mp_fd, (struct sockaddr *)&un, sizeof(un)) < 0) {
 312                 RTE_LOG(ERR, EAL, "failed to bind %s: %s\n",
 313                         un.sun_path, strerror(errno));
 314                 close(mp_fd);
 315                 return -1;
 316         }
 317
 318         RTE_LOG(INFO, EAL, "Multi-process socket %s\n", un.sun_path);
 319         return mp_fd;
 320 }
 321
 322 static int
 323 unlink_sockets(const char *filter)
 324 {
 325         int dir_fd;
 326         DIR *mp_dir;
 327         struct dirent *ent;
 328
 329         mp_dir = opendir(mp_dir_path);
 330         if (!mp_dir) {
 331                 RTE_LOG(ERR, EAL, "Unable to open directory %s\n", mp_dir_path);
 332                 return -1;
 333         }
 334         dir_fd = dirfd(mp_dir);
 335
 336         while ((ent = readdir(mp_dir))) {
 337                 if (fnmatch(filter, ent->d_name, 0) == 0)
 338                         unlinkat(dir_fd, ent->d_name, 0);
 339         }
 340
 341         closedir(mp_dir);
 342         return 0;
 343 }
 344
 345 static void
 346 unlink_socket_by_path(const char *path)
 347 {
 348         char *filename;
 349         char *fullpath = strdup(path);
 350
 351         if (!fullpath)
 352                 return;
 353         filename = basename(fullpath);
 354         unlink_sockets(filename);
 355         free(fullpath);
 356         RTE_LOG(INFO, EAL, "Remove socket %s\n", path);
 357 }
 358
 359 int
 360 rte_mp_channel_init(void)
 361 {
 362         char thread_name[RTE_MAX_THREAD_NAME_LEN];
 363         char *path;
 364         pthread_t tid;
 365
 366         snprintf(mp_filter, PATH_MAX, ".%s_unix_*",
 367                  internal_config.hugefile_prefix);
 368
 369         path = strdup(eal_mp_socket_path());
 370         snprintf(mp_dir_path, PATH_MAX, "%s", dirname(path));
 371         free(path);
 372
 373         if (rte_eal_process_type() == RTE_PROC_PRIMARY &&
 374             unlink_sockets(mp_filter)) {
 375                 RTE_LOG(ERR, EAL, "failed to unlink mp sockets\n");
 376                 return -1;
 377         }
 378
 379         if (open_socket_fd() < 0)
 380                 return -1;
 381
 382         if (pthread_create(&tid, NULL, mp_handle, NULL) < 0) {
 383                 RTE_LOG(ERR, EAL, "failed to create mp thead: %s\n",
 384                         strerror(errno));
 385                 close(mp_fd);
 386                 mp_fd = -1;
 387                 return -1;
 388         }
 389
 390         /* try best to set thread name */
 391         snprintf(thread_name, RTE_MAX_THREAD_NAME_LEN, "rte_mp_handle");
 392         rte_thread_setname(tid, thread_name);
 393         return 0;
 394 }
 395
 396 /**
 397  * Return -1, as fail to send message and it's caused by the local side.
 398  * Return 0, as fail to send message and it's caused by the remote side.
 399  * Return 1, as succeed to send message.
 400  *
 401  */
 402 static int
 403 send_msg(const char *dst_path, struct rte_mp_msg *msg, int type)
 404 {
 405         int snd;
 406         struct iovec iov;
 407         struct msghdr msgh;
 408         struct cmsghdr *cmsg;
 409         struct sockaddr_un dst;
 410         struct mp_msg_internal m;
 411         int fd_size = msg->num_fds * sizeof(int);
 412         char control[CMSG_SPACE(fd_size)];
 413
 414         m.type = type;
 415         memcpy(&m.msg, msg, sizeof(*msg));
 416
 417         memset(&dst, 0, sizeof(dst));
 418         dst.sun_family = AF_UNIX;
 419         snprintf(dst.sun_path, sizeof(dst.sun_path), "%s", dst_path);
 420
 421         memset(&msgh, 0, sizeof(msgh));
 422         memset(control, 0, sizeof(control));
 423
 424         iov.iov_base = &m;
 425         iov.iov_len = sizeof(m) - sizeof(msg->fds);
 426
 427         msgh.msg_name = &dst;
 428         msgh.msg_namelen = sizeof(dst);
 429         msgh.msg_iov = &iov;
 430         msgh.msg_iovlen = 1;
 431         msgh.msg_control = control;
 432         msgh.msg_controllen = sizeof(control);
 433
 434         cmsg = CMSG_FIRSTHDR(&msgh);
 435         cmsg->cmsg_len = CMSG_LEN(fd_size);
 436         cmsg->cmsg_level = SOL_SOCKET;
 437         cmsg->cmsg_type = SCM_RIGHTS;
 438         memcpy(CMSG_DATA(cmsg), msg->fds, fd_size);
 439
 440         do {
 441                 snd = sendmsg(mp_fd, &msgh, 0);
 442         } while (snd < 0 && errno == EINTR);
 443
 444         if (snd < 0) {
 445                 rte_errno = errno;
 446                 /* Check if it caused by peer process exits */
 447                 if (errno == -ECONNREFUSED) {
 448                         /* We don't unlink the primary's socket here */
 449                         if (rte_eal_process_type() == RTE_PROC_PRIMARY)
 450                                 unlink_socket_by_path(dst_path);
 451                         return 0;
 452                 }
 453                 if (errno == -ENOBUFS) {
 454                         RTE_LOG(ERR, EAL, "Peer cannot receive message %s\n",
 455                                 dst_path);
 456                         return 0;
 457                 }
 458                 RTE_LOG(ERR, EAL, "failed to send to (%s) due to %s\n",
 459                         dst_path, strerror(errno));
 460                 return -1;
 461         }
 462
 463         return 1;
 464 }
 465
 466 static int
 467 mp_send(struct rte_mp_msg *msg, const char *peer, int type)
 468 {
 469         int ret = 0;
 470         DIR *mp_dir;
 471         struct dirent *ent;
 472
 473         if (!peer && (rte_eal_process_type() == RTE_PROC_SECONDARY))
 474                 peer = eal_mp_socket_path();
 475
 476         if (peer) {
 477                 if (send_msg(peer, msg, type) < 0)
 478                         return -1;
 479                 else
 480                         return 0;
 481         }
 482
 483         /* broadcast to all secondary processes */
 484         mp_dir = opendir(mp_dir_path);
 485         if (!mp_dir) {
 486                 RTE_LOG(ERR, EAL, "Unable to open directory %s\n",
 487                                 mp_dir_path);
 488                 rte_errno = errno;
 489                 return -1;
 490         }
 491         while ((ent = readdir(mp_dir))) {
 492                 char path[PATH_MAX];
 493
 494                 if (fnmatch(mp_filter, ent->d_name, 0) != 0)
 495                         continue;
 496
 497                 snprintf(path, sizeof(path), "%s/%s", mp_dir_path,
 498                          ent->d_name);
 499                 if (send_msg(path, msg, type) < 0)
 500                         ret = -1;
 501         }
 502
 503         closedir(mp_dir);
 504         return ret;
 505 }
 506
 507 static bool
 508 check_input(const struct rte_mp_msg *msg)
 509 {
 510         if (msg == NULL) {
 511                 RTE_LOG(ERR, EAL, "Msg cannot be NULL\n");
 512                 rte_errno = EINVAL;
 513                 return false;
 514         }
 515
 516         if (validate_action_name(msg->name))
 517                 return false;
 518
 519         if (msg->len_param > RTE_MP_MAX_PARAM_LEN) {
 520                 RTE_LOG(ERR, EAL, "Message data is too long\n");
 521                 rte_errno = E2BIG;
 522                 return false;
 523         }
 524
 525         if (msg->num_fds > RTE_MP_MAX_FD_NUM) {
 526                 RTE_LOG(ERR, EAL, "Cannot send more than %d FDs\n",
 527                         RTE_MP_MAX_FD_NUM);
 528                 rte_errno = E2BIG;
 529                 return false;
 530         }
 531
 532         return true;
 533 }
 534
 535 int __rte_experimental
 536 rte_mp_sendmsg(struct rte_mp_msg *msg)
 537 {
 538         if (!check_input(msg))
 539                 return -1;
 540
 541         RTE_LOG(DEBUG, EAL, "sendmsg: %s\n", msg->name);
 542         return mp_send(msg, NULL, MP_MSG);
 543 }
 544
 545 static int
 546 mp_request_one(const char *dst, struct rte_mp_msg *req,
 547                struct rte_mp_reply *reply, const struct timespec *ts)
 548 {
 549         int ret;
 550         struct timeval now;
 551         struct rte_mp_msg msg, *tmp;
 552         struct sync_request sync_req, *exist;
 553
 554         sync_req.reply_received = 0;
 555         strcpy(sync_req.dst, dst);
 556         sync_req.request = req;
 557         sync_req.reply = &msg;
 558         pthread_cond_init(&sync_req.cond, NULL);
 559
 560         pthread_mutex_lock(&sync_requests.lock);
 561         exist = find_sync_request(dst, req->name);
 562         if (!exist)
 563                 TAILQ_INSERT_TAIL(&sync_requests.requests, &sync_req, next);
 564         pthread_mutex_unlock(&sync_requests.lock);
 565         if (exist) {
 566                 RTE_LOG(ERR, EAL, "A pending request %s:%s\n", dst, req->name);
 567                 rte_errno = EEXIST;
 568                 return -1;
 569         }
 570
 571         ret = send_msg(dst, req, MP_REQ);
 572         if (ret < 0) {
 573                 RTE_LOG(ERR, EAL, "Fail to send request %s:%s\n",
 574                         dst, req->name);
 575                 return -1;
 576         } else if (ret == 0)
 577                 return 0;
 578
 579         reply->nb_sent++;
 580
 581         pthread_mutex_lock(&sync_requests.lock);
 582         do {
 583                 pthread_cond_timedwait(&sync_req.cond, &sync_requests.lock, ts);
 584                 /* Check spurious wakeups */
 585                 if (sync_req.reply_received == 1)
 586                         break;
 587                 /* Check if time is out */
 588                 if (gettimeofday(&now, NULL) < 0)
 589                         break;
 590                 if (ts->tv_sec < now.tv_sec)
 591                         break;
 592                 else if (now.tv_sec == ts->tv_sec &&
 593                          now.tv_usec * 1000 < ts->tv_nsec)
 594                         break;
 595         } while (1);
 596         /* We got the lock now */
 597         TAILQ_REMOVE(&sync_requests.requests, &sync_req, next);
 598         pthread_mutex_unlock(&sync_requests.lock);
 599
 600         if (sync_req.reply_received == 0) {
 601                 RTE_LOG(ERR, EAL, "Fail to recv reply for request %s:%s\n",
 602                         dst, req->name);
 603                 rte_errno = ETIMEDOUT;
 604                 return -1;
 605         }
 606
 607         tmp = realloc(reply->msgs, sizeof(msg) * (reply->nb_received + 1));
 608         if (!tmp) {
 609                 RTE_LOG(ERR, EAL, "Fail to alloc reply for request %s:%s\n",
 610                         dst, req->name);
 611                 rte_errno = ENOMEM;
 612                 return -1;
 613         }
 614         memcpy(&tmp[reply->nb_received], &msg, sizeof(msg));
 615         reply->msgs = tmp;
 616         reply->nb_received++;
 617         return 0;
 618 }
 619
 620 int __rte_experimental
 621 rte_mp_request(struct rte_mp_msg *req, struct rte_mp_reply *reply,
 622                 const struct timespec *ts)
 623 {
 624         int ret = 0;
 625         DIR *mp_dir;
 626         struct dirent *ent;
 627         struct timeval now;
 628         struct timespec end;
 629
 630         RTE_LOG(DEBUG, EAL, "request: %s\n", req->name);
 631
 632         if (check_input(req) == false)
 633                 return -1;
 634         if (gettimeofday(&now, NULL) < 0) {
 635                 RTE_LOG(ERR, EAL, "Faile to get current time\n");
 636                 rte_errno = errno;
 637                 return -1;
 638         }
 639
 640         end.tv_nsec = (now.tv_usec * 1000 + ts->tv_nsec) % 1000000000;
 641         end.tv_sec = now.tv_sec + ts->tv_sec +
 642                         (now.tv_usec * 1000 + ts->tv_nsec) / 1000000000;
 643
 644         reply->nb_sent = 0;
 645         reply->nb_received = 0;
 646         reply->msgs = NULL;
 647
 648         /* for secondary process, send request to the primary process only */
 649         if (rte_eal_process_type() == RTE_PROC_SECONDARY)
 650                 return mp_request_one(eal_mp_socket_path(), req, reply, &end);
 651
 652         /* for primary process, broadcast request, and collect reply 1 by 1 */
 653         mp_dir = opendir(mp_dir_path);
 654         if (!mp_dir) {
 655                 RTE_LOG(ERR, EAL, "Unable to open directory %s\n", mp_dir_path);
 656                 rte_errno = errno;
 657                 return -1;
 658         }
 659
 660         while ((ent = readdir(mp_dir))) {
 661                 char path[PATH_MAX];
 662
 663                 if (fnmatch(mp_filter, ent->d_name, 0) != 0)
 664                         continue;
 665
 666                 snprintf(path, sizeof(path), "%s/%s", mp_dir_path,
 667                          ent->d_name);
 668
 669                 if (mp_request_one(path, req, reply, &end))
 670                         ret = -1;
 671         }
 672
 673         closedir(mp_dir);
 674         return ret;
 675 }
 676
 677 int __rte_experimental
 678 rte_mp_reply(struct rte_mp_msg *msg, const char *peer)
 679 {
 680
 681         RTE_LOG(DEBUG, EAL, "reply: %s\n", msg->name);
 682
 683         if (check_input(msg) == false)
 684                 return -1;
 685
 686         if (peer == NULL) {
 687                 RTE_LOG(ERR, EAL, "peer is not specified\n");
 688                 rte_errno = EINVAL;
 689                 return -1;
 690         }
 691
 692         return mp_send(msg, peer, MP_REP);
 693 }