C++ 管理进程

Question

我有一个进程创建了 10 个子进程，它们都共享一个内存段。 10个子进程可以并发运行，但是我想让父亲在这段时间 sleep 。 children 完成后就去 sleep 了。当所有 child 都完成后，父亲进行一些计算并叫醒 children 进行另一轮(然后他再次休眠，直到所有 child 都完成计算)。我想我可以在以 check_children_finished() 为条件的 while 循环内暂停 ()。 children 知道通过 kill() 向 parent 发出信号，他们有他的 pid(作为参数发送给他们)，实际上父亲也有 child 的 pid(存储在共享内存中)这是一些代码 -

---

父代码:

---

#include <csignal>
#include <cstdio>
#include <cstdlib>
#include <sys/types.h>

#include <sys/ipc.h> 
#include <sys/shm.h> 
#include <sys/stat.h>
#include <unistd.h>
#include <cstring>
#include <string>
#include <sstream>
#include <iostream>

using namespace std;

static void sig_usr(int);
static volatile sig_atomic_t *sharedArray;
static int segmentId; 
int number_of_children = 10;

bool check_children_finished(){
  for (int i=1; i<number_of_children; i++)
    if (sharedArray[i*2] == -1)
      return false;
  return true;
}

void sig_usr (int signo)
{
  /*
  // This is signals handler (when a signal fires this is what runs)
  // we expect SIGUSR1 from the child process
  if(signo == SIGUSR1)
    cout << "(parent: " << (int)getpid() << ") --- caught SIGUSR1" << endl;
  if (signo == SIGUSR2){
    cout << "(parent: " << (int)getpid() << ") --- caught SIGUSR2" << endl;
  }else
    perror("unexpected signal fired");
  //  return;
  */
}


int main()
{

  // size of the shared memory array
  int arrSize = 100;
  const int shareSize = sizeof(sig_atomic_t) * (arrSize);
  /* Allocate shared memory segment */  
  segmentId = shmget(IPC_PRIVATE, shareSize, S_IRUSR | S_IWUSR); 
  sharedArray = (sig_atomic_t *) shmat(segmentId, NULL, 0);

  // feel shared memory with -1
  for (int i=0; i<arrSize; i++)
    sharedArray[i] = -1;

  // binding SIGUSR1 & SIGUSR2 to sig_usr method
  // we need to override SIGUSR2 because of group signaling
  signal(SIGUSR1, sig_usr);
  //  signal(SIGUSR2,sig_usr);

  fprintf(stderr, "\n (parnet) myPid=%d ; segId=%d\n",(int)getpid(), segmentId);
  sharedArray[0] = 123;
  int kids = 0; // this is the number of child processes
  // we send to the child (as shell parameters) the parent pid, shared segment address , and index to the shared memory(this is where he will write his pid and in index+1 the heuristic value)
  char* kidsParams[5];

  // takes care of param[0]=command to run
  string exec_line = "./child";
  kidsParams[0] = new char[exec_line.size()+1];
  memcpy(kidsParams[0], exec_line.c_str(), exec_line.size());

  // takes care of param[1]=parent pid
  kidsParams[1] = new char[100]; // = malloc(100*sizeof(char));
  sprintf( kidsParams[1],"%d",(int)getpid());

  // takes care of param[2]=shared mem segment address
  kidsParams[2] = new char[100];
  sprintf( kidsParams[2],"%d",segmentId);

  // takes care of param[3]=the child private index in shared mem
  kidsParams[3] = new char[100];

  kidsParams[4] = NULL;  // needed as end of array
  int index = 0;
  for(; kids<number_of_children; kids++) {
    sprintf( kidsParams[3],"%d",index);
    index+=2;
    pid_t childpid = fork();
    if(childpid==0){
      execv(kidsParams[0],kidsParams);
    }
   }
  cout << "(parent) --- just finished creating " << number_of_children << " kids" << endl;
  cout << "(parent) entering to while {...} pause" << endl;
  for (int i=0; i<number_of_children; i++)
    cout << "[" << i << "] = " << sharedArray[i];
  cout << endl;
  // going to sleep --- here I want while loop with conditioning that all children finished
  while ( ! check_children_finished() ) {
    cout << "(parent) now will signal the group" << endl;
    // killpg sends signal to the group (all the children). note that the group has the same pid as the father
    killpg(getpid(),SIGUSR2);
    cout << "(parent) just finished signaling the group" << endl;
    pause();
    for (int i=0; i<number_of_children; i++)
       cout << "[" << i << "] = " << sharedArray[i];
    cout << endl;
  }
  cout << "(parent) exited the while{...} paused" << endl;


  // removes shared memory
  //  shmdt (sharedArray);  
  //  shmctl (segmentId, IPC_RMID, NULL);  
  // note that all children must also shmctl (...IPC_RMID...);

}

