AAE-CNAS-Labs/Topic-7/task_7.c

// master generates tasks and place them in a queue

// slave executes tasks (and sends the result back to the consumer)

#include <errno.h>
#include <fcntl.h>
#include <mqueue.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <time.h>
#include <unistd.h>

void handle_arguments(int argc, char *argv[]) {
	if (!(argc == 2 || argc == 4)) {
		fprintf(stderr, "Usage: %s #_OF_TASKS [#_OF_PRODUCERS] [#_OF_CONSUMERS]\n", argv[0]);
		exit(EXIT_FAILURE);
	}

	if (atoi(argv[1]) < 1) {
		fprintf(stderr, "I need at least one task!\n");
		exit(EXIT_FAILURE);
	}

	// if (atoi(argv[1]) > MAXMSG) {
	// 	fprintf(stderr, "I can handle up to %d tasks!\n", MAXMSG);
	// 	exit(EXIT_FAILURE);
	// }

	if (argc == 4) {
		if (atoi(argv[2]) < 1) {
			fprintf(stderr, "I need at least one producer!\n");
			exit(EXIT_FAILURE);
		}

		if (atoi(argv[3]) < 1) {
			fprintf(stderr, "I need at least one consumer!\n");
			exit(EXIT_FAILURE);
		}

	}

	if (atoi(argv[2]) > 32) {
		fprintf(stderr, "I am lazy so I can create up to 32 producers!\n");
			exit(EXIT_FAILURE);
	}
	if (atoi(argv[3]) > 32) {
		fprintf(stderr, "I am lazy so I can create up to 32 consumers!\n");
			exit(EXIT_FAILURE);
	}



}

struct Task {
	pid_t prod_pid;
	int sqc;
	int a;
	int b;
	int result;
};

struct Task generate_task(int sqc) {
	struct Task task;
	task.prod_pid = getpid();
	task.sqc = sqc;
	task.a = random()/2;
	task.b = random()/2;
	task.result = 0;

	return task;
};

void print_task(struct Task task) {
	printf("====v TASK v====\n");
	printf("Producer PID: %d\n", task.prod_pid);
	printf("Sequence no:  %d\n", task.sqc);
	printf("A:            %d\n", task.a);
	printf("B:            %d\n", task.b);
	if (task.result != 0) {	// Cloud replace it with bool solved in Task
	printf("A + B =       %d\n", task.result);
	}
	printf("====^ TASK ^====\n");
}

void producer(int n_tasks, int prio) {
	printf("A producer has been called with PID of %d\n", getpid());

    // printf("The queue was created with fd = %d\n", queue);

    mqd_t queue = mq_open("/tasks", O_WRONLY);
	if (queue == (mqd_t) -1) {
        perror("mq_open: ");
        exit(EXIT_FAILURE);
    }

    struct Task tasks[n_tasks];

    for (int i = 0; i < n_tasks; ++i) {
    	tasks[i] = generate_task(i);
    	// print_task(tasks[i]);
    	if (mq_send(queue, (const char *)&tasks[i], sizeof(struct Task), prio)) {
        	perror("mq_send: ");
        	exit(EXIT_FAILURE);
    	}
    	printf("Task %d of %d sent...\n", i + 1, n_tasks);
    }

    mqd_t results_queue;
    do {
    	char pid_s[10];
		sprintf(pid_s, "%d", getpid());
		char results_name[30] = "/results";
		strcat(results_name, pid_s);
    	results_queue = mq_open(results_name, O_RDONLY);
    	if (results_queue == (mqd_t) -1) {
    		if (errno != ENOENT) {
    			perror("mq_open('/results'): ");
    			exit(EXIT_FAILURE);
    		}
    	}
    } while(results_queue == (mqd_t) -1);

    struct Task tmp;

    for (int i = 0; i < n_tasks; ++i) {
	    int received_bytes = mq_receive(results_queue, (char *) &tmp, sizeof(struct Task), NULL);
	    if (received_bytes < 0) {
	    	perror("producer mq_receive('/results')");
	    	exit(EXIT_FAILURE);
	    }
	    tasks[tmp.sqc] = tmp;
	    printf("%d received %d of %d results...\n", tmp.prod_pid, i + 1, n_tasks);
    }

    for (int i = 0; i < n_tasks; ++i) {
    	print_task(tasks[i]);
    }


    mq_unlink("/test");
}

