org.apache.giraph.ooc.data

Class DiskBackedDataStore<T>

java.lang.Object

org.apache.giraph.ooc.data.DiskBackedDataStore<T>

Type Parameters:

T - raw data format of the data store subclassing this class

Direct Known Subclasses:

DiskBackedEdgeStore, DiskBackedMessageStore, DiskBackedPartitionStore

public abstract class DiskBackedDataStore<T>
extends Object

This class provides basic operations for data structures that have to participate in out-of-core mechanism. Essential subclasses of this class are: - DiskBackedPartitionStore (for partition data) - DiskBackedMessageStore (for messages) - DiskBackedEdgeStore (for edges read in INPUT_SUPERSTEP) Basically, any data structure that may cause OOM to happen can be implemented as a subclass of this class. There are two different terms used in the rest of this class: - "data store" refers to in-memory representation of data. Usually this is stored per-partition in in-memory implementations of data structures. For instance, "data store" of a DiskBackedPartitionStore would collection of all partitions kept in the in-memory partition store within the DiskBackedPartitionStore. - "raw data buffer" refers to raw data which were supposed to be de-serialized and added to the data store, but they remain 'as is' in the memory because their corresponding partition is offloaded to disk and is not available in the data store.

Field Summary

Fields

Modifier and Type	Field and Description
protected Set<Integer>	hasPartitionDataOnFile Set containing ids of all partitions where the partition data is in some file on disk.
static IntConfOption	MINIMUM_BUFFER_SIZE_TO_FLUSH Minimum size of a buffer (in bytes) to flush to disk.
protected OutOfCoreEngine	oocEngine Out-of-core engine

Method Summary

Modifier and Type	Method and Description
protected void	addEntry(int partitionId, T entry) Adds a data entry for a given partition to the current data store.
protected abstract void	addEntryToInMemoryPartitionData(int partitionId, T entry) Adds a single entry for a given partition to the in-memory data store.
protected abstract int	entrySerializedSize(T entry) Gets the size of a given entry in bytes.
Set<Integer>	getCandidateBuffersToOffload(int ioThreadId) Looks through all partitions that their data is not in the data store (is offloaded to disk), and sees if any of them has enough raw data buffer in memory.
protected abstract long	loadInMemoryPartitionData(int partitionId, int ioThreadId, DataIndex index) Loads data of a partition into data store.
abstract long	loadPartitionData(int partitionId) Loads and assembles all data for a given partition, and put it into the data store.
protected long	loadPartitionDataProxy(int partitionId, DataIndex index) The proxy method that does the actual operation for `loadPartitionData`, but uses the data index given by the caller.
abstract 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	offloadBuffersProxy(int partitionId, DataIndex index) The proxy method that does the actual operation for `offloadBuffers`, but uses the data index given by the caller.
protected abstract long	offloadInMemoryPartitionData(int partitionId, int ioThreadId, DataIndex index) Offloads data of a partition in data store to disk.
abstract 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.
protected long	offloadPartitionDataProxy(int partitionId, DataIndex index) The proxy method that does the actual operation for `offloadPartitionData`, but uses the data index given by the caller.
protected abstract T	readNextEntry(DataInput in) Reads the next available raw entry from a given input stream.
protected abstract void	writeEntry(T entry, DataOutput out) Writes a single raw entry to a given output stream.

Methods inherited from class java.lang.Object

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

Field Detail

MINIMUM_BUFFER_SIZE_TO_FLUSH

public static final IntConfOption MINIMUM_BUFFER_SIZE_TO_FLUSH

Minimum size of a buffer (in bytes) to flush to disk. This is used to decide whether vertex/edge buffers are large enough to flush to disk.

oocEngine

protected final OutOfCoreEngine oocEngine

Out-of-core engine

hasPartitionDataOnFile

protected final Set<Integer> hasPartitionDataOnFile

