bssolver/master.c

#define _SVID_SOURCE 1
//#define _DEFAULT_SOURCE 1

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <math.h>
#include <sys/stat.h>
#include <sys/time.h>
#include <sys/types.h>
#include <sys/select.h>
#include <fcntl.h>
#include <sys/shm.h>
#include "error.h"
#include "shr_mem.h"
#include <semaphore.h>

#define SLAVE_NUM 5
#define MAX_SIZE 200
#define MAX_SHM_SIZE 0x4B000

void createPipes(int pipesOutput[SLAVE_NUM][2], int pipesInput[SLAVE_NUM][2]);
int writeOutput(char * output, FILE * outputFile, void * shmP);
void closePipes(int pipesInput[SLAVE_NUM][2], int pipesOutput[SLAVE_NUM][2], int slave);
char sendTask(struct stat statBuf, char * fileName, int pipesInput[SLAVE_NUM][2], int slave);

int main(int argc, char *argv[]) {
    if (setvbuf(stdout, NULL, _IONBF, 0) != 0)
        printSystemError("setvbuf() -- ERROR");

    if (argc == 1)
        printError("Error: no arguments provided");

    sem_unlink("/FULL");
    sem_unlink("/EMPTY");

    FILE * outputFile;
    if ((outputFile = fopen("outputFile.txt", "w")) == NULL)
        printSystemError("fopen() -- ERROR");

    char * shmPointer;
    int shmFd = createShm(MAX_SHM_SIZE, &shmPointer);
    void * shmStart = shmPointer;

    printf("%d\n", argc - 1);
    sleep(2);

    int pipesOutput[SLAVE_NUM][2], pipesInput[SLAVE_NUM][2];
    createPipes(pipesOutput, pipesInput);

    struct stat statBuf;
    int tasks = 1;
    for (int j = 0; tasks <= SLAVE_NUM && tasks < argc; tasks++) {
        if (sendTask(statBuf, argv[tasks], pipesInput, j++) == 0)
            continue;
    }

    char buf[MAX_SIZE];

    fd_set pipesOutputSet;
    FD_ZERO(&pipesOutputSet);

    int r, readBytes, flag, count;
    if (tasks != SLAVE_NUM) {
        for (int k = tasks - 1; k < SLAVE_NUM; k++) {
            closePipes(pipesInput, pipesOutput, k);
        }
    }
    do {
        flag = 0;
        FD_ZERO(&pipesOutputSet);
        for (int n = 0; n < SLAVE_NUM; n++) {
            if (pipesOutput[n][0] != -1) {
                FD_SET(pipesOutput[n][0], &pipesOutputSet);
                flag = 1;
            }
        }
        if (flag) {
            if ((count = select(pipesOutput[SLAVE_NUM - 1][1] + 1, &pipesOutputSet, NULL, NULL, NULL)) <= 0)
                printSystemError("select() -- ERROR");

            for (r = 0; r < SLAVE_NUM && count > 0; r++) {
                if (FD_ISSET(pipesOutput[r][0], &pipesOutputSet)) {
                    readBytes = 0;
                    do {
                        if ((readBytes += read(pipesOutput[r][0], buf + readBytes, MAX_SIZE)) < 0)
                            printSystemError("read() -- ERROR");
                    } while (buf[readBytes - 1] != '\n');
                    buf[readBytes] = 0;

                    // printf("\n%s\n", buf);
                    shmPointer += writeOutput(buf, outputFile, shmPointer);
                    count--;

                    if (tasks < argc) {
                        if (sendTask(statBuf, argv[tasks++], pipesInput, r) == 0)
                            continue;
                    } else {
                        closePipes(pipesInput, pipesOutput, r);
                    }
                }
            }
        }
    } while (flag);

    close(shmFd);
    fclose(outputFile);
    return EXIT_SUCCESS;
}

char sendTask(struct stat statBuf, char * fileName, int pipesInput[SLAVE_NUM][2], int slave) {
    if (stat(fileName, &statBuf) == -1)
        printSystemError("Error, one of the specified paths is incorrect");

    if (!S_ISREG(statBuf.st_mode))
        return EXIT_FAILURE;

    if ((write(pipesInput[slave][1], fileName, strlen(fileName) + 1)) < 0)
        printSystemError("write() -- ERROR");

    return EXIT_SUCCESS;
}

void closePipes(int pipesInput[SLAVE_NUM][2], int pipesOutput[SLAVE_NUM][2], int slave) {
    if (close(pipesOutput[slave][0]) < 0)
        printSystemError("close() -- ERROR");
    if (close(pipesInput[slave][1]) < 0)
        printSystemError("close() -- ERROR");

    pipesOutput[slave][0] = -1;
    pipesInput[slave][0] = -1;
}

int writeOutput(char * output, FILE * outputFile, void * shmPointer) {
    int length = strlen(output);

    if (fwrite(output, sizeof(char), length, outputFile) != length)
        printSystemError("fwrite() -- ERROR");

    writeShm(shmPointer, output, length);

    return length + 1;
}

void createPipes(int pipesOutput[SLAVE_NUM][2], int pipesInput[SLAVE_NUM][2]) {
    pid_t pid[SLAVE_NUM];

    for (int i = 0; i < SLAVE_NUM; i++) {
        if (pipe(pipesOutput[i]) < 0)
            printSystemError("pipe() -- ERROR");
        if (pipe(pipesInput[i]) < 0)
            printSystemError("pipe() -- ERROR");
        if ((pid[i] = fork()) == 0) {
            if (close(pipesOutput[i][0]) < 0)
                printSystemError("close() -- ERROR");
            if (dup2(pipesOutput[i][1], STDOUT_FILENO) < 0)
                printSystemError("dup2() -- ERROR 1");
            if (close(pipesOutput[i][1]) < 0)
                printSystemError("close() -- ERROR");

            if (close(pipesInput[i][1]) < 0)
                printSystemError("close() -- ERROR");
            if (dup2(pipesInput[i][0], STDIN_FILENO) < 0)
                printSystemError("dup2() -- ERROR 2");
            if (close(pipesInput[i][0]) < 0)
                printSystemError("close() -- ERROR");

	    char * param[] = {"slave.o", NULL};
            if (execv("slave.o", param) < 0)
                printSystemError("execv() -- ERROR");
        }
        else if (pid[i] < 0)
            printSystemError("fork() -- ERROR");

        if (close(pipesInput[i][0]) < 0)
            printSystemError("close() -- ERROR");
        if (close(pipesOutput[i][1]) < 0)
            printSystemError("close() -- ERROR");
    }
}