# examples/PIPS/opencproducerconsumerex/src/consumer.c

Go to the documentation of this 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
    00021 #include "CommanHeader.h"
    00022
    00023 /*****************************************************************************
    00024 *  ConsumerThreadEntryPoint
    00025 *  Function: Consumer Thread that consumes as amny as needed items when
    00026 *  Producer has enough items produced
    00027 *  It also informs the Observer thread about item consumed
    00028 *******************************************************************************/
    00029
    00030 void* ConsumerThreadEntryPoint( void* aParam )
    00031 {
    00032         ThreadParam* args = (ThreadParam*) aParam;
    00033         int yetToConsume = args->noOfItems;
    00034         int clientId = args->noOfConsumers;
    00035         ProducedItem* consumeItem;
    00036         int ret = 0;
    00037         key_t msgQFd = -1;
    00038
    00039         /* Construct the message to be send thru msg q */
    00040         struct msgbuf* sendMsg = (struct msgbuf*)malloc(KMAXSENDMSG);
    00041         sendMsg->mtype = 1;
    00042
    00043         /* Get the Handler to Observer Msg Q */
    00044         msgQFd = msgget(KMSGQKEY, IPC_CREAT);
    00045
    00046         while (yetToConsume != 0)
    00047         {
    00048                 /* Acquire the Lock before checking for items available from Producer */
    00049                 sem_wait(&args->itemLock);
    00050                 /* Pop item from Producved Item-Stack */
    00051                 consumeItem = PopFromStack();
    00052                 if (consumeItem == NULL)
    00053                 {
    00054                         /* No items in the Stack */
    00055                         sem_post(&args->itemLock);
    00056                         continue;
    00057                 }
    00058                 yetToConsume--;
    00059                 /* Inform the Observer about Item Consumption */
    00060                 sprintf(&sendMsg->mtext, " Consumer#%d Consuming Produced %s\0", clientId, consumeItem->itemName);
    00061                 ret = msgsnd(msgQFd, sendMsg, strlen(sendMsg->mtext)+4, 0);
    00062
    00063                 /* Release the Lock once after getting an item */
    00064                 sem_post(&args->itemLock);
    00065                 FreeItem(consumeItem);
    00066         }
    00067
    00068         free( sendMsg );
    00069         return (int*)0;
    00070 }
    00071
    00072
    00073 /*  End of File */