ProcessMonitor Class Reference

Class to monitor all started framework processes (input, workers, output), kill them if requested and handle the signals from the OS. More...

#include <ProcessMonitor.h>

Public Member Functions
void	subscribe (const std::string &pubSocketAddress, const std::string &subSocketAddress, const std::string &controlSocketAddress)
	Start listening for process management information on the given address.
void	waitForRunningInput (int timeout)
	Block until either the input process is running or the timeout (in seconds) is raised.
void	waitForRunningOutput (int timeout)
	Block until either the output process is running or the timeout (in seconds) is raised.
void	waitForRunningWorker (int timeout)
	Block until either the worker process is running or the timeout (in seconds) is raised.
void	killProcesses (unsigned int timeout)
	Ask all processes to terminate. If not, kill them after timeout seconds.
void	initialize (unsigned int requestedNumberOfWorkers)
	Init the processing with that many workers.
void	terminate ()
	Terminate the processing.
void	reset ()
	Reset the internal state.
void	checkMulticast (int timeout=0)
	check multicast for messages and kill workers if requested
void	checkChildProcesses ()
	check the child processes, if one has died
void	checkSignals (int g_signalReceived)
	check if we have received any signal from the user or OS. Kill the processes if not SIGINT.
bool	hasEnded () const
	If we have received a SIGINT signal or the last process is gone, we can end smoothly.
unsigned int	needMoreWorkers () const
	Compare our current list of workers of how many we want to have.
bool	hasWorkers () const
	Check if there is at least one running worker.

Private Member Functions
unsigned int	processesWithType (const ProcType &procType) const
	Cound the number of processes with a certain type.
template<class ASocket >
void	processMulticast (const ASocket &socket)
	Process a message from the multicast.

Private Attributes
ZMQClient	m_client
	The client used for message processing.
unsigned int	m_requestedNumberOfWorkers = 0
	How many workers we should request to start.
std::map< int, ProcType >	m_processList
	The current list of pid -> process types (to be compared to the proc handler)
bool	m_hasEnded = false
	Someone requested us to end the processing.
StreamHelper	m_streamer
	The data store streamer.
bool	m_receivedStatistics = false
	Did we already receive the statistics?

Detailed Description

Class to monitor all started framework processes (input, workers, output), kill them if requested and handle the signals from the OS.

Is used in the ZMQ event processor and can only be used together with ZMQ.

Definition at line 28 of file ProcessMonitor.h.

Member Function Documentation

checkChildProcesses()

void checkChildProcesses

(

)

check the child processes, if one has died

Definition at line 266 of file ProcessMonitor.cc.

{
  if (hasEnded()) {
    return;
  }
 
  // Copy is intended, as we do not want the signal handler to change our list
  std::vector<int> currentProcessList = GlobalProcHandler::getPIDList();
  // Check for processes, which where there last time but are gone now (so they died)
  for (auto iter = m_processList.begin(); iter != m_processList.end();) {
    const auto& pair = *iter;
 
    B2ASSERT("This pid should not be in our list!", pair.first != 0);
    //once a process is gone from the global list, remove them from our own, too.
    if (std::find(currentProcessList.begin(), currentProcessList.end(), pair.first) != currentProcessList.end()) {
      ++iter;
      continue;
    }
 
    // if the process has gone down properly, it should now be set to "Stopped"
    if (pair.second == ProcType::c_Input) {
      B2ERROR("An input process has died unexpected! Need to go down.");
      m_hasEnded = true;
      return;
    } else if (pair.second == ProcType::c_Output) {
      B2ERROR("An output process has died unexpected! Need to go down.");
      m_hasEnded = true;
      return;
    } else if (pair.second == ProcType::c_Proxy) {
      B2ERROR("A proxy process has died unexpected! Need to go down.");
      m_hasEnded = true;
      return;
    } else if (pair.second == ProcType::c_Worker) {
      //      B2WARNING("A worker process has died unexpected. If you have requested, I will now restart the workers.");
      B2ERROR("A worker process has died unexpectedly.");
      B2ASSERT("A worker died but none was present?", processesWithType(ProcType::c_Worker) != 0);
      auto pcbMulticastMessage = ZMQMessageFactory::createMessage(EMessageTypes::c_deleteWorkerMessage, pair.first);
      m_client.publish(std::move(pcbMulticastMessage));
    } else if (pair.second == ProcType::c_Stopped) {
      B2DEBUG(30, "An children process has died expectedly.");
    }
 
    iter = m_processList.erase(iter);
  }
 
  m_hasEnded = false;
  // The processing should be finished, if...
  if (m_processList.empty()) {
    // .. there is no process around anymore
    m_hasEnded = true;
  } else if (m_processList.size() == 1 and m_processList.begin()->second == ProcType::c_Output) {
    // ... the single remaining process is an output process (can happen, if we do not use the event backup)
    m_hasEnded = true;
  } else {
    // ... there are only workers or stopped processes around
    m_hasEnded = true;
    for (const auto& pair : m_processList) {
      if (pair.second == ProcType::c_Input or pair.second == ProcType::c_Output) {
        m_hasEnded = false;
        break;
      }
    }
  }
 
  if (m_hasEnded) {
    B2DEBUG(30, "No input and no output process around. Will go home now!");
  }
}

