HDFS-5288. Close idle connections in portmap. Contributed by Haohui Mai

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/trunk@1544352 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Brandon Li 2013-11-21 22:43:55 +00:00
parent 9863cb1905
commit a379a7bbb9
6 changed files with 254 additions and 206 deletions

View File

@ -17,42 +17,111 @@
*/ */
package org.apache.hadoop.portmap; package org.apache.hadoop.portmap;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.apache.commons.logging.Log; import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory; import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.oncrpc.RpcProgram; import org.apache.hadoop.oncrpc.RpcProgram;
import org.apache.hadoop.oncrpc.SimpleTcpServer; import org.apache.hadoop.oncrpc.RpcUtil;
import org.apache.hadoop.oncrpc.SimpleUdpServer;
import org.apache.hadoop.util.StringUtils; import org.apache.hadoop.util.StringUtils;
import org.jboss.netty.bootstrap.ConnectionlessBootstrap;
import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.group.ChannelGroup;
import org.jboss.netty.channel.group.DefaultChannelGroup;
import org.jboss.netty.channel.socket.nio.NioDatagramChannelFactory;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
import org.jboss.netty.handler.timeout.IdleStateHandler;
import org.jboss.netty.util.HashedWheelTimer;
import com.google.common.annotations.VisibleForTesting;
/** /**
* Portmap service for binding RPC protocols. See RFC 1833 for details. * Portmap service for binding RPC protocols. See RFC 1833 for details.
*/ */
public class Portmap { final class Portmap {
public static final Log LOG = LogFactory.getLog(Portmap.class); private static final Log LOG = LogFactory.getLog(Portmap.class);
private static final int DEFAULT_IDLE_TIME_MILLISECONDS = 5000;
private static void startUDPServer(RpcProgramPortmap rpcProgram) { private ConnectionlessBootstrap udpServer;
rpcProgram.register(PortmapMapping.TRANSPORT_UDP, RpcProgram.RPCB_PORT); private ServerBootstrap tcpServer;
SimpleUdpServer udpServer = new SimpleUdpServer(RpcProgram.RPCB_PORT, private ChannelGroup allChannels = new DefaultChannelGroup();
rpcProgram, 1); private Channel udpChannel;
udpServer.run(); private Channel tcpChannel;
} private final RpcProgramPortmap handler = new RpcProgramPortmap(allChannels);
private static void startTCPServer(final RpcProgramPortmap rpcProgram) {
rpcProgram.register(PortmapMapping.TRANSPORT_TCP, RpcProgram.RPCB_PORT);
SimpleTcpServer tcpServer = new SimpleTcpServer(RpcProgram.RPCB_PORT,
rpcProgram, 1);
tcpServer.run();
}
public static void main(String[] args) { public static void main(String[] args) {
StringUtils.startupShutdownMessage(Portmap.class, args, LOG); StringUtils.startupShutdownMessage(Portmap.class, args, LOG);
RpcProgramPortmap program = new RpcProgramPortmap();
final int port = RpcProgram.RPCB_PORT;
Portmap pm = new Portmap();
try { try {
startUDPServer(program); pm.start(DEFAULT_IDLE_TIME_MILLISECONDS,
startTCPServer(program); new InetSocketAddress(port), new InetSocketAddress(port));
} catch (Throwable e) { } catch (Throwable e) {
LOG.fatal("Start server failure"); LOG.fatal("Failed to start the server. Cause:" + e.getMessage());
pm.shutdown();
System.exit(-1); System.exit(-1);
} }
} }
void shutdown() {
allChannels.close().awaitUninterruptibly();
tcpServer.releaseExternalResources();
udpServer.releaseExternalResources();
}
@VisibleForTesting
SocketAddress getTcpServerLocalAddress() {
return tcpChannel.getLocalAddress();
}
@VisibleForTesting
SocketAddress getUdpServerLoAddress() {
return udpChannel.getLocalAddress();
}
@VisibleForTesting
RpcProgramPortmap getHandler() {
return handler;
}
void start(final int idleTimeMilliSeconds, final SocketAddress tcpAddress,
final SocketAddress udpAddress) {
tcpServer = new ServerBootstrap(new NioServerSocketChannelFactory(
Executors.newCachedThreadPool(), Executors.newCachedThreadPool()));
tcpServer.setPipelineFactory(new ChannelPipelineFactory() {
private final HashedWheelTimer timer = new HashedWheelTimer();
private final IdleStateHandler idleStateHandler = new IdleStateHandler(
timer, 0, 0, idleTimeMilliSeconds, TimeUnit.MILLISECONDS);
@Override
public ChannelPipeline getPipeline() throws Exception {
return Channels.pipeline(RpcUtil.constructRpcFrameDecoder(),
RpcUtil.STAGE_RPC_MESSAGE_PARSER, idleStateHandler, handler,
RpcUtil.STAGE_RPC_TCP_RESPONSE);
}
});
udpServer = new ConnectionlessBootstrap(new NioDatagramChannelFactory(
Executors.newCachedThreadPool()));
udpServer.setPipeline(Channels.pipeline(RpcUtil.STAGE_RPC_MESSAGE_PARSER,
handler, RpcUtil.STAGE_RPC_UDP_RESPONSE));
tcpChannel = tcpServer.bind(tcpAddress);
udpChannel = udpServer.bind(udpAddress);
allChannels.add(tcpChannel);
allChannels.add(udpChannel);
LOG.info("Portmap server started at tcp://" + tcpChannel.getLocalAddress()
+ ", udp://" + udpChannel.getLocalAddress());
}
} }

