ProcessFlow

In this project I implemented, in C, a process orchestrator: ProcessFlow. It registers tasks (a name bound to a program with its arguments) and runs them by creating child processes with fork, exec, waitpid, dup2 and pipe, without delegating to any shell. The execution modes are sequential, parallel and chained through pipes, with input and output redirection to files.

How the program works

The program has two modes: interactive, with the processflow> prompt, and workflow, reading a .pf file passed as an argument (each line is printed before being executed). Both fall into the same loop: I read a line, strip the \n, tokenize with strtok by spaces and tabs and dispatch on the first token:

while (1) {
    if (interativo) {
        printf("processflow> ");
        fflush(stdout);
    }
    if (!fgets(linha, sizeof(linha), input)) break;
    linha[strcspn(linha, "\n")] = '\0';
    if (strlen(linha) == 0) continue;
    if (!interativo) {
        printf("%s\n", linha);
        fflush(stdout);  // descarrega o buffer na hora
    }
    if (strcmp(linha, "exit") == 0) break;

    char *tokens[MAX_TOKENS];
    int n = 0;
    char *palavra = strtok(linha, " \t");
    while (palavra != NULL && n < MAX_TOKENS - 1) {
        tokens[n++] = palavra;
        palavra = strtok(NULL, " \t");
    }
    tokens[n] = NULL;
    // despacho por tokens[0]: task, run, input, output, append, workdir, start, jobs, wait
}

The tasks live in a fixed array tasks[MAX_TASKS] (limit of 104, guarded), each one with a unique name: a task with a repeated name is rejected and the original definition is kept.

A buffering bug in the workflow echo

In workflow mode I print each line of the file before executing it, to meet the requirement that every line read be printed before being processed. For a while that echo was just a printf("%s\n", linha) without fflush, and the program passed my local tests. The problem is that printf does not write straight to the output: it copies the data into an internal stdio buffer, and the actual write (the write syscall) only happens when that buffer is flushed. When that happens depends on the buffering mode of stdout:

• stderr is unbuffered, it writes right away;
• stdout on a terminal is line buffered, it flushes on every \n;
• stdout to a file or pipe is fully buffered, it only flushes when the buffer fills up, or on fflush, or at the end of the program.

The libc chooses the mode by the destination of the descriptor, not by the content. I only tested on the terminal (line buffered), so every \n already flushed and the order came out right, the echo always came before the execution. It was the autocheck that exposed the problem: the io_static checker runs the program with the output captured by a pipe, which switches stdout to fully buffered, and compares the expected output with the obtained one. In all three checkers (Simple execution + exit, Pipe and Redirection) the verdict was the same, saída divergente: in the expected output each echo came before its execution, but in the obtained output the children’s outputs appeared first (they write straight to fd 1, without going through the stdio buffer) and the workflow echoes only came at the end, all together, in the final flush of the parent’s buffer.

A detail that confused me at first: the child inherits a copy of the stdio buffer on fork, but exec replaces the process image and discards that copy without flushing. In other words, the inversion did not come from the child, but from the parent’s buffer, which was only flushed when the program exited.

The fix is to force the flush right after the echo:

if (!interativo) {
    printf("%s\n", linha);
    fflush(stdout);  
}

With fflush(stdout), the write happens at that instant, and the echo reaches the pipe before any later execution, whether the destination is a terminal or not.

Task registration (task)

Registering does not execute anything, it only records. I store the name, the program and the arguments with strdup, with the list terminated in NULL for execvp:

if (numTasks == MAX_TASKS) {
    fprintf(stderr, "Erro: limite de tasks atingido\n");
    continue;
}

Task *t = &tasks[numTasks];
t->nome = strdup(tokens[1]);
t->programa = strdup(tokens[2]);
t->args[0] = strdup(tokens[2]);       // args[0] = nome do programa
int j = 1;
for (int i = 3; i < n; i++)
    t->args[j++] = strdup(tokens[i]);
t->args[j] = NULL;                    // terminador p/ execvp
t->inputFile = NULL;
t->outputFile = NULL;
t->append = 0;
numTasks++;

Sequential and parallel execution (run)

run alone becomes run sequential of one task; with the explicit mode it accepts several. For each task I do fork: the child executes directly (rodarNoFilho) and the parent decides what to wait for. In sequential the waitpid is immediate, one process at a time; in parallel I store the pids and wait for the whole group at the end:

pid_t pid = fork();
if (pid < 0) {
    fprintf(stderr, "Erro: fork falhou.\n");
    continue;
} else if (pid == 0) {
    rodarNoFilho(t);          // o filho não volta daqui
} else if (pid > 0) {
    if (strcmp(modo, "sequential") == 0) {
        waitpid(pid, &status, 0);
        if (WIFEXITED(status) && WEXITSTATUS(status) != 0)
            fprintf(stderr, "Erro: tarefa '%s' terminou com código %d\n",
                    t->nome, WEXITSTATUS(status));
    } else if (strcmp(modo, "parallel") == 0 || strcmp(modo, "pipe") == 0) {
        pids[k++] = pid;      // espero o grupo no fim
    }
}

