linux/src/transport/TransportRpmsg.c

   1 /*
   2  * Copyright (c) 2014-2015 Texas Instruments Incorporated - http://www.ti.com
   3  * All rights reserved.
   4  *
   5  * Redistribution and use in source and binary forms, with or without
   6  * modification, are permitted provided that the following conditions
   7  * are met:
   8  *
   9  * *  Redistributions of source code must retain the above copyright
  10  *    notice, this list of conditions and the following disclaimer.
  11  *
  12  * *  Redistributions in binary form must reproduce the above copyright
  13  *    notice, this list of conditions and the following disclaimer in the
  14  *    documentation and/or other materials provided with the distribution.
  15  *
  16  * *  Neither the name of Texas Instruments Incorporated nor the names of
  17  *    its contributors may be used to endorse or promote products derived
  18  *    from this software without specific prior written permission.
  19  *
  20  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
  21  * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO,
  22  * THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
  23  * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR
  24  * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
  25  * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
  26  * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS;
  27  * OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
  28  * WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR
  29  * OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE,
  30  * EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  31  */
  32 /*
  33  *  ======== TransportRpmsg.c ========
  34  *  Implementation of functions specified in the IMessageQTransport interface.
  35  */
  36
  37 /* Socket Headers */
  38 #include <sys/socket.h>
  39 #include <sys/select.h>
  40 #include <sys/eventfd.h>
  41 #include <stdio.h>
  42 #include <stdlib.h>
  43 #include <unistd.h>
  44 #include <errno.h>
  45 #include <string.h>
  46 #include <fcntl.h>
  47 #include <pthread.h>
  48
  49 /* Socket Protocol Family */
  50 #include <net/rpmsg.h>
  51
  52
  53 /* IPC headers */
  54 #include <ti/ipc/Std.h>
  55 #include <SocketFxns.h>         /* Socket utils: */
  56 #include <ti/ipc/Ipc.h>
  57 #include <ti/ipc/MessageQ.h>
  58 #include <ti/ipc/MultiProc.h>
  59 #include <ti/ipc/transports/TransportRpmsg.h>
  60 #include <_MessageQ.h>
  61 #include <_lad.h>
  62
  63 /* More magic rpmsg port numbers: */
  64 #define MESSAGEQ_RPMSG_PORT       61
  65 #define MESSAGEQ_RPMSG_MAXSIZE   512
  66
  67 #define TransportRpmsg_GROWSIZE 32
  68
  69 /* traces in this file are controlled via _TransportMessageQ_verbose */
  70 Bool _TransportMessageQ_verbose = FALSE;
  71 #define verbose _TransportMessageQ_verbose
  72
  73 Int TransportRpmsg_bind(Void *handle, UInt32 queueId);
  74 Int TransportRpmsg_unbind(Void *handle, UInt32 queueId);
  75 Bool TransportRpmsg_put(Void *handle, Ptr msg);
  76
  77 typedef struct TransportRpmsg_Module {
  78     int             sock[MultiProc_MAXPROCESSORS];
  79     fd_set          rfds;
  80     int             maxFd;
  81     int             inFds[1024];
  82     int             nInFds;
  83     pthread_mutex_t gate;
  84     int             unblockEvent;    /* eventFd for unblocking socket */
  85     pthread_t       threadId;        /* ID returned by pthread_create() */
  86     Bool            threadStarted;
  87
  88     TransportRpmsg_Handle *inst;    /* array of instances */
  89 } TransportRpmsg_Module;
  90
  91 IMessageQTransport_Fxns TransportRpmsg_fxns = {
  92     .bind    = TransportRpmsg_bind,
  93     .unbind  = TransportRpmsg_unbind,
  94     .put     = TransportRpmsg_put
  95 };
  96
  97 typedef struct TransportRpmsg_Object {
  98     IMessageQTransport_Object base;
  99     Int status;
 100     UInt16 rprocId;
 101     int numQueues;
 102     int *qIndexToFd;
 103 } TransportRpmsg_Object;
 104
 105 TransportRpmsg_Module TransportRpmsg_state = {
 106     .sock = {0},
 107     .threadStarted = FALSE,
 108     .inst = NULL
 109 };
 110 TransportRpmsg_Module *TransportRpmsg_module = &TransportRpmsg_state;
 111
 112 static Int attach(UInt16 rprocId);
 113 static Int detach(UInt16 rprocId);
 114 static void *rpmsgThreadFxn(void *arg);
 115 static Int transportGet(int sock, MessageQ_Msg *retMsg);
 116 static Void bindFdToQueueIndex(TransportRpmsg_Object *obj,
 117                                Int fd,
 118                                UInt16 qIndex);
 119 static Void unbindQueueIndex(TransportRpmsg_Object *obj, UInt16 qIndex);
 120 static Int queueIndexToFd(TransportRpmsg_Object *obj, UInt16 qIndex);
 121
 122 Int TransportRpmsg_Factory_create(Void);
 123 Void TransportRpmsg_Factory_delete(Void);
 124
 125 Ipc_TransportFactoryFxns TransportRpmsg_Factory = {
 126     .createFxn = TransportRpmsg_Factory_create,
 127     .deleteFxn = TransportRpmsg_Factory_delete
 128 };
 129
 130 /* -------------------------------------------------------------------------- */
 131
 132 /* instance convertors */
 133 IMessageQTransport_Handle TransportRpmsg_upCast(TransportRpmsg_Handle handle)
 134 {
 135     TransportRpmsg_Object *obj = (TransportRpmsg_Object *)handle;
 136     return ((IMessageQTransport_Handle)&obj->base);
 137 }
 138
 139 TransportRpmsg_Handle TransportRpmsg_downCast(IMessageQTransport_Handle base)
 140 {
 141     return ((TransportRpmsg_Handle)base);
 142 }
 143
 144 TransportRpmsg_Handle TransportRpmsg_create(TransportRpmsg_Params *params,
 145                                             Int *attachStatus)
 146 {
 147     TransportRpmsg_Object *obj;
 148     Int rv;
 149
 150     rv = attach(params->rprocId);
 151     if (attachStatus) {
 152         *attachStatus = rv;
 153     }
 154
 155     if (rv != MessageQ_S_SUCCESS) {
 156         return NULL;
 157     }
 158
 159     obj = calloc(1, sizeof (TransportRpmsg_Object));
 160
 161     /* structure copy */
 162     obj->base.base.interfaceType = IMessageQTransport_TypeId;
 163     obj->base.fxns = &TransportRpmsg_fxns;
 164     obj->rprocId = params->rprocId;
 165
 166     obj->qIndexToFd = calloc(TransportRpmsg_GROWSIZE, sizeof (Int));
 167     obj->numQueues = TransportRpmsg_GROWSIZE;
 168
 169     return (TransportRpmsg_Handle)obj;
 170 }
 171
 172 Void TransportRpmsg_delete(TransportRpmsg_Handle *handlep)
 173 {
 174     TransportRpmsg_Object *obj = *(TransportRpmsg_Object **)handlep;
 175
 176     detach(obj->rprocId);
 177
 178     free(obj->qIndexToFd);
 179     free(obj);
 180
 181     *handlep = NULL;
 182 }
 183
 184 static Int attach(UInt16 rprocId)
 185 {
 186     Int     status = MessageQ_S_SUCCESS;
 187     int     sock;
 188     UInt16  clusterId;
 189
 190
 191     clusterId = rprocId - MultiProc_getBaseIdOfCluster();
 192
 193     /* Create the socket for sending messages to the remote proc: */
 194     sock = socket(AF_RPMSG, SOCK_SEQPACKET, 0);
 195     if (sock < 0) {
 196         status = MessageQ_E_FAIL;
 197         printf("attach: socket failed: %d (%s)\n",
 198                errno, strerror(errno));
 199
 200         goto exit;
 201     }
 202
 203     PRINTVERBOSE1("attach: created send socket: %d\n", sock)
 204
 205     /* Attempt to connect: */
 206     status = ConnectSocket(sock, rprocId, MESSAGEQ_RPMSG_PORT);
 207     if (status < 0) {
 208         /* is it ok to "borrow" this error code from MessageQ? */
 209         status = MessageQ_E_RESOURCE;
 210
 211         /* don't hard-printf or exit since this is no longer fatal */
 212         PRINTVERBOSE1("attach: ConnectSocket(rprocId:%d) failed\n", rprocId)
 213
 214         goto exitSock;
 215     }
 216
 217     TransportRpmsg_module->sock[clusterId] = sock;
 218
 219     if (TransportRpmsg_module->threadStarted == FALSE) {
 220         /* create a module wide event to unblock the socket select thread */
 221         TransportRpmsg_module->unblockEvent = eventfd(0, 0);
 222         if (TransportRpmsg_module->unblockEvent == -1) {
 223             printf("attach: unblock socket failed: %d (%s)\n",
 224                    errno, strerror(errno));
 225             status = MessageQ_E_FAIL;
 226
 227             goto exitSock;
 228         }
 229
 230         PRINTVERBOSE1("attach: created unblock event %d\n",
 231                       TransportRpmsg_module->unblockEvent)
 232
 233         FD_ZERO(&TransportRpmsg_module->rfds);
 234         FD_SET(TransportRpmsg_module->unblockEvent,
 235                &TransportRpmsg_module->rfds);
 236         TransportRpmsg_module->maxFd = TransportRpmsg_module->unblockEvent;
 237         TransportRpmsg_module->nInFds = 0;
 238
 239         pthread_mutex_init(&TransportRpmsg_module->gate, NULL);
 240
 241         status = pthread_create(&TransportRpmsg_module->threadId, NULL,
 242                                 &rpmsgThreadFxn, NULL);
 243         if (status < 0) {
 244             status = MessageQ_E_FAIL;
 245             printf("attach: failed to spawn thread\n");
 246
 247             goto exitEvent;
 248         }
 249         else {
 250             TransportRpmsg_module->threadStarted = TRUE;
 251         }
 252     }
 253
 254     goto exit;
 255
 256 exitEvent:
 257     close(TransportRpmsg_module->unblockEvent);
 258
 259     FD_ZERO(&TransportRpmsg_module->rfds);
 260     TransportRpmsg_module->maxFd = 0;
 261
 262 exitSock:
 263     close(sock);
 264     TransportRpmsg_module->sock[clusterId] = 0;
 265
 266 exit:
 267     return status;
 268 }
 269
 270 static Int detach(UInt16 rprocId)
 271 {
 272
 273     Int     status = -1;
 274     int     sock;
 275     UInt16  clusterId;
 276
 277     clusterId = rprocId - MultiProc_getBaseIdOfCluster();
 278     sock = TransportRpmsg_module->sock[clusterId];
 279
 280     if (sock) {
 281         PRINTVERBOSE1("detach: closing socket: %d\n", sock)
 282
 283         status = close(sock);
 284     }
 285
 286     return status;
 287 }
 288
 289 Int TransportRpmsg_bind(Void *handle, UInt32 queueId)
 290 {
 291     TransportRpmsg_Object *obj = (TransportRpmsg_Object *)handle;
 292     UInt16   queueIndex = queueId & 0x0000ffff;
 293     int      fd;
 294     int      err;
 295     uint64_t buf;
 296     UInt16   rprocId;
 297
 298     rprocId = obj->rprocId;
 299
 300     PRINTVERBOSE2("TransportRpmsg_bind: creating endpoint for rprocId %d "
 301             "queueIndex %d\n", rprocId, queueIndex)
 302
 303     /*  Create the socket to receive messages for this messageQ. */
 304     fd = socket(AF_RPMSG, SOCK_SEQPACKET, 0);
 305     if (fd < 0) {
 306         printf("TransportRpmsg_bind: socket call failed: %d (%s)\n",
 307                 errno, strerror(errno));
 308         goto exitClose;
 309     }
 310
 311     PRINTVERBOSE1("TransportRpmsg_bind: created socket fd %d\n", fd)
 312
 313     err = SocketBindAddr(fd, rprocId, (UInt32)queueIndex);
 314     if (err < 0) {
 315         /* don't hard-printf since this is no longer fatal */
 316         PRINTVERBOSE2("TransportRpmsg_bind: bind failed: %d (%s)\n",
 317                       errno, strerror(errno))
 318
 319         close(fd);
 320
 321         return -1;
 322     }
 323
 324     pthread_mutex_lock(&TransportRpmsg_module->gate);
 325
 326     /* add to our fat fd array and update select() parameters */
 327     TransportRpmsg_module->inFds[TransportRpmsg_module->nInFds++] = fd;
 328     TransportRpmsg_module->maxFd = MAX(TransportRpmsg_module->maxFd, fd);
 329     FD_SET(fd, &TransportRpmsg_module->rfds);
 330
 331     pthread_mutex_unlock(&TransportRpmsg_module->gate);
 332
 333     bindFdToQueueIndex(obj, fd, queueIndex);
 334
 335     /*
 336      * Even though we use the unblock event as just a signalling event with
 337      * no related payload, we need to write some non-zero value.  Might as
 338      * well make it the fd (which the reader could decide to use if needed).
 339      */
 340     buf = fd;
 341     write(TransportRpmsg_module->unblockEvent, &buf, sizeof (buf));
 342
 343     goto exit;
 344
 345 exitClose:
 346     TransportRpmsg_unbind(handle, fd);
 347     fd = 0;
 348
 349 exit:
 350     return fd;
 351 }
 352
 353 Int TransportRpmsg_unbind(Void *handle, UInt32 queueId)
 354 {
 355     TransportRpmsg_Object *obj = (TransportRpmsg_Object *)handle;
 356     UInt16 queueIndex = queueId & 0x0000ffff;
 357     uint64_t buf;
 358     Int    status = MessageQ_S_SUCCESS;
 359     int    maxFd;
 360     int    fd;
 361     int    i;
 362     int    j;
 363
 364     fd = queueIndexToFd(obj, queueIndex);
 365     if (!fd) {
 366         PRINTVERBOSE1("TransportRpmsg_unbind: queueId 0x%x not bound\n",
 367                       queueId)
 368
 369         return -1;
 370     }
 371
 372     PRINTVERBOSE1("TransportRpmsg_unbind: closing socket %d\n", fd)
 373
 374     pthread_mutex_lock(&TransportRpmsg_module->gate);
 375
 376     /* remove from input fd array */
 377     for (i = 0; i < TransportRpmsg_module->nInFds; i++) {
 378         if (TransportRpmsg_module->inFds[i] == fd) {
 379             TransportRpmsg_module->nInFds--;
 380
 381             /* shift subsequent elements down */
 382             for (j = i; j < TransportRpmsg_module->nInFds; j++) {
 383                 TransportRpmsg_module->inFds[j] =
 384                     TransportRpmsg_module->inFds[j + 1];
 385             }
 386             TransportRpmsg_module->inFds[TransportRpmsg_module->nInFds] = 0;
 387
 388             FD_CLR(fd, &TransportRpmsg_module->rfds);
 389             if (fd == TransportRpmsg_module->maxFd) {
 390                 /* find new max fd */
 391                 maxFd = TransportRpmsg_module->unblockEvent;
 392                 for (j = 0; j < TransportRpmsg_module->nInFds; j++) {
 393                     maxFd = MAX(TransportRpmsg_module->inFds[j], maxFd);
 394                 }
 395                 TransportRpmsg_module->maxFd = maxFd;
 396             }
 397
 398             /*
 399              * Even though we use the unblock event as just a signalling
 400              * event with no related payload, we need to write some non-zero
 401              * value.  Might as well make it the fd (which the reader could
 402              * decide to use if needed).
 403              */
 404             buf = fd;
 405             write(TransportRpmsg_module->unblockEvent, &buf, sizeof (buf));
 406
 407             break;
 408         }
 409
 410         close(fd);
 411     }
 412
 413     unbindQueueIndex(obj, queueIndex);
 414
 415     pthread_mutex_unlock(&TransportRpmsg_module->gate);
 416
 417     return status;
 418 }
 419
 420 Bool TransportRpmsg_put(Void *handle, Ptr pmsg)
 421 {
 422     MessageQ_Msg msg  = (MessageQ_Msg)pmsg;
 423     Int     status    = TRUE;
 424     int     sock;
 425     int     err;
 426     UInt16  clusterId;
 427
 428     /*
 429      * Retrieve the socket for the AF_SYSLINK protocol associated with this
 430      * transport.
 431      */
 432     clusterId = msg->dstProc - MultiProc_getBaseIdOfCluster();
 433     sock = TransportRpmsg_module->sock[clusterId];
 434     if (!sock) {
 435         return FALSE;
 436     }
 437
 438     PRINTVERBOSE2("Sending msgId: %d via sock: %d\n", msg->msgId, sock)
 439
 440     err = send(sock, msg, msg->msgSize, 0);
 441     if (err < 0) {
 442         printf("TransportRpmsg_put: send failed: %d (%s)\n",
 443                errno, strerror(errno));
 444         status = FALSE;
 445
 446         goto exit;
 447     }
 448
 449     /*
 450      * Free the message, as this is a copy transport, we maintain MessageQ
 451      * semantics.
 452      */
 453     MessageQ_free(msg);
 454
 455 exit:
 456     return status;
 457 }
 458
 459 Bool TransportRpmsg_control(Void *handle, UInt cmd, UArg cmdArg)
 460 {
 461     return FALSE;
 462 }
 463
 464 void *rpmsgThreadFxn(void *arg)
 465 {
 466     static int lastFdx = 0;
 467     int      curFdx = 0;
 468     Int      status = MessageQ_S_SUCCESS;
 469     Int      tmpStatus;
 470     int      retval;
 471     uint64_t buf;
 472     fd_set   rfds;
 473     int      maxFd;
 474     int      nfds;
 475     MessageQ_Msg     retMsg;
 476     MessageQ_QueueId queueId;
 477
 478     while (TRUE) {
 479         pthread_mutex_lock(&TransportRpmsg_module->gate);
 480
 481         maxFd = TransportRpmsg_module->maxFd;
 482         rfds = TransportRpmsg_module->rfds;
 483         nfds = TransportRpmsg_module->nInFds;
 484
 485         pthread_mutex_unlock(&TransportRpmsg_module->gate);
 486
 487         PRINTVERBOSE3("rpmsgThreadFxn: maxFd %d rfds[1:0] 0x%08x%08x\n", maxFd,
 488                       (int)rfds.fds_bits[1], (int)rfds.fds_bits[0])
 489
 490         retval = select(maxFd + 1, &rfds, NULL, NULL, NULL);
 491         if (retval) {
 492             if (FD_ISSET(TransportRpmsg_module->unblockEvent, &rfds)) {
 493                 /*
 494                  * Our event was signalled by TransportRpmsg_bind()
 495                  * or TransportRpmsg_unbind() to tell us that the set of
 496                  * fds has changed.
 497                  */
 498                 PRINTVERBOSE0("rpmsgThreadFxn: got unblock event\n")
 499
 500                 /* we don't need the written value */
 501                 read(TransportRpmsg_module->unblockEvent, &buf, sizeof (buf));
 502             }
 503             else {
 504                 /* start where we last left off */
 505                 curFdx = lastFdx;
 506
 507                 /*
 508                  * The set of fds that's used by select has been recorded
 509                  * locally, but the array of fds that are scanned below is
 510                  * a changing set (MessageQ_create/delete() can change it).
 511                  * While this might present an issue in itself, one key
 512                  * takeaway is that 'nfds' must not be zero else the % below
 513                  * will cause a divide-by-zero exception.  We won't even get
 514                  * here if nfds == 0 since it's a local copy of the module's
 515                  * 'nInFds' which has to be > 0 for us to get here.  So, even
 516                  * though the module's 'nInFds' might go to 0 during this loop,
 517                  * the loop itself will still remain intact.
 518                  */
 519                 do {
 520                     if (FD_ISSET(TransportRpmsg_module->inFds[curFdx], &rfds)) {
 521
 522                         PRINTVERBOSE1("rpmsgThreadFxn: getting from fd %d\n",
 523                                       TransportRpmsg_module->inFds[curFdx])
 524
 525                         /* transport input fd was signalled: get the message */
 526                         tmpStatus = transportGet(
 527                             TransportRpmsg_module->inFds[curFdx], &retMsg);
 528                         if (tmpStatus < 0) {
 529                             printf("rpmsgThreadFxn: transportGet failed.");
 530                             status = MessageQ_E_FAIL;
 531                         }
 532                         else {
 533                             queueId = MessageQ_getDstQueue(retMsg);
 534
 535                             PRINTVERBOSE1("rpmsgThreadFxn: got message, "
 536                                     "delivering to queueId 0x%x\n", queueId)
 537
 538                             MessageQ_put(queueId, retMsg);
 539                         }
 540
 541                         lastFdx = (curFdx + 1) % nfds;
 542
 543                         break;
 544                     }
 545
 546                     curFdx = (curFdx + 1) % nfds;
 547                 } while (curFdx != lastFdx);
 548             }
 549         }
 550     }
 551
 552     return (void *)status;
 553 }
 554
 555 /*
 556  * ======== transportGet ========
 557  *  Retrieve a message waiting in the socket's queue.
 558 */
 559 static Int transportGet(int sock, MessageQ_Msg *retMsg)
 560 {
 561     Int           status    = MessageQ_S_SUCCESS;
 562     MessageQ_Msg  msg;
 563     struct sockaddr_rpmsg fromAddr;  /* [Socket address of sender] */
 564     unsigned int  len;
 565     int           byteCount;
 566
 567     /*
 568      * We have no way of peeking to see what message size we'll get, so we
 569      * allocate a message of max size to receive contents from the rpmsg socket
 570      * (currently, a copy transport)
 571      */
 572     msg = MessageQ_alloc(0, MESSAGEQ_RPMSG_MAXSIZE);
 573     if (!msg) {
 574         status = MessageQ_E_MEMORY;
 575         goto exit;
 576     }
 577
 578     memset(&fromAddr, 0, sizeof (fromAddr));
 579     len = sizeof (fromAddr);
 580
 581     byteCount = recvfrom(sock, msg, MESSAGEQ_RPMSG_MAXSIZE, 0,
 582                          (struct sockaddr *)&fromAddr, &len);
 583     if (len != sizeof (fromAddr)) {
 584         printf("recvfrom: got bad addr len (%d)\n", len);
 585         status = MessageQ_E_FAIL;
 586         goto exit;
 587     }
 588     if (byteCount < 0) {
 589         printf("recvfrom failed: %s (%d)\n", strerror(errno), errno);
 590         status = MessageQ_E_FAIL;
 591         goto exit;
 592     }
 593     else {
 594          /*
 595           * Update the allocated message size (even though this may waste
 596           * space when the actual message is smaller than the maximum rpmsg
 597           * size, the message will be freed soon anyway, and it avoids an
 598           * extra copy).
 599           */
 600          msg->msgSize = byteCount;
 601
 602          /*
 603           * If the message received was statically allocated, reset the
 604           * heapId, so the app can free it.
 605           */
 606          if (msg->heapId == MessageQ_STATICMSG)  {
 607              msg->heapId = 0;  /* for a copy transport, heap id is 0. */
 608          }
 609     }
 610
 611     PRINTVERBOSE1("transportGet: recvfrom socket: fd: %d\n", sock)
 612     PRINTVERBOSE3("\tReceived a msg: byteCount: %d, rpmsg addr: %d, rpmsg "
 613             "proc: %d\n", byteCount, fromAddr.addr, fromAddr.vproc_id)
 614     PRINTVERBOSE2("\tMessage Id: %d, Message size: %d\n", msg->msgId,
 615             msg->msgSize)
 616
 617     *retMsg = msg;
 618
 619 exit:
 620     return status;
 621 }
 622
 623 Void bindFdToQueueIndex(TransportRpmsg_Object *obj, Int fd, UInt16 qIndex)
 624 {
 625     Int *queues;
 626     Int *oldQueues;
 627     UInt oldSize;
 628
 629     if (qIndex >= obj->numQueues) {
 630         PRINTVERBOSE1("TransportRpmsg_bind: growing numQueues to %d\n",
 631                       qIndex + TransportRpmsg_GROWSIZE)
 632
 633         oldSize = obj->numQueues * sizeof (Int);
 634
 635         queues = calloc(qIndex + TransportRpmsg_GROWSIZE, sizeof (Int));
 636         memcpy(queues, obj->qIndexToFd, oldSize);
 637
 638         oldQueues = obj->qIndexToFd;
 639         obj->qIndexToFd = queues;
 640         obj->numQueues = qIndex + TransportRpmsg_GROWSIZE;
 641
 642         free(oldQueues);
 643     }
 644
 645     obj->qIndexToFd[qIndex] = fd;
 646 }
 647
 648 Void unbindQueueIndex(TransportRpmsg_Object *obj, UInt16 qIndex)
 649 {
 650     obj->qIndexToFd[qIndex] = 0;
 651 }
 652
 653 Int queueIndexToFd(TransportRpmsg_Object *obj, UInt16 qIndex)
 654 {
 655     return obj->qIndexToFd[qIndex];
 656 }
 657
 658 /*
 659  *  ======== TransportRpmsg_Factory_create ========
 660  *  Create the transport instances
 661  *
 662  *  Attach to all remote processors. For now, must attach to
 663  *  at least one to tolerate MessageQ_E_RESOURCE failures.
 664  *
 665  *  This function implements the IPC Factory interface, so it
 666  *  returns Ipc status codes.
 667  */
 668 Int TransportRpmsg_Factory_create(Void)
 669 {
 670     Int     status;
 671     Int     attachStatus;
 672     Int     i;
 673     UInt16  procId;
 674     Int32   attachedAny;
 675     UInt16  clusterSize;
 676     UInt16  clusterBase;
 677
 678     TransportRpmsg_Handle      *inst;
 679     TransportRpmsg_Handle       transport;
 680     TransportRpmsg_Params       params;
 681     IMessageQTransport_Handle   iMsgQTrans;
 682
 683
 684     status = Ipc_S_SUCCESS;
 685     attachedAny = FALSE;
 686
 687     /* needed to enumerate processors in cluster */
 688     clusterSize = MultiProc_getNumProcsInCluster();
 689     clusterBase = MultiProc_getBaseIdOfCluster();
 690
 691     /* allocate the instance array */
 692     inst = calloc(clusterSize, sizeof(TransportRpmsg_Handle));
 693
 694     if (inst == NULL) {
 695         printf("Error: TransportRpmsg_Factory_create failed, no memory\n");
 696         status = Ipc_E_MEMORY;
 697         goto exit;
 698     }
 699
 700     TransportRpmsg_module->inst = inst;
 701
 702     /* create transport instance for all processors in cluster */
 703     for (i = 0, procId = clusterBase; i < clusterSize; i++, procId++) {
 704
 705         if (MultiProc_self() == procId) {
 706             continue;
 707         }
 708
 709         params.rprocId = procId;
 710         transport = TransportRpmsg_create(&params, &attachStatus);
 711
 712         if (transport != NULL) {
 713             iMsgQTrans = TransportRpmsg_upCast(transport);
 714             MessageQ_registerTransport(iMsgQTrans, procId, 0);
 715             attachedAny = TRUE;
 716         }
 717         else {
 718             if (attachStatus == MessageQ_E_RESOURCE) {
 719                 continue;
 720             }
 721             printf("TransportRpmsg_Factory_create: failed to attach to "
 722                     "procId=%d status=%d\n", procId, attachStatus);
 723             status = Ipc_E_FAIL;
 724             break;
 725         }
 726
 727         TransportRpmsg_module->inst[i] = transport;
 728     }
 729
 730     if (!attachedAny) {
 731         status = Ipc_E_FAIL;
 732     }
 733
 734 exit:
 735     return (status);
 736 }
 737
 738 /*
 739  *  ======== TransportRpmsg_Factory_delete ========
 740  *  Finalize the transport instances
 741  */
 742 Void TransportRpmsg_Factory_delete(Void)
 743 {
 744     Int     i;
 745     UInt16  procId;
 746     UInt16  clusterSize;
 747     UInt16  clusterBase;
 748
 749     /* needed to enumerate processors in cluster */
 750     clusterSize = MultiProc_getNumProcsInCluster();
 751     clusterBase = MultiProc_getBaseIdOfCluster();
 752
 753     /* detach from all remote processors, assuming they are up */
 754     for (i = 0, procId = clusterBase; i < clusterSize; i++, procId++) {
 755
 756         if (MultiProc_self() == procId) {
 757             continue;
 758         }
 759
 760         if (TransportRpmsg_module->inst[i] != NULL) {
 761             MessageQ_unregisterTransport(procId, 0);
 762             TransportRpmsg_delete(&(TransportRpmsg_module->inst[i]));
 763         }
 764     }
 765
 766     return;
 767 }