Skip to content

Commit

Permalink
HBASE-19372 Remove the Span object in SyncFuture as it is useless now
Browse files Browse the repository at this point in the history 
Signed-off-by: Michael Stack <stack@apache.org>
  • Loading branch information
@Apache9 @saintstack
Apache9 authored and saintstack committed Nov 29, 2017
1 parent 8b32d37 commit f81ac42
Show file tree
Hide file tree
Showing 5 changed files with 47 additions and 93 deletions.
34 changes: 17 additions & 17 deletions hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AbstractFSWAL.java
View file
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand All @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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);
}

Expand Down Expand Up @@ -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);
Expand All @@ -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();
Expand Down Expand Up @@ -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();
Expand All @@ -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<RingBufferTruck> ringBuffer)
protected final long stampSequenceIdAndPublishToRingBuffer(RegionInfo hri, WALKey key,
WALEdit edits, boolean inMemstore, RingBuffer<RingBufferTruck> 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(() -> {
Expand All @@ -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 {
Expand Down
32 changes: 10 additions & 22 deletions hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AsyncFSWAL.java
View file
Expand Up @@ -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);
}
}

Expand All @@ -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);
}
}

Expand Down
42 changes: 16 additions & 26 deletions hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/FSHLog.java
View file
Expand Up @@ -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;
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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);
Expand All @@ -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);
}

/**
Expand Down
28 changes: 2 additions & 26 deletions hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/SyncFuture.java
View file
Expand Up @@ -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
Expand Down Expand Up @@ -68,19 +67,14 @@ 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
* @param span current span, detached from caller. Don't forget to attach it when resuming after a
* 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();
}
Expand All @@ -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;
}
Expand All @@ -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).
Expand Down
4 changes: 2 additions & 2 deletions hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestSyncFuture.java
View file
Expand Up @@ -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);
}
}

0 comments on commit f81ac42

Please sign in to comment.