diff --git a/modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpServerTransport.java b/modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpServerTransport.java index 5fae4cb4beac2..fcaebcb61c010 100644 --- a/modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpServerTransport.java +++ b/modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpServerTransport.java @@ -61,8 +61,9 @@ import org.elasticsearch.http.HttpServerChannel; import org.elasticsearch.http.netty4.cors.Netty4CorsHandler; import org.elasticsearch.threadpool.ThreadPool; -import org.elasticsearch.transport.SharedGroupFactory; import org.elasticsearch.transport.NettyAllocator; +import org.elasticsearch.transport.NettyByteBufSizer; +import org.elasticsearch.transport.SharedGroupFactory; import org.elasticsearch.transport.netty4.Netty4Utils; import java.net.InetSocketAddress; @@ -284,6 +285,7 @@ public ChannelHandler configureServerChannelHandler() { protected static class HttpChannelHandler extends ChannelInitializer { private final Netty4HttpServerTransport transport; + private final NettyByteBufSizer byteBufSizer; private final Netty4HttpRequestCreator requestCreator; private final Netty4HttpRequestHandler requestHandler; private final HttpHandlingSettings handlingSettings; @@ -291,6 +293,7 @@ protected static class HttpChannelHandler extends ChannelInitializer { protected HttpChannelHandler(final Netty4HttpServerTransport transport, final HttpHandlingSettings handlingSettings) { this.transport = transport; this.handlingSettings = handlingSettings; + this.byteBufSizer = new NettyByteBufSizer(); this.requestCreator = new Netty4HttpRequestCreator(); this.requestHandler = new Netty4HttpRequestHandler(transport); } @@ -299,6 +302,7 @@ protected HttpChannelHandler(final Netty4HttpServerTransport transport, final Ht protected void initChannel(Channel ch) throws Exception { Netty4HttpChannel nettyHttpChannel = new Netty4HttpChannel(ch); ch.attr(HTTP_CHANNEL_KEY).set(nettyHttpChannel); + ch.pipeline().addLast("byte_buf_sizer", byteBufSizer); ch.pipeline().addLast("read_timeout", new ReadTimeoutHandler(transport.readTimeoutMillis, TimeUnit.MILLISECONDS)); final HttpRequestDecoder decoder = new HttpRequestDecoder( handlingSettings.getMaxInitialLineLength(), diff --git a/modules/transport-netty4/src/main/java/org/elasticsearch/transport/NettyByteBufSizer.java b/modules/transport-netty4/src/main/java/org/elasticsearch/transport/NettyByteBufSizer.java new file mode 100644 index 0000000000000..e2eadf5690be2 --- /dev/null +++ b/modules/transport-netty4/src/main/java/org/elasticsearch/transport/NettyByteBufSizer.java @@ -0,0 +1,43 @@ +/* + * Licensed to Elasticsearch under one or more contributor + * license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright + * ownership. Elasticsearch 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.elasticsearch.transport; + +import io.netty.buffer.ByteBuf; +import io.netty.channel.ChannelHandler; +import io.netty.channel.ChannelHandlerContext; +import io.netty.handler.codec.MessageToMessageDecoder; + +import java.util.List; + +@ChannelHandler.Sharable +public class NettyByteBufSizer extends MessageToMessageDecoder { + + @Override + protected void decode(ChannelHandlerContext ctx, ByteBuf buf, List out) { + int readableBytes = buf.readableBytes(); + if (buf.capacity() >= 1024) { + ByteBuf resized = buf.discardReadBytes().capacity(readableBytes); + assert resized.readableBytes() == readableBytes; + out.add(resized.retain()); + } else { + out.add(buf.retain()); + } + } +} diff --git a/modules/transport-netty4/src/main/java/org/elasticsearch/transport/netty4/Netty4Transport.java b/modules/transport-netty4/src/main/java/org/elasticsearch/transport/netty4/Netty4Transport.java index 9c9147b17401b..001392f500554 100644 --- a/modules/transport-netty4/src/main/java/org/elasticsearch/transport/netty4/Netty4Transport.java +++ b/modules/transport-netty4/src/main/java/org/elasticsearch/transport/netty4/Netty4Transport.java @@ -53,8 +53,9 @@ import org.elasticsearch.core.internal.net.NetUtils; import org.elasticsearch.indices.breaker.CircuitBreakerService; import org.elasticsearch.threadpool.ThreadPool; -import org.elasticsearch.transport.SharedGroupFactory; import org.elasticsearch.transport.NettyAllocator; +import org.elasticsearch.transport.NettyByteBufSizer; +import org.elasticsearch.transport.SharedGroupFactory; import org.elasticsearch.transport.TcpTransport; import org.elasticsearch.transport.TransportSettings; @@ -327,6 +328,7 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws E protected class ServerChannelInitializer extends ChannelInitializer { protected final String name; + private final NettyByteBufSizer sizer = new NettyByteBufSizer(); protected ServerChannelInitializer(String name) { this.name = name; @@ -337,6 +339,7 @@ protected void initChannel(Channel ch) throws Exception { addClosedExceptionLogger(ch); Netty4TcpChannel nettyTcpChannel = new Netty4TcpChannel(ch, true, name, ch.newSucceededFuture()); ch.attr(CHANNEL_KEY).set(nettyTcpChannel); + ch.pipeline().addLast("byte_buf_sizer", sizer); ch.pipeline().addLast("logging", new ESLoggingHandler()); ch.pipeline().addLast("dispatcher", new Netty4MessageChannelHandler(pageCacheRecycler, Netty4Transport.this)); serverAcceptedChannel(nettyTcpChannel);