org.apache.giraph.comm.messages

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

java.lang.Object

org.apache.giraph.comm.messages.SimpleMessageStore<I,M,T>

Type Parameters:

I - Vertex id

M - Message data

T - Type of object which holds messages for one vertex

All Implemented Interfaces:

MessageStore<I,M>

Direct Known Subclasses:

AbstractListPerVertexStore, ByteArrayMessagesPerVertexStore, OneMessagePerVertexStore

public abstract class SimpleMessageStore<I extends org.apache.hadoop.io.WritableComparable,M extends org.apache.hadoop.io.Writable,T>
extends Object
implements MessageStore<I,M>

Abstract class for MessageStore which allows any kind of object to hold messages for one vertex. Simple in memory message store implemented with a two level concurrent hash map.

Field Summary

Fields

Modifier and Type	Field and Description
protected ImmutableClassesGiraphConfiguration<I,?,?>	config Giraph configuration
protected ConcurrentMap<Integer,ConcurrentMap<I,T>>	map Map from partition id to map from vertex id to messages for that vertex
protected MessageValueFactory<M>	messageValueFactory Message class
protected PartitionSplitInfo<I>	partitionInfo Partition split info

Constructor Summary

Constructors

Constructor and Description
SimpleMessageStore(MessageValueFactory<M> messageValueFactory, PartitionSplitInfo<I> partitionInfo, ImmutableClassesGiraphConfiguration<I,?,?> config) Constructor

Method Summary

Modifier and Type	Method and Description
void	clearAll() Clears all resources used by this store.
void	clearPartition(int partitionId) Clears messages for a partition.
void	clearVertexMessages(I vertexId) Clears messages for a vertex.
void	finalizeStore() Called before start of computation in bspworker Since it is run from a single thread while the store is not being accessed by any other thread - this is ensured to be thread-safe
protected abstract Iterable<M>	getMessagesAsIterable(T messages) Get messages as an iterable from message storage
protected abstract int	getNumberOfMessagesIn(ConcurrentMap<I,T> partitionMap) Get number of messages in partition map
protected ConcurrentMap<I,T>	getOrCreatePartitionMap(int partitionId) If there is already a map of messages related to the partition id return that map, otherwise create a new one, put it in global map and return it.
Iterable<I>	getPartitionDestinationVertices(int partitionId) Gets vertex ids from selected partition which we have messages for
protected int	getPartitionId(I vertexId) Get id of partition which holds vertex with selected id
Iterable<M>	getVertexMessages(I vertexId) Gets messages for a vertex.
boolean	hasMessagesForPartition(int partitionId) Check if we have messages for some partition
boolean	hasMessagesForVertex(I vertexId) Check if we have messages for some vertex
protected abstract T	readFieldsForMessages(DataInput in) Read message storage from DataInput
void	readFieldsForPartition(DataInput in, int partitionId) Deserialize messages for one partition
protected abstract void	writeMessages(T messages, DataOutput out) Write message storage to DataOutput
void	writePartition(DataOutput out, int partitionId) Serialize messages for one partition.

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.apache.giraph.comm.messages.MessageStore

addMessage, addPartitionMessages, isPointerListEncoding

Field Detail

messageValueFactory

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

Message class

partitionInfo

protected final PartitionSplitInfo<I extends org.apache.hadoop.io.WritableComparable> partitionInfo

Partition split info

map

protected final ConcurrentMap<Integer,ConcurrentMap<I extends org.apache.hadoop.io.WritableComparable,T>> map

Map from partition id to map from vertex id to messages for that vertex

config

protected final ImmutableClassesGiraphConfiguration<I extends org.apache.hadoop.io.WritableComparable,?,?> config

Giraph configuration

Constructor Detail

SimpleMessageStore

public SimpleMessageStore(MessageValueFactory<M> messageValueFactory,
                          PartitionSplitInfo<I> partitionInfo,
                          ImmutableClassesGiraphConfiguration<I,?,?> config)

Constructor

Parameters:

messageValueFactory - Message class held in the store

partitionInfo - Partition split info

config - Giraph configuration

Method Detail

getMessagesAsIterable

protected abstract Iterable<M> getMessagesAsIterable(T messages)

Get messages as an iterable from message storage

Parameters:

messages - Message storage

Returns:

Messages as an iterable

getNumberOfMessagesIn

protected abstract int getNumberOfMessagesIn(ConcurrentMap<I,T> partitionMap)

Get number of messages in partition map

Parameters:

partitionMap - Partition map in which to count messages

Returns:

Number of messages in partition map

writeMessages

protected abstract void writeMessages(T messages,
                                      DataOutput out)
                               throws IOException

Write message storage to DataOutput

Parameters:

messages - Message storage

out - Data output