int solve_difficult_problem(int a, int b) {
	sleep(random()%3 + 1);
	return a + b;
}

void consumer() {
	printf("A consumer has been called with PID of %d\n", getpid());
	mqd_t queue;
	do {
		queue = mq_open("/tasks", O_RDONLY);
		if (queue == (mqd_t) -1) {
			if (errno != ENOENT) {
        		perror("mq_open('/tasks'): ");
        		exit(EXIT_FAILURE);
			}
    	}
    } while(queue == (mqd_t) -1);


    struct Task task;

    struct timespec interval;
    interval.tv_nsec = 0;

    while (1) {
    	interval.tv_sec = time(NULL) + 5;
	    int received_bytes = mq_timedreceive(queue, (char *) &task, sizeof(struct Task), NULL, &interval);
	    if (received_bytes < 0) {
	    	perror("consumer mq_receive('/tasks'): ");
	    	exit(EXIT_FAILURE);
	    }

	    task.result = solve_difficult_problem(task.a, task.b);

	    if (received_bytes < 0) {
	    	perror("mq_receive: ");
	    	exit(EXIT_FAILURE);
	    }
		char prod_pid[10];
		sprintf(prod_pid, "%d", task.prod_pid);
		char results_name[30] = "/results";
		strcat(results_name, prod_pid);
	    mqd_t results_queue = mq_open(results_name, O_WRONLY);
	    if (results_queue == (mqd_t) -1) {
	        perror("consumer mq_open('/results'): ");
	        exit(EXIT_FAILURE);
	    }
	    if (mq_send(results_queue, (const char *)&task, sizeof(struct Task), 0)) {
        	perror("mq_send: ");
        	exit(EXIT_FAILURE);
    	}

    }


    mq_unlink("/results");

}

int main(int argc, char *argv[]) {
	srandom(time(NULL));
	handle_arguments(argc, argv);

	int n_tasks = atoi(argv[1]);
	int n_producers = atoi(argv[2]);
	int n_consumers = atoi(argv[3]);
	char tasks_name[] = "/tasks";
	// char results_name[] = "/results";

	struct mq_attr attr;
	attr.mq_flags = 0;
	attr.mq_maxmsg = 10;	// Probably more can be obtained by modifying /proc/sys/fs/mqueue/msg_max
	attr.mq_msgsize = sizeof(struct Task);
	attr.mq_curmsgs = 0;

	mq_unlink(tasks_name);
	// mq_unlink(results_name);

	mqd_t tasks_queue = mq_open(tasks_name, O_RDWR | O_CREAT, S_IRWXU, &attr);
	if (tasks_queue == (mqd_t) -1) {
        perror("mq_open: ");
        exit(EXIT_FAILURE);
    }

	pid_t pid;
	int status[n_producers + n_consumers];
	int producers_pids[n_producers];
	int consumers_pids[n_consumers];
	
	for (int i = 0; i < n_producers; ++i) {
		pid = fork();
		if (pid == -1) {
			perror("fork: ");
			exit(EXIT_FAILURE);
		}
		if (pid == 0) {
			producer(n_tasks, i);
			exit(EXIT_SUCCESS);
		}

		char pid_s[10];
		sprintf(pid_s, "%d", pid);
		char results_name[30] = "/results";
		strcat(results_name, pid_s);

		unlink(results_name);
		mqd_t results_queue = mq_open(results_name, O_RDWR | O_CREAT, S_IRWXU, &attr);
		if (results_queue == (mqd_t) -1) {
	        perror("mq_open: ");
	        exit(EXIT_FAILURE);
	    }

		producers_pids[i] = pid;
	}
	for (int i = 0; i < n_consumers; ++i) {
		pid = fork();
		if (pid == -1) {
			perror("fork: ");
			exit(EXIT_FAILURE);
		}
		if (pid == 0) {
			consumer();
			exit(EXIT_SUCCESS);
		}
		consumers_pids[i] = pid;
	}

	for (int i = 0; i < n_producers; ++i) {
		waitpid(producers_pids[i], &status[i], 0);
	}
	for (int i = 0; i < n_consumers; ++i) {
		waitpid(consumers_pids[i], &status[i], 0);
	}

	printf("This is the end!\n");

	return 0;
}