checkMulticast()

void checkMulticast

(

int

timeout = 0

)

check multicast for messages and kill workers if requested

Definition at line 210 of file ProcessMonitor.cc.

{
  if (hasEnded()) {
    return;
  }
 
  const auto multicastAnswer = [this](const auto & socket) {
    processMulticast(socket);
    return false;
  };
  m_client.pollMulticast(timeout * 1000, multicastAnswer);
}

checkSignals()

void checkSignals

(

int

g_signalReceived

)

check if we have received any signal from the user or OS. Kill the processes if not SIGINT.

Definition at line 335 of file ProcessMonitor.cc.

{
  if (g_signalReceived > 0) {
    B2DEBUG(30, "Received signal " << g_signalReceived);
    m_hasEnded = true;
  }
}

hasEnded()

bool hasEnded

(

)

const

If we have received a SIGINT signal or the last process is gone, we can end smoothly.

Definition at line 343 of file ProcessMonitor.cc.

{
  return m_hasEnded;
}

hasWorkers()

bool hasWorkers

(

)

const

Check if there is at least one running worker.

Definition at line 348 of file ProcessMonitor.cc.

{
  return processesWithType(ProcType::c_Worker) > 0;
}

initialize()

void initialize

(

unsigned int

requestedNumberOfWorkers

)

Init the processing with that many workers.

Definition at line 205 of file ProcessMonitor.cc.

{
  m_requestedNumberOfWorkers = requestedNumberOfWorkers;
}

killProcesses()

void killProcesses

(

unsigned int

timeout

)

Ask all processes to terminate. If not, kill them after timeout seconds.

Definition at line 106 of file ProcessMonitor.cc.

{
  B2ASSERT("Only the monitoring process is allowed to kill processes", GlobalProcHandler::isProcess(ProcType::c_Monitor)
           or GlobalProcHandler::isProcess(ProcType::c_Init));
 
  if (not m_processList.empty() and m_client.isOnline()) {
    B2DEBUG(30, "Try to kill the processes gently...");
    // Try to kill them gently...
    auto pcbMulticastMessage = ZMQMessageFactory::createMessage(EMessageTypes::c_terminateMessage);
    m_client.publish(std::move(pcbMulticastMessage));
 
    checkChildProcesses();
 
    const auto multicastAnswer = [this](const auto & socket) {
      processMulticast(socket);
      for (const auto& pair : m_processList) {
        if (pair.second != ProcType::c_Stopped) {
          B2DEBUG(30, "Process pid " << pair.first << " of type " << static_cast<char>(pair.second) << " is still alive");
          return true;
        }
      }
      return not m_receivedStatistics;
    };
 
    bool allProcessesStopped = true;
    for (const auto& pair : m_processList) {
      if (pair.second != ProcType::c_Stopped) {
        allProcessesStopped = false;
        break;
      }
    }
 
    if (not allProcessesStopped) {
      B2DEBUG(30, "Start waiting for processes to go down.");
      m_client.pollMulticast(timeout, multicastAnswer);
      B2DEBUG(30, "Finished waiting for processes to go down.");
    }
  }
 
  if (m_client.isOnline()) {
    B2DEBUG(30, "Will kill the proxy now.");
    // Kill the proxy and give it some time to terminate
    auto message = ZMQMessageHelper::createZMQMessage("TERMINATE");
    m_client.send(message);
    std::this_thread::sleep_for(std::chrono::milliseconds(10));
  }
 
  // If everything did not help, we will kill all of them
  GlobalProcHandler::killAllProcesses();
}

needMoreWorkers()

unsigned int needMoreWorkers

(

)

const

Compare our current list of workers of how many we want to have.

Definition at line 353 of file ProcessMonitor.cc.

{
  const int neededWorkers = m_requestedNumberOfWorkers - processesWithType(ProcType::c_Worker);
  if (neededWorkers < 0) {
    B2FATAL("Something went completely wrong here! I have more workers as requested...");
  }
  if (neededWorkers > 0) {
    B2DEBUG(30, "I need to restart " << neededWorkers << " workers");
  }
  return static_cast<unsigned int>(neededWorkers);
}

