Process based circular buffer not working
I'm converting the program I wrote from the previous thread to use processes instead of threads. Since I got the pthread to work with the help of Salem, I don't need to worry about the looping problems I had before.
The problem I'm having now is sharing the values across the processes. I'm using mmap() to do this, but it's proving to be a challenge for the buffer and the sharing struct.
I'm having difficulty initializing share_t as an array and buffer is not properly sharing variables from request_t() to lift() (turning up as zeros from the calloc).
Can someone help with getting this thing to work, please.
Here is the code.
llist.h
Code:
#include <stdbool.h>
typedef struct Buffer
{
int from;
int to;
}buffer_t; //buffer arrary to store from and to values from sim_input
typedef struct Output
{
int p; //previous floor
int rf; //request from
int rt; //request to
int total_m[4]; //total movement
int c; // current position
int m; //movement
int total_r[4]; //total requests
}write_t;
typedef struct shared_memory
{
void* in;
void* out;
int id; //name of thread (more relieable than pthread_self())
}share_t; //Shared memory to be used across functions (Allow multiple arguements in pthread_create)
lift_sim_B.c
Code:
#include <stdio.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <unistd.h>
#include <semaphore.h>
#include <sys/mman.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#include <sys/stat.h>
#include <fcntl.h>
#include "list.h"
sem_t *full, *empty; //Semaphore mutex intializer
static int *counter; //number of requests made
static int size; //buffer size (must only change once at beginning)
//Doesn't need to be in shared memory because it will be the same for all processes.
static int sec; //Time in seconds, needed for lift to execute operation
static int* length; //Number of values in buffer
//Will be shared across processes using mmap.
static int *in; //index of reader
static int *out_i; //index of writer
static buffer_t *A;
write_t write1; //Used for printing the information from buffer.
void *lift(void* vargp) //Consumer
{
share_t* share = vargp;
while (true)
{
while (*length <= 0)
{
sem_wait(&empty);
printf("Waiting\n");
}
sleep(sec);
printf("lifting information\n");
int id = ((share_t*)vargp)->id;
//gather information to print
if (write1.p == 0) //only for when system begins (Assign Previous to the first record `from` value)
{
write1.p = A[*out_i].from;
}
write1.rf = A[*out_i].from;
write1.rt = A[*out_i].to;
write1.m = (write1.p - A[*out_i].from) + (A[*out_i].to - A[*out_i].from);
write1.c = A[*out_i].to;
//Now write the information
write1.total_r[id] = write1.total_r[id] + 1;
write1.total_m[id] = write1.total_m[id] + 1;
fprintf(share->out, "Lift-%d Operation\n"
"Previous position: Floor %d\n"
"Request: Floor %d to Floor %d\n"
"Detail operations:\n"
" Go from Floor %d to Floor %d\n"
" Go from Floor %d to Floor %d\n"
" #movement for this request: %d\n"
" #request: %d\n"
" Total #movement: %d\n"
"Current Position: Floor %d\n"
"\n"
"----------------------------\n", id, write1.p, write1.rf, write1.rt, write1.p, write1.rf, write1.rf, write1.rt
, write1.m, write1.total_r[id], write1.total_m[id], write1.c);
write1.p = write1.c; //for next statement
*length = *length - 1;
*out_i = (*out_i + 1) % size;
if (*length == 0)
{
printf("Signalling\n");
sem_post(&full);
}
}
return NULL;
}
void *request_t(void *vargp) //producer
{
share_t* share = vargp;
while(1)
{
while (*length >= size)
{
sem_wait(&full);
printf("Waiting\n");
}
printf("Requesting information\n");
//read the input line by line and into the buffer
if (fscanf(share->in, "%d %d\n", &A[*in].from, &A[*in].to) == EOF)
{
break;//stop the loop when file is empty
}
//If requests go out of bounds, they will be automatically registered as either
//maximum level or minimum level
if (A[*in].to > 20)
{
A[*in].to = 20;
}
else if (A[*in].to < 1)
{
A[*in].to = 1;
}
else if (A[*in].from > 20)
{
A[*in].from = 20;
}
else if (A[*in].from < 1)
{
A[*in].from = 1;
}
//Print buffer information to sim_output
fprintf(share->out, "New Lift Request from Floor %d to Floor %d \n"
"Request No %d \n"
"----------------------------\n", A[*in].from, A[*in].to, * counter = *counter + 1);
*length = *length + 1;
*in = (*in + 1) % size;
if (*length == 1)
{
printf("Signalling\n");
sem_post(&empty);
}
}
return NULL;
}
void main(int argc, char *argv[]) // to avoid segmentation fault
{
//establish values for use (convert argv to int values)
if (argc != 3) //Must have both m and t
{
perror("Insufficient arguements\n");
exit(EXIT_FAILURE);
}
*counter = 1;
//Create a shared memory for the file streams
//it is important the algorithm keeps a consistent stream of values
share_t *share[4];
share = mmap(NULL, sizeof * share, PROT_READ | PROT_WRITE,
MAP_SHARED | MAP_ANONYMOUS, -1, 0);
in = mmap(NULL, sizeof * in, PROT_READ | PROT_WRITE,
MAP_SHARED | MAP_ANONYMOUS, -1, 0);
out_i = mmap(NULL, sizeof * out_i, PROT_READ | PROT_WRITE,
MAP_SHARED | MAP_ANONYMOUS, -1, 0);
*in = 0;
*out_i = 0;
long arg = strtol(argv[1], NULL, 10);
size = arg;
if (size <= 1)
{
printf("buffer size too small\n");
exit(0);
}
else
{
//allow variables to be shared among processes without duplicates being made
length = mmap(NULL, sizeof * length, PROT_READ | PROT_WRITE,
MAP_SHARED | MAP_ANONYMOUS, -1, 0);
counter = mmap(NULL, sizeof * counter, PROT_READ | PROT_WRITE,
MAP_SHARED | MAP_ANONYMOUS, -1, 0);
*length = size;
*counter = 0;
//initialize Buffer as empty array and as shared memory.
A = mmap(NULL, sizeof * A, PROT_READ | PROT_WRITE,
MAP_SHARED | MAP_ANONYMOUS, -1, 0);
A = calloc(size, sizeof *A);
}
arg = strtol(argv[2], NULL, 10);
sec = arg; //time required by each lift to serve a request
if (sec <= 0)
{
printf("insufficient time\n");
exit(0);
}
//open files for reading and writing
FILE *out = fopen("sim_output.txt", "w");
FILE *in_t = fopen("sim_input.txt", "r");
if (in == NULL && out == NULL) //Check if files exists before we can use them
{
printf("Input file empty\n");
exit(0);
}
for (int i = 0; i < 4; i++)//Load files into share for transporting.
{
share[i].in = in_t;
share[i].out = out;
share[i].id = i;
}
//Create processes
printf("Creating processes\n");
int shared_mem_id = shmget(IPC_PRIVATE, sizeof(sem_t), IPC_CREAT);
full = shmat(shared_mem_id, NULL, 0);
empty = shmat(shared_mem_id, NULL, 0);
sem_init(&mutex, 1, 1);
if (fork() == 0) { // Lift-R Child Proccess
full = shmat(shared_mem_id, NULL, 0);
request_t(share);
shmdt(mutex);
exit(0);
}
if (fork() == 0) { // lift-1 Child Proccess
empty = shmat(shared_mem_id, NULL, 0);
lift(share[1]);
shmdt(mutex);
exit(0);
}
wait(NULL);
if (fork() == 0) { // lift-2 Child Proccess
empty = shmat(shared_mem_id, NULL, 0);
lift(share[2]);
shmdt(mutex);
exit(0);
}
wait(NULL);
if (fork() == 0) { // lift-3 Child Proccess
empty = shmat(shared_mem_id, NULL, 0);
lift(share[3]);
shmdt(mutex);
exit(0);
}
wait(NULL);
int sum_r = write1.total_r[1] + write1.total_r[2] + write1.total_r[3],
sum_t = write1.total_m[1] + write1.total_m[2] + write1.total_m[3]; //Create total sums of requests and movement
fprintf(out, "\n"
"Total number of requests %d \n"
"Total number of movememts %d \n", sum_r, sum_t); //Print at the very last line after all requests were made
fclose(in_t);
fclose(out);
printf("finished\n");
exit(0);
}