Modulo checking using server and clients, created in C using POSIX message queues in preparation for the Operating Systems class

posix-mq-modulo-checking-client.c

#include <stdlib.h>
#include <stdio.h>
#include <mqueue.h>
#include <unistd.h>
#include <string.h>
#include <signal.h>
#include <errno.h>
#include <sys/time.h>

#define MAX_QUEUE_NAME_LENGTH 100
#define MESSAGE_LENGTH 100
#define MAX_MESSAGES 10
#define MESSAGE_SEPARATOR "|"
#define TIMEOUT 200

#define ERR(source) ( fprintf(stderr, "%s:%d\n", __FILE__, __LINE__), \
						perror(source), exit(EXIT_FAILURE) )

void usage(char *fileName)
{
	fprintf(stderr, "Usage: %s queueName\n", fileName);
	exit(EXIT_FAILURE);
}

void setHandler(int signal, void (*handler)(int))
{
	struct sigaction action;
	action.sa_handler = handler;
	if (sigaction(signal, &action, NULL) < 0)
		ERR("sigaction");
}

int main(int argc, char **argv)
{
	if (argc != 2)
		usage(argv[0]);
	
	char *outputQueueName = argv[1];
	char inputQueueName[MAX_QUEUE_NAME_LENGTH];
	if (snprintf(inputQueueName, MAX_QUEUE_NAME_LENGTH, "/%d", getpid()) < 0)
		ERR("snprintf");
	
	struct mq_attr queueAttributes;
	queueAttributes.mq_maxmsg = MAX_MESSAGES;
	queueAttributes.mq_msgsize = MESSAGE_LENGTH;
	mqd_t inputQueueDes = mq_open(inputQueueName, O_RDONLY | O_CREAT, 0600, &queueAttributes);
	mqd_t outputQueueDes = mq_open(outputQueueName, O_WRONLY);
	if (inputQueueDes < 0 || outputQueueDes < 0)
		ERR("mq_open");
	
	
	setHandler(SIGALRM, SIG_IGN);
	setHandler(SIGINT, SIG_IGN);
	
	printf("Queue opened, reading number from stdin\n");
	char message[MESSAGE_LENGTH];
	if (snprintf(message, MESSAGE_LENGTH, "%s%s", inputQueueName, MESSAGE_SEPARATOR) < 0)
		ERR("snprintf");
	int* numberPlaceholder = (int*)(message + strlen(inputQueueName) + strlen(MESSAGE_SEPARATOR));
	char numberToCheck[MESSAGE_LENGTH];
	struct itimerval timeoutTimer, resetTimer;
	timeoutTimer.it_interval.tv_sec = 0;
	timeoutTimer.it_interval.tv_usec = 0;
	timeoutTimer.it_value.tv_sec = TIMEOUT / 1000;
	timeoutTimer.it_value.tv_usec = (TIMEOUT - timeoutTimer.it_value.tv_sec * 1000) * 1000;
	memset(&resetTimer, 0, sizeof(struct itimerval));
	while (fgets(numberToCheck, MESSAGE_LENGTH, stdin) != NULL)
	{
		*numberPlaceholder = atoi(numberToCheck);
		printf("Sending %d to %s (%d)\n", *numberPlaceholder, outputQueueName, outputQueueDes);
		if (mq_send(outputQueueDes, message, MESSAGE_LENGTH, 0) < 0)
		{
			if (errno == EINTR)
				break;
			ERR("mq_send");
		}
		printf("Message sent, waiting %d ms for response\n", TIMEOUT);
		
		if (setitimer(ITIMER_REAL, &timeoutTimer, NULL) < 0)
			ERR("setitimer");
		
		char response[MESSAGE_LENGTH];
		if (mq_receive(inputQueueDes, response, MESSAGE_LENGTH, NULL) < 0)
		{
			if (errno == EINTR)
			{
				printf("Timeout - no response came for the client's message queue\n");
				break;
			}
		}
		if (setitimer(ITIMER_REAL, &resetTimer, NULL) < 0)
			ERR("setitimer");
		
		printf("Received message: %s\n", response);
	}
	
	
	if (mq_close(inputQueueDes) < 0 || mq_close(outputQueueDes) < 0)
		ERR("mq_close");
	if (mq_unlink(inputQueueName) < 0)
		ERR("mq_unlink");
	return EXIT_SUCCESS;
}

posix-mq-modulo-checking-server.c

#include <stdlib.h>
#include <stdio.h>
#include <mqueue.h>
#include <unistd.h>
#include <string.h>
#include <signal.h>
#include <errno.h>

#define DIVIDERS_COUNT 4
#define MAX_QUEUE_NAME_LENGTH 100
#define MESSAGE_LENGTH 100
#define MAX_MESSAGES 10
#define MESSAGE_SEPARATOR "|"

#define ERR(source) ( fprintf(stderr, "%s:%d\n", __FILE__, __LINE__), \
						perror(source), exit(EXIT_FAILURE) )

