Question

*Help Please with the code**

CSIT 345 Lab 2 Process Programming Your lab requirement is to write program codes for 3.21 and shared memory program for producer and consumer as shown in the following. You can start with the code provided in the virtual machine in the virtual box you installed. The code can be found in /home/oscreader/osc9e-src/ch3 a. For 3.21, you can start with the newprocposix.c and modify the code to meet your requirement. Then type: gcc neypCOCROSİS.c to compile, then type ./a.outto execute Programming Problems 155 321 The Collatz conjecture concerns what happens when we take any positive integer n and apply the following algorithm: if n is even n/2 3 x n+1, if n is odd The conjecture states that when this algorithm is continually applied, all positive integers will eventually reach 1. For example, if n 35, the sequence is 35, 106, 53, 160, 80, 40, 20, 10, 5, 16, 8, 4, 2, 1 Write a C program using the forkO system call that generates this sequence in the child process. The starting number will be provided from the command line. For example, if 8 is passed as a parameter on the command line, the child process will output 8.4. 2, 1. Because the parent and child processes have their own copies of the data, it will be mecessary for the child to output the sequence. Have the parent invoke the wait() call to wait for the child process the ne program. Perform necessary error checking to ensure that a int to complete before exiting positive er is passed on the command line. output the sequence of s must b. You will write two programs, producer and consumer. The producer will generate the sequence of numbers using Collatz conjecture and write it to the shared memory. The consumer will read the sequence of number out from the shared memory. Please start with the code under /osc9e-src/ch3: shm-posix-consumer.c and shm-posix producer.c as the template. Modify the code as needed to fit your needs. When you compile the code, please run the following two commands: gcc_shm posix-consumer.c-o consumer -lrt gcc shm-posix-producer.c-o producer lrt Then you should type-/producer to execute the producer program and then type ./consumer to execute the consumer program. The consumer program will print out the number sequence on the screen.

Answer 1

a)

#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include </usr/include/semaphore.h>

// for sleep
#include <unistd.h>

#define BUFF_SIZE   5           /* total number of slots */
#define NP          3           /* total number of producers */
#define NC          3           /* total number of consumers */
#define NITERS      4           /* number of items produced/consumed */

typedef struct
{
    int buf[BUFF_SIZE];   /* shared var */
    int in;               /* buf[in%BUFF_SIZE] is the first empty slot */
    int out;              /* buf[out%BUFF_SIZE] is the first full slot */
    sem_t full;           /* keep track of the number of full spots */
    sem_t empty;          /* keep track of the number of empty spots */

    // use correct type here
    pthread_mutex_t mutex;          /* enforce mutual exclusion to shared data */
} sbuf_t;

sbuf_t shared;


void *Producer(void *arg)
{
    int i, item, index;

    index = (int)arg;


    for (i=0; i < NITERS; i++)
    {

        /* Produce item */
        item = i;

        /* Prepare to write item to buf */

        /* If there are no empty slots, wait */
        sem_wait(&shared.empty);
        /* If another thread uses the buffer, wait */
        pthread_mutex_lock(&shared.mutex);
        shared.buf[shared.in] = item;
        shared.in = (shared.in+1)%BUFF_SIZE;
        printf("[P%d] Producing %d ...\n", index, item);
        fflush(stdout);
        /* Release the buffer */
        pthread_mutex_unlock(&shared.mutex);
        /* Increment the number of full slots */
        sem_post(&shared.full);

        /* Interleave  producer and consumer execution */
        if (i % 2 == 1) sleep(1);
    }
    return NULL;
}

void *Consumer(void *arg)
{
    int i, item, index;

    index = (int)arg;
    for (i=NITERS; i > 0; i--) {
        sem_wait(&shared.full);
        pthread_mutex_lock(&shared.mutex);
        item=i;
        item=shared.buf[shared.out];
        shared.out = (shared.out+1)%BUFF_SIZE;
        printf("[C%d] Consuming  %d ...\n", index, item);
        fflush(stdout);
        /* Release the buffer */
        pthread_mutex_unlock(&shared.mutex);
        /* Increment the number of full slots */
        sem_post(&shared.empty);

        /* Interleave  producer and consumer execution */
        if (i % 2 == 1) sleep(1);
    }
    return NULL;
}

int main()
{
    pthread_t idP, idC;
    int index;

    sem_init(&shared.full, 0, 0);
    sem_init(&shared.empty, 0, BUFF_SIZE);
    pthread_mutex_init(&shared.mutex, NULL);
    for (index = 0; index < NP; index++)
    {
        /* Create a new producer */
        pthread_create(&idP, NULL, Producer, (void*)index);
    }
    /*create a new Consumer*/
    for(index=0; index<NC; index++)
    {
        pthread_create(&idC, NULL, Consumer, (void*)index);
    }



    pthread_exit(NULL);
}

