ProcHandler Class Reference

A class to manage processes for parallel processing. More...

#include <ProcHandler.h>

Public Member Functions
	ProcHandler (unsigned int nWorkerProc, bool markChildrenAsLocal=false)
	Constructor.
	~ProcHandler ()
	Destructor.
void	startInputProcess ()
	Fork and initialize an input process.
void	startWorkerProcesses ()
	Fork and initialize worker processes.
void	startOutputProcess ()
	There is no real output process, but marks current process as output.
bool	waitForAllProcesses ()
	Wait until all forked processes handled by this ProcHandler terminate.
std::set< int >	processList () const
	Return list of PIDs managed by this ProcHandler instance.

Static Public Member Functions
static bool	parallelProcessingUsed ()
	Returns true if multiple processes have been spawned, false in single-core mode.
static bool	isInputProcess ()
	Return true if the process is an input process.
static bool	isWorkerProcess ()
	Return true if the process is a worker process.
static bool	isOutputProcess ()
	Return true if the process is an output process.
static int	numEventProcesses ()
	Return number of worker processes (configured value, not current)
static std::set< int >	globalProcessList ()
	Return list of all PIDs (from all ProcHandler instances).
static int	EvtProcID ()
	Return ID of the current process.
static void	setProcessID (int processID)
	Set the process ID of this process.
static std::string	getProcessName ()
	Get a name for this process.

Private Member Functions
bool	startProc (std::set< int > *processList, const std::string &procType, int id)
	Start a new process, adding its PID to processList, and setting s_processID = id.

Private Attributes
bool	m_markChildrenAsLocal
	Anormal termination of child will not stop parent, waitForAllProcesses() returns status.
std::set< int >	m_processList
	PIDs of processes controlled by this ProcHandler.
unsigned int	m_numWorkerProcesses
	Number of worker processes controlled by this ProcHandler.

Detailed Description

A class to manage processes for parallel processing.

Definition at line 21 of file ProcHandler.h.

Constructor & Destructor Documentation

ProcHandler()

ProcHandler ( unsigned int  nWorkerProc, bool  markChildrenAsLocal = false  )

explicit

Constructor.

Children marked as local will not cause the parent process to die in case of anormal termination. Errors can be detected using the return value of waitForAllProcesses(). Note that you probably shouldn't have multiple ProcHandlers with local children running at the same time.

Definition at line 188 of file ProcHandler.cc.

                                                                          :
  m_markChildrenAsLocal(markChildrenAsLocal),
  m_numWorkerProcesses(nWorkerProc)
{
  if ((int)nWorkerProc > s_numEventProcesses)
    s_numEventProcesses = nWorkerProc;
 
  if (!pidListEmpty())
    B2FATAL("Constructing ProcHandler after forking is not allowed!");
 
  //s_pidVector size shouldn't be changed once processes are forked (race condition)
  s_pidVector.reserve(s_pidVector.size() + nWorkerProc + 2);
  s_pids = s_pidVector.data();
  setsid();
 
}

Member Function Documentation

EvtProcID()

int EvtProcID ( )

static

Return ID of the current process.

Return values mean: -1: no parallel processing used <10000 one of the worker processes (between input and output paths). In range 0..Environment::getInstance().getNumberProcesses()-1. >=10000 input path >=20000 output path

Definition at line 248 of file ProcHandler.cc.

{ return s_processID; }

getProcessName()

std::string getProcessName ( )

static

Get a name for this process.

(input, event, output...).

Definition at line 252 of file ProcHandler.cc.

{
  if (isWorkerProcess())
    return "worker";
  if (isInputProcess())
    return "input";
  if (isOutputProcess())
    return "output";
 
  //shouldn't happen
  return "???";
}

globalProcessList()

std::set< int > globalProcessList ( )

static

Return list of all PIDs (from all ProcHandler instances).

Definition at line 239 of file ProcHandler.cc.

{
  return std::set<int>(s_pidVector.begin(), s_pidVector.end());
}

isInputProcess()

bool isInputProcess ( )

static

Return true if the process is an input process.

Definition at line 228 of file ProcHandler.cc.

{ return (s_processID >= 10000 and s_processID < 20000) or GlobalProcHandler::isInputProcess(); }

isOutputProcess()

bool isOutputProcess ( )

static

Return true if the process is an output process.

Definition at line 232 of file ProcHandler.cc.

{ return s_processID >= 20000 or GlobalProcHandler::isOutputProcess(); }

isWorkerProcess()

bool isWorkerProcess ( )

static

Return true if the process is a worker process.

