Gabriel Rocha
Projetos

ProcessFlow

Orquestrador de processos em C que executa tarefas de forma sequencial, paralela ou encadeada por pipe.

2026Repositório

Neste projeto implementei, em C, um orquestrador de processos, o ProcessFlow. Ele cadastra tarefas (um nome ligado a um programa com seus argumentos) e as executa criando processos filhos com fork, exec, waitpid, dup2 e pipe, sem delegar para nenhum shell. Os modos de execução são sequencial, paralelo e encadeado por pipe, com redirecionamento de entrada e saída para arquivos.

Como o programa funciona

O programa tem dois modos: interativo, com o prompt processflow>, e workflow, lendo um arquivo .pf passado como argumento (cada linha é impressa antes de executar). Os dois caem no mesmo laço: leio uma linha, tiro o \n, tokenizo com strtok por espaços e tabulações e despacho pelo primeiro 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
}

As tarefas ficam em um array fixo tasks[MAX_TASKS] (limite de 104, guardado), cada uma com nome único: uma task com nome repetido é rejeitada e a definição original é mantida.

Um bug de buffering no eco do workflow

No modo workflow eu imprimo cada linha do arquivo antes de executá-la, para cumprir o requisito de cada linha lida ser impressa antes de ser processada. Por um tempo esse eco foi só um printf("%s\n", linha) sem fflush, e o programa passava nos meus testes locais. O problema é que o printf não escreve direto na saída: ele copia os dados para um buffer interno da stdio, e a escrita real (o syscall write) só acontece quando esse buffer é descarregado. Quando isso ocorre depende do modo de buffering do stdout:

A libc escolhe o modo pelo destino do descritor, não pelo conteúdo. Eu testei só no terminal (line buffered), então cada \n já descarregava e a ordem saía certa, o eco sempre vinha antes da execução. Foi o autocheck que expôs o problema: o checker io_static roda o programa com a saída capturada por um pipe, o que muda o stdout para fully buffered, e compara o esperado com o obtido. Nos três checkers (Execução simples + exit, Pipe e Redirecionamento) o veredito foi o mesmo, saída divergente: no esperado cada eco vinha antes da sua execução, mas no obtido as saídas dos filhos apareciam primeiro (eles escrevem direto no fd 1, sem passar pelo buffer da stdio) e os ecos do workflow só vinham no fim, todos juntos, no descarregamento final do buffer do pai.

Um detalhe que me confundiu no começo: o filho herda uma cópia do buffer da stdio no fork, mas o exec substitui a imagem do processo e descarta essa cópia sem descarregar. Ou seja, a inversão não vinha do filho, e sim do buffer do pai, que só era descarregado no encerramento do programa.

A correção é forçar o descarregamento logo depois do eco:

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

Com o fflush(stdout), o write acontece naquele instante, e o eco chega ao pipe antes de qualquer execução posterior, o destino sendo terminal ou não.

Cadastro de tarefas (task)

Cadastrar não executa nada, só registra. Guardo o nome, o programa e os argumentos com strdup, com a lista terminada em NULL para o 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++;

Execução sequencial e paralela (run)

run sozinho vira run sequential de uma tarefa; com o modo explícito ele aceita várias. Para cada tarefa eu faço fork: o filho executa direto (rodarNoFilho) e o pai decide o que esperar. No sequencial o waitpid é imediato, um processo por vez; no paralelo eu guardo os pids e espero o grupo todo no fim:

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
    }
}

Execução encadeada por pipe (run pipe)

Para o pipe eu crio n - 1 pipes antes de fazer os fork (um entre cada par de estágios) e cada filho conecta a ponta certa com dup2: o estágio da posição pos lê do pipe anterior e escreve no próximo. Todos os descritores extras são fechados no filho e, depois, no pai, sem isso o último estágio nunca recebe 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);
}

Assim, run pipe listar ordenar contar equivale a ls -l | sort | wc -l, com cada tarefa num processo próprio.

Redirecionamento e rodarNoFilho

input, output e append não executam, eles só gravam inputFile, outputFile e append na Task; o efeito aparece quando a tarefa roda. No filho eu abro o arquivo e aponto o descritor certo para a entrada ou a saída padrão antes do 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);
}

O output zera o append e o append o seta, então os dois são mutuamente exclusivos (um sobrescreve, o outro acrescenta).

Jobs em background (start, jobs, wait)

Além dos run, existe start para rodar em background: o pai registra o pid num vetor de Job, imprime um id e devolve o prompt. jobs consulta cada um com waitpid(WNOHANG) e wait <id> bloqueia até aquele job terminar. Detalhe: o README ainda marca workdir, start, jobs e wait como “em desenvolvimento”, mas todos já estão implementados no código:

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++;
}

Particularidades e pontos que revisaria

  1. Vetor de jobs sem limite. As tasks têm a guarda de 104, mas o start escreve em jobs[numJobs] sem checar se o vetor encheu; mais de 104 jobs em background estouram o array. Faltou o mesmo if (numJobs == MAX_TASKS) que existe no cadastro de tarefas:

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

  2. Memória nunca liberada. Tudo que vem do strdup (nome, programa, argumentos, arquivos) fica vivo até o processo sair, e redefinir input/output/append vaza a string antiga. Numa sessão curta isso é benigno, mas em um workflow longo vai acumulando.

  3. Morte por sinal não é reportada. No waitpid eu só olho WIFEXITED/WEXITSTATUS; se um filho for morto por um sinal, o erro passa despercebido. Faltaria o WIFSIGNALED.

  4. Excesso de tokens é truncado em silêncio. A tokenização para em MAX_TOKENS - 1 e ignora o resto da linha; uma linha com muitos argumentos perde o final sem nenhum aviso. O certo seria rejeitar com uma mensagem ou alocar dinamicamente.

  5. Um main grande. O despacho de comandos é uma cadeia de if no mesmo laço do main; funciona, mas o arquivo fica difícil de navegar. Vale separar cada comando em uma função.

Pontos de melhoria