Implement byte array reusage in NioTransport (#27696)

This is related to #27563. This commit modifies the
InboundChannelBuffer to support releasable byte pages. These byte
pages are provided by the PageCacheRecycler. The PageCacheRecycler
must be passed to the Transport with this change.

core/src/main/java/org/elasticsearch/client/transport/TransportClient.java

@@ -43,6 +43,7 @@
 import org.elasticsearch.common.transport.TransportAddress;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.BigArrays;
+import org.elasticsearch.common.util.PageCacheRecycler;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.indices.breaker.CircuitBreakerService;
 import org.elasticsearch.node.InternalSettingsPreparer;
@@ -168,11 +169,12 @@ private static ClientTemplate buildTemplate(Settings providedSettings, Settings
             CircuitBreakerService circuitBreakerService = Node.createCircuitBreakerService(settingsModule.getSettings(),
                 settingsModule.getClusterSettings());
             resourcesToClose.add(circuitBreakerService);
-            BigArrays bigArrays = new BigArrays(settings, circuitBreakerService);
+            PageCacheRecycler pageCacheRecycler = new PageCacheRecycler(settings);
+            BigArrays bigArrays = new BigArrays(pageCacheRecycler, circuitBreakerService);
             resourcesToClose.add(bigArrays);
             modules.add(settingsModule);
             NetworkModule networkModule = new NetworkModule(settings, true, pluginsService.filterPlugins(NetworkPlugin.class), threadPool,
-                bigArrays, circuitBreakerService, namedWriteableRegistry, xContentRegistry, networkService, null);
+                bigArrays, pageCacheRecycler, circuitBreakerService, namedWriteableRegistry, xContentRegistry, networkService, null);
             final Transport transport = networkModule.getTransportSupplier().get();
             final TransportService transportService = new TransportService(settings, transport, threadPool,
                 networkModule.getTransportInterceptor(),

core/src/main/java/org/elasticsearch/common/network/NetworkModule.java

@@ -34,6 +34,7 @@
 import org.elasticsearch.common.settings.Setting.Property;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.BigArrays;
+import org.elasticsearch.common.util.PageCacheRecycler;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.http.HttpServerTransport;
@@ -107,6 +108,7 @@ public final class NetworkModule {
      */
     public NetworkModule(Settings settings, boolean transportClient, List<NetworkPlugin> plugins, ThreadPool threadPool,
                          BigArrays bigArrays,
+                         PageCacheRecycler pageCacheRecycler,
                          CircuitBreakerService circuitBreakerService,
                          NamedWriteableRegistry namedWriteableRegistry,
                          NamedXContentRegistry xContentRegistry,
@@ -121,9 +123,9 @@ public NetworkModule(Settings settings, boolean transportClient, List<NetworkPlu
                     registerHttpTransport(entry.getKey(), entry.getValue());
                 }
             }
-            Map<String, Supplier<Transport>> httpTransportFactory = plugin.getTransports(settings, threadPool, bigArrays,
+            Map<String, Supplier<Transport>> transportFactory = plugin.getTransports(settings, threadPool, bigArrays, pageCacheRecycler,
                 circuitBreakerService, namedWriteableRegistry, networkService);
-            for (Map.Entry<String, Supplier<Transport>> entry : httpTransportFactory.entrySet()) {
+            for (Map.Entry<String, Supplier<Transport>> entry : transportFactory.entrySet()) {
                 registerTransport(entry.getKey(), entry.getValue());
             }
             List<TransportInterceptor> transportInterceptors = plugin.getTransportInterceptors(namedWriteableRegistry,

core/src/main/java/org/elasticsearch/common/util/BigArrays.java

@@ -372,9 +372,9 @@ public T set(long index, T value) {
     final boolean checkBreaker;
     private final BigArrays circuitBreakingInstance;
 
-    public BigArrays(Settings settings, @Nullable final CircuitBreakerService breakerService) {
+    public BigArrays(PageCacheRecycler recycler, @Nullable final CircuitBreakerService breakerService) {
         // Checking the breaker is disabled if not specified
-        this(new PageCacheRecycler(settings), breakerService, false);
+        this(recycler, breakerService, false);
     }
 
     // public for tests

core/src/main/java/org/elasticsearch/common/util/PageCacheRecycler.java

@@ -65,10 +65,10 @@ public void close() {
         Releasables.close(true, bytePage, intPage, longPage, objectPage);
     }
 
-    protected PageCacheRecycler(Settings settings) {
+    public PageCacheRecycler(Settings settings) {
         super(settings);
-        final Type type = TYPE_SETTING .get(settings);
-        final long limit = LIMIT_HEAP_SETTING .get(settings).getBytes();
+        final Type type = TYPE_SETTING.get(settings);
+        final long limit = LIMIT_HEAP_SETTING.get(settings).getBytes();
         final int availableProcessors = EsExecutors.numberOfProcessors(settings);
 
         // We have a global amount of memory that we need to divide across data types.

core/src/main/java/org/elasticsearch/node/Node.java

@@ -80,6 +80,7 @@
 import org.elasticsearch.common.transport.TransportAddress;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.BigArrays;
+import org.elasticsearch.common.util.PageCacheRecycler;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.discovery.Discovery;
 import org.elasticsearch.discovery.DiscoveryModule;
@@ -364,7 +365,8 @@ protected Node(final Environment environment, Collection<Class<? extends Plugin>
             modules.add(new GatewayModule());
 
 
-            BigArrays bigArrays = createBigArrays(settings, circuitBreakerService);
+            PageCacheRecycler pageCacheRecycler = createPageCacheRecycler(settings);
+            BigArrays bigArrays = createBigArrays(pageCacheRecycler, circuitBreakerService);
             resourcesToClose.add(bigArrays);
             modules.add(settingsModule);
             List<NamedWriteableRegistry.Entry> namedWriteables = Stream.of(
@@ -404,7 +406,8 @@ protected Node(final Environment environment, Collection<Class<? extends Plugin>
 
             final RestController restController = actionModule.getRestController();
             final NetworkModule networkModule = new NetworkModule(settings, false, pluginsService.filterPlugins(NetworkPlugin.class),
-                    threadPool, bigArrays, circuitBreakerService, namedWriteableRegistry, xContentRegistry, networkService, restController);
+                threadPool, bigArrays, pageCacheRecycler, circuitBreakerService, namedWriteableRegistry, xContentRegistry,
+                networkService, restController);
             Collection<UnaryOperator<Map<String, MetaData.Custom>>> customMetaDataUpgraders =
                 pluginsService.filterPlugins(Plugin.class).stream()
                     .map(Plugin::getCustomMetaDataUpgrader)
@@ -896,8 +899,16 @@ public static CircuitBreakerService createCircuitBreakerService(Settings setting
      * Creates a new {@link BigArrays} instance used for this node.
      * This method can be overwritten by subclasses to change their {@link BigArrays} implementation for instance for testing
      */
-    BigArrays createBigArrays(Settings settings, CircuitBreakerService circuitBreakerService) {
-        return new BigArrays(settings, circuitBreakerService);
+    BigArrays createBigArrays(PageCacheRecycler pageCacheRecycler, CircuitBreakerService circuitBreakerService) {
+        return new BigArrays(pageCacheRecycler, circuitBreakerService);
+    }
+
+    /**
+     * Creates a new {@link BigArrays} instance used for this node.
+     * This method can be overwritten by subclasses to change their {@link BigArrays} implementation for instance for testing
+     */
+    PageCacheRecycler createPageCacheRecycler(Settings settings) {
+        return new PageCacheRecycler(settings);
     }
 
     /**

core/src/main/java/org/elasticsearch/plugins/NetworkPlugin.java

@@ -27,6 +27,7 @@
 import org.elasticsearch.common.network.NetworkService;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.BigArrays;
+import org.elasticsearch.common.util.PageCacheRecycler;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.http.HttpServerTransport;
@@ -58,6 +59,7 @@ default List<TransportInterceptor> getTransportInterceptors(NamedWriteableRegist
      * See {@link org.elasticsearch.common.network.NetworkModule#TRANSPORT_TYPE_KEY} to configure a specific implementation.
      */
     default Map<String, Supplier<Transport>> getTransports(Settings settings, ThreadPool threadPool, BigArrays bigArrays,
+                                                           PageCacheRecycler pageCacheRecycler,
                                                            CircuitBreakerService circuitBreakerService,
                                                            NamedWriteableRegistry namedWriteableRegistry,
                                                            NetworkService networkService) {

core/src/test/java/org/elasticsearch/common/io/stream/ReleasableBytesStreamOutputTests.java

@@ -21,6 +21,7 @@
 
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.MockBigArrays;
+import org.elasticsearch.common.util.MockPageCacheRecycler;
 import org.elasticsearch.indices.breaker.NoneCircuitBreakerService;
 import org.elasticsearch.test.ESTestCase;
 
@@ -30,7 +31,7 @@ public class ReleasableBytesStreamOutputTests extends ESTestCase {
 
     public void testRelease() throws Exception {
         MockBigArrays mockBigArrays =
-            new MockBigArrays(Settings.EMPTY, new NoneCircuitBreakerService());
+            new MockBigArrays(new MockPageCacheRecycler(Settings.EMPTY), new NoneCircuitBreakerService());
         try (ReleasableBytesStreamOutput output =
                  getRandomReleasableBytesStreamOutput(mockBigArrays)) {
             output.writeBoolean(randomBoolean());

core/src/test/java/org/elasticsearch/common/network/NetworkModuleTests.java

@@ -27,6 +27,7 @@
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.transport.BoundTransportAddress;
 import org.elasticsearch.common.util.BigArrays;
+import org.elasticsearch.common.util.PageCacheRecycler;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.http.HttpInfo;
@@ -133,6 +134,7 @@ public void testRegisterTransport() {
         NetworkPlugin plugin = new NetworkPlugin() {
             @Override
             public Map<String, Supplier<Transport>> getTransports(Settings settings, ThreadPool threadPool, BigArrays bigArrays,
+                                                                  PageCacheRecycler pageCacheRecycler,
                                                                   CircuitBreakerService circuitBreakerService,
                                                                   NamedWriteableRegistry namedWriteableRegistry,
                                                                   NetworkService networkService) {
@@ -193,6 +195,7 @@ public void testOverrideDefault() {
         NetworkModule module = newNetworkModule(settings, false, new NetworkPlugin() {
             @Override
             public Map<String, Supplier<Transport>> getTransports(Settings settings, ThreadPool threadPool, BigArrays bigArrays,
+                                                                  PageCacheRecycler pageCacheRecycler,
                                                                   CircuitBreakerService circuitBreakerService,
                                                                   NamedWriteableRegistry namedWriteableRegistry,
                                                                   NetworkService networkService) {
@@ -227,6 +230,7 @@ public void testDefaultKeys() {
         NetworkModule module = newNetworkModule(settings, false, new NetworkPlugin() {
             @Override
             public Map<String, Supplier<Transport>> getTransports(Settings settings, ThreadPool threadPool, BigArrays bigArrays,
+                                                                  PageCacheRecycler pageCacheRecycler,
                                                                   CircuitBreakerService circuitBreakerService,
                                                                   NamedWriteableRegistry namedWriteableRegistry,
                                                                   NetworkService networkService) {
@@ -306,7 +310,7 @@ public List<TransportInterceptor> getTransportInterceptors(NamedWriteableRegistr
     }
 
     private NetworkModule newNetworkModule(Settings settings, boolean transportClient, NetworkPlugin... plugins) {
-        return new NetworkModule(settings, transportClient, Arrays.asList(plugins), threadPool, null, null, null, xContentRegistry(), null,
-            new NullDispatcher());
+        return new NetworkModule(settings, transportClient, Arrays.asList(plugins), threadPool, null, null, null, null,
+            xContentRegistry(), null, new NullDispatcher());
     }
 }

core/src/test/java/org/elasticsearch/common/util/BigArraysTests.java

@@ -42,7 +42,7 @@
 public class BigArraysTests extends ESTestCase {
 
     private BigArrays randombigArrays() {
-        return new MockBigArrays(Settings.EMPTY, new NoneCircuitBreakerService());
+        return new MockBigArrays(new MockPageCacheRecycler(Settings.EMPTY), new NoneCircuitBreakerService());
     }
 
     private BigArrays bigArrays;

core/src/test/java/org/elasticsearch/common/util/BytesRefHashTests.java

@@ -41,7 +41,7 @@ public class BytesRefHashTests extends ESSingleNodeTestCase {
     BytesRefHash hash;
 
     private BigArrays randombigArrays() {
-        return new MockBigArrays(Settings.EMPTY, new NoneCircuitBreakerService());
+        return new MockBigArrays(new MockPageCacheRecycler(Settings.EMPTY), new NoneCircuitBreakerService());
     }
 
     private void newHash() {

core/src/test/java/org/elasticsearch/common/util/LongHashTests.java

@@ -36,7 +36,7 @@ public class LongHashTests extends ESSingleNodeTestCase {
     LongHash hash;
 
     private BigArrays randombigArrays() {
-        return new MockBigArrays(Settings.EMPTY, new NoneCircuitBreakerService());
+        return new MockBigArrays(new MockPageCacheRecycler(Settings.EMPTY), new NoneCircuitBreakerService());
     }
 
     private void newHash() {

core/src/test/java/org/elasticsearch/common/util/LongObjectHashMapTests.java

@@ -27,7 +27,7 @@
 public class LongObjectHashMapTests extends ESSingleNodeTestCase {
 
     private BigArrays randombigArrays() {
-        return new MockBigArrays(Settings.EMPTY, new NoneCircuitBreakerService());
+        return new MockBigArrays(new MockPageCacheRecycler(Settings.EMPTY), new NoneCircuitBreakerService());
     }
 
     public void testDuel() {

core/src/test/java/org/elasticsearch/index/IndexModuleTests.java

@@ -39,6 +39,7 @@
 import org.elasticsearch.common.settings.Setting.Property;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.BigArrays;
+import org.elasticsearch.common.util.PageCacheRecycler;
 import org.elasticsearch.env.Environment;
 import org.elasticsearch.env.NodeEnvironment;
 import org.elasticsearch.env.ShardLock;
@@ -124,7 +125,8 @@ public void setUp() throws Exception {
                 emptyMap(), emptyMap(), emptyMap());
         threadPool = new TestThreadPool("test");
         circuitBreakerService = new NoneCircuitBreakerService();
-        bigArrays = new BigArrays(settings, circuitBreakerService);
+        PageCacheRecycler pageCacheRecycler = new PageCacheRecycler(settings);
+        bigArrays = new BigArrays(pageCacheRecycler, circuitBreakerService);
         scriptService = new ScriptService(settings, Collections.emptyMap(), Collections.emptyMap());
         clusterService = ClusterServiceUtils.createClusterService(threadPool);
         nodeEnvironment = new NodeEnvironment(settings, environment);

core/src/test/java/org/elasticsearch/search/aggregations/MultiBucketAggregatorWrapperTests.java

@@ -25,6 +25,7 @@
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.common.util.MockBigArrays;
+import org.elasticsearch.common.util.MockPageCacheRecycler;
 import org.elasticsearch.indices.breaker.NoneCircuitBreakerService;
 import org.elasticsearch.search.internal.SearchContext;
 import org.elasticsearch.test.ESTestCase;
@@ -46,7 +47,7 @@ public class MultiBucketAggregatorWrapperTests extends ESTestCase {
     public void testNoNullScorerIsDelegated() throws Exception {
         LeafReaderContext leafReaderContext = MemoryIndex.fromDocument(Collections.emptyList(), new MockAnalyzer(random()))
                 .createSearcher().getIndexReader().leaves().get(0);
-        BigArrays bigArrays = new MockBigArrays(Settings.EMPTY, new NoneCircuitBreakerService());
+        BigArrays bigArrays = new MockBigArrays(new MockPageCacheRecycler(Settings.EMPTY), new NoneCircuitBreakerService());
         SearchContext searchContext = mock(SearchContext.class);
         when(searchContext.bigArrays()).thenReturn(bigArrays);
 

core/src/test/java/org/elasticsearch/search/aggregations/bucket/sampler/BestDocsDeferringCollectorTests.java

@@ -33,6 +33,7 @@
 import org.apache.lucene.store.Directory;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.MockBigArrays;
+import org.elasticsearch.common.util.MockPageCacheRecycler;
 import org.elasticsearch.indices.breaker.NoneCircuitBreakerService;
 import org.elasticsearch.search.aggregations.AggregatorTestCase;
 import org.elasticsearch.search.aggregations.BucketCollector;
@@ -63,8 +64,8 @@ public void testReplay() throws Exception {
         TermQuery termQuery = new TermQuery(new Term("field", String.valueOf(randomInt(maxNumValues))));
         TopDocs topDocs = indexSearcher.search(termQuery, numDocs);
 
-        BestDocsDeferringCollector collector =
-                new BestDocsDeferringCollector(numDocs, new MockBigArrays(Settings.EMPTY, new NoneCircuitBreakerService()));
+        BestDocsDeferringCollector collector = new BestDocsDeferringCollector(numDocs,
+            new MockBigArrays(new MockPageCacheRecycler(Settings.EMPTY), new NoneCircuitBreakerService()));
         Set<Integer> deferredCollectedDocIds = new HashSet<>();
         collector.setDeferredCollector(Collections.singleton(testCollector(deferredCollectedDocIds)));
         collector.preCollection();

core/src/test/java/org/elasticsearch/search/aggregations/bucket/terms/TermsAggregatorTests.java

@@ -38,6 +38,7 @@
 import org.elasticsearch.common.network.InetAddresses;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.MockBigArrays;
+import org.elasticsearch.common.util.MockPageCacheRecycler;
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.mapper.IpFieldMapper;
 import org.elasticsearch.index.mapper.KeywordFieldMapper;
@@ -913,7 +914,7 @@ public void testMixLongAndDouble() throws Exception {
                 dir.close();
             }
             InternalAggregation.ReduceContext ctx =
-                new InternalAggregation.ReduceContext(new MockBigArrays(Settings.EMPTY,
+                new InternalAggregation.ReduceContext(new MockBigArrays(new MockPageCacheRecycler(Settings.EMPTY),
                     new NoneCircuitBreakerService()), null, true);
             for (InternalAggregation internalAgg : aggs) {
                 InternalAggregation mergedAggs = internalAgg.doReduce(aggs, ctx);

core/src/test/java/org/elasticsearch/search/aggregations/metrics/cardinality/InternalCardinalityTests.java

@@ -25,6 +25,7 @@
 import org.elasticsearch.common.lease.Releasables;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.MockBigArrays;
+import org.elasticsearch.common.util.MockPageCacheRecycler;
 import org.elasticsearch.indices.breaker.NoneCircuitBreakerService;
 import org.elasticsearch.search.aggregations.ParsedAggregation;
 import org.elasticsearch.search.aggregations.pipeline.PipelineAggregator;
@@ -60,7 +61,7 @@ public void tearDown() throws Exception {
     protected InternalCardinality createTestInstance(String name,
             List<PipelineAggregator> pipelineAggregators, Map<String, Object> metaData) {
         HyperLogLogPlusPlus hllpp = new HyperLogLogPlusPlus(p,
-                new MockBigArrays(Settings.EMPTY, new NoneCircuitBreakerService()), 1);
+                new MockBigArrays(new MockPageCacheRecycler(Settings.EMPTY), new NoneCircuitBreakerService()), 1);
         algos.add(hllpp);
         for (int i = 0; i < 100; i++) {
             hllpp.collect(0, BitMixer.mix64(randomIntBetween(1, 100)));
@@ -107,7 +108,7 @@ protected InternalCardinality mutateInstance(InternalCardinality instance) {
             break;
         case 1:
             HyperLogLogPlusPlus newState = new HyperLogLogPlusPlus(state.precision(),
-                    new MockBigArrays(Settings.EMPTY, new NoneCircuitBreakerService()), 0);
+                    new MockBigArrays(new MockPageCacheRecycler(Settings.EMPTY), new NoneCircuitBreakerService()), 0);
             newState.merge(0, state, 0);
             int extraValues = between(10, 100);
             for (int i = 0; i < extraValues; i++) {