processesWithType()

unsigned int processesWithType ( const ProcType &  procType ) const

private

Cound the number of processes with a certain type.

Definition at line 365 of file ProcessMonitor.cc.

{
  auto correctProcType = [&procType](const auto & pair) {
    return pair.second == procType;
  };
  return std::count_if(m_processList.begin(), m_processList.end(), correctProcType);
}

processMulticast()

void processMulticast ( const ASocket &  socket )

private

Process a message from the multicast.

Definition at line 224 of file ProcessMonitor.cc.

{
  auto pcbMulticastMessage = ZMQMessageFactory::fromSocket<ZMQNoIdMessage>(socket);
  if (pcbMulticastMessage->isMessage(EMessageTypes::c_helloMessage)) {
    const int pid = std::stoi(pcbMulticastMessage->getData());
    const ProcType procType = GlobalProcHandler::getProcType(pid);
    m_processList[pid] = procType;
    B2DEBUG(30, "Now having " << processesWithType(ProcType::c_Input) << " input processes.");
    B2DEBUG(30, "Now having " << processesWithType(ProcType::c_Output) << " output processes.");
    B2DEBUG(30, "Now having " << processesWithType(ProcType::c_Worker) << " worker processes.");
  } else if (pcbMulticastMessage->isMessage(EMessageTypes::c_goodbyeMessage)) {
    const int pid = std::stoi(pcbMulticastMessage->getData());
    const auto& processIt = m_processList.find(pid);
    if (processIt == m_processList.end()) {
      B2WARNING("An unknown PID died!");
      return;
    }
    const ProcType procType = processIt->second;
    if (procType == ProcType::c_Worker) {
      m_requestedNumberOfWorkers--;
      B2DEBUG(30, "Now we will only need " << m_requestedNumberOfWorkers << " of workers anymore");
    }
    processIt->second = ProcType::c_Stopped;
    B2DEBUG(30, "Now having " << processesWithType(ProcType::c_Input) << " input processes.");
    B2DEBUG(30, "Now having " << processesWithType(ProcType::c_Output) << " output processes.");
    B2DEBUG(30, "Now having " << processesWithType(ProcType::c_Worker) << " worker processes.");
  } else if (pcbMulticastMessage->isMessage(EMessageTypes::c_killWorkerMessage)) {
    const int workerPID = atoi(pcbMulticastMessage->getData().c_str());
    B2DEBUG(30, "Got message to kill worker " << workerPID);
    if (kill(workerPID, SIGKILL) == 0) {
      B2WARNING("Needed to kill worker  " << workerPID << " as it was requested.");
    } else {
      B2ERROR("Try to kill process " << workerPID << ", but process is already gone.");
    }
    // TODO: Do we want to decrease the number of workers here or later in the check of the processes?
  } else if (pcbMulticastMessage->isMessage(EMessageTypes::c_statisticMessage)) {
    m_streamer.read(std::move(pcbMulticastMessage));
    B2DEBUG(30, "Having received the process statistics");
    m_receivedStatistics = true;
  }
}

reset()

void reset

(

)

Reset the internal state.

Definition at line 101 of file ProcessMonitor.cc.

{
  m_client.reset();
}

subscribe()

void subscribe	(	const std::string &	pubSocketAddress,
		const std::string &	subSocketAddress,
		const std::string &	controlSocketAddress
	)

Start listening for process management information on the given address.

Definition at line 24 of file ProcessMonitor.cc.

