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 <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 static Bool verbose = FALSE;
229 /* =============================================================================
230 * Globals
231 * =============================================================================
232 */
233 static MessageQ_ModuleObject MessageQ_state =
234 {
235 .refCount = 0,
236 .nameServer = NULL,
237 .putHookFxn = NULL
238 };
240 /*!
241 * @var MessageQ_module
242 *
243 * @brief Pointer to the MessageQ module state.
244 */
245 MessageQ_ModuleObject * MessageQ_module = &MessageQ_state;
248 /* =============================================================================
249 * Forward declarations of internal functions
250 * =============================================================================
251 */
253 /* This is a helper function to initialize a message. */
254 static Int transportCreateEndpoint(int * fd, UInt16 queueIndex);
255 static Int transportCloseEndpoint(int fd);
256 static Int transportGet(int fd, MessageQ_Msg * retMsg);
257 static Int transportPut(MessageQ_Msg msg, UInt16 dstId, UInt16 dstProcId);
259 /* =============================================================================
260 * APIS
261 * =============================================================================
262 */
263 /* Function to get default configuration for the MessageQ module.
264 *
265 */
266 Void MessageQ_getConfig (MessageQ_Config * cfg)
267 {
268 Int status;
269 MessageQDrv_CmdArgs cmdArgs;
271 assert (cfg != NULL);
273 cmdArgs.args.getConfig.config = cfg;
274 status = MessageQDrv_ioctl (CMD_MESSAGEQ_GETCONFIG, &cmdArgs);
276 if (status < 0) {
277 PRINTVERBOSE1("MessageQ_getConfig: API (through IOCTL) failed, \
278 status=%d\n", status)
279 }
281 return;
282 }
284 /* Function to setup the MessageQ module. */
285 Int MessageQ_setup (const MessageQ_Config * cfg)
286 {
287 Int status;
288 MessageQDrv_CmdArgs cmdArgs;
290 Int i;
292 cmdArgs.args.setup.config = (MessageQ_Config *) cfg;
293 status = MessageQDrv_ioctl(CMD_MESSAGEQ_SETUP, &cmdArgs);
294 if (status < 0) {
295 PRINTVERBOSE1("MessageQ_setup: API (through IOCTL) failed, \
296 status=%d\n", status)
297 return status;
298 }
300 MessageQ_module->nameServer = cmdArgs.args.setup.nameServerHandle;
301 MessageQ_module->seqNum = 0;
303 /* Create a default local gate. */
304 pthread_mutex_init (&(MessageQ_module->gate), NULL);
306 /* Clear ipcFd array. */
307 for (i = 0; i < MultiProc_MAXPROCESSORS; i++) {
308 MessageQ_module->ipcFd[i] = -1;
309 }
311 return status;
312 }
314 /*
315 * Function to destroy the MessageQ module.
316 */
317 Int MessageQ_destroy (void)
318 {
319 Int status;
320 MessageQDrv_CmdArgs cmdArgs;
322 status = MessageQDrv_ioctl (CMD_MESSAGEQ_DESTROY, &cmdArgs);
323 if (status < 0) {
324 PRINTVERBOSE1("MessageQ_destroy: API (through IOCTL) failed, \
325 status=%d\n", status)
326 }
328 return status;
329 }
331 /*
332 * ======== MessageQ_Params_init ========
333 * Legacy implementation.
334 */
335 Void MessageQ_Params_init(MessageQ_Params *params)
336 {
337 ((MessageQ_Params_Legacy *)params)->synchronizer = NULL;
338 }
340 /*
341 * ======== MessageQ_Params_init__S ========
342 * New implementation which is version aware.
343 */
344 Void MessageQ_Params_init__S(MessageQ_Params *params, Int version)
345 {
346 MessageQ_Params_Version2 *params2;
348 switch (version) {
350 case MessageQ_Params_VERSION_2:
351 params2 = (MessageQ_Params_Version2 *)params;
352 params2->__version = MessageQ_Params_VERSION_2;
353 params2->synchronizer = NULL;
354 params2->queueIndex = MessageQ_ANY;
355 break;
357 default:
358 assert(FALSE);
359 break;
360 }
361 }
363 /*
364 * Function to create a MessageQ object for receiving.
365 *
366 * Create a file descriptor and bind the source address
367 * (local ProcId/MessageQ ID) in
368 * order to get messages dispatched to this messageQ.
369 */
370 MessageQ_Handle MessageQ_create (String name, const MessageQ_Params * pp)
371 {
372 Int status = MessageQ_S_SUCCESS;
373 MessageQ_Object * obj = NULL;
374 UInt16 queuePort = 0u;
375 MessageQDrv_CmdArgs cmdArgs;
376 int fildes[2];
377 MessageQ_Params ps;
379 MessageQ_Params_init__S(&ps, MessageQ_Params_VERSION);
381 /* copy the given params into the current params structure */
382 if (pp != NULL) {
384 /* snoop the params pointer to see if it's a legacy structure */
385 if ((pp->__version == 0) || (pp->__version > 100)) {
386 ps.synchronizer = ((MessageQ_Params_Legacy *)pp)->synchronizer;
387 }
389 /* not legacy structure, use params version field */
390 else if (pp->__version == MessageQ_Params_VERSION_2) {
391 ps.__version = ((MessageQ_Params_Version2 *)pp)->__version;
392 ps.synchronizer = ((MessageQ_Params_Version2 *)pp)->synchronizer;
393 ps.queueIndex = ((MessageQ_Params_Version2 *)pp)->queueIndex;
394 }
395 else {
396 assert(FALSE);
397 }
398 }
400 cmdArgs.args.create.params = &ps;
401 cmdArgs.args.create.name = name;
403 if (name != NULL) {
404 cmdArgs.args.create.nameLen = (strlen (name) + 1);
405 }
406 else {
407 cmdArgs.args.create.nameLen = 0;
408 }
410 /* Create the generic obj */
411 obj = (MessageQ_Object *)calloc(1, sizeof (MessageQ_Object));
412 if (obj == NULL) {
413 PRINTVERBOSE0("MessageQ_create: memory allocation failed\n")
414 return NULL;
415 }
417 status = MessageQDrv_ioctl (CMD_MESSAGEQ_CREATE, &cmdArgs);
418 if (status < 0) {
419 PRINTVERBOSE1("MessageQ_create: API (through IOCTL) failed, \
420 status=%d\n", status)
421 goto cleanup;
422 }
424 /* Populate the params member */
425 memcpy(&obj->params, &ps, sizeof(ps));
427 obj->queue = cmdArgs.args.create.queueId;
428 obj->serverHandle = cmdArgs.args.create.handle;
430 /* Get the queue port # (queueIndex + PORT_OFFSET) */
431 queuePort = obj->queue & 0x0000FFFF;
433 PRINTVERBOSE2("MessageQ_create: creating endpoint for: %s"
434 "queuePort %d\n", (name == NULL) ? "NULL" : name , queuePort)
435 status = transportCreateEndpoint(&obj->ipcFd, queuePort);
436 if (status < 0) {
437 goto cleanup;
438 }
440 /*
441 * Now, to support MessageQ_unblock() functionality, create an event object.
442 * Writing to this event will unblock the select() call in MessageQ_get().
443 */
444 if (pipe(fildes) == -1) {
445 printf ("MessageQ_create: pipe creation failed: %d, %s\n",
446 errno, strerror(errno));
447 status = MessageQ_E_FAIL;
448 obj->unblockFdW = obj->unblockFdR = -1;
449 }
450 else {
451 obj->unblockFdW = fildes[1];
452 obj->unblockFdR = fildes[0];
453 }
455 cleanup:
456 /* Cleanup if fail: */
457 if (status < 0) {
458 MessageQ_delete((MessageQ_Handle *)&obj);
459 }
461 return ((MessageQ_Handle) obj);
462 }
464 /*
465 * Function to delete a MessageQ object for a specific slave processor.
466 *
467 * Deletes the file descriptors associated with this MessageQ object.
468 */
469 Int MessageQ_delete (MessageQ_Handle * handlePtr)
470 {
471 Int status = MessageQ_S_SUCCESS;
472 MessageQ_Object * obj = NULL;
473 MessageQDrv_CmdArgs cmdArgs;
475 assert(handlePtr != NULL);
476 obj = (MessageQ_Object *) (*handlePtr);
477 assert(obj != NULL);
479 if (obj->serverHandle != NULL) {
480 cmdArgs.args.deleteMessageQ.handle = obj->serverHandle;
481 status = MessageQDrv_ioctl (CMD_MESSAGEQ_DELETE, &cmdArgs);
482 if (status < 0) {
483 PRINTVERBOSE1("MessageQ_delete: API (through IOCTL) failed, \
484 status=%d\n", status)
485 }
486 }
488 /* Close the fds used for MessageQ_unblock(): */
489 if (obj->unblockFdW >= 0) {
490 close(obj->unblockFdW);
491 }
492 if (obj->unblockFdR >= 0) {
493 close(obj->unblockFdR);
494 }
496 /* Close the communication endpoint: */
497 if (obj->ipcFd >= 0) {
498 transportCloseEndpoint(obj->ipcFd);
499 }
501 /* Now free the obj */
502 free (obj);
503 *handlePtr = NULL;
505 return (status);
506 }
508 /*
509 * Opens an instance of MessageQ for sending.
510 *
511 * We need not create a tiipc file descriptor here; the file descriptors for
512 * all remote processors were created during MessageQ_attach(), and will be
513 * retrieved during MessageQ_put().
514 */
515 Int MessageQ_open (String name, MessageQ_QueueId * queueId)
516 {
517 Int status = MessageQ_S_SUCCESS;
519 status = NameServer_getUInt32 (MessageQ_module->nameServer,
520 name, queueId, NULL);
522 if (status == NameServer_E_NOTFOUND) {
523 /* Set return queue ID to invalid. */
524 *queueId = MessageQ_INVALIDMESSAGEQ;
525 status = MessageQ_E_NOTFOUND;
526 }
527 else if (status >= 0) {
528 /* Override with a MessageQ status code. */
529 status = MessageQ_S_SUCCESS;
530 }
531 else {
532 /* Set return queue ID to invalid. */
533 *queueId = MessageQ_INVALIDMESSAGEQ;
534 /* Override with a MessageQ status code. */
535 if (status == NameServer_E_TIMEOUT) {
536 status = MessageQ_E_TIMEOUT;
537 }
538 else {
539 status = MessageQ_E_FAIL;
540 }
541 }
543 return (status);
544 }
546 /*
547 * ======== MessageQ_openQueueId ========
548 */
549 MessageQ_QueueId MessageQ_openQueueId(UInt16 queueIndex, UInt16 procId)
550 {
551 MessageQ_QueueIndex queuePort;
552 MessageQ_QueueId queueId;
554 /* queue port is embedded in the queueId */
555 queuePort = queueIndex + MessageQ_PORTOFFSET;
556 queueId = ((MessageQ_QueueId)(procId) << 16) | queuePort;
558 return (queueId);
559 }
561 /* Closes previously opened instance of MessageQ module. */
562 Int MessageQ_close (MessageQ_QueueId * queueId)
563 {
564 Int32 status = MessageQ_S_SUCCESS;
566 /* Nothing more to be done for closing the MessageQ. */
567 *queueId = MessageQ_INVALIDMESSAGEQ;
569 return (status);
570 }
572 /*
573 * Place a message onto a message queue.
574 *
575 * Calls TransportShm_put(), which handles the sending of the message using the
576 * appropriate kernel interface (socket, device ioctl) call for the remote
577 * procId encoded in the queueId argument.
578 *
579 */
580 Int MessageQ_put (MessageQ_QueueId queueId, MessageQ_Msg msg)
581 {
582 Int status;
583 UInt16 dstProcId = (UInt16)(queueId >> 16);
584 UInt16 queuePort = queueId & 0x0000ffff;
586 /* use the queue port # for destination address */
587 msg->dstId = queuePort;
588 msg->dstProc = dstProcId;
590 /* invoke put hook function after addressing the message */
591 if (MessageQ_module->putHookFxn != NULL) {
592 MessageQ_module->putHookFxn(queueId, msg);
593 }
595 status = transportPut(msg, queuePort, dstProcId);
597 return (status);
598 }
600 /*
601 * Gets a message for a message queue and blocks if the queue is empty.
602 * If a message is present, it returns it. Otherwise it blocks
603 * waiting for a message to arrive.
604 * When a message is returned, it is owned by the caller.
605 *
606 * We block using select() on the receiving tiipc file descriptor, then
607 * get the waiting message via a read.
608 * We use the file descriptors stored in the messageQ object via a previous
609 * call to MessageQ_create().
610 *
611 * Note: We currently do not support messages to be sent between threads on the
612 * lcoal processor.
613 *
614 */
615 Int MessageQ_get (MessageQ_Handle handle, MessageQ_Msg * msg ,UInt timeout)
616 {
617 Int status = MessageQ_S_SUCCESS;
618 Int tmpStatus;
619 MessageQ_Object * obj = (MessageQ_Object *) handle;
620 int retval;
621 int nfds;
622 fd_set rfds;
623 struct timeval tv;
624 void *timevalPtr;
625 int maxfd = 0;
627 /* Wait (with timeout) and retreive message */
628 FD_ZERO(&rfds);
629 FD_SET(obj->ipcFd, &rfds);
630 maxfd = obj->ipcFd;
632 /* Wait also on the event fd, which may be written by MessageQ_unblock(): */
633 FD_SET(obj->unblockFdR, &rfds);
635 if (timeout == MessageQ_FOREVER) {
636 timevalPtr = NULL;
637 }
638 else {
639 /* Timeout given in msec: convert: */
640 tv.tv_sec = timeout / 1000;
641 tv.tv_usec = (timeout % 1000) * 1000;
642 timevalPtr = &tv;
643 }
644 /* Add one to last fd created: */
645 nfds = ((maxfd > obj->unblockFdR) ? maxfd : obj->unblockFdR) + 1;
647 retval = select(nfds, &rfds, NULL, NULL, timevalPtr);
648 if (retval) {
649 if (FD_ISSET(obj->unblockFdR, &rfds)) {
650 /*
651 * Our event was signalled by MessageQ_unblock().
652 *
653 * This is typically done during a shutdown sequence, where
654 * the intention of the client would be to ignore (i.e. not fetch)
655 * any pending messages in the transport's queue.
656 * Thus, we shall not check for nor return any messages.
657 */
658 *msg = NULL;
659 status = MessageQ_E_UNBLOCKED;
660 }
661 else {
662 if (FD_ISSET(obj->ipcFd, &rfds)) {
663 /* Our transport's fd was signalled: Get the message: */
664 tmpStatus = transportGet(obj->ipcFd, msg);
665 if (tmpStatus < 0) {
666 printf ("MessageQ_get: tranposrtshm_get failed.");
667 status = MessageQ_E_FAIL;
668 }
669 }
670 }
671 }
672 else if (retval == 0) {
673 *msg = NULL;
674 status = MessageQ_E_TIMEOUT;
675 }
677 return (status);
678 }
680 /*
681 * Return a count of the number of messages in the queue
682 *
683 * TBD: To be implemented. Return -1 for now.
684 */
685 Int MessageQ_count (MessageQ_Handle handle)
686 {
687 Int count = -1;
688 return (count);
689 }
691 /* Initializes a message not obtained from MessageQ_alloc. */
692 Void MessageQ_staticMsgInit (MessageQ_Msg msg, UInt32 size)
693 {
694 /* Fill in the fields of the message */
695 MessageQ_msgInit (msg);
696 msg->heapId = MessageQ_STATICMSG;
697 msg->msgSize = size;
698 }
700 /*
701 * Allocate a message and initialize the needed fields (note some
702 * of the fields in the header are set via other APIs or in the
703 * MessageQ_put function,
704 */
705 MessageQ_Msg MessageQ_alloc (UInt16 heapId, UInt32 size)
706 {
707 MessageQ_Msg msg = NULL;
709 /*
710 * heapId not used for local alloc (as this is over a copy transport), but
711 * we need to send to other side as heapId is used in BIOS transport:
712 */
713 msg = (MessageQ_Msg)calloc (1, size);
714 MessageQ_msgInit (msg);
715 msg->msgSize = size;
716 msg->heapId = heapId;
718 return msg;
719 }
721 /* Frees the message back to the heap that was used to allocate it. */
722 Int MessageQ_free (MessageQ_Msg msg)
723 {
724 UInt32 status = MessageQ_S_SUCCESS;
726 /* Check to ensure this was not allocated by user: */
727 if (msg->heapId == MessageQ_STATICMSG) {
728 status = MessageQ_E_CANNOTFREESTATICMSG;
729 }
730 else {
731 free (msg);
732 }
734 return status;
735 }
737 /* Register a heap with MessageQ. */
738 Int MessageQ_registerHeap (Ptr heap, UInt16 heapId)
739 {
740 Int status = MessageQ_S_SUCCESS;
742 /* Do nothing, as this uses a copy transport: */
744 return status;
745 }
747 /* Unregister a heap with MessageQ. */
748 Int MessageQ_unregisterHeap (UInt16 heapId)
749 {
750 Int status = MessageQ_S_SUCCESS;
752 /* Do nothing, as this uses a copy transport: */
754 return status;
755 }
757 /* Unblocks a MessageQ */
758 Void MessageQ_unblock (MessageQ_Handle handle)
759 {
760 MessageQ_Object * obj = (MessageQ_Object *) handle;
761 char buf = 'n';
763 /* Write to pipe to awaken any threads blocked on this messageQ: */
764 write(obj->unblockFdW, &buf, 1);
765 }
767 /* Embeds a source message queue into a message. */
768 Void MessageQ_setReplyQueue (MessageQ_Handle handle, MessageQ_Msg msg)
769 {
770 MessageQ_Object * obj = (MessageQ_Object *) handle;
772 msg->replyId = (UInt16)(obj->queue);
773 msg->replyProc = (UInt16)(obj->queue >> 16);
774 }
776 /* Returns the QueueId associated with the handle. */
777 MessageQ_QueueId MessageQ_getQueueId (MessageQ_Handle handle)
778 {
779 MessageQ_Object * obj = (MessageQ_Object *) handle;
780 UInt32 queueId;
782 queueId = (obj->queue);
784 return queueId;
785 }
787 /* Sets the tracing of a message */
788 Void MessageQ_setMsgTrace (MessageQ_Msg msg, Bool traceFlag)
789 {
790 msg->flags = (msg->flags & ~TRACEMASK) | (traceFlag << TRACESHIFT);
791 }
793 /*
794 * Returns the amount of shared memory used by one transport instance.
795 *
796 * The MessageQ module itself does not use any shared memory but the
797 * underlying transport may use some shared memory.
798 */
799 SizeT MessageQ_sharedMemReq (Ptr sharedAddr)
800 {
801 SizeT memReq = 0u;
803 /* Do nothing, as this is a copy transport. */
805 return (memReq);
806 }
808 /*
809 * Opens a file descriptor for this remote proc.
810 *
811 * Only opens it if one does not already exist for this procId.
812 *
813 * Note: remoteProcId may be MultiProc_Self() for loopback case.
814 */
815 Int MessageQ_attach (UInt16 remoteProcId, Ptr sharedAddr)
816 {
817 Int status = MessageQ_S_SUCCESS;
818 int ipcFd;
819 int err;
821 PRINTVERBOSE1("MessageQ_attach: remoteProcId: %d\n", remoteProcId)
823 if (remoteProcId >= MultiProc_MAXPROCESSORS) {
824 status = MessageQ_E_INVALIDPROCID;
825 goto exit;
826 }
828 pthread_mutex_lock (&(MessageQ_module->gate));
830 /* Only open a fd if one doesn't exist: */
831 if (MessageQ_module->ipcFd[remoteProcId] == -1) {
832 /* Create a fd for sending messages to the remote proc: */
833 ipcFd = open("/dev/tiipc", O_RDWR);
834 if (ipcFd < 0) {
835 status = MessageQ_E_FAIL;
836 printf ("MessageQ_attach: open of tiipc device failed: %d, %s\n",
837 errno, strerror(errno));
838 }
839 else {
840 PRINTVERBOSE1("MessageQ_attach: opened tiipc fd for sending: %d\n",
841 ipcFd)
842 MessageQ_module->ipcFd[remoteProcId] = ipcFd;
843 /*
844 * Connect to the remote endpoint and bind any reserved address as
845 * local endpoint
846 */
847 Connect(ipcFd, remoteProcId, MESSAGEQ_RPMSG_PORT);
848 /* Bind to any port # above 1024 (MessageQCopy_MAXRESERVEDEPT) */
849 err = BindAddr(ipcFd, TIIPC_ADDRANY);
850 if (err < 0) {
851 status = MessageQ_E_FAIL;
852 printf ("MessageQ_attach: bind failed: %d, %s\n",
853 errno, strerror(errno));
854 }
855 }
856 }
857 else {
858 status = MessageQ_E_ALREADYEXISTS;
859 }
861 pthread_mutex_unlock (&(MessageQ_module->gate));
863 exit:
864 return (status);
865 }
867 /*
868 * Close the fd for this remote proc.
869 *
870 */
871 Int MessageQ_detach (UInt16 remoteProcId)
872 {
873 Int status = MessageQ_S_SUCCESS;
874 int ipcFd;
876 if (remoteProcId >= MultiProc_MAXPROCESSORS) {
877 status = MessageQ_E_INVALIDPROCID;
878 goto exit;
879 }
881 pthread_mutex_lock (&(MessageQ_module->gate));
883 ipcFd = MessageQ_module->ipcFd[remoteProcId];
884 if (close (ipcFd)) {
885 status = MessageQ_E_OSFAILURE;
886 printf("MessageQ_detach: close failed: %d, %s\n",
887 errno, strerror(errno));
888 }
889 else {
890 PRINTVERBOSE1("MessageQ_detach: closed fd: %d\n", ipcFd)
891 MessageQ_module->ipcFd[remoteProcId] = -1;
892 }
894 pthread_mutex_unlock (&(MessageQ_module->gate));
896 exit:
897 return (status);
898 }
900 /*
901 * This is a helper function to initialize a message.
902 */
903 Void MessageQ_msgInit (MessageQ_Msg msg)
904 {
905 msg->reserved0 = 0; /* We set this to distinguish from NameServerMsg */
906 msg->replyId = (UInt16)MessageQ_INVALIDMESSAGEQ;
907 msg->msgId = MessageQ_INVALIDMSGID;
908 msg->dstId = (UInt16)MessageQ_INVALIDMESSAGEQ;
909 msg->flags = MessageQ_HEADERVERSION | MessageQ_NORMALPRI;
910 msg->srcProc = MultiProc_self();
912 pthread_mutex_lock(&(MessageQ_module->gate));
913 msg->seqNum = MessageQ_module->seqNum++;
914 pthread_mutex_unlock(&(MessageQ_module->gate));
915 }
917 /*
918 * =============================================================================
919 * Transport: Fxns kept here until need for a transport layer is realized.
920 * =============================================================================
921 */
922 /*
923 * ======== transportCreateEndpoint ========
924 *
925 * Create a communication endpoint to receive messages.
926 */
927 static Int transportCreateEndpoint(int * fd, UInt16 queuePort)
928 {
929 Int status = MessageQ_S_SUCCESS;
930 int err;
932 /* Create a fd to the ti-ipc to receive messages for this messageQ */
933 *fd= open("/dev/tiipc", O_RDWR);
934 if (*fd < 0) {
935 status = MessageQ_E_FAIL;
936 printf ("transportCreateEndpoint: Couldn't open tiipc device: %d, %s\n",
937 errno, strerror(errno));
939 goto exit;
940 }
942 PRINTVERBOSE1("transportCreateEndpoint: opened fd: %d\n", *fd)
944 /* Bind to this port # in the transport */
945 err = BindAddr(*fd, (UInt32)queuePort);
946 if (err < 0) {
947 status = MessageQ_E_FAIL;
948 printf("transportCreateEndpoint: bind failed: %d, %s\n",
949 errno, strerror(errno));
951 close(*fd);
952 goto exit;
953 }
955 exit:
956 return (status);
957 }
959 /*
960 * ======== transportCloseEndpoint ========
961 *
962 * Close the communication endpoint.
963 */
964 static Int transportCloseEndpoint(int fd)
965 {
966 Int status = MessageQ_S_SUCCESS;
968 PRINTVERBOSE1("transportCloseEndpoint: closing fd: %d\n", fd)
970 /* Stop communication to this endpoint */
971 close(fd);
973 return (status);
974 }
976 /*
977 * ======== transportGet ========
978 * Retrieve a message waiting in the queue.
979 */
980 static Int transportGet(int fd, MessageQ_Msg * retMsg)
981 {
982 Int status = MessageQ_S_SUCCESS;
983 MessageQ_Msg msg;
984 int ret;
985 int byteCount;
986 tiipc_remote_params remote;
988 /*
989 * We have no way of peeking to see what message size we'll get, so we
990 * allocate a message of max size to receive contents from tiipc
991 * (currently, a copy transport)
992 */
993 msg = MessageQ_alloc (0, MESSAGEQ_RPMSG_MAXSIZE);
994 if (!msg) {
995 status = MessageQ_E_MEMORY;
996 goto exit;
997 }
999 /* Get message */
1000 byteCount = read(fd, msg, MESSAGEQ_RPMSG_MAXSIZE);
1001 if (byteCount < 0) {
1002 printf("read failed: %s (%d)\n", strerror(errno), errno);
1003 status = MessageQ_E_FAIL;
1004 goto exit;
1005 }
1006 else {
1007 /* Update the allocated message size (even though this may waste space
1008 * when the actual message is smaller than the maximum rpmsg size,
1009 * the message will be freed soon anyway, and it avoids an extra copy).
1010 */
1011 msg->msgSize = byteCount;
1013 /*
1014 * If the message received was statically allocated, reset the
1015 * heapId, so the app can free it.
1016 */
1017 if (msg->heapId == MessageQ_STATICMSG) {
1018 msg->heapId = 0; /* for a copy transport, heap id is 0. */
1019 }
1020 }
1022 PRINTVERBOSE1("transportGet: read from fd: %d\n", fd)
1023 ret = ioctl(fd, TIIPC_IOCGETREMOTE, &remote);
1024 if (ret == -1) {
1025 printf("ioctl failed: %s (%d)\n", strerror(errno), errno);
1026 status = MessageQ_E_FAIL;
1027 goto exit;
1028 }
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,
1032 msg->msgSize)
1034 *retMsg = msg;
1036 exit:
1037 return (status);
1038 }
1040 /*
1041 * ======== transportPut ========
1042 *
1043 * Write to tiipc file descriptor associated with
1044 * with this destination procID.
1045 */
1046 static Int transportPut(MessageQ_Msg msg, UInt16 dstId, UInt16 dstProcId)
1047 {
1048 Int status = MessageQ_S_SUCCESS;
1049 int ipcFd;
1050 int err;
1052 /*
1053 * Retrieve the tiipc file descriptor associated with this
1054 * transport for the destination processor.
1055 */
1056 ipcFd = MessageQ_module->ipcFd[dstProcId];
1058 PRINTVERBOSE2("Sending msgId: %d via fd: %d\n", msg->msgId, ipcFd)
1060 /* send response message to remote processor */
1061 err = write(ipcFd, msg, msg->msgSize);
1062 if (err < 0) {
1063 printf ("transportPut: write failed: %d, %s\n",
1064 errno, strerror(errno));
1065 status = MessageQ_E_FAIL;
1066 goto exit;
1067 }
1069 /*
1070 * Free the message, as this is a copy transport, we maintain MessageQ
1071 * semantics.
1072 */
1073 MessageQ_free (msg);
1075 exit:
1076 return (status);
1077 }