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