diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AbstractFSWAL.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AbstractFSWAL.java index f7fbd8680886..64f44cd69e57 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AbstractFSWAL.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AbstractFSWAL.java @@ -21,6 +21,8 @@ import static org.apache.hadoop.hbase.shaded.com.google.common.base.Preconditions.checkNotNull; import static org.apache.hadoop.hbase.wal.AbstractFSWALProvider.WAL_FILE_NAME_DELIMITER; +import com.lmax.disruptor.RingBuffer; + import java.io.IOException; import java.io.InterruptedIOException; import java.lang.management.MemoryType; @@ -59,7 +61,6 @@ import org.apache.hadoop.hbase.exceptions.TimeoutIOException; import org.apache.hadoop.hbase.io.util.MemorySizeUtil; import org.apache.hadoop.hbase.regionserver.MultiVersionConcurrencyControl; -import org.apache.hadoop.hbase.shaded.com.google.common.annotations.VisibleForTesting; import org.apache.hadoop.hbase.trace.TraceUtil; import org.apache.hadoop.hbase.util.Bytes; import org.apache.hadoop.hbase.util.CollectionUtils; @@ -74,11 +75,10 @@ import org.apache.hadoop.hbase.wal.WALProvider.WriterBase; import org.apache.hadoop.hdfs.protocol.DatanodeInfo; import org.apache.hadoop.util.StringUtils; -import org.apache.htrace.core.Span; import org.apache.htrace.core.TraceScope; import org.apache.yetus.audience.InterfaceAudience; -import com.lmax.disruptor.RingBuffer; +import org.apache.hadoop.hbase.shaded.com.google.common.annotations.VisibleForTesting; /** * Implementation of {@link WAL} to go against {@link FileSystem}; i.e. keep WALs in HDFS. Only one @@ -696,13 +696,12 @@ Path replaceWriter(Path oldPath, Path newPath, W nextWriter) throws IOException } } - protected Span blockOnSync(final SyncFuture syncFuture) throws IOException { + protected final void blockOnSync(SyncFuture syncFuture) throws IOException { // Now we have published the ringbuffer, halt the current thread until we get an answer back. try { if (syncFuture != null) { syncFuture.get(walSyncTimeoutNs); } - return (syncFuture == null) ? null : syncFuture.getSpan(); } catch (TimeoutIOException tioe) { // SyncFuture reuse by thread, if TimeoutIOException happens, ringbuffer // still refer to it, so if this thread use it next time may get a wrong @@ -792,7 +791,8 @@ public void requestLogRoll() { * Get the backing files associated with this WAL. * @return may be null if there are no files. */ - protected FileStatus[] getFiles() throws IOException { + @VisibleForTesting + FileStatus[] getFiles() throws IOException { return CommonFSUtils.listStatus(fs, walDir, ourFiles); } @@ -862,13 +862,13 @@ public void updateStore(byte[] encodedRegionName, byte[] familyName, Long sequen sequenceIdAccounting.updateStore(encodedRegionName, familyName, sequenceid, onlyIfGreater); } - protected SyncFuture getSyncFuture(long sequence, Span span) { + protected final SyncFuture getSyncFuture(long sequence) { return CollectionUtils .computeIfAbsent(syncFuturesByHandler, Thread.currentThread(), SyncFuture::new) - .reset(sequence, span); + .reset(sequence); } - protected void requestLogRoll(boolean tooFewReplicas) { + protected final void requestLogRoll(boolean tooFewReplicas) { if (!this.listeners.isEmpty()) { for (WALActionsListener i : this.listeners) { i.logRollRequested(tooFewReplicas); @@ -894,7 +894,7 @@ void atHeadOfRingBufferEventHandlerAppend() { // Noop } - protected boolean append(W writer, FSWALEntry entry) throws IOException { + protected final boolean append(W writer, FSWALEntry entry) throws IOException { // TODO: WORK ON MAKING THIS APPEND FASTER. DOING WAY TOO MUCH WORK WITH CPs, PBing, etc. atHeadOfRingBufferEventHandlerAppend(); long start = EnvironmentEdgeManager.currentTime(); @@ -940,7 +940,7 @@ private long postAppend(final Entry e, final long elapsedTime) throws IOExceptio return len; } - protected void postSync(final long timeInNanos, final int handlerSyncs) { + protected final void postSync(final long timeInNanos, final int handlerSyncs) { if (timeInNanos > this.slowSyncNs) { String msg = new StringBuilder().append("Slow sync cost: ").append(timeInNanos / 1000000) .append(" ms, current pipeline: ").append(Arrays.toString(getPipeline())).toString(); @@ -954,11 +954,12 @@ protected void postSync(final long timeInNanos, final int handlerSyncs) { } } - protected long stampSequenceIdAndPublishToRingBuffer(RegionInfo hri, WALKey key, WALEdit edits, - boolean inMemstore, RingBuffer ringBuffer) + protected final long stampSequenceIdAndPublishToRingBuffer(RegionInfo hri, WALKey key, + WALEdit edits, boolean inMemstore, RingBuffer ringBuffer) throws IOException { if (this.closed) { - throw new IOException("Cannot append; log is closed, regionName = " + hri.getRegionNameAsString()); + throw new IOException( + "Cannot append; log is closed, regionName = " + hri.getRegionNameAsString()); } MutableLong txidHolder = new MutableLong(); MultiVersionConcurrencyControl.WriteEntry we = key.getMvcc().begin(() -> { @@ -968,10 +969,9 @@ protected long stampSequenceIdAndPublishToRingBuffer(RegionInfo hri, WALKey key, try (TraceScope scope = TraceUtil.createTrace(implClassName + ".append")) { FSWALEntry entry = new FSWALEntry(txid, key, edits, hri, inMemstore); entry.stampRegionSequenceId(we); - if(scope!=null){ + if (scope != null) { ringBuffer.get(txid).load(entry, scope.getSpan()); - } - else{ + } else { ringBuffer.get(txid).load(entry, null); } } finally { diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AsyncFSWAL.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AsyncFSWAL.java index 9baf80387fd2..9aad2bc17ed6 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AsyncFSWAL.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AsyncFSWAL.java @@ -565,24 +565,18 @@ public long append(RegionInfo hri, WALKey key, WALEdit edits, boolean inMemstore public void sync() throws IOException { try (TraceScope scope = TraceUtil.createTrace("AsyncFSWAL.sync")){ long txid = waitingConsumePayloads.next(); - SyncFuture future = null; + SyncFuture future; try { - if (scope != null) { - future = getSyncFuture(txid, scope.getSpan()); - RingBufferTruck truck = waitingConsumePayloads.get(txid); - truck.load(future); - } + future = getSyncFuture(txid); + RingBufferTruck truck = waitingConsumePayloads.get(txid); + truck.load(future); } finally { waitingConsumePayloads.publish(txid); } if (shouldScheduleConsumer()) { eventLoop.execute(consumer); } - //TODO handle htrace API change, see HBASE-18895 - //scope = Trace.continueSpan(blockOnSync(future)); - if (future != null) { - blockOnSync(future); - } + blockOnSync(future); } } @@ -594,24 +588,18 @@ public void sync(long txid) throws IOException { try (TraceScope scope = TraceUtil.createTrace("AsyncFSWAL.sync")) { // here we do not use ring buffer sequence as txid long sequence = waitingConsumePayloads.next(); - SyncFuture future = null; + SyncFuture future; try { - if(scope!= null) { - future = getSyncFuture(txid, scope.getSpan()); - RingBufferTruck truck = waitingConsumePayloads.get(sequence); - truck.load(future); - } + future = getSyncFuture(txid); + RingBufferTruck truck = waitingConsumePayloads.get(sequence); + truck.load(future); } finally { waitingConsumePayloads.publish(sequence); } if (shouldScheduleConsumer()) { eventLoop.execute(consumer); } - //TODO handle htrace API change, see HBASE-18895 - //scope = Trace.continueSpan(blockOnSync(future)); - if (future != null) { - blockOnSync(future); - } + blockOnSync(future); } } diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/FSHLog.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/FSHLog.java index c4e23dac8022..15a6a4181f30 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/FSHLog.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/FSHLog.java @@ -17,6 +17,14 @@ */ package org.apache.hadoop.hbase.regionserver.wal; +import com.lmax.disruptor.BlockingWaitStrategy; +import com.lmax.disruptor.EventHandler; +import com.lmax.disruptor.ExceptionHandler; +import com.lmax.disruptor.LifecycleAware; +import com.lmax.disruptor.TimeoutException; +import com.lmax.disruptor.dsl.Disruptor; +import com.lmax.disruptor.dsl.ProducerType; + import java.io.FileNotFoundException; import java.io.IOException; import java.io.OutputStream; @@ -55,18 +63,10 @@ import org.apache.hadoop.hdfs.DFSOutputStream; import org.apache.hadoop.hdfs.client.HdfsDataOutputStream; import org.apache.hadoop.hdfs.protocol.DatanodeInfo; -import org.apache.htrace.core.Span; import org.apache.htrace.core.TraceScope; import org.apache.yetus.audience.InterfaceAudience; -import org.apache.hadoop.hbase.shaded.com.google.common.annotations.VisibleForTesting; -import com.lmax.disruptor.BlockingWaitStrategy; -import com.lmax.disruptor.EventHandler; -import com.lmax.disruptor.ExceptionHandler; -import com.lmax.disruptor.LifecycleAware; -import com.lmax.disruptor.TimeoutException; -import com.lmax.disruptor.dsl.Disruptor; -import com.lmax.disruptor.dsl.ProducerType; +import org.apache.hadoop.hbase.shaded.com.google.common.annotations.VisibleForTesting; /** * The default implementation of FSWAL. @@ -702,22 +702,18 @@ private boolean checkLowReplication() { return logRollNeeded; } - private SyncFuture publishSyncOnRingBuffer(long sequence) { - return publishSyncOnRingBuffer(sequence, null); - } - private long getSequenceOnRingBuffer() { return this.disruptor.getRingBuffer().next(); } - private SyncFuture publishSyncOnRingBuffer(Span span) { - long sequence = this.disruptor.getRingBuffer().next(); - return publishSyncOnRingBuffer(sequence, span); + private SyncFuture publishSyncOnRingBuffer() { + long sequence = getSequenceOnRingBuffer(); + return publishSyncOnRingBuffer(sequence); } - private SyncFuture publishSyncOnRingBuffer(long sequence, Span span) { + private SyncFuture publishSyncOnRingBuffer(long sequence) { // here we use ring buffer sequence as transaction id - SyncFuture syncFuture = getSyncFuture(sequence, span); + SyncFuture syncFuture = getSyncFuture(sequence); try { RingBufferTruck truck = this.disruptor.getRingBuffer().get(sequence); truck.load(syncFuture); @@ -729,14 +725,8 @@ private SyncFuture publishSyncOnRingBuffer(long sequence, Span span) { // Sync all known transactions private void publishSyncThenBlockOnCompletion(TraceScope scope) throws IOException { - if (scope != null) { - SyncFuture syncFuture = publishSyncOnRingBuffer(scope.getSpan()); - blockOnSync(syncFuture); - } - else { - SyncFuture syncFuture = publishSyncOnRingBuffer(null); - blockOnSync(syncFuture); - } + SyncFuture syncFuture = publishSyncOnRingBuffer(); + blockOnSync(syncFuture); } /** diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/SyncFuture.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/SyncFuture.java index 0dbd0208e304..bd30797f557f 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/SyncFuture.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/SyncFuture.java @@ -20,9 +20,8 @@ import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; -import org.apache.yetus.audience.InterfaceAudience; import org.apache.hadoop.hbase.exceptions.TimeoutIOException; -import org.apache.htrace.core.Span; +import org.apache.yetus.audience.InterfaceAudience; /** * A Future on a filesystem sync call. It given to a client or 'Handler' for it to wait on till the @@ -68,11 +67,6 @@ class SyncFuture { private Thread t; - /** - * Optionally carry a disconnected scope to the SyncRunner. - */ - private Span span; - /** * Call this method to clear old usage and get it ready for new deploy. * @param txid the new transaction id @@ -80,7 +74,7 @@ class SyncFuture { * call to {@link #get(long)}. * @return this */ - synchronized SyncFuture reset(final long txid, Span span) { + synchronized SyncFuture reset(long txid) { if (t != null && t != Thread.currentThread()) { throw new IllegalStateException(); } @@ -90,7 +84,6 @@ synchronized SyncFuture reset(final long txid, Span span) { } this.doneTxid = NOT_DONE; this.txid = txid; - this.span = span; this.throwable = null; return this; } @@ -104,23 +97,6 @@ synchronized long getTxid() { return this.txid; } - /** - * Retrieve the {@code span} instance from this Future. EventHandler calls this method to continue - * the span. Thread waiting on this Future musn't call this method until AFTER calling - * {@link #get(long)} and the future has been released back to the originating thread. - */ - synchronized Span getSpan() { - return this.span; - } - - /** - * Used to re-attach a {@code span} to the Future. Called by the EventHandler after a it has - * completed processing and detached the span from its scope. - */ - synchronized void setSpan(Span span) { - this.span = span; - } - /** * @param txid the transaction id at which this future 'completed'. * @param t Can be null. Set if we are 'completing' on error (and this 't' is the error). diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestSyncFuture.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestSyncFuture.java index 50825f818fbb..809263659def 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestSyncFuture.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestSyncFuture.java @@ -32,10 +32,10 @@ public class TestSyncFuture { public void testGet() throws Exception { long timeout = 5000; long txid = 100000; - SyncFuture syncFulture = new SyncFuture().reset(txid, null); + SyncFuture syncFulture = new SyncFuture().reset(txid); syncFulture.done(txid, null); assertEquals(txid, syncFulture.get(timeout)); - syncFulture.reset(txid, null).get(timeout); + syncFulture.reset(txid).get(timeout); } }