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

#include <GlobalProcHandler.h>

Collaboration diagram for GlobalProcHandler:

Static Public Member Functions
static void	initialize (unsigned int nWorkerProc)
	Create a new process handler, which will handle nWorkerProc processes.

static bool	startInputProcess ()
	Fork and initialize an input process.

static bool	startWorkerProcesses (unsigned int numProcesses)
	Fork and initialize as many worker processes as requested.

static bool	startOutputProcess (bool local=false)
	Fork and initialize an output process.

static bool	startProxyProcess ()
	Fork and initialize a proxy process.

static bool	startMonitoringProcess ()
	Fork and initialize a monitoring process.

static void	waitForAllProcesses ()
	Wait until all forked processes handled by this GlobalProcHandler.

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 of type c_Input.

static bool	isWorkerProcess ()
	Return true if the process is of type c_Worker.

static bool	isOutputProcess ()
	Return true if the process is of type c_Output.

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

static int	EvtProcID ()
	Get the ID of this process. Attention: this ID may not be a stable API feature.

static std::string	getProcessName ()
	Get a human readable name for this process. (input, event, output...).

static void	killAllProcesses ()
	Hard kill all processes.

static const std::vector< int > &	getPIDList ()
	Return the PID list handled by the running GlobalProcHandler. This PID list is updated using the signal handlers and the startup of processes.

static bool	isProcess (ProcType procType)
	Return true if the process is of type procType.

static ProcType	getProcType (int pid)
	Return the proc type of this process.

Static Private Member Functions
static bool	startProc (ProcType procType, int id)
	Start a new process, sets the type and id and returns true if in this new process.

static void	addPID (int pid)
	Add a new PID. Is called when forking.

static bool	findPID (int pid)
	Find a PID in the list and return true, if found.

static void	removePID (int pid)
	Remove a PID from the list by setting it to 0.

static void	clearPIDs ()
	Remove all PIDs.

static bool	pidListEmpty ()
	Check if the PID list is empty (only 0).

static void	childSignalHandler (int)
	This function is called on SIG_CLD.

Static Private Attributes
static ProcType	s_procType = ProcType::c_Init
	Our current proc type.

static int	s_processID = -1
	Our current process id.

static int	s_numEventProcesses = 0
	How many processes are handled in this GlobalProcHandler.

static std::vector< int >	s_pidVector
	global list of PIDs managed by GlobalProcHandler.

static std::map< int, ProcType >	s_startedPIDs
	Which PIDs were started with which types.

Detailed Description

A class to manage processes for parallel processing.

Definition at line 22 of file GlobalProcHandler.h.

Member Function Documentation

◆ addPID()

void addPID ( int pid )

staticprivate

Add a new PID. Is called when forking.

Definition at line 83 of file GlobalProcHandler.cc.

{
  // if possible, insert pid into gap in list
  for (int& pid : s_pidVector) {
    if (pid == 0) {
      pid = newPID;
      return;
    }
  }
 
  B2FATAL("PID vector at capacity. This produces a race condition, make sure GlobalProcHandler is created early.");
}

◆ childSignalHandler()

void childSignalHandler ( int )

staticprivate

This function is called on SIG_CLD.

Definition at line 40 of file GlobalProcHandler.cc.

{
  while (!GlobalProcHandler::pidListEmpty()) {
    int status;
    int pid = waitpid(-1, &status, WNOHANG);
    if (pid == -1) {
      if (errno == EINTR) {
        continue; // interrupted, try again
      } else if (errno == ECHILD) {
        // We don't have any child processes?
        EventProcessor::writeToStdErr("\n Called waitpid() without any children left. This shouldn't happen and and indicates a problem.\n");
        // actually, this is ok in case we already called waitpid() somewhere else. (but we don't do that...)
        // In case I want to avoid this, waitid() and WNOWAIT might help, but require libc >= 2.12 (not present in SL5)
        GlobalProcHandler::clearPIDs();
        return;
      } else {
        // also shouldn't happen
        EventProcessor::writeToStdErr("\nwaitpid() failed.\n");
      }
    } else if (pid == 0) {
      // should not happen because of waitpid(-1,...)
      // further children exist, but no state change yet
      break;
    } else {
      // state change
      // get signed PID
      if (not GlobalProcHandler::findPID(pid)) {
        // unknown child process died, ignore
        continue;
      }
 
      // errors?
      if (WIFSIGNALED(status) or (WIFEXITED(status) and WEXITSTATUS(status) != 0)) {
        EventProcessor::writeToStdErr("\nSub-process exited with non-zero exit status. Please check other log messages for details.\n");
      }
 
      // remove pid from global list
      GlobalProcHandler::removePID(pid);
    }
  }
}

◆ clearPIDs()

void clearPIDs ( )

staticprivate

