org.apache.giraph.ooc.data

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

java.lang.Object

org.apache.giraph.ooc.data.DiskBackedDataStore<VertexIdMessages<I,M>>

org.apache.giraph.ooc.data.DiskBackedMessageStore<I,M>

Type Parameters:

I - Vertex id

M - Message data

All Implemented Interfaces:

MessageStore<I,M>

public class DiskBackedMessageStore<I extends org.apache.hadoop.io.WritableComparable,M extends org.apache.hadoop.io.Writable>
extends DiskBackedDataStore<VertexIdMessages<I,M>>
implements MessageStore<I,M>

Implementation of a message store used for out-of-core mechanism.

Field Summary

Fields inherited from class org.apache.giraph.ooc.data.DiskBackedDataStore

hasPartitionDataOnFile, MINIMUM_BUFFER_SIZE_TO_FLUSH, oocEngine

Constructor Summary

Constructors

Constructor and Description
DiskBackedMessageStore(ImmutableClassesGiraphConfiguration<I,?,?> config, OutOfCoreEngine oocEngine, MessageStore<I,M> messageStore, boolean useMessageCombiner, long superstep) Constructor

Method Summary

Modifier and Type	Method and Description
protected void	addEntryToInMemoryPartitionData(int partitionId, VertexIdMessages<I,M> messages) Adds a single entry for a given partition to the in-memory data store.
void	addMessage(I vertexId, M message) Adds a message for a particular vertex The method is used by InternalMessageStore to send local messages; for the general case, use a more efficient addPartitionMessages
void	addPartitionMessages(int partitionId, VertexIdMessages<I,M> messages) Adds messages for partition
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.
protected int	entrySerializedSize(VertexIdMessages<I,M> messages) Gets the size of a given entry in bytes.
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
Iterable<I>	getPartitionDestinationVertices(int partitionId) Gets vertex ids from selected partition which we have messages for
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
boolean	isPointerListEncoding() True if this message-store encodes messages as a list of long pointers to compact serialized messages
protected long	loadInMemoryPartitionData(int partitionId, int ioThreadId, DataIndex index) Loads data of a partition into data store.
long	loadPartitionData(int partitionId) Loads and assembles all data for a given partition, and put it into the data store.
long	offloadBuffers(int partitionId) Offloads raw data buffers of a given partition to disk, and returns the number of bytes offloaded from memory to disk.
protected long	offloadInMemoryPartitionData(int partitionId, int ioThreadId, DataIndex index) Offloads data of a partition in data store to disk.
long	offloadPartitionData(int partitionId) Offloads partition data of a given partition in the data store to disk, and returns the number of bytes offloaded from memory to disk.
void	readFieldsForPartition(DataInput in, int partitionId) Deserialize messages for one partition
protected VertexIdMessages<I,M>	readNextEntry(DataInput in) Reads the next available raw entry from a given input stream.
protected void	writeEntry(VertexIdMessages<I,M> messages, DataOutput out) Writes a single raw entry to a given output stream.
void	writePartition(DataOutput out, int partitionId) Serialize messages for one partition.

Methods inherited from class org.apache.giraph.ooc.data.DiskBackedDataStore

addEntry, getCandidateBuffersToOffload, loadPartitionDataProxy, offloadBuffersProxy, offloadPartitionDataProxy

Methods inherited from class java.lang.Object

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

Constructor Detail

DiskBackedMessageStore

public DiskBackedMessageStore(ImmutableClassesGiraphConfiguration<I,?,?> config,
                              OutOfCoreEngine oocEngine,
                              MessageStore<I,M> messageStore,
                              boolean useMessageCombiner,
                              long superstep)

Constructor

Parameters:

config - Configuration

oocEngine - Out-of-core engine

messageStore - In-memory message store for which out-of-core message store would be wrapper

useMessageCombiner - Whether message combiner is used for this message store

superstep - superstep number this messages store is used for

Method Detail

isPointerListEncoding

public boolean isPointerListEncoding()

Description copied from interface: MessageStore

True if this message-store encodes messages as a list of long pointers to compact serialized messages

Specified by:

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

Returns:

true if we encode messages as a list of pointers

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

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

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>

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

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

