qnx/src/api/MessageQ.c

   1 /*
   2  * Copyright (c) 2012-2014, 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 <ti/ipc/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 <ti/syslink/inc/_MultiProc.h>
 133 #define MessageQ_internal 1     /* must be defined before include file */
 134 #include <ti/ipc/MessageQ.h>
 135 #include <_MessageQ.h>
 136 #include <_IpcLog.h>
 137 #include <ti/syslink/inc/MessageQDrvDefs.h>
 138
 139 #include <sys/select.h>
 140 #include <sys/time.h>
 141 #include <sys/types.h>
 142 #include <sys/param.h>
 143
 144 #include <errno.h>
 145 #include <stdio.h>
 146 #include <string.h>
 147 #include <stdlib.h>
 148 #include <unistd.h>
 149 #include <assert.h>
 150 #include <fcntl.h>
 151
 152 #include <ti/syslink/inc/usr/Qnx/MessageQDrv.h>
 153
 154 /* TI IPC utils: */
 155 #include <TiIpcFxns.h>
 156
 157 #include <ti/syslink/inc/ti/ipc/ti_ipc.h>
 158
 159 /* =============================================================================
 160  * Macros/Constants
 161  * =============================================================================
 162  */
 163
 164 /*!
 165  *  @brief  Name of the reserved NameServer used for MessageQ.
 166  */
 167 #define MessageQ_NAMESERVER  "MessageQ"
 168
 169 /* More magic rpmsg port numbers: */
 170 #define MESSAGEQ_RPMSG_PORT       61
 171 #define MESSAGEQ_RPMSG_MAXSIZE    512
 172 #define RPMSG_RESERVED_ADDRESSES  (1024)
 173
 174 /* MessageQ needs local address bound to be a 16-bit value */
 175 #define MAX_LOCAL_ADDR            0x10000
 176
 177 /* Trace flag settings: */
 178 #define TRACESHIFT    12
 179 #define TRACEMASK     0x1000
 180
 181 /* =============================================================================
 182  * Structures & Enums
 183  * =============================================================================
 184  */
 185
 186 /* params structure evolution */
 187 typedef struct {
 188     Void *synchronizer;
 189 } MessageQ_Params_Legacy;
 190
 191 typedef struct {
 192     Int __version;
 193     Void *synchronizer;
 194     MessageQ_QueueIndex queueIndex;
 195 } MessageQ_Params_Version2;
 196
 197 /* structure for MessageQ module state */
 198 typedef struct MessageQ_ModuleObject {
 199     Int                 refCount;
 200     /*!< Reference count */
 201     NameServer_Handle   nameServer;
 202     /*!< Handle to the local NameServer used for storing GP objects */
 203     pthread_mutex_t     gate;
 204     /*!< Handle of gate to be used for local thread safety */
 205     int                 ipcFd[MultiProc_MAXPROCESSORS];
 206     /*!< File Descriptors for sending to each remote processor */
 207     int                 seqNum;
 208     /*!< Process-specific sequence number */
 209     MessageQ_PutHookFxn putHookFxn;
 210     /*!< hook function for MessageQ_put method */
 211 } MessageQ_ModuleObject;
 212
 213 /*!
 214  *  @brief  Structure for the Handle for the MessageQ.
 215  */
 216 typedef struct MessageQ_Object_tag {
 217     MessageQ_Params         params;
 218     /*! Instance specific creation parameters */
 219     MessageQ_QueueId        queue;
 220     /* Unique id */
 221     int                     ipcFd;
 222     /* File Descriptors to receive from a message queue. */
 223     int                     unblockFdW;
 224     /* Write this fd to unblock the select() call in MessageQ _get() */
 225     int                     unblockFdR;
 226      /* File Descriptor to block on to listen to unblockFdW. */
 227     void                    *serverHandle;
 228 } MessageQ_Object;
 229
 230 static Bool verbose = FALSE;
 231
 232 /* =============================================================================
 233  *  Globals
 234  * =============================================================================
 235  */
 236 static MessageQ_ModuleObject MessageQ_state =
 237 {
 238     .refCount   = 0,
 239     .nameServer = NULL,
 240     .putHookFxn = NULL
 241 };
 242
 243 /*!
 244  *  @var    MessageQ_module
 245  *
 246  *  @brief  Pointer to the MessageQ module state.
 247  */
 248 MessageQ_ModuleObject * MessageQ_module = &MessageQ_state;
 249
 250
 251 /* =============================================================================
 252  * Forward declarations of internal functions
 253  * =============================================================================
 254  */
 255
 256 /* This is a helper function to initialize a message. */
 257 static Int transportCreateEndpoint(int * fd, UInt16 * queueIndex);
 258 static Int transportCloseEndpoint(int fd);
 259 static Int transportGet(int fd, MessageQ_Msg * retMsg);
 260 static Int transportPut(MessageQ_Msg msg, UInt16 dstId, UInt16 dstProcId);
 261
 262 /* =============================================================================
 263  * APIS
 264  * =============================================================================
 265  */
 266 /* Function to get default configuration for the MessageQ module.
 267  *
 268  */
 269 Void MessageQ_getConfig (MessageQ_Config * cfg)
 270 {
 271     Int status;
 272     MessageQDrv_CmdArgs cmdArgs;
 273
 274     assert (cfg != NULL);
 275
 276     cmdArgs.args.getConfig.config = cfg;
 277     status = MessageQDrv_ioctl (CMD_MESSAGEQ_GETCONFIG, &cmdArgs);
 278
 279     if (status < 0) {
 280         PRINTVERBOSE1("MessageQ_getConfig: API (through IOCTL) failed, \
 281             status=%d\n", status)
 282     }
 283
 284     return;
 285 }
 286
 287 /* Function to setup the MessageQ module. */
 288 Int MessageQ_setup (const MessageQ_Config * cfg)
 289 {
 290     Int status;
 291     MessageQDrv_CmdArgs cmdArgs;
 292
 293     Int i;
 294
 295     cmdArgs.args.setup.config = (MessageQ_Config *) cfg;
 296     status = MessageQDrv_ioctl(CMD_MESSAGEQ_SETUP, &cmdArgs);
 297     if (status < 0) {
 298         PRINTVERBOSE1("MessageQ_setup: API (through IOCTL) failed, \
 299             status=%d\n", status)
 300         return status;
 301     }
 302
 303     MessageQ_module->nameServer = cmdArgs.args.setup.nameServerHandle;
 304     MessageQ_module->seqNum = 0;
 305
 306     /* Create a default local gate. */
 307     pthread_mutex_init (&(MessageQ_module->gate), NULL);
 308
 309     /* Clear ipcFd array. */
 310     for (i = 0; i < MultiProc_MAXPROCESSORS; i++) {
 311        MessageQ_module->ipcFd[i]      = -1;
 312     }
 313
 314     return status;
 315 }
 316
 317 /*
 318  * Function to destroy the MessageQ module.
 319  */
 320 Int MessageQ_destroy (void)
 321 {
 322     Int status;
 323     MessageQDrv_CmdArgs    cmdArgs;
 324
 325     status = MessageQDrv_ioctl (CMD_MESSAGEQ_DESTROY, &cmdArgs);
 326     if (status < 0) {
 327         PRINTVERBOSE1("MessageQ_destroy: API (through IOCTL) failed, \
 328             status=%d\n", status)
 329     }
 330
 331     return status;
 332 }
 333
 334 /*
 335  *  ======== MessageQ_Params_init ========
 336  *  Legacy implementation.
 337  */
 338 Void MessageQ_Params_init(MessageQ_Params *params)
 339 {
 340     ((MessageQ_Params_Legacy *)params)->synchronizer = NULL;
 341 }
 342
 343 /*
 344  *  ======== MessageQ_Params_init__S ========
 345  *  New implementation which is version aware.
 346  */
 347 Void MessageQ_Params_init__S(MessageQ_Params *params, Int version)
 348 {
 349     MessageQ_Params_Version2 *params2;
 350
 351     switch (version) {
 352
 353         case MessageQ_Params_VERSION_2:
 354             params2 = (MessageQ_Params_Version2 *)params;
 355             params2->__version = MessageQ_Params_VERSION_2;
 356             params2->synchronizer = NULL;
 357             params2->queueIndex = MessageQ_ANY;
 358             break;
 359
 360         default:
 361             assert(FALSE);
 362             break;
 363     }
 364 }
 365
 366 /*
 367  *   Function to create a MessageQ object for receiving.
 368  *
 369  *   Create a file descriptor and bind the source address
 370  *   (local ProcId/MessageQ ID) in
 371  *   order to get messages dispatched to this messageQ.
 372  */
 373 MessageQ_Handle MessageQ_create (String name, const MessageQ_Params * pp)
 374 {
 375     Int                   status    = MessageQ_S_SUCCESS;
 376     MessageQ_Object *     obj    = NULL;
 377     UInt16                queueIndex = 0u;
 378     UInt16                procId;
 379     MessageQDrv_CmdArgs   cmdArgs;
 380     int                   fildes[2];
 381     MessageQ_Params       ps;
 382
 383     MessageQ_Params_init__S(&ps, MessageQ_Params_VERSION);
 384
 385     /* copy the given params into the current params structure */
 386     if (pp != NULL) {
 387
 388         /* snoop the params pointer to see if it's a legacy structure */
 389         if ((pp->__version == 0) || (pp->__version > 100)) {
 390             ps.synchronizer = ((MessageQ_Params_Legacy *)pp)->synchronizer;
 391         }
 392
 393         /* not legacy structure, use params version field */
 394         else if (pp->__version == MessageQ_Params_VERSION_2) {
 395             ps.__version = ((MessageQ_Params_Version2 *)pp)->__version;
 396             ps.synchronizer = ((MessageQ_Params_Version2 *)pp)->synchronizer;
 397             ps.queueIndex = ((MessageQ_Params_Version2 *)pp)->queueIndex;
 398         }
 399         else {
 400             assert(FALSE);
 401         }
 402     }
 403
 404     cmdArgs.args.create.params = &ps;
 405     cmdArgs.args.create.name = name;
 406
 407     if (name != NULL) {
 408         cmdArgs.args.create.nameLen = (strlen (name) + 1);
 409     }
 410     else {
 411         cmdArgs.args.create.nameLen = 0;
 412     }
 413
 414     /* Create the generic obj */
 415     obj = (MessageQ_Object *)calloc(1, sizeof (MessageQ_Object));
 416     if (obj == NULL) {
 417         PRINTVERBOSE0("MessageQ_create: memory allocation failed\n")
 418         return NULL;
 419     }
 420
 421     PRINTVERBOSE2("MessageQ_create: creating endpoint for: %s, \
 422        queueIndex: %d\n", name, queueIndex)
 423     status = transportCreateEndpoint(&obj->ipcFd, &queueIndex);
 424     if (status < 0) {
 425         goto cleanup;
 426     }
 427
 428     /*
 429      * We expect the endpoint creation to return a port number from
 430      * the MessageQCopy layer. This port number will be greater than
 431      * 1024 and less than 0x10000. Use this number as the queueIndex.
 432      */
 433     cmdArgs.args.create.queueId = queueIndex;
 434
 435     status = MessageQDrv_ioctl (CMD_MESSAGEQ_CREATE, &cmdArgs);
 436     if (status < 0) {
 437         PRINTVERBOSE1("MessageQ_create: API (through IOCTL) failed, \
 438             status=%d\n", status)
 439         goto cleanup;
 440     }
 441
 442    /* Populate the params member */
 443     memcpy(&obj->params, &ps, sizeof(ps));
 444
 445     procId = MultiProc_self();
 446     obj->queue = cmdArgs.args.create.queueId;
 447     obj->serverHandle = cmdArgs.args.create.handle;
 448
 449     /*
 450      * Now, to support MessageQ_unblock() functionality, create an event object.
 451      * Writing to this event will unblock the select() call in MessageQ_get().
 452      */
 453     if (pipe(fildes) == -1) {
 454         printf ("MessageQ_create: pipe creation failed: %d, %s\n",
 455                    errno, strerror(errno));
 456         status = MessageQ_E_FAIL;
 457         obj->unblockFdW = obj->unblockFdR = -1;
 458     }
 459     else {
 460         obj->unblockFdW = fildes[1];
 461         obj->unblockFdR = fildes[0];
 462     }
 463
 464 cleanup:
 465     /* Cleanup if fail: */
 466     if (status < 0) {
 467         MessageQ_delete((MessageQ_Handle *)&obj);
 468     }
 469
 470     return ((MessageQ_Handle) obj);
 471 }
 472
 473 /*
 474  * Function to delete a MessageQ object for a specific slave processor.
 475  *
 476  * Deletes the file descriptors associated with this MessageQ object.
 477  */
 478 Int MessageQ_delete (MessageQ_Handle * handlePtr)
 479 {
 480     Int               status    = MessageQ_S_SUCCESS;
 481     MessageQ_Object * obj       = NULL;
 482     MessageQDrv_CmdArgs cmdArgs;
 483
 484     assert(handlePtr != NULL);
 485     obj = (MessageQ_Object *) (*handlePtr);
 486     assert(obj != NULL);
 487
 488     if (obj->serverHandle != NULL) {
 489         cmdArgs.args.deleteMessageQ.handle = obj->serverHandle;
 490         status = MessageQDrv_ioctl (CMD_MESSAGEQ_DELETE, &cmdArgs);
 491         if (status < 0) {
 492             PRINTVERBOSE1("MessageQ_delete: API (through IOCTL) failed, \
 493                 status=%d\n", status)
 494         }
 495     }
 496
 497     /* Close the fds used for MessageQ_unblock(): */
 498     if (obj->unblockFdW >= 0) {
 499         close(obj->unblockFdW);
 500     }
 501     if (obj->unblockFdR >= 0) {
 502         close(obj->unblockFdR);
 503     }
 504
 505     /* Close the communication endpoint: */
 506     if (obj->ipcFd >= 0) {
 507         transportCloseEndpoint(obj->ipcFd);
 508     }
 509
 510     /* Now free the obj */
 511     free (obj);
 512     *handlePtr = NULL;
 513
 514     return (status);
 515 }
 516
 517 /*
 518  *  Opens an instance of MessageQ for sending.
 519  *
 520  *  We need not create a tiipc file descriptor here; the file descriptors for
 521  *  all remote processors were created during MessageQ_attach(), and will be
 522  *  retrieved during MessageQ_put().
 523  */
 524 Int MessageQ_open (String name, MessageQ_QueueId * queueId)
 525 {
 526     Int status = MessageQ_S_SUCCESS;
 527
 528     status = NameServer_getUInt32 (MessageQ_module->nameServer,
 529                                      name, queueId, NULL);
 530
 531     if (status == NameServer_E_NOTFOUND) {
 532         /* Set return queue ID to invalid. */
 533         *queueId = MessageQ_INVALIDMESSAGEQ;
 534         status = MessageQ_E_NOTFOUND;
 535     }
 536     else if (status >= 0) {
 537         /* Override with a MessageQ status code. */
 538         status = MessageQ_S_SUCCESS;
 539     }
 540     else {
 541         /* Set return queue ID to invalid. */
 542         *queueId = MessageQ_INVALIDMESSAGEQ;
 543         /* Override with a MessageQ status code. */
 544         if (status == NameServer_E_TIMEOUT) {
 545             status = MessageQ_E_TIMEOUT;
 546         }
 547         else {
 548             status = MessageQ_E_FAIL;
 549         }
 550     }
 551
 552     return (status);
 553 }
 554
 555 /* Closes previously opened instance of MessageQ module. */
 556 Int MessageQ_close (MessageQ_QueueId * queueId)
 557 {
 558     Int32 status = MessageQ_S_SUCCESS;
 559
 560     /* Nothing more to be done for closing the MessageQ. */
 561     *queueId = MessageQ_INVALIDMESSAGEQ;
 562
 563     return (status);
 564 }
 565
 566 /*
 567  * Place a message onto a message queue.
 568  *
 569  * Calls TransportShm_put(), which handles the sending of the message using the
 570  * appropriate kernel interface (socket, device ioctl) call for the remote
 571  * procId encoded in the queueId argument.
 572  *
 573  */
 574 Int MessageQ_put (MessageQ_QueueId queueId, MessageQ_Msg msg)
 575 {
 576     Int      status;
 577     UInt16   dstProcId  = (UInt16)(queueId >> 16);
 578     UInt16   queueIndex = (MessageQ_QueueIndex)(queueId & 0x0000ffff);
 579
 580     msg->dstId     = queueIndex;
 581     msg->dstProc   = dstProcId;
 582
 583     /* invoke put hook function after addressing the message */
 584     if (MessageQ_module->putHookFxn != NULL) {
 585         MessageQ_module->putHookFxn(queueId, msg);
 586     }
 587
 588     status = transportPut(msg, queueIndex, dstProcId);
 589
 590     return (status);
 591 }
 592
 593 /*
 594  * Gets a message for a message queue and blocks if the queue is empty.
 595  * If a message is present, it returns it.  Otherwise it blocks
 596  * waiting for a message to arrive.
 597  * When a message is returned, it is owned by the caller.
 598  *
 599  * We block using select() on the receiving tiipc file descriptor, then
 600  * get the waiting message via a read.
 601  * We use the file descriptors stored in the messageQ object via a previous
 602  * call to MessageQ_create().
 603  *
 604  * Note: We currently do not support messages to be sent between threads on the
 605  * lcoal processor.
 606  *
 607  */
 608 Int MessageQ_get (MessageQ_Handle handle, MessageQ_Msg * msg ,UInt timeout)
 609 {
 610     Int     status = MessageQ_S_SUCCESS;
 611     Int     tmpStatus;
 612     MessageQ_Object * obj = (MessageQ_Object *) handle;
 613     int     retval;
 614     int     nfds;
 615     fd_set  rfds;
 616     struct  timeval tv;
 617     void    *timevalPtr;
 618     int     maxfd = 0;
 619
 620     /* Wait (with timeout) and retreive message */
 621     FD_ZERO(&rfds);
 622     FD_SET(obj->ipcFd, &rfds);
 623     maxfd = obj->ipcFd;
 624
 625     /* Wait also on the event fd, which may be written by MessageQ_unblock(): */
 626     FD_SET(obj->unblockFdR, &rfds);
 627
 628     if (timeout == MessageQ_FOREVER) {
 629         timevalPtr = NULL;
 630     }
 631     else {
 632         /* Timeout given in msec: convert:  */
 633         tv.tv_sec = timeout / 1000;
 634         tv.tv_usec = (timeout % 1000) * 1000;
 635         timevalPtr = &tv;
 636     }
 637     /* Add one to last fd created: */
 638     nfds = ((maxfd > obj->unblockFdR) ? maxfd : obj->unblockFdR) + 1;
 639
 640     retval = select(nfds, &rfds, NULL, NULL, timevalPtr);
 641     if (retval)  {
 642         if (FD_ISSET(obj->unblockFdR, &rfds))  {
 643             /*
 644              * Our event was signalled by MessageQ_unblock().
 645              *
 646              * This is typically done during a shutdown sequence, where
 647              * the intention of the client would be to ignore (i.e. not fetch)
 648              * any pending messages in the transport's queue.
 649              * Thus, we shall not check for nor return any messages.
 650              */
 651             *msg = NULL;
 652             status = MessageQ_E_UNBLOCKED;
 653         }
 654         else {
 655             if (FD_ISSET(obj->ipcFd, &rfds)) {
 656                 /* Our transport's fd was signalled: Get the message: */
 657                 tmpStatus = transportGet(obj->ipcFd, msg);
 658                 if (tmpStatus < 0) {
 659                     printf ("MessageQ_get: tranposrtshm_get failed.");
 660                     status = MessageQ_E_FAIL;
 661                 }
 662             }
 663         }
 664     }
 665     else if (retval == 0) {
 666         *msg = NULL;
 667         status = MessageQ_E_TIMEOUT;
 668     }
 669
 670     return (status);
 671 }
 672
 673 /*
 674  * Return a count of the number of messages in the queue
 675  *
 676  * TBD: To be implemented. Return -1 for now.
 677  */
 678 Int MessageQ_count (MessageQ_Handle handle)
 679 {
 680     Int               count = -1;
 681     return (count);
 682 }
 683
 684 /* Initializes a message not obtained from MessageQ_alloc. */
 685 Void MessageQ_staticMsgInit (MessageQ_Msg msg, UInt32 size)
 686 {
 687     /* Fill in the fields of the message */
 688     MessageQ_msgInit (msg);
 689     msg->heapId  = MessageQ_STATICMSG;
 690     msg->msgSize = size;
 691 }
 692
 693 /*
 694  * Allocate a message and initialize the needed fields (note some
 695  * of the fields in the header are set via other APIs or in the
 696  * MessageQ_put function,
 697  */
 698 MessageQ_Msg MessageQ_alloc (UInt16 heapId, UInt32 size)
 699 {
 700     MessageQ_Msg msg       = NULL;
 701
 702     /*
 703      * heapId not used for local alloc (as this is over a copy transport), but
 704      * we need to send to other side as heapId is used in BIOS transport:
 705      */
 706     msg = (MessageQ_Msg)calloc (1, size);
 707     MessageQ_msgInit (msg);
 708     msg->msgSize = size;
 709     msg->heapId  = heapId;
 710
 711     return msg;
 712 }
 713
 714 /* Frees the message back to the heap that was used to allocate it. */
 715 Int MessageQ_free (MessageQ_Msg msg)
 716 {
 717     UInt32         status = MessageQ_S_SUCCESS;
 718
 719     /* Check to ensure this was not allocated by user: */
 720     if (msg->heapId == MessageQ_STATICMSG)  {
 721         status =  MessageQ_E_CANNOTFREESTATICMSG;
 722     }
 723     else {
 724         free (msg);
 725     }
 726
 727     return status;
 728 }
 729
 730 /* Register a heap with MessageQ. */
 731 Int MessageQ_registerHeap (Ptr heap, UInt16 heapId)
 732 {
 733     Int  status = MessageQ_S_SUCCESS;
 734
 735     /* Do nothing, as this uses a copy transport: */
 736
 737     return status;
 738 }
 739
 740 /* Unregister a heap with MessageQ. */
 741 Int MessageQ_unregisterHeap (UInt16 heapId)
 742 {
 743     Int  status = MessageQ_S_SUCCESS;
 744
 745     /* Do nothing, as this uses a copy transport: */
 746
 747     return status;
 748 }
 749
 750 /* Unblocks a MessageQ */
 751 Void MessageQ_unblock (MessageQ_Handle handle)
 752 {
 753     MessageQ_Object * obj   = (MessageQ_Object *) handle;
 754     char         buf = 'n';
 755     int          numBytes;
 756
 757     /* Write to pipe to awaken any threads blocked on this messageQ: */
 758     numBytes = write(obj->unblockFdW, &buf, 1);
 759 }
 760
 761 /* Embeds a source message queue into a message. */
 762 Void MessageQ_setReplyQueue (MessageQ_Handle handle, MessageQ_Msg msg)
 763 {
 764     MessageQ_Object * obj   = (MessageQ_Object *) handle;
 765
 766     msg->replyId   = (UInt16)(obj->queue);
 767     msg->replyProc = (UInt16)(obj->queue >> 16);
 768 }
 769
 770 /* Returns the QueueId associated with the handle. */
 771 MessageQ_QueueId MessageQ_getQueueId (MessageQ_Handle handle)
 772 {
 773     MessageQ_Object * obj = (MessageQ_Object *) handle;
 774     UInt32            queueId;
 775
 776     queueId = (obj->queue);
 777
 778     return queueId;
 779 }
 780
 781 /* Sets the tracing of a message */
 782 Void MessageQ_setMsgTrace (MessageQ_Msg msg, Bool traceFlag)
 783 {
 784     msg->flags = (msg->flags & ~TRACEMASK) |   (traceFlag << TRACESHIFT);
 785 }
 786
 787 /*
 788  *  Returns the amount of shared memory used by one transport instance.
 789  *
 790  *  The MessageQ module itself does not use any shared memory but the
 791  *  underlying transport may use some shared memory.
 792  */
 793 SizeT MessageQ_sharedMemReq (Ptr sharedAddr)
 794 {
 795     SizeT memReq = 0u;
 796
 797     /* Do nothing, as this is a copy transport. */
 798
 799     return (memReq);
 800 }
 801
 802 /*
 803  *  Opens a file descriptor for this remote proc.
 804  *
 805  *  Only opens it if one does not already exist for this procId.
 806  *
 807  *  Note: remoteProcId may be MultiProc_Self() for loopback case.
 808  */
 809 Int MessageQ_attach (UInt16 remoteProcId, Ptr sharedAddr)
 810 {
 811     Int     status = MessageQ_S_SUCCESS;
 812     UInt32  localAddr;
 813     int     ipcFd;
 814     int     err;
 815
 816     PRINTVERBOSE1("MessageQ_attach: remoteProcId: %d\n", remoteProcId)
 817
 818     if (remoteProcId >= MultiProc_MAXPROCESSORS) {
 819         status = MessageQ_E_INVALIDPROCID;
 820         goto exit;
 821     }
 822
 823     pthread_mutex_lock (&(MessageQ_module->gate));
 824
 825     /* Only open a fd if one doesn't exist: */
 826     if (MessageQ_module->ipcFd[remoteProcId] == -1)  {
 827         /* Create a fd for sending messages to the remote proc: */
 828         ipcFd = open("/dev/tiipc", O_RDWR);
 829         if (ipcFd < 0) {
 830             status = MessageQ_E_FAIL;
 831             printf ("MessageQ_attach: open of tiipc device failed: %d, %s\n",
 832                        errno, strerror(errno));
 833         }
 834         else  {
 835             PRINTVERBOSE1("MessageQ_attach: opened tiipc fd for sending: %d\n",
 836                 ipcFd)
 837             MessageQ_module->ipcFd[remoteProcId] = ipcFd;
 838             /*
 839              * Connect to the remote endpoint and bind any reserved address as
 840              * local endpoint
 841              */
 842             Connect(ipcFd, remoteProcId, MESSAGEQ_RPMSG_PORT);
 843             err = BindAddr(ipcFd, &localAddr);
 844             if (err < 0) {
 845                 status = MessageQ_E_FAIL;
 846                 printf ("MessageQ_attach: bind failed: %d, %s\n",
 847                     errno, strerror(errno));
 848             }
 849         }
 850     }
 851     else {
 852         status = MessageQ_E_ALREADYEXISTS;
 853     }
 854
 855     pthread_mutex_unlock (&(MessageQ_module->gate));
 856
 857 exit:
 858     return (status);
 859 }
 860
 861 /*
 862  *  Close the fd for this remote proc.
 863  *
 864  */
 865 Int MessageQ_detach (UInt16 remoteProcId)
 866 {
 867     Int status = MessageQ_S_SUCCESS;
 868     int ipcFd;
 869
 870     if (remoteProcId >= MultiProc_MAXPROCESSORS) {
 871         status = MessageQ_E_INVALIDPROCID;
 872         goto exit;
 873     }
 874
 875     pthread_mutex_lock (&(MessageQ_module->gate));
 876
 877     ipcFd = MessageQ_module->ipcFd[remoteProcId];
 878     if (close (ipcFd)) {
 879         status = MessageQ_E_OSFAILURE;
 880         printf("MessageQ_detach: close failed: %d, %s\n",
 881                        errno, strerror(errno));
 882     }
 883     else {
 884         PRINTVERBOSE1("MessageQ_detach: closed fd: %d\n", ipcFd)
 885         MessageQ_module->ipcFd[remoteProcId] = -1;
 886     }
 887
 888     pthread_mutex_unlock (&(MessageQ_module->gate));
 889
 890 exit:
 891     return (status);
 892 }
 893
 894 /*
 895  * This is a helper function to initialize a message.
 896  */
 897 Void MessageQ_msgInit (MessageQ_Msg msg)
 898 {
 899     msg->reserved0 = 0;  /* We set this to distinguish from NameServerMsg */
 900     msg->replyId   = (UInt16)MessageQ_INVALIDMESSAGEQ;
 901     msg->msgId     = MessageQ_INVALIDMSGID;
 902     msg->dstId     = (UInt16)MessageQ_INVALIDMESSAGEQ;
 903     msg->flags     = MessageQ_HEADERVERSION | MessageQ_NORMALPRI;
 904     msg->srcProc   = MultiProc_self();
 905
 906     pthread_mutex_lock(&(MessageQ_module->gate));
 907     msg->seqNum  = MessageQ_module->seqNum++;
 908     pthread_mutex_unlock(&(MessageQ_module->gate));
 909 }
 910
 911 /*
 912  * =============================================================================
 913  * Transport: Fxns kept here until need for a transport layer is realized.
 914  * =============================================================================
 915  */
 916 /*
 917  * ======== transportCreateEndpoint ========
 918  *
 919  * Create a communication endpoint to receive messages.
 920  */
 921 static Int transportCreateEndpoint(int * fd, UInt16 * queueIndex)
 922 {
 923     Int          status    = MessageQ_S_SUCCESS;
 924     int          err;
 925     UInt32       localAddr;
 926
 927     /* Create a fd to the ti-ipc to receive messages for this messageQ */
 928     *fd= open("/dev/tiipc", O_RDWR);
 929     if (*fd < 0) {
 930         status = MessageQ_E_FAIL;
 931         printf ("transportCreateEndpoint: Couldn't open tiipc device: %d, %s\n",
 932                   errno, strerror(errno));
 933
 934         goto exit;
 935     }
 936
 937     PRINTVERBOSE1("transportCreateEndpoint: opened fd: %d\n", *fd)
 938
 939     err = BindAddr(*fd, &localAddr);
 940     if (err < 0) {
 941         status = MessageQ_E_FAIL;
 942         printf("transportCreateEndpoint: bind failed: %d, %s\n",
 943                   errno, strerror(errno));
 944
 945         close(*fd);
 946         goto exit;
 947     }
 948
 949     if (localAddr >= MAX_LOCAL_ADDR) {
 950         status = MessageQ_E_FAIL;
 951         printf("transportCreateEndpoint: local address returned is"
 952             "by BindAddr is greater than max supported\n");
 953
 954         close(*fd);
 955         goto exit;
 956     }
 957
 958     *queueIndex = localAddr;
 959
 960 exit:
 961     return (status);
 962 }
 963
 964 /*
 965  * ======== transportCloseEndpoint ========
 966  *
 967  *  Close the communication endpoint.
 968  */
 969 static Int transportCloseEndpoint(int fd)
 970 {
 971     Int status = MessageQ_S_SUCCESS;
 972
 973     PRINTVERBOSE1("transportCloseEndpoint: closing fd: %d\n", fd)
 974
 975     /* Stop communication to this endpoint */
 976     close(fd);
 977
 978     return (status);
 979 }
 980
 981 /*
 982  * ======== transportGet ========
 983  *  Retrieve a message waiting in the queue.
 984 */
 985 static Int transportGet(int fd, MessageQ_Msg * retMsg)
 986 {
 987     Int           status    = MessageQ_S_SUCCESS;
 988     MessageQ_Msg  msg;
 989     int           ret;
 990     int           byteCount;
 991     tiipc_remote_params remote;
 992
 993     /*
 994      * We have no way of peeking to see what message size we'll get, so we
 995      * allocate a message of max size to receive contents from tiipc
 996      * (currently, a copy transport)
 997      */
 998     msg = MessageQ_alloc (0, MESSAGEQ_RPMSG_MAXSIZE);
 999     if (!msg)  {
1000         status = MessageQ_E_MEMORY;
1001         goto exit;
1002     }
1003
1004     /* Get message */
1005     byteCount = read(fd, msg, MESSAGEQ_RPMSG_MAXSIZE);
1006     if (byteCount < 0) {
1007         printf("read failed: %s (%d)\n", strerror(errno), errno);
1008         status = MessageQ_E_FAIL;
1009         goto exit;
1010     }
1011     else {
1012          /* Update the allocated message size (even though this may waste space
1013           * when the actual message is smaller than the maximum rpmsg size,
1014           * the message will be freed soon anyway, and it avoids an extra copy).
1015           */
1016          msg->msgSize = byteCount;
1017
1018          /*
1019           * If the message received was statically allocated, reset the
1020           * heapId, so the app can free it.
1021           */
1022          if (msg->heapId == MessageQ_STATICMSG)  {
1023              msg->heapId = 0;  /* for a copy transport, heap id is 0. */
1024          }
1025     }
1026
1027     PRINTVERBOSE1("transportGet: read from fd: %d\n", fd)
1028     ret = ioctl(fd, TIIPC_IOCGETREMOTE, &remote);
1029     PRINTVERBOSE3("\tReceived a msg: byteCount: %d, rpmsg addr: %d, rpmsg \
1030         proc: %d\n", byteCount, remote.remote_addr, remote.remote_proc)
1031     PRINTVERBOSE2("\tMessage Id: %d, Message size: %d\n", msg->msgId, msg->msgSize)
1032
1033     *retMsg = msg;
1034
1035 exit:
1036     return (status);
1037 }
1038
1039 /*
1040  * ======== transportPut ========
1041  *
1042  * Write to tiipc file descriptor associated with
1043  * with this destination procID.
1044  */
1045 static Int transportPut(MessageQ_Msg msg, UInt16 dstId, UInt16 dstProcId)
1046 {
1047     Int     status    = MessageQ_S_SUCCESS;
1048     int     ipcFd;
1049     int     err;
1050
1051     /*
1052      * Retrieve the tiipc file descriptor associated with this
1053      * transport for the destination processor.
1054      */
1055     ipcFd = MessageQ_module->ipcFd[dstProcId];
1056
1057     PRINTVERBOSE2("Sending msgId: %d via fd: %d\n", msg->msgId, ipcFd)
1058
1059     /* send response message to remote processor */
1060     err = write(ipcFd, msg, msg->msgSize);
1061     if (err < 0) {
1062         printf ("transportPut: write failed: %d, %s\n",
1063                   errno, strerror(errno));
1064         status = MessageQ_E_FAIL;
1065         goto exit;
1066     }
1067
1068     /*
1069      * Free the message, as this is a copy transport, we maintain MessageQ
1070      * semantics.
1071      */
1072     MessageQ_free (msg);
1073
1074 exit:
1075     return (status);
1076 }