Question

In c programming

1. The Consumer

2. Submits processing requests to the producer by supplying a file name, its location and a character. It also outputs the contents of the file provided by the producer to the standard output.

3. The Producer
   Accepts multiple consumer requests and processes each request by creating the following four threads.

   1. The reader thread will read an input file, one line at a time. It will pass each line of input to the character thread through a queue of messages.

   2. The character thread component will scan the line and replace each blank character by the character supplied by the client. It will then pass the line to the toUpper thread through another queue of messages.

   3. The toUpper thread component will scan the line and convert all lower case letters to upper case. It will then pass the converted line to the writer thread through another queue of messages.

   4. The writer thread will write the line to an output file.

   5. The producer will also return both the file name as well as its location to the consumer when the end of input file is reached.

      Implementation Details

   6. You should develop a module that implements a queue of character string buffers.

   7. This structure will be an array of pointers to strings with integers to indicate the head and tail of the list.

   8. The maximum size of the buffer array will be 10.

   9. Buffers will be created by the reader thread and destroyed by the writer thread.

   10. Threads should terminate when end of input file is reached.

This is what I have so far. It compiles but doesn't work. Need help!

#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <string.h>
#include <fcntl.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <pthread.h>
#include <limits.h>
#define MAX 10

// Declaring global variables
int consumerCount = 0;
int producerCount1 = 0;
int producerCount2 = 0;
int producerCount3 = 0;
int producerCount4 = 0;
char line[1000];
char mychar;

// A structure to represent a queue
struct element {
    char *name;
    struct element *next;
};

struct element *tail;

void init_queue (void);
void enqueue (char *name);
int dequeue (char *name);
void print_queue (void);
void error (char *msg);
void* consume();
void* reader();
void* character();
void* toUpper();
void* writer();

void init_queue (void)
{
    tail = NULL;
}

void enqueue (char *name)
{
    struct element *ptr;
    char *cp;

    if ((ptr = (struct element *) malloc (sizeof (struct element))) == NULL)
        error ("malloc");
    if ((cp = (char *) malloc (strlen (name) + 1)) == NULL)
        error ("malloc");

    strcpy (cp, name);
    ptr -> name = cp;

    if (tail == NULL) {
        ptr -> next = ptr;
    }
    else
    {
        ptr -> next = tail -> next;
        tail -> next = ptr;
    }
    tail = ptr;
}

int dequeue (char *name) // returns -1 on error
{
    struct element *ptr;
    char *cp;

    if (!tail) {
        fprintf (stderr, "Queue is empty\n");
        return -1;
    }
    // get the head
    ptr = tail -> next;
    cp = ptr -> name;

    if (ptr == tail)
        tail = NULL;
    else
        tail -> next = ptr -> next;
    free (ptr);
    strcpy (name, cp);
    free (cp);
    return 0;
}

int main(int argc, char **argv){
FILE* fp = fopen(argv[1], "r");

// Getting the mutex
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

pthread_cond_t dataNotProduced =
                   PTHREAD_COND_INITIALIZER;
pthread_cond_t dataNotConsumed =
                   PTHREAD_COND_INITIALIZER;

void* consume(void* arg){
    while (1){
        pthread_mutex_lock(&mutex);
    printf("Enter character: ");
    scanf("%c", &mychar);
    consumerCount++;
    }
    pthread_mutex_unlock(&mutex);
}

void* reader(void* arg)
{
  

  
       // Getting the lock on queue using mutex
       pthread_mutex_lock(&mutex);

       if (consumerCount < MAX)
       {
           //while (!feof(fp)) {
              if (NULL != fgets(line, 1000, fp)) {
              enqueue(line);                                      
        }
   }

       // Get the mutex unlocked
       pthread_mutex_unlock(&mutex);
   }

// Function definition for consumer thread B
void* character(void* arg)
{

   while (1) {

       // Getting the lock on queue using mutex
       pthread_mutex_lock(&mutex);

        dequeue(line);
        for(int i = 0; i < strlen(line); i++){
            if(line[i] == ' '){
                line[i] = mychar;
            }
        enqueue(line);
        printf("%s\n", line);
        }
      
          

       // Get the mutex unlocked
       pthread_mutex_unlock(&mutex);
   }

// Function definition for consumer thread C
void* toUpper()
{

   while (1) {

       // Getting the lock on queue using mutex
       pthread_mutex_lock(&mutex);
      
            dequeue(line);
            for(int i = 0; i < strlen(line); i++){
                toUpper(line[i]);
            }
            enqueue(line);

           producerCount3++;
          
           // Get the mutex unlocked
           pthread_mutex_unlock(&mutex);
       }

  

  
   }
}
void* writer(void* arg)
{

   while (1) {

       // Getting the lock on queue using mutex
       pthread_mutex_lock(&mutex);
      
       dequeue(line);
       fprintf(fp, "%s\n", line);

       // Get the mutex unlocked
       pthread_mutex_unlock(&mutex);
   }

init_queue();

   // Declaring integers used to
   // identify the thread in the system
   pthread_t consumerThread, producerThread1, producerThread2, producerThread3, producerThread4;

   //Function to create a threads
   //(pthread_create() takes 4 arguments)
   printf("hello\n");
   int retConsumer = pthread_create(&consumerThread,
                   NULL, consume, NULL);
   int retProducer1 = pthread_create(&producerThread1,
                   NULL, reader, NULL);
   int retProducer2 = pthread_create(&producerThread2,
                   NULL, character, NULL);
   int retProducer3 = pthread_create(&producerThread3,
                   NULL, toUpper, NULL);
   int retProducer4 = pthread_create(&producerThread4,
                   NULL, writer, NULL);

   // pthread_join suspends execution of the calling
   // thread until the target thread terminates
   if (!retConsumer)
       pthread_join(consumerThread, NULL);
   if (!retProducer1)
       pthread_join(producerThread1, NULL);
   if (!retProducer2)
       pthread_join(producerThread2, NULL);
   if (!retProducer3)
       pthread_join(producerThread3, NULL);
    if (!retProducer4)
       pthread_join(producerThread4, NULL);
  
}
}