Throws:

IOException

readFieldsForMessages

protected abstract T readFieldsForMessages(DataInput in)
                                    throws IOException

Read message storage from DataInput

Parameters:

in - Data input

Returns:

Message storage

Throws:

IOException

getPartitionId

protected int getPartitionId(I vertexId)

Get id of partition which holds vertex with selected id

Parameters:

vertexId - Id of vertex

Returns:

Id of partiton

getOrCreatePartitionMap

protected ConcurrentMap<I,T> getOrCreatePartitionMap(int partitionId)

If there is already a map of messages related to the partition id return that map, otherwise create a new one, put it in global map and return it.

Parameters:

partitionId - Id of partition

Returns:

Message map for this partition

finalizeStore

public void finalizeStore()

Description copied from interface: MessageStore

Called before start of computation in bspworker Since it is run from a single thread while the store is not being accessed by any other thread - this is ensured to be thread-safe

Specified by:

finalizeStore in interface MessageStore<I extends org.apache.hadoop.io.WritableComparable,M extends org.apache.hadoop.io.Writable>

getPartitionDestinationVertices

public Iterable<I> getPartitionDestinationVertices(int partitionId)

Description copied from interface: MessageStore

Gets vertex ids from selected partition which we have messages for

Specified by:

getPartitionDestinationVertices in interface MessageStore<I extends org.apache.hadoop.io.WritableComparable,M extends org.apache.hadoop.io.Writable>

Parameters:

partitionId - Id of partition

Returns:

Iterable over vertex ids which we have messages for

hasMessagesForVertex

public boolean hasMessagesForVertex(I vertexId)

Description copied from interface: MessageStore

Check if we have messages for some vertex

Specified by:

hasMessagesForVertex in interface MessageStore<I extends org.apache.hadoop.io.WritableComparable,M extends org.apache.hadoop.io.Writable>

Parameters:

vertexId - Id of vertex which we want to check

Returns:

True iff we have messages for vertex with required id

getVertexMessages

public Iterable<M> getVertexMessages(I vertexId)

Description copied from interface: MessageStore

Gets messages for a vertex. The lifetime of every message is only guaranteed until the iterator's next() method is called. Do not hold references to objects returned by this iterator.

Specified by:

getVertexMessages in interface MessageStore<I extends org.apache.hadoop.io.WritableComparable,M extends org.apache.hadoop.io.Writable>

Parameters:

vertexId - Vertex id for which we want to get messages

Returns:

Iterable of messages for a vertex id

writePartition

public void writePartition(DataOutput out,
                           int partitionId)
                    throws IOException

Description copied from interface: MessageStore

Serialize messages for one partition.

Specified by:

writePartition in interface MessageStore<I extends org.apache.hadoop.io.WritableComparable,M extends org.apache.hadoop.io.Writable>

Parameters:

out - DataOutput to serialize this object into

partitionId - Id of partition

Throws:

IOException

readFieldsForPartition

public void readFieldsForPartition(DataInput in,
                                   int partitionId)
                            throws IOException

Description copied from interface: MessageStore

Deserialize messages for one partition

Specified by:

readFieldsForPartition in interface MessageStore<I extends org.apache.hadoop.io.WritableComparable,M extends org.apache.hadoop.io.Writable>

Parameters:

in - DataInput to deserialize this object from.

partitionId - Id of partition

Throws:

IOException

clearVertexMessages

public void clearVertexMessages(I vertexId)

Description copied from interface: MessageStore

Clears messages for a vertex.

Specified by:

clearVertexMessages in interface MessageStore<I extends org.apache.hadoop.io.WritableComparable,M extends org.apache.hadoop.io.Writable>

Parameters:

vertexId - Vertex id for which we want to clear messages

clearPartition

public void clearPartition(int partitionId)

Description copied from interface: MessageStore

Clears messages for a partition.

Specified by:

clearPartition in interface MessageStore<I extends org.apache.hadoop.io.WritableComparable,M extends org.apache.hadoop.io.Writable>

Parameters:

partitionId - Partition id for which we want to clear messages

hasMessagesForPartition

public boolean hasMessagesForPartition(int partitionId)

Description copied from interface: MessageStore

Check if we have messages for some partition

Specified by:

hasMessagesForPartition in interface MessageStore<I extends org.apache.hadoop.io.WritableComparable,M extends org.apache.hadoop.io.Writable>

Parameters:

partitionId - Id of partition which we want to check

Returns:

True iff we have messages for the given partition

clearAll

public void clearAll()

Description copied from interface: MessageStore

Clears all resources used by this store.

Specified by:

clearAll in interface MessageStore<I extends org.apache.hadoop.io.WritableComparable,M extends org.apache.hadoop.io.Writable>