Android: Fix compilation warnings
[ipc/ipcdev.git] / linux / src / transport / TransportRpmsg.c
1 /*
2  * Copyright (c) 2014-2015 Texas Instruments Incorporated - http://www.ti.com
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  *  ======== TransportRpmsg.c ========
34  *  Implementation of functions specified in the IMessageQTransport interface.
35  */
37 /* Socket Headers */
38 #include <sys/socket.h>
39 #include <sys/select.h>
40 #include <sys/eventfd.h>
41 #include <stdio.h>
42 #include <stdlib.h>
43 #include <unistd.h>
44 #include <errno.h>
45 #include <string.h>
46 #include <fcntl.h>
47 #include <pthread.h>
50 /* Socket Protocol Family */
51 #include <net/rpmsg.h>
54 /* IPC headers */
55 #include <ti/ipc/Std.h>
56 #include <SocketFxns.h>         /* Socket utils: */
57 #include <ti/ipc/Ipc.h>
58 #include <ti/ipc/MessageQ.h>
59 #include <ti/ipc/MultiProc.h>
60 #include <ti/ipc/transports/TransportRpmsg.h>
61 #include <_MessageQ.h>
62 #include <_lad.h>
64 #if !defined(EFD_SEMAPHORE)
65 #  define EFD_SEMAPHORE (1 << 0)
66 #endif
68 /* More magic rpmsg port numbers: */
69 #define MESSAGEQ_RPMSG_PORT       61
70 #define MESSAGEQ_RPMSG_MAXSIZE   512
72 #define TransportRpmsg_GROWSIZE 32
73 #define INVALIDSOCKET (-1)
75 #define TransportRpmsg_Event_ACK        (1 << 0)
76 #define TransportRpmsg_Event_PAUSE      (1 << 1)
77 #define TransportRpmsg_Event_CONTINUE   (1 << 2)
78 #define TransportRpmsg_Event_SHUTDOWN   (1 << 3)
81 #define _MAX(a,b) (((a)>(b))?(a):(b))
83 /* traces in this file are controlled via _TransportMessageQ_verbose */
84 Bool _TransportMessageQ_verbose = FALSE;
85 #define verbose _TransportMessageQ_verbose
87 Int TransportRpmsg_bind(Void *handle, UInt32 queueId);
88 Int TransportRpmsg_unbind(Void *handle, UInt32 queueId);
89 Bool TransportRpmsg_put(Void *handle, Ptr msg);
91 typedef struct TransportRpmsg_Module {
92     int             sock[MultiProc_MAXPROCESSORS];
93     fd_set          rfds;
94     int             maxFd;
95     struct {
96         int     fd;
97         UInt32  qId;
98     } inFds[1024];
99     int             nInFds;
100     pthread_mutex_t gate;
101     int             unblockEvent;    /* unblock the dispatch thread */
102     int             waitEvent;       /* block the client thread */
103     pthread_t       threadId;        /* ID returned by pthread_create() */
104     Bool            threadStarted;
106     TransportRpmsg_Handle *inst;    /* array of instances */
107 } TransportRpmsg_Module;
109 IMessageQTransport_Fxns TransportRpmsg_fxns = {
110     .bind    = TransportRpmsg_bind,
111     .unbind  = TransportRpmsg_unbind,
112     .put     = TransportRpmsg_put
113 };
115 typedef struct TransportRpmsg_Object {
116     IMessageQTransport_Object base;
117     Int status;
118     UInt16 rprocId;
119     int numQueues;
120     int *qIndexToFd;
121 } TransportRpmsg_Object;
123 TransportRpmsg_Module TransportRpmsg_state = {
124     .sock = {INVALIDSOCKET},
125     .unblockEvent = -1,
126     .waitEvent = -1,
127     .threadStarted = FALSE,
128     .inst = NULL
129 };
130 TransportRpmsg_Module *TransportRpmsg_module = &TransportRpmsg_state;
132 static void *rpmsgThreadFxn(void *arg);
133 static Int transportGet(int sock, MessageQ_Msg *retMsg);
134 static Void bindFdToQueueIndex(TransportRpmsg_Object *obj,
135                                Int fd,
136                                UInt16 qIndex);
137 static Void unbindQueueIndex(TransportRpmsg_Object *obj, UInt16 qIndex);
138 static Int queueIndexToFd(TransportRpmsg_Object *obj, UInt16 qIndex);
140 /* factory functions */
141 Int TransportRpmsg_Factory_create(Void);
142 Void TransportRpmsg_Factory_delete(Void);
143 Int TransportRpmsg_Factory_attach(UInt16 procId);
144 Int TransportRpmsg_Factory_detach(UInt16 procId);
146 Ipc_TransportFactoryFxns TransportRpmsg_Factory = {
147     .createFxn = TransportRpmsg_Factory_create,
148     .deleteFxn = TransportRpmsg_Factory_delete,
149     .attachFxn = TransportRpmsg_Factory_attach,
150     .detachFxn = TransportRpmsg_Factory_detach
151 };
153 /* -------------------------------------------------------------------------- */
155 /* instance convertors */
156 IMessageQTransport_Handle TransportRpmsg_upCast(TransportRpmsg_Handle handle)
158     TransportRpmsg_Object *obj = (TransportRpmsg_Object *)handle;
159     return ((IMessageQTransport_Handle)&obj->base);
162 TransportRpmsg_Handle TransportRpmsg_downCast(IMessageQTransport_Handle base)
164     return ((TransportRpmsg_Handle)base);
167 /*
168  *  ======== TransportRpmsg_create ========
169  */
170 TransportRpmsg_Handle TransportRpmsg_create(TransportRpmsg_Params *params)
172     Int status = MessageQ_S_SUCCESS;
173     TransportRpmsg_Object *obj = NULL;
174     int sock;
175     int flags;
176     UInt16 clusterId;
177     int i;
180     clusterId = params->rprocId - MultiProc_getBaseIdOfCluster();
182     /* create socket for sending messages to remote processor */
183     sock = socket(AF_RPMSG, SOCK_SEQPACKET, 0);
185     if (sock < 0) {
186         status = Ipc_E_FAIL;
187         fprintf(stderr,
188                 "TransportRpmsg_create: socket failed: %d (%s)\n", errno,
189                 strerror(errno));
190         goto done;
191     }
192     TransportRpmsg_module->sock[clusterId] = sock;
193     PRINTVERBOSE1("attach: created send socket: %d\n", sock)
195     status = ConnectSocket(sock, params->rprocId, MESSAGEQ_RPMSG_PORT);
197     if (status < 0) {
198         status = Ipc_E_FAIL;
199         fprintf(stderr,
200                 "TransportRpmsg_create: connect failed: %d (%s) procId: %d\n",
201                 errno, strerror(errno), params->rprocId);
202         close(sock);
203         TransportRpmsg_module->sock[clusterId] = INVALIDSOCKET;
204         goto done;
205     }
207     /* make sure socket fd doesn't exist for 'fork() -> exec*()'ed child */
208     flags = fcntl(sock, F_GETFD);
209     if (flags != -1) {
210         fcntl(sock, F_SETFD, flags | FD_CLOEXEC);
211     }
213     /* create the instance object */
214     obj = calloc(1, sizeof(TransportRpmsg_Object));
216     if (obj == NULL) {
217         status = Ipc_E_MEMORY;
218         close(sock);
219         TransportRpmsg_module->sock[clusterId] = INVALIDSOCKET;
220         goto done;
221     }
223     /* initialize the instance */
224     obj->base.base.interfaceType = IMessageQTransport_TypeId;
225     obj->base.fxns = &TransportRpmsg_fxns;
226     obj->rprocId = params->rprocId;
227     obj->numQueues = TransportRpmsg_GROWSIZE;
229     obj->qIndexToFd = calloc(TransportRpmsg_GROWSIZE, sizeof(int));
231     if (obj->qIndexToFd == NULL) {
232         status = Ipc_E_MEMORY;
233         goto done;
234     }
236     /* must initialize array */
237     for (i = 0; i < TransportRpmsg_GROWSIZE; i++) {
238         obj->qIndexToFd[i] = -1;
239     }
241 done:
242     if (status < 0) {
243         TransportRpmsg_delete((TransportRpmsg_Handle *)&obj);
244     }
246     return (TransportRpmsg_Handle)obj;
249 /*
250  *  ======== TransportRpmsg_delete ========
251  */
252 Void TransportRpmsg_delete(TransportRpmsg_Handle *pHandle)
254     TransportRpmsg_Object *obj = *(TransportRpmsg_Object **)pHandle;
255     UInt16 clusterId;
256     int sock;
258     if (obj == NULL) {
259         goto done;
260     }
262     clusterId = obj->rprocId - MultiProc_getBaseIdOfCluster();
264     /* close the socket for the given transport instance */
265     sock = TransportRpmsg_module->sock[clusterId];
266     if (sock != INVALIDSOCKET) {
267         PRINTVERBOSE1("detach: closing socket: %d\n", sock)
268         close(sock);
269     }
270     TransportRpmsg_module->sock[clusterId] = INVALIDSOCKET;
272     if ((obj != NULL) && (obj->qIndexToFd != NULL)) {
273         free(obj->qIndexToFd);
274         obj->qIndexToFd = NULL;
275     }
277     if (obj != NULL) {
278         free(obj);
279         obj = NULL;
280     }
282 done:
283     *pHandle = NULL;
286 /*
287  *  ======== TransportRpmsg_bind ========
288  */
289 Int TransportRpmsg_bind(Void *handle, UInt32 queueId)
291     TransportRpmsg_Object *obj = (TransportRpmsg_Object *)handle;
292     UInt16 queuePort = queueId & 0x0000ffff;
293     int fd;
294     int flags;
295     int err;
296     uint64_t event;
297     UInt16 rprocId;
298     pthread_t tid;
299     Int status = MessageQ_S_SUCCESS;
301     tid = pthread_self();
302     rprocId = obj->rprocId;
304     PRINTVERBOSE3("TransportRpmsg_bind: creating endpoint for rprocId %d "
305             "queuePort 0x%x, tid=0x%x\n", rprocId, queuePort, (unsigned int)tid)
307     pthread_mutex_lock(&TransportRpmsg_module->gate);
309     /*  Check if binding already exists.
310      *
311      *  There is a race condition between a thread calling MessageQ_create
312      *  and another thread calling Ipc_attach. Must make sure we don't bind
313      *  the same queue twice.
314      */
315     if (queueIndexToFd(obj, queueId) != -1) {
316         goto done;
317     }
319     /*  Create the socket to receive messages for this messageQ. */
320     fd = socket(AF_RPMSG, SOCK_SEQPACKET, 0);
321     if (fd < 0) {
322         fprintf(stderr, "TransportRpmsg_bind: socket call failed: %d (%s)\n",
323                 errno, strerror(errno));
324         status = MessageQ_E_OSFAILURE;
325         goto done;
326     }
327     PRINTVERBOSE2("TransportRpmsg_bind: created socket fd %d, tdi=0x%x\n", fd,
328             (unsigned int)tid);
330     err = SocketBindAddr(fd, rprocId, (UInt32)queuePort);
331     if (err < 0) {
332         /* don't hard-printf since this is no longer fatal */
333         PRINTVERBOSE2("TransportRpmsg_bind: bind failed: %d (%s)\n",
334                       errno, strerror(errno));
335         close(fd);
336         status = MessageQ_E_OSFAILURE;
337         goto done;
338     }
340     /* make sure socket fd doesn't exist for 'fork() -> exec*()'ed child */
341     flags = fcntl(fd, F_GETFD);
342     if (flags != -1) {
343         fcntl(fd, F_SETFD, flags | FD_CLOEXEC);
344     }
346     /*  pause the dispatch thread */
347     PRINTVERBOSE1("TransportRpmsg_bind: sending PAUSE event, tid=0x%x\n",
348             (unsigned int)tid);
349     event = TransportRpmsg_Event_PAUSE;
350     write(TransportRpmsg_module->unblockEvent, &event, sizeof(event));
352     /* wait for ACK event */
353     read(TransportRpmsg_module->waitEvent, &event, sizeof(event));
354     PRINTVERBOSE2("TransportRpmsg_bind: received ACK event (%d), tid=0x%x\n",
355             (int)event, (unsigned int)tid);
357     /* add to our fat fd array and update select() parameters */
358     TransportRpmsg_module->inFds[TransportRpmsg_module->nInFds].fd = fd;
359     TransportRpmsg_module->inFds[TransportRpmsg_module->nInFds++].qId = queueId;
360     TransportRpmsg_module->maxFd = _MAX(TransportRpmsg_module->maxFd, fd);
361     FD_SET(fd, &TransportRpmsg_module->rfds);
362     bindFdToQueueIndex(obj, fd, queuePort);
364     /* release the dispatch thread */
365     PRINTVERBOSE1("TransportRpmsg_bind: sending CONTINUE event, tid=0x%x\n",
366             (unsigned int)tid);
367     event = TransportRpmsg_Event_CONTINUE;
368     write(TransportRpmsg_module->unblockEvent, &event, sizeof(event));
370 done:
371     pthread_mutex_unlock(&TransportRpmsg_module->gate);
373     return (status);
376 /*
377  *  ======== TransportRpmsg_unbind ========
378  */
379 Int TransportRpmsg_unbind(Void *handle, UInt32 queueId)
381     TransportRpmsg_Object *obj = (TransportRpmsg_Object *)handle;
382     UInt16 queuePort = queueId & 0x0000ffff;
383     uint64_t event;
384     Int    status = MessageQ_S_SUCCESS;
385     int    maxFd;
386     int    fd;
387     int    i;
388     int    j;
390     pthread_mutex_lock(&TransportRpmsg_module->gate);
392     /*  pause the dispatch thread */
393     event = TransportRpmsg_Event_PAUSE;
394     write(TransportRpmsg_module->unblockEvent, &event, sizeof(event));
396     /* wait for ACK event */
397     read(TransportRpmsg_module->waitEvent, &event, sizeof(event));
399     /*  Check if binding already deleted.
400      *
401      *  There is a race condition between a thread calling MessageQ_delete
402      *  and another thread calling Ipc_detach. Must make sure we don't unbind
403      *  the same queue twice.
404      */
405     if ((fd = queueIndexToFd(obj, queuePort)) == -1) {
406         goto done;
407     }
408     PRINTVERBOSE1("TransportRpmsg_unbind: closing socket %d\n", fd)
410     /* guarenteed to work because queueIndexToFd above succeeded */
411     unbindQueueIndex(obj, queuePort);
413     /* remove from input fd array */
414     for (i = 0; i < TransportRpmsg_module->nInFds; i++) {
415         if (TransportRpmsg_module->inFds[i].fd == fd) {
416             TransportRpmsg_module->nInFds--;
418             /* shift subsequent elements down */
419             for (j = i; j < TransportRpmsg_module->nInFds; j++) {
420                 TransportRpmsg_module->inFds[j] =
421                         TransportRpmsg_module->inFds[j + 1];
422             }
423             TransportRpmsg_module->inFds[TransportRpmsg_module->nInFds].fd = -1;
424             TransportRpmsg_module->inFds[TransportRpmsg_module->nInFds].qId = 0;
425             break;
426         }
427     }
429     /* remove fd from the descriptor set, compute new max value */
430     FD_CLR(fd, &TransportRpmsg_module->rfds);
431     if (fd == TransportRpmsg_module->maxFd) {
432         /* find new max fd */
433         maxFd = TransportRpmsg_module->unblockEvent;
434         for (i = 0; i < TransportRpmsg_module->nInFds; i++) {
435             maxFd = _MAX(TransportRpmsg_module->inFds[i].fd, maxFd);
436         }
437         TransportRpmsg_module->maxFd = maxFd;
438     }
440     close(fd);
442     /* release the dispatch thread */
443     event = TransportRpmsg_Event_CONTINUE;
444     write(TransportRpmsg_module->unblockEvent, &event, sizeof(event));
446 done:
447     pthread_mutex_unlock(&TransportRpmsg_module->gate);
449     return (status);
452 /*
453  *  ======== TransportRpmsg_put ========
454  */
455 Bool TransportRpmsg_put(Void *handle, Ptr pmsg)
457     MessageQ_Msg msg  = (MessageQ_Msg)pmsg;
458     Int     status    = TRUE;
459     int     sock;
460     int     err;
461     UInt16  clusterId;
462     (Void)handle;
464     /*
465      * Retrieve the socket for the AF_SYSLINK protocol associated with this
466      * transport.
467      */
468     clusterId = msg->dstProc - MultiProc_getBaseIdOfCluster();
469     sock = TransportRpmsg_module->sock[clusterId];
470     if (sock == INVALIDSOCKET) {
471         return FALSE;
472     }
474     PRINTVERBOSE2("Sending msgId: %d via sock: %d\n", msg->msgId, sock)
476     err = send(sock, msg, msg->msgSize, 0);
477     if (err < 0) {
478         fprintf(stderr, "TransportRpmsg_put: send failed: %d (%s)\n",
479                 errno, strerror(errno));
480         status = FALSE;
482         goto exit;
483     }
485     /*
486      * Free the message, as this is a copy transport, we maintain MessageQ
487      * semantics.
488      */
489     MessageQ_free(msg);
491 exit:
492     return status;
495 /*
496  *  ======== TransportRpmsg_control ========
497  */
498 Bool TransportRpmsg_control(Void *handle, UInt cmd, UArg cmdArg)
500     (Void)handle;
501     (Void)cmd;
502     (Void)cmdArg;
503     return FALSE;
506 /*
507  *  ======== rpmsgThreadFxn ========
508  */
509 void *rpmsgThreadFxn(void *arg)
511     Int      status = MessageQ_S_SUCCESS;
512     Int      tmpStatus;
513     int      retval;
514     uint64_t event;
515     fd_set   rfds;
516     int      maxFd;
517     int      nfds;
518     MessageQ_Msg     retMsg = NULL;
519     MessageQ_QueueId queueId;
520     MessageQ_Handle handle;
521     Bool run = TRUE;
522     int i;
523     int j;
524     int fd;
525     (Void)arg;
527     while (run) {
528         maxFd = TransportRpmsg_module->maxFd;
529         rfds = TransportRpmsg_module->rfds;
530         nfds = TransportRpmsg_module->nInFds;
532         PRINTVERBOSE3("rpmsgThreadFxn: maxFd %d rfds[1:0] 0x%08x%08x\n", maxFd,
533                 (int)rfds.fds_bits[1], (int)rfds.fds_bits[0])
535         retval = select(maxFd + 1, &rfds, NULL, NULL, NULL);
537         /* if error, try again */
538         if (retval < 0) {
539             printf("Warning: rpmsgThreadFxn: select failed, trying again\n");
540             continue;
541         }
543         /* dispatch all pending messages, do this first */
544         for (i = 0; i < nfds; i++) {
545             fd = TransportRpmsg_module->inFds[i].fd;
547             if (FD_ISSET(fd, &rfds)) {
548                 PRINTVERBOSE1("rpmsgThreadFxn: getting from fd %d\n",
549                         TransportRpmsg_module->inFds[i].fd);
551                 /* transport input fd was signalled: get the message */
552                 tmpStatus = transportGet(fd, &retMsg);
553                 if (tmpStatus < 0 && tmpStatus != MessageQ_E_SHUTDOWN) {
554                     fprintf(stderr,
555                             "rpmsgThreadFxn: transportGet failed on fd %d, "
556                             "returned %d\n", fd, tmpStatus);
557                 }
558                 else if (tmpStatus == MessageQ_E_SHUTDOWN) {
559                     fprintf(stderr,
560                             "rpmsgThreadFxn: transportGet failed on fd %d, "
561                             "returned %d\n", fd, tmpStatus);
563                     retval = pthread_mutex_trylock(&TransportRpmsg_module->gate);
564                     if (retval != 0) {
565                         printf("Warning: rpmsgThreadFxn: "
566                                "unable to get the lock during shutdown, "
567                                "will try again\n");
568                         continue;
569                     }
571                     /*
572                      * Don't close(fd) at this time since it will get closed
573                      * later when MessageQ_delete() is called in response to
574                      * this failure.  Just remove fd's bit from the select mask
575                      * 'rfds' for now, but don't remove it from inFds[].
576                      */
577                     FD_CLR(fd, &TransportRpmsg_module->rfds);
578                     if (fd == TransportRpmsg_module->maxFd) {
579                         /* find new max fd */
580                         maxFd = TransportRpmsg_module->unblockEvent;
581                         for (j = 0; j < TransportRpmsg_module->nInFds; j++) {
582                             maxFd = _MAX(TransportRpmsg_module->inFds[j].fd,
583                                          maxFd);
584                         }
585                         TransportRpmsg_module->maxFd = maxFd;
586                     }
587                     queueId = TransportRpmsg_module->inFds[i].qId;
589                     pthread_mutex_unlock(&TransportRpmsg_module->gate);
591                     handle = MessageQ_getLocalHandle(queueId);
593                     PRINTVERBOSE2("rpmsgThreadFxn: shutting down MessageQ "
594                                   "%p (queueId 0x%x)...\n", handle, queueId)
596                     if (handle != NULL) {
597                         MessageQ_shutdown(handle);
598                     }
599                     else {
600                         fprintf(stderr,
601                                 "rpmsgThreadFxn: MessageQ_getLocalHandle(0x%x) "
602                                 "returned NULL, can't shutdown\n", queueId);
603                     }
604                 }
605                 else {
606                     queueId = MessageQ_getDstQueue(retMsg);
607                     PRINTVERBOSE1("rpmsgThreadFxn: got message, "
608                             "delivering to queueId 0x%x\n", queueId)
609                     MessageQ_put(queueId, retMsg);
610                 }
611             }
612         }
614         /* check for events */
615         if (FD_ISSET(TransportRpmsg_module->unblockEvent, &rfds)) {
617             read(TransportRpmsg_module->unblockEvent, &event, sizeof(event));
619             do {
620                 if (event & TransportRpmsg_Event_SHUTDOWN) {
621                     PRINTVERBOSE0("rpmsgThreadFxn: event SHUTDOWN\n");
622                     run = FALSE;
623                     break; /* highest priority, stop processing events */
624                 }
625                 if (event & TransportRpmsg_Event_CONTINUE) {
626                     PRINTVERBOSE1("rpmsgThreadFxn: event CONTINUE (%d)\n",
627                             (int)event);
628                     event &= ~TransportRpmsg_Event_CONTINUE;
629                 }
630                 if (event & TransportRpmsg_Event_PAUSE) {
631                     /*  Our event was signalled by TransportRpmsg_bind()
632                      *  or TransportRpmsg_unbind() to tell us that the set
633                      *  of file descriptors has changed.
634                      */
635                     PRINTVERBOSE0("rpmsgThreadFxn: event PAUSE\n");
636                     /* send the acknowledgement */
637                     event = TransportRpmsg_Event_ACK;
638                     write(TransportRpmsg_module->waitEvent, &event,
639                             sizeof(event));
640                     /* now wait to be released */
641                     read(TransportRpmsg_module->unblockEvent, &event,
642                             sizeof(event));
643                 }
644             } while (event != 0);
645         }
646     }
648     return (void *)status;
651 /*
652  * ======== transportGet ========
653  *  Retrieve a message waiting in the socket's queue.
654  */
655 static Int transportGet(int sock, MessageQ_Msg *retMsg)
657     Int           status    = MessageQ_S_SUCCESS;
658     MessageQ_Msg  msg;
659     struct sockaddr_rpmsg fromAddr;  /* [Socket address of sender] */
660     socklen_t     len;
661     int           byteCount;
663     /*
664      * We have no way of peeking to see what message size we'll get, so we
665      * allocate a message of max size to receive contents from the rpmsg socket
666      * (currently, a copy transport)
667      */
668     msg = MessageQ_alloc(0, MESSAGEQ_RPMSG_MAXSIZE);
669     if (!msg) {
670         status = MessageQ_E_MEMORY;
671         goto exit;
672     }
674     memset(&fromAddr, 0, sizeof (fromAddr));
675     len = sizeof (fromAddr);
677     byteCount = recvfrom(sock, msg, MESSAGEQ_RPMSG_MAXSIZE, 0,
678                          (struct sockaddr *)&fromAddr, &len);
679     if (len != sizeof (fromAddr)) {
680         fprintf(stderr, "recvfrom: got bad addr len (%d)\n", len);
681         status = MessageQ_E_FAIL;
682         goto freeMsg;
683     }
684     if (byteCount < 0) {
685         fprintf(stderr, "recvfrom failed: %s (%d)\n", strerror(errno), errno);
686         if (errno == ENOLINK) {
687             status = MessageQ_E_SHUTDOWN;
688         }
689         else {
690             status = MessageQ_E_FAIL;
691         }
692         goto freeMsg;
693     }
694     else {
695          /*
696           * Update the allocated message size (even though this may waste
697           * space when the actual message is smaller than the maximum rpmsg
698           * size, the message will be freed soon anyway, and it avoids an
699           * extra copy).
700           */
701          msg->msgSize = byteCount;
703          /* set the heapId in the message header to match allocation above */
704          msg->heapId = 0;
705     }
707     PRINTVERBOSE1("transportGet: recvfrom socket: fd: %d\n", sock)
708     PRINTVERBOSE3("\tReceived a msg: byteCount: %d, rpmsg addr: %d, rpmsg "
709             "proc: %d\n", byteCount, fromAddr.addr, fromAddr.vproc_id)
710     PRINTVERBOSE2("\tMessage Id: %d, Message size: %d\n", msg->msgId,
711             msg->msgSize)
713     *retMsg = msg;
715     goto exit;
717 freeMsg:
718     MessageQ_free(msg);
720 exit:
721     return status;
724 /*
725  *  ======== bindFdToQueueIndex ========
726  *
727  *  Precondition: caller must be inside the module gate
728  */
729 Void bindFdToQueueIndex(TransportRpmsg_Object *obj, Int fd, UInt16 queuePort)
731     Int *queues;
732     Int *oldQueues;
733     UInt oldSize;
734     UInt newCount;
735     UInt queueIndex;
736     UInt i;
738     /* subtract port offset from queue index */
739     queueIndex = queuePort - MessageQ_PORTOFFSET;
741     if (queueIndex >= (UInt)obj->numQueues) {
742         newCount = queueIndex + TransportRpmsg_GROWSIZE;
743         PRINTVERBOSE1("TransportRpmsg_bind: growing numQueues to %d\n",
744                 newCount);
746         /* allocate larger table */
747         oldSize = obj->numQueues * sizeof(int);
748         queues = calloc(newCount, sizeof(int));
750         /* copy contents from old table int new table */
751         memcpy(queues, obj->qIndexToFd, oldSize);
753         /* initialize remaining entries of new (larger) table */
754         for (i = obj->numQueues; i < newCount; i++) {
755             queues[i] = -1;
756         }
758         /* swap in new table, delete old table */
759         oldQueues = obj->qIndexToFd;
760         obj->qIndexToFd = queues;
761         obj->numQueues = newCount;
762         free(oldQueues);
763     }
765     /* add new entry */
766     obj->qIndexToFd[queueIndex] = fd;
769 /*
770  *  ======== unbindQueueIndex ========
771  *
772  *  Precondition: caller must be inside the module gate
773  */
774 Void unbindQueueIndex(TransportRpmsg_Object *obj, UInt16 queuePort)
776     UInt queueIndex;
778     /* subtract port offset from queue index */
779     queueIndex = queuePort - MessageQ_PORTOFFSET;
781     /* clear table entry */
782     obj->qIndexToFd[queueIndex] = -1;
785 /*
786  *  ======== queueIndexToFd ========
787  *
788  *  Precondition: caller must be inside the module gate
789  */
790 Int queueIndexToFd(TransportRpmsg_Object *obj, UInt16 queuePort)
792     UInt queueIndex;
794     /* subtract port offset from queue index */
795     queueIndex = queuePort - MessageQ_PORTOFFSET;
797     /* return file descriptor */
798     return (obj->qIndexToFd[queueIndex]);
801 /*
802  *  ======== TransportRpmsg_Factory_create ========
803  *  Create the transport instances
804  *
805  *  Attach to all remote processors. For now, must attach to
806  *  at least one to tolerate MessageQ_E_RESOURCE failures.
807  *
808  *  This function implements the IPC Factory interface, so it
809  *  returns Ipc status codes.
810  */
811 Int TransportRpmsg_Factory_create(Void)
813     Int status = Ipc_S_SUCCESS;
814     Int i;
815     UInt16 clusterSize;
816     TransportRpmsg_Handle *inst;
817     int flags;
820     /* needed to enumerate processors in cluster */
821     clusterSize = MultiProc_getNumProcsInCluster();
823     /* allocate the instance array */
824     inst = calloc(clusterSize, sizeof(TransportRpmsg_Handle));
826     if (inst == NULL) {
827         fprintf(stderr,
828                 "Error: TransportRpmsg_Factory_create failed, no memory\n");
829         status = Ipc_E_MEMORY;
830         goto done;
831     }
833     for (i = 0; i < clusterSize; i++) {
834         inst[i] = NULL;
835     }
837     TransportRpmsg_module->inst = inst;
839     /* counter event object for passing commands to dispatch thread */
840     TransportRpmsg_module->unblockEvent = eventfd(0, 0);
842     if (TransportRpmsg_module->unblockEvent == -1) {
843         fprintf(stderr, "create: unblock event failed: %d (%s)\n",
844                 errno, strerror(errno));
845         status = Ipc_E_FAIL;
846         goto done;
847     }
849     PRINTVERBOSE1("create: created unblock event %d\n",
850             TransportRpmsg_module->unblockEvent)
852     /* semaphore event object for acknowledging client thread */
853     TransportRpmsg_module->waitEvent = eventfd(0, EFD_SEMAPHORE);
855     if (TransportRpmsg_module->waitEvent == -1) {
856         fprintf(stderr,
857                 "create: wait event failed: %d (%s)\n", errno, strerror(errno));
858         status = Ipc_E_FAIL;
859         goto done;
860     }
862     PRINTVERBOSE1("create: created wait event %d\n",
863             TransportRpmsg_module->waitEvent)
865     /* make sure eventfds don't exist for 'fork() -> exec*()'ed child */
866     flags = fcntl(TransportRpmsg_module->waitEvent, F_GETFD);
867     if (flags != -1) {
868         fcntl(TransportRpmsg_module->waitEvent, F_SETFD, flags | FD_CLOEXEC);
869     }
870     flags = fcntl(TransportRpmsg_module->unblockEvent, F_GETFD);
871     if (flags != -1) {
872         fcntl(TransportRpmsg_module->unblockEvent, F_SETFD, flags | FD_CLOEXEC);
873     }
875     FD_ZERO(&TransportRpmsg_module->rfds);
876     FD_SET(TransportRpmsg_module->unblockEvent,
877             &TransportRpmsg_module->rfds);
878     TransportRpmsg_module->maxFd = TransportRpmsg_module->unblockEvent;
879     TransportRpmsg_module->nInFds = 0;
881     pthread_mutex_init(&TransportRpmsg_module->gate, NULL);
883     status = pthread_create(&TransportRpmsg_module->threadId, NULL,
884             &rpmsgThreadFxn, NULL);
886     if (status < 0) {
887         status = Ipc_E_FAIL;
888         fprintf(stderr, "create: failed to spawn thread\n");
889         goto done;
890     }
891     TransportRpmsg_module->threadStarted = TRUE;
893 done:
894     if (status < 0) {
895         TransportRpmsg_Factory_delete();
896     }
898     return (status);
901 /*
902  *  ======== TransportRpmsg_Factory_delete ========
903  *  Finalize the transport instances
904  */
905 Void TransportRpmsg_Factory_delete(Void)
907     uint64_t event;
910     /* shutdown the message dispatch thread */
911     if (TransportRpmsg_module->threadStarted) {
912         event = TransportRpmsg_Event_SHUTDOWN;
913         write(TransportRpmsg_module->unblockEvent, &event, sizeof(event));
915         /* wait for dispatch thread to exit */
916         pthread_join(TransportRpmsg_module->threadId, NULL);
917     }
919     /* destroy the mutex object */
920     pthread_mutex_destroy(&TransportRpmsg_module->gate);
922     /* close the client wait event */
923     if (TransportRpmsg_module->waitEvent != -1) {
924         close(TransportRpmsg_module->waitEvent);
925         TransportRpmsg_module->waitEvent = -1;
926     }
928     /* close the dispatch thread unblock event */
929     if (TransportRpmsg_module->unblockEvent != -1) {
930         close(TransportRpmsg_module->unblockEvent);
931         TransportRpmsg_module->unblockEvent = -1;
932     }
934     /* free the instance handle array */
935     if (TransportRpmsg_module->inst != NULL) {
936         free(TransportRpmsg_module->inst);
937         TransportRpmsg_module->inst = NULL;
938     }
940     return;
943 /*
944  *  ======== TransportRpmsg_Factory_attach ========
945  */
946 Int TransportRpmsg_Factory_attach(UInt16 procId)
948     Int status = Ipc_S_SUCCESS;
949     UInt16 clusterId;
950     TransportRpmsg_Params params;
951     TransportRpmsg_Handle transport;
952     IMessageQTransport_Handle iMsgQTrans;
954     /* cannot attach to yourself */
955     if (MultiProc_self() == procId) {
956         status = Ipc_E_INVALIDARG;
957         goto done;
958     }
960     /* processor must be a member of the cluster */
961     clusterId = procId - MultiProc_getBaseIdOfCluster();
963     if (clusterId >= MultiProc_getNumProcsInCluster()) {
964         status = Ipc_E_INVALIDARG;
965         goto done;
966     }
968     /* create transport instance for given processor */
969     params.rprocId = procId;
970     transport = TransportRpmsg_create(&params);
972     if (transport == NULL) {
973         status = Ipc_E_FAIL;
974         goto done;
975     }
977     /* register transport instance with MessageQ */
978     iMsgQTrans = TransportRpmsg_upCast(transport);
979     TransportRpmsg_module->inst[clusterId] = transport;
980     MessageQ_registerTransport(iMsgQTrans, procId, 0);
982 done:
983     return (status);
986 /*
987  *  ======== TransportRpmsg_Factory_detach ========
988  */
989 Int TransportRpmsg_Factory_detach(UInt16 procId)
991     Int status = Ipc_S_SUCCESS;
992     UInt16 clusterId;
994     /* cannot detach from yourself */
995     if (MultiProc_self() == procId) {
996         status = Ipc_E_INVALIDARG;
997         goto done;
998     }
1000     /* processor must be a member of the cluster */
1001     clusterId = procId - MultiProc_getBaseIdOfCluster();
1003     if (clusterId >= MultiProc_getNumProcsInCluster()) {
1004         status = Ipc_E_INVALIDARG;
1005         goto done;
1006     }
1008     /* must be attached in order to detach */
1009     if (TransportRpmsg_module->inst[clusterId] == NULL) {
1010         status = Ipc_E_INVALIDSTATE;
1011         goto done;
1012     }
1014     /* unregister from MessageQ, delete the transport instance */
1015     MessageQ_unregisterTransport(procId, 0);
1016     TransportRpmsg_delete(&(TransportRpmsg_module->inst[clusterId]));
1018 done:
1019     return (status);