Answer 1

Note : This code is my own code according to your problem, Please try the following code

Code :

/* buffer.h */
typedef int buffer_item;
#define BUFFER_SIZE 5

/* main.c */

#include <stdlib.h>
#include <stdio.h>
#include <pthread.h>
#include <semaphore.h>
#include "buffer.h"

#define RAND_DIVISOR 100000000
#define TRUE 1

/* The mutex lock */
pthread_mutex_t mutex;

/* the semaphores */
sem_t full, empty;

/* the buffer */
buffer_item buffer[BUFFER_SIZE];

/* buffer counter */
int counter;

pthread_t tid;       //Thread ID
pthread_attr_t attr; //Set of thread attributes

void *producer(void *param); /* the producer thread */
void *consumer(void *param); /* the consumer thread */

void initializeData() {

   /* Create the mutex lock */
   pthread_mutex_init(&mutex, NULL);

   /* Create the full semaphore and initialize to 0 */
   sem_init(&full, 0, 0);

   /* Create the empty semaphore and initialize to BUFFER_SIZE */
   sem_init(&empty, 0, BUFFER_SIZE);

   /* Get the default attributes */
   pthread_attr_init(&attr);

   /* init buffer */
   counter = 0;
}

/* Producer Thread */
void *producer(void *param) {
   buffer_item item;

   while(TRUE) {
      /* sleep for a random period of time */
      int rNum = rand() / RAND_DIVISOR;
      sleep(rNum);

      /* generate a random number */
      item = rand();

      /* acquire the empty lock */
      sem_wait(&empty);
      /* acquire the mutex lock */
      pthread_mutex_lock(&mutex);

      if(insert_item(item)) {
         fprintf(stderr, " Producer report error condition\n");
      }
      else {
         printf("producer produced %d\n", item);
      }
      /* release the mutex lock */
      pthread_mutex_unlock(&mutex);
      /* signal full */
      sem_post(&full);
   }
}

/* Consumer Thread */
void *consumer(void *param) {
   buffer_item item;

   while(TRUE) {
      /* sleep for a random period of time */
      int rNum = rand() / RAND_DIVISOR;
      sleep(rNum);

      /* aquire the full lock */
      sem_wait(&full);
      /* aquire the mutex lock */
      pthread_mutex_lock(&mutex);
      if(remove_item(&item)) {
         fprintf(stderr, "Consumer report error condition\n");
      }
      else {
         printf("consumer consumed %d\n", item);
      }
      /* release the mutex lock */
      pthread_mutex_unlock(&mutex);
      /* signal empty */
      sem_post(&empty);
   }
}

/* Add an item to the buffer */
int insert_item(buffer_item item) {
   /* When the buffer is not full add the item
      and increment the counter*/
   if(counter < BUFFER_SIZE) {
      buffer[counter] = item;
      counter++;
      return 0;
   }
   else { /* Error the buffer is full */
      return -1;
   }
}

/* Remove an item from the buffer */
int remove_item(buffer_item *item) {
   /* When the buffer is not empty remove the item
      and decrement the counter */
   if(counter > 0) {
      *item = buffer[(counter-1)];
      counter--;
      return 0;
   }
   else { /* Error buffer empty */
      return -1;
   }
}

int main(int argc, char *argv[]) {
   /* Loop counter */
   int i;

   /* Verify the correct number of arguments were passed in */
   if(argc != 4) {
      fprintf(stderr, "USAGE:./main.out <INT> <INT> <INT>\n");
   }

   int mainSleepTime = atoi(argv[1]); /* Time in seconds for main to sleep */
   int numProd = atoi(argv[2]); /* Number of producer threads */
   int numCons = atoi(argv[3]); /* Number of consumer threads */

   /* Initialize the app */
   initializeData();

   /* Create the producer threads */
   for(i = 0; i < numProd; i++) {
      /* Create the thread */
      pthread_create(&tid,&attr,producer,NULL);
    }

   /* Create the consumer threads */
   for(i = 0; i < numCons; i++) {
      /* Create the thread */
      pthread_create(&tid,&attr,consumer,NULL);
   }

   /* Sleep for the specified amount of time in milliseconds */
   sleep(mainSleepTime);

   /* Exit the program */
   printf("Exit the program\n");
   exit(0);
}

OUTPUT:
producer produced 35005211
consumer consumed 35005211
producer produced 1726956429
consumer consumed 1726956429
producer produced 278722862
consumer consumed 278722862
producer produced 468703135
producer produced 1801979802
producer produced 635723058
producer produced 1125898167
consumer consumed 1125898167
Exit the program