org.apache.giraph.comm

Class SendMessageCache<I extends org.apache.hadoop.io.WritableComparable,M extends org.apache.hadoop.io.Writable>

java.lang.Object

org.apache.giraph.comm.SendDataCache<B>

org.apache.giraph.comm.SendVertexIdDataCache<I,M,VertexIdMessages<I,M>>

org.apache.giraph.comm.SendMessageCache<I,M>

Type Parameters:

I - Vertex id

M - Message data

Direct Known Subclasses:

SendOneMessageToManyCache

public class SendMessageCache<I extends org.apache.hadoop.io.WritableComparable,M extends org.apache.hadoop.io.Writable>
extends SendVertexIdDataCache<I,M,VertexIdMessages<I,M>>

Aggregates the messages to be sent to workers so they can be sent in bulk. Not thread-safe.

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	SendMessageCache.TargetVertexIdIterator<I extends org.apache.hadoop.io.WritableComparable> An iterator wrapper on edges to return target vertex ids.

Field Summary

Fields

Modifier and Type	Field and Description
protected NettyWorkerClientRequestProcessor<I,?,?>	clientProcessor NettyWorkerClientRequestProcessor for message sending
protected int	maxMessagesSizePerWorker Max message size sent to a worker
protected MessageValueFactory<M>	messageValueFactory Cached message value factory
protected long	totalMsgBytesSentInSuperstep Message bytes sent during the last superstep
protected long	totalMsgsSentInSuperstep Messages sent during the last superstep

Constructor Summary

Constructors

Constructor and Description
SendMessageCache(ImmutableClassesGiraphConfiguration conf, CentralizedServiceWorker<?,?,?> serviceWorker, NettyWorkerClientRequestProcessor<I,?,?> processor, int maxMsgSize) Constructor

Method Summary

Modifier and Type	Method and Description
protected int	addMessage(WorkerInfo workerInfo, int partitionId, byte[] serializedId, int idSerializerPos, M message) Add a message to the cache with serialized ids.
int	addMessage(WorkerInfo workerInfo, int partitionId, I destVertexId, M message) Add a message to the cache.
VertexIdMessages<I,M>	createVertexIdData() Create a new VertexIdData specialized for the use case.
void	flush() Flush the rest of the messages to the workers.
protected PairList<Integer,VertexIdMessages<I,M>>	removeWorkerMessages(WorkerInfo workerInfo) Gets the messages for a worker and removes it from the cache.
long	resetMessageBytesCount() Reset the message bytes count per superstep.
long	resetMessageCount() Reset the message count per superstep.
void	sendMessageRequest(I destVertexId, M message) Send a message to a target vertex id.
void	sendMessageToAllRequest(Iterator<I> vertexIdIterator, M message) Send message to the target ids in the iterator
void	sendMessageToAllRequest(Vertex<I,?,?> vertex, M message) Send message to all its neighbors

Methods inherited from class org.apache.giraph.comm.SendVertexIdDataCache

addData, addData

Methods inherited from class org.apache.giraph.comm.SendDataCache

getConf, getData, getInitialBufferSize, getNumWorkers, getSendWorkerInitialBufferSize, getServiceWorker, getWorkerPartitions, incrDataSize, removeAllData, removeWorkerData, setData

Methods inherited from class java.lang.Object

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

Field Detail

totalMsgsSentInSuperstep

protected long totalMsgsSentInSuperstep

Messages sent during the last superstep

totalMsgBytesSentInSuperstep

protected long totalMsgBytesSentInSuperstep

Message bytes sent during the last superstep

maxMessagesSizePerWorker

protected final int maxMessagesSizePerWorker

Max message size sent to a worker

clientProcessor

protected final NettyWorkerClientRequestProcessor<I extends org.apache.hadoop.io.WritableComparable,?,?> clientProcessor

NettyWorkerClientRequestProcessor for message sending

messageValueFactory

protected MessageValueFactory<M extends org.apache.hadoop.io.Writable> messageValueFactory

Cached message value factory

Constructor Detail

SendMessageCache

public SendMessageCache(ImmutableClassesGiraphConfiguration conf,
                        CentralizedServiceWorker<?,?,?> serviceWorker,
                        NettyWorkerClientRequestProcessor<I,?,?> processor,
                        int maxMsgSize)

Constructor

Parameters:

conf - Giraph configuration

serviceWorker - Service worker

processor - NettyWorkerClientRequestProcessor

maxMsgSize - Max message size sent to a worker

Method Detail

createVertexIdData

public VertexIdMessages<I,M> createVertexIdData()

Description copied from class: SendVertexIdDataCache

Create a new VertexIdData specialized for the use case.

Specified by:

createVertexIdData in class SendVertexIdDataCache<I extends org.apache.hadoop.io.WritableComparable,M extends org.apache.hadoop.io.Writable,VertexIdMessages<I extends org.apache.hadoop.io.WritableComparable,M extends org.apache.hadoop.io.Writable>>

Returns:

A new instance of VertexIdData

addMessage

public int addMessage(WorkerInfo workerInfo,
                      int partitionId,
                      I destVertexId,
                      M message)

Add a message to the cache.

Parameters:

workerInfo - the remote worker destination

partitionId - the remote Partition this message belongs to

destVertexId - vertex id that is ultimate destination

message - Message to send to remote worker

Returns:

Size of messages for the worker.

addMessage

protected int addMessage(WorkerInfo workerInfo,
                         int partitionId,
                         byte[] serializedId,
                         int idSerializerPos,
                         M message)

Add a message to the cache with serialized ids.

Parameters:

workerInfo - The remote worker destination

partitionId - The remote Partition this message belongs to

serializedId - Serialized vertex id that is ultimate destination

idSerializerPos - The end position of serialized id in the byte array

message - Message to send to remote worker

Returns:

Size of messages for the worker.

removeWorkerMessages

protected PairList<Integer,VertexIdMessages<I,M>> removeWorkerMessages(WorkerInfo workerInfo)

Gets the messages for a worker and removes it from the cache.

Parameters:

workerInfo - the address of the worker who owns the data partitions that are receiving the messages

Returns:

List of pairs (partitionId, ByteArrayVertexIdMessages), where all partition ids belong to workerInfo

sendMessageRequest

public void sendMessageRequest(I destVertexId,
                               M message)

Send a message to a target vertex id.

Parameters:

destVertexId - Target vertex id

message - The message sent to the target

sendMessageToAllRequest

public void sendMessageToAllRequest(Vertex<I,?,?> vertex,
                                    M message)

Send message to all its neighbors

Parameters:

vertex - The source vertex

message - The message sent to a worker

sendMessageToAllRequest

public void sendMessageToAllRequest(Iterator<I> vertexIdIterator,
                                    M message)

Send message to the target ids in the iterator

Parameters:

vertexIdIterator - The iterator of target vertex ids

message - The message sent to a worker

flush

public void flush()

Flush the rest of the messages to the workers.

resetMessageCount

public long resetMessageCount()

Reset the message count per superstep.

Returns:

The message count sent in last superstep

resetMessageBytesCount

public long resetMessageBytesCount()

Reset the message bytes count per superstep.

Returns:

The message count sent in last superstep