Set containing ids of all partitions where the partition data is in some file on disk. Note that the out-of-core mechanism may decide to put the data for a partition on disk, while the partition data is empty. For instance, at the beginning of a superstep, out-of-core mechanism may decide to put incoming messages of a partition on disk, while the partition has not received any messages. In such scenarios, the "out-of-core mechanism" thinks that the partition data is on disk, while disk-backed data stores may want to optimize for IO/metadata accesses and decide not to create/write anything on files on disk. In summary, there is a subtle difference between this field and `hasPartitionOnDisk` field. Basically, this field is used for optimizing IO (mainly metadata) accesses by disk-backed stores, while `hasPartitionDataOnDisk` is the view that out-of-core mechanism has regarding partition storage statuses. Since out-of-core mechanism does not know about the actual data for a partition, these two fields have to be separate.

Method Detail

addEntry

protected void addEntry(int partitionId,
                        T entry)

Adds a data entry for a given partition to the current data store. If data of a given partition in data store is already offloaded to disk, adds the data entry to appropriate raw data buffer list.

Parameters:

partitionId - id of the partition to add the data entry to

entry - data entry to add

loadPartitionData

public abstract long loadPartitionData(int partitionId)
                                throws IOException

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.

Parameters:

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

Returns:

number of bytes loaded from disk to memory

Throws:

IOException

loadPartitionDataProxy

protected long loadPartitionDataProxy(int partitionId,
                                      DataIndex index)
                               throws IOException

The proxy method that does the actual operation for `loadPartitionData`, but uses the data index given by the caller.

Parameters:

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

index - data index chain for the data to load

Returns:

number of bytes loaded from disk to memory

Throws:

IOException

offloadPartitionData

public abstract long offloadPartitionData(int partitionId)
                                   throws IOException

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

Parameters:

partitionId - id of the partition to offload its data

Returns:

number of bytes offloaded from memory to disk

Throws:

IOException

offloadPartitionDataProxy

protected long offloadPartitionDataProxy(int partitionId,
                                         DataIndex index)
                                  throws IOException

The proxy method that does the actual operation for `offloadPartitionData`, but uses the data index given by the caller.

Parameters:

partitionId - id of the partition to offload its data

index - data index chain for the data to offload

Returns:

number of bytes offloaded from memory to disk

Throws:

IOException

offloadBuffers

public abstract long offloadBuffers(int partitionId)
                             throws IOException

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

Parameters:

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

Returns:

number of bytes offloaded from memory to disk

Throws:

IOException

offloadBuffersProxy

protected long offloadBuffersProxy(int partitionId,
                                   DataIndex index)
                            throws IOException

The proxy method that does the actual operation for `offloadBuffers`, but uses the data index given by the caller.

Parameters:

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

index - data index chain for the data to offload its buffers

Returns:

number of bytes offloaded from memory to disk

Throws:

IOException

getCandidateBuffersToOffload

public Set<Integer> getCandidateBuffersToOffload(int ioThreadId)

Looks through all partitions that their data is not in the data store (is offloaded to disk), and sees if any of them has enough raw data buffer in memory. If so, puts that partition in a list to return.

Parameters:

ioThreadId - Id of the IO thread who would offload the buffers

Returns:

Set of partition ids of all partition raw buffers where the aggregate size of buffers are large enough and it is worth flushing those buffers to disk

writeEntry

protected abstract void writeEntry(T entry,
                                   DataOutput out)
                            throws IOException

Writes a single raw entry to a given output stream.

Parameters:

entry - entry to write to output

out - output stream to write the entry to

Throws:

IOException

readNextEntry

protected abstract T readNextEntry(DataInput in)
                            throws IOException

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

Parameters:

in - input stream to read the entry from

Returns:

entry read from an input stream

Throws:

IOException

loadInMemoryPartitionData

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

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

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 abstract long offloadInMemoryPartitionData(int partitionId,
                                                     int ioThreadId,
                                                     DataIndex index)
                                              throws IOException

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

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 abstract int entrySerializedSize(T entry)

Gets the size of a given entry in bytes.

Parameters:

entry - input entry to find its size

Returns:

size of given input entry in bytes

addEntryToInMemoryPartitionData

protected abstract void addEntryToInMemoryPartitionData(int partitionId,
                                                        T entry)

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

Parameters:

partitionId - id of the partition to add the data to

entry - input entry to add to the data store