Chained execution through pipes (run pipe)

For the pipe I create n - 1 pipes before doing the forks (one between each pair of stages) and each child connects the right end with dup2: the stage at position pos reads from the previous pipe and writes to the next one. All extra descriptors are closed in the child and, afterwards, in the parent; without that the last stage never receives EOF:

if (strcmp(modo, "pipe") == 0)
    for (int i = 0; i < n - start - 1; i++)
        pipe(pipes[i]);
...
else if (pid == 0) {
    if (strcmp(modo, "pipe") == 0) {
        int pos = i - start;
        if (pos > 0) dup2(pipes[pos - 1][0], 0);          // lê da saída do anterior
        if (pos < n - start - 1) dup2(pipes[pos][1], 1);  // escreve para o próximo
        for (int p = 0; p < n - start - 1; p++) {
            close(pipes[p][0]);
            close(pipes[p][1]);
        }
    }
    rodarNoFilho(t);
}

So run pipe listar ordenar contar is equivalent to ls -l | sort | wc -l, with each task in its own process.

Redirection and rodarNoFilho

input, output and append do not execute, they only record inputFile, outputFile and append in the Task; the effect appears when the task runs. In the child I open the file and point the right descriptor to standard input or standard output before execvp:

void rodarNoFilho(Task *t) {
    if (t->inputFile != NULL) {
        int fd = open(t->inputFile, O_RDONLY);
        if (fd != -1) {
            dup2(fd, 0);
            close(fd);
        } else {
            perror("Erro: Arquivo não existe.");
            _exit(1);
        }
    }
    if (t->outputFile != NULL && !t->append) {
        int fd = open(t->outputFile, O_WRONLY | O_CREAT | O_TRUNC, 0644);
        if (fd != -1) {
            dup2(fd, 1);
            close(fd);
        } else {
            perror("Erro: Falha ao escrever no arquivo");
            _exit(1);
        }
    }
    if (t->outputFile != NULL && t->append) {
        int fd = open(t->outputFile, O_WRONLY | O_CREAT | O_APPEND, 0644);
        if (fd != -1) {
            dup2(fd, 1);
            close(fd);
        } else {
            perror("Erro: Falha ao escrever no arquivo");
            _exit(1);
        }
    }
    execvp(t->programa, t->args);
    perror("Erro: exec falhou.");
    _exit(1);
}

output clears append and append sets it, so the two are mutually exclusive (one overwrites, the other appends).

Background jobs (start, jobs, wait)

Besides the runs, there is start to run in the background: the parent records the pid in a vector of Job, prints an id and returns the prompt. jobs queries each one with waitpid(WNOHANG) and wait <id> blocks until that job finishes. Detail: the README still marks workdir, start, jobs and wait as “in development”, but all of them are already implemented in the code:

else if (pid > 0) {
    jobs[numJobs].id = numJobs + 1;
    jobs[numJobs].pid = pid;
    jobs[numJobs].nome = t->nome;
    jobs[numJobs].terminou = 0;
    printf("[%d] %d\n", jobs[numJobs].id, jobs[numJobs].pid);
    numJobs++;
}

Particularities and points I would revisit

1. Job vector without a limit. The tasks have the 104 guard, but start writes to jobs[numJobs] without checking whether the vector has filled up; more than 104 background jobs overflow the array. The same if (numJobs == MAX_TASKS) that exists in task registration was missing:

   // start: escreve sem checar o limite de jobs
   jobs[numJobs].id = numJobs + 1;
   jobs[numJobs].pid = pid;
   ...
   numJobs++;

2. Memory never freed. Everything that comes from strdup (name, program, arguments, files) stays alive until the process exits, and redefining input/output/append leaks the old string. In a short session that is benign, but in a long workflow it accumulates.

3. Death by signal is not reported. In waitpid I only look at WIFEXITED/WEXITSTATUS; if a child is killed by a signal, the error goes unnoticed. WIFSIGNALED would be missing.

4. Excess tokens are silently truncated. Tokenization stops at MAX_TOKENS - 1 and ignores the rest of the line; a line with too many arguments loses the end without any warning. The right thing would be to reject it with a message or allocate dynamically.

5. A large main. Command dispatch is a chain of ifs in the same loop as main; it works, but the file becomes hard to navigate. It is worth splitting each command into a function.

Improvement points

• Apply the limit guard in start (a warning when numJobs reaches the ceiling).
• Report death by signal (WIFSIGNALED) in addition to the exit code.
• Free the strdup strings when redefining files and on exit, or use a memory pool.
• Reject a line with excess tokens instead of silently truncating.
• Break main into per-command dispatch functions.