{
  if (GlobalProcHandler::startProxyProcess()) {
    B2DEBUG(30, "ProcessMonitor starting.....");
    m_client.reset();
 
    // The default will be to not do anything on signals...
    EventProcessor::installMainSignalHandlers(SIG_IGN);
 
    // We open a new context here in the new process
    zmq::context_t context(1);
 
    // current directory
    char currentdir[256];
    getcwd(currentdir, 256);
 
    zmq::socket_t pubSocket(context, ZMQ_XPUB);
    // ATTENTION: this is switched on intention!
    try {
      pubSocket.bind(subSocketAddress);
    } catch (zmq::error_t& err) {
      B2DEBUG(30, "ZMQ bind error!! " + std::string(err.what()));
    }
 
    pubSocket.set(zmq::sockopt::linger, 0);
 
    zmq::socket_t subSocket(context, ZMQ_XSUB);
    // ATTENTION: this is switched on intention!
    subSocket.bind(pubSocketAddress);
    subSocket.set(zmq::sockopt::linger, 0);
 
    zmq::socket_t controlSocket(context, ZMQ_SUB);
    controlSocket.bind(controlSocketAddress);
    controlSocket.set(zmq::sockopt::linger, 0);
    controlSocket.set(zmq::sockopt::subscribe, "");
 
    B2DEBUG(30, "Will now start the proxy..");
    bool running = true;
    while (running) {
      try {
        zmq::socket_ref nullsocket;
        zmq::proxy_steerable(pubSocket, subSocket, nullsocket, controlSocket);
        running = false;
      } catch (zmq::error_t& ex) {
        if (ex.num() != EINTR) {
          B2ERROR("There was an error during the proxy event: " << ex.what());
          running = false;
        }
      }
    }
    controlSocket.close();
    pubSocket.close();
    subSocket.close();
    context.close();
    B2DEBUG(30, "Proxy has finished");
    exit(0);
  }
 
  // Time to setup the proxy
  std::this_thread::sleep_for(std::chrono::milliseconds(500));
 
  m_streamer.initialize(0, true);
  m_client.initialize<ZMQ_PUB>(pubSocketAddress, subSocketAddress, controlSocketAddress, false);
  m_client.subscribe(EMessageTypes::c_helloMessage);
  m_client.subscribe(EMessageTypes::c_statisticMessage);
  m_client.subscribe(EMessageTypes::c_killWorkerMessage);
  m_client.subscribe(EMessageTypes::c_goodbyeMessage);
 
  B2DEBUG(30, "Started multicast publishing on " << pubSocketAddress << " and subscribing on " << subSocketAddress);
}

terminate()

void terminate

(

)

Terminate the processing.

Definition at line 96 of file ProcessMonitor.cc.

{
  m_client.terminate(false);
}

waitForRunningInput()

void waitForRunningInput

(

int

timeout

)

Block until either the input process is running or the timeout (in seconds) is raised.

Definition at line 157 of file ProcessMonitor.cc.

{
  if (processesWithType(ProcType::c_Input) < 1) {
    const auto multicastAnswer = [this](const auto & socket) {
      processMulticast(socket);
      return processesWithType(ProcType::c_Input) < 1;
    };
 
    const auto pullResult = m_client.pollMulticast(timeout, multicastAnswer);
    if (not pullResult) {
      B2ERROR("Input process did not start properly!");
      m_hasEnded = true;
    }
  }
}

waitForRunningOutput()

void waitForRunningOutput

(

int

timeout

)

Block until either the output process is running or the timeout (in seconds) is raised.

Definition at line 189 of file ProcessMonitor.cc.

{
  if (processesWithType(ProcType::c_Output) < 1) {
    const auto multicastAnswer = [this](const auto & socket) {
      processMulticast(socket);
      return processesWithType(ProcType::c_Output) < 1;
    };
 
    const auto pullResult = m_client.pollMulticast(timeout, multicastAnswer);
    if (not pullResult) {
      B2ERROR("Output process did not start properly!");
      m_hasEnded = true;
    }
  }
}

waitForRunningWorker()

void waitForRunningWorker

(

int

timeout

)

Block until either the worker process is running or the timeout (in seconds) is raised.

Definition at line 173 of file ProcessMonitor.cc.

{
  if (processesWithType(ProcType::c_Worker) < m_requestedNumberOfWorkers) {
    const auto multicastAnswer = [this](const auto & socket) {
      processMulticast(socket);
      return processesWithType(ProcType::c_Worker) < m_requestedNumberOfWorkers;
    };
 
    const auto pullResult = m_client.pollMulticast(timeout, multicastAnswer);
    if (not pullResult) {
      B2ERROR("Worker process did not start properly!");
      m_hasEnded = true;
    }
  }
}

Member Data Documentation

m_client

ZMQClient m_client

private

The client used for message processing.

Definition at line 69 of file ProcessMonitor.h.

m_hasEnded

bool m_hasEnded = false

private

Someone requested us to end the processing.

Definition at line 76 of file ProcessMonitor.h.

m_processList

std::map<int, ProcType> m_processList

private

The current list of pid -> process types (to be compared to the proc handler)

Definition at line 74 of file ProcessMonitor.h.

m_receivedStatistics

bool m_receivedStatistics = false

private

Did we already receive the statistics?

Definition at line 82 of file ProcessMonitor.h.

m_requestedNumberOfWorkers

unsigned int m_requestedNumberOfWorkers = 0

private

How many workers we should request to start.

Definition at line 72 of file ProcessMonitor.h.

m_streamer

StreamHelper m_streamer

private

The data store streamer.

Definition at line 79 of file ProcessMonitor.h.

---

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

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