org.apache.hadoop.hbase.regionserver.wal

Class AsyncFSWAL

java.lang.Object

org.apache.hadoop.hbase.regionserver.wal.AbstractFSWAL<WALProvider.AsyncWriter>

org.apache.hadoop.hbase.regionserver.wal.AsyncFSWAL

All Implemented Interfaces:

Closeable, AutoCloseable, WALFileLengthProvider, WAL

@InterfaceAudience.LimitedPrivate(value="Configuration")
public class AsyncFSWAL
extends AbstractFSWAL<WALProvider.AsyncWriter>

An asynchronous implementation of FSWAL.

Here 'waitingConsumePayloads' acts as the RingBuffer in FSHLog.

For append, we process it as follow:

1. In the caller thread(typically, in the rpc handler thread):
   1. Insert the entry into 'waitingConsumePayloads'. Use ringbuffer sequence as txid.
   2. Schedule the consumer task if needed. See shouldScheduleConsumer() for more details.
2. In the consumer task(executed in a single threaded thread pool)
   1. Poll the entry from waitingConsumePayloads and insert it into toWriteAppends
   2. Poll the entry from toWriteAppends, append it to the AsyncWriter, and insert it into unackedAppends
   3. If the buffered size reaches batchSize, or there is a sync request, then we call sync on the AsyncWriter.
   4. In the callback methods:
      • If succeeded, poll the entry from unackedAppends and drop it.
      • If failed, add all the entries in unackedAppends back to toWriteAppends and wait for writing them again.

For sync, the processing stages are almost same. And different from FSHLog, we will open a new writer and rewrite unacked entries to the new writer and sync again if we hit a sync error.

Here we only describe the logic of doReplaceWriter. The main logic of rollWriter is same with FSHLog.
For a normal roll request(for example, we have reached the log roll size):

1. In the log roller thread, we will set waitingRoll(int) to true and readyForRolling to false, and then wait on readyForRolling(see waitForSafePoint()).
2. In the consumer thread, we will stop polling entries from waitingConsumePayloads if waitingRoll(int) is true, and also stop writing the entries in toWriteAppends out.
3. If there are unflush data in the writer, sync them.
4. When all out-going sync request is finished, i.e, the unackedAppends is empty, signal the readyForRollingCond.
5. Back to the log roller thread, now we can confirm that there are no out-going entries, i.e., we reach a safe point. So it is safe to replace old writer with new writer now.
6. Set writerBroken(int) and waitingRoll(int) to false.
7. Schedule the consumer task.
8. Schedule a background task to close the old writer.

For a broken writer roll request, the only difference is that we can bypass the wait for safe point stage.

Nested Class Summary

Nested classes/interfaces inherited from interface org.apache.hadoop.hbase.wal.WAL

WAL.Entry, WAL.Reader

Field Summary

Fields

Modifier and Type	Field and Description
static String	ASYNC_WAL_USE_SHARED_EVENT_LOOP
static String	ASYNC_WAL_WAIT_ON_SHUTDOWN_IN_SECONDS
private long	batchSize
private Class<? extends org.apache.hbase.thirdparty.io.netty.channel.Channel>	channelClass
private ExecutorService	closeExecutor
private ExecutorService	consumeExecutor
private Lock	consumeLock
private Runnable	consumer
private AtomicBoolean	consumerScheduled
static boolean	DEFAULT_ASYNC_WAL_USE_SHARED_EVENT_LOOP
static int	DEFAULT_ASYNC_WAL_WAIT_ON_SHUTDOWN_IN_SECONDS
static long	DEFAULT_WAL_BATCH_SIZE
private int	epochAndState
private org.apache.hbase.thirdparty.io.netty.channel.EventLoopGroup	eventLoopGroup
private long	fileLengthAtLastSync
private AsyncFSOutput	fsOut
private Supplier<Boolean>	hasConsumerTask
private long	highestProcessedAppendTxid
private long	highestProcessedAppendTxidAtLastSync
private static org.slf4j.Logger	LOG
private static int	MAX_EPOCH
private boolean	readyForRolling
private Condition	readyForRollingCond
private static Comparator<SyncFuture>	SEQ_COMPARATOR
private SortedSet<SyncFuture>	syncFutures
private Deque<FSWALEntry>	toWriteAppends
private Deque<FSWALEntry>	unackedAppends
private com.lmax.disruptor.RingBuffer<RingBufferTruck>	waitingConsumePayloads
private com.lmax.disruptor.Sequence	waitingConsumePayloadsGatingSequence
private int	waitOnShutdownInSeconds
static String	WAL_BATCH_SIZE

