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