Repository: hadoop Updated Branches: refs/heads/branch-2 842a54a5f -> bbdd99086
HDFS-7280. Use netty 4 in WebImageViewer. Contributed by Haohui Mai. Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/bbdd9908 Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/bbdd9908 Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/bbdd9908 Branch: refs/heads/branch-2 Commit: bbdd990864d677e99b8fc73bdf720d66e2187d2c Parents: 842a54a Author: Haohui Mai <[email protected]> Authored: Tue Oct 28 16:53:53 2014 -0700 Committer: Haohui Mai <[email protected]> Committed: Tue Oct 28 16:54:04 2014 -0700 ---------------------------------------------------------------------- hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt | 2 + hadoop-hdfs-project/hadoop-hdfs/pom.xml | 2 +- .../offlineImageViewer/FSImageHandler.java | 145 ++++++++++--------- .../OfflineImageViewerPB.java | 13 +- .../offlineImageViewer/WebImageViewer.java | 109 +++++++------- .../TestOfflineImageViewer.java | 7 +- .../TestOfflineImageViewerForAcl.java | 5 +- hadoop-project/pom.xml | 6 + 8 files changed, 156 insertions(+), 133 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hadoop/blob/bbdd9908/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt b/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt index d04324a..27f47fc 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt +++ b/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt @@ -62,6 +62,8 @@ Release 2.7.0 - UNRELEASED HDFS-7213. processIncrementalBlockReport performance degradation. (Eric Payne via kihwal) + HDFS-7280. Use netty 4 in WebImageViewer. (wheat9) + OPTIMIZATIONS BUG FIXES http://git-wip-us.apache.org/repos/asf/hadoop/blob/bbdd9908/hadoop-hdfs-project/hadoop-hdfs/pom.xml ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/pom.xml b/hadoop-hdfs-project/hadoop-hdfs/pom.xml index 35ccd57..4c3ce8a 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/pom.xml +++ b/hadoop-hdfs-project/hadoop-hdfs/pom.xml @@ -182,7 +182,7 @@ http://maven.apache.org/xsd/maven-4.0.0.xsd";> </dependency> <dependency> <groupId>io.netty</groupId> - <artifactId>netty</artifactId> + <artifactId>netty-all</artifactId> <scope>compile</scope> </dependency> <dependency> http://git-wip-us.apache.org/repos/asf/hadoop/blob/bbdd9908/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/offlineImageViewer/FSImageHandler.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/offlineImageViewer/FSImageHandler.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/offlineImageViewer/FSImageHandler.java index dea6422..eb93c87 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/offlineImageViewer/FSImageHandler.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/offlineImageViewer/FSImageHandler.java @@ -22,105 +22,118 @@ import java.io.IOException; import java.util.List; import java.util.Map; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.netty.channel.ChannelFutureListener; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.SimpleChannelInboundHandler; +import io.netty.channel.group.ChannelGroup; +import io.netty.handler.codec.http.DefaultFullHttpResponse; +import io.netty.handler.codec.http.DefaultHttpResponse; +import static io.netty.handler.codec.http.HttpResponseStatus.*; + +import io.netty.handler.codec.http.HttpMethod; +import io.netty.handler.codec.http.HttpRequest; +import io.netty.handler.codec.http.HttpResponseStatus; +import static io.netty.handler.codec.http.HttpVersion.*; +import io.netty.handler.codec.http.QueryStringDecoder; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.hadoop.hdfs.web.JsonUtil; -import org.apache.hadoop.ipc.RemoteException; -import org.jboss.netty.channel.ChannelFuture; -import org.jboss.netty.channel.ChannelFutureListener; -import org.jboss.netty.channel.ChannelHandlerContext; -import org.jboss.netty.channel.MessageEvent; -import org.jboss.netty.channel.SimpleChannelUpstreamHandler; -import org.jboss.netty.handler.codec.http.DefaultHttpResponse; -import org.jboss.netty.handler.codec.http.HttpHeaders; -import org.jboss.netty.handler.codec.http.HttpMethod; -import org.jboss.netty.handler.codec.http.HttpRequest; -import org.jboss.netty.handler.codec.http.HttpResponse; -import org.jboss.netty.handler.codec.http.HttpResponseStatus; -import org.jboss.netty.handler.codec.http.HttpVersion; -import org.jboss.netty.handler.codec.http.QueryStringDecoder; - -import javax.management.Query; +import org.apache.hadoop.hdfs.web.resources.ExceptionHandler; + /** * Implement the read-only WebHDFS API for fsimage. */ -class FSImageHandler extends SimpleChannelUpstreamHandler { +class FSImageHandler extends SimpleChannelInboundHandler<HttpRequest> { public static final Log LOG = LogFactory.getLog(FSImageHandler.class); private final FSImageLoader image; - - FSImageHandler(FSImageLoader image) throws IOException { - this.image = image; - } + private final ChannelGroup activeChannels; @Override - public void messageReceived( - ChannelHandlerContext ctx, MessageEvent e) throws Exception { - ChannelFuture future = e.getFuture(); - try { - future = handleOperation(e); - } finally { - future.addListener(ChannelFutureListener.CLOSE); - } + public void channelActive(ChannelHandlerContext ctx) throws Exception { + activeChannels.add(ctx.channel()); } - private ChannelFuture handleOperation(MessageEvent e) - throws IOException { - HttpRequest request = (HttpRequest) e.getMessage(); - HttpResponse response = new DefaultHttpResponse( - HttpVersion.HTTP_1_1, HttpResponseStatus.OK); - response.setHeader(HttpHeaders.Names.CONTENT_TYPE, "application/json"); + FSImageHandler(FSImageLoader image, ChannelGroup activeChannels) throws IOException { + this.image = image; + this.activeChannels = activeChannels; + } + @Override + public void channelRead0(ChannelHandlerContext ctx, HttpRequest request) + throws Exception { if (request.getMethod() != HttpMethod.GET) { - response.setStatus(HttpResponseStatus.METHOD_NOT_ALLOWED); - return e.getChannel().write(response); + DefaultHttpResponse resp = new DefaultHttpResponse(HTTP_1_1, + METHOD_NOT_ALLOWED); + resp.headers().set("Connection", "close"); + ctx.write(resp).addListener(ChannelFutureListener.CLOSE); + return; } QueryStringDecoder decoder = new QueryStringDecoder(request.getUri()); final String op = getOp(decoder); - String content; - String path = null; - try { - path = getPath(decoder); - if ("GETFILESTATUS".equals(op)) { - content = image.getFileStatus(path); - } else if ("LISTSTATUS".equals(op)) { - content = image.listStatus(path); - } else if ("GETACLSTATUS".equals(op)) { - content = image.getAclStatus(path); - } else { - throw new IllegalArgumentException("Invalid value for webhdfs parameter" + " \"op\""); - } - } catch (IllegalArgumentException ex) { - response.setStatus(HttpResponseStatus.BAD_REQUEST); - content = JsonUtil.toJsonString(ex); - } catch (FileNotFoundException ex) { - response.setStatus(HttpResponseStatus.NOT_FOUND); - content = JsonUtil.toJsonString(ex); - } catch (Exception ex) { - content = JsonUtil.toJsonString(ex); + final String content; + String path = getPath(decoder); + if ("GETFILESTATUS".equals(op)) { + content = image.getFileStatus(path); + } else if ("LISTSTATUS".equals(op)) { + content = image.listStatus(path); + } else if ("GETACLSTATUS".equals(op)) { + content = image.getAclStatus(path); + } else { + throw new IllegalArgumentException("Invalid value for webhdfs parameter" + " \"op\""); } - HttpHeaders.setContentLength(response, content.length()); - e.getChannel().write(response); - ChannelFuture future = e.getChannel().write(content); + LOG.info("op=" + op + " target=" + path); + + DefaultFullHttpResponse resp = new DefaultFullHttpResponse( + HTTP_1_1, HttpResponseStatus.OK, + Unpooled.wrappedBuffer(content.getBytes())); + resp.headers().set("Content-Type", "application/json"); + resp.headers().set("Content-Length", resp.content().readableBytes()); + resp.headers().set("Connection", "close"); + ctx.write(resp).addListener(ChannelFutureListener.CLOSE); + } - LOG.info(response.getStatus().getCode() + " method=" - + request.getMethod().getName() + " op=" + op + " target=" + path); + @Override + public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { + ctx.flush(); + } + + @Override + public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) + throws Exception { + Exception e = cause instanceof Exception ? (Exception) cause : new + Exception(cause); + final String output = JsonUtil.toJsonString(e); + ByteBuf content = Unpooled.wrappedBuffer(output.getBytes()); + final DefaultFullHttpResponse resp = new DefaultFullHttpResponse( + HTTP_1_1, INTERNAL_SERVER_ERROR, content); + + resp.headers().set("Content-Type", "application/json"); + if (e instanceof IllegalArgumentException) { + resp.setStatus(BAD_REQUEST); + } else if (e instanceof FileNotFoundException) { + resp.setStatus(NOT_FOUND); + } - return future; + resp.headers().set("Content-Length", resp.content().readableBytes()); + resp.headers().set("Connection", "close"); + ctx.write(resp).addListener(ChannelFutureListener.CLOSE); } private static String getOp(QueryStringDecoder decoder) { - Map<String, List<String>> parameters = decoder.getParameters(); + Map<String, List<String>> parameters = decoder.parameters(); return parameters.containsKey("op") ? parameters.get("op").get(0).toUpperCase() : null; } private static String getPath(QueryStringDecoder decoder) throws FileNotFoundException { - String path = decoder.getPath(); + String path = decoder.path(); if (path.startsWith("/webhdfs/v1/")) { return path.substring(11); } else { http://git-wip-us.apache.org/repos/asf/hadoop/blob/bbdd9908/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/offlineImageViewer/OfflineImageViewerPB.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/offlineImageViewer/OfflineImageViewerPB.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/offlineImageViewer/OfflineImageViewerPB.java index 8f3906e..c5d442b 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/offlineImageViewer/OfflineImageViewerPB.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/offlineImageViewer/OfflineImageViewerPB.java @@ -112,12 +112,12 @@ public class OfflineImageViewerPB { * Command line options * @throws IOException */ - public static void main(String[] args) throws IOException { + public static void main(String[] args) throws Exception { int status = run(args); System.exit(status); } - public static int run(String[] args) throws IOException { + public static int run(String[] args) throws Exception { Options options = buildOptions(); if (args.length == 0) { printUsage(); @@ -159,8 +159,13 @@ public class OfflineImageViewerPB { "r")); } else if (processor.equals("Web")) { String addr = cmd.getOptionValue("addr", "localhost:5978"); - new WebImageViewer(NetUtils.createSocketAddr(addr)) - .initServerAndWait(inputFile); + WebImageViewer viewer = new WebImageViewer(NetUtils.createSocketAddr + (addr)); + try { + viewer.start(inputFile); + } finally { + viewer.close(); + } } return 0; } catch (EOFException e) { http://git-wip-us.apache.org/repos/asf/hadoop/blob/bbdd9908/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/offlineImageViewer/WebImageViewer.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/offlineImageViewer/WebImageViewer.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/offlineImageViewer/WebImageViewer.java index f86bd10..087972f 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/offlineImageViewer/WebImageViewer.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/offlineImageViewer/WebImageViewer.java @@ -17,48 +17,51 @@ */ package org.apache.hadoop.hdfs.tools.offlineImageViewer; -import java.io.IOException; -import java.net.InetSocketAddress; -import java.util.concurrent.Executors; - +import com.google.common.annotations.VisibleForTesting; +import io.netty.bootstrap.ServerBootstrap; +import io.netty.channel.Channel; +import io.netty.channel.ChannelInitializer; +import io.netty.channel.ChannelPipeline; +import io.netty.channel.EventLoopGroup; +import io.netty.channel.group.ChannelGroup; +import io.netty.channel.group.DefaultChannelGroup; +import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.channel.socket.SocketChannel; +import io.netty.channel.socket.nio.NioServerSocketChannel; +import io.netty.handler.codec.http.HttpRequestDecoder; +import io.netty.handler.codec.http.HttpResponseEncoder; +import io.netty.handler.codec.string.StringEncoder; +import io.netty.util.concurrent.GlobalEventExecutor; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.jboss.netty.bootstrap.ServerBootstrap; -import org.jboss.netty.channel.Channel; -import org.jboss.netty.channel.ChannelFactory; -import org.jboss.netty.channel.ChannelHandlerContext; -import org.jboss.netty.channel.ChannelPipeline; -import org.jboss.netty.channel.ChannelStateEvent; -import org.jboss.netty.channel.Channels; -import org.jboss.netty.channel.SimpleChannelUpstreamHandler; -import org.jboss.netty.channel.group.ChannelGroup; -import org.jboss.netty.channel.group.DefaultChannelGroup; -import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory; -import org.jboss.netty.handler.codec.http.HttpRequestDecoder; -import org.jboss.netty.handler.codec.http.HttpResponseEncoder; -import org.jboss.netty.handler.codec.string.StringEncoder; -import com.google.common.annotations.VisibleForTesting; +import java.io.Closeable; +import java.io.IOException; +import java.net.InetSocketAddress; /** * WebImageViewer loads a fsimage and exposes read-only WebHDFS API for its * namespace. */ -public class WebImageViewer { +public class WebImageViewer implements Closeable { public static final Log LOG = LogFactory.getLog(WebImageViewer.class); private Channel channel; private InetSocketAddress address; - private final ChannelFactory factory = - new NioServerSocketChannelFactory(Executors.newCachedThreadPool(), - Executors.newCachedThreadPool(), 1); - private final ServerBootstrap bootstrap = new ServerBootstrap(factory); - static final ChannelGroup allChannels = - new DefaultChannelGroup("WebImageViewer"); + private final ServerBootstrap bootstrap; + private final EventLoopGroup bossGroup; + private final EventLoopGroup workerGroup; + private final ChannelGroup allChannels; public WebImageViewer(InetSocketAddress address) { this.address = address; + this.bossGroup = new NioEventLoopGroup(); + this.workerGroup = new NioEventLoopGroup(); + this.allChannels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE); + this.bootstrap = new ServerBootstrap() + .group(bossGroup, workerGroup) + .channel(NioServerSocketChannel.class); } /** @@ -66,13 +69,13 @@ public class WebImageViewer { * @param fsimage the fsimage to load. * @throws IOException if failed to load the fsimage. */ - public void initServerAndWait(String fsimage) throws IOException { - initServer(fsimage); + public void start(String fsimage) throws IOException { try { - channel.getCloseFuture().await(); + initServer(fsimage); + channel.closeFuture().await(); } catch (InterruptedException e) { LOG.info("Interrupted. Stopping the WebImageViewer."); - shutdown(); + close(); } } @@ -82,37 +85,26 @@ public class WebImageViewer { * @throws IOException if fail to load the fsimage. */ @VisibleForTesting - public void initServer(String fsimage) throws IOException { - FSImageLoader loader = FSImageLoader.load(fsimage); + public void initServer(String fsimage) + throws IOException, InterruptedException { + final FSImageLoader loader = FSImageLoader.load(fsimage); - ChannelPipeline pipeline = Channels.pipeline(); - pipeline.addLast("channelTracker", new SimpleChannelUpstreamHandler() { + bootstrap.childHandler(new ChannelInitializer<SocketChannel>() { @Override - public void channelOpen(ChannelHandlerContext ctx, ChannelStateEvent e) - throws Exception { - allChannels.add(e.getChannel()); + protected void initChannel(SocketChannel ch) throws Exception { + ChannelPipeline p = ch.pipeline(); + p.addLast(new HttpRequestDecoder(), + new StringEncoder(), + new HttpResponseEncoder(), + new FSImageHandler(loader, allChannels)); } }); - pipeline.addLast("httpDecoder", new HttpRequestDecoder()); - pipeline.addLast("requestHandler", new FSImageHandler(loader)); - pipeline.addLast("stringEncoder", new StringEncoder()); - pipeline.addLast("httpEncoder", new HttpResponseEncoder()); - bootstrap.setPipeline(pipeline); - channel = bootstrap.bind(address); - allChannels.add(channel); - address = (InetSocketAddress) channel.getLocalAddress(); - LOG.info("WebImageViewer started. Listening on " + address.toString() - + ". Press Ctrl+C to stop the viewer."); - } + channel = bootstrap.bind(address).sync().channel(); + allChannels.add(channel); - /** - * Stop WebImageViewer. - */ - @VisibleForTesting - public void shutdown() { - allChannels.close().awaitUninterruptibly(); - factory.releaseExternalResources(); + address = (InetSocketAddress) channel.localAddress(); + LOG.info("WebImageViewer started. Listening on " + address.toString() + ". Press Ctrl+C to stop the viewer."); } /** @@ -123,4 +115,11 @@ public class WebImageViewer { public int getPort() { return address.getPort(); } + + @Override + public void close() { + allChannels.close().awaitUninterruptibly(); + bossGroup.shutdownGracefully(); + workerGroup.shutdownGracefully(); + } } http://git-wip-us.apache.org/repos/asf/hadoop/blob/bbdd9908/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/tools/offlineImageViewer/TestOfflineImageViewer.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/tools/offlineImageViewer/TestOfflineImageViewer.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/tools/offlineImageViewer/TestOfflineImageViewer.java index c7e09ea..5c0d26c 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/tools/offlineImageViewer/TestOfflineImageViewer.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/tools/offlineImageViewer/TestOfflineImageViewer.java @@ -237,7 +237,7 @@ public class TestOfflineImageViewer { } @Test - public void testFileDistributionCalculatorWithOptions() throws IOException { + public void testFileDistributionCalculatorWithOptions() throws Exception { int status = OfflineImageViewerPB.run(new String[] {"-i", originalFsimage.getAbsolutePath(), "-o", "-", "-p", "FileDistribution", "-maxSize", "512", "-step", "8"}); @@ -258,8 +258,7 @@ public class TestOfflineImageViewer { } @Test - public void testWebImageViewer() throws IOException, InterruptedException, - URISyntaxException { + public void testWebImageViewer() throws Exception { WebImageViewer viewer = new WebImageViewer( NetUtils.createSocketAddr("localhost:0")); try { @@ -319,7 +318,7 @@ public class TestOfflineImageViewer { connection.getResponseCode()); } finally { // shutdown the viewer - viewer.shutdown(); + viewer.close(); } } http://git-wip-us.apache.org/repos/asf/hadoop/blob/bbdd9908/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/tools/offlineImageViewer/TestOfflineImageViewerForAcl.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/tools/offlineImageViewer/TestOfflineImageViewerForAcl.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/tools/offlineImageViewer/TestOfflineImageViewerForAcl.java index 7b4804a..56963a6 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/tools/offlineImageViewer/TestOfflineImageViewerForAcl.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/tools/offlineImageViewer/TestOfflineImageViewerForAcl.java @@ -157,8 +157,7 @@ public class TestOfflineImageViewerForAcl { } @Test - public void testWebImageViewerForAcl() throws IOException, - InterruptedException, URISyntaxException { + public void testWebImageViewerForAcl() throws Exception { WebImageViewer viewer = new WebImageViewer( NetUtils.createSocketAddr("localhost:0")); try { @@ -200,7 +199,7 @@ public class TestOfflineImageViewerForAcl { connection.getResponseCode()); } finally { // shutdown the viewer - viewer.shutdown(); + viewer.close(); } } } http://git-wip-us.apache.org/repos/asf/hadoop/blob/bbdd9908/hadoop-project/pom.xml ---------------------------------------------------------------------- diff --git a/hadoop-project/pom.xml b/hadoop-project/pom.xml index c26d366..a0d5420 100644 --- a/hadoop-project/pom.xml +++ b/hadoop-project/pom.xml @@ -530,6 +530,12 @@ </dependency> <dependency> + <groupId>io.netty</groupId> + <artifactId>netty-all</artifactId> + <version>4.0.23.Final</version> + </dependency> + + <dependency> <groupId>commons-io</groupId> <artifactId>commons-io</artifactId> <version>2.4</version>
