lib/librte_eal/common/malloc_mp.c

   1 /* SPDX-License-Identifier: BSD-3-Clause
   2  * Copyright(c) 2018 Intel Corporation
   3  */
   4
   5 #include <string.h>
   6 #include <sys/time.h>
   7
   8 #include <rte_alarm.h>
   9 #include <rte_errno.h>
  10 #include <rte_string_fns.h>
  11
  12 #include "eal_memalloc.h"
  13 #include "eal_memcfg.h"
  14
  15 #include "malloc_elem.h"
  16 #include "malloc_mp.h"
  17
  18 #define MP_ACTION_SYNC "mp_malloc_sync"
  19 /**< request sent by primary process to notify of changes in memory map */
  20 #define MP_ACTION_ROLLBACK "mp_malloc_rollback"
  21 /**< request sent by primary process to notify of changes in memory map. this is
  22  * essentially a regular sync request, but we cannot send sync requests while
  23  * another one is in progress, and we might have to - therefore, we do this as
  24  * a separate callback.
  25  */
  26 #define MP_ACTION_REQUEST "mp_malloc_request"
  27 /**< request sent by secondary process to ask for allocation/deallocation */
  28 #define MP_ACTION_RESPONSE "mp_malloc_response"
  29 /**< response sent to secondary process to indicate result of request */
  30
  31 /* forward declarations */
  32 static int
  33 handle_sync_response(const struct rte_mp_msg *request,
  34                 const struct rte_mp_reply *reply);
  35 static int
  36 handle_rollback_response(const struct rte_mp_msg *request,
  37                 const struct rte_mp_reply *reply);
  38
  39 #define MP_TIMEOUT_S 5 /**< 5 seconds timeouts */
  40
  41 /* when we're allocating, we need to store some state to ensure that we can
  42  * roll back later
  43  */
  44 struct primary_alloc_req_state {
  45         struct malloc_heap *heap;
  46         struct rte_memseg **ms;
  47         int ms_len;
  48         struct malloc_elem *elem;
  49         void *map_addr;
  50         size_t map_len;
  51 };
  52
  53 enum req_state {
  54         REQ_STATE_INACTIVE = 0,
  55         REQ_STATE_ACTIVE,
  56         REQ_STATE_COMPLETE
  57 };
  58
  59 struct mp_request {
  60         TAILQ_ENTRY(mp_request) next;
  61         struct malloc_mp_req user_req; /**< contents of request */
  62         pthread_cond_t cond; /**< variable we use to time out on this request */
  63         enum req_state state; /**< indicate status of this request */
  64         struct primary_alloc_req_state alloc_state;
  65 };
  66
  67 /*
  68  * We could've used just a single request, but it may be possible for
  69  * secondaries to timeout earlier than the primary, and send a new request while
  70  * primary is still expecting replies to the old one. Therefore, each new
  71  * request will get assigned a new ID, which is how we will distinguish between
  72  * expected and unexpected messages.
  73  */
  74 TAILQ_HEAD(mp_request_list, mp_request);
  75 static struct {
  76         struct mp_request_list list;
  77         pthread_mutex_t lock;
  78 } mp_request_list = {
  79         .list = TAILQ_HEAD_INITIALIZER(mp_request_list.list),
  80         .lock = PTHREAD_MUTEX_INITIALIZER
  81 };
  82
  83 /**
  84  * General workflow is the following:
  85  *
  86  * Allocation:
  87  * S: send request to primary
  88  * P: attempt to allocate memory
  89  *    if failed, sendmsg failure
  90  *    if success, send sync request
  91  * S: if received msg of failure, quit
  92  *    if received sync request, synchronize memory map and reply with result
  93  * P: if received sync request result
  94  *    if success, sendmsg success
  95  *    if failure, roll back allocation and send a rollback request
  96  * S: if received msg of success, quit
  97  *    if received rollback request, synchronize memory map and reply with result
  98  * P: if received sync request result
  99  *    sendmsg sync request result
 100  * S: if received msg, quit
 101  *
 102  * Aside from timeouts, there are three points where we can quit:
 103  *  - if allocation failed straight away
 104  *  - if allocation and sync request succeeded
 105  *  - if allocation succeeded, sync request failed, allocation rolled back and
 106  *    rollback request received (irrespective of whether it succeeded or failed)
 107  *
 108  * Deallocation:
 109  * S: send request to primary
 110  * P: attempt to deallocate memory
 111  *    if failed, sendmsg failure
 112  *    if success, send sync request
 113  * S: if received msg of failure, quit
 114  *    if received sync request, synchronize memory map and reply with result
 115  * P: if received sync request result
 116  *    sendmsg sync request result
 117  * S: if received msg, quit
 118  *
 119  * There is no "rollback" from deallocation, as it's safe to have some memory
 120  * mapped in some processes - it's absent from the heap, so it won't get used.
 121  */
 122
 123 static struct mp_request *
 124 find_request_by_id(uint64_t id)
 125 {
 126         struct mp_request *req;
 127         TAILQ_FOREACH(req, &mp_request_list.list, next) {
 128                 if (req->user_req.id == id)
 129                         break;
 130         }
 131         return req;
 132 }
 133
 134 /* this ID is, like, totally guaranteed to be absolutely unique. pinky swear. */
 135 static uint64_t
 136 get_unique_id(void)
 137 {
 138         uint64_t id;
 139         do {
 140                 id = rte_rand();
 141         } while (find_request_by_id(id) != NULL);
 142         return id;
 143 }
 144
 145 /* secondary will respond to sync requests thusly */
 146 static int
 147 handle_sync(const struct rte_mp_msg *msg, const void *peer)
 148 {
 149         struct rte_mp_msg reply;
 150         const struct malloc_mp_req *req =
 151                         (const struct malloc_mp_req *)msg->param;
 152         struct malloc_mp_req *resp =
 153                         (struct malloc_mp_req *)reply.param;
 154         int ret;
 155
 156         if (req->t != REQ_TYPE_SYNC) {
 157                 RTE_LOG(ERR, EAL, "Unexpected request from primary\n");
 158                 return -1;
 159         }
 160
 161         memset(&reply, 0, sizeof(reply));
 162
 163         reply.num_fds = 0;
 164         strlcpy(reply.name, msg->name, sizeof(reply.name));
 165         reply.len_param = sizeof(*resp);
 166
 167         ret = eal_memalloc_sync_with_primary();
 168
 169         resp->t = REQ_TYPE_SYNC;
 170         resp->id = req->id;
 171         resp->result = ret == 0 ? REQ_RESULT_SUCCESS : REQ_RESULT_FAIL;
 172
 173         rte_mp_reply(&reply, peer);
 174
 175         return 0;
 176 }
 177
 178 static int
 179 handle_alloc_request(const struct malloc_mp_req *m,
 180                 struct mp_request *req)
 181 {
 182         const struct malloc_req_alloc *ar = &m->alloc_req;
 183         struct malloc_heap *heap;
 184         struct malloc_elem *elem;
 185         struct rte_memseg **ms;
 186         size_t alloc_sz;
 187         int n_segs;
 188         void *map_addr;
 189
 190         alloc_sz = RTE_ALIGN_CEIL(ar->align + ar->elt_size +
 191                         MALLOC_ELEM_TRAILER_LEN, ar->page_sz);
 192         n_segs = alloc_sz / ar->page_sz;
 193
 194         heap = ar->heap;
 195
 196         /* we can't know in advance how many pages we'll need, so we malloc */
 197         ms = malloc(sizeof(*ms) * n_segs);
 198         if (ms == NULL) {
 199                 RTE_LOG(ERR, EAL, "Couldn't allocate memory for request state\n");
 200                 goto fail;
 201         }
 202         memset(ms, 0, sizeof(*ms) * n_segs);
 203
 204         elem = alloc_pages_on_heap(heap, ar->page_sz, ar->elt_size, ar->socket,
 205                         ar->flags, ar->align, ar->bound, ar->contig, ms,
 206                         n_segs);
 207
 208         if (elem == NULL)
 209                 goto fail;
 210
 211         map_addr = ms[0]->addr;
 212
 213         eal_memalloc_mem_event_notify(RTE_MEM_EVENT_ALLOC, map_addr, alloc_sz);
 214
 215         /* we have succeeded in allocating memory, but we still need to sync
 216          * with other processes. however, since DPDK IPC is single-threaded, we
 217          * send an asynchronous request and exit this callback.
 218          */
 219
 220         req->alloc_state.ms = ms;
 221         req->alloc_state.ms_len = n_segs;
 222         req->alloc_state.map_addr = map_addr;
 223         req->alloc_state.map_len = alloc_sz;
 224         req->alloc_state.elem = elem;
 225         req->alloc_state.heap = heap;
 226
 227         return 0;
 228 fail:
 229         free(ms);
 230         return -1;
 231 }
 232
 233 /* first stage of primary handling requests from secondary */
 234 static int
 235 handle_request(const struct rte_mp_msg *msg, const void *peer __rte_unused)
 236 {
 237         const struct malloc_mp_req *m =
 238                         (const struct malloc_mp_req *)msg->param;
 239         struct mp_request *entry;
 240         int ret;
 241
 242         /* lock access to request */
 243         pthread_mutex_lock(&mp_request_list.lock);
 244
 245         /* make sure it's not a dupe */
 246         entry = find_request_by_id(m->id);
 247         if (entry != NULL) {
 248                 RTE_LOG(ERR, EAL, "Duplicate request id\n");
 249                 goto fail;
 250         }
 251
 252         entry = malloc(sizeof(*entry));
 253         if (entry == NULL) {
 254                 RTE_LOG(ERR, EAL, "Unable to allocate memory for request\n");
 255                 goto fail;
 256         }
 257
 258         /* erase all data */
 259         memset(entry, 0, sizeof(*entry));
 260
 261         if (m->t == REQ_TYPE_ALLOC) {
 262                 ret = handle_alloc_request(m, entry);
 263         } else if (m->t == REQ_TYPE_FREE) {
 264                 eal_memalloc_mem_event_notify(RTE_MEM_EVENT_FREE,
 265                                 m->free_req.addr, m->free_req.len);
 266
 267                 ret = malloc_heap_free_pages(m->free_req.addr,
 268                                 m->free_req.len);
 269         } else {
 270                 RTE_LOG(ERR, EAL, "Unexpected request from secondary\n");
 271                 goto fail;
 272         }
 273
 274         if (ret != 0) {
 275                 struct rte_mp_msg resp_msg;
 276                 struct malloc_mp_req *resp =
 277                                 (struct malloc_mp_req *)resp_msg.param;
 278
 279                 /* send failure message straight away */
 280                 resp_msg.num_fds = 0;
 281                 resp_msg.len_param = sizeof(*resp);
 282                 strlcpy(resp_msg.name, MP_ACTION_RESPONSE,
 283                                 sizeof(resp_msg.name));
 284
 285                 resp->t = m->t;
 286                 resp->result = REQ_RESULT_FAIL;
 287                 resp->id = m->id;
 288
 289                 if (rte_mp_sendmsg(&resp_msg)) {
 290                         RTE_LOG(ERR, EAL, "Couldn't send response\n");
 291                         goto fail;
 292                 }
 293                 /* we did not modify the request */
 294                 free(entry);
 295         } else {
 296                 struct rte_mp_msg sr_msg;
 297                 struct malloc_mp_req *sr =
 298                                 (struct malloc_mp_req *)sr_msg.param;
 299                 struct timespec ts;
 300
 301                 memset(&sr_msg, 0, sizeof(sr_msg));
 302
 303                 /* we can do something, so send sync request asynchronously */
 304                 sr_msg.num_fds = 0;
 305                 sr_msg.len_param = sizeof(*sr);
 306                 strlcpy(sr_msg.name, MP_ACTION_SYNC, sizeof(sr_msg.name));
 307
 308                 ts.tv_nsec = 0;
 309                 ts.tv_sec = MP_TIMEOUT_S;
 310
 311                 /* sync requests carry no data */
 312                 sr->t = REQ_TYPE_SYNC;
 313                 sr->id = m->id;
 314
 315                 /* there may be stray timeout still waiting */
 316                 do {
 317                         ret = rte_mp_request_async(&sr_msg, &ts,
 318                                         handle_sync_response);
 319                 } while (ret != 0 && rte_errno == EEXIST);
 320                 if (ret != 0) {
 321                         RTE_LOG(ERR, EAL, "Couldn't send sync request\n");
 322                         if (m->t == REQ_TYPE_ALLOC)
 323                                 free(entry->alloc_state.ms);
 324                         goto fail;
 325                 }
 326
 327                 /* mark request as in progress */
 328                 memcpy(&entry->user_req, m, sizeof(*m));
 329                 entry->state = REQ_STATE_ACTIVE;
 330
 331                 TAILQ_INSERT_TAIL(&mp_request_list.list, entry, next);
 332         }
 333         pthread_mutex_unlock(&mp_request_list.lock);
 334         return 0;
 335 fail:
 336         pthread_mutex_unlock(&mp_request_list.lock);
 337         free(entry);
 338         return -1;
 339 }
 340
 341 /* callback for asynchronous sync requests for primary. this will either do a
 342  * sendmsg with results, or trigger rollback request.
 343  */
 344 static int
 345 handle_sync_response(const struct rte_mp_msg *request,
 346                 const struct rte_mp_reply *reply)
 347 {
 348         enum malloc_req_result result;
 349         struct mp_request *entry;
 350         const struct malloc_mp_req *mpreq =
 351                         (const struct malloc_mp_req *)request->param;
 352         int i;
 353
 354         /* lock the request */
 355         pthread_mutex_lock(&mp_request_list.lock);
 356
 357         entry = find_request_by_id(mpreq->id);
 358         if (entry == NULL) {
 359                 RTE_LOG(ERR, EAL, "Wrong request ID\n");
 360                 goto fail;
 361         }
 362
 363         result = REQ_RESULT_SUCCESS;
 364
 365         if (reply->nb_received != reply->nb_sent)
 366                 result = REQ_RESULT_FAIL;
 367
 368         for (i = 0; i < reply->nb_received; i++) {
 369                 struct malloc_mp_req *resp =
 370                                 (struct malloc_mp_req *)reply->msgs[i].param;
 371
 372                 if (resp->t != REQ_TYPE_SYNC) {
 373                         RTE_LOG(ERR, EAL, "Unexpected response to sync request\n");
 374                         result = REQ_RESULT_FAIL;
 375                         break;
 376                 }
 377                 if (resp->id != entry->user_req.id) {
 378                         RTE_LOG(ERR, EAL, "Response to wrong sync request\n");
 379                         result = REQ_RESULT_FAIL;
 380                         break;
 381                 }
 382                 if (resp->result == REQ_RESULT_FAIL) {
 383                         result = REQ_RESULT_FAIL;
 384                         break;
 385                 }
 386         }
 387
 388         if (entry->user_req.t == REQ_TYPE_FREE) {
 389                 struct rte_mp_msg msg;
 390                 struct malloc_mp_req *resp = (struct malloc_mp_req *)msg.param;
 391
 392                 memset(&msg, 0, sizeof(msg));
 393
 394                 /* this is a free request, just sendmsg result */
 395                 resp->t = REQ_TYPE_FREE;
 396                 resp->result = result;
 397                 resp->id = entry->user_req.id;
 398                 msg.num_fds = 0;
 399                 msg.len_param = sizeof(*resp);
 400                 strlcpy(msg.name, MP_ACTION_RESPONSE, sizeof(msg.name));
 401
 402                 if (rte_mp_sendmsg(&msg))
 403                         RTE_LOG(ERR, EAL, "Could not send message to secondary process\n");
 404
 405                 TAILQ_REMOVE(&mp_request_list.list, entry, next);
 406                 free(entry);
 407         } else if (entry->user_req.t == REQ_TYPE_ALLOC &&
 408                         result == REQ_RESULT_SUCCESS) {
 409                 struct malloc_heap *heap = entry->alloc_state.heap;
 410                 struct rte_mp_msg msg;
 411                 struct malloc_mp_req *resp =
 412                                 (struct malloc_mp_req *)msg.param;
 413
 414                 memset(&msg, 0, sizeof(msg));
 415
 416                 heap->total_size += entry->alloc_state.map_len;
 417
 418                 /* result is success, so just notify secondary about this */
 419                 resp->t = REQ_TYPE_ALLOC;
 420                 resp->result = result;
 421                 resp->id = entry->user_req.id;
 422                 msg.num_fds = 0;
 423                 msg.len_param = sizeof(*resp);
 424                 strlcpy(msg.name, MP_ACTION_RESPONSE, sizeof(msg.name));
 425
 426                 if (rte_mp_sendmsg(&msg))
 427                         RTE_LOG(ERR, EAL, "Could not send message to secondary process\n");
 428
 429                 TAILQ_REMOVE(&mp_request_list.list, entry, next);
 430                 free(entry->alloc_state.ms);
 431                 free(entry);
 432         } else if (entry->user_req.t == REQ_TYPE_ALLOC &&
 433                         result == REQ_RESULT_FAIL) {
 434                 struct rte_mp_msg rb_msg;
 435                 struct malloc_mp_req *rb =
 436                                 (struct malloc_mp_req *)rb_msg.param;
 437                 struct timespec ts;
 438                 struct primary_alloc_req_state *state =
 439                                 &entry->alloc_state;
 440                 int ret;
 441
 442                 memset(&rb_msg, 0, sizeof(rb_msg));
 443
 444                 /* we've failed to sync, so do a rollback */
 445                 eal_memalloc_mem_event_notify(RTE_MEM_EVENT_FREE,
 446                                 state->map_addr, state->map_len);
 447
 448                 rollback_expand_heap(state->ms, state->ms_len, state->elem,
 449                                 state->map_addr, state->map_len);
 450
 451                 /* send rollback request */
 452                 rb_msg.num_fds = 0;
 453                 rb_msg.len_param = sizeof(*rb);
 454                 strlcpy(rb_msg.name, MP_ACTION_ROLLBACK, sizeof(rb_msg.name));
 455
 456                 ts.tv_nsec = 0;
 457                 ts.tv_sec = MP_TIMEOUT_S;
 458
 459                 /* sync requests carry no data */
 460                 rb->t = REQ_TYPE_SYNC;
 461                 rb->id = entry->user_req.id;
 462
 463                 /* there may be stray timeout still waiting */
 464                 do {
 465                         ret = rte_mp_request_async(&rb_msg, &ts,
 466                                         handle_rollback_response);
 467                 } while (ret != 0 && rte_errno == EEXIST);
 468                 if (ret != 0) {
 469                         RTE_LOG(ERR, EAL, "Could not send rollback request to secondary process\n");
 470
 471                         /* we couldn't send rollback request, but that's OK -
 472                          * secondary will time out, and memory has been removed
 473                          * from heap anyway.
 474                          */
 475                         TAILQ_REMOVE(&mp_request_list.list, entry, next);
 476                         free(state->ms);
 477                         free(entry);
 478                         goto fail;
 479                 }
 480         } else {
 481                 RTE_LOG(ERR, EAL, " to sync request of unknown type\n");
 482                 goto fail;
 483         }
 484
 485         pthread_mutex_unlock(&mp_request_list.lock);
 486         return 0;
 487 fail:
 488         pthread_mutex_unlock(&mp_request_list.lock);
 489         return -1;
 490 }
 491
 492 static int
 493 handle_rollback_response(const struct rte_mp_msg *request,
 494                 const struct rte_mp_reply *reply __rte_unused)
 495 {
 496         struct rte_mp_msg msg;
 497         struct malloc_mp_req *resp = (struct malloc_mp_req *)msg.param;
 498         const struct malloc_mp_req *mpreq =
 499                         (const struct malloc_mp_req *)request->param;
 500         struct mp_request *entry;
 501
 502         /* lock the request */
 503         pthread_mutex_lock(&mp_request_list.lock);
 504
 505         memset(&msg, 0, sizeof(msg));
 506
 507         entry = find_request_by_id(mpreq->id);
 508         if (entry == NULL) {
 509                 RTE_LOG(ERR, EAL, "Wrong request ID\n");
 510                 goto fail;
 511         }
 512
 513         if (entry->user_req.t != REQ_TYPE_ALLOC) {
 514                 RTE_LOG(ERR, EAL, "Unexpected active request\n");
 515                 goto fail;
 516         }
 517
 518         /* we don't care if rollback succeeded, request still failed */
 519         resp->t = REQ_TYPE_ALLOC;
 520         resp->result = REQ_RESULT_FAIL;
 521         resp->id = mpreq->id;
 522         msg.num_fds = 0;
 523         msg.len_param = sizeof(*resp);
 524         strlcpy(msg.name, MP_ACTION_RESPONSE, sizeof(msg.name));
 525
 526         if (rte_mp_sendmsg(&msg))
 527                 RTE_LOG(ERR, EAL, "Could not send message to secondary process\n");
 528
 529         /* clean up */
 530         TAILQ_REMOVE(&mp_request_list.list, entry, next);
 531         free(entry->alloc_state.ms);
 532         free(entry);
 533
 534         pthread_mutex_unlock(&mp_request_list.lock);
 535         return 0;
 536 fail:
 537         pthread_mutex_unlock(&mp_request_list.lock);
 538         return -1;
 539 }
 540
 541 /* final stage of the request from secondary */
 542 static int
 543 handle_response(const struct rte_mp_msg *msg, const void *peer  __rte_unused)
 544 {
 545         const struct malloc_mp_req *m =
 546                         (const struct malloc_mp_req *)msg->param;
 547         struct mp_request *entry;
 548
 549         pthread_mutex_lock(&mp_request_list.lock);
 550
 551         entry = find_request_by_id(m->id);
 552         if (entry != NULL) {
 553                 /* update request status */
 554                 entry->user_req.result = m->result;
 555
 556                 entry->state = REQ_STATE_COMPLETE;
 557
 558                 /* trigger thread wakeup */
 559                 pthread_cond_signal(&entry->cond);
 560         }
 561
 562         pthread_mutex_unlock(&mp_request_list.lock);
 563
 564         return 0;
 565 }
 566
 567 /* synchronously request memory map sync, this is only called whenever primary
 568  * process initiates the allocation.
 569  */
 570 int
 571 request_sync(void)
 572 {
 573         struct rte_mp_msg msg;
 574         struct rte_mp_reply reply;
 575         struct malloc_mp_req *req = (struct malloc_mp_req *)msg.param;
 576         struct timespec ts;
 577         int i, ret = -1;
 578
 579         memset(&msg, 0, sizeof(msg));
 580         memset(&reply, 0, sizeof(reply));
 581
 582         /* no need to create tailq entries as this is entirely synchronous */
 583
 584         msg.num_fds = 0;
 585         msg.len_param = sizeof(*req);
 586         strlcpy(msg.name, MP_ACTION_SYNC, sizeof(msg.name));
 587
 588         /* sync request carries no data */
 589         req->t = REQ_TYPE_SYNC;
 590         req->id = get_unique_id();
 591
 592         ts.tv_nsec = 0;
 593         ts.tv_sec = MP_TIMEOUT_S;
 594
 595         /* there may be stray timeout still waiting */
 596         do {
 597                 ret = rte_mp_request_sync(&msg, &reply, &ts);
 598         } while (ret != 0 && rte_errno == EEXIST);
 599         if (ret != 0) {
 600                 /* if IPC is unsupported, behave as if the call succeeded */
 601                 if (rte_errno != ENOTSUP)
 602                         RTE_LOG(ERR, EAL, "Could not send sync request to secondary process\n");
 603                 else
 604                         ret = 0;
 605                 goto out;
 606         }
 607
 608         if (reply.nb_received != reply.nb_sent) {
 609                 RTE_LOG(ERR, EAL, "Not all secondaries have responded\n");
 610                 goto out;
 611         }
 612
 613         for (i = 0; i < reply.nb_received; i++) {
 614                 struct malloc_mp_req *resp =
 615                                 (struct malloc_mp_req *)reply.msgs[i].param;
 616                 if (resp->t != REQ_TYPE_SYNC) {
 617                         RTE_LOG(ERR, EAL, "Unexpected response from secondary\n");
 618                         goto out;
 619                 }
 620                 if (resp->id != req->id) {
 621                         RTE_LOG(ERR, EAL, "Wrong request ID\n");
 622                         goto out;
 623                 }
 624                 if (resp->result != REQ_RESULT_SUCCESS) {
 625                         RTE_LOG(ERR, EAL, "Secondary process failed to synchronize\n");
 626                         goto out;
 627                 }
 628         }
 629
 630         ret = 0;
 631 out:
 632         free(reply.msgs);
 633         return ret;
 634 }
 635
 636 /* this is a synchronous wrapper around a bunch of asynchronous requests to
 637  * primary process. this will initiate a request and wait until responses come.
 638  */
 639 int
 640 request_to_primary(struct malloc_mp_req *user_req)
 641 {
 642         struct rte_mp_msg msg;
 643         struct malloc_mp_req *msg_req = (struct malloc_mp_req *)msg.param;
 644         struct mp_request *entry;
 645         struct timespec ts;
 646         struct timeval now;
 647         int ret;
 648
 649         memset(&msg, 0, sizeof(msg));
 650         memset(&ts, 0, sizeof(ts));
 651
 652         pthread_mutex_lock(&mp_request_list.lock);
 653
 654         entry = malloc(sizeof(*entry));
 655         if (entry == NULL) {
 656                 RTE_LOG(ERR, EAL, "Cannot allocate memory for request\n");
 657                 goto fail;
 658         }
 659
 660         memset(entry, 0, sizeof(*entry));
 661
 662         if (gettimeofday(&now, NULL) < 0) {
 663                 RTE_LOG(ERR, EAL, "Cannot get current time\n");
 664                 goto fail;
 665         }
 666
 667         ts.tv_nsec = (now.tv_usec * 1000) % 1000000000;
 668         ts.tv_sec = now.tv_sec + MP_TIMEOUT_S +
 669                         (now.tv_usec * 1000) / 1000000000;
 670
 671         /* initialize the request */
 672         pthread_cond_init(&entry->cond, NULL);
 673
 674         msg.num_fds = 0;
 675         msg.len_param = sizeof(*msg_req);
 676         strlcpy(msg.name, MP_ACTION_REQUEST, sizeof(msg.name));
 677
 678         /* (attempt to) get a unique id */
 679         user_req->id = get_unique_id();
 680
 681         /* copy contents of user request into the message */
 682         memcpy(msg_req, user_req, sizeof(*msg_req));
 683
 684         if (rte_mp_sendmsg(&msg)) {
 685                 RTE_LOG(ERR, EAL, "Cannot send message to primary\n");
 686                 goto fail;
 687         }
 688
 689         /* copy contents of user request into active request */
 690         memcpy(&entry->user_req, user_req, sizeof(*user_req));
 691
 692         /* mark request as in progress */
 693         entry->state = REQ_STATE_ACTIVE;
 694
 695         TAILQ_INSERT_TAIL(&mp_request_list.list, entry, next);
 696
 697         /* finally, wait on timeout */
 698         do {
 699                 ret = pthread_cond_timedwait(&entry->cond,
 700                                 &mp_request_list.lock, &ts);
 701         } while (ret != 0 && ret != ETIMEDOUT);
 702
 703         if (entry->state != REQ_STATE_COMPLETE) {
 704                 RTE_LOG(ERR, EAL, "Request timed out\n");
 705                 ret = -1;
 706         } else {
 707                 ret = 0;
 708                 user_req->result = entry->user_req.result;
 709         }
 710         TAILQ_REMOVE(&mp_request_list.list, entry, next);
 711         free(entry);
 712
 713         pthread_mutex_unlock(&mp_request_list.lock);
 714         return ret;
 715 fail:
 716         pthread_mutex_unlock(&mp_request_list.lock);
 717         free(entry);
 718         return -1;
 719 }
 720
 721 int
 722 register_mp_requests(void)
 723 {
 724         if (rte_eal_process_type() == RTE_PROC_PRIMARY) {
 725                 /* it's OK for primary to not support IPC */
 726                 if (rte_mp_action_register(MP_ACTION_REQUEST, handle_request) &&
 727                                 rte_errno != ENOTSUP) {
 728                         RTE_LOG(ERR, EAL, "Couldn't register '%s' action\n",
 729                                 MP_ACTION_REQUEST);
 730                         return -1;
 731                 }
 732         } else {
 733                 if (rte_mp_action_register(MP_ACTION_SYNC, handle_sync)) {
 734                         RTE_LOG(ERR, EAL, "Couldn't register '%s' action\n",
 735                                 MP_ACTION_SYNC);
 736                         return -1;
 737                 }
 738                 if (rte_mp_action_register(MP_ACTION_ROLLBACK, handle_sync)) {
 739                         RTE_LOG(ERR, EAL, "Couldn't register '%s' action\n",
 740                                 MP_ACTION_SYNC);
 741                         return -1;
 742                 }
 743                 if (rte_mp_action_register(MP_ACTION_RESPONSE,
 744                                 handle_response)) {
 745                         RTE_LOG(ERR, EAL, "Couldn't register '%s' action\n",
 746                                 MP_ACTION_RESPONSE);
 747                         return -1;
 748                 }
 749         }
 750         return 0;
 751 }