RooFit::MultiProcess::Channel Class Reference

One endpoint of a full-duplex interprocess message pipe.

A Channel wraps one end of an AF_UNIX socketpair() created before forking the child processes, and provides framed, whole-message send and receive operations on top of the byte stream. Each frame is preceded by an 8-byte header containing the payload size and a "more" bit that marks all but the last frame of a multipart message.

Sends never block: bytes that the kernel socket buffer does not accept immediately are stored in a per-channel pending-output buffer, which is flushed opportunistically whenever any Channel in the process waits for input (see wait()). This mimics the previous ZeroMQ setup with an unlimited high-water mark and avoids send-send deadlocks between processes.

Definition at line 55 of file Channel.h.

Public Member Functions
	Channel ()=default
	Channel (Channel &&other) noexcept
	Channel (const Channel &)=delete
	Channel (int fd)
	Takes ownership of fd (one end of a socketpair) and makes it non-blocking.
	~Channel ()
int	fd () const
bool	has_pending_output () const
Channel &	operator= (Channel &&other) noexcept
Channel &	operator= (const Channel &)=delete
Message	recv_frame (bool *more=nullptr)
	Blocking receive of one complete frame, interruptible by SIGTERM (throws ppoll_error_t, like the poll functions).
void	send_frame (const void *data, std::size_t size, bool more)
	Queue one frame for sending and write out as much as the socket accepts.
bool	try_flush ()
	Write out pending output; returns true when all of it has been written.
bool	try_recv_frame (Message &msg, bool *more)
	Non-blocking receive attempt.
bool	valid () const

Static Public Member Functions
static std::vector< std::size_t >	wait (const std::vector< const Channel * > &read_channels, int timeout_ms)
	Wait until at least one of read_channels has input available, flushing the pending output of all live Channels in this process meanwhile.

Private Member Functions
void	close_fd ()

Static Private Member Functions
static void	throw_connection_closed ()
	Handle end-of-stream / closed-connection conditions; never returns.

Private Attributes
int	fd_ = -1
bool	in_have_header_ = false
std::uint64_t	in_header_ = 0
std::size_t	in_header_bytes_ = 0
Message	in_msg_
std::size_t	in_msg_bytes_ = 0
std::vector< char >	out_buf_
std::size_t	out_pos_ = 0

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

Constructor & Destructor Documentation

Channel() [1/4]

RooFit::MultiProcess::Channel::Channel ( )

default

Channel() [2/4]

RooFit::MultiProcess::Channel::Channel ( int fd )

explicit

Takes ownership of fd (one end of a socketpair) and makes it non-blocking.

Definition at line 68 of file Channel.cxx.

~Channel()

RooFit::MultiProcess::Channel::~Channel

(

)

Definition at line 82 of file Channel.cxx.

Channel() [3/4]

RooFit::MultiProcess::Channel::Channel ( const Channel & )

delete

Channel() [4/4]

RooFit::MultiProcess::Channel::Channel ( Channel && other )

noexcept

Definition at line 90 of file Channel.cxx.

Member Function Documentation

close_fd()

void RooFit::MultiProcess::Channel::close_fd ( )

private

Definition at line 131 of file Channel.cxx.

fd()

int RooFit::MultiProcess::Channel::fd ( ) const

inline

Definition at line 68 of file Channel.h.

has_pending_output()

bool RooFit::MultiProcess::Channel::has_pending_output ( ) const

inline

Definition at line 81 of file Channel.h.

operator=() [1/2]

Channel & RooFit::MultiProcess::Channel::operator= ( Channel && other )

noexcept

Definition at line 107 of file Channel.cxx.

operator=() [2/2]

Channel & RooFit::MultiProcess::Channel::operator= ( const Channel & )

delete

recv_frame()

Message RooFit::MultiProcess::Channel::recv_frame

(

bool *

more = nullptr

)

Blocking receive of one complete frame, interruptible by SIGTERM (throws ppoll_error_t, like the poll functions).

Definition at line 255 of file Channel.cxx.

send_frame()

void RooFit::MultiProcess::Channel::send_frame	(	const void *	data,
		std::size_t	size,
		bool	more )

Queue one frame for sending and write out as much as the socket accepts.

Definition at line 158 of file Channel.cxx.

throw_connection_closed()

void RooFit::MultiProcess::Channel::throw_connection_closed ( )

staticprivate

Handle end-of-stream / closed-connection conditions; never returns.

Definition at line 139 of file Channel.cxx.

try_flush()

bool RooFit::MultiProcess::Channel::try_flush

(

)

Write out pending output; returns true when all of it has been written.

Definition at line 177 of file Channel.cxx.

try_recv_frame()

bool RooFit::MultiProcess::Channel::try_recv_frame	(	Message &	msg,
		bool *	more )

Non-blocking receive attempt.

Returns true and fills msg/more when a complete frame was received; returns false if more bytes are needed.

Definition at line 198 of file Channel.cxx.

valid()

bool RooFit::MultiProcess::Channel::valid ( ) const

inline

Definition at line 67 of file Channel.h.

wait()

std::vector< std::size_t > RooFit::MultiProcess::Channel::wait ( const std::vector< const Channel * > & read_channels, int timeout_ms )

static

Wait until at least one of read_channels has input available, flushing the pending output of all live Channels in this process meanwhile.

Returns the indices into read_channels that are readable. A negative timeout means wait forever; otherwise the result may be empty after timeout_ms milliseconds. Throws ppoll_error_t with num() == EINTR when interrupted by a signal (including the SIGTERM self-pipe wake-up).

Definition at line 264 of file Channel.cxx.

Member Data Documentation

fd_

int RooFit::MultiProcess::Channel::fd_ = -1

private

Definition at line 98 of file Channel.h.

in_have_header_

bool RooFit::MultiProcess::Channel::in_have_header_ = false

private

Definition at line 107 of file Channel.h.

in_header_

std::uint64_t RooFit::MultiProcess::Channel::in_header_ = 0

private

Definition at line 105 of file Channel.h.

in_header_bytes_

std::size_t RooFit::MultiProcess::Channel::in_header_bytes_ = 0

private

Definition at line 106 of file Channel.h.

in_msg_

Message RooFit::MultiProcess::Channel::in_msg_

private

Definition at line 108 of file Channel.h.

in_msg_bytes_

std::size_t RooFit::MultiProcess::Channel::in_msg_bytes_ = 0

private

Definition at line 109 of file Channel.h.

out_buf_

std::vector<char> RooFit::MultiProcess::Channel::out_buf_

private

Definition at line 101 of file Channel.h.

out_pos_

std::size_t RooFit::MultiProcess::Channel::out_pos_ = 0

private

Definition at line 102 of file Channel.h.