Lab1 task from the SOP2 course

pipe-network-simulation.c

#define _GNU_SOURCE
#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <unistd.h>
#include <signal.h>
#include <sys/wait.h>
#include <time.h>
#include <string.h>
#include <limits.h>

#define CLOSE_MESSAGE "[-]"

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

void safeClosePipe(int fd)
{
	if (TEMP_FAILURE_RETRY(close(fd)) < 0)
		ERR("close");
}

int setHandler(void (*f)(int), int sigNo)
{
	struct sigaction act;
	memset(&act, 0, sizeof(struct sigaction));
	act.sa_handler = f;
	if (-1 == sigaction(sigNo, &act, NULL))
		return -1;
	return 0;
}

void handleZombie(int sig)
{
	pid_t pid;
	for(;;){
		pid = waitpid(0, NULL, WNOHANG);
		if(pid == 0) return;
		if(pid < 0) {
				if(errno == ECHILD) return;
				ERR("waitpid");
		}
	}
}

volatile sig_atomic_t lastSignal = 0;
void sigHandler(int sig) {
	lastSignal = sig;
}

void usage(char *name)
{
	fprintf(stderr,"Usage: %s n k\n", name);
	fprintf(stderr,"n, k > 0\n");
	exit(EXIT_FAILURE);
}

// You must free the result if result is non-NULL.
char *str_replace(char *orig, char *rep, char *with) {
    char *result; // the return string
    char *ins;    // the next insert point
    char *tmp;    // varies
    int len_rep;  // length of rep (the string to remove)
    int len_with; // length of with (the string to replace rep with)
    int len_front; // distance between rep and end of last rep
    int count;    // number of replacements

    // sanity checks and initialization
    if (!orig || !rep)
        return NULL;
    len_rep = strlen(rep);
    if (len_rep == 0)
        return NULL; // empty rep causes infinite loop during count
    if (!with)
        with = "";
    len_with = strlen(with);

    // count the number of replacements needed
    ins = orig;
    for (count = 0; (tmp = strstr(ins, rep)); ++count) {
        ins = tmp + len_rep;
    }

    tmp = result = malloc(strlen(orig) + (len_with - len_rep) * count + 1);

    if (!result)
        return NULL;

    // first time through the loop, all the variable are set correctly
    // from here on,
    //    tmp points to the end of the result string
    //    ins points to the next occurrence of rep in orig
    //    orig points to the remainder of orig after "end of rep"
    while (count--) {
        ins = strstr(orig, rep);
        len_front = ins - orig;
        tmp = strncpy(tmp, orig, len_front) + len_front;
        tmp = strcpy(tmp, with) + len_with;
        orig += len_front + len_rep; // move to next "end of rep"
    }
    strcpy(tmp, orig);
    return result;
}

void initialClosePipes(int **pipes, int n, int childID)
{
	// Zamykanie pipe'ów do odczytu
	for (int i=0; i <= n; i++)
	{
		if (i == childID)
			continue;
		//printf("[%d] closes read-end of pipe %d\n", childID, i);
		safeClosePipe(pipes[i][0]);
	}
	
	// Zamykanie pipe do zapisu do siebie
	//printf("[%d] closes write-end of pipe %d\n", childID, childID);
	safeClosePipe(pipes[childID][1]);
}

void cleanupClosePipes(int **pipes, int n, int childID)
{
	// Zamykanie swojego pipe do odczytu
	safeClosePipe(pipes[childID][0]);
	
	// Zamykanie pipe'ów do zapisu i zwalnianie pamięci
	for (int i=0; i <= n; i++)
	{
		if (i != childID)
			safeClosePipe(pipes[i][1]);
		free(pipes[i]);
	}
	free(pipes);
}

int isCloseMessage(char buffer[PIPE_BUF])
{
	return strstr(CLOSE_MESSAGE, buffer) != NULL;
}