Remove all PIDs.

Definition at line 111 of file GlobalProcHandler.cc.

{
  std::fill(s_pidVector.begin(), s_pidVector.end(), 0);
}

◆ EvtProcID()

int EvtProcID ( )

static

Get the ID of this process. Attention: this ID may not be a stable API feature.

Definition at line 238 of file GlobalProcHandler.cc.

{
  return s_processID;
}

◆ findPID()

bool findPID ( int pid )

staticprivate

Find a PID in the list and return true, if found.

Definition at line 96 of file GlobalProcHandler.cc.

{
  return std::find(s_pidVector.begin(), s_pidVector.end(), pid) != s_pidVector.end();
}

◆ getPIDList()

const std::vector< int > & getPIDList ( )

static

Return the PID list handled by the running GlobalProcHandler. This PID list is updated using the signal handlers and the startup of processes.

Definition at line 257 of file GlobalProcHandler.cc.

{
  return s_pidVector;
}

◆ getProcessName()

std::string getProcessName ( )

static

Get a human readable name for this process. (input, event, output...).

Definition at line 221 of file GlobalProcHandler.cc.

{
  if (isProcess(ProcType::c_Worker))
    return "worker";
  if (isProcess(ProcType::c_Input))
    return "input";
  if (isProcess(ProcType::c_Output))
    return "output";
  if (isProcess(ProcType::c_Init))
    return "init";
  if (isProcess(ProcType::c_Monitor))
    return "monitor";
 
  //shouldn't happen
  return "???";
}

◆ getProcType()

ProcType getProcType ( int pid )

static

Return the proc type of this process.

Definition at line 262 of file GlobalProcHandler.cc.

{
  const auto procTypeIt = s_startedPIDs.find(pid);
  if (procTypeIt == s_startedPIDs.end()) {
    B2FATAL("Asking for a non-existing PID");
  }
  return procTypeIt->second;
}

◆ initialize()

void initialize ( unsigned int nWorkerProc )

static

Create a new process handler, which will handle nWorkerProc processes.

Definition at line 126 of file GlobalProcHandler.cc.

{
  B2ASSERT("Constructing GlobalProcHandler after forking is not allowed!", pidListEmpty());
 
  s_numEventProcesses = nWorkerProc;
 
  // s_pidVector size shouldn't be changed once processes are forked (race condition)
  s_pidVector.resize(s_pidVector.size() + nWorkerProc + 3, 0); // num worker + input + output + proxy
}

◆ isInputProcess()

bool isInputProcess ( )

static

Return true if the process is of type c_Input.

Definition at line 286 of file GlobalProcHandler.cc.

{
  return isProcess(ProcType::c_Input);
}

◆ isOutputProcess()

bool isOutputProcess ( )

static

Return true if the process is of type c_Output.

Definition at line 276 of file GlobalProcHandler.cc.

{
  return isProcess(ProcType::c_Output);
}

◆ isProcess()

bool isProcess ( ProcType procType )

static

Return true if the process is of type procType.

Definition at line 177 of file GlobalProcHandler.cc.

{
  return (procType == s_procType);
}

◆ isWorkerProcess()

bool isWorkerProcess ( )

static

Return true if the process is of type c_Worker.

Definition at line 281 of file GlobalProcHandler.cc.

{
  return isProcess(ProcType::c_Worker);
}

◆ killAllProcesses()

void killAllProcesses ( )

static

Hard kill all processes.

Definition at line 243 of file GlobalProcHandler.cc.

{
  for (int& pid : s_pidVector) {
    if (pid != 0) {
      if (kill(pid, SIGKILL) >= 0) {
        B2DEBUG(100, "hard killed process " << pid);
      } else {
        B2DEBUG(100, "no process " << pid << " found, already gone?");
      }
      pid = 0;
    }
  }
}

◆ numEventProcesses()

int numEventProcesses ( )

static

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

Definition at line 271 of file GlobalProcHandler.cc.

{
  return s_numEventProcesses;
}

◆ parallelProcessingUsed()

bool parallelProcessingUsed ( )

static

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

Definition at line 172 of file GlobalProcHandler.cc.

{
  return s_processID != -1;
}

◆ pidListEmpty()

bool pidListEmpty ( )

staticprivate

Check if the PID list is empty (only 0).

Definition at line 116 of file GlobalProcHandler.cc.

{
  for (const int& pid : s_pidVector) {
    if (pid != 0) {
      return false;
    }
  }
  return true;
}

◆ removePID()

void removePID ( int pid )

staticprivate

Remove a PID from the list by setting it to 0.

Definition at line 101 of file GlobalProcHandler.cc.

{
  for (int& pid : s_pidVector) {
    if (pid == oldPID) {
      pid = 0;
      return;
    }
  }
}

◆ startInputProcess()

