Apache
/
hadoop
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HADOOP-12672. RPC timeout should not override IPC ping interval (iwasakims)

This commit is contained in:
Masatake Iwasaki 2016-03-11 15:12:22 +09:00
parent cff6db1f92
commit acafc950d9
4 changed files with 94 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
hadoop-common-project/hadoop-common/CHANGES.txt
View File

@ -902,6 +902,8 @@ Release 2.6.5 - UNRELEASED
HADOOP-12789. log classpath of ApplicationClassLoader at INFO level
(Sangjin Lee via mingma)
HADOOP-12672. RPC timeout should not override IPC ping interval (iwasakims)
OPTIMIZATIONS
BUG FIXES

35
hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Client.java
View File

@ -383,15 +383,16 @@ public class Client {
private Socket socket = null; // connected socket
private DataInputStream in;
private DataOutputStream out;
private int rpcTimeout;
private final int rpcTimeout;
private int maxIdleTime; //connections will be culled if it was idle for
//maxIdleTime msecs
private final RetryPolicy connectionRetryPolicy;
private final int maxRetriesOnSasl;
private int maxRetriesOnSocketTimeouts;
private boolean tcpNoDelay; // if T then disable Nagle's Algorithm
private boolean doPing; //do we need to send ping message
private int pingInterval; // how often sends ping to the server in msecs
private final boolean tcpNoDelay; // if T then disable Nagle's Algorithm
private final boolean doPing; //do we need to send ping message
private final int pingInterval; // how often sends ping to the server
private final int soTimeout; // used by ipc ping and rpc timeout
private ByteArrayOutputStream pingRequest; // ping message
// currently active calls
@ -429,6 +430,9 @@ public class Client {
pingHeader.writeDelimitedTo(pingRequest);
}
this.pingInterval = remoteId.getPingInterval();
this.soTimeout =
(rpcTimeout == 0 || (doPing && pingInterval < rpcTimeout))?
this.pingInterval : this.rpcTimeout;
this.serviceClass = serviceClass;
if (LOG.isDebugEnabled()) {
LOG.debug("The ping interval is " + this.pingInterval + " ms.");
@ -479,12 +483,12 @@ public class Client {
/* Process timeout exception
* if the connection is not going to be closed or
* is not configured to have a RPC timeout, send a ping.
* (if rpcTimeout is not set to be 0, then RPC should timeout.
* otherwise, throw the timeout exception.
* the RPC is not timed out yet, send a ping.
*/
private void handleTimeout(SocketTimeoutException e) throws IOException {
if (shouldCloseConnection.get() || !running.get() || rpcTimeout > 0) {
private void handleTimeout(SocketTimeoutException e, int waiting)
throws IOException {
if (shouldCloseConnection.get() || !running.get() ||
(0 < rpcTimeout && rpcTimeout <= waiting)) {
throw e;
} else {
sendPing();
@ -498,11 +502,13 @@ public class Client {
*/
@Override
public int read() throws IOException {
int waiting = 0;
do {
try {
return super.read();
} catch (SocketTimeoutException e) {
handleTimeout(e);
waiting += soTimeout;
handleTimeout(e, waiting);
}
} while (true);
}
@ -515,11 +521,13 @@ public class Client {
*/
@Override
public int read(byte[] buf, int off, int len) throws IOException {
int waiting = 0;
do {
try {
return super.read(buf, off, len);
} catch (SocketTimeoutException e) {
handleTimeout(e);
waiting += soTimeout;
handleTimeout(e, waiting);
}
} while (true);
}
@ -612,10 +620,7 @@ public class Client {
}
NetUtils.connect(this.socket, server, connectionTimeout);
if (rpcTimeout > 0) {
pingInterval = rpcTimeout; // rpcTimeout overwrites pingInterval
}
this.socket.setSoTimeout(pingInterval);
this.socket.setSoTimeout(soTimeout);
return;
} catch (ConnectTimeoutException toe) {
/* Check for an address change and update the local reference.

9
hadoop-common-project/hadoop-common/src/main/resources/core-default.xml
View File

@ -980,7 +980,7 @@ for ldap providers in the same way as above does.
<value>true</value>
<description>Send a ping to the server when timeout on reading the response,
if set to true. If no failure is detected, the client retries until at least
a byte is read.
a byte is read or the time given by ipc.client.rpc-timeout.ms is passed.
</description>
</property>
@ -997,10 +997,9 @@ for ldap providers in the same way as above does.
<name>ipc.client.rpc-timeout.ms</name>
<value>0</value>
<description>Timeout on waiting response from server, in milliseconds.
Currently this timeout works only when ipc.client.ping is set to true
because it uses the same facilities with IPC ping.
The timeout overrides the ipc.ping.interval and client will throw exception
instead of sending ping when the interval is passed.
If ipc.client.ping is set to true and this rpc-timeout is greater than
the value of ipc.ping.interval, the effective value of the rpc-timeout is
rounded up to multiple of ipc.ping.interval.
</description>
</property>

68
hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestRPC.java
View File

@ -1107,6 +1107,74 @@ public class TestRPC {
}
}
/**
* Test RPC timeout when ipc.client.ping is false.
*/
@Test(timeout=30000)
public void testClientRpcTimeoutWithoutPing() throws Exception {
final Server server = new RPC.Builder(conf)
.setProtocol(TestProtocol.class).setInstance(new TestImpl())
.setBindAddress(ADDRESS).setPort(0)
.setQueueSizePerHandler(1).setNumHandlers(1).setVerbose(true)
.build();
server.start();
final Configuration conf = new Configuration();
conf.setBoolean(CommonConfigurationKeys.IPC_CLIENT_PING_KEY, false);
conf.setInt(CommonConfigurationKeys.IPC_CLIENT_RPC_TIMEOUT_KEY, 1000);
final TestProtocol proxy =
RPC.getProxy(TestProtocol.class, TestProtocol.versionID,
NetUtils.getConnectAddress(server), conf);
try {
proxy.sleep(3000);
fail("RPC should time out.");
} catch (SocketTimeoutException e) {
LOG.info("got expected timeout.", e);
} finally {
server.stop();
RPC.stopProxy(proxy);
}
}
/**
* Test RPC timeout greater than ipc.ping.interval.
*/
@Test(timeout=30000)
public void testClientRpcTimeoutGreaterThanPingInterval() throws Exception {
final Server server = new RPC.Builder(conf)
.setProtocol(TestProtocol.class).setInstance(new TestImpl())
.setBindAddress(ADDRESS).setPort(0)
.setQueueSizePerHandler(1).setNumHandlers(1).setVerbose(true)
.build();
server.start();
final Configuration conf = new Configuration();
conf.setBoolean(CommonConfigurationKeys.IPC_CLIENT_PING_KEY, true);
conf.setInt(CommonConfigurationKeys.IPC_PING_INTERVAL_KEY, 800);
conf.setInt(CommonConfigurationKeys.IPC_CLIENT_RPC_TIMEOUT_KEY, 1000);
final TestProtocol proxy =
RPC.getProxy(TestProtocol.class, TestProtocol.versionID,
NetUtils.getConnectAddress(server), conf);
// should not time out.
proxy.sleep(300);
// should not time out because effective rpc-timeout is
// multiple of ping interval: 1600 (= 800 * (1000 / 800 + 1))
proxy.sleep(1300);
try {
proxy.sleep(2000);
fail("RPC should time out.");
} catch (SocketTimeoutException e) {
LOG.info("got expected timeout.", e);
} finally {
server.stop();
RPC.stopProxy(proxy);
}
}
public static void main(String[] args) throws IOException {
new TestRPC().testCallsInternal(conf);