void markMessage(char buffer[PIPE_BUF], int childID)
{
	char substringToReplace[10];
	if (sprintf(substringToReplace, " %d", childID) <= 0)
		ERR("sprintf");
	char *markedMessage = str_replace(buffer, substringToReplace, " x");
	if (markedMessage == NULL)
		ERR("str_replace");
	strncpy(buffer, markedMessage, PIPE_BUF);
	buffer[strlen(markedMessage)] = '\0';
	free(markedMessage);
}

int getRandomRecepient(int n, int childID)
{
	int recepient = rand() % (n + 1);
	if (recepient == childID)
	{
		recepient += 1;
		recepient %= (n + 1);
	}
	
	return recepient;
}

void sendInitialMessage(char buffer[PIPE_BUF], int n, int messageNumber, int destinationFd)
{
	if (sprintf(buffer, "[%d]", messageNumber) <= 0)
		ERR("sprintf");
	for (int i=0; i <= n; i++)
	{
		if (sprintf(buffer, "%s %d", buffer, i) <= 0)
			ERR("sprintf");
	}
	
	ssize_t bytesWritten = write(destinationFd, buffer, PIPE_BUF);
	if (bytesWritten < 0)
		ERR("write");
}

int childWork(int **pipes, int n, int childID)
{
	initialClosePipes(pipes, n, childID);
	srand(getpid());
	
	char buffer[PIPE_BUF];
	ssize_t bytesRead;
	do
	{
		bytesRead = read(pipes[childID][0], buffer, PIPE_BUF);
		if (bytesRead > 0)
		{
			printf("%d received %s\n", childID, buffer);
			if (isCloseMessage(buffer))
			{
				printf("%d - close message received\n", childID);
				break;
			}
			markMessage(buffer, childID);
			printf("%d sends %s\n", childID, buffer);
			int recepient = getRandomRecepient(n, childID);
			if (write(pipes[recepient][1], buffer, PIPE_BUF) <= 0)
				ERR("write");
		}
	} while (bytesRead > 0);
	
	printf("[%d] exits\n", childID);
	cleanupClosePipes(pipes, n, childID);
	return EXIT_SUCCESS;
}

int main(int argc, char **argv)
{
	int n, k;
	if (argc != 3)
		usage(argv[0]);
	
	n = atoi(argv[1]);
	k = atoi(argv[2]);
	
	if (n <= 0 || k <= 0)
		usage(argv[0]);
	
	int **pipes = malloc((n + 1) * sizeof(int*));
	if (pipes == NULL)
		ERR("malloc");
	for (int i=0; i <= n; i++)
	{
		pipes[i] = malloc(2 * sizeof(int));
		if (pipes[i] == NULL)
			ERR("malloc");
		if (pipe(pipes[i]) < 0)
			ERR("pipe");
	}

	for (int i=1; i <= n; i++)
	{
		pid_t pid = fork();
		if (pid < 0)
			ERR("fork");
		else if (pid == 0)
			return childWork(pipes, n, i);
	}
	srand(getpid());
	initialClosePipes(pipes, n, 0);
	
	char buffer[PIPE_BUF];
	for (int i=1; i <= k; i++)
	{
		int recepientID = getRandomRecepient(n, 0);
		printf("\nparent sending message number %d\n", i);
		sendInitialMessage(buffer, n, i, pipes[recepientID][1]);
		ssize_t bytesRead;
		do 
		{
			bytesRead = read(pipes[0][0], buffer, PIPE_BUF);
			if (bytesRead > 0)
			{
				printf("parent read message: %s\n", buffer);
				markMessage(buffer, 0);
				break;
			}
		} while (bytesRead > 0);
	}
	
	printf("parent sends close message\n");
	if (sprintf(buffer, "%s", CLOSE_MESSAGE) <= 0)
		ERR("sprintf");
	for (int i=1; i <= n; i++)
	{
		if (write(pipes[i][1], buffer, PIPE_BUF) <= 0)
			ERR("write");
	}
	
	printf("parent waits for children and exits\n");
	cleanupClosePipes(pipes, n, 0);
	while (wait(NULL) > 0)
		continue;
	
	return EXIT_SUCCESS;
}