com.bigdata.io.writecache

Class WriteCache

java.lang.Object

com.bigdata.io.writecache.WriteCache

All Implemented Interfaces:

IWriteCache

Direct Known Subclasses:

WriteCache.FileChannelScatteredWriteCache, WriteCache.FileChannelWriteCache, WriteCache.ReadCache

public abstract class WriteCache
extends Object
implements IWriteCache

This class provides a write cache with read-through for NIO writes on a FileChannel (and potentially on a remote service). This class is designed to maximize the opportunity for efficient NIO by combining many writes onto a single direct ByteBuffer and then efficiently transferring those writes onto the backing channel in a channel dependent manner. In general, there are three use cases for a WriteCache:

1. Gathered writes. This case is used by the RWStore.
2. Pure append of sequentially allocated records. This case is used by the WORMStrategy (WORM) and by the IndexSegmentBuilder.
3. Write of a single large buffer owned by the caller. This case may be used when the caller wants to manage the buffers or when the caller's buffer is larger than the write cache.

The caller is responsible for managing which buffers are being written on and read on, when they are flushed, and when they are reset. It is perfectly reasonable to have more than one WriteCache and to read through on any WriteCache until it has been recycled. A WriteCache must be reset before it is put into play again for new writes.

Note: For an append-only model (WORM), the caller MUST serialize writes onto the IRawStore and the WriteCache. This is required in order to ensure that the records are laid out in a dense linear fashion on the WriteCache and permits the backing buffer to be transferred in a single IO to the backing file.

Note: For a RWStore, the caller must take more responsibility for managing the WriteCache(s) which are in play and scheduling their eviction onto the backing store. The caller can track the space remaining in each WriteCache and decide when to flush a WriteCache based on that information.

Version:

$Id$

Author:

Bryan Thompson

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	WriteCache.FileChannelScatteredWriteCache The scattered write cache is used by the RWStore since the writes can be made to any part of the file assigned for data allocation.
static class	WriteCache.FileChannelWriteCache A WriteCache implementation suitable for an append-only file such as the WORMStrategy or the output file of the IndexSegmentBuilder.
static class	WriteCache.HAPackage Used to retrieve the HAWriteMessage AND the associated ByteBuffer.
static class	WriteCache.ReadCache
static class	WriteCache.RecordMetadata The metadata associated with a record in the WriteCache.

Field Summary

Fields

Modifier and Type	Field and Description
protected AtomicReference<WriteCacheCounters>	counters The current performance counters.
protected static org.apache.log4j.Logger	log
protected ConcurrentMap<Long,WriteCache.RecordMetadata>	recordMap An index into the write cache used for read through on the cache.

Constructor Summary

Constructors

Constructor and Description
WriteCache(IBufferAccess buf, boolean prefixWrites, boolean useChecksum, boolean isHighlyAvailable, boolean bufferHasData, long fileExtent) Create a WriteCache from either a caller supplied buffer or a direct ByteBuffer allocated from the DirectBufferPool.

Method Summary

Methods

Modifier and Type	Method and Description
ByteBuffer	allocate(int nbytes) Allocate space for a record of the given length on this WriteCache.
int	bytesWritten() The #of bytes written on the backing buffer.
int	capacity() The maximum length of a record which could be inserted into the buffer.
void	close() Permanently take the WriteCache instance out of service.
void	closeForWrites() Called from WriteCacheService to lock buffer content immediately prior to flushing and HA pipline replication.
boolean	contains(long offset) Checks if cache recordMap contains address offset
int	decrementReferenceCount() Although public, it is designed to be used by the WriteCacheService with a memoizer pattern to support concurrent reads to read cache buffers.
void	flush(boolean force) Flush the writes to the backing channel but DOES NOT sync the channel and DOES NOT reset() the WriteCache.
boolean	flush(boolean force, long timeout, TimeUnit unit) Flush the writes to the backing channel but DOES NOT sync the channel and DOES NOT reset() the WriteCache.
protected String	getCompressorKey() Return the optional key for the CompressorRegistry which identifies the IRecordCompressor to be applied.
CounterSet	getCounters() Return the performance counters for the write cacher.
long	getFileExtent()
long	getLastOffset() Used by the HAWriteMessage to retrieve the nextOffset as implied by the recordMap
int	getReferenceCount()
int	incrementReferenceCount() Called when a new reference is acquired
boolean	isClosedForWrites()
ByteBuffer	read(long offset, int nbytes) Read a record from the write cache.
protected void	registerWriteStatus(long offset, int length, char action)
void	reset() Reset the write cache, discarding any writes which have not been written through to the backing channel yet.
void	resetRecordMapFromBuffer() Hook to rebuild RecordMetadata after buffer has been transferred.
protected void	resetRecordMapFromBuffer(ByteBuffer buf, Map<Long,WriteCache.RecordMetadata> recordMap) Overridden by WriteCache.FileChannelScatteredWriteCache.resetRecordMapFromBuffer(ByteBuffer, Map) .
void	setFileExtent(long fileExtent) Set the current extent of the backing file on the WriteCache object.
protected void	setFirstOffset(long firstOffset) Exposed to the WORM for HA support.
void	setRecordMap(Collection<WriteCache.RecordMetadata> map)
String	toString() Adds some debugging information.
boolean	write(long offset, ByteBuffer data, int chk) Write the record on the cache.
protected abstract boolean	writeOnChannel(ByteBuffer buf, long firstOffset, Map<Long,WriteCache.RecordMetadata> recordMap, long nanos) Write the data from the buffer onto the channel.

