A optimized version of parallel message manager. More...

#include <parallel_message_manager_opt.h>

Inheritance diagram for grape::ParallelMessageManagerOpt:

Public Member Functions
void	Init (MPI_Comm comm) override
	Inherit.

MessageBufferPool &	GetPool ()

void	Start () override
	Inherit.

void	StartARound () override
	Inherit.

void	FinishARound () override
	Inherit.

bool	ToTerminate () override
	Inherit.

void	Finalize () override
	Inherit.

void	ForceContinue () override
	Inherit.

void	ForceTerminate (const std::string &terminate_info) override
	Inherit.

const TerminateInfo &	GetTerminateInfo () const override
	Inherit.

size_t	GetMsgSize () const override
	Inherit.

void	InitChannels (int channel_num=1, size_t block_size=default_msg_send_block_size, size_t block_cap=default_msg_send_block_capacity)
	Init a set of channels, each channel is a thread local message buffer.

std::vector< ThreadLocalMessageBufferOpt< ParallelMessageManagerOpt > > &	Channels ()

void	SendMicroBufferByFid (fid_t fid, MicroBuffer &&arc)
	Send a buffer to a fragment.

template<typename MESSAGE_T >
void	SendToFragment (fid_t dst_fid, const MESSAGE_T &msg, int channel_id=0)
	Send message to a fragment.

template<typename GRAPH_T , typename MESSAGE_T >
void	SyncStateOnOuterVertex (const GRAPH_T &frag, const typename GRAPH_T::vertex_t &v, const MESSAGE_T &msg, int channel_id=0)
	SyncStateOnOuterVertex on a channel.

template<typename GRAPH_T >
void	SyncStateOnOuterVertex (const GRAPH_T &frag, const typename GRAPH_T::vertex_t &v, int channel_id=0)

template<typename GRAPH_T , typename MESSAGE_T >
void	SendMsgThroughIEdges (const GRAPH_T &frag, const typename GRAPH_T::vertex_t &v, const MESSAGE_T &msg, int channel_id=0)
	SendMsgThroughIEdges on a channel.

template<typename GRAPH_T , typename MESSAGE_T >
void	SendMsgThroughOEdges (const GRAPH_T &frag, const typename GRAPH_T::vertex_t &v, const MESSAGE_T &msg, int channel_id=0)
	SendMsgThroughOEdges on a channel.

template<typename GRAPH_T , typename MESSAGE_T >
void	SendMsgThroughEdges (const GRAPH_T &frag, const typename GRAPH_T::vertex_t &v, const MESSAGE_T &msg, int channel_id=0)
	SendMsgThroughEdges on a channel.

bool	GetMessageInBuffer (MessageInBuffer &buf)
	Get a bunch of messages, stored in a MessageInBuffer.

template<typename GRAPH_T , typename MESSAGE_T , typename FUNC_T >
void	ParallelProcess (int thread_num, const GRAPH_T &frag, const FUNC_T &func)
	Parallel process all incoming messages with given function of last round.

template<typename GRAPH_T , typename MESSAGE_T , typename FUNC_T >
size_t	ParallelProcessCount (int thread_num, const GRAPH_T &frag, const FUNC_T &func)

template<typename MESSAGE_T , typename FUNC_T >
void	ParallelProcess (int thread_num, const FUNC_T &func)
	Parallel process all incoming messages with given function of last round.

Private Member Functions
void	startSendThread ()

void	probeAllIncomingMessages ()

int	probeIncomingMessages ()

void	startRecvThread ()

void	stopRecvThread ()

size_t	finishMsgFilling ()

void	resetRecvQueue ()

void	waitSend ()

Private Attributes
fid_t	fid_

fid_t	fnum_

CommSpec	comm_spec_

MPI_Comm	comm_

std::vector< MicroBuffer >	to_self_

std::vector< ThreadLocalMessageBufferOpt< ParallelMessageManagerOpt > >	channels_

int	round_

BlockingQueue< std::pair< fid_t, MicroBuffer > >	sending_queue_

std::thread	send_thread_

std::array< BlockingQueue< MicroBuffer >, 2 >	recv_queues_

