You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by sz...@apache.org on 2016/06/06 08:50:59 UTC
[4/8] hadoop git commit: Revert "Revert "HADOOP-13168. Support
Future.get with timeout in ipc async calls.""
Revert "Revert "HADOOP-13168. Support Future.get with timeout in ipc async calls.""
This reverts commit e4450d47f19131818e1c040b6bd8d85ae8250475.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/574dcd34
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/574dcd34
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/574dcd34
Branch: refs/heads/trunk
Commit: 574dcd34c0da1903d25e37dc5757642a584dc3d0
Parents: cba9a01
Author: Tsz-Wo Nicholas Sze <sz...@hortonworks.com>
Authored: Mon Jun 6 16:31:23 2016 +0800
Committer: Tsz-Wo Nicholas Sze <sz...@hortonworks.com>
Committed: Mon Jun 6 16:31:23 2016 +0800
----------------------------------------------------------------------
.../main/java/org/apache/hadoop/ipc/Client.java | 119 ++++++++----------
.../apache/hadoop/ipc/ProtobufRpcEngine.java | 62 +++++-----
.../apache/hadoop/util/concurrent/AsyncGet.java | 60 +++++++++
.../hadoop/util/concurrent/AsyncGetFuture.java | 73 +++++++++++
.../org/apache/hadoop/ipc/TestAsyncIPC.java | 124 +++++++++++--------
.../hadoop/hdfs/AsyncDistributedFileSystem.java | 24 +---
.../ClientNamenodeProtocolTranslatorPB.java | 33 ++---
7 files changed, 310 insertions(+), 185 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/574dcd34/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Client.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Client.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Client.java
index 9be4649..d1d5b17 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Client.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Client.java
@@ -18,46 +18,10 @@
package org.apache.hadoop.ipc;
-import static org.apache.hadoop.ipc.RpcConstants.*;
-
-import java.io.BufferedInputStream;
-import java.io.BufferedOutputStream;
-import java.io.ByteArrayOutputStream;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.EOFException;
-import java.io.FilterInputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.InterruptedIOException;
-import java.io.OutputStream;
-import java.net.InetAddress;
-import java.net.InetSocketAddress;
-import java.net.Socket;
-import java.net.SocketTimeoutException;
-import java.net.UnknownHostException;
-import java.security.PrivilegedExceptionAction;
-import java.util.Arrays;
-import java.util.Hashtable;
-import java.util.Iterator;
-import java.util.Map.Entry;
-import java.util.Random;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-import java.util.concurrent.RejectedExecutionException;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
-
-import javax.net.SocketFactory;
-import javax.security.sasl.Sasl;
-
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import com.google.protobuf.CodedOutputStream;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience;
@@ -93,14 +57,25 @@ import org.apache.hadoop.util.ProtoUtil;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.util.Time;
+import org.apache.hadoop.util.concurrent.AsyncGet;
+import org.apache.hadoop.util.concurrent.AsyncGetFuture;
import org.apache.htrace.core.Span;
import org.apache.htrace.core.Tracer;
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Preconditions;
-import com.google.common.util.concurrent.AbstractFuture;
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
-import com.google.protobuf.CodedOutputStream;
+import javax.net.SocketFactory;
+import javax.security.sasl.Sasl;
+import java.io.*;
+import java.net.*;
+import java.security.PrivilegedExceptionAction;
+import java.util.*;
+import java.util.Map.Entry;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.apache.hadoop.ipc.RpcConstants.CONNECTION_CONTEXT_CALL_ID;
+import static org.apache.hadoop.ipc.RpcConstants.PING_CALL_ID;
/** A client for an IPC service. IPC calls take a single {@link Writable} as a
* parameter, and return a {@link Writable} as their value. A service runs on
@@ -119,8 +94,8 @@ public class Client implements AutoCloseable {
private static final ThreadLocal<Integer> callId = new ThreadLocal<Integer>();
private static final ThreadLocal<Integer> retryCount = new ThreadLocal<Integer>();
- private static final ThreadLocal<Future<?>>
- RETURN_RPC_RESPONSE = new ThreadLocal<>();
+ private static final ThreadLocal<Future<?>> ASYNC_RPC_RESPONSE
+ = new ThreadLocal<>();
private static final ThreadLocal<Boolean> asynchronousMode =
new ThreadLocal<Boolean>() {
@Override
@@ -131,8 +106,8 @@ public class Client implements AutoCloseable {
@SuppressWarnings("unchecked")
@Unstable
- public static <T> Future<T> getReturnRpcResponse() {
- return (Future<T>) RETURN_RPC_RESPONSE.get();
+ public static <T> Future<T> getAsyncRpcResponse() {
+ return (Future<T>) ASYNC_RPC_RESPONSE.get();
}
/** Set call id and retry count for the next call. */
@@ -379,6 +354,11 @@ public class Client implements AutoCloseable {
}
}
+ @Override
+ public String toString() {
+ return getClass().getSimpleName() + id;
+ }
+
/** Indicate when the call is complete and the
* value or error are available. Notifies by default. */
protected synchronized void callComplete() {
@@ -1413,27 +1393,32 @@ public class Client implements AutoCloseable {
}
if (isAsynchronousMode()) {
- Future<Writable> returnFuture = new AbstractFuture<Writable>() {
- private final AtomicBoolean callled = new AtomicBoolean(false);
+ final AsyncGet<Writable, IOException> asyncGet
+ = new AsyncGet<Writable, IOException>() {
@Override
- public Writable get() throws InterruptedException, ExecutionException {
- if (callled.compareAndSet(false, true)) {
- try {
- set(getRpcResponse(call, connection));
- } catch (IOException ie) {
- setException(ie);
- } finally {
+ public Writable get(long timeout, TimeUnit unit)
+ throws IOException, TimeoutException{
+ boolean done = true;
+ try {
+ final Writable w = getRpcResponse(call, connection, timeout, unit);
+ if (w == null) {
+ done = false;
+ throw new TimeoutException(call + " timed out "
+ + timeout + " " + unit);
+ }
+ return w;
+ } finally {
+ if (done) {
releaseAsyncCall();
}
}
- return super.get();
}
};
- RETURN_RPC_RESPONSE.set(returnFuture);
+ ASYNC_RPC_RESPONSE.set(new AsyncGetFuture<>(asyncGet));
return null;
} else {
- return getRpcResponse(call, connection);
+ return getRpcResponse(call, connection, -1, null);
}
}
@@ -1469,12 +1454,18 @@ public class Client implements AutoCloseable {
return asyncCallCounter.get();
}
- private Writable getRpcResponse(final Call call, final Connection connection)
- throws IOException {
+ /** @return the rpc response or, in case of timeout, null. */
+ private Writable getRpcResponse(final Call call, final Connection connection,
+ final long timeout, final TimeUnit unit) throws IOException {
synchronized (call) {
while (!call.done) {
try {
- call.wait(); // wait for the result
+ final long waitTimeout = AsyncGet.Util.asyncGetTimeout2WaitTimeout(
+ timeout, unit);
+ call.wait(waitTimeout); // wait for the result
+ if (waitTimeout > 0 && !call.done) {
+ return null;
+ }
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
throw new InterruptedIOException("Call interrupted");
http://git-wip-us.apache.org/repos/asf/hadoop/blob/574dcd34/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtobufRpcEngine.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtobufRpcEngine.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtobufRpcEngine.java
index 8fcdb78..0f43fc6 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtobufRpcEngine.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtobufRpcEngine.java
@@ -18,21 +18,9 @@
package org.apache.hadoop.ipc;
-import java.io.DataInput;
-import java.io.DataOutput;
-import java.io.IOException;
-import java.io.OutputStream;
-import java.lang.reflect.Method;
-import java.lang.reflect.Proxy;
-import java.net.InetSocketAddress;
-import java.util.Map;
-import java.util.concurrent.Callable;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Future;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import javax.net.SocketFactory;
-
+import com.google.common.annotations.VisibleForTesting;
+import com.google.protobuf.*;
+import com.google.protobuf.Descriptors.MethodDescriptor;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience;
@@ -52,17 +40,23 @@ import org.apache.hadoop.security.token.SecretManager;
import org.apache.hadoop.security.token.TokenIdentifier;
import org.apache.hadoop.util.ProtoUtil;
import org.apache.hadoop.util.Time;
+import org.apache.hadoop.util.concurrent.AsyncGet;
import org.apache.htrace.core.TraceScope;
import org.apache.htrace.core.Tracer;
-import com.google.common.annotations.VisibleForTesting;
-import com.google.protobuf.BlockingService;
-import com.google.protobuf.CodedOutputStream;
-import com.google.protobuf.Descriptors.MethodDescriptor;
-import com.google.protobuf.GeneratedMessage;
-import com.google.protobuf.Message;
-import com.google.protobuf.ServiceException;
-import com.google.protobuf.TextFormat;
+import javax.net.SocketFactory;
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.lang.reflect.Method;
+import java.lang.reflect.Proxy;
+import java.net.InetSocketAddress;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
/**
* RPC Engine for for protobuf based RPCs.
@@ -70,8 +64,8 @@ import com.google.protobuf.TextFormat;
@InterfaceStability.Evolving
public class ProtobufRpcEngine implements RpcEngine {
public static final Log LOG = LogFactory.getLog(ProtobufRpcEngine.class);
- private static final ThreadLocal<Callable<?>>
- RETURN_MESSAGE_CALLBACK = new ThreadLocal<>();
+ private static final ThreadLocal<AsyncGet<Message, Exception>>
+ ASYNC_RETURN_MESSAGE = new ThreadLocal<>();
static { // Register the rpcRequest deserializer for WritableRpcEngine
org.apache.hadoop.ipc.Server.registerProtocolEngine(
@@ -81,10 +75,9 @@ public class ProtobufRpcEngine implements RpcEngine {
private static final ClientCache CLIENTS = new ClientCache();
- @SuppressWarnings("unchecked")
@Unstable
- public static <T> Callable<T> getReturnMessageCallback() {
- return (Callable<T>) RETURN_MESSAGE_CALLBACK.get();
+ public static AsyncGet<Message, Exception> getAsyncReturnMessage() {
+ return ASYNC_RETURN_MESSAGE.get();
}
public <T> ProtocolProxy<T> getProxy(Class<T> protocol, long clientVersion,
@@ -263,14 +256,17 @@ public class ProtobufRpcEngine implements RpcEngine {
}
if (Client.isAsynchronousMode()) {
- final Future<RpcResponseWrapper> frrw = Client.getReturnRpcResponse();
- Callable<Message> callback = new Callable<Message>() {
+ final Future<RpcResponseWrapper> frrw = Client.getAsyncRpcResponse();
+ final AsyncGet<Message, Exception> asyncGet
+ = new AsyncGet<Message, Exception>() {
@Override
- public Message call() throws Exception {
- return getReturnMessage(method, frrw.get());
+ public Message get(long timeout, TimeUnit unit) throws Exception {
+ final RpcResponseWrapper rrw = timeout < 0?
+ frrw.get(): frrw.get(timeout, unit);
+ return getReturnMessage(method, rrw);
}
};
- RETURN_MESSAGE_CALLBACK.set(callback);
+ ASYNC_RETURN_MESSAGE.set(asyncGet);
return null;
} else {
return getReturnMessage(method, val);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/574dcd34/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/concurrent/AsyncGet.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/concurrent/AsyncGet.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/concurrent/AsyncGet.java
new file mode 100644
index 0000000..5eac869
--- /dev/null
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/concurrent/AsyncGet.java
@@ -0,0 +1,60 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.util.concurrent;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+/**
+ * This interface defines an asynchronous {@link #get(long, TimeUnit)} method.
+ *
+ * When the return value is still being computed, invoking
+ * {@link #get(long, TimeUnit)} will result in a {@link TimeoutException}.
+ * The method should be invoked again and again
+ * until the underlying computation is completed.
+ *
+ * @param <R> The type of the return value.
+ * @param <E> The exception type that the underlying implementation may throw.
+ */
+public interface AsyncGet<R, E extends Throwable> {
+ /**
+ * Get the result.
+ *
+ * @param timeout The maximum time period to wait.
+ * When timeout == 0, it does not wait at all.
+ * When timeout < 0, it waits indefinitely.
+ * @param unit The unit of the timeout value
+ * @return the result, which is possibly null.
+ * @throws E an exception thrown by the underlying implementation.
+ * @throws TimeoutException if it cannot return after the given time period.
+ * @throws InterruptedException if the thread is interrupted.
+ */
+ R get(long timeout, TimeUnit unit)
+ throws E, TimeoutException, InterruptedException;
+
+ /** Utility */
+ class Util {
+ /**
+ * @return {@link Object#wait(long)} timeout converted
+ * from {@link #get(long, TimeUnit)} timeout.
+ */
+ public static long asyncGetTimeout2WaitTimeout(long timeout, TimeUnit unit){
+ return timeout < 0? 0: timeout == 0? 1:unit.toMillis(timeout);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/574dcd34/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/concurrent/AsyncGetFuture.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/concurrent/AsyncGetFuture.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/concurrent/AsyncGetFuture.java
new file mode 100644
index 0000000..d687867
--- /dev/null
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/concurrent/AsyncGetFuture.java
@@ -0,0 +1,73 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.util.concurrent;
+
+import com.google.common.util.concurrent.AbstractFuture;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/** A {@link Future} implemented using an {@link AsyncGet} object. */
+public class AsyncGetFuture<T, E extends Throwable> extends AbstractFuture<T> {
+ public static final Log LOG = LogFactory.getLog(AsyncGetFuture.class);
+
+ private final AtomicBoolean called = new AtomicBoolean(false);
+ private final AsyncGet<T, E> asyncGet;
+
+ public AsyncGetFuture(AsyncGet<T, E> asyncGet) {
+ this.asyncGet = asyncGet;
+ }
+
+ private void callAsyncGet(long timeout, TimeUnit unit) {
+ if (!isCancelled() && called.compareAndSet(false, true)) {
+ try {
+ set(asyncGet.get(timeout, unit));
+ } catch (TimeoutException te) {
+ LOG.trace("TRACE", te);
+ called.compareAndSet(true, false);
+ } catch (Throwable e) {
+ LOG.trace("TRACE", e);
+ setException(e);
+ }
+ }
+ }
+
+ @Override
+ public T get() throws InterruptedException, ExecutionException {
+ callAsyncGet(-1, TimeUnit.MILLISECONDS);
+ return super.get();
+ }
+
+ @Override
+ public T get(long timeout, TimeUnit unit)
+ throws InterruptedException, TimeoutException, ExecutionException {
+ callAsyncGet(timeout, unit);
+ return super.get(0, TimeUnit.MILLISECONDS);
+ }
+
+ @Override
+ public boolean isDone() {
+ callAsyncGet(0, TimeUnit.MILLISECONDS);
+ return super.isDone();
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/574dcd34/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestAsyncIPC.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestAsyncIPC.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestAsyncIPC.java
index 8ee3a2c..0ad191b 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestAsyncIPC.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestAsyncIPC.java
@@ -18,20 +18,6 @@
package org.apache.hadoop.ipc;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
-
-import java.io.IOException;
-import java.net.InetSocketAddress;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Future;
-
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
@@ -48,6 +34,17 @@ import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.*;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+
public class TestAsyncIPC {
private static Configuration conf;
@@ -87,26 +84,50 @@ public class TestAsyncIPC {
try {
final long param = TestIPC.RANDOM.nextLong();
TestIPC.call(client, param, server, conf);
- Future<LongWritable> returnFuture = Client.getReturnRpcResponse();
- returnFutures.put(i, returnFuture);
+ returnFutures.put(i, Client.getAsyncRpcResponse());
expectedValues.put(i, param);
} catch (Exception e) {
- LOG.fatal("Caught: " + StringUtils.stringifyException(e));
failed = true;
+ throw new RuntimeException(e);
}
}
}
- public void waitForReturnValues() throws InterruptedException,
- ExecutionException {
+ void assertReturnValues() throws InterruptedException, ExecutionException {
for (int i = 0; i < count; i++) {
LongWritable value = returnFutures.get(i).get();
- if (expectedValues.get(i) != value.get()) {
- LOG.fatal(String.format("Call-%d failed!", i));
- failed = true;
- break;
+ Assert.assertEquals("call" + i + " failed.",
+ expectedValues.get(i).longValue(), value.get());
+ }
+ Assert.assertFalse(failed);
+ }
+
+ void assertReturnValues(long timeout, TimeUnit unit)
+ throws InterruptedException, ExecutionException {
+ final boolean[] checked = new boolean[count];
+ for(boolean done = false; !done;) {
+ done = true;
+ for (int i = 0; i < count; i++) {
+ if (checked[i]) {
+ continue;
+ } else {
+ done = false;
+ }
+
+ final LongWritable value;
+ try {
+ value = returnFutures.get(i).get(timeout, unit);
+ } catch (TimeoutException e) {
+ LOG.info("call" + i + " caught ", e);
+ continue;
+ }
+
+ Assert.assertEquals("call" + i + " failed.",
+ expectedValues.get(i).longValue(), value.get());
+ checked[i] = true;
}
}
+ Assert.assertFalse(failed);
}
}
@@ -183,8 +204,7 @@ public class TestAsyncIPC {
private void doCall(final int idx, final long param) throws IOException {
TestIPC.call(client, param, server, conf);
- Future<LongWritable> returnFuture = Client.getReturnRpcResponse();
- returnFutures.put(idx, returnFuture);
+ returnFutures.put(idx, Client.getAsyncRpcResponse());
expectedValues.put(idx, param);
}
@@ -233,10 +253,7 @@ public class TestAsyncIPC {
}
for (int i = 0; i < callerCount; i++) {
callers[i].join();
- callers[i].waitForReturnValues();
- String msg = String.format("Expected not failed for caller-%d: %s.", i,
- callers[i]);
- assertFalse(msg, callers[i].failed);
+ callers[i].assertReturnValues();
}
for (int i = 0; i < clientCount; i++) {
clients[i].stop();
@@ -258,25 +275,37 @@ public class TestAsyncIPC {
try {
AsyncCaller caller = new AsyncCaller(client, addr, callCount);
caller.run();
+ caller.assertReturnValues();
+ caller.assertReturnValues();
+ caller.assertReturnValues();
+ Assert.assertEquals(asyncCallCount, client.getAsyncCallCount());
+ } finally {
+ client.stop();
+ server.stop();
+ }
+ }
- caller.waitForReturnValues();
- String msg = String.format(
- "First time, expected not failed for caller: %s.", caller);
- assertFalse(msg, caller.failed);
+ @Test(timeout = 60000)
+ public void testFutureGetWithTimeout() throws IOException,
+ InterruptedException, ExecutionException {
+// GenericTestUtils.setLogLevel(AsyncGetFuture.LOG, Level.ALL);
+ final Server server = new TestIPC.TestServer(10, true, conf);
+ final InetSocketAddress addr = NetUtils.getConnectAddress(server);
+ server.start();
- caller.waitForReturnValues();
- assertTrue(asyncCallCount == client.getAsyncCallCount());
- msg = String.format("Second time, expected not failed for caller: %s.",
- caller);
- assertFalse(msg, caller.failed);
+ final Client client = new Client(LongWritable.class, conf);
- assertTrue(asyncCallCount == client.getAsyncCallCount());
+ try {
+ final AsyncCaller caller = new AsyncCaller(client, addr, 10);
+ caller.run();
+ caller.assertReturnValues(10, TimeUnit.MILLISECONDS);
} finally {
client.stop();
server.stop();
}
}
+
public void internalTestAsyncCallLimit(int handlerCount, boolean handlerSleep,
int clientCount, int callerCount, int callCount) throws IOException,
InterruptedException, ExecutionException {
@@ -367,9 +396,7 @@ public class TestAsyncIPC {
server.start();
final AsyncCaller caller = new AsyncCaller(client, addr, 4);
caller.run();
- caller.waitForReturnValues();
- String msg = String.format("Expected not failed for caller: %s.", caller);
- assertFalse(msg, caller.failed);
+ caller.assertReturnValues();
} finally {
client.stop();
server.stop();
@@ -406,9 +433,7 @@ public class TestAsyncIPC {
server.start();
final AsyncCaller caller = new AsyncCaller(client, addr, 10);
caller.run();
- caller.waitForReturnValues();
- String msg = String.format("Expected not failed for caller: %s.", caller);
- assertFalse(msg, caller.failed);
+ caller.assertReturnValues();
} finally {
client.stop();
server.stop();
@@ -443,9 +468,7 @@ public class TestAsyncIPC {
server.start();
final AsyncCaller caller = new AsyncCaller(client, addr, 10);
caller.run();
- caller.waitForReturnValues();
- String msg = String.format("Expected not failed for caller: %s.", caller);
- assertFalse(msg, caller.failed);
+ caller.assertReturnValues();
} finally {
client.stop();
server.stop();
@@ -489,10 +512,7 @@ public class TestAsyncIPC {
}
for (int i = 0; i < callerCount; ++i) {
callers[i].join();
- callers[i].waitForReturnValues();
- String msg = String.format("Expected not failed for caller-%d: %s.", i,
- callers[i]);
- assertFalse(msg, callers[i].failed);
+ callers[i].assertReturnValues();
}
} finally {
client.stop();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/574dcd34/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/AsyncDistributedFileSystem.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/AsyncDistributedFileSystem.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/AsyncDistributedFileSystem.java
index 4fe0861..6bfd71d 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/AsyncDistributedFileSystem.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/AsyncDistributedFileSystem.java
@@ -19,20 +19,16 @@
package org.apache.hadoop.hdfs;
import java.io.IOException;
-import java.util.concurrent.Callable;
-import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
-import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.hadoop.classification.InterfaceStability.Unstable;
import org.apache.hadoop.fs.Options;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolTranslatorPB;
+import org.apache.hadoop.util.concurrent.AsyncGetFuture;
import org.apache.hadoop.ipc.Client;
-import com.google.common.util.concurrent.AbstractFuture;
-
/****************************************************************
* Implementation of the asynchronous distributed file system.
* This instance of this class is the way end-user code interacts
@@ -52,22 +48,8 @@ public class AsyncDistributedFileSystem {
}
static <T> Future<T> getReturnValue() {
- final Callable<T> returnValueCallback = ClientNamenodeProtocolTranslatorPB
- .getReturnValueCallback();
- Future<T> returnFuture = new AbstractFuture<T>() {
- private final AtomicBoolean called = new AtomicBoolean(false);
- public T get() throws InterruptedException, ExecutionException {
- if (called.compareAndSet(false, true)) {
- try {
- set(returnValueCallback.call());
- } catch (Exception e) {
- setException(e);
- }
- }
- return super.get();
- }
- };
- return returnFuture;
+ return new AsyncGetFuture<>(
+ ClientNamenodeProtocolTranslatorPB.getAsyncReturnValue());
}
/**
http://git-wip-us.apache.org/repos/asf/hadoop/blob/574dcd34/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolTranslatorPB.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolTranslatorPB.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolTranslatorPB.java
index faa925c..939c1ac 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolTranslatorPB.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolTranslatorPB.java
@@ -24,7 +24,8 @@ import java.util.EnumSet;
import java.util.List;
import com.google.common.collect.Lists;
-import java.util.concurrent.Callable;
+
+import java.util.concurrent.TimeUnit;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
@@ -198,6 +199,7 @@ import org.apache.hadoop.security.token.Token;
import com.google.protobuf.ByteString;
import com.google.protobuf.Message;
import com.google.protobuf.ServiceException;
+import org.apache.hadoop.util.concurrent.AsyncGet;
/**
* This class forwards NN's ClientProtocol calls as RPC calls to the NN server
@@ -209,8 +211,8 @@ import com.google.protobuf.ServiceException;
public class ClientNamenodeProtocolTranslatorPB implements
ProtocolMetaInterface, ClientProtocol, Closeable, ProtocolTranslator {
final private ClientNamenodeProtocolPB rpcProxy;
- private static final ThreadLocal<Callable<?>>
- RETURN_VALUE_CALLBACK = new ThreadLocal<>();
+ private static final ThreadLocal<AsyncGet<?, Exception>>
+ ASYNC_RETURN_VALUE = new ThreadLocal<>();
static final GetServerDefaultsRequestProto VOID_GET_SERVER_DEFAULT_REQUEST =
GetServerDefaultsRequestProto.newBuilder().build();
@@ -246,8 +248,8 @@ public class ClientNamenodeProtocolTranslatorPB implements
@SuppressWarnings("unchecked")
@Unstable
- public static <T> Callable<T> getReturnValueCallback() {
- return (Callable<T>) RETURN_VALUE_CALLBACK.get();
+ public static <T> AsyncGet<T, Exception> getAsyncReturnValue() {
+ return (AsyncGet<T, Exception>) ASYNC_RETURN_VALUE.get();
}
@Override
@@ -369,7 +371,7 @@ public class ClientNamenodeProtocolTranslatorPB implements
try {
if (Client.isAsynchronousMode()) {
rpcProxy.setPermission(null, req);
- setReturnValueCallback();
+ setAsyncReturnValue();
} else {
rpcProxy.setPermission(null, req);
}
@@ -378,17 +380,18 @@ public class ClientNamenodeProtocolTranslatorPB implements
}
}
- private void setReturnValueCallback() {
- final Callable<Message> returnMessageCallback = ProtobufRpcEngine
- .getReturnMessageCallback();
- Callable<Void> callBack = new Callable<Void>() {
+ private void setAsyncReturnValue() {
+ final AsyncGet<Message, Exception> asyncReturnMessage
+ = ProtobufRpcEngine.getAsyncReturnMessage();
+ final AsyncGet<Void, Exception> asyncGet
+ = new AsyncGet<Void, Exception>() {
@Override
- public Void call() throws Exception {
- returnMessageCallback.call();
+ public Void get(long timeout, TimeUnit unit) throws Exception {
+ asyncReturnMessage.get(timeout, unit);
return null;
}
};
- RETURN_VALUE_CALLBACK.set(callBack);
+ ASYNC_RETURN_VALUE.set(asyncGet);
}
@Override
@@ -403,7 +406,7 @@ public class ClientNamenodeProtocolTranslatorPB implements
try {
if (Client.isAsynchronousMode()) {
rpcProxy.setOwner(null, req.build());
- setReturnValueCallback();
+ setAsyncReturnValue();
} else {
rpcProxy.setOwner(null, req.build());
}
@@ -536,7 +539,7 @@ public class ClientNamenodeProtocolTranslatorPB implements
try {
if (Client.isAsynchronousMode()) {
rpcProxy.rename2(null, req);
- setReturnValueCallback();
+ setAsyncReturnValue();
} else {
rpcProxy.rename2(null, req);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org