Task 7 completed

Topic-7/task_7.c

@@ -0,0 +1,285 @@
+// 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;
+}