addPartitionMessages

public void addPartitionMessages(int partitionId,
                                 VertexIdMessages<I,M> messages)

Description copied from interface: MessageStore

Adds messages for partition

Specified by:

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

Parameters:

partitionId - Id of partition

messages - Collection of vertex ids and messages we want to add

addMessage

public void addMessage(I vertexId,
                       M message)
                throws IOException

Description copied from interface: MessageStore

Adds a message for a particular vertex The method is used by InternalMessageStore to send local messages; for the general case, use a more efficient addPartitionMessages

Specified by:

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

Parameters:

vertexId - Id of target vertex

message - A message to send

Throws:

IOException

loadPartitionData

public long loadPartitionData(int partitionId)
                       throws IOException

Description copied from class: DiskBackedDataStore

Loads and assembles all data for a given partition, and put it into the data store. Returns the number of bytes transferred from disk to memory in the loading process.

Specified by:

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

Parameters:

partitionId - id of the partition to load and assemble all data for

Returns:

number of bytes loaded from disk to memory

Throws:

IOException

offloadPartitionData

public long offloadPartitionData(int partitionId)
                          throws IOException

Description copied from class: DiskBackedDataStore

Offloads partition data of a given partition in the data store to disk, and returns the number of bytes offloaded from memory to disk.

Specified by:

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

Parameters:

partitionId - id of the partition to offload its data

Returns:

number of bytes offloaded from memory to disk

Throws:

IOException

offloadBuffers

public long offloadBuffers(int partitionId)
                    throws IOException

Description copied from class: DiskBackedDataStore

Offloads raw data buffers of a given partition to disk, and returns the number of bytes offloaded from memory to disk.

Specified by:

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

Parameters:

partitionId - id of the partition to offload its raw data buffers

Returns:

number of bytes offloaded from memory to disk

Throws:

IOException

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

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

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

writeEntry

protected void writeEntry(VertexIdMessages<I,M> messages,
                          DataOutput out)
                   throws IOException

Description copied from class: DiskBackedDataStore

Writes a single raw entry to a given output stream.

Specified by:

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

Parameters:

messages - entry to write to output

out - output stream to write the entry to

Throws:

IOException

readNextEntry

protected VertexIdMessages<I,M> readNextEntry(DataInput in)
                                       throws IOException

Description copied from class: DiskBackedDataStore

Reads the next available raw entry from a given input stream.

Specified by:

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

Parameters:

in - input stream to read the entry from

Returns:

entry read from an input stream

Throws:

IOException

loadInMemoryPartitionData

protected long loadInMemoryPartitionData(int partitionId,
                                         int ioThreadId,
                                         DataIndex index)
                                  throws IOException

Description copied from class: DiskBackedDataStore

Loads data of a partition into data store. Returns number of bytes loaded.

Specified by:

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

Parameters:

partitionId - id of the partition to load its data

ioThreadId - id of the IO thread performing the load

index - data index chain for the data to load

Returns:

number of bytes loaded from disk to memory

Throws:

IOException

offloadInMemoryPartitionData

protected long offloadInMemoryPartitionData(int partitionId,
                                            int ioThreadId,
                                            DataIndex index)
                                     throws IOException

Description copied from class: DiskBackedDataStore

Offloads data of a partition in data store to disk. Returns the number of bytes offloaded to disk

Specified by:

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

Parameters:

partitionId - id of the partition to offload to disk

ioThreadId - id of the IO thread performing the offload

index - data index chain for the data to offload

Returns:

number of bytes offloaded from memory to disk

Throws:

IOException

entrySerializedSize

protected int entrySerializedSize(VertexIdMessages<I,M> messages)

Description copied from class: DiskBackedDataStore

Gets the size of a given entry in bytes.

Specified by:

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

Parameters:

messages - input entry to find its size

Returns:

size of given input entry in bytes

addEntryToInMemoryPartitionData

protected void addEntryToInMemoryPartitionData(int partitionId,
                                               VertexIdMessages<I,M> messages)

Description copied from class: DiskBackedDataStore

Adds a single entry for a given partition to the in-memory data store.

Specified by:

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

Parameters:

partitionId - id of the partition to add the data to

messages - input entry to add to the data store