b)

######## pc.h ############

#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <semaphore.h>
#include <fcntl.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/sem.h>
#include <sys/shm.h>

//buffers of shared memory
struct buffer{
        struct buf{
        int size;
        char data [128];
        }buf[100];
};  

//for semaphores
union semun {
        int val;
        struct semid_ds *buf;
        unsigned short *array;
};

int S;          //semaphore S: initialized to 1, to allow 1 processs to access the shared memory
int N;          //semaphore N: initialized to 0, as the shared memory begins empty
int E;          //semaphore E: initialized to 100 as the shared memory begins with 100 available buffers

int sizeOfFile;

//semaphore wait
int sem_w(int semaphore)
{
        //printf("Waiting on a semaphore\n");
    struct sembuf sem_b;
    
    sem_b.sem_num = 0;
    sem_b.sem_op = -1; /* "wait" operation: decrese semaphore valuse by 1 */
    sem_b.sem_flg = 0;
    if (semop(semaphore, &sem_b, 1) == -1) {
        fprintf(stderr, "semaphore_p failed\n");
        return(0);
    }
    return(1);  
}

//semaphore signal
int sem_s(int semaphore)
{
        //printf("Signalling a semaphore\n");
struct sembuf sem_b;
    
    sem_b.sem_num = 0;
    sem_b.sem_op = 1; /* "signal" operation: increases semaphore value by 1 */
    sem_b.sem_flg = 0;
    if (semop(semaphore, &sem_b, 1) == -1) {
        fprintf(stderr, "semaphore_v failed\n");
        return(0);
    }
    return(1);
}

###### producer.c #############

#include "pc.h"

int BytesWritten = 0;   //count of the total number of bytes written to the shared memory
int TotalBytes; //the total number of bytes in the file

char input_file [100];
FILE* in;
char input_buffer [BUFSIZ];

int main(int argc, const char * argv[]) 
{
        /* get input file name */
        if(argc < 2) //if no command line arguments input
        {
                strcpy(input_file, "input.txt");        //use defult file input.txt
        }
        else
        {
                strcpy(input_file, argv[1]);    //get file name from command like argument
        }


        /* open file */
        printf("Opening file: %s\n", input_file);
        
        in = fopen(input_file, "r");

        if(in == NULL)
        {
                printf("File '%s' does not exist in this location.\n Closing producer.\n", input_file);
                return;
        }
        
        //get the total number of bytes in the file
        fseek(in, 0, SEEK_END);
        TotalBytes = ftell(in);
        fseek(in, 0, SEEK_SET); 


        printf("Total bytes in file = %d\n", TotalBytes);


        /* set up/connect to shared memory */
        void *shared_memory = (void *)0;
        char buffer[BUFSIZ];
        int shmid;

        shmid = shmget((key_t)1234, sizeof(struct buffer), 0666 | IPC_CREAT);

         if (shmid == -1) {
                fprintf(stderr, "shmget failed\n");
                exit(EXIT_FAILURE);
        }

        shared_memory = shmat(shmid, (void *)0, 0);
         if (shared_memory == (void *)-1) {
                fprintf(stderr, "shmat failed\n");
                exit(EXIT_FAILURE);
        }

        printf("Memory attached at %X\n", (int)shared_memory);
        

        /* set up/connect to semaphores */
        union semun sem_union;
        
        S = semget((key_t) 1234, 1, 0666 | IPC_CREAT);
        sem_union.val = 1;
        semctl(S, 0, SETVAL, sem_union);

        N = semget((key_t) 4321, 1, 0666 | IPC_CREAT);
        sem_union.val = 0;
        semctl(N, 0, SETVAL, sem_union);
        
        E = semget((key_t) 1234, 1, 0666 | IPC_CREAT);
        sem_union.val = 100;
        semctl(E, 0, SETVAL, sem_union);


        /* prepare variables for counts and connection to shared memory */
        size_t count;
        int begin_index;
        int buffer_index = 0;
        
        struct buffer *producer_buffer;
        producer_buffer = (struct buffer *)shared_memory;
        

        /* send total bytes in the file the consumer */
        sem_w(S);       
        producer_buffer->buf[buffer_index].size = TotalBytes;
        sem_s(S);
        sem_s(N);
        sleep(1);

        /* read the bytes from the file in section of size BUFSIZ into the input_buffer, 
        divide this buffer into 128 byte parts and write these to the shared memory */
        while(BytesWritten < TotalBytes)     //while not at the end of the input file
        {       
                /* read from file into input buffer */
                count = fread(input_buffer, 1, BUFSIZ + BytesWritten, in);

                                        //printf("Read %d bytes from file: %s\n", count, input_buffer); 
                begin_index = 0;
                int i = count / 128;    //two indexes to send the count of bytes per buffer
                int f = i;
                //printf("i = %d", i);
                /* Divide the data in the input buffer into 128 byte parts and write to the shared memory */
                while(begin_index < count)   
                {       
                        buffer_index ++;

                        sem_w(E);       //wait on semaphore E: wait for buffer not full
                        sem_w(S);       //wait on semaphore S: wait for shared memory not in use

                        //"read" from input buffer into shared memory buffer
                        strncpy(producer_buffer->buf[buffer_index].data, input_buffer + begin_index, 128);
                        
                        //put count of the number of bytes written in shared buffer into the buffer
                        if(i == 0) {
                                //printf("Bytes sent = %d\n", count-(128*f));
                                producer_buffer->buf[buffer_index].size = count-(128*f);
                        }
                        else {
                                //printf("Bytes sent = %d\n", 128);
                                producer_buffer->buf[buffer_index].size = 128;
                                i--;
                        }
                                //printf("Written to buffer[%d]: %s\n", buffer_index, producer_buffer->buf[buffer_index].data);
                        sem_s(S);       //signal semaphore S: shared memory now available
                        sem_s(N);       //signal semaphore N: additional full buffer available in shared memory
                        begin_index += 128;             
                }

                BytesWritten += count;
        }
        //once reached EOF, print and then close file 
                printf("Reached end of file. %d bytes written to shared memory.\n", BytesWritten);
        fclose(in);
}