---

这是一个子代码:

---

(same includes...)
using namespace std;

// declare the function proptotype (needed in signal function)
static void sig_usr(int);

// handles the signal (what happens when the signal fires) --- here I want to solve the search problem
void sig_usr (int signo)
{
  /*
  if(signo == SIGUSR1){
    cout << "(child: " << (int)getpid() << ") --- caught SIGUSR1" << endl;
  }else if(signo == SIGUSR2){
    cout << "(child: " << (int)getpid() << ") --- caught SIGUSR2" << endl;
  }else
    perror("eerrrr");
  //  return;
  */

}

int main(int argc, char** argv){
  // binding the signal to the handler
  signal(SIGUSR2,sig_usr);
  int segmentId;  
  volatile sig_atomic_t *sharedArray ;
  int myIndex;
  int myData =  5; 
  int parentPid;
  // read parameters
  parentPid = atoi(argv[1]);
  segmentId = atoi(argv[2]);
  myIndex = atoi(argv[3]);

  // declare a pointer to the shared memory
  sharedArray = (sig_atomic_t *) shmat(segmentId, NULL, 0);
  sharedArray[myIndex] =(int)getpid();
  sharedArray[myIndex+1] = myData;
  //  fprintf(stderr, "My Group Pid(child): %d\n",(int)getpgrp());
  cout << "(child: " << (int)getpid() << ") --- going to sleep" << endl;
  pause();
  cout << "(child: " << (int)getpid() << ") --- I woke up" << endl;
  //calc data

  //fprintf(stderr, "My Pid(child): %d\n",(int)getpid());
  //fprintf(stderr, "I got %d (child)\n",sharedArray[0] );


  // this signals the father
  kill(parentPid,SIGUSR1);
  cout << "fired SIGUSR1"<< endl;
}

---

这是一个典型的输出:

---

 (parnet) myPid=3104 ; segId=22872080
(parent) --- just finished creating 10 kids
(parent) entering to while {...} pause
[0] = 123[1] = -1[2] = -1[3] = -1[4] = -1[5] = -1[6] = -1[7] = -1[8] = -1[9] = -1
(parent) now will signal the group
User defined signal 2

---

有时我会得到类似的东西:

---

 (parnet) myPid=3126 ; segId=22937618
(child: 3129) --- going to sleep
(child: 3127) --- going to sleep
(parent) --- just finished creating 10 kids
(parent) entering to while {...} pause
[0] = 3127[1] = 5[2] = 3128[3] = 5[4] = 3129[5] = 5[6] = -1[7] = -1[8] = -1[9] = -1
(parent) now will signal the group
User defined signal 2
(child: 3127) --- I woke up
fired SIGUSR1
(child: 3128) --- going to sleep
(child: 3132) --- going to sleep
(child: 3129) --- I woke up
fired SIGUSR1

谁能提出解决方案？谢谢!-- 里昂

Answer 1

我建议在监控进程和其他工作进程之间使用管道，并使用简单的文本协议(protocol)进行控制通信。

^{(因为管道上的文本协议(protocol)比信号更可靠——信号可以“丢失”或“合并”)}

所以，在 fork worker 之前，我会调用(例如 10 次，或者如果你想要他们两种方式，可能是 2*10 次)pipe(2)创建控制通信。然后你可以使用 ppoll(2) (或只是 poll )来复用管道。

但是您是否考虑过使用现有框架，例如Open-MPI (一个实现 MPI =“消息传递接口(interface)”)。这并不意味着放弃使用共享内存段(只需将 MPI 用于控制和同步问题)。或者也许使用 OpenMP或者只是 p-threads ？

MPI 在高性能数值计算中的应用非常广泛，许多 super 计算机都拥有它(使用专门的硬件来实现非常快速的消息传递)。

^{(当然， super 计算机是集群，因此您实际上并没有在数千个内核之间共享内存；您将通过在代码中仅使用 MPI 来利用它们...)}

(您也可以通过 OpenCL 调查使用 GPGPU)