Task from the first SOP2 laboratory

pipe-tree-hierarchy.c

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

#define LEVELS 3
#define ITERATIONS 100
#define SLEEP_DURATION 10

#define TOTAL_NODES ((2 << LEVELS) - 1)
#define PARENT(i) ((i - 1) / 2)
#define LEFT(i) (2 * i + 1)
#define RIGHT(i) (2 * i + 2)


#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");
}

void freePipes(int *pipes[TOTAL_NODES + 1])
{
	for (int i=0; i <= TOTAL_NODES; i++)
		free(pipes[i]);
}

void closeRecursiveWithExclusion(int *pipes[TOTAL_NODES + 1], int startingIndex, int excludedIndex, int childrenOnly)
{
	if (startingIndex > TOTAL_NODES || startingIndex == excludedIndex)
		return;
	
	if (!childrenOnly)
	{
		safeClosePipe(pipes[startingIndex][0]);
		safeClosePipe(pipes[startingIndex][1]);
	}
	
	// Rekurencja na koncu nieusunieta. Kotowski by dal 0 pkt
	closeRecursiveWithExclusion(pipes, LEFT(startingIndex), excludedIndex, 0);
	closeRecursiveWithExclusion(pipes, RIGHT(startingIndex), excludedIndex, 0);
}

int childWork(int index, int *pipes[TOTAL_NODES + 1])
{
	int level = log((double)index + 1) / log(2.0);
	
	srand(getpid());
	
	// Zamknij wszysktie pipe poza soba i bezposrednimi dziecmi
	closeRecursiveWithExclusion(pipes, 0, index, 0);
	closeRecursiveWithExclusion(pipes, LEFT(index), index, 1);
	closeRecursiveWithExclusion(pipes, RIGHT(index), index, 1);
	
	// A takze pipe do pisania do siebie
	safeClosePipe(pipes[index][1]);
	
	
	int hasChildren = level < LEVELS ? 1 : 0;
	if (hasChildren)
	{
		// Zamknij pipe do odczytu dla dzieci
		safeClosePipe(pipes[LEFT(index)][0]);
		safeClosePipe(pipes[RIGHT(index)][0]);
	}
	
	printf("[%d] start at level [%d]\n", getpid(), level);
	if (hasChildren)
		printf("[%d] will listen on %d, write to %d and %d\n", getpid(), pipes[index][0], pipes[LEFT(index)][1], pipes[RIGHT(index)][1]);
	else
		printf("[%d] has no children, will listen on %d\n", getpid(), pipes[index][0]);
	
	char buffer[PIPE_BUF];
	ssize_t bytesRead;
	int counter = 0;
	do
	{
		bytesRead = read(pipes[index][0], buffer, PIPE_BUF);
		if (bytesRead < 0)
			ERR("read");
		else if (bytesRead > 0)
		{
			int numberRead = *(int*)buffer;
			// printf("%d read %d\n", index, numberRead);
			if (numberRead == 0)
			{
				if (hasChildren)
				{
					ssize_t bytesWrittenToLeft = TEMP_FAILURE_RETRY(write(pipes[LEFT(index)][1], buffer, PIPE_BUF)),
						bytesWrittenToRight = TEMP_FAILURE_RETRY(write(pipes[RIGHT(index)][1], buffer, PIPE_BUF));
					if (bytesWrittenToLeft < 0 || bytesWrittenToRight < 0)
						ERR("write");
				}
				break;
			}
			
			// Prześlij dalej albo zwiększ licznik
			if (!hasChildren)
			{
				counter++;
				// printf("[%d] counter %d\n", getpid(), counter);
			}
			else
			{
				int receiverIndex = LEFT(index) + rand() % 2;
				// printf("%d sends 1 to %d\n", index, receiverIndex);
				
				ssize_t bytesWritten = TEMP_FAILURE_RETRY(write(pipes[receiverIndex][1], buffer, PIPE_BUF));
				if (bytesWritten < 0)
					ERR("write");
			}
		}
	} while (bytesRead > 0);
	
	if (hasChildren)
	{
		safeClosePipe(pipes[LEFT(index)][1]);
		safeClosePipe(pipes[RIGHT(index)][1]);
	}
	else
	{
		printf("[%d] counted to %d\n", getpid(), counter);
	}
	safeClosePipe(pipes[index][0]);
	
	printf("[%d] stop\n", getpid());
	freePipes(pipes);
	return EXIT_SUCCESS;
}

int main(int argc, char **argv)
{
	// indeksowanie jak w kopcu binarnym
	int *pipes[TOTAL_NODES + 1];
	for (int i=0; i <= TOTAL_NODES; 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 < TOTAL_NODES; i++)
	{
		pid_t pid = fork();
		if (pid < 0)
			ERR("fork");
		else if (pid == 0)
			return childWork(i, pipes);
	}
	
	// Pisanie i czytanie do samego siebie
	safeClosePipe(pipes[0][0]);
	safeClosePipe(pipes[0][1]);
	// Wszystko ponizej dzieci
	closeRecursiveWithExclusion(pipes, 1, 0, 1);
	closeRecursiveWithExclusion(pipes, 2, 0, 1);
	// Pisanie do dzieci
	safeClosePipe(pipes[1][0]);
	safeClosePipe(pipes[2][0]);
	
	srand(time(NULL));
	char buffer[PIPE_BUF];
	*(int*)buffer = 1;
	struct timespec sleepDuration;
	
	for (int i=0; i < ITERATIONS; i++)
	{
		sleepDuration.tv_sec = 0;
		sleepDuration.tv_nsec = SLEEP_DURATION * 1000;
		int receiverIndex = 1 + rand() % 2;
		ssize_t bytesWritten = TEMP_FAILURE_RETRY(write(pipes[receiverIndex][1], buffer, PIPE_BUF));
		if (bytesWritten < 0)
			ERR("write");
		while (nanosleep(&sleepDuration, &sleepDuration) != 0)
			continue;
	}
	*(int*)buffer = 0;
	ssize_t bytesWrittenToLeft = TEMP_FAILURE_RETRY(write(pipes[1][1], buffer, PIPE_BUF)),
		bytesWrittenToRight = TEMP_FAILURE_RETRY(write(pipes[2][1], buffer, PIPE_BUF));
	if (bytesWrittenToLeft < 0 || bytesWrittenToRight < 0)
		ERR("write");
	safeClosePipe(pipes[1][1]);
	safeClosePipe(pipes[2][1]);
	
	printf("Parent waits for children\n");
	while (wait(NULL) >= 0)
		continue;
	
	printf("Parent exits\n");
	freePipes(pipes);
	return EXIT_SUCCESS;
}