####### consumer.h ############

#include "pc.h"

int main(int argc, const char * argv[]) 
{
        char output_file [100];
        FILE* out;

        //establish output file
        if(argc < 2) //if no command line arguments input
        {
                strcpy(output_file, "output.txt");      //use defult file input.txt
        }
        else
        {
                strcpy(output_file, argv[1]);   //get file name from command like argument
        }
        printf("Opening file: %s\n", output_file);
        
        out = fopen(output_file, "w");
        //open file
        if(out == NULL)
        {
                printf("File '%s' does not exist in this location.\n Closing consumer.\n", output_file);
                return;
        }


        //set up/connect to shared memory
        void *shared_memory = (void *)0;
        int shmid;
        shmid = shmget((key_t)1234, sizeof(struct buffer), 0666 | IPC_CREAT);

         if (shmid == -1) {
                fprintf(stderr, "shmget failed\n");
                exit(EXIT_FAILURE);
        }

        shared_memory = shmat(shmid, (void *)0, 0);
         if (shared_memory == (void *)-1) {
                fprintf(stderr, "shmat failed\n");
                exit(EXIT_FAILURE);
        }

        printf("Memory attached at %X\n", (int)shared_memory);

        //set up/connect to semaphores
        union semun sem_union;
        
        S = semget((key_t) 1234, 1, 0666 | IPC_CREAT);
        N = semget((key_t) 4321, 1, 0666 | IPC_CREAT);
        E = semget((key_t) 1234, 1, 0666 | IPC_CREAT);
        
        sem_union.val = 1;
        semctl(S, 0, SETVAL, sem_union);

        sem_union.val = 0;
        semctl(N, 0, SETVAL, sem_union);
        
        sem_union.val = 100;
        semctl(E, 0, SETVAL, sem_union);

        struct buffer * consumer_buffer;
        consumer_buffer = (struct buffer *)shared_memory;
        
        int buffer_index = 0;

        //get total bytes in file from producer
        printf("Waiting for producer to connect and send the size of its input file...\n");
        sem_w(N);
        sem_s(S);
        int totalBytesInFile = consumer_buffer->buf[buffer_index].size;
        sem_w(S);
        
        int totalByteCount = 0;
        int count;

        while(totalByteCount < totalBytesInFile)
        {
                buffer_index ++;
                //printf("waitng on N\n");
                sem_w(N);       //wait on semaphore N: shared memory not empty
                //printf("waiting on S\n");
                //sem_w(S);     //wait on semaphore S: shared memory not in use

                //"consume" buffer from shared memory
                //printf("Read from buffer[%d]: %s\n", buffer_index,  consumer_buffer->buf[buffer_index].data);
                //printf("Bytes read = %d\n", consumer_buffer->buf[buffer_index].size);
                count = fwrite(consumer_buffer->buf[buffer_index].data, 1, consumer_buffer->buf[buffer_index].size, out);
                if(count != consumer_buffer->buf[buffer_index].size)
                        printf("Error in writing to file!!");
                totalByteCount += count;

                //sem_s(S);     //signal semaphore S
                sem_s(E);       //signal semaphore E

        }
        //once reached end of file, close output file
        printf("Total bytes written to file %s = %d.\n", output_file, totalByteCount);
        fclose(out);

}