qnx/src/tests/MessageQMulti/MessageQMulti.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   MessageQApp.c
  34  *
  35  *  @brief  Sample application for MessageQ module between MPU and Remote Proc
  36  *
  37  *  ============================================================================
  38  */
  39
  40 /* Standard headers */
  41 #include <pthread.h>
  42 #include <stdio.h>
  43 #include <errno.h>
  44 #include <string.h>
  45 #include <stdlib.h>
  46
  47 /* IPC Headers */
  48 #include <ti/ipc/Std.h>
  49 #include <ti/ipc/Ipc.h>
  50 #include <ti/ipc/MessageQ.h>
  51
  52 /* App defines: Must match on remote proc side: */
  53 #define MSGSIZE                     64u
  54 #define HEAPID                      0u
  55 #define SLAVE_MESSAGEQNAME          "SLAVE"
  56 #define HOST_MESSAGEQNAME           "HOST"
  57
  58 /** ============================================================================
  59  *  Macros and types
  60  *  ============================================================================
  61  */
  62
  63 #define  NUM_LOOPS_DFLT   1000
  64 #define  NUM_THREADS_DFLT 10
  65 #define  MAX_NUM_THREADS  55
  66 #define  ONE_PROCESS_ONLY (-1)
  67
  68 /** ============================================================================
  69  *  Globals
  70  *  ============================================================================
  71  */
  72 static Int     numLoops, numThreads, procNum;
  73
  74 struct thread_info {    /* Used as argument to thread_start() */
  75     pthread_t thread_id;        /* ID returned by pthread_create() */
  76     int       thread_num;       /* Application-defined thread # */
  77 };
  78
  79 static void * pingThreadFxn(void *arg);
  80
  81 /** ============================================================================
  82  *  Functions
  83  *  ============================================================================
  84  */
  85
  86 static Void * pingThreadFxn(void *arg)
  87 {
  88     Int                      threadNum = *(int *)arg;
  89     Int32                    status     = 0;
  90     MessageQ_Msg             msg        = NULL;
  91     MessageQ_Params          msgParams;
  92     UInt16                   i;
  93     MessageQ_Handle          handle;
  94     MessageQ_QueueId         queueId = MessageQ_INVALIDMESSAGEQ;
  95
  96     char             remoteQueueName[64];
  97     char             hostQueueName[64];
  98
  99     printf ("Entered pingThreadFxn: %d\n", threadNum);
 100
 101     sprintf(remoteQueueName, "%s_%d", SLAVE_MESSAGEQNAME, threadNum );
 102     sprintf(hostQueueName,   "%s_%d", HOST_MESSAGEQNAME,  threadNum );
 103
 104     /* Create the local Message Queue for receiving. */
 105     MessageQ_Params_init (&msgParams);
 106     handle = MessageQ_create (hostQueueName, &msgParams);
 107     if (handle == NULL) {
 108         printf ("Error in MessageQ_create\n");
 109         goto exit;
 110     }
 111     else {
 112         printf ("thread: %d, Local Message: %s, QId: 0x%x\n",
 113             threadNum, hostQueueName, MessageQ_getQueueId(handle));
 114     }
 115
 116     /* Poll until remote side has it's messageQ created before we send: */
 117     do {
 118         status = MessageQ_open (remoteQueueName, &queueId);
 119         sleep (1);
 120     } while (status == MessageQ_E_NOTFOUND);
 121     if (status < 0) {
 122         printf ("Error in MessageQ_open [0x%x]\n", status);
 123         goto cleanup;
 124     }
 125     else {
 126         printf ("thread: %d, Remote queue: %s, QId: 0x%x\n",
 127                  threadNum, remoteQueueName, queueId);
 128     }
 129
 130     printf ("\nthread: %d: Exchanging messages with remote processor...\n",
 131             threadNum);
 132     for (i = 0 ; i < numLoops ; i++) {
 133         /* Allocate message. */
 134         msg = MessageQ_alloc (HEAPID, MSGSIZE);
 135         if (msg == NULL) {
 136             printf ("Error in MessageQ_alloc\n");
 137             break;
 138         }
 139
 140         MessageQ_setMsgId (msg, i);
 141
 142         /* Have the remote proc reply to this message queue */
 143         MessageQ_setReplyQueue (handle, msg);
 144
 145         status = MessageQ_put (queueId, msg);
 146         if (status < 0) {
 147             printf ("Error in MessageQ_put [0x%x]\n", status);
 148             break;
 149         }
 150
 151         status = MessageQ_get(handle, &msg, MessageQ_FOREVER);
 152         if (status < 0) {
 153             printf ("Error in MessageQ_get [0x%x]\n", status);
 154             break;
 155         }
 156         else {
 157             /* Validate the returned message. */
 158             if ((msg != NULL) && (MessageQ_getMsgId (msg) != i)) {
 159                 printf ("Data integrity failure!\n"
 160                         "    Expected %d\n"
 161                         "    Received %d\n",
 162                         i, MessageQ_getMsgId (msg));
 163                 break;
 164             }
 165
 166             status = MessageQ_free (msg);
 167         }
 168
 169         printf ("thread: %d: Exchanged %d msgs\n", threadNum, (i+1));
 170     }
 171
 172     printf ("thread: %d: pingThreadFxn successfully completed!\n", threadNum);
 173
 174     MessageQ_close (&queueId);
 175
 176 cleanup:
 177     /* Clean-up */
 178     status = MessageQ_delete (&handle);
 179     if (status < 0) {
 180         printf ("Error in MessageQ_delete [0x%x]\n", status);
 181     }
 182
 183 exit:
 184
 185     return ((void *)status);
 186 }
 187
 188 int main (int argc, char ** argv)
 189 {
 190     struct thread_info threads[MAX_NUM_THREADS];
 191     int ret,i;
 192     void *res;
 193     Int32   status = 0;
 194
 195     /* Parse Args: */
 196     numLoops = NUM_LOOPS_DFLT;
 197     numThreads = NUM_THREADS_DFLT;
 198     procNum = ONE_PROCESS_ONLY;
 199     switch (argc) {
 200         case 1:
 201            /* use defaults */
 202            break;
 203         case 2:
 204            numThreads = atoi(argv[1]);
 205            break;
 206         case 3:
 207            numThreads = atoi(argv[1]);
 208            numLoops   = atoi(argv[2]);
 209            break;
 210         case 4:
 211            /* We force numThreads = 1 if doing a multiProcess test: */
 212            numThreads = 1;
 213            numLoops   = atoi(argv[2]);
 214            procNum = atoi(argv[3]);
 215            break;
 216         default:
 217            printf("Usage: %s [<numThreads>] [<numLoops>] [<Process #]>\n",
 218                argv[0]);
 219            printf("\tDefaults: numThreads: %d, numLoops: %d\n",
 220                NUM_THREADS_DFLT, NUM_LOOPS_DFLT);
 221            printf("\tMax Threads: %d\n", MAX_NUM_THREADS);
 222            exit(0);
 223     }
 224
 225     if (numThreads > MAX_NUM_THREADS) {
 226         printf("Error: Maximum number of threads supported is %d\n",
 227             MAX_NUM_THREADS);
 228         exit(EXIT_FAILURE);
 229     }
 230
 231     printf("Using numThreads: %d, numLoops: %d\n", numThreads, numLoops);
 232     if (procNum != ONE_PROCESS_ONLY) {
 233         printf("ProcNum: %d\n", procNum);
 234     }
 235
 236     status = Ipc_start();
 237     if (status < 0) {
 238         printf ("Ipc_start failed: status = 0x%x\n", status);
 239         goto exit;
 240     }
 241
 242     /* Launch multiple threads: */
 243     for (i = 0; i < numThreads; i++) {
 244         /* Create the test thread: */
 245         printf ("creating pingThreadFxn: %d\n", i);
 246         threads[i].thread_num = (procNum == ONE_PROCESS_ONLY)? i: procNum;
 247         ret = pthread_create(&threads[i].thread_id, NULL, &pingThreadFxn,
 248                            &(threads[i].thread_num));
 249         if (ret) {
 250             printf("MessageQMulti: can't spawn thread: %d, %s\n",
 251                     i, strerror(ret));
 252         }
 253     }
 254
 255     /* Join all threads: */
 256     for (i = 0; i < numThreads; i++) {
 257         ret = pthread_join(threads[i].thread_id, &res);
 258         if (ret != 0) {
 259             printf("MessageQMulti: failed to join thread #%d, %s\n",
 260                     threads[i].thread_num, strerror(ret));
 261         }
 262         else {
 263             printf("MessageQMulti: Joined with thread %d; "
 264                 "returned value was %p\n", threads[i].thread_num, res);
 265         }
 266         free(res);      /* Free memory allocated by thread */
 267     }
 268
 269     Ipc_stop();
 270
 271 exit:
 272
 273     return (0);
 274 }