View File

@ -17,9 +17,6 @@
*/ */
package org.apache.hadoop.portmap; package org.apache.hadoop.portmap;
import java.util.Arrays;
import java.util.Collection;
import org.apache.hadoop.oncrpc.RpcAcceptedReply; import org.apache.hadoop.oncrpc.RpcAcceptedReply;
import org.apache.hadoop.oncrpc.XDR; import org.apache.hadoop.oncrpc.XDR;
import org.apache.hadoop.oncrpc.security.VerifierNone; import org.apache.hadoop.oncrpc.security.VerifierNone;
@ -45,18 +42,13 @@ public static XDR booleanReply(XDR xdr, int xid, boolean value) {
return xdr; return xdr;
} }
public static XDR pmapList(XDR xdr, int xid, Collection<PortmapMapping> list) { public static XDR pmapList(XDR xdr, int xid, PortmapMapping[] list) {
RpcAcceptedReply.getAcceptInstance(xid, new VerifierNone()).write(xdr); RpcAcceptedReply.getAcceptInstance(xid, new VerifierNone()).write(xdr);
for (PortmapMapping mapping : list) { for (PortmapMapping mapping : list) {
System.out.println(mapping);
xdr.writeBoolean(true); // Value follows xdr.writeBoolean(true); // Value follows
mapping.serialize(xdr); mapping.serialize(xdr);
} }
xdr.writeBoolean(false); // No value follows xdr.writeBoolean(false); // No value follows
return xdr; return xdr;
} }
public static XDR pmapList(XDR xdr, int xid, PortmapMapping[] list) {
return pmapList(xdr, xid, Arrays.asList(list));
}
} }

View File