Fields inherited from class org.apache.hadoop.hbase.regionserver.wal.AbstractFSWAL

blocksize, closed, conf, coprocessorHost, DEFAULT_SLOW_SYNC_TIME_MS, filenum, fs, highestSyncedTxid, highestUnsyncedTxid, implClassName, listeners, LOG_NAME_COMPARATOR, logrollsize, maxLogs, numEntries, ourFiles, prefixPathStr, rollRequested, rollWriterLock, sequenceIdAccounting, shutdown, slowSyncNs, totalLogSize, walArchiveDir, walDir, walFile2Props, walFilePrefix, walFileSuffix, writer

Constructor Summary

Constructors

Constructor and Description

AsyncFSWAL(org.apache.hadoop.fs.FileSystem fs, org.apache.hadoop.fs.Path rootDir, String logDir, String archiveDir, org.apache.hadoop.conf.Configuration conf, List<WALActionsListener> listeners, boolean failIfWALExists, String prefix, String suffix, org.apache.hbase.thirdparty.io.netty.channel.EventLoopGroup eventLoopGroup, Class<? extends org.apache.hbase.thirdparty.io.netty.channel.Channel> channelClass)

Method Summary

Modifier and Type	Method and Description
private void	addTimeAnnotation(SyncFuture future, String annotation)
long	append(RegionInfo hri, WALKeyImpl key, WALEdit edits, boolean inMemstore) NOTE: This append, at a time that is usually after this call returns, starts an mvcc transaction by calling 'begin' wherein which we assign this update a sequenceid.
private void	appendAndSync()
private long	closeWriter()
private void	consume()
protected WALProvider.AsyncWriter	createWriterInstance(org.apache.hadoop.fs.Path path)
protected void	doAppend(WALProvider.AsyncWriter writer, FSWALEntry entry)
protected boolean	doCheckLogLowReplication()
protected void	doReplaceWriter(org.apache.hadoop.fs.Path oldPath, org.apache.hadoop.fs.Path newPath, WALProvider.AsyncWriter nextWriter) Notice that you need to clear the AbstractFSWAL.rollRequested flag in this method, as the new writer will begin to work before returning from this method.
protected void	doShutdown()
private static int	epoch(int epochAndState)
private int	finishSync(boolean addSyncTrace)
private int	finishSyncLowerThanTxid(long txid, boolean addSyncTrace)
(package private) int	getLogReplication() This method gets the datanode replication count for the current WAL.
(package private) org.apache.hadoop.hdfs.protocol.DatanodeInfo[]	getPipeline() This method gets the pipeline for the current WAL.
private boolean	shouldScheduleConsumer()
void	sync() Sync what we have in the WAL.
void	sync(long txid) Sync the WAL if the txId was not already sync'd.
private void	sync(WALProvider.AsyncWriter writer)
private void	syncCompleted(WALProvider.AsyncWriter writer, long processedTxid, long startTimeNs)
private void	syncFailed(long epochWhenSync, Throwable error)
private boolean	trySetReadyForRolling()
private void	waitForSafePoint()
private static boolean	waitingRoll(int epochAndState)
private static boolean	writerBroken(int epochAndState)

Methods inherited from class org.apache.hadoop.hbase.regionserver.wal.AbstractFSWAL

abortCacheFlush, append, atHeadOfRingBufferEventHandlerAppend, blockOnSync, checkLogLowReplication, close, completeCacheFlush, computeFilename, findRegionsToForceFlush, getCoprocessorHost, getCurrentFileName, getEarliestMemStoreSeqNum, getEarliestMemStoreSeqNum, getFilenum, getFileNumFromFileName, getFiles, getLogFileSize, getLogFileSizeIfBeingWritten, getNumLogFiles, getNumRolledLogFiles, getOldPath, getPreallocatedEventCount, getSyncFuture, getUnflushedEntriesCount, getWALArchivePath, isLogRollRequested, isUnflushedEntries, logRollAndSetupWalProps, main, postSync, registerWALActionsListener, replaceWriter, requestLogRoll, requestLogRoll, rollWriter, rollWriter, shutdown, stampSequenceIdAndPublishToRingBuffer, startCacheFlush, startCacheFlush, toString, unregisterWALActionsListener, updateStore

Methods inherited from class java.lang.Object

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

Field Detail

LOG

private static final org.slf4j.Logger LOG

SEQ_COMPARATOR

private static final Comparator<SyncFuture> SEQ_COMPARATOR

WAL_BATCH_SIZE

public static final String WAL_BATCH_SIZE

See Also:

Constant Field Values

DEFAULT_WAL_BATCH_SIZE

public static final long DEFAULT_WAL_BATCH_SIZE

See Also:

Constant Field Values

ASYNC_WAL_USE_SHARED_EVENT_LOOP

public static final String ASYNC_WAL_USE_SHARED_EVENT_LOOP

See Also:

Constant Field Values

DEFAULT_ASYNC_WAL_USE_SHARED_EVENT_LOOP

public static final boolean DEFAULT_ASYNC_WAL_USE_SHARED_EVENT_LOOP

See Also:

Constant Field Values

ASYNC_WAL_WAIT_ON_SHUTDOWN_IN_SECONDS

public static final String ASYNC_WAL_WAIT_ON_SHUTDOWN_IN_SECONDS

See Also:

Constant Field Values

DEFAULT_ASYNC_WAL_WAIT_ON_SHUTDOWN_IN_SECONDS

public static final int DEFAULT_ASYNC_WAL_WAIT_ON_SHUTDOWN_IN_SECONDS

See Also:

Constant Field Values

eventLoopGroup

private final org.apache.hbase.thirdparty.io.netty.channel.EventLoopGroup eventLoopGroup

consumeExecutor

private final ExecutorService consumeExecutor

channelClass

private final Class<? extends org.apache.hbase.thirdparty.io.netty.channel.Channel> channelClass

consumeLock

private final Lock consumeLock

consumer

private final Runnable consumer

hasConsumerTask

private final Supplier<Boolean> hasConsumerTask

MAX_EPOCH

private static final int MAX_EPOCH

See Also:

Constant Field Values

epochAndState

private volatile int epochAndState

readyForRolling

private boolean readyForRolling

readyForRollingCond

private final Condition readyForRollingCond

waitingConsumePayloads

private final com.lmax.disruptor.RingBuffer<RingBufferTruck> waitingConsumePayloads

waitingConsumePayloadsGatingSequence

private final com.lmax.disruptor.Sequence waitingConsumePayloadsGatingSequence

consumerScheduled

private final AtomicBoolean consumerScheduled

batchSize

private final long batchSize

closeExecutor

private final ExecutorService closeExecutor

fsOut

private volatile AsyncFSOutput fsOut

toWriteAppends

private final Deque<FSWALEntry> toWriteAppends

unackedAppends

private final Deque<FSWALEntry> unackedAppends

syncFutures

private final SortedSet<SyncFuture> syncFutures

highestProcessedAppendTxid

private long highestProcessedAppendTxid

fileLengthAtLastSync

private long fileLengthAtLastSync

highestProcessedAppendTxidAtLastSync

private long highestProcessedAppendTxidAtLastSync

waitOnShutdownInSeconds

private final int waitOnShutdownInSeconds

Constructor Detail

AsyncFSWAL

public AsyncFSWAL(org.apache.hadoop.fs.FileSystem fs,
                  org.apache.hadoop.fs.Path rootDir,
                  String logDir,
                  String archiveDir,
                  org.apache.hadoop.conf.Configuration conf,
                  List<WALActionsListener> listeners,
                  boolean failIfWALExists,
                  String prefix,
                  String suffix,
                  org.apache.hbase.thirdparty.io.netty.channel.EventLoopGroup eventLoopGroup,
                  Class<? extends org.apache.hbase.thirdparty.io.netty.channel.Channel> channelClass)
           throws FailedLogCloseException,
                  IOException

Throws:

FailedLogCloseException

IOException

Method Detail

waitingRoll

private static boolean waitingRoll(int epochAndState)

writerBroken

private static boolean writerBroken(int epochAndState)

epoch

private static int epoch(int epochAndState)

trySetReadyForRolling

private boolean trySetReadyForRolling()

syncFailed

private void syncFailed(long epochWhenSync,
                        Throwable error)

syncCompleted

private void syncCompleted(WALProvider.AsyncWriter writer,
                           long processedTxid,
                           long startTimeNs)

sync

private void sync(WALProvider.AsyncWriter writer)

addTimeAnnotation

private void addTimeAnnotation(SyncFuture future,
                               String annotation)