Methods inherited from class java.lang.Object

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

Field Detail

log

protected static final org.apache.log4j.Logger log

recordMap

protected final ConcurrentMap<Long,WriteCache.RecordMetadata> recordMap

An index into the write cache used for read through on the cache. The keys are the file offsets that would be used to read the corresponding record. The values describe the position in buffer where that record is found and the length of the record.

Note: Exposed to inner classes.

counters

protected final AtomicReference<WriteCacheCounters> counters

The current performance counters.

Constructor Detail

WriteCache

public WriteCache(IBufferAccess buf,
          boolean prefixWrites,
          boolean useChecksum,
          boolean isHighlyAvailable,
          boolean bufferHasData,
          long fileExtent)
           throws InterruptedException

Create a WriteCache from either a caller supplied buffer or a direct ByteBuffer allocated from the DirectBufferPool.

Note: The application MUST ensure that it close()s the WriteCache or it can leak direct ByteBuffers!

Note: NIO operations are performed using a direct ByteBuffer (that is, one use backing bytes are allocated on the C heap). When the caller supplies a ByteBuffer that is allocated on the Java heap as opposed to in native memory, a temporary direct ByteBuffer will be allocated for the IO operation by Java. The JVM can fail to release this temporary direct ByteBuffer, resulting in a memory leak. For this reason, the WriteCache SHOULD use a direct ByteBuffer.

Parameters:

buf - A ByteBuffer to be used as the write cache (optional). When null a buffer will be allocated for you from the DirectBufferPool. Buffers allocated on your behalf will be automatically released by close().

prefixWrites - true iff the implementation uses scattered writes. The RW store uses scattered writes since its updates are written to different parts of the backing file. The WORM store does not since all updates are written to the end of the user extent in the backing file.

useChecksum - true iff the write cache will store the caller's checksum for a record and validate it on read.

isHighlyAvailable - when true the whole record checksum is maintained for use when replicating the write cache along the write pipeline. This needs to be true for HA1 as well since we need to write the HALog.

bufferHasData - when true the caller asserts that the buffer has data (from a replicated write), in which case the position should be the start of the data in the buffer and the limit the #of bytes with valid data. when false, the caller's buffer will be cleared. The code presumes that the WriteCache instance will be used to lay down a single buffer worth of data onto the backing file.

fileExtent - The then current extent of the backing file.

Throws:

InterruptedException

See Also:

ab76d1d4479fffffffffa5abfb09c719a30?bug_id=6210541

Method Detail

setFirstOffset

protected void setFirstOffset(long firstOffset)

Exposed to the WORM for HA support.

Parameters:

firstOffset - The first offset (from the HA message).

toString

public String toString()

Adds some debugging information.

Overrides:

toString in class Object

capacity

public final int capacity()

The maximum length of a record which could be inserted into the buffer.

Note: When checksums are enabled, this is 4 bytes less than the actual capacity of the underlying buffer since each record requires an additional four bytes for the checksum field.

bytesWritten

public final int bytesWritten()

The #of bytes written on the backing buffer.

