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