c - 按任务同步 fork 的子进程

Question

我对如何使用不同的同步方法感到很困惑。

我正在使用这个通用代码结构:

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);

确保所有流程首先执行 A 部分，然后是 B 部分，然后是 C 部分并且永远不会出现故障的最简单方法是什么？我的任务涉及在每个部分使用管道发送数组数据，我的印象是读取将阻塞，直到读取数组为止。我有一个从 parent 到每个 child 的 2 个管道，然后每个 child 之间有一个管道，它变得非常困惑，因为我有很多读写操作。一定有更简单的方法吗？

Answer 1

我整理了一些示例代码(在 Linux 和其他一些系统上，使用 -pthread 标志编译):

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

它不是 完全防弹的，因为一个进程可以突然进入和离开一个部分，从而将下一个部分标记为“清除”。但是，一旦您在每个部分中都有一个阻塞调用，这种情况就不会发生。

现在它是防弹的，引入了第二个信号量，确保没有进程可以在所有进程进入之前离开一个部分。

进一步编辑:现在只使用信号量。失去了只有锁定线程才能解锁的安全性，但这在这里不是一个大问题。好处是不对非线程代码使用任何 pthread 函数。在 Linux 上，您仍然需要 libpthread，其他一些系统(例如 FreeBSD)在标准 C 库中具有信号量函数。