Note: in order to rely on this value the caller MUST have exclusive access to the buffer. This API does not provide the means for acquiring that exclusive access. This is something that the caller has to arrange for themselves, which is why this is a package private method.

setFileExtent

public void setFileExtent(long fileExtent)

Set the current extent of the backing file on the WriteCache object. When used as part of an HA write pipeline, the receiver is responsible for adjusting its local file size to match the file extent in each WriteCache message.

Parameters:

fileExtent - The current extent of the file.

Throws:

IllegalArgumentException - if the file extent is negative.

See Also:

WriteCacheService.setExtent(long)

getFileExtent

public long getFileExtent()

write

public boolean write(long offset,
            ByteBuffer data,
            int chk)
              throws InterruptedException

Write the record on the cache. This interface DOES NOT provide any guarantee about the ordering of writes. Callers who require a specific ordering must coordinate that ordering themselves, e.g., by synchronizing across their writes onto the cache.

Specified by:

write in interface IWriteCache

Parameters:

offset - The file offset of that record in the backing file.

data - The record. The bytes from the current Buffer.position() to the Buffer.limit() will be written and the Buffer.position() will be advanced to the Buffer.limit() . The caller may subsequently modify the contents of the buffer without changing the state of the cache (i.e., the data are copied into the cache).

chk - The checksum of the data (optional). When checksums are not enabled this should be ZERO (0). When checksums are enabled, #read(long) will validate the checksum before returning data.

Returns:

true iff the caller's record was transferred to the cache. When false, there is not enough room left in the write cache for this record.

Throws:

IllegalStateException - If the buffer is closed.

IllegalArgumentException - If the caller's record is larger than the maximum capacity of cache (the record could not fit within the cache). The caller should check for this and provide special handling for such large records. For example, they can be written directly onto the backing channel.

InterruptedException

read

public ByteBuffer read(long offset,
              int nbytes)
                throws InterruptedException,
                       ChecksumError

Read a record from the write cache.

Specified by:

read in interface IWriteCache

Parameters:

offset - The file offset of that record in the backing file.

nbytes - The length of the record (decoded from the address by the caller).

Returns:

