RooFit::MultiProcess::ProcessManager Class Reference

Fork processes for queue and workers.

This class manages three types of processes:

1. master: the initial main process. It defines and enqueues tasks and processes results.
2. workers: a pool of processes that will try to take tasks from the queue. These are forked from master.
3. queue: This process runs the queue_loop and maintains the queue of tasks. It is also forked from master.

Parameters

N_workers

Number of worker processes to spawn.

Definition at line 27 of file ProcessManager.h.

Public Member Functions
	ProcessManager (std::size_t N_workers)
	~ProcessManager ()
int	claim_mq_fd ()
	Hand over the master-queue channel end for the current process type.
int	claim_mw_fd (std::size_t worker_ix)
	Hand over the master-worker channel end for the current process type.
int	claim_qw_fd (std::size_t worker_ix)
	Hand over the queue-worker channel end for the current process type.
pid_t	get_queue_pid () const
std::vector< pid_t >	get_worker_pids ()
void	identify_processes () const
	Print to stdout which type of process we are on and what its PID is (for debugging)
bool	is_initialized () const
bool	is_master () const
bool	is_queue () const
bool	is_worker () const
std::size_t	N_workers () const
void	terminate () noexcept
	Shutdown forked processes if on master and if this process manager is initialized.
void	wait_for_sigterm_then_exit ()
std::size_t	worker_id () const

Static Public Member Functions
static void	handle_sigterm (int signum)
	We need this to tell the children to die, because we can't talk to them anymore during JobManager destruction, because that kills the Messenger first.
static bool	sigterm_received ()
static int	sigterm_wake_fd ()
	Read end of the self-pipe that the SIGTERM handler writes to (or -1 on the master process, which installs no handler); used by Channel::wait.

