TB9.2 Example Applications: examples/PIPS/opencproducerconsumerex/src/producer.c Source File

00001 
+00015 /* INCLUDE FILES */
+00016 #include <sys/types.h>
+00017 #include <sys/ipc.h>
+00018 #include <sys/msg.h>
+00019 #include <stdlib.h>
+00020 #include <stdio.h>
+00021 
+00022 #include "CommanHeader.h"
+00023 
+00024 #define MIN(X, Y) (X < Y) ? X : Y;
+00025 
+00026 /*****************************************************************************
+00027 *  ProducerThreadEntryPoint
+00028 *  Function: Producer Thread that produces M number of items at a time 
+00029 *  where M is number of Consumers.
+00030 *  This thread will produce N number of items where N is total number of items 
+00031 *  needed by all the Consumers.
+00032 *  It also informs the Observer thread about item production
+00033 *******************************************************************************/
+00034 
+00035 void* ProducerThreadEntryPoint( void* aParam ) 
+00036 {
+00037         ThreadParam* args = (ThreadParam*) aParam;
+00038         int yetToProduce = args->noOfItems;
+00039         int produceNow = 0;
+00040         int noOfConsumers = args->noOfConsumers;
+00041         ProducedItem item;
+00042         key_t msgQFd = -1;
+00043         int ret = 0;
+00044         /* Construct the message to be send thru msg q */
+00045         struct msgbuf* sendMsg = (struct msgbuf*)malloc(KMAXSENDMSG);
+00046         sendMsg->mtype = 1;
+00047         
+00048         /* Get the Handler to Observer Msg Q */
+00049         msgQFd = msgget(KMSGQKEY, IPC_CREAT);
+00050 
+00051         item.itemNum = 1;
+00052 
+00053         while (yetToProduce != 0) 
+00054         {
+00055                 produceNow = MIN(yetToProduce, noOfConsumers);
+00056                 yetToProduce -= produceNow;
+00057 
+00058                 /* Acquire the Lock so that Producer can Produce "produceNow" items once. */
+00059                 sem_wait(&args->itemLock);
+00060                 for(;produceNow != 0; produceNow--) 
+00061                 {
+00062                         sprintf(item.itemName, "Item-Number: #%d", item.itemNum);
+00063                         /* Push this item on to Stack so that Consumetrs can take it.*/
+00064                         PushOntoStack(&item);
+00065                         item.itemNum++;
+00066                         /* Inform the Observer about Item Production */
+00067                         sprintf(sendMsg->mtext, " Producer Produced %s\0", item.itemName);
+00068                         ret = msgsnd(msgQFd, sendMsg, strlen(sendMsg->mtext)+4, 0);
+00069                 }
+00070                 /* Release the Lock so that Consumers can consume produced items. */
+00071                 sem_post(&args->itemLock);
+00072         }
+00073 
+00074         free( sendMsg );
+00075         return (int*)0;
+00076 }
+00077 
+00078 /*  End of File */
+