void usage(char *fileName)
{
	fprintf(stderr, "Usage: %s d1 ... d%d\n", fileName, DIVIDERS_COUNT);
	fprintf(stderr, "d1, ..., d%d - dividers, d_i > 0\n", DIVIDERS_COUNT);
	exit(EXIT_FAILURE);
}

int parseDivider(char *argument, char *fileName)
{
	int divider = atoi(argument);
	if (divider <= 0)
		usage(fileName);
	return divider;
}

void setHandler(int signal, void (*handler)(int))
{
	struct sigaction action;
	action.sa_handler = handler;
	//action.sa_sigaction = handler;
	//action.sa_flags = SA_SIGINFO;
	if (sigaction(signal, &action, NULL) < 0)
		ERR("sigaction");
}


void registerNotification(mqd_t queueDes);
void checkDivision(mqd_t queueDes, int divisor)
{
	registerNotification(queueDes);
	
	struct mq_attr queueAttributes;
	while (mq_getattr(queueDes, &queueAttributes) == 0 && queueAttributes.mq_curmsgs > 0)
	{
		char message[MESSAGE_LENGTH];
		if (mq_receive(queueDes, message, MESSAGE_LENGTH, NULL) < 0)
			ERR("mq_receive");
		
		char *separator = strstr(message, MESSAGE_SEPARATOR);
		message[separator - message] = '\0';
		char *receiverQueue = message;
		int numberToCheck = *(int*)(separator + strlen(MESSAGE_SEPARATOR));
		
		printf("checking %d and returning to %s\n", numberToCheck, receiverQueue);
		
		mqd_t receiverQueueDes = mq_open(receiverQueue, O_WRONLY);
		if (receiverQueueDes < 0)
			ERR("mq_open");
		if (mq_send(receiverQueueDes, message, MESSAGE_LENGTH, 0) < 0)
			ERR("mq_send");
		if (mq_close(receiverQueueDes) < 0)
			ERR("mq_close");
		printf("response sent successfully\n");
	}
}

void registerNotification(mqd_t queueDes)
{
	struct sigevent notification;
	memset(&notification, 0, sizeof(struct sigevent));
	notification.sigev_notify = SIGEV_SIGNAL;
	notification.sigev_value.sival_int = queueDes;
	notification.sigev_signo = SIGRTMIN;
	//~ notification.sigev_notify_function = checkDivision;
	
	if (mq_notify(queueDes, &notification) < 0 && errno != EBUSY)
		ERR("mq_notify");
}

int main(int argc, char **argv)
{
	if (argc != DIVIDERS_COUNT + 1)
		usage(argv[0]);
	
	int dividers[DIVIDERS_COUNT];
	for (int i=0; i < DIVIDERS_COUNT; i++)
		dividers[i] = parseDivider(argv[1 + i], argv[0]);
	
	sigset_t blockMask, prevMask;
	sigaddset(&blockMask, SIGRTMIN);
	sigaddset(&blockMask, SIGINT);
	if (sigprocmask(SIG_BLOCK, &blockMask, &prevMask) < 0)
		ERR("sigprocmask");
	
	pid_t pid = getpid();
	char queueNames[DIVIDERS_COUNT][MAX_QUEUE_NAME_LENGTH];
	mqd_t queueDes[DIVIDERS_COUNT];
	struct mq_attr queueAttributes;
	queueAttributes.mq_maxmsg = MAX_MESSAGES;
	queueAttributes.mq_msgsize = MESSAGE_LENGTH;
	for (int i=0; i < DIVIDERS_COUNT; i++)
	{
		if (snprintf(queueNames[i], MAX_QUEUE_NAME_LENGTH, "/%d_%d", pid, dividers[i]) <= 0)
			ERR("snprintf");
		queueDes[i] = mq_open(queueNames[i], O_RDONLY | O_CREAT, 0600, &queueAttributes);
		if (queueDes[i] < 0)
			ERR("mq_open");
		registerNotification(queueDes[i]);
		printf("queue %s for divider %d created\n", queueNames[i], dividers[i]);
	}
	
	while (1)
	{
		int lastSignal;
		if (sigwait(&blockMask, &lastSignal) < 0)
			ERR("sigwait");
		if (lastSignal == SIGRTMIN)
		{
			printf("checking queues\n");
			for (int i=0; i < DIVIDERS_COUNT; i++)
				checkDivision(queueDes[i], dividers[i]);
		}
		else if (lastSignal == SIGINT)
		{
			printf("C-c received, cleaning up\n");
			break;
		}
	}
	
	printf("closing and removing all the queues\n");
	for (int i=0; i < DIVIDERS_COUNT; i++)
	{
		if (mq_close(queueDes[i]) < 0)
			ERR("mq_close");
		if (mq_unlink(queueNames[i]) < 0)
			ERR("mq_unlink");
	}
	printf("queues removed successfully\n");
	
	if (sigprocmask(SIG_SETMASK, &prevMask, NULL) < 0)
		ERR("sigprocmask");
	return EXIT_SUCCESS;
}