The data read -or- null iff the record does not lie within the IWriteCache. When non-null, this will be a newly allocated exact fit mutable ByteBuffer backed by a Java byte[]. The buffer will be flipped to prepare for reading (the position will be zero and the limit will be the #of bytes read). The data DOES NOT include the bytes used to code checksum even when checksums are enabled.

Throws:

IllegalStateException - If the buffer is closed.

InterruptedException

ChecksumError - if checksums are enabled and the checksum for the record could not be validated.

flush

public void flush(boolean force)
           throws IOException,
                  InterruptedException

Flush the writes to the backing channel but DOES NOT sync the channel and DOES NOT reset() the WriteCache. reset() is a separate operation because a common use is to retain recently flushed instances for read-back.

Specified by:

flush in interface IWriteCache

Parameters:

force - When true, the data will be forced to stable media.

Throws:

IOException

InterruptedException

flush

public boolean flush(boolean force,
            long timeout,
            TimeUnit unit)
              throws IOException,
                     TimeoutException,
                     InterruptedException

Flush the writes to the backing channel but DOES NOT sync the channel and DOES NOT reset() the WriteCache. reset() is a separate operation because a common use is to retain recently flushed instances for read-back.

Specified by:

flush in interface IWriteCache

Parameters:

force - When true, the data will be forced to stable media.

Throws:

IOException

TimeoutException

InterruptedException

writeOnChannel

protected abstract boolean writeOnChannel(ByteBuffer buf,
                     long firstOffset,
                     Map<Long,WriteCache.RecordMetadata> recordMap,
                     long nanos)
                                   throws InterruptedException,
                                          TimeoutException,
                                          IOException

Write the data from the buffer onto the channel. This method provides a uniform means to request that the buffer write itself onto the backing channel, regardless of whether the channel is backed by a file, a socket, etc.

Implementations of this method MAY support gathered writes, depending on the channel. The necessary information to perform a gathered write is present in the recordMap. On the other hand, the implementation MAY require that the records in the cache are laid out for a WORM, in which case getFirstOffset() provides the starting offset for the data to be written. The application MUST coordinate the requirements for a R/W or WORM store with the use of the WriteCache and the means to write on the backing channel.

Parameters:

buf - The data to be written. Only the dirty bytes are visible in this view. The implementation should write all bytes from the current position to the limit.

firstOffset - The offset of the first record in the recordMap into the file (may be relative to a base offset within the file). This is provided as an optimization for the WORM which writes its records contiguously on the backing store.

recordMap - The mapping of record offsets onto metadata about those records.

nanos - The timeout for the operation in nanoseconds.

Returns:

true if the operation was completed successfully within the time alloted.

Throws:

InterruptedException - if the thread was interrupted.

IOException - if there was an IO problem.

TimeoutException

reset

public void reset()
           throws InterruptedException

Reset the write cache, discarding any writes which have not been written through to the backing channel yet. This method IS NOT responsible for discarding writes which have been written through since those are in general available for reading directly from the backing channel. The abort protocol at the IAtomicStore level is responsible for ensuring that processes do not see old data after an abort. This is generally handled by re-loading the appropriate root block and reinitializing various things from that root block..

This implementation clears the buffer, the record map, and other internal metadata such that the WriteCache is prepared to receive new writes.

Specified by:

reset in interface IWriteCache

Throws:

IllegalStateException - if the write cache is closed.

InterruptedException

close

public void close()
           throws InterruptedException

Permanently take the WriteCache instance out of service. If the buffer was allocated by the WriteCache then it is released back to the DirectBufferPool. After this method is called, records can no longer be read from nor written onto the WriteCache. It is safe to invoke this method more than once.

Concurrent read(long, int) requests will be serviced if the already hold the the read lock but requests will fail once the

Specified by:

close in interface IWriteCache

Throws:

InterruptedException

getCompressorKey

protected String getCompressorKey()

Return the optional key for the CompressorRegistry which identifies the IRecordCompressor to be applied.

getCounters

public CounterSet getCounters()

Return the performance counters for the write cacher.

registerWriteStatus

protected void registerWriteStatus(long offset,
                       int length,
                       char action)

setRecordMap

public void setRecordMap(Collection<WriteCache.RecordMetadata> map)

getLastOffset

public long getLastOffset()

Used by the HAWriteMessage to retrieve the nextOffset as implied by the recordMap

Returns:

the last offset value

resetRecordMapFromBuffer

public void resetRecordMapFromBuffer()
                              throws InterruptedException

Hook to rebuild RecordMetadata after buffer has been transferred. For the default WriteCache this is a single entry using firstOffset and current position. For scattered writes, it uses a map with the addr, size, and data inlined.

Throws:

InterruptedException

See Also:

WriteCache.FileChannelScatteredWriteCache

resetRecordMapFromBuffer

protected void resetRecordMapFromBuffer(ByteBuffer buf,
                            Map<Long,WriteCache.RecordMetadata> recordMap)

Overridden by WriteCache.FileChannelScatteredWriteCache.resetRecordMapFromBuffer(ByteBuffer, Map) .

Parameters:

buf -

recordMap -

closeForWrites

public void closeForWrites()
                    throws IllegalStateException,
                           InterruptedException

Called from WriteCacheService to lock buffer content immediately prior to flushing and HA pipline replication. Neither the internal buffer state nor the recordMap may be changed once the WriteCache has been closed for writes. This is necessary to provide 100% binary replication. Otherwise the stores can differ in the data in freed allocation slots.

Throws:

InterruptedException

IllegalStateException

isClosedForWrites

public boolean isClosedForWrites()

allocate

public ByteBuffer allocate(int nbytes)
                    throws IllegalStateException,
                           InterruptedException

Allocate space for a record of the given length on this WriteCache.

Parameters:

nbytes - The size of the record.

Returns:

A view of the allocation on the WriteCache -or- null if there is not enough room in this WriteCache for the allocation.

Throws:

IllegalStateException - if the WriteCache has been closed.

InterruptedException - if the lock could not be acquired.

getReferenceCount

public int getReferenceCount()

incrementReferenceCount

public int incrementReferenceCount()

Called when a new reference is acquired

Returns:

current reference count

decrementReferenceCount

public int decrementReferenceCount()

Although public, it is designed to be used by the WriteCacheService with a memoizer pattern to support concurrent reads to read cache buffers. Called when a reference is released. The return value should be tested and if zero the cache should be returned to the clean list.

Returns:

current reference count

contains

public boolean contains(long offset)

Checks if cache recordMap contains address offset

Parameters:

offset -

Returns: