org.apache.hadoop.hbase.regionserver.wal

Class FSHLog.RingBufferEventHandler

java.lang.Object

org.apache.hadoop.hbase.regionserver.wal.FSHLog.RingBufferEventHandler

All Implemented Interfaces:

com.lmax.disruptor.EventHandler<RingBufferTruck>, com.lmax.disruptor.LifecycleAware

Enclosing class:

FSHLog

class FSHLog.RingBufferEventHandler
extends Object
implements com.lmax.disruptor.EventHandler<RingBufferTruck>, com.lmax.disruptor.LifecycleAware

Handler that is run by the disruptor ringbuffer consumer. Consumer is a SINGLE 'writer/appender' thread. Appends edits and starts up sync runs. Tries its best to batch up syncs. There is no discernible benefit batching appends so we just append as they come in because it simplifies the below implementation. See metrics for batching effectiveness (In measurement, at 100 concurrent handlers writing 1k, we are batching > 10 appends and 10 handler sync invocations for every actual dfsclient sync call; at 10 concurrent handlers, YMMV).

Herein, we have an array into which we store the sync futures as they come in. When we have a 'batch', we'll then pass what we have collected to a SyncRunner thread to do the filesystem sync. When it completes, it will then call SyncFuture.done(long, Throwable) on each of SyncFutures in the batch to release blocked Handler threads.

I've tried various effects to try and make latencies low while keeping throughput high. I've tried keeping a single Queue of SyncFutures in this class appending to its tail as the syncs coming and having sync runner threads poll off the head to 'finish' completed SyncFutures. I've tried linkedlist, and various from concurrent utils whether LinkedBlockingQueue or ArrayBlockingQueue, etc. The more points of synchronization, the more 'work' (according to 'perf stats') that has to be done; small increases in stall percentages seem to have a big impact on throughput/latencies. The below model where we have an array into which we stash the syncs and then hand them off to the sync thread seemed like a decent compromise. See HBASE-8755 for more detail.

Field Summary

Fields

Modifier and Type	Field and Description
private Exception	exception Set if we get an exception appending or syncing so that all subsequence appends and syncs on this WAL fail until WAL is replaced.
private Object	safePointWaiter Object to block on while waiting on safe point.
private boolean	shutdown
private SyncFuture[]	syncFutures
private AtomicInteger	syncFuturesCount
private int	syncRunnerIndex Which syncrunner to use next.
private FSHLog.SyncRunner[]	syncRunners
private FSHLog.SafePointZigZagLatch	zigzagLatch

Constructor Summary

Constructors

Constructor and Description
RingBufferEventHandler(int syncRunnerCount, int maxBatchCount)

Method Summary

Modifier and Type	Method and Description
(package private) void	append(FSWALEntry entry) Append to the WAL.
(package private) FSHLog.SafePointZigZagLatch	attainSafePoint()
private void	attainSafePoint(long currentSequence) Check if we should attain safe point.
private void	cleanupOutstandingSyncsOnException(long sequence, Exception e)
private boolean	isOutstandingSyncs() Returns True if outstanding sync futures still
private boolean	isOutstandingSyncsFromRunners()
private void	offerDoneSyncsBackToCache() Offers the finished syncs back to the cache for reuse.
void	onEvent(RingBufferTruck truck, long sequence, boolean endOfBatch)
void	onShutdown()
void	onStart()

Methods inherited from class java.lang.Object

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

Field Detail

syncRunners

private final FSHLog.SyncRunner[] syncRunners

syncFutures

private final SyncFuture[] syncFutures

syncFuturesCount

private AtomicInteger syncFuturesCount

zigzagLatch

private volatile FSHLog.SafePointZigZagLatch zigzagLatch

exception

private Exception exception

Set if we get an exception appending or syncing so that all subsequence appends and syncs on this WAL fail until WAL is replaced.

safePointWaiter

private final Object safePointWaiter

Object to block on while waiting on safe point.

shutdown

private volatile boolean shutdown

syncRunnerIndex

private int syncRunnerIndex

Which syncrunner to use next.

Constructor Detail

RingBufferEventHandler

RingBufferEventHandler(int syncRunnerCount,
                       int maxBatchCount)

Method Detail

cleanupOutstandingSyncsOnException

private void cleanupOutstandingSyncsOnException(long sequence,
                                                Exception e)

offerDoneSyncsBackToCache

private void offerDoneSyncsBackToCache()

Offers the finished syncs back to the cache for reuse.

isOutstandingSyncs

private boolean isOutstandingSyncs()

Returns True if outstanding sync futures still

isOutstandingSyncsFromRunners

private boolean isOutstandingSyncsFromRunners()

onEvent

public void onEvent(RingBufferTruck truck,
                    long sequence,
                    boolean endOfBatch)
             throws Exception

Specified by:

onEvent in interface com.lmax.disruptor.EventHandler<RingBufferTruck>

Throws:

Exception

attainSafePoint

FSHLog.SafePointZigZagLatch attainSafePoint()

attainSafePoint

private void attainSafePoint(long currentSequence)

Check if we should attain safe point. If so, go there and then wait till signalled before we proceeding.

append

void append(FSWALEntry entry)
     throws Exception

Append to the WAL. Does all CP and WAL listener calls.

Throws:

Exception

onStart

public void onStart()

Specified by:

onStart in interface com.lmax.disruptor.LifecycleAware

onShutdown

public void onShutdown()

Specified by:

onShutdown in interface com.lmax.disruptor.LifecycleAware