@ -18,8 +18,6 @@
package org.apache.hadoop.portmap; package org.apache.hadoop.portmap;
import java.util.HashMap; import java.util.HashMap;
import java.util.Map.Entry;
import java.util.Set;
import org.apache.commons.logging.Log; import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory; import org.apache.commons.logging.LogFactory;
@ -34,31 +32,34 @@
import org.jboss.netty.buffer.ChannelBuffer; import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers; import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.channel.ChannelHandlerContext; import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.group.ChannelGroup;
import org.jboss.netty.handler.timeout.IdleState;
import org.jboss.netty.handler.timeout.IdleStateAwareChannelUpstreamHandler;
import org.jboss.netty.handler.timeout.IdleStateEvent;
/** final class RpcProgramPortmap extends IdleStateAwareChannelUpstreamHandler implements PortmapInterface {
* An rpcbind request handler. static final int PROGRAM = 100000;
*/ static final int VERSION = 2;
public class RpcProgramPortmap extends RpcProgram implements PortmapInterface {
public static final int PROGRAM = 100000;
public static final int VERSION = 2;
private static final Log LOG = LogFactory.getLog(RpcProgramPortmap.class); private static final Log LOG = LogFactory.getLog(RpcProgramPortmap.class);
/** Map synchronized usis monitor lock of this instance */ /** Map synchronized usis monitor lock of this instance */
private final HashMap<String, PortmapMapping> map; private final HashMap<String, PortmapMapping> map;
public RpcProgramPortmap() { /** ChannelGroup that remembers all active channels for gracefully shutdown. */
super("portmap", "localhost", RPCB_PORT, PROGRAM, VERSION, VERSION); private final ChannelGroup allChannels;
map = new HashMap<String, PortmapMapping>(256);
}
/** Dump all the register RPC services */ RpcProgramPortmap(ChannelGroup allChannels) {
private synchronized void dumpRpcServices() { this.allChannels = allChannels;
Set<Entry<String, PortmapMapping>> entrySet = map.entrySet(); map = new HashMap<String, PortmapMapping>(256);
for (Entry<String, PortmapMapping> entry : entrySet) { PortmapMapping m = new PortmapMapping(PROGRAM, VERSION,
LOG.info("Service: " + entry.getKey() + " portmapping: " PortmapMapping.TRANSPORT_TCP, RpcProgram.RPCB_PORT);
+ entry.getValue()); PortmapMapping m1 = new PortmapMapping(PROGRAM, VERSION,
} PortmapMapping.TRANSPORT_UDP, RpcProgram.RPCB_PORT);
map.put(PortmapMapping.key(m), m);
map.put(PortmapMapping.key(m1), m1);
} }
@Override @Override
@ -77,7 +78,6 @@ public XDR set(int xid, XDR in, XDR out) {
PortmapMapping value = null; PortmapMapping value = null;
synchronized(this) { synchronized(this) {
map.put(key, mapping); map.put(key, mapping);
dumpRpcServices();
value = map.get(key); value = map.get(key);
} }
return PortmapResponse.intReply(out, xid, value.getPort()); return PortmapResponse.intReply(out, xid, value.getPort());
@ -126,21 +126,15 @@ public synchronized XDR dump(int xid, XDR in, XDR out) {
} }
@Override @Override
public void register(PortmapMapping mapping) { public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
String key = PortmapMapping.key(mapping); throws Exception {
synchronized(this) {
map.put(key, mapping);
}
}
@Override RpcInfo info = (RpcInfo) e.getMessage();
public void handleInternal(ChannelHandlerContext ctx, RpcInfo info) {
RpcCall rpcCall = (RpcCall) info.header(); RpcCall rpcCall = (RpcCall) info.header();
final Procedure portmapProc = Procedure.fromValue(rpcCall.getProcedure()); final Procedure portmapProc = Procedure.fromValue(rpcCall.getProcedure());
int xid = rpcCall.getXid(); int xid = rpcCall.getXid();
byte[] data = new byte[info.data().readableBytes()]; XDR in = new XDR(info.data().toByteBuffer().asReadOnlyBuffer(),
info.data().readBytes(data); XDR.State.READING);
XDR in = new XDR(data);
XDR out = new XDR(); XDR out = new XDR();
if (portmapProc == Procedure.PMAPPROC_NULL) { if (portmapProc == Procedure.PMAPPROC_NULL) {
@ -162,13 +156,29 @@ public void handleInternal(ChannelHandlerContext ctx, RpcInfo info) {
reply.write(out); reply.write(out);
} }
ChannelBuffer buf = ChannelBuffers.wrappedBuffer(out.asReadOnlyWrap().buffer()); ChannelBuffer buf = ChannelBuffers.wrappedBuffer(out.asReadOnlyWrap()
.buffer());
RpcResponse rsp = new RpcResponse(buf, info.remoteAddress()); RpcResponse rsp = new RpcResponse(buf, info.remoteAddress());
RpcUtil.sendRpcResponse(ctx, rsp); RpcUtil.sendRpcResponse(ctx, rsp);
} }
@Override @Override
protected boolean isIdempotent(RpcCall call) { public void channelOpen(ChannelHandlerContext ctx, ChannelStateEvent e)
return false; throws Exception {
allChannels.add(e.getChannel());
}
@Override
public void channelIdle(ChannelHandlerContext ctx, IdleStateEvent e)
throws Exception {
if (e.getState() == IdleState.ALL_IDLE) {
e.getChannel().close();
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) {
LOG.warn("Encountered ", e.getCause());
e.getChannel().close();
} }
} }

View File

@ -0,0 +1,116 @@
/**
* 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.portmap;
import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.HashMap;
import junit.framework.Assert;
import org.apache.hadoop.oncrpc.RpcCall;
import org.apache.hadoop.oncrpc.XDR;
import org.apache.hadoop.oncrpc.security.CredentialsNone;
import org.apache.hadoop.oncrpc.security.VerifierNone;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import org.mockito.internal.util.reflection.Whitebox;
public class TestPortmap {
private static Portmap pm = new Portmap();
private static final int SHORT_TIMEOUT_MILLISECONDS = 10;
private static final int RETRY_TIMES = 5;
private int xid;
@BeforeClass
public static void setup() {
pm.start(SHORT_TIMEOUT_MILLISECONDS, new InetSocketAddress("localhost", 0),
new InetSocketAddress("localhost", 0));
}
@AfterClass
public static void tearDown() {
pm.shutdown();
}
@Test(timeout = 1000)
public void testIdle() throws InterruptedException, IOException {
Socket s = new Socket();
try {
s.connect(pm.getTcpServerLocalAddress());
int i = 0;
while (!s.isConnected() && i < RETRY_TIMES) {
++i;
Thread.sleep(SHORT_TIMEOUT_MILLISECONDS);
}
Assert.assertTrue("Failed to connect to the server", s.isConnected()
&& i < RETRY_TIMES);
int b = s.getInputStream().read();
Assert.assertTrue("The server failed to disconnect", b == -1);
} finally {
s.close();
}
}
@Test(timeout = 1000)
public void testRegistration() throws IOException, InterruptedException {
XDR req = new XDR();
RpcCall.getInstance(++xid, RpcProgramPortmap.PROGRAM,
RpcProgramPortmap.VERSION,
PortmapInterface.Procedure.PMAPPROC_SET.getValue(),
new CredentialsNone(), new VerifierNone()).write(req);
PortmapMapping sent = new PortmapMapping(90000, 1,
PortmapMapping.TRANSPORT_TCP, 1234);
sent.serialize(req);
byte[] reqBuf = req.getBytes();
DatagramSocket s = new DatagramSocket();
DatagramPacket p = new DatagramPacket(reqBuf, reqBuf.length,
pm.getUdpServerLoAddress());
try {
s.send(p);
} finally {
s.close();
}
// Give the server a chance to process the request
Thread.sleep(100);
boolean found = false;
@SuppressWarnings("unchecked")
HashMap<String, PortmapMapping> map = (HashMap<String, PortmapMapping>) Whitebox
.getInternalState(pm.getHandler(), "map");
for (PortmapMapping m : map.values()) {
if (m.getPort() == sent.getPort()
&& PortmapMapping.key(m).equals(PortmapMapping.key(sent))) {
found = true;
break;
}
}
Assert.assertTrue("Registration failed", found);
}
}

View File

@ -1,141 +0,0 @@
/**
* 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.hdfs.nfs;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hdfs.nfs.mount.RpcProgramMountd;
import org.apache.hadoop.nfs.nfs3.Nfs3Constant;
import org.apache.hadoop.oncrpc.RegistrationClient;
import org.apache.hadoop.oncrpc.RpcCall;
import org.apache.hadoop.oncrpc.XDR;
import org.apache.hadoop.oncrpc.security.CredentialsNone;
import org.apache.hadoop.oncrpc.security.VerifierNone;
import org.apache.hadoop.portmap.PortmapMapping;
import org.apache.hadoop.portmap.PortmapRequest;
public class TestPortmapRegister {
public static final Log LOG = LogFactory.getLog(TestPortmapRegister.class);
static void testRequest(XDR request, XDR request2) {
RegistrationClient registrationClient = new RegistrationClient(
"localhost", Nfs3Constant.SUN_RPCBIND, request);
registrationClient.run();
}
public static void main(String[] args) throws InterruptedException {
PortmapMapping mapEntry = new PortmapMapping(RpcProgramMountd.PROGRAM,
RpcProgramMountd.VERSION_1, PortmapMapping.TRANSPORT_UDP,
RpcProgramMountd.PORT);
XDR mappingRequest = PortmapRequest.create(mapEntry);
RegistrationClient registrationClient = new RegistrationClient(
"localhost", Nfs3Constant.SUN_RPCBIND, mappingRequest);
registrationClient.run();
Thread t1 = new Runtest1();
//Thread t2 = testa.new Runtest2();
t1.start();
//t2.start();
t1.join();
//t2.join();
//testDump();
}
static class Runtest1 extends Thread {
@Override
public void run() {
//testGetportMount();
PortmapMapping mapEntry = new PortmapMapping(RpcProgramMountd.PROGRAM,
RpcProgramMountd.VERSION_1, PortmapMapping.TRANSPORT_UDP,
RpcProgramMountd.PORT);
XDR req = PortmapRequest.create(mapEntry);
testRequest(req, req);
}
}
static class Runtest2 extends Thread {
@Override
public void run() {
testDump();
}
}
static void createPortmapXDRheader(XDR xdr_out, int procedure) {
// TODO: Move this to RpcRequest
RpcCall.getInstance(0, 100000, 2, procedure, new CredentialsNone(),
new VerifierNone()).write(xdr_out);
/*
xdr_out.putInt(1); //unix auth
xdr_out.putVariableOpaque(new byte[20]);
xdr_out.putInt(0);
xdr_out.putInt(0);
*/
}
static void testGetportMount() {
XDR xdr_out = new XDR();
createPortmapXDRheader(xdr_out, 3);
xdr_out.writeInt(100005);
xdr_out.writeInt(1);
xdr_out.writeInt(6);
xdr_out.writeInt(0);
XDR request2 = new XDR();
createPortmapXDRheader(xdr_out, 3);
request2.writeInt(100005);
request2.writeInt(1);
request2.writeInt(6);
request2.writeInt(0);
testRequest(xdr_out, request2);
}
static void testGetport() {
XDR xdr_out = new XDR();
createPortmapXDRheader(xdr_out, 3);
xdr_out.writeInt(100003);
xdr_out.writeInt(3);
xdr_out.writeInt(6);
xdr_out.writeInt(0);
XDR request2 = new XDR();
createPortmapXDRheader(xdr_out, 3);
request2.writeInt(100003);
request2.writeInt(3);
request2.writeInt(6);
request2.writeInt(0);
testRequest(xdr_out, request2);
}
static void testDump() {
XDR xdr_out = new XDR();
createPortmapXDRheader(xdr_out, 4);
testRequest(xdr_out, xdr_out);
}
}

View File

@ -714,6 +714,8 @@ Release 2.2.1 - UNRELEASED
HDFS-5014. Process register commands with out holding BPOfferService lock. HDFS-5014. Process register commands with out holding BPOfferService lock.
(Vinaykumar B via umamahesh) (Vinaykumar B via umamahesh)
HDFS-5288. Close idle connections in portmap (Haohui Mai via brandonli)
Release 2.2.0 - 2013-10-13 Release 2.2.0 - 2013-10-13
INCOMPATIBLE CHANGES INCOMPATIBLE CHANGES