Add experimental nio-transport plugin with http

core/src/main/java/org/elasticsearch/common/settings/ClusterSettings.java

@@ -245,6 +245,11 @@ public void apply(Settings value, Settings current, Settings previous) {
                     HttpTransportSettings.SETTING_HTTP_MAX_HEADER_SIZE,
                     HttpTransportSettings.SETTING_HTTP_MAX_INITIAL_LINE_LENGTH,
                     HttpTransportSettings.SETTING_HTTP_RESET_COOKIES,
+                    HttpTransportSettings.SETTING_HTTP_TCP_NO_DELAY,
+                    HttpTransportSettings.SETTING_HTTP_TCP_KEEP_ALIVE,
+                    HttpTransportSettings.SETTING_HTTP_TCP_REUSE_ADDRESS,
+                    HttpTransportSettings.SETTING_HTTP_TCP_SEND_BUFFER_SIZE,
+                    HttpTransportSettings.SETTING_HTTP_TCP_RECEIVE_BUFFER_SIZE,
                     HierarchyCircuitBreakerService.TOTAL_CIRCUIT_BREAKER_LIMIT_SETTING,
                     HierarchyCircuitBreakerService.FIELDDATA_CIRCUIT_BREAKER_LIMIT_SETTING,
                     HierarchyCircuitBreakerService.FIELDDATA_CIRCUIT_BREAKER_OVERHEAD_SETTING,

core/src/main/java/org/elasticsearch/http/HttpTransportSettings.java

@@ -20,6 +20,7 @@
 package org.elasticsearch.http;
 
 import org.elasticsearch.common.Booleans;
+import org.elasticsearch.common.network.NetworkService;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Setting.Property;
 import org.elasticsearch.common.transport.PortsRange;
@@ -30,6 +31,7 @@
 import java.util.function.Function;
 
 import static java.util.Collections.emptyList;
+import static org.elasticsearch.common.settings.Setting.boolSetting;
 import static org.elasticsearch.common.settings.Setting.listSetting;
 
 public final class HttpTransportSettings {
@@ -91,6 +93,17 @@ public final class HttpTransportSettings {
     public static final Setting<Boolean> SETTING_HTTP_RESET_COOKIES =
         Setting.boolSetting("http.reset_cookies", false, Property.NodeScope);
 
+    public static final Setting<Boolean> SETTING_HTTP_TCP_NO_DELAY =
+        boolSetting("http.tcp_no_delay", NetworkService.TCP_NO_DELAY, Setting.Property.NodeScope);
+    public static final Setting<Boolean> SETTING_HTTP_TCP_KEEP_ALIVE =
+        boolSetting("http.tcp.keep_alive", NetworkService.TCP_KEEP_ALIVE, Setting.Property.NodeScope);
+    public static final Setting<Boolean> SETTING_HTTP_TCP_REUSE_ADDRESS =
+        boolSetting("http.tcp.reuse_address", NetworkService.TCP_REUSE_ADDRESS, Setting.Property.NodeScope);
+    public static final Setting<ByteSizeValue> SETTING_HTTP_TCP_SEND_BUFFER_SIZE =
+        Setting.byteSizeSetting("http.tcp.send_buffer_size", NetworkService.TCP_SEND_BUFFER_SIZE, Setting.Property.NodeScope);
+    public static final Setting<ByteSizeValue> SETTING_HTTP_TCP_RECEIVE_BUFFER_SIZE =
+        Setting.byteSizeSetting("http.tcp.receive_buffer_size", NetworkService.TCP_RECEIVE_BUFFER_SIZE, Setting.Property.NodeScope);
+
     private HttpTransportSettings() {
     }
 }

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpRequest.java

@@ -39,7 +39,7 @@
 import java.util.Set;
 import java.util.stream.Collectors;
 
-public class Netty4HttpRequest extends RestRequest {
+class Netty4HttpRequest extends RestRequest {
 
     private final FullHttpRequest request;
     private final Channel channel;

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpServerTransport.java

@@ -108,6 +108,11 @@
 import static org.elasticsearch.http.HttpTransportSettings.SETTING_HTTP_PUBLISH_HOST;
 import static org.elasticsearch.http.HttpTransportSettings.SETTING_HTTP_PUBLISH_PORT;
 import static org.elasticsearch.http.HttpTransportSettings.SETTING_HTTP_RESET_COOKIES;
+import static org.elasticsearch.http.HttpTransportSettings.SETTING_HTTP_TCP_KEEP_ALIVE;
+import static org.elasticsearch.http.HttpTransportSettings.SETTING_HTTP_TCP_NO_DELAY;
+import static org.elasticsearch.http.HttpTransportSettings.SETTING_HTTP_TCP_RECEIVE_BUFFER_SIZE;
+import static org.elasticsearch.http.HttpTransportSettings.SETTING_HTTP_TCP_REUSE_ADDRESS;
+import static org.elasticsearch.http.HttpTransportSettings.SETTING_HTTP_TCP_SEND_BUFFER_SIZE;
 import static org.elasticsearch.http.HttpTransportSettings.SETTING_PIPELINING;
 import static org.elasticsearch.http.HttpTransportSettings.SETTING_PIPELINING_MAX_EVENTS;
 import static org.elasticsearch.http.netty4.cors.Netty4CorsHandler.ANY_ORIGIN;
@@ -125,17 +130,6 @@ public class Netty4HttpServerTransport extends AbstractLifecycleComponent implem
         (s) -> Integer.toString(EsExecutors.numberOfProcessors(s) * 2),
         (s) -> Setting.parseInt(s, 1, "http.netty.worker_count"), Property.NodeScope);
 
-    public static final Setting<Boolean> SETTING_HTTP_TCP_NO_DELAY =
-        boolSetting("http.tcp_no_delay", NetworkService.TCP_NO_DELAY, Property.NodeScope);
-    public static final Setting<Boolean> SETTING_HTTP_TCP_KEEP_ALIVE =
-        boolSetting("http.tcp.keep_alive", NetworkService.TCP_KEEP_ALIVE, Property.NodeScope);
-    public static final Setting<Boolean> SETTING_HTTP_TCP_REUSE_ADDRESS =
-        boolSetting("http.tcp.reuse_address", NetworkService.TCP_REUSE_ADDRESS, Property.NodeScope);
-
-    public static final Setting<ByteSizeValue> SETTING_HTTP_TCP_SEND_BUFFER_SIZE =
-        Setting.byteSizeSetting("http.tcp.send_buffer_size", NetworkService.TCP_SEND_BUFFER_SIZE, Property.NodeScope);
-    public static final Setting<ByteSizeValue> SETTING_HTTP_TCP_RECEIVE_BUFFER_SIZE =
-        Setting.byteSizeSetting("http.tcp.receive_buffer_size", NetworkService.TCP_RECEIVE_BUFFER_SIZE, Property.NodeScope);
     public static final Setting<ByteSizeValue> SETTING_HTTP_NETTY_RECEIVE_PREDICTOR_SIZE =
         Setting.byteSizeSetting("http.netty.receive_predictor_size", new ByteSizeValue(64, ByteSizeUnit.KB), Property.NodeScope);
     public static final Setting<ByteSizeValue> SETTING_HTTP_NETTY_RECEIVE_PREDICTOR_MIN =
@@ -249,7 +243,7 @@ public Netty4HttpServerTransport(Settings settings, NetworkService networkServic
         this.compressionLevel = SETTING_HTTP_COMPRESSION_LEVEL.get(settings);
         this.pipelining = SETTING_PIPELINING.get(settings);
         this.pipeliningMaxEvents = SETTING_PIPELINING_MAX_EVENTS.get(settings);
-        this.corsConfig = buildCorsConfig(settings);
+        this.corsConfig = Netty4CorsConfig.buildCorsConfig(settings);
 
         // validate max content length
         if (maxContentLength.getBytes() > Integer.MAX_VALUE) {
@@ -373,40 +367,6 @@ static int resolvePublishPort(Settings settings, List<TransportAddress> boundAdd
         return publishPort;
     }
 
-    // package private for testing
-    static Netty4CorsConfig buildCorsConfig(Settings settings) {
-        if (SETTING_CORS_ENABLED.get(settings) == false) {
-            return Netty4CorsConfigBuilder.forOrigins().disable().build();
-        }
-        String origin = SETTING_CORS_ALLOW_ORIGIN.get(settings);
-        final Netty4CorsConfigBuilder builder;
-        if (Strings.isNullOrEmpty(origin)) {
-            builder = Netty4CorsConfigBuilder.forOrigins();
-        } else if (origin.equals(ANY_ORIGIN)) {
-            builder = Netty4CorsConfigBuilder.forAnyOrigin();
-        } else {
-            Pattern p = RestUtils.checkCorsSettingForRegex(origin);
-            if (p == null) {
-                builder = Netty4CorsConfigBuilder.forOrigins(RestUtils.corsSettingAsArray(origin));
-            } else {
-                builder = Netty4CorsConfigBuilder.forPattern(p);
-            }
-        }
-        if (SETTING_CORS_ALLOW_CREDENTIALS.get(settings)) {
-            builder.allowCredentials();
-        }
-        String[] strMethods = Strings.tokenizeToStringArray(SETTING_CORS_ALLOW_METHODS.get(settings), ",");
-        HttpMethod[] methods = Arrays.asList(strMethods)
-            .stream()
-            .map(HttpMethod::valueOf)
-            .toArray(size -> new HttpMethod[size]);
-        return builder.allowedRequestMethods(methods)
-            .maxAge(SETTING_CORS_MAX_AGE.get(settings))
-            .allowedRequestHeaders(Strings.tokenizeToStringArray(SETTING_CORS_ALLOW_HEADERS.get(settings), ","))
-            .shortCircuit()
-            .build();
-    }
-
     private TransportAddress bindAddress(final InetAddress hostAddress) {
         final AtomicReference<Exception> lastException = new AtomicReference<>();
         final AtomicReference<InetSocketAddress> boundSocket = new AtomicReference<>();

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/cors/Netty4CorsConfig.java

@@ -23,7 +23,11 @@
 import io.netty.handler.codec.http.EmptyHttpHeaders;
 import io.netty.handler.codec.http.HttpHeaders;
 import io.netty.handler.codec.http.HttpMethod;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.rest.RestUtils;
 
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.LinkedHashSet;
 import java.util.Map;
@@ -32,6 +36,14 @@
 import java.util.concurrent.Callable;
 import java.util.regex.Pattern;
 
+import static org.elasticsearch.http.HttpTransportSettings.SETTING_CORS_ALLOW_CREDENTIALS;
+import static org.elasticsearch.http.HttpTransportSettings.SETTING_CORS_ALLOW_HEADERS;
+import static org.elasticsearch.http.HttpTransportSettings.SETTING_CORS_ALLOW_METHODS;
+import static org.elasticsearch.http.HttpTransportSettings.SETTING_CORS_ALLOW_ORIGIN;
+import static org.elasticsearch.http.HttpTransportSettings.SETTING_CORS_ENABLED;
+import static org.elasticsearch.http.HttpTransportSettings.SETTING_CORS_MAX_AGE;
+import static org.elasticsearch.http.netty4.cors.Netty4CorsHandler.ANY_ORIGIN;
+
 /**
  * Configuration for Cross-Origin Resource Sharing (CORS).
  *
@@ -232,4 +244,37 @@ public String toString() {
             ", preflightHeaders=" + preflightHeaders + ']';
     }
 
+    public static Netty4CorsConfig buildCorsConfig(Settings settings) {
+        if (SETTING_CORS_ENABLED.get(settings) == false) {
+            return Netty4CorsConfigBuilder.forOrigins().disable().build();
+        }
+        String origin = SETTING_CORS_ALLOW_ORIGIN.get(settings);
+        final Netty4CorsConfigBuilder builder;
+        if (Strings.isNullOrEmpty(origin)) {
+            builder = Netty4CorsConfigBuilder.forOrigins();
+        } else if (origin.equals(ANY_ORIGIN)) {
+            builder = Netty4CorsConfigBuilder.forAnyOrigin();
+        } else {
+            Pattern p = RestUtils.checkCorsSettingForRegex(origin);
+            if (p == null) {
+                builder = Netty4CorsConfigBuilder.forOrigins(RestUtils.corsSettingAsArray(origin));
+            } else {
+                builder = Netty4CorsConfigBuilder.forPattern(p);
+            }
+        }
+        if (SETTING_CORS_ALLOW_CREDENTIALS.get(settings)) {
+            builder.allowCredentials();
+        }
+        String[] strMethods = Strings.tokenizeToStringArray(SETTING_CORS_ALLOW_METHODS.get(settings), ",");
+        HttpMethod[] methods = Arrays.asList(strMethods)
+            .stream()
+            .map(HttpMethod::valueOf)
+            .toArray(size -> new HttpMethod[size]);
+        return builder.allowedRequestMethods(methods)
+            .maxAge(SETTING_CORS_MAX_AGE.get(settings))
+            .allowedRequestHeaders(Strings.tokenizeToStringArray(SETTING_CORS_ALLOW_HEADERS.get(settings), ","))
+            .shortCircuit()
+            .build();
+    }
+
 }

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/pipelining/HttpPipelinedResponse.java

@@ -23,7 +23,7 @@
 import io.netty.handler.codec.http.FullHttpResponse;
 import io.netty.util.ReferenceCounted;
 
-class HttpPipelinedResponse implements Comparable<HttpPipelinedResponse>, ReferenceCounted {
+public class HttpPipelinedResponse implements Comparable<HttpPipelinedResponse>, ReferenceCounted {
 
     private final FullHttpResponse response;
     private final ChannelPromise promise;

modules/transport-netty4/src/main/java/org/elasticsearch/transport/Netty4Plugin.java

@@ -55,11 +55,6 @@ public List<Setting<?>> getSettings() {
         return Arrays.asList(
             Netty4HttpServerTransport.SETTING_HTTP_NETTY_MAX_COMPOSITE_BUFFER_COMPONENTS,
             Netty4HttpServerTransport.SETTING_HTTP_WORKER_COUNT,
-            Netty4HttpServerTransport.SETTING_HTTP_TCP_NO_DELAY,
-            Netty4HttpServerTransport.SETTING_HTTP_TCP_KEEP_ALIVE,
-            Netty4HttpServerTransport.SETTING_HTTP_TCP_REUSE_ADDRESS,
-            Netty4HttpServerTransport.SETTING_HTTP_TCP_SEND_BUFFER_SIZE,
-            Netty4HttpServerTransport.SETTING_HTTP_TCP_RECEIVE_BUFFER_SIZE,
             Netty4HttpServerTransport.SETTING_HTTP_NETTY_RECEIVE_PREDICTOR_SIZE,
             Netty4HttpServerTransport.SETTING_HTTP_NETTY_RECEIVE_PREDICTOR_MIN,
             Netty4HttpServerTransport.SETTING_HTTP_NETTY_RECEIVE_PREDICTOR_MAX,

modules/transport-netty4/src/main/java/org/elasticsearch/transport/netty4/ByteBufBytesReference.java

@@ -27,7 +27,7 @@
 import java.io.OutputStream;
 import java.nio.charset.StandardCharsets;
 
-final class ByteBufBytesReference extends BytesReference {
+public final class ByteBufBytesReference extends BytesReference {
 
     private final ByteBuf buffer;
     private final int length;
@@ -65,7 +65,7 @@ public void writeTo(OutputStream os) throws IOException {
         buffer.getBytes(offset, os, length);
     }
 
-    ByteBuf toByteBuf() {
+    public ByteBuf toByteBuf() {
         return buffer.duplicate();
     }
 

modules/transport-netty4/src/test/java/org/elasticsearch/http/netty4/Netty4HttpServerTransportTests.java

@@ -117,7 +117,7 @@ public void testCorsConfig() {
                                       .put(SETTING_CORS_ALLOW_HEADERS.getKey(), collectionToDelimitedString(headers, ",", prefix, ""))
                                       .put(SETTING_CORS_ALLOW_CREDENTIALS.getKey(), true)
                                       .build();
-        final Netty4CorsConfig corsConfig = Netty4HttpServerTransport.buildCorsConfig(settings);
+        final Netty4CorsConfig corsConfig = Netty4CorsConfig.buildCorsConfig(settings);
         assertTrue(corsConfig.isAnyOriginSupported());
         assertEquals(headers, corsConfig.allowedRequestHeaders());
         assertEquals(methods, corsConfig.allowedRequestMethods().stream().map(HttpMethod::name).collect(Collectors.toSet()));
@@ -128,7 +128,7 @@ public void testCorsConfigWithDefaults() {
         final Set<String> headers = Strings.commaDelimitedListToSet(SETTING_CORS_ALLOW_HEADERS.getDefault(Settings.EMPTY));
         final long maxAge = SETTING_CORS_MAX_AGE.getDefault(Settings.EMPTY);
         final Settings settings = Settings.builder().put(SETTING_CORS_ENABLED.getKey(), true).build();
-        final Netty4CorsConfig corsConfig = Netty4HttpServerTransport.buildCorsConfig(settings);
+        final Netty4CorsConfig corsConfig = Netty4CorsConfig.buildCorsConfig(settings);
         assertFalse(corsConfig.isAnyOriginSupported());
         assertEquals(Collections.emptySet(), corsConfig.origins().get());
         assertEquals(headers, corsConfig.allowedRequestHeaders());

plugins/transport-nio/build.gradle

@@ -0,0 +1,33 @@
+/*
+ * 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.
+ */
+
+esplugin {
+  description 'The nio transport.'
+  classname 'org.elasticsearch.transport.nio.NioPlugin'
+}
+
+dependencyLicenses.enabled = false
+
+compileJava.options.compilerArgs << "-Xlint:-try"
+compileTestJava.options.compilerArgs << "-Xlint:-rawtypes,-unchecked"
+
+dependencies {
+  compile project(path: ':modules:transport-netty4', configuration: 'runtime')
+  compile project(path: ':test:framework', configuration: 'runtime')
+}

plugins/transport-nio/src/main/java/org/elasticsearch/transport/nio/NioPlugin.java

@@ -0,0 +1,85 @@
+/*
+ * 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.nio;
+
+import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
+import org.elasticsearch.common.network.NetworkModule;
+import org.elasticsearch.common.network.NetworkService;
+import org.elasticsearch.common.settings.Setting;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.BigArrays;
+import org.elasticsearch.common.xcontent.NamedXContentRegistry;
+import org.elasticsearch.http.HttpServerTransport;
+import org.elasticsearch.transport.nio.http.NioHttpTransport;
+import org.elasticsearch.indices.breaker.CircuitBreakerService;
+import org.elasticsearch.plugins.NetworkPlugin;
+import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.Transport;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Supplier;
+
+public class NioPlugin extends Plugin implements NetworkPlugin {
+
+    public static final String NIO_TRANSPORT_NAME = "nio-transport";
+    public static final String NIO_HTTP_TRANSPORT_NAME = "nio-http-transport";
+
+    @Override
+    public List<Setting<?>> getSettings() {
+        return Arrays.asList(
+            NioHttpTransport.NIO_HTTP_WORKER_COUNT,
+            NioHttpTransport.NIO_HTTP_ACCEPTOR_COUNT,
+            NioTransport.NIO_WORKER_COUNT,
+            NioTransport.NIO_ACCEPTOR_COUNT
+        );
+    }
+
+    @Override
+    public Settings additionalSettings() {
+        final Settings.Builder settingsBuilder = Settings.builder();
+        settingsBuilder.put(NetworkModule.TRANSPORT_TYPE_KEY, NIO_TRANSPORT_NAME);
+        settingsBuilder.put(NetworkModule.HTTP_TYPE_KEY, NIO_HTTP_TRANSPORT_NAME);
+        return settingsBuilder.build();
+    }
+
+    @Override
+    public Map<String, Supplier<Transport>> getTransports(Settings settings, ThreadPool threadPool, BigArrays bigArrays,
+                                                          CircuitBreakerService circuitBreakerService,
+                                                          NamedWriteableRegistry namedWriteableRegistry,
+                                                          NetworkService networkService) {
+        return Collections.singletonMap(NIO_TRANSPORT_NAME, () -> new NioTransport(settings, threadPool, networkService, bigArrays,
+            namedWriteableRegistry, circuitBreakerService));
+    }
+
+    @Override
+    public Map<String, Supplier<HttpServerTransport>> getHttpTransports(Settings settings, ThreadPool threadPool, BigArrays bigArrays,
+                                                                        CircuitBreakerService circuitBreakerService,
+                                                                        NamedWriteableRegistry namedWriteableRegistry,
+                                                                        NamedXContentRegistry xContentRegistry,
+                                                                        NetworkService networkService,
+                                                                        HttpServerTransport.Dispatcher dispatcher) {
+        return Collections.singletonMap(NIO_HTTP_TRANSPORT_NAME,
+            () -> new NioHttpTransport(settings, networkService, bigArrays, threadPool, xContentRegistry, dispatcher));
+    }
+}