big-data/hadoop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

HDFS-16175.Improve the configurable value of Server #PURGE_INTERVAL_NANOS. (#3307)

Browse Source 
Co-authored-by: zhujianghua <zhujianghua@zhujianghuadeMacBook-Pro.local>
Reviewed-by: Ayush Saxena <ayushsaxena@apache.org>
This commit is contained in:
jianghuazhu 2021-08-25 17:34:45 +08:00 committed by GitHub
parent 1b9927afe1
commit ad54f5195c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 55 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/CommonConfigurationKeysPublic.java
View File

@ -494,6 +494,10 @@ public class CommonConfigurationKeysPublic {
"ipc.server.log.slow.rpc";
public static final boolean IPC_SERVER_LOG_SLOW_RPC_DEFAULT = false;
public static final String IPC_SERVER_PURGE_INTERVAL_MINUTES_KEY =
"ipc.server.purge.interval";
public static final int IPC_SERVER_PURGE_INTERVAL_MINUTES_DEFAULT = 15;
/**
* @see
* <a href="{@docRoot}/../hadoop-project-dist/hadoop-common/core-default.xml">

29
hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java
View File

@ -488,6 +488,8 @@ protected ResponseBuffer initialValue() {
volatile private boolean running = true; // true while server runs
private CallQueueManager<Call> callQueue;
private long purgeIntervalNanos;
// maintains the set of client connections and handles idle timeouts
private ConnectionManager connectionManager;
private Listener listener = null;
@ -517,6 +519,20 @@ protected void setLogSlowRPC(boolean logSlowRPCFlag) {
this.logSlowRPC = logSlowRPCFlag;
}
private void setPurgeIntervalNanos(int purgeInterval) {
int tmpPurgeInterval = CommonConfigurationKeysPublic.
IPC_SERVER_PURGE_INTERVAL_MINUTES_DEFAULT;
if (purgeInterval > 0) {
tmpPurgeInterval = purgeInterval;
}
this.purgeIntervalNanos = TimeUnit.NANOSECONDS.convert(
tmpPurgeInterval, TimeUnit.MINUTES);
}
@VisibleForTesting
public long getPurgeIntervalNanos() {
return this.purgeIntervalNanos;
}
/**
* Logs a Slow RPC Request.
@ -1591,9 +1607,6 @@ Reader getReader() {
}
}
private final static long PURGE_INTERVAL_NANOS = TimeUnit.NANOSECONDS.convert(
15, TimeUnit.MINUTES);
// Sends responses of RPC back to clients.
private class Responder extends Thread {
private final Selector writeSelector;
@ -1629,7 +1642,7 @@ private void doRunLoop() {
try {
waitPending(); // If a channel is being registered, wait.
writeSelector.select(
TimeUnit.NANOSECONDS.toMillis(PURGE_INTERVAL_NANOS));
TimeUnit.NANOSECONDS.toMillis(purgeIntervalNanos));
Iterator<SelectionKey> iter = writeSelector.selectedKeys().iterator();
while (iter.hasNext()) {
SelectionKey key = iter.next();
@ -1652,7 +1665,7 @@ private void doRunLoop() {
}
}
long nowNanos = Time.monotonicNowNanos();
if (nowNanos < lastPurgeTimeNanos + PURGE_INTERVAL_NANOS) {
if (nowNanos < lastPurgeTimeNanos + purgeIntervalNanos) {
continue;
}
lastPurgeTimeNanos = nowNanos;
@ -1730,7 +1743,7 @@ private void doPurge(RpcCall call, long now) {
Iterator<RpcCall> iter = responseQueue.listIterator(0);
while (iter.hasNext()) {
call = iter.next();
if (now > call.responseTimestampNanos + PURGE_INTERVAL_NANOS) {
if (now > call.responseTimestampNanos + purgeIntervalNanos) {
closeConnection(call.connection);
break;
}
@ -3239,6 +3252,10 @@ protected Server(String bindAddress, int port,
CommonConfigurationKeysPublic.IPC_SERVER_LOG_SLOW_RPC,
CommonConfigurationKeysPublic.IPC_SERVER_LOG_SLOW_RPC_DEFAULT));
this.setPurgeIntervalNanos(conf.getInt(
CommonConfigurationKeysPublic.IPC_SERVER_PURGE_INTERVAL_MINUTES_KEY,
CommonConfigurationKeysPublic.IPC_SERVER_PURGE_INTERVAL_MINUTES_DEFAULT));
// Create the responder here
responder = new Responder();

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

@ -2460,6 +2460,14 @@
</description>
</property>
<property>
<name>ipc.server.purge.interval</name>
<value>15</value>
<description>Define how often calls are cleaned up in the server.
The default is 15 minutes. The unit is minutes.
</description>
</property>
<property>
<name>ipc.maximum.data.length</name>
<value>134217728</value>

20
hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestServer.java
View File

@ -25,8 +25,10 @@
import java.net.BindException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.util.concurrent.TimeUnit;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.ipc.Server.Call;
@ -184,4 +186,22 @@ public void testExceptionsHandlerSuppressed() {
assertTrue(handler.isSuppressedLog(IpcException.class));
assertFalse(handler.isSuppressedLog(RpcClientException.class));
}
@Test (timeout=300000)
public void testPurgeIntervalNanosConf() throws Exception {
Configuration conf = new Configuration();
conf.setInt(CommonConfigurationKeysPublic.
IPC_SERVER_PURGE_INTERVAL_MINUTES_KEY, 3);
Server server = new Server("0.0.0.0", 0, LongWritable.class,
1, conf) {
@Override
public Writable call(
RPC.RpcKind rpcKind, String protocol, Writable param,
long receiveTime) throws Exception {
return null;
}
};
long purgeInterval = TimeUnit.NANOSECONDS.convert(3, TimeUnit.MINUTES);
assertEquals(server.getPurgeIntervalNanos(), purgeInterval);
}
}