bool startInputProcess ( )

static

Fork and initialize an input process.

Definition at line 141 of file GlobalProcHandler.cc.

{
  return startProc(ProcType::c_Input, 10000);
}

◆ startMonitoringProcess()

bool startMonitoringProcess ( )

static

Fork and initialize a monitoring process.

Definition at line 166 of file GlobalProcHandler.cc.

{
  s_procType = ProcType::c_Monitor;
  return true;
}

◆ startOutputProcess()

bool startOutputProcess ( bool local = false )

static

Fork and initialize an output process.

If local is true, do not fork.

Definition at line 156 of file GlobalProcHandler.cc.

{
  if (local) {
    s_procType = ProcType::c_Output;
    return true;
  } else {
    return (startProc(ProcType::c_Output, 20000));
  }
}

◆ startProc()

bool startProc	(	ProcType	procType,
		int	id
	)

staticprivate

Start a new process, sets the type and id and returns true if in this new process.

Also adds the signal handling.

Definition at line 182 of file GlobalProcHandler.cc.

{
  EventProcessor::installSignalHandler(SIGCHLD, childSignalHandler);
 
  fflush(stdout);
  fflush(stderr);
  pid_t pid = fork();
  if (pid > 0) {
    // Mother process
    addPID(pid);
    s_startedPIDs[pid] = procType;
    fflush(stdout);
  } else if (pid < 0) {
    B2FATAL("fork() failed: " << strerror(errno));
  } else {
    // Child process
    // do NOT handle SIGCHLD in forked processes!
    EventProcessor::installSignalHandler(SIGCHLD, SIG_IGN);
 
    s_procType = procType;
 
    if (id == 0)
      s_processID = getpid();
    else
      s_processID = id;
 
    ProcHandler::setProcessID(s_processID);    // Interface to existing ProcHandler
 
    // 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;
}

◆ startProxyProcess()

bool startProxyProcess ( )

static

Fork and initialize a proxy process.

Definition at line 136 of file GlobalProcHandler.cc.

{
  return startProc(ProcType::c_Proxy, 30000);
}

◆ startWorkerProcesses()

bool startWorkerProcesses ( unsigned int numProcesses )

static

Fork and initialize as many worker processes as requested.

Definition at line 146 of file GlobalProcHandler.cc.

{
  for (unsigned int i = 0; i < numProcesses; i++) {
    if (startProc(ProcType::c_Worker, 0)) {
      return true;
    }
  }
  return false;
}

◆ waitForAllProcesses()

void waitForAllProcesses ( )

static

Wait until all forked processes handled by this GlobalProcHandler.

Definition at line 291 of file GlobalProcHandler.cc.

{
  while (true) {
    if (pidListEmpty()) {
      return;
    }
 
    std::this_thread::sleep_for(std::chrono::milliseconds(1));
  }
}

Member Data Documentation

◆ s_numEventProcesses

int s_numEventProcesses = 0

staticprivate

How many processes are handled in this GlobalProcHandler.

Definition at line 77 of file GlobalProcHandler.h.

◆ s_pidVector

std::vector< int > s_pidVector

staticprivate

global list of PIDs managed by GlobalProcHandler.

Definition at line 80 of file GlobalProcHandler.h.

◆ s_processID

int s_processID = -1

staticprivate

Our current process id.

Definition at line 75 of file GlobalProcHandler.h.

◆ s_procType

ProcType s_procType = ProcType::c_Init

staticprivate

Our current proc type.

Definition at line 73 of file GlobalProcHandler.h.

◆ s_startedPIDs

std::map< int, ProcType > s_startedPIDs

staticprivate

Which PIDs were started with which types.

Definition at line 82 of file GlobalProcHandler.h.

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

framework/pcore/include/GlobalProcHandler.h
framework/pcore/src/GlobalProcHandler.cc

Static Public Member Functions

Static Private Member Functions

Static Private Attributes

Detailed Description

Member Function Documentation

◆ addPID()

◆ childSignalHandler()

◆ clearPIDs()

◆ EvtProcID()

◆ findPID()

◆ getPIDList()

◆ getProcessName()

◆ getProcType()

◆ initialize()

◆ isInputProcess()

◆ isOutputProcess()

◆ isProcess()

◆ isWorkerProcess()

◆ killAllProcesses()

◆ numEventProcesses()

◆ parallelProcessingUsed()

◆ pidListEmpty()

◆ removePID()

◆ startInputProcess()

◆ startMonitoringProcess()

◆ startOutputProcess()

◆ startProc()

◆ startProxyProcess()

◆ startWorkerProcesses()

◆ waitForAllProcesses()

Member Data Documentation

◆ s_numEventProcesses

◆ s_pidVector

◆ s_processID

◆ s_procType

◆ s_startedPIDs