std::array< MessageBuffer, 2 >	recv_bufs_

std::array< size_t, 2 >	recv_bufs_loc_

std::array< std::vector< MessageBuffer >, 2 >	recv_bufs_stash_

std::thread	recv_thread_

bool	force_continue_

size_t	sent_size_

size_t	total_sent_size_

bool	force_terminate_

TerminateInfo	terminate_info_

MessageBufferPool	pool_

Static Private Attributes
static constexpr size_t	default_msg_send_block_size = 2 * 1023 * 1024

static constexpr size_t	default_msg_send_block_capacity = 2 * 1024 * 1024

Detailed Description

A optimized version of parallel message manager.

ParallelMessageManagerOpt support multi-threads to send messages concurrently with channels. Each channel contains a thread local message buffer.

For each thread local message buffer, when accumulated a given amount of messages, the buffer will be sent through MPI.

After a round of evaluation, there is a global barrier to determine whether the fixed point is reached.

Member Function Documentation

◆ Finalize()

void grape::ParallelMessageManagerOpt::Finalize ( )

inlineoverridevirtual

Inherit.

Implements grape::MessageManagerBase.

◆ FinishARound()

void grape::ParallelMessageManagerOpt::FinishARound ( )

inlineoverridevirtual

Inherit.

Implements grape::MessageManagerBase.

◆ ForceContinue()

void grape::ParallelMessageManagerOpt::ForceContinue ( )

inlineoverridevirtual

Inherit.

Implements grape::MessageManagerBase.

◆ ForceTerminate()

void grape::ParallelMessageManagerOpt::ForceTerminate ( const std::string & terminate_info )

inlineoverridevirtual

Inherit.

Implements grape::MessageManagerBase.

◆ GetMessageInBuffer()

bool grape::ParallelMessageManagerOpt::GetMessageInBuffer ( MessageInBuffer & buf )

inline

Get a bunch of messages, stored in a MessageInBuffer.

Parameters

buf	Message buffer which holds a grape::OutArchive.

◆ GetMsgSize()

size_t grape::ParallelMessageManagerOpt::GetMsgSize ( ) const

inlineoverridevirtual

Inherit.

Implements grape::MessageManagerBase.

◆ GetTerminateInfo()

const TerminateInfo & grape::ParallelMessageManagerOpt::GetTerminateInfo ( ) const

inlineoverridevirtual

Inherit.

Implements grape::MessageManagerBase.

◆ Init()

void grape::ParallelMessageManagerOpt::Init ( MPI_Comm comm )

inlineoverridevirtual

Inherit.

Implements grape::MessageManagerBase.

◆ InitChannels()

void grape::ParallelMessageManagerOpt::InitChannels	(	int	channel_num = `1`,
		size_t	block_size = `default_msg_send_block_size`,
		size_t	block_cap = `default_msg_send_block_capacity`
	)

inline

Init a set of channels, each channel is a thread local message buffer.

Parameters

channel_num	Number of channels.
block_size	Size of each channel.
block_cap	Capacity of each channel.

◆ ParallelProcess() [1/2]

template<typename MESSAGE_T , typename FUNC_T >

void grape::ParallelMessageManagerOpt::ParallelProcess	(	int	thread_num,
		const FUNC_T &	func
	)

inline

Parallel process all incoming messages with given function of last round.

Template Parameters

GRAPH_T	Graph type.
MESSAGE_T	Message type.
FUNC_T	Function type.

Parameters

thread_num	Number of threads.
frag
func

◆ ParallelProcess() [2/2]

template<typename GRAPH_T , typename MESSAGE_T , typename FUNC_T >

void grape::ParallelMessageManagerOpt::ParallelProcess	(	int	thread_num,
		const GRAPH_T &	frag,
		const FUNC_T &	func
	)

inline

Parallel process all incoming messages with given function of last round.

Template Parameters

GRAPH_T	Graph type.
MESSAGE_T	Message type.
FUNC_T	Function type.

Parameters

thread_num	Number of threads.
frag
func

◆ SendMicroBufferByFid()

void grape::ParallelMessageManagerOpt::SendMicroBufferByFid	(	fid_t	fid,
		MicroBuffer &&	arc
	)

