synchronize forked child processes by tasks

Question

I am very confused on how to use different synchronization methods.

I am working with this general code structure:

int i, pnum;
pid_t pid;

for(i = 0; i < NUM_PROCS; ++i) {
  if ((pid = fork()) < 0) {
    perror("fork failed. . .\n");
    exit(-1);
  }      
  else if (pid == 0) {
    pnum = i;
    break;
  }
}


if (pid == 0) {
  for (i = 0; i < NUM_ITER; ++i) {

    // DO SECTION A

    // DO SECTION B

    // DO SECTION C

  }

  exit(0);
}
else {
  for (i = 0; i < NUM_ITER; ++i) {
    // DO SECTION A

    // DO SECTION B

    // DO SECTION C
  }
}

for (i = 0; i < NUM_PROCS; ++i) wait(NULL);

What is the easiest way to ensure all processes first do section A, then B, then C and never out of order? My assignment involves using pipes at each section to send array data over, and I was under the impression that a read will block until the array has been read. I have a 2 pipes from the parent to each child, and then a pipe between each child and its getting very confusing as I have ALOT of reads and writes. There must be an easier way?

Answer 1

I hacked together some example code (On Linux and maybe some other systems, compile with -pthread flag):

#define _XOPEN_SOURCE
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#include <semaphore.h>

#define SECTIONS 3
#define NUM_PROCS 5

struct sectionguards
{
    sem_t enter_clear;
    sem_t leave_clear;
    sem_t count_lock;
    int count;
};

static struct sectionguards *guards;

static void init_guards(void)
{
    int shmid, i;
    void *shmaddr;

    shmid = shmget(IPC_PRIVATE, SECTIONS * sizeof(struct sectionguards), 00600);
    if (shmid < 0)
    {
        perror("shmget");
        exit(EXIT_FAILURE);
    }

    shmaddr = shmat(shmid, NULL, 0);
    if (shmaddr == (void *)-1)
    {
        perror("shmat");
        exit(EXIT_FAILURE);
    }

    guards = shmaddr;

    for (i = 0; i < SECTIONS; ++i)
    {
        if (sem_init(&(guards[i].enter_clear), 1, 1) < 0)
        {
            perror("sem_init");
            exit(EXIT_FAILURE);
        }

        if (sem_init(&(guards[i].leave_clear), 1, 0) < 0)
        {
            perror("sem_init");
            exit(EXIT_FAILURE);
        }

        if (sem_init(&(guards[i].count_lock), 1, 1) < 0)
        {
            perror("sem_init");
            exit(EXIT_FAILURE);
        }

        guards[i].count = 0;
    }
}

static void enter(int section)
{
    int next_section = section + 1;
    if (next_section == SECTIONS) next_section = 0;

    sem_wait(&(guards[section].enter_clear));
    sem_post(&(guards[section].enter_clear));
    sem_wait(&(guards[section].count_lock));
    if (!(guards[section].count)++)
    {
        sem_wait(&(guards[next_section].enter_clear));
    }
    if (guards[section].count == NUM_PROCS)
    {
        sem_post(&(guards[section].leave_clear));
    }
    sem_post(&(guards[section].count_lock));
}

static void leave(int section)
{
    int next_section = section + 1;
    if (next_section == SECTIONS) next_section = 0;

    sem_wait(&(guards[section].leave_clear));
    sem_post(&(guards[section].leave_clear));
    sem_wait(&(guards[section].count_lock));
    if (!--(guards[section].count))
    {
        sem_post(&(guards[next_section].enter_clear));
        sem_wait(&(guards[section].leave_clear));
    }
    sem_post(&(guards[section].count_lock));
}

int main(void)
{
    int i, pnum;
    pid_t pid;

    init_guards();

    pnum = 5;
    for(i = 1; i < NUM_PROCS; ++i) {
      if ((pid = fork()) < 0) {
        perror("fork failed. . .\n");
        exit(-1);
      }      
      else if (pid == 0) {
        pnum = i;
        break;
      }
    }


    if (pid == 0) {
      for (i = 0; i < 5; ++i) {

          enter(0);
          printf("Worker %d in a\n", pnum);
          leave(0);

          enter(1);
          printf("Worker %d in b\n", pnum);
          leave(1);

          enter(2);
          printf("Worker %d in c\n", pnum);
          leave(2);

      }

      exit(0);
    }
    else {
      for (i = 0; i < 5; ++i) {

          enter(0);
          printf("Worker %d in a\n", pnum);
          leave(0);

          enter(1);
          printf("Worker %d in b\n", pnum);
          leave(1);

          enter(2);
          printf("Worker %d in c\n", pnum);
          leave(2);

      }
    }

    for (i = 1; i < NUM_PROCS; ++i) wait(NULL);
    return EXIT_SUCCESS;
}

It's not entirely bullet-proof because a process could enter and leave a section in a burst, thus marking the next section "clear". But this won't happen as soon as you have a blocking call inside each section.

Now it is bullet-proof, introducing a second semaphore that makes sure no process can leave a section before all entered it.

Further edit: using only semaphores now. The safety that only the locking thread can unlock is lost, but this is not a big concern here. The benefit is not using any pthread functions for non-threaded code. On Linux, you still need libpthread, some other systems (e.g. FreeBSD) have the semaphore functions in the standard C library.