Private Member Functions
void	close_channel_fds ()
	Close all channel ends still owned by this ProcessManager (i.e.
void	close_unused_channel_fds ()
	Close the channel ends that do not belong to the current process type.
void	create_channel_fds ()
	Create the socketpairs that connect the processes.
void	initialize_processes (bool cpu_pinning=true)
	Fork processes and activate CPU pinning.
void	shutdown_processes ()
	Shutdown forked processes if on master.

Private Attributes
bool	initialized_ = false
bool	is_master_ = false
bool	is_queue_ = false
bool	is_worker_ = false
std::array< int, 2 >	mq_fds_ {{-1, -1}}
std::vector< std::array< int, 2 > >	mw_fds_
std::size_t	N_workers_
friend	Queue
pid_t	queue_pid_
std::vector< std::array< int, 2 > >	qw_fds_
std::size_t	worker_id_
std::vector< pid_t >	worker_pids_

Static Private Attributes
static volatile sig_atomic_t	sigterm_received_ = 0
static int	sigterm_wake_read_fd_ = -1
static int	sigterm_wake_write_fd_ = -1

#include </github/home/ROOT-CI/src/roofit/multiprocess/res/RooFit/MultiProcess/ProcessManager.h>

Constructor & Destructor Documentation

ProcessManager()

RooFit::MultiProcess::ProcessManager::ProcessManager ( std::size_t N_workers )

explicit

Definition at line 43 of file ProcessManager.cxx.

~ProcessManager()

RooFit::MultiProcess::ProcessManager::~ProcessManager

(

)

Definition at line 52 of file ProcessManager.cxx.

Member Function Documentation

claim_mq_fd()

int RooFit::MultiProcess::ProcessManager::claim_mq_fd

(

)

Hand over the master-queue channel end for the current process type.

Definition at line 216 of file ProcessManager.cxx.

claim_mw_fd()

int RooFit::MultiProcess::ProcessManager::claim_mw_fd

(

std::size_t

worker_ix

)

Hand over the master-worker channel end for the current process type.

Definition at line 228 of file ProcessManager.cxx.

claim_qw_fd()

int RooFit::MultiProcess::ProcessManager::claim_qw_fd

(

std::size_t

worker_ix

)

Hand over the queue-worker channel end for the current process type.

Definition at line 222 of file ProcessManager.cxx.

close_channel_fds()

void RooFit::MultiProcess::ProcessManager::close_channel_fds ( )

private

Close all channel ends still owned by this ProcessManager (i.e.

not claimed by a Messenger).

Definition at line 204 of file ProcessManager.cxx.

close_unused_channel_fds()

void RooFit::MultiProcess::ProcessManager::close_unused_channel_fds ( )

private

Close the channel ends that do not belong to the current process type.

Definition at line 179 of file ProcessManager.cxx.

create_channel_fds()

void RooFit::MultiProcess::ProcessManager::create_channel_fds ( )

private

Create the socketpairs that connect the processes.

Must be called before forking; every process then keeps only the ends it needs (see close_unused_channel_fds).

Definition at line 167 of file ProcessManager.cxx.

get_queue_pid()

pid_t RooFit::MultiProcess::ProcessManager::get_queue_pid ( ) const

inline

Definition at line 62 of file ProcessManager.h.

get_worker_pids()

std::vector< pid_t > RooFit::MultiProcess::ProcessManager::get_worker_pids ( )

inline

Definition at line 63 of file ProcessManager.h.

handle_sigterm()

void RooFit::MultiProcess::ProcessManager::handle_sigterm ( int signum )

static

We need this to tell the children to die, because we can't talk to them anymore during JobManager destruction, because that kills the Messenger first.

We do that with SIGTERMs. The sigterm_received() should be checked in message loops to stop them when it's true. The handler also writes to a self-pipe, so that a poll that is entered after the flag check but before signal delivery still wakes up.

Definition at line 74 of file ProcessManager.cxx.

identify_processes()

void RooFit::MultiProcess::ProcessManager::identify_processes

(

)

const

Print to stdout which type of process we are on and what its PID is (for debugging)

Definition at line 470 of file ProcessManager.cxx.

initialize_processes()

void RooFit::MultiProcess::ProcessManager::initialize_processes ( bool cpu_pinning = true )

private

Fork processes and activate CPU pinning.

Parameters

cpu_pinning

Activate CPU pinning if true. Effective on Linux only.

Definition at line 236 of file ProcessManager.cxx.

is_initialized()

bool RooFit::MultiProcess::ProcessManager::is_initialized

(

)

const

Definition at line 356 of file ProcessManager.cxx.

is_master()

bool RooFit::MultiProcess::ProcessManager::is_master

(

)

const

Definition at line 444 of file ProcessManager.cxx.

is_queue()

bool RooFit::MultiProcess::ProcessManager::is_queue

(

)

const

Definition at line 449 of file ProcessManager.cxx.

is_worker()

bool RooFit::MultiProcess::ProcessManager::is_worker

(

)

const

Definition at line 454 of file ProcessManager.cxx.

N_workers()

std::size_t RooFit::MultiProcess::ProcessManager::N_workers

(

)

const

Definition at line 464 of file ProcessManager.cxx.

shutdown_processes()

void RooFit::MultiProcess::ProcessManager::shutdown_processes ( )

private

Shutdown forked processes if on master.

Definition at line 418 of file ProcessManager.cxx.

sigterm_received()

bool RooFit::MultiProcess::ProcessManager::sigterm_received ( )

static

Definition at line 92 of file ProcessManager.cxx.

sigterm_wake_fd()

int RooFit::MultiProcess::ProcessManager::sigterm_wake_fd ( )

static

Read end of the self-pipe that the SIGTERM handler writes to (or -1 on the master process, which installs no handler); used by Channel::wait.

Definition at line 86 of file ProcessManager.cxx.

terminate()

void RooFit::MultiProcess::ProcessManager::terminate ( )

noexcept

Shutdown forked processes if on master and if this process manager is initialized.

Definition at line 362 of file ProcessManager.cxx.

wait_for_sigterm_then_exit()

void RooFit::MultiProcess::ProcessManager::wait_for_sigterm_then_exit

(

)

Definition at line 374 of file ProcessManager.cxx.

worker_id()

std::size_t RooFit::MultiProcess::ProcessManager::worker_id

(

)

const

Definition at line 459 of file ProcessManager.cxx.

Member Data Documentation

initialized_

bool RooFit::MultiProcess::ProcessManager::initialized_ = false

private

Definition at line 82 of file ProcessManager.h.

is_master_

bool RooFit::MultiProcess::ProcessManager::is_master_ = false

private

Definition at line 72 of file ProcessManager.h.

is_queue_

bool RooFit::MultiProcess::ProcessManager::is_queue_ = false

private

Definition at line 73 of file ProcessManager.h.

is_worker_

bool RooFit::MultiProcess::ProcessManager::is_worker_ = false

private

Definition at line 74 of file ProcessManager.h.

mq_fds_

std::array<int, 2> RooFit::MultiProcess::ProcessManager::mq_fds_ {{-1, -1}}

private

Definition at line 87 of file ProcessManager.h.

mw_fds_

std::vector<std::array<int, 2> > RooFit::MultiProcess::ProcessManager::mw_fds_

private

Definition at line 89 of file ProcessManager.h.

N_workers_

std::size_t RooFit::MultiProcess::ProcessManager::N_workers_

private

Definition at line 76 of file ProcessManager.h.

Queue

friend RooFit::MultiProcess::ProcessManager::Queue

private

Definition at line 28 of file ProcessManager.h.

queue_pid_

pid_t RooFit::MultiProcess::ProcessManager::queue_pid_

private

Definition at line 80 of file ProcessManager.h.

qw_fds_

std::vector<std::array<int, 2> > RooFit::MultiProcess::ProcessManager::qw_fds_

private

Definition at line 88 of file ProcessManager.h.

sigterm_received_

volatile sig_atomic_t RooFit::MultiProcess::ProcessManager::sigterm_received_ = 0

staticprivate

Definition at line 91 of file ProcessManager.h.

sigterm_wake_read_fd_

int RooFit::MultiProcess::ProcessManager::sigterm_wake_read_fd_ = -1

staticprivate

Definition at line 92 of file ProcessManager.h.

sigterm_wake_write_fd_

int RooFit::MultiProcess::ProcessManager::sigterm_wake_write_fd_ = -1

staticprivate

Definition at line 93 of file ProcessManager.h.

worker_id_

std::size_t RooFit::MultiProcess::ProcessManager::worker_id_

private

Definition at line 75 of file ProcessManager.h.

worker_pids_

std::vector<pid_t> RooFit::MultiProcess::ProcessManager::worker_pids_

private

Definition at line 79 of file ProcessManager.h.