Definition at line 230 of file ProcHandler.cc.

{ return (parallelProcessingUsed() and s_processID < 10000) or GlobalProcHandler::isWorkerProcess(); }

numEventProcesses()

int numEventProcesses ( )

static

Return number of worker processes (configured value, not current)

Definition at line 234 of file ProcHandler.cc.

{
  return s_numEventProcesses;
}

parallelProcessingUsed()

bool parallelProcessingUsed ( )

static

Returns true if multiple processes have been spawned, false in single-core mode.

Definition at line 226 of file ProcHandler.cc.

{ return s_processID != -1 or GlobalProcHandler::parallelProcessingUsed(); }

processList()

std::set< int > processList

(

)

const

Return list of PIDs managed by this ProcHandler instance.

Definition at line 243 of file ProcHandler.cc.

{
  return m_processList;
}

setProcessID()

void setProcessID ( int  processID )

static

Set the process ID of this process.

Definition at line 250 of file ProcHandler.cc.

{ s_processID = processID; }

startInputProcess()

void startInputProcess

(

)

Fork and initialize an input process.

Definition at line 207 of file ProcHandler.cc.

{
  startProc(&m_processList, "input", 10000);
}

startOutputProcess()

void startOutputProcess

(

)

There is no real output process, but marks current process as output.

Definition at line 220 of file ProcHandler.cc.

{
  if (s_processID == -1)
    s_processID = 20000;
}

startProc()

bool startProc ( std::set< int > *  processList, const std::string &  procType, int  id  )

private

Start a new process, adding its PID to processList, and setting s_processID = id.

Returns true in child process.

Definition at line 154 of file ProcHandler.cc.

{
  EventProcessor::installSignalHandler(SIGCHLD, sigChldHandler);
 
  fflush(stdout);
  fflush(stderr);
  PyOS_BeforeFork();
  pid_t pid = fork();
  if (pid > 0) {   // Mother process
    PyOS_AfterFork_Parent();
    if (m_markChildrenAsLocal)
      pid = -pid;
    processList->insert(pid);
    addPID(pid);
    B2INFO("ProcHandler: " << procType << " process forked. pid = " << pid);
    fflush(stdout);
  } else if (pid < 0) {
    B2FATAL("fork() failed: " << strerror(errno));
  } else {
    //do NOT handle SIGCHLD in forked processes!
    EventProcessor::installSignalHandler(SIGCHLD, SIG_IGN);
 
    s_processID = id;
    //Reset some python state: signals, threads, gil in the child
    PyOS_AfterFork_Child();
    //InputController becomes useless in child process
    InputController::resetForChildProcess();
    //die when parent dies
    prctl(PR_SET_PDEATHSIG, SIGHUP);
    return true;
  }
  return false;
}

startWorkerProcesses()

void startWorkerProcesses

(

)

Fork and initialize worker processes.

Definition at line 212 of file ProcHandler.cc.

{
  for (unsigned int i = 0; i < m_numWorkerProcesses; i++) {
    if (startProc(&m_processList, "worker", i))
      break; // in child process
  }
}

waitForAllProcesses()

bool waitForAllProcesses

(

)

Wait until all forked processes handled by this ProcHandler terminate.

In case the markChildrenAsLocal option was set when a process was started and this process terminated anormaly (via signal or non-zero exit code), the return value is set to false. Otherwise, true is returned.

Definition at line 266 of file ProcHandler.cc.

{
  bool ok = true;
  while (!m_processList.empty()) {
    for (int pid : m_processList) {
      //once a process is gone from the global list, remove them from our own, too.
      if (findPID(pid) == 0) {
        m_processList.erase(pid);
        if (m_markChildrenAsLocal and pid < 0 and s_localChildrenWithErrors != 0) {
          ok = false;
          s_localChildrenWithErrors--;
        }
        break;
      }
    }
 
    usleep(100);
  }
  return ok;
}

Member Data Documentation

m_markChildrenAsLocal

bool m_markChildrenAsLocal

private

Anormal termination of child will not stop parent, waitForAllProcesses() returns status.

Definition at line 88 of file ProcHandler.h.

m_numWorkerProcesses

unsigned int m_numWorkerProcesses

private

Number of worker processes controlled by this ProcHandler.

Definition at line 90 of file ProcHandler.h.

m_processList

std::set<int> m_processList

private

PIDs of processes controlled by this ProcHandler.

Definition at line 89 of file ProcHandler.h.

---

The documentation for this class was generated from the following files:

• framework/pcore/include/ProcHandler.h
• framework/pcore/src/ProcHandler.cc