inline

Send a buffer to a fragment.

Parameters

fid	Destination fragment id.
arc	Message buffer.

◆ SendMsgThroughEdges()

template<typename GRAPH_T , typename MESSAGE_T >

void grape::ParallelMessageManagerOpt::SendMsgThroughEdges	(	const GRAPH_T &	frag,
		const typename GRAPH_T::vertex_t &	v,
		const MESSAGE_T &	msg,
		int	channel_id = `0`
	)

inline

SendMsgThroughEdges on a channel.

Template Parameters

GRAPH_T	Graph type.
MESSAGE_T	Message type.

Parameters

frag	Source fragment.
v	Source vertex.
msg
channel_id

◆ SendMsgThroughIEdges()

template<typename GRAPH_T , typename MESSAGE_T >

void grape::ParallelMessageManagerOpt::SendMsgThroughIEdges	(	const GRAPH_T &	frag,
		const typename GRAPH_T::vertex_t &	v,
		const MESSAGE_T &	msg,
		int	channel_id = `0`
	)

inline

SendMsgThroughIEdges on a channel.

Template Parameters

GRAPH_T	Graph type.
MESSAGE_T	Message type.

Parameters

frag	Source fragment.
v	Source vertex.
msg
channel_id

◆ SendMsgThroughOEdges()

template<typename GRAPH_T , typename MESSAGE_T >

void grape::ParallelMessageManagerOpt::SendMsgThroughOEdges	(	const GRAPH_T &	frag,
		const typename GRAPH_T::vertex_t &	v,
		const MESSAGE_T &	msg,
		int	channel_id = `0`
	)

inline

SendMsgThroughOEdges on a channel.

Template Parameters

GRAPH_T	Graph type.
MESSAGE_T	Message type.

Parameters

frag	Source fragment.
v	Source vertex.
msg
channel_id

◆ SendToFragment()

template<typename MESSAGE_T >

void grape::ParallelMessageManagerOpt::SendToFragment	(	fid_t	dst_fid,
		const MESSAGE_T &	msg,
		int	channel_id = `0`
	)

inline

Send message to a fragment.

Template Parameters

MESSAGE_T Message type.

Parameters

dst_fid	Destination fragment id.
msg
channelId

◆ Start()

void grape::ParallelMessageManagerOpt::Start ( )

inlineoverridevirtual

Inherit.

Implements grape::MessageManagerBase.

◆ StartARound()

void grape::ParallelMessageManagerOpt::StartARound ( )

inlineoverridevirtual

Inherit.

Implements grape::MessageManagerBase.

◆ SyncStateOnOuterVertex()

template<typename GRAPH_T , typename MESSAGE_T >

void grape::ParallelMessageManagerOpt::SyncStateOnOuterVertex	(	const GRAPH_T &	frag,
		const typename GRAPH_T::vertex_t &	v,
		const MESSAGE_T &	msg,
		int	channel_id = `0`
	)

inline

SyncStateOnOuterVertex on a channel.

Template Parameters

GRAPH_T	Graph type.
MESSAGE_T	Message type.

Parameters

frag	Source fragment.
v	Source vertex.
msg
channel_id

◆ ToTerminate()

bool grape::ParallelMessageManagerOpt::ToTerminate ( )

inlineoverridevirtual

Inherit.

Implements grape::MessageManagerBase.

Public Member Functions

Private Member Functions

Private Attributes

Static Private Attributes

Detailed Description

Member Function Documentation

◆ Finalize()

◆ FinishARound()

◆ ForceContinue()

◆ ForceTerminate()

◆ GetMessageInBuffer()

◆ GetMsgSize()

◆ GetTerminateInfo()

◆ Init()

◆ InitChannels()

◆ ParallelProcess() [1/2]

◆ ParallelProcess() [2/2]

◆ SendMicroBufferByFid()

◆ SendMsgThroughEdges()

◆ SendMsgThroughIEdges()

◆ SendMsgThroughOEdges()

◆ SendToFragment()

◆ Start()

◆ StartARound()

◆ SyncStateOnOuterVertex()

◆ ToTerminate()