Skip to content

Commit 1387999

Browse files
Tim-BrooksTim-Brooks
committed
Share netty event loops between transports (elastic#46346)
Currently Elasticsearch creates independent event loop groups for each transport (http and internal) transport type. This is unnecessary and can lead to contention when different threads access shared resources (ex: allocators). This commit moves to a model where, by default, the event loops are shared between the transports. The previous behavior can be attained by specifically setting the http worker count.
1 parent 659edb9 commit 1387999

File tree

18 files changed

+342
-76
lines changed

18 files changed

+342
-76
lines changed

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

Lines changed: 12 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,6 @@
2929
import io.netty.channel.ChannelOption;
3030
import io.netty.channel.FixedRecvByteBufAllocator;
3131
import io.netty.channel.RecvByteBufAllocator;
32-
import io.netty.channel.nio.NioEventLoopGroup;
3332
import io.netty.channel.socket.nio.NioChannelOption;
3433
import io.netty.handler.codec.ByteToMessageDecoder;
3534
import io.netty.handler.codec.http.HttpContentCompressor;
@@ -62,14 +61,14 @@
6261
import org.elasticsearch.http.HttpServerChannel;
6362
import org.elasticsearch.http.netty4.cors.Netty4CorsHandler;
6463
import org.elasticsearch.threadpool.ThreadPool;
64+
import org.elasticsearch.transport.SharedGroupFactory;
6565
import org.elasticsearch.transport.NettyAllocator;
6666
import org.elasticsearch.transport.netty4.Netty4Utils;
6767

6868
import java.net.InetSocketAddress;
6969
import java.net.SocketOption;
7070
import java.util.concurrent.TimeUnit;
7171

72-
import static org.elasticsearch.common.util.concurrent.EsExecutors.daemonThreadFactory;
7372
import static org.elasticsearch.http.HttpTransportSettings.SETTING_HTTP_MAX_CHUNK_SIZE;
7473
import static org.elasticsearch.http.HttpTransportSettings.SETTING_HTTP_MAX_CONTENT_LENGTH;
7574
import static org.elasticsearch.http.HttpTransportSettings.SETTING_HTTP_MAX_HEADER_SIZE;
@@ -126,9 +125,7 @@ public class Netty4HttpServerTransport extends AbstractHttpServerTransport {
126125
// Netty's CompositeByteBuf implementation does not allow less than two components.
127126
}, s -> Setting.parseInt(s, 2, Integer.MAX_VALUE, SETTING_KEY_HTTP_NETTY_MAX_COMPOSITE_BUFFER_COMPONENTS), Property.NodeScope);
128127

129-
public static final Setting<Integer> SETTING_HTTP_WORKER_COUNT = new Setting<>("http.netty.worker_count",
130-
(s) -> Integer.toString(EsExecutors.allocatedProcessors(s) * 2),
131-
(s) -> Setting.parseInt(s, 1, "http.netty.worker_count"), Property.NodeScope);
128+
public static final Setting<Integer> SETTING_HTTP_WORKER_COUNT = Setting.intSetting("http.netty.worker_count", 0, Property.NodeScope);
132129