finishSyncLowerThanTxid

private int finishSyncLowerThanTxid(long txid,
                                    boolean addSyncTrace)

finishSync

private int finishSync(boolean addSyncTrace)

appendAndSync

private void appendAndSync()

consume

private void consume()

shouldScheduleConsumer

private boolean shouldScheduleConsumer()

append

public long append(RegionInfo hri,
                   WALKeyImpl key,
                   WALEdit edits,
                   boolean inMemstore)
            throws IOException

Description copied from class: AbstractFSWAL

NOTE: This append, at a time that is usually after this call returns, starts an mvcc transaction by calling 'begin' wherein which we assign this update a sequenceid. At assignment time, we stamp all the passed in Cells inside WALEdit with their sequenceId. You must 'complete' the transaction this mvcc transaction by calling MultiVersionConcurrencyControl#complete(...) or a variant otherwise mvcc will get stuck. Do it in the finally of a try/finally block within which this append lives and any subsequent operations like sync or update of memstore, etc. Get the WriteEntry to pass mvcc out of the passed in WALKey walKey parameter. Be warned that the WriteEntry is not immediately available on return from this method. It WILL be available subsequent to a sync of this append; otherwise, you will just have to wait on the WriteEntry to get filled in.

Specified by:

append in interface WAL

Specified by:

append in class AbstractFSWAL<WALProvider.AsyncWriter>

Parameters:

hri - the regioninfo associated with append

key - Modified by this call; we add to it this edits region edit/sequence id.

edits - Edits to append. MAY CONTAIN NO EDITS for case where we want to get an edit sequence id that is after all currently appended edits.

inMemstore - Always true except for case where we are writing a compaction completion record into the WAL; in this case the entry is just so we can finish an unfinished compaction -- it is not an edit for memstore.

Returns:

Returns a 'transaction id' and key will have the region edit/sequence id in it.

Throws:

IOException

sync

public void sync()
          throws IOException

Description copied from interface: WAL

Sync what we have in the WAL.

Throws:

IOException

sync

public void sync(long txid)
          throws IOException

Description copied from interface: WAL

Sync the WAL if the txId was not already sync'd.

Parameters:

txid - Transaction id to sync to.

Throws:

IOException

createWriterInstance

protected WALProvider.AsyncWriter createWriterInstance(org.apache.hadoop.fs.Path path)
                                                throws IOException

Specified by:

createWriterInstance in class AbstractFSWAL<WALProvider.AsyncWriter>

Throws:

IOException

waitForSafePoint

private void waitForSafePoint()

closeWriter

private long closeWriter()

doReplaceWriter

protected void doReplaceWriter(org.apache.hadoop.fs.Path oldPath,
                               org.apache.hadoop.fs.Path newPath,
                               WALProvider.AsyncWriter nextWriter)
                        throws IOException

Description copied from class: AbstractFSWAL

Notice that you need to clear the AbstractFSWAL.rollRequested flag in this method, as the new writer will begin to work before returning from this method. If we clear the flag after returning from this call, we may miss a roll request. The implementation class should choose a proper place to clear the AbstractFSWAL.rollRequested flag so we do not miss a roll request, typically before you start writing to the new writer.

Specified by:

doReplaceWriter in class AbstractFSWAL<WALProvider.AsyncWriter>

Throws:

IOException

doShutdown

protected void doShutdown()
                   throws IOException

Specified by:

doShutdown in class AbstractFSWAL<WALProvider.AsyncWriter>

Throws:

IOException

doAppend

protected void doAppend(WALProvider.AsyncWriter writer,
                        FSWALEntry entry)

Specified by:

doAppend in class AbstractFSWAL<WALProvider.AsyncWriter>

getPipeline

org.apache.hadoop.hdfs.protocol.DatanodeInfo[] getPipeline()

Description copied from class: AbstractFSWAL

This method gets the pipeline for the current WAL.

Specified by:

getPipeline in class AbstractFSWAL<WALProvider.AsyncWriter>

getLogReplication

int getLogReplication()

Description copied from class: AbstractFSWAL

This method gets the datanode replication count for the current WAL.

Specified by:

getLogReplication in class AbstractFSWAL<WALProvider.AsyncWriter>

doCheckLogLowReplication

protected boolean doCheckLogLowReplication()

Specified by:

doCheckLogLowReplication in class AbstractFSWAL<WALProvider.AsyncWriter>