Package com.alibaba.graphscope.parallel.mm.impl

Class AbstractMessageManager<OID_T extends org.apache.hadoop.io.WritableComparable,VDATA_T extends org.apache.hadoop.io.Writable,EDATA_T extends org.apache.hadoop.io.Writable,IN_MSG_T extends org.apache.hadoop.io.Writable,OUT_MSG_T extends org.apache.hadoop.io.Writable,GS_VID_T,GS_OID_T>

java.lang.Object

com.alibaba.graphscope.parallel.mm.impl.AbstractMessageManager<OID_T,VDATA_T,EDATA_T,IN_MSG_T,OUT_MSG_T,GS_VID_T,GS_OID_T>

Type Parameters:

OID_T -

VDATA_T -

EDATA_T -

IN_MSG_T -

OUT_MSG_T -

GS_VID_T -

GS_OID_T -

All Implemented Interfaces:

GiraphMessageManager<OID_T,VDATA_T,EDATA_T,IN_MSG_T,OUT_MSG_T,GS_VID_T,GS_OID_T>

Direct Known Subclasses:

GiraphMpiMessageManager, GiraphNettyMessageManager

public abstract class AbstractMessageManager<OID_T extends org.apache.hadoop.io.WritableComparable,VDATA_T extends org.apache.hadoop.io.Writable,EDATA_T extends org.apache.hadoop.io.Writable,IN_MSG_T extends org.apache.hadoop.io.Writable,OUT_MSG_T extends org.apache.hadoop.io.Writable,GS_VID_T,GS_OID_T>
extends Object
implements GiraphMessageManager<OID_T,VDATA_T,EDATA_T,IN_MSG_T,OUT_MSG_T,GS_VID_T,GS_OID_T>

skeletal implementation for giraph message manager. providing common utils for message store(cache).

Field Summary

Fields

Modifier and Type	Field	Description
protected MessageStore<OID_T,IN_MSG_T,GS_VID_T>	currentIncomingMessageStore
protected int	fragId
protected IFragment<GS_OID_T,GS_VID_T,?,?>	fragment
protected int	fragNum
protected DefaultMessageManager	grapeMessager
protected Vertex<GS_VID_T>	grapeVertex
protected MessageStore<OID_T,IN_MSG_T,GS_VID_T>	nextIncomingMessageStore

Constructor Summary

Constructors

Constructor	Description
AbstractMessageManager(IFragment<GS_OID_T,GS_VID_T,?,?> fragment, DefaultMessageManager mm, ImmutableClassesGiraphConfiguration<OID_T,VDATA_T,EDATA_T> conf, FFICommunicator communicator)

Method Summary

Modifier and Type	Method	Description
boolean	anyMessageReceived()	As this is called after superStep and before presuperStep's swapping, we check nextIncomingMessage Store.
protected void	checkLid(long lid)
void	forceContinue()
FFICommunicator	getCommunicator()
ImmutableClassesGiraphConfiguration<OID_T,VDATA_T,EDATA_T>	getConf()
IFragment<GS_OID_T,GS_VID_T,?,?>	getFragment()
Iterable<IN_MSG_T>	getMessages(long lid)	Get the messages received from last round.
boolean	messageAvailable(long lid)	Check any message available on this vertex.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface com.alibaba.graphscope.parallel.mm.GiraphMessageManager

finishMessageSending, postApplication, postSuperstep, preSuperstep, receiveMessages, sendMessage, sendMessageToAllEdges

Field Detail

fragment

protected IFragment<GS_OID_T,GS_VID_T,?,?> fragment

grapeMessager

protected DefaultMessageManager grapeMessager

grapeVertex

protected Vertex<GS_VID_T> grapeVertex

fragId

protected int fragId

fragNum

protected int fragNum

nextIncomingMessageStore

protected volatile MessageStore<OID_T extends org.apache.hadoop.io.WritableComparable,IN_MSG_T extends org.apache.hadoop.io.Writable,GS_VID_T> nextIncomingMessageStore

currentIncomingMessageStore

protected volatile MessageStore<OID_T extends org.apache.hadoop.io.WritableComparable,IN_MSG_T extends org.apache.hadoop.io.Writable,GS_VID_T> currentIncomingMessageStore

Constructor Detail

AbstractMessageManager

public AbstractMessageManager(IFragment<GS_OID_T,GS_VID_T,?,?> fragment,
                              DefaultMessageManager mm,
                              ImmutableClassesGiraphConfiguration<OID_T,VDATA_T,EDATA_T> conf,
                              FFICommunicator communicator)

Method Detail

getConf

public ImmutableClassesGiraphConfiguration<OID_T,VDATA_T,EDATA_T> getConf()

getFragment

public IFragment<GS_OID_T,GS_VID_T,?,?> getFragment()

getCommunicator

public FFICommunicator getCommunicator()

getMessages

public Iterable<IN_MSG_T> getMessages(long lid)

Get the messages received from last round.

Specified by:

getMessages in interface GiraphMessageManager<OID_T extends org.apache.hadoop.io.WritableComparable,VDATA_T extends org.apache.hadoop.io.Writable,EDATA_T extends org.apache.hadoop.io.Writable,IN_MSG_T extends org.apache.hadoop.io.Writable,OUT_MSG_T extends org.apache.hadoop.io.Writable,GS_VID_T,GS_OID_T>

Parameters:

lid - local id.

Returns:

received msg.

messageAvailable

public boolean messageAvailable(long lid)

Check any message available on this vertex.

Specified by:

messageAvailable in interface GiraphMessageManager<OID_T extends org.apache.hadoop.io.WritableComparable,VDATA_T extends org.apache.hadoop.io.Writable,EDATA_T extends org.apache.hadoop.io.Writable,IN_MSG_T extends org.apache.hadoop.io.Writable,OUT_MSG_T extends org.apache.hadoop.io.Writable,GS_VID_T,GS_OID_T>

Parameters:

lid - local id

Returns:

true if received messages.

anyMessageReceived

public boolean anyMessageReceived()

As this is called after superStep and before presuperStep's swapping, we check nextIncomingMessage Store.

Specified by:

anyMessageReceived in interface GiraphMessageManager<OID_T extends org.apache.hadoop.io.WritableComparable,VDATA_T extends org.apache.hadoop.io.Writable,EDATA_T extends org.apache.hadoop.io.Writable,IN_MSG_T extends org.apache.hadoop.io.Writable,OUT_MSG_T extends org.apache.hadoop.io.Writable,GS_VID_T,GS_OID_T>

Returns:

true if message received

forceContinue

public void forceContinue()

Specified by:

forceContinue in interface GiraphMessageManager<OID_T extends org.apache.hadoop.io.WritableComparable,VDATA_T extends org.apache.hadoop.io.Writable,EDATA_T extends org.apache.hadoop.io.Writable,IN_MSG_T extends org.apache.hadoop.io.Writable,OUT_MSG_T extends org.apache.hadoop.io.Writable,GS_VID_T,GS_OID_T>

checkLid

protected void checkLid(long lid)