133130
public static final Setting<ByteSizeValue> SETTING_HTTP_NETTY_RECEIVE_PREDICTOR_SIZE =
134131
Setting.byteSizeSetting("http.netty.receive_predictor_size", new ByteSizeValue(64, ByteSizeUnit.KB), Property.NodeScope);
@@ -137,29 +134,30 @@ public class Netty4HttpServerTransport extends AbstractHttpServerTransport {
137134
private final ByteSizeValue maxHeaderSize;
138135
private final ByteSizeValue maxChunkSize;
139136

140-
private final int workerCount;
141-
142137
private final int pipeliningMaxEvents;
143138

139+
private final SharedGroupFactory sharedGroupFactory;
144140
private final RecvByteBufAllocator recvByteBufAllocator;
145141
private final int readTimeoutMillis;
146142

147143
private final int maxCompositeBufferComponents;
148144

149145
private volatile ServerBootstrap serverBootstrap;
146+
private volatile SharedGroupFactory.SharedGroup sharedGroup;
150147

151148
public Netty4HttpServerTransport(Settings settings, NetworkService networkService, BigArrays bigArrays, ThreadPool threadPool,
152-
NamedXContentRegistry xContentRegistry, Dispatcher dispatcher, ClusterSettings clusterSettings) {
149+
NamedXContentRegistry xContentRegistry, Dispatcher dispatcher, ClusterSettings clusterSettings,
150+
SharedGroupFactory sharedGroupFactory) {
153151
super(settings, networkService, bigArrays, threadPool, xContentRegistry, dispatcher, clusterSettings);
154152
Netty4Utils.setAvailableProcessors(EsExecutors.NODE_PROCESSORS_SETTING.get(settings));
153+
this.sharedGroupFactory = sharedGroupFactory;
155154

156155
this.maxChunkSize = SETTING_HTTP_MAX_CHUNK_SIZE.get(settings);
157156
this.maxHeaderSize = SETTING_HTTP_MAX_HEADER_SIZE.get(settings);
158157
this.maxInitialLineLength = SETTING_HTTP_MAX_INITIAL_LINE_LENGTH.get(settings);
159158
this.pipeliningMaxEvents = SETTING_PIPELINING_MAX_EVENTS.get(settings);
160159

161160
this.maxCompositeBufferComponents = SETTING_HTTP_NETTY_MAX_COMPOSITE_BUFFER_COMPONENTS.get(settings);
162-
this.workerCount = SETTING_HTTP_WORKER_COUNT.get(settings);
163161

164162
this.readTimeoutMillis = Math.toIntExact(SETTING_HTTP_READ_TIMEOUT.get(settings).getMillis());
165163

@@ -180,10 +178,10 @@ public Settings settings() {
180178
protected void doStart() {
181179
boolean success = false;
182180
try {
181+
sharedGroup = sharedGroupFactory.getHttpGroup();
183182
serverBootstrap = new ServerBootstrap();
184183

185-
serverBootstrap.group(new NioEventLoopGroup(workerCount, daemonThreadFactory(settings,
186-
HTTP_SERVER_WORKER_THREAD_NAME_PREFIX)));
184+
serverBootstrap.group(sharedGroup.getLowLevelGroup());
187185

188186
// NettyAllocator will return the channel type designed to work with the configuredAllocator
189187
serverBootstrap.channel(NettyAllocator.getServerChannelType());
@@ -260,9 +258,9 @@ protected HttpServerChannel bind(InetSocketAddress socketAddress) throws Excepti
260258

261259
@Override
262260
protected void stopInternal() {
263-
if (serverBootstrap != null) {
264-
serverBootstrap.config().group().shutdownGracefully(0, 5, TimeUnit.SECONDS).awaitUninterruptibly();
265-
serverBootstrap = null;
261+
if (sharedGroup != null) {
262+
sharedGroup.shutdown();
263+
sharedGroup = null;
266264
}
267265
}
268266

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

Lines changed: 16 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919

2020
package org.elasticsearch.transport;
2121

22+
import org.apache.lucene.util.SetOnce;
2223
import org.elasticsearch.Version;
2324
import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
2425
import org.elasticsearch.common.network.NetworkModule;
@@ -48,6 +49,8 @@ public class Netty4Plugin extends Plugin implements NetworkPlugin {
4849
public static final String NETTY_TRANSPORT_NAME = "netty4";
4950
public static final String NETTY_HTTP_TRANSPORT_NAME = "netty4";
5051

52+
private final SetOnce<SharedGroupFactory> groupFactory = new SetOnce<>();
53+
5154
@Override
5255
public List<Setting<?>> getSettings() {
5356
return Arrays.asList(
@@ -77,7 +80,7 @@ public Map<String, Supplier<Transport>> getTransports(Settings settings, ThreadP
7780
CircuitBreakerService circuitBreakerService,
7881
NamedWriteableRegistry namedWriteableRegistry, NetworkService networkService) {
7982
return Collections.singletonMap(NETTY_TRANSPORT_NAME, () -> new Netty4Transport(settings, Version.CURRENT, threadPool,
80-
networkService, pageCacheRecycler, namedWriteableRegistry, circuitBreakerService));
83+
networkService, pageCacheRecycler, namedWriteableRegistry, circuitBreakerService, getSharedGroupFactory(settings)));
8184
}
8285

8386
@Override
@@ -90,6 +93,17 @@ public Map<String, Supplier<HttpServerTransport>> getHttpTransports(Settings set
9093
ClusterSettings clusterSettings) {
9194
return Collections.singletonMap(NETTY_HTTP_TRANSPORT_NAME,
9295
() -> new Netty4HttpServerTransport(settings, networkService, bigArrays, threadPool, xContentRegistry, dispatcher,
93-
clusterSettings));
96+
clusterSettings, getSharedGroupFactory(settings)));
97+
}
98+
99+
private SharedGroupFactory getSharedGroupFactory(Settings settings) {
100+
SharedGroupFactory groupFactory = this.groupFactory.get();
101+
if (groupFactory != null) {
102+
assert groupFactory.getSettings().equals(settings) : "Different settings than originally provided";
103+
return groupFactory;
104+
} else {
105+
this.groupFactory.set(new SharedGroupFactory(settings));
106+
return this.groupFactory.get();
107+
}
94108
}
95109
}

‎modules/transport-netty4/src/main/java/org/elasticsearch/transport/SharedGroupFactory.java‎

Lines changed: 141 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,141 @@
1+
/*
2+
* Licensed to Elasticsearch under one or more contributor
3+
* license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright
5+
* ownership. Elasticsearch licenses this file to you under
6+
* the Apache License, Version 2.0 (the "License"); you may
7+
* not use this file except in compliance with the License.
8+
* You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package org.elasticsearch.transport;
21+
22+
import io.netty.channel.EventLoopGroup;
23+
import io.netty.channel.nio.NioEventLoopGroup;
24+
import io.netty.util.concurrent.Future;
25+
import org.apache.logging.log4j.LogManager;
26+
import org.apache.logging.log4j.Logger;
27+
import org.elasticsearch.common.settings.Settings;
28+
import org.elasticsearch.common.util.concurrent.AbstractRefCounted;
29+
import org.elasticsearch.http.HttpServerTransport;
30+
import org.elasticsearch.http.netty4.Netty4HttpServerTransport;
31+
import org.elasticsearch.transport.netty4.Netty4Transport;
32+
33+
import java.util.concurrent.TimeUnit;
34+
import java.util.concurrent.atomic.AtomicBoolean;
35+
36+
import static org.elasticsearch.common.util.concurrent.EsExecutors.daemonThreadFactory;
37+
38+
/**
39+
* Creates and returns {@link io.netty.channel.EventLoopGroup} instances. It will return a shared group for
40+
* both {@link #getHttpGroup()} and {@link #getTransportGroup()} if
41+
* {@link org.elasticsearch.http.netty4.Netty4HttpServerTransport#SETTING_HTTP_WORKER_COUNT} is configured to be 0.
42+
* If that setting is not 0, then it will return a different group in the {@link #getHttpGroup()} call.
43+
*/
44+
public final class SharedGroupFactory {
45+
46+
private static final Logger logger = LogManager.getLogger(SharedGroupFactory.class);
47+
48+
private final Settings settings;
49+
private final int workerCount;
50+
private final int httpWorkerCount;
51+
52+
private RefCountedGroup genericGroup;
53+
private SharedGroup dedicatedHttpGroup;
54+
55+
public SharedGroupFactory(Settings settings) {
56+
this.settings = settings;
57+
this.workerCount = Netty4Transport.WORKER_COUNT.get(settings);
58+
this.httpWorkerCount = Netty4HttpServerTransport.SETTING_HTTP_WORKER_COUNT.get(settings);
59+
}
60+
61+
public Settings getSettings() {
62+
return settings;
63+
}
64+
65+
public int getTransportWorkerCount() {
66+
return workerCount;
67+
}
68+
69+
public synchronized SharedGroup getTransportGroup() {
70+
return getGenericGroup();
71+
}
72+
73+
public synchronized SharedGroup getHttpGroup() {
74+
if (httpWorkerCount == 0) {
75+
return getGenericGroup();
76+
} else {
77+
if (dedicatedHttpGroup == null) {
78+
NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup(httpWorkerCount,
79+
daemonThreadFactory(settings, HttpServerTransport.HTTP_SERVER_WORKER_THREAD_NAME_PREFIX));
80+
dedicatedHttpGroup = new SharedGroup(new RefCountedGroup(eventLoopGroup));
81+
}
82+
return dedicatedHttpGroup;
83+
}
84+
}
85+
86+
private SharedGroup getGenericGroup() {
87+
if (genericGroup == null) {
88+
EventLoopGroup eventLoopGroup = new NioEventLoopGroup(workerCount,
89+
daemonThreadFactory(settings, TcpTransport.TRANSPORT_WORKER_THREAD_NAME_PREFIX));
90+
this.genericGroup = new RefCountedGroup(eventLoopGroup);
91+
} else {
92+
genericGroup.incRef();
93+
}
94+
return new SharedGroup(genericGroup);
95+
}
96+
97+
private static class RefCountedGroup extends AbstractRefCounted {
98+
99+
public static final String NAME = "ref-counted-event-loop-group";
100+
private final EventLoopGroup eventLoopGroup;
101+
102+
private RefCountedGroup(EventLoopGroup eventLoopGroup) {
103+
super(NAME);
104+
this.eventLoopGroup = eventLoopGroup;
105+
}
106+
107+
@Override
108+
protected void closeInternal() {
109+
Future<?> shutdownFuture = eventLoopGroup.shutdownGracefully(0, 5, TimeUnit.SECONDS);
110+
shutdownFuture.awaitUninterruptibly();
111+
if (shutdownFuture.isSuccess() == false) {
112+
logger.warn("Error closing netty event loop group", shutdownFuture.cause());
113+
}
114+
}
115+
}
116+
117+
/**
118+
* Wraps the {@link RefCountedGroup}. Calls {@link RefCountedGroup#decRef()} on close. After close,
119+
* this wrapped instance can no longer be used.
120+
*/
121+
public static class SharedGroup {
122+
123+
private final RefCountedGroup refCountedGroup;
124+
125+
private final AtomicBoolean isOpen = new AtomicBoolean(true);
126+
127+
private SharedGroup(RefCountedGroup refCountedGroup) {
128+
this.refCountedGroup = refCountedGroup;
129+
}
130+
131+
public EventLoopGroup getLowLevelGroup() {
132+
return refCountedGroup.eventLoopGroup;
133+
}
134+
135+
public void shutdown() {
136+
if (isOpen.compareAndSet(true, false)) {
137+
refCountedGroup.decRef();
138+
}
139+
}
140+
}
141+
}

0 commit comments

Comments
 (0)