qnx/src/api/MessageQ.c

   1 /*
   2  * Copyright (c) 2012-2013, Texas Instruments Incorporated
   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  *  @file   MessageQ.c
  34  *
  35  *  @brief  MessageQ module "client" implementation
  36  *
  37  *  This implementation is geared for use in a "client/server" model, whereby
  38  *  system-wide data is maintained in a "server" component and process-
  39  *  specific data is handled here.  At the moment, this implementation
  40  *  connects and communicates with LAD for the server connection.
  41  *
  42  *  The MessageQ module supports the structured sending and receiving of
  43  *  variable length messages. This module can be used for homogeneous or
  44  *  heterogeneous multi-processor messaging.
  45  *
  46  *  MessageQ provides more sophisticated messaging than other modules. It is
  47  *  typically used for complex situations such as multi-processor messaging.
  48  *
  49  *  The following are key features of the MessageQ module:
  50  *  -Writers and readers can be relocated to another processor with no
  51  *   runtime code changes.
  52  *  -Timeouts are allowed when receiving messages.
  53  *  -Readers can determine the writer and reply back.
  54  *  -Receiving a message is deterministic when the timeout is zero.
  55  *  -Messages can reside on any message queue.
  56  *  -Supports zero-copy transfers.
  57  *  -Can send and receive from any type of thread.
  58  *  -Notification mechanism is specified by application.
  59  *  -Allows QoS (quality of service) on message buffer pools. For example,
  60  *   using specific buffer pools for specific message queues.
  61  *
  62  *  Messages are sent and received via a message queue. A reader is a thread
  63  *  that gets (reads) messages from a message queue. A writer is a thread that
  64  *  puts (writes) a message to a message queue. Each message queue has one
  65  *  reader and can have many writers. A thread may read from or write to multiple
  66  *  message queues.
  67  *
  68  *  Conceptually, the reader thread owns a message queue. The reader thread
  69  *  creates a message queue. Writer threads  a created message queues to
  70  *  get access to them.
  71  *
  72  *  Message queues are identified by a system-wide unique name. Internally,
  73  *  MessageQ uses the NameServermodule for managing
  74  *  these names. The names are used for opening a message queue. Using
  75  *  names is not required.
  76  *
  77  *  Messages must be allocated from the MessageQ module. Once a message is
  78  *  allocated, it can be sent on any message queue. Once a message is sent, the
  79  *  writer loses ownership of the message and should not attempt to modify the
  80  *  message. Once the reader receives the message, it owns the message. It
  81  *  may either free the message or re-use the message.
  82  *
  83  *  Messages in a message queue can be of variable length. The only
  84  *  requirement is that the first field in the definition of a message must be a
  85  *  MsgHeader structure. For example:
  86  *  typedef struct MyMsg {
  87  *      MessageQ_MsgHeader header;
  88  *      ...
  89  *  } MyMsg;
  90  *
  91  *  The MessageQ API uses the MessageQ_MsgHeader internally. Your application
  92  *  should not modify or directly access the fields in the MessageQ_MsgHeader.
  93  *
  94  *  All messages sent via the MessageQ module must be allocated from a
  95  *  Heap implementation. The heap can be used for
  96  *  other memory allocation not related to MessageQ.
  97  *
  98  *  An application can use multiple heaps. The purpose of having multiple
  99  *  heaps is to allow an application to regulate its message usage. For
 100  *  example, an application can allocate critical messages from one heap of fast
 101  *  on-chip memory and non-critical messages from another heap of slower
 102  *  external memory
 103  *
 104  *  MessageQ does support the usage of messages that allocated via the
 105  *  alloc function. Please refer to the staticMsgInit
 106  *  function description for more details.
 107  *
 108  *  In a multiple processor system, MessageQ communications to other
 109  *  processors via MessageQTransport instances. There must be one and
 110  *  only one MessageQTransport instance for each processor where communication
 111  *  is desired.
 112  *  So on a four processor system, each processor must have three
 113  *  MessageQTransport instance.
 114  *
 115  *  The user only needs to create the MessageQTransport instances. The instances
 116  *  are responsible for registering themselves with MessageQ.
 117  *  This is accomplished via the registerTransport function.
 118  *
 119  *  ============================================================================
 120  */
 121
 122
 123 /* Standard headers */
 124 #include <Std.h>
 125
 126 /* Linux specific header files, replacing OSAL: */
 127 #include <pthread.h>
 128
 129 /* Module level headers */
 130 #include <ti/ipc/NameServer.h>
 131 #include <ti/ipc/MultiProc.h>
 132 #include <_MultiProc.h>
 133 #include <ti/ipc/MessageQ.h>
 134 #include <_MessageQ.h>
 135 #include <_log.h>
 136 #include <ti/syslink/inc/MessageQDrvDefs.h>
 137
 138 #include <sys/select.h>
 139 #include <sys/time.h>
 140 #include <sys/types.h>
 141 #include <sys/param.h>
 142
 143 #include <errno.h>
 144 #include <stdio.h>
 145 #include <string.h>
 146 #include <stdlib.h>
 147 #include <unistd.h>
 148 #include <assert.h>
 149 #include <fcntl.h>
 150
 151 #include <ti/syslink/inc/usr/Qnx/MessageQDrv.h>
 152
 153 /* TI IPC utils: */
 154 #include <TiIpcFxns.h>
 155
 156 #include <ti/syslink/inc/ti/ipc/ti_ipc.h>
 157
 158 /* =============================================================================
 159  * Macros/Constants
 160  * =============================================================================
 161  */
 162
 163 /*!
 164  *  @brief  Name of the reserved NameServer used for MessageQ.
 165  */
 166 #define MessageQ_NAMESERVER  "MessageQ"
 167
 168 /* More magic rpmsg port numbers: */
 169 #define MESSAGEQ_RPMSG_PORT       61
 170 #define MESSAGEQ_RPMSG_MAXSIZE   512
 171 #define RPMSG_RESERVED_ADDRESSES  (1024)
 172
 173 /* Trace flag settings: */
 174 #define TRACESHIFT    12
 175 #define TRACEMASK     0x1000
 176
 177 /* =============================================================================
 178  * Structures & Enums
 179  * =============================================================================
 180  */
 181
 182 /* structure for MessageQ module state */
 183 typedef struct MessageQ_ModuleObject {
 184     Int                 refCount;
 185     /*!< Reference count */
 186     NameServer_Handle   nameServer;
 187     /*!< Handle to the local NameServer used for storing GP objects */
 188     pthread_mutex_t     gate;
 189     /*!< Handle of gate to be used for local thread safety */
 190     MessageQ_Params     defaultInstParams;
 191     /*!< Default instance creation parameters */
 192     int                 ipcFd[MultiProc_MAXPROCESSORS];
 193     /*!< File Descriptors for sending to each remote processor */
 194     int                 seqNum;
 195     /*!< Process-specific sequence number */
 196 } MessageQ_ModuleObject;
 197
 198 /*!
 199  *  @brief  Structure for the Handle for the MessageQ.
 200  */
 201 typedef struct MessageQ_Object_tag {
 202     MessageQ_Params         params;
 203     /*! Instance specific creation parameters */
 204     MessageQ_QueueId        queue;
 205     /* Unique id */
 206     int                     ipcFd;
 207     /* File Descriptors to receive from a message queue. */
 208     int                     unblockFdW;
 209     /* Write this fd to unblock the select() call in MessageQ _get() */
 210     int                     unblockFdR;
 211      /* File Descriptor to block on to listen to unblockFdW. */
 212     void                    *serverHandle;
 213 } MessageQ_Object;
 214
 215 static Bool verbose = FALSE;
 216
 217 /* =============================================================================
 218  *  Globals
 219  * =============================================================================
 220  */
 221 static MessageQ_ModuleObject MessageQ_state =
 222 {
 223     .refCount               = 0,
 224     .nameServer             = NULL,
 225 };
 226
 227 /*!
 228  *  @var    MessageQ_module
 229  *
 230  *  @brief  Pointer to the MessageQ module state.
 231  */
 232 MessageQ_ModuleObject * MessageQ_module = &MessageQ_state;
 233
 234
 235 /* =============================================================================
 236  * Forward declarations of internal functions
 237  * =============================================================================
 238  */
 239
 240 /* This is a helper function to initialize a message. */
 241 static Int transportCreateEndpoint(int * fd, UInt16 queueIndex);
 242 static Int transportCloseEndpoint(int fd);
 243 static Int transportGet(int fd, MessageQ_Msg * retMsg);
 244 static Int transportPut(MessageQ_Msg msg, UInt16 dstId, UInt16 dstProcId);
 245
 246 /* =============================================================================
 247  * APIS
 248  * =============================================================================
 249  */
 250 /* Function to get default configuration for the MessageQ module.
 251  *
 252  */
 253 Void MessageQ_getConfig (MessageQ_Config * cfg)
 254 {
 255     Int status;
 256     MessageQDrv_CmdArgs cmdArgs;
 257
 258     assert (cfg != NULL);
 259
 260     cmdArgs.args.getConfig.config = cfg;
 261     status = MessageQDrv_ioctl (CMD_MESSAGEQ_GETCONFIG, &cmdArgs);
 262
 263     if (status < 0) {
 264         PRINTVERBOSE1("MessageQ_getConfig: API (through IOCTL) failed, \
 265             status=%d\n", status)
 266     }
 267
 268     return;
 269 }
 270
 271 /* Function to setup the MessageQ module. */
 272 Int MessageQ_setup (const MessageQ_Config * cfg)
 273 {
 274     Int status;
 275     MessageQDrv_CmdArgs cmdArgs;
 276
 277     Int i;
 278
 279     cmdArgs.args.setup.config = (MessageQ_Config *) cfg;
 280     status = MessageQDrv_ioctl(CMD_MESSAGEQ_SETUP, &cmdArgs);
 281     if (status < 0) {
 282         PRINTVERBOSE1("MessageQ_setup: API (through IOCTL) failed, \
 283             status=%d\n", status)
 284         return status;
 285     }
 286
 287     MessageQ_module->nameServer = cmdArgs.args.setup.nameServerHandle;
 288     MessageQ_module->seqNum = 0;
 289
 290     /* Create a default local gate. */
 291     pthread_mutex_init (&(MessageQ_module->gate), NULL);
 292
 293     /* Clear ipcFd array. */
 294     for (i = 0; i < MultiProc_MAXPROCESSORS; i++) {
 295        MessageQ_module->ipcFd[i]      = -1;
 296     }
 297
 298     return status;
 299 }
 300
 301 /*
 302  * Function to destroy the MessageQ module.
 303  */
 304 Int MessageQ_destroy (void)
 305 {
 306     Int status;
 307     MessageQDrv_CmdArgs    cmdArgs;
 308
 309     status = MessageQDrv_ioctl (CMD_MESSAGEQ_DESTROY, &cmdArgs);
 310     if (status < 0) {
 311         PRINTVERBOSE1("MessageQ_destroy: API (through IOCTL) failed, \
 312             status=%d\n", status)
 313     }
 314
 315     return status;
 316 }
 317
 318 /* Function to initialize the parameters for the MessageQ instance. */
 319 Void MessageQ_Params_init (MessageQ_Params * params)
 320 {
 321     memcpy (params, &(MessageQ_module->defaultInstParams),
 322             sizeof (MessageQ_Params));
 323
 324     return;
 325 }
 326
 327 /*
 328  *   Function to create a MessageQ object for receiving.
 329  *
 330  *   Create a file descriptor and bind the source address
 331  *   (local ProcId/MessageQ ID) in
 332  *   order to get messages dispatched to this messageQ.
 333  */
 334 MessageQ_Handle MessageQ_create (String name, const MessageQ_Params * params)
 335 {
 336     Int                   status    = MessageQ_S_SUCCESS;
 337     MessageQ_Object *     obj    = NULL;
 338     UInt16                queueIndex = 0u;
 339     UInt16                procId;
 340     MessageQDrv_CmdArgs   cmdArgs;
 341     int                   fildes[2];
 342
 343     cmdArgs.args.create.params = (MessageQ_Params *) params;
 344     cmdArgs.args.create.name = name;
 345     if (name != NULL) {
 346         cmdArgs.args.create.nameLen = (strlen (name) + 1);
 347     }
 348     else {
 349         cmdArgs.args.create.nameLen = 0;
 350     }
 351
 352     status = MessageQDrv_ioctl (CMD_MESSAGEQ_CREATE, &cmdArgs);
 353     if (status < 0) {
 354         PRINTVERBOSE1("MessageQ_create: API (through IOCTL) failed, \
 355             status=%d\n", status)
 356         return NULL;
 357     }
 358
 359     /* Create the generic obj */
 360     obj = (MessageQ_Object *)calloc(1, sizeof (MessageQ_Object));
 361
 362     if (params != NULL) {
 363        /* Populate the params member */
 364         memcpy((Ptr) &obj->params, (Ptr)params, sizeof (MessageQ_Params));
 365     }
 366
 367     procId = MultiProc_self();
 368     queueIndex = (MessageQ_QueueIndex)cmdArgs.args.create.queueId;
 369     obj->queue = cmdArgs.args.create.queueId;
 370     obj->serverHandle = cmdArgs.args.create.handle;
 371
 372     PRINTVERBOSE2("MessageQ_create: creating endpoint for: %s, \
 373        queueIndex: %d\n", name, queueIndex)
 374     status = transportCreateEndpoint(&obj->ipcFd, queueIndex);
 375     if (status < 0) {
 376        goto cleanup;
 377     }
 378
 379     /*
 380      * Now, to support MessageQ_unblock() functionality, create an event object.
 381      * Writing to this event will unblock the select() call in MessageQ_get().
 382      */
 383     if (pipe(fildes) == -1) {
 384         printf ("MessageQ_create: pipe creation failed: %d, %s\n",
 385                    errno, strerror(errno));
 386         status = MessageQ_E_FAIL;
 387     }
 388     obj->unblockFdW = fildes[1];
 389     obj->unblockFdR = fildes[0];
 390
 391 cleanup:
 392     /* Cleanup if fail: */
 393     if (status < 0) {
 394         MessageQ_delete((MessageQ_Handle *)&obj);
 395     }
 396
 397     return ((MessageQ_Handle) obj);
 398 }
 399
 400 /*
 401  * Function to delete a MessageQ object for a specific slave processor.
 402  *
 403  * Deletes the file descriptors associated with this MessageQ object.
 404  */
 405 Int MessageQ_delete (MessageQ_Handle * handlePtr)
 406 {
 407     Int               status    = MessageQ_S_SUCCESS;
 408     MessageQ_Object * obj       = NULL;
 409     MessageQDrv_CmdArgs cmdArgs;
 410
 411     obj = (MessageQ_Object *) (*handlePtr);
 412
 413     cmdArgs.args.deleteMessageQ.handle = obj->serverHandle;
 414     status = MessageQDrv_ioctl (CMD_MESSAGEQ_DELETE, &cmdArgs);
 415     if (status < 0) {
 416         PRINTVERBOSE1("MessageQ_delete: API (through IOCTL) failed, \
 417             status=%d\n", status)
 418     }
 419
 420     /* Close the fds used for MessageQ_unblock(): */
 421     close(obj->unblockFdW);
 422     close(obj->unblockFdR);
 423
 424     /* Close the communication endpoint: */
 425     status = transportCloseEndpoint(obj->ipcFd);
 426
 427     /* Now free the obj */
 428     free (obj);
 429     *handlePtr = NULL;
 430
 431     return (status);
 432 }
 433
 434 /*
 435  *  Opens an instance of MessageQ for sending.
 436  *
 437  *  We need not create a tiipc file descriptor here; the file descriptors for
 438  *  all remote processors were created during MessageQ_attach(), and will be
 439  *  retrieved during MessageQ_put().
 440  */
 441 Int MessageQ_open (String name, MessageQ_QueueId * queueId)
 442 {
 443     Int status = MessageQ_S_SUCCESS;
 444
 445     status = NameServer_getUInt32 (MessageQ_module->nameServer,
 446                                      name, queueId, NULL);
 447
 448     if (status == NameServer_E_NOTFOUND) {
 449         /* Set return queue ID to invalid. */
 450         *queueId = MessageQ_INVALIDMESSAGEQ;
 451         status = MessageQ_E_NOTFOUND;
 452     }
 453     else if (status >= 0) {
 454         /* Override with a MessageQ status code. */
 455         status = MessageQ_S_SUCCESS;
 456     }
 457     else {
 458         /* Set return queue ID to invalid. */
 459         *queueId = MessageQ_INVALIDMESSAGEQ;
 460         /* Override with a MessageQ status code. */
 461         if (status == NameServer_E_TIMEOUT) {
 462             status = MessageQ_E_TIMEOUT;
 463         }
 464         else {
 465             status = MessageQ_E_FAIL;
 466         }
 467     }
 468
 469     return (status);
 470 }
 471
 472 /* Closes previously opened instance of MessageQ module. */
 473 Int MessageQ_close (MessageQ_QueueId * queueId)
 474 {
 475     Int32 status = MessageQ_S_SUCCESS;
 476
 477     /* Nothing more to be done for closing the MessageQ. */
 478     *queueId = MessageQ_INVALIDMESSAGEQ;
 479
 480     return (status);
 481 }
 482
 483 /*
 484  * Place a message onto a message queue.
 485  *
 486  * Calls TransportShm_put(), which handles the sending of the message using the
 487  * appropriate kernel interface (socket, device ioctl) call for the remote
 488  * procId encoded in the queueId argument.
 489  *
 490  */
 491 Int MessageQ_put (MessageQ_QueueId queueId, MessageQ_Msg msg)
 492 {
 493     Int      status;
 494     UInt16   dstProcId  = (UInt16)(queueId >> 16);
 495     UInt16   queueIndex = (MessageQ_QueueIndex)(queueId & 0x0000ffff);
 496
 497     msg->dstId     = queueIndex;
 498     msg->dstProc   = dstProcId;
 499
 500     status = transportPut(msg, queueIndex, dstProcId);
 501
 502     return (status);
 503 }
 504
 505 /*
 506  * Gets a message for a message queue and blocks if the queue is empty.
 507  * If a message is present, it returns it.  Otherwise it blocks
 508  * waiting for a message to arrive.
 509  * When a message is returned, it is owned by the caller.
 510  *
 511  * We block using select() on the receiving tiipc file descriptor, then
 512  * get the waiting message via a read.
 513  * We use the file descriptors stored in the messageQ object via a previous
 514  * call to MessageQ_create().
 515  *
 516  * Note: We currently do not support messages to be sent between threads on the
 517  * lcoal processor.
 518  *
 519  */
 520 Int MessageQ_get (MessageQ_Handle handle, MessageQ_Msg * msg ,UInt timeout)
 521 {
 522     Int     status = MessageQ_S_SUCCESS;
 523     Int     tmpStatus;
 524     MessageQ_Object * obj = (MessageQ_Object *) handle;
 525     int     retval;
 526     int     nfds;
 527     fd_set  rfds;
 528     struct  timeval tv;
 529     void    *timevalPtr;
 530     int     maxfd = 0;
 531
 532     /* Wait (with timeout) and retreive message */
 533     FD_ZERO(&rfds);
 534     FD_SET(obj->ipcFd, &rfds);
 535     maxfd = obj->ipcFd;
 536
 537     /* Wait also on the event fd, which may be written by MessageQ_unblock(): */
 538     FD_SET(obj->unblockFdR, &rfds);
 539
 540     if (timeout == MessageQ_FOREVER) {
 541         timevalPtr = NULL;
 542     }
 543     else {
 544         /* Timeout given in msec: convert:  */
 545         tv.tv_sec = timeout / 1000;
 546         tv.tv_usec = (timeout % 1000) * 1000;
 547         timevalPtr = &tv;
 548     }
 549     /* Add one to last fd created: */
 550     nfds = ((maxfd > obj->unblockFdR) ? maxfd : obj->unblockFdR) + 1;
 551
 552     retval = select(nfds, &rfds, NULL, NULL, timevalPtr);
 553     if (retval)  {
 554         if (FD_ISSET(obj->unblockFdR, &rfds))  {
 555             /*
 556              * Our event was signalled by MessageQ_unblock().
 557              *
 558              * This is typically done during a shutdown sequence, where
 559              * the intention of the client would be to ignore (i.e. not fetch)
 560              * any pending messages in the transport's queue.
 561              * Thus, we shall not check for nor return any messages.
 562              */
 563             *msg = NULL;
 564             status = MessageQ_E_UNBLOCKED;
 565         }
 566         else {
 567             if (FD_ISSET(obj->ipcFd, &rfds)) {
 568                 /* Our transport's fd was signalled: Get the message: */
 569                 tmpStatus = transportGet(obj->ipcFd, msg);
 570                 if (tmpStatus < 0) {
 571                     printf ("MessageQ_get: tranposrtshm_get failed.");
 572                     status = MessageQ_E_FAIL;
 573                 }
 574             }
 575         }
 576     }
 577     else if (retval == 0) {
 578         *msg = NULL;
 579         status = MessageQ_E_TIMEOUT;
 580     }
 581
 582     return (status);
 583 }
 584
 585 /*
 586  * Return a count of the number of messages in the queue
 587  *
 588  * TBD: To be implemented. Return -1 for now.
 589  */
 590 Int MessageQ_count (MessageQ_Handle handle)
 591 {
 592     Int               count = -1;
 593     return (count);
 594 }
 595
 596 /* Initializes a message not obtained from MessageQ_alloc. */
 597 Void MessageQ_staticMsgInit (MessageQ_Msg msg, UInt32 size)
 598 {
 599     /* Fill in the fields of the message */
 600     MessageQ_msgInit (msg);
 601     msg->heapId  = MessageQ_STATICMSG;
 602     msg->msgSize = size;
 603 }
 604
 605 /*
 606  * Allocate a message and initialize the needed fields (note some
 607  * of the fields in the header are set via other APIs or in the
 608  * MessageQ_put function,
 609  */
 610 MessageQ_Msg MessageQ_alloc (UInt16 heapId, UInt32 size)
 611 {
 612     MessageQ_Msg msg       = NULL;
 613
 614     /*
 615      * heapId not used for local alloc (as this is over a copy transport), but
 616      * we need to send to other side as heapId is used in BIOS transport:
 617      */
 618     msg = (MessageQ_Msg)calloc (1, size);
 619     MessageQ_msgInit (msg);
 620     msg->msgSize = size;
 621     msg->heapId  = heapId;
 622
 623     return msg;
 624 }
 625
 626 /* Frees the message back to the heap that was used to allocate it. */
 627 Int MessageQ_free (MessageQ_Msg msg)
 628 {
 629     UInt32         status = MessageQ_S_SUCCESS;
 630
 631     /* Check to ensure this was not allocated by user: */
 632     if (msg->heapId == MessageQ_STATICMSG)  {
 633         status =  MessageQ_E_CANNOTFREESTATICMSG;
 634     }
 635     else {
 636         free (msg);
 637     }
 638
 639     return status;
 640 }
 641
 642 /* Register a heap with MessageQ. */
 643 Int MessageQ_registerHeap (Ptr heap, UInt16 heapId)
 644 {
 645     Int  status = MessageQ_S_SUCCESS;
 646
 647     /* Do nothing, as this uses a copy transport: */
 648
 649     return status;
 650 }
 651
 652 /* Unregister a heap with MessageQ. */
 653 Int MessageQ_unregisterHeap (UInt16 heapId)
 654 {
 655     Int  status = MessageQ_S_SUCCESS;
 656
 657     /* Do nothing, as this uses a copy transport: */
 658
 659     return status;
 660 }
 661
 662 /* Unblocks a MessageQ */
 663 Void MessageQ_unblock (MessageQ_Handle handle)
 664 {
 665     MessageQ_Object * obj   = (MessageQ_Object *) handle;
 666     char         buf = 'n';
 667     int          numBytes;
 668
 669     /* Write to pipe to awaken any threads blocked on this messageQ: */
 670     numBytes = write(obj->unblockFdW, &buf, 1);
 671 }
 672
 673 /* Embeds a source message queue into a message. */
 674 Void MessageQ_setReplyQueue (MessageQ_Handle handle, MessageQ_Msg msg)
 675 {
 676     MessageQ_Object * obj   = (MessageQ_Object *) handle;
 677
 678     msg->replyId   = (UInt16)(obj->queue);
 679     msg->replyProc = (UInt16)(obj->queue >> 16);
 680 }
 681
 682 /* Returns the QueueId associated with the handle. */
 683 MessageQ_QueueId MessageQ_getQueueId (MessageQ_Handle handle)
 684 {
 685     MessageQ_Object * obj = (MessageQ_Object *) handle;
 686     UInt32            queueId;
 687
 688     queueId = (obj->queue);
 689
 690     return queueId;
 691 }
 692
 693 /* Sets the tracing of a message */
 694 Void MessageQ_setMsgTrace (MessageQ_Msg msg, Bool traceFlag)
 695 {
 696     msg->flags = (msg->flags & ~TRACEMASK) |   (traceFlag << TRACESHIFT);
 697 }
 698
 699 /*
 700  *  Returns the amount of shared memory used by one transport instance.
 701  *
 702  *  The MessageQ module itself does not use any shared memory but the
 703  *  underlying transport may use some shared memory.
 704  */
 705 SizeT MessageQ_sharedMemReq (Ptr sharedAddr)
 706 {
 707     SizeT memReq = 0u;
 708
 709     /* Do nothing, as this is a copy transport. */
 710
 711     return (memReq);
 712 }
 713
 714 /*
 715  *  Opens a file descriptor for this remote proc.
 716  *
 717  *  Only opens it if one does not already exist for this procId.
 718  *
 719  *  Note: remoteProcId may be MultiProc_Self() for loopback case.
 720  */
 721 Int MessageQ_attach (UInt16 remoteProcId, Ptr sharedAddr)
 722 {
 723     Int     status = MessageQ_S_SUCCESS;
 724     int     ipcFd;
 725
 726     PRINTVERBOSE1("MessageQ_attach: remoteProcId: %d\n", remoteProcId)
 727
 728     if (remoteProcId >= MultiProc_MAXPROCESSORS) {
 729         status = MessageQ_E_INVALIDPROCID;
 730         goto exit;
 731     }
 732
 733     pthread_mutex_lock (&(MessageQ_module->gate));
 734
 735     /* Only open a fd if one doesn't exist: */
 736     if (MessageQ_module->ipcFd[remoteProcId] == -1)  {
 737         /* Create a fd for sending messages to the remote proc: */
 738         ipcFd = open("/dev/tiipc", O_RDWR);
 739         if (ipcFd < 0) {
 740             status = MessageQ_E_FAIL;
 741             printf ("MessageQ_attach: open of tiipc device failed: %d, %s\n",
 742                        errno, strerror(errno));
 743         }
 744         else  {
 745             PRINTVERBOSE1("MessageQ_attach: opened tiipc fd for sending: %d\n",
 746                 ipcFd)
 747             MessageQ_module->ipcFd[remoteProcId] = ipcFd;
 748             /*
 749              * Connect to the remote endpoint and bind a reserved address as
 750              * local endpoint
 751              */
 752             Connect(ipcFd, remoteProcId, MESSAGEQ_RPMSG_PORT);
 753         }
 754     }
 755     else {
 756         status = MessageQ_E_ALREADYEXISTS;
 757     }
 758
 759     pthread_mutex_unlock (&(MessageQ_module->gate));
 760
 761 exit:
 762     return (status);
 763 }
 764
 765 /*
 766  *  Close the fd for this remote proc.
 767  *
 768  */
 769 Int MessageQ_detach (UInt16 remoteProcId)
 770 {
 771     Int status = MessageQ_S_SUCCESS;
 772     int ipcFd;
 773
 774     if (remoteProcId >= MultiProc_MAXPROCESSORS) {
 775         status = MessageQ_E_INVALIDPROCID;
 776         goto exit;
 777     }
 778
 779     pthread_mutex_lock (&(MessageQ_module->gate));
 780
 781     ipcFd = MessageQ_module->ipcFd[remoteProcId];
 782     if (close (ipcFd)) {
 783         status = MessageQ_E_OSFAILURE;
 784         printf("MessageQ_detach: close failed: %d, %s\n",
 785                        errno, strerror(errno));
 786     }
 787     else {
 788         PRINTVERBOSE1("MessageQ_detach: closed fd: %d\n", ipcFd)
 789         MessageQ_module->ipcFd[remoteProcId] = -1;
 790     }
 791
 792     pthread_mutex_unlock (&(MessageQ_module->gate));
 793
 794 exit:
 795     return (status);
 796 }
 797
 798 /*
 799  * This is a helper function to initialize a message.
 800  */
 801 Void MessageQ_msgInit (MessageQ_Msg msg)
 802 {
 803     msg->reserved0 = 0;  /* We set this to distinguish from NameServerMsg */
 804     msg->replyId   = (UInt16)MessageQ_INVALIDMESSAGEQ;
 805     msg->msgId     = MessageQ_INVALIDMSGID;
 806     msg->dstId     = (UInt16)MessageQ_INVALIDMESSAGEQ;
 807     msg->flags     = MessageQ_HEADERVERSION | MessageQ_NORMALPRI;
 808     msg->srcProc   = MultiProc_self();
 809
 810     pthread_mutex_lock(&(MessageQ_module->gate));
 811     msg->seqNum  = MessageQ_module->seqNum++;
 812     pthread_mutex_unlock(&(MessageQ_module->gate));
 813 }
 814
 815 /*
 816  * =============================================================================
 817  * Transport: Fxns kept here until need for a transport layer is realized.
 818  * =============================================================================
 819  */
 820 /*
 821  * ======== transportCreateEndpoint ========
 822  *
 823  * Create a communication endpoint to receive messages.
 824  */
 825 static Int transportCreateEndpoint(int * fd, UInt16 queueIndex)
 826 {
 827     Int          status    = MessageQ_S_SUCCESS;
 828     int          err;
 829
 830     /* Create a fd to the ti-ipc to receive messages for this messageQ */
 831     *fd= open("/dev/tiipc", O_RDWR);
 832     if (*fd < 0) {
 833         status = MessageQ_E_FAIL;
 834         printf ("transportCreateEndpoint: Couldn't open tiipc device: %d, %s\n",
 835                   errno, strerror(errno));
 836
 837         goto exit;
 838     }
 839
 840     PRINTVERBOSE1("transportCreateEndpoint: opened fd: %d\n", *fd)
 841
 842     err = BindAddr(*fd, (UInt32)queueIndex);
 843     if (err < 0) {
 844         status = MessageQ_E_FAIL;
 845         printf ("transportCreateEndpoint: bind failed: %d, %s\n",
 846                   errno, strerror(errno));
 847     }
 848
 849 exit:
 850     return (status);
 851 }
 852
 853 /*
 854  * ======== transportCloseEndpoint ========
 855  *
 856  *  Close the communication endpoint.
 857  */
 858 static Int transportCloseEndpoint(int fd)
 859 {
 860     Int status = MessageQ_S_SUCCESS;
 861
 862     PRINTVERBOSE1("transportCloseEndpoint: closing fd: %d\n", fd)
 863
 864     /* Stop communication to this endpoint */
 865     close(fd);
 866
 867     return (status);
 868 }
 869
 870 /*
 871  * ======== transportGet ========
 872  *  Retrieve a message waiting in the queue.
 873 */
 874 static Int transportGet(int fd, MessageQ_Msg * retMsg)
 875 {
 876     Int           status    = MessageQ_S_SUCCESS;
 877     MessageQ_Msg  msg;
 878     int           ret;
 879     int           byteCount;
 880     tiipc_remote_params remote;
 881
 882     /*
 883      * We have no way of peeking to see what message size we'll get, so we
 884      * allocate a message of max size to receive contents from tiipc
 885      * (currently, a copy transport)
 886      */
 887     msg = MessageQ_alloc (0, MESSAGEQ_RPMSG_MAXSIZE);
 888     if (!msg)  {
 889         status = MessageQ_E_MEMORY;
 890         goto exit;
 891     }
 892
 893     /* Get message */
 894     byteCount = read(fd, msg, MESSAGEQ_RPMSG_MAXSIZE);
 895     if (byteCount < 0) {
 896         printf("read failed: %s (%d)\n", strerror(errno), errno);
 897         status = MessageQ_E_FAIL;
 898         goto exit;
 899     }
 900     else {
 901          /* Update the allocated message size (even though this may waste space
 902           * when the actual message is smaller than the maximum rpmsg size,
 903           * the message will be freed soon anyway, and it avoids an extra copy).
 904           */
 905          msg->msgSize = byteCount;
 906
 907          /*
 908           * If the message received was statically allocated, reset the
 909           * heapId, so the app can free it.
 910           */
 911          if (msg->heapId == MessageQ_STATICMSG)  {
 912              msg->heapId = 0;  /* for a copy transport, heap id is 0. */
 913          }
 914     }
 915
 916     PRINTVERBOSE1("transportGet: read from fd: %d\n", fd)
 917     ret = ioctl(fd, TIIPC_IOCGETREMOTE, &remote);
 918     PRINTVERBOSE3("\tReceived a msg: byteCount: %d, rpmsg addr: %d, rpmsg \
 919         proc: %d\n", byteCount, remote.remote_addr, remote.remote_proc)
 920     PRINTVERBOSE2("\tMessage Id: %d, Message size: %d\n", msg->msgId, msg->msgSize)
 921
 922     *retMsg = msg;
 923
 924 exit:
 925     return (status);
 926 }
 927
 928 /*
 929  * ======== transportPut ========
 930  *
 931  * Write to tiipc file descriptor associated with
 932  * with this destination procID.
 933  */
 934 static Int transportPut(MessageQ_Msg msg, UInt16 dstId, UInt16 dstProcId)
 935 {
 936     Int     status    = MessageQ_S_SUCCESS;
 937     int     ipcFd;
 938     int     err;
 939
 940     /*
 941      * Retrieve the tiipc file descriptor associated with this
 942      * transport for the destination processor.
 943      */
 944     ipcFd = MessageQ_module->ipcFd[dstProcId];
 945
 946     PRINTVERBOSE2("Sending msgId: %d via fd: %d\n", msg->msgId, ipcFd)
 947
 948     /* send response message to remote processor */
 949     err = write(ipcFd, msg, msg->msgSize);
 950     if (err < 0) {
 951         printf ("transportPut: write failed: %d, %s\n",
 952                   errno, strerror(errno));
 953         status = MessageQ_E_FAIL;
 954     }
 955
 956     /*
 957      * Free the message, as this is a copy transport, we maintain MessageQ
 958      * semantics.
 959      */
 960     MessageQ_free (msg);
 961
 962     return (status);
 963 }