Skip to content

Commit aec44a0

Browse files
Tim-BrooksTim-Brooks
committed
Remove tcp profile from low level nio channel (#27441)
This is related to #27260. Currently every nio channel has a profile field. Profile is a concept that only relates to the tcp transport. Http channels will not have profiles. This commit moves the profile from the nio channel to the read context. The context is the level that protocol specific features and logic should live.
1 parent 01bac8d commit aec44a0

15 files changed

+79
-73
lines changed

‎test/framework/src/main/java/org/elasticsearch/transport/nio/NioTransport.java‎

Lines changed: 10 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -66,7 +66,6 @@ public class NioTransport extends TcpTransport {
6666
intSetting("transport.nio.acceptor_count", 1, 1, Setting.Property.NodeScope);
6767

6868
protected final OpenChannels openChannels = new OpenChannels(logger);
69-
private final Consumer<NioSocketChannel> contextSetter;
7069
private final ConcurrentMap<String, ChannelFactory> profileToChannelFactory = newConcurrentMap();
7170
private final ArrayList<AcceptingSelector> acceptors = new ArrayList<>();
7271
private final ArrayList<SocketSelector> socketSelectors = new ArrayList<>();
@@ -77,7 +76,6 @@ public class NioTransport extends TcpTransport {
7776
public NioTransport(Settings settings, ThreadPool threadPool, NetworkService networkService, BigArrays bigArrays,
7877
NamedWriteableRegistry namedWriteableRegistry, CircuitBreakerService circuitBreakerService) {
7978
super("nio", settings, threadPool, bigArrays, circuitBreakerService, namedWriteableRegistry, networkService);
80-
contextSetter = (c) -> c.setContexts(new TcpReadContext(c, new TcpReadHandler(this)), new TcpWriteContext(c));
8179
}
8280

8381
@Override
@@ -89,7 +87,7 @@ public long getNumOpenServerConnections() {
8987
protected NioServerSocketChannel bind(String name, InetSocketAddress address) throws IOException {
9088
ChannelFactory channelFactory = this.profileToChannelFactory.get(name);
9189
AcceptingSelector selector = acceptors.get(++acceptorNumber % NioTransport.NIO_ACCEPTOR_COUNT.get(settings));
92-
return channelFactory.openNioServerSocketChannel(name, address, selector);
90+
return channelFactory.openNioServerSocketChannel(address, selector);
9391
}
9492

9593
@Override
@@ -119,8 +117,9 @@ protected void doStart() {
119117
}
120118
}
121119

120+
Consumer<NioSocketChannel> clientContextSetter = getContextSetter("client-socket");
122121
clientSelectorSupplier = new RoundRobinSelectorSupplier(socketSelectors);
123-
clientChannelFactory = new ChannelFactory(new ProfileSettings(settings, "default"), contextSetter);
122+
clientChannelFactory = new ChannelFactory(new ProfileSettings(settings, "default"), clientContextSetter);
124123

125124
if (NetworkService.NETWORK_SERVER.get(settings)) {
126125
int acceptorCount = NioTransport.NIO_ACCEPTOR_COUNT.get(settings);
@@ -142,7 +141,9 @@ protected void doStart() {
142141

143142
// loop through all profiles and start them up, special handling for default one
144143
for (ProfileSettings profileSettings : profileSettings) {
145-
profileToChannelFactory.putIfAbsent(profileSettings.profileName, new ChannelFactory(profileSettings, contextSetter));
144+
String profileName = profileSettings.profileName;
145+
Consumer<NioSocketChannel> contextSetter = getContextSetter(profileName);
146+
profileToChannelFactory.putIfAbsent(profileName, new ChannelFactory(profileSettings, contextSetter));
146147
bindServer(profileSettings);
147148
}
148149
}
@@ -174,4 +175,8 @@ protected SocketEventHandler getSocketEventHandler() {
174175
final void exceptionCaught(NioSocketChannel channel, Exception exception) {
175176
onException(channel, exception);
176177
}
178+
179+
private Consumer<NioSocketChannel> getContextSetter(String profileName) {
180+
return (c) -> c.setContexts(new TcpReadContext(c, new TcpReadHandler(profileName,this)), new TcpWriteContext(c));
181+
}
177182
}

‎test/framework/src/main/java/org/elasticsearch/transport/nio/TcpReadHandler.java‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -26,15 +26,17 @@
2626

2727
public class TcpReadHandler {
2828

29+
private final String profile;
2930
private final NioTransport transport;
3031

31-
public TcpReadHandler(NioTransport transport) {
32+
public TcpReadHandler(String profile, NioTransport transport) {
33+
this.profile = profile;
3234
this.transport = transport;
3335
}
3436

3537
public void handleMessage(BytesReference reference, NioSocketChannel channel, int messageBytesLength) {
3638
try {
37-
transport.messageReceived(reference, channel, channel.getProfile(), channel.getRemoteAddress(), messageBytesLength);
39+
transport.messageReceived(reference, channel, profile, channel.getRemoteAddress(), messageBytesLength);
3840
} catch (IOException e) {
3941
handleException(channel, e);
4042
}

‎test/framework/src/main/java/org/elasticsearch/transport/nio/channel/AbstractNioChannel.java‎

Lines changed: 1 addition & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -56,13 +56,11 @@ public abstract class AbstractNioChannel<S extends SelectableChannel & NetworkCh
5656
final AtomicBoolean isClosing = new AtomicBoolean(false);
5757

5858
private final InetSocketAddress localAddress;
59-
private final String profile;
6059
private final CompletableFuture<Void> closeContext = new CompletableFuture<>();
6160
private final ESSelector selector;
6261
private SelectionKey selectionKey;
6362

64-
AbstractNioChannel(String profile, S socketChannel, ESSelector selector) throws IOException {
65-
this.profile = profile;
63+
AbstractNioChannel(S socketChannel, ESSelector selector) throws IOException {
6664
this.socketChannel = socketChannel;
6765
this.localAddress = (InetSocketAddress) socketChannel.getLocalAddress();
6866
this.selector = selector;
@@ -78,11 +76,6 @@ public InetSocketAddress getLocalAddress() {
7876
return localAddress;
7977
}
8078

81-
@Override
82-
public String getProfile() {
83-
return profile;
84-
}
85-
8679
/**
8780
* Schedules a channel to be closed by the selector event loop with which it is registered.
8881
* <p>

‎test/framework/src/main/java/org/elasticsearch/transport/nio/channel/ChannelFactory.java‎

Lines changed: 40 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,6 @@
2020
package org.elasticsearch.transport.nio.channel;
2121

2222

23-
import org.apache.lucene.util.IOUtils;
2423
import org.elasticsearch.mocksocket.PrivilegedSocketAccess;
2524
import org.elasticsearch.transport.TcpTransport;
2625
import org.elasticsearch.transport.nio.AcceptingSelector;
@@ -64,33 +63,51 @@ public ChannelFactory(TcpTransport.ProfileSettings profileSettings, Consumer<Nio
6463

6564
public NioSocketChannel openNioChannel(InetSocketAddress remoteAddress, SocketSelector selector) throws IOException {
6665
SocketChannel rawChannel = rawChannelFactory.openNioChannel(remoteAddress);
67-
NioSocketChannel channel = new NioSocketChannel(NioChannel.CLIENT, rawChannel, selector);
68-
setContexts(channel);
66+
NioSocketChannel channel = createChannel(selector, rawChannel);
6967
scheduleChannel(channel, selector);
7068
return channel;
7169
}
7270

7371
public NioSocketChannel acceptNioChannel(NioServerSocketChannel serverChannel, SocketSelector selector) throws IOException {
7472
SocketChannel rawChannel = rawChannelFactory.acceptNioChannel(serverChannel);
75-
NioSocketChannel channel = new NioSocketChannel(serverChannel.getProfile(), rawChannel, selector);
76-
setContexts(channel);
73+
NioSocketChannel channel = createChannel(selector, rawChannel);
7774
scheduleChannel(channel, selector);
7875
return channel;
7976
}
8077

81-
public NioServerSocketChannel openNioServerSocketChannel(String profileName, InetSocketAddress address, AcceptingSelector selector)
78+
public NioServerSocketChannel openNioServerSocketChannel(InetSocketAddress address, AcceptingSelector selector)
8279
throws IOException {
8380
ServerSocketChannel rawChannel = rawChannelFactory.openNioServerSocketChannel(address);
84-
NioServerSocketChannel serverChannel = new NioServerSocketChannel(profileName, rawChannel, this, selector);
81+
NioServerSocketChannel serverChannel = createServerChannel(selector, rawChannel);
8582
scheduleServerChannel(serverChannel, selector);
8683
return serverChannel;
8784
}
8885

86+
private NioSocketChannel createChannel(SocketSelector selector, SocketChannel rawChannel) throws IOException {
87+
try {
88+
NioSocketChannel channel = new NioSocketChannel(rawChannel, selector);
89+
setContexts(channel);
90+
return channel;
91+
} catch (Exception e) {
92+
closeRawChannel(rawChannel, e);
93+
throw e;
94+
}
95+
}
96+
97+
private NioServerSocketChannel createServerChannel(AcceptingSelector selector, ServerSocketChannel rawChannel) throws IOException {
98+
try {
99+
return new NioServerSocketChannel(rawChannel, this, selector);
100+
} catch (Exception e) {
101+
closeRawChannel(rawChannel, e);
102+
throw e;
103+
}
104+
}
105+
89106
private void scheduleChannel(NioSocketChannel channel, SocketSelector selector) {
90107
try {
91108
selector.scheduleForRegistration(channel);
92109
} catch (IllegalStateException e) {
93-
IOUtils.closeWhileHandlingException(channel.getRawChannel());
110+
closeRawChannel(channel.getRawChannel(), e);
94111
throw e;
95112
}
96113
}
@@ -99,7 +116,7 @@ private void scheduleServerChannel(NioServerSocketChannel channel, AcceptingSele
99116
try {
100117
selector.scheduleForRegistration(channel);
101118
} catch (IllegalStateException e) {
102-
IOUtils.closeWhileHandlingException(channel.getRawChannel());
119+
closeRawChannel(channel.getRawChannel(), e);
103120
throw e;
104121
}
105122
}
@@ -110,6 +127,14 @@ private void setContexts(NioSocketChannel channel) {
110127
assert channel.getWriteContext() != null : "write context should have been set on channel";
111128
}
112129

130+
private static void closeRawChannel(Closeable c, Exception e) {
131+
try {
132+
c.close();
133+
} catch (IOException closeException) {
134+
e.addSuppressed(closeException);
135+
}
136+
}
137+
113138
static class RawChannelFactory {
114139

115140
private final boolean tcpNoDelay;
@@ -142,7 +167,12 @@ SocketChannel openNioChannel(InetSocketAddress remoteAddress) throws IOException
142167
SocketChannel acceptNioChannel(NioServerSocketChannel serverChannel) throws IOException {
143168
ServerSocketChannel serverSocketChannel = serverChannel.getRawChannel();
144169
SocketChannel socketChannel = PrivilegedSocketAccess.accept(serverSocketChannel);
145-
configureSocketChannel(socketChannel);
170+
try {
171+
configureSocketChannel(socketChannel);
172+
} catch (IOException e) {
173+
closeRawChannel(socketChannel, e);
174+
throw e;
175+
}
146176
return socketChannel;
147177
}
148178

@@ -160,14 +190,6 @@ ServerSocketChannel openNioServerSocketChannel(InetSocketAddress address) throws
160190
return serverSocketChannel;
161191
}
162192

163-
private void closeRawChannel(Closeable c, IOException e) {
164-
try {
165-
c.close();
166-
} catch (IOException closeException) {
167-
e.addSuppressed(closeException);
168-
}
169-
}
170-
171193
private void configureSocketChannel(SocketChannel channel) throws IOException {
172194
channel.configureBlocking(false);
173195
Socket socket = channel.socket();

‎test/framework/src/main/java/org/elasticsearch/transport/nio/channel/NioChannel.java‎

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -30,12 +30,8 @@
3030

3131
public interface NioChannel extends TcpChannel {
3232

33-
String CLIENT = "client-socket";
34-
3533
InetSocketAddress getLocalAddress();
3634

37-
String getProfile();
38-
3935
void close();
4036

4137
void closeFromSelector() throws IOException;

‎test/framework/src/main/java/org/elasticsearch/transport/nio/channel/NioServerSocketChannel.java‎

Lines changed: 4 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -30,9 +30,9 @@ public class NioServerSocketChannel extends AbstractNioChannel<ServerSocketChann
3030

3131
private final ChannelFactory channelFactory;
3232

33-
public NioServerSocketChannel(String profile, ServerSocketChannel socketChannel, ChannelFactory channelFactory,
34-
AcceptingSelector selector) throws IOException {
35-
super(profile, socketChannel, selector);
33+
public NioServerSocketChannel(ServerSocketChannel socketChannel, ChannelFactory channelFactory, AcceptingSelector selector)
34+
throws IOException {
35+
super(socketChannel, selector);
3636
this.channelFactory = channelFactory;
3737
}
3838

@@ -48,8 +48,7 @@ public void sendMessage(BytesReference reference, ActionListener<Void> listener)
4848
@Override
4949
public String toString() {
5050
return "NioServerSocketChannel{" +
51-
"profile=" + getProfile() +
52-
", localAddress=" + getLocalAddress() +
51+
"localAddress=" + getLocalAddress() +
5352
'}';
5453
}
5554
}

‎test/framework/src/main/java/org/elasticsearch/transport/nio/channel/NioSocketChannel.java‎

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -41,8 +41,8 @@ public class NioSocketChannel extends AbstractNioChannel<SocketChannel> {
4141
private ReadContext readContext;
4242
private Exception connectException;
4343

44-
public NioSocketChannel(String profile, SocketChannel socketChannel, SocketSelector selector) throws IOException {
45-
super(profile, socketChannel, selector);
44+
public NioSocketChannel(SocketChannel socketChannel, SocketSelector selector) throws IOException {
45+
super(socketChannel, selector);
4646
this.remoteAddress = (InetSocketAddress) socketChannel.getRemoteAddress();
4747
this.socketSelector = selector;
4848
}
@@ -169,8 +169,7 @@ public void addConnectListener(ActionListener<Void> listener) {
169169
@Override
170170
public String toString() {
171171
return "NioSocketChannel{" +
172-
"profile=" + getProfile() +
173-
", localAddress=" + getLocalAddress() +
172+
"localAddress=" + getLocalAddress() +
174173
", remoteAddress=" + remoteAddress +
175174
'}';
176175
}

‎test/framework/src/test/java/org/elasticsearch/transport/nio/AcceptorEventHandlerTests.java‎

Lines changed: 3 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -23,13 +23,11 @@
2323
import org.elasticsearch.test.ESTestCase;
2424
import org.elasticsearch.transport.nio.channel.ChannelFactory;
2525
import org.elasticsearch.transport.nio.channel.DoNotRegisterServerChannel;
26-
import org.elasticsearch.transport.nio.channel.NioChannel;
2726
import org.elasticsearch.transport.nio.channel.NioServerSocketChannel;
2827
import org.elasticsearch.transport.nio.channel.NioSocketChannel;
2928
import org.elasticsearch.transport.nio.channel.ReadContext;
3029
import org.elasticsearch.transport.nio.channel.WriteContext;
3130
import org.junit.Before;
32-
import org.mockito.ArgumentCaptor;
3331

3432
import java.io.IOException;
3533
import java.nio.channels.SelectionKey;
@@ -67,7 +65,7 @@ public void setUpHandler() throws IOException {
6765
handler = new AcceptorEventHandler(logger, openChannels, new RoundRobinSelectorSupplier(selectors), acceptedChannelCallback);
6866

6967
AcceptingSelector selector = mock(AcceptingSelector.class);
70-
channel = new DoNotRegisterServerChannel("", mock(ServerSocketChannel.class), channelFactory, selector);
68+
channel = new DoNotRegisterServerChannel(mock(ServerSocketChannel.class), channelFactory, selector);
7169
channel.register();
7270
}
7371

@@ -88,7 +86,7 @@ public void testHandleRegisterSetsOP_ACCEPTInterest() {
8886
}
8987

9088
public void testHandleAcceptCallsChannelFactory() throws IOException {
91-
NioSocketChannel childChannel = new NioSocketChannel("", mock(SocketChannel.class), socketSelector);
89+
NioSocketChannel childChannel = new NioSocketChannel(mock(SocketChannel.class), socketSelector);
9290
when(channelFactory.acceptNioChannel(same(channel), same(socketSelector))).thenReturn(childChannel);
9391

9492
handler.acceptChannel(channel);
@@ -100,7 +98,7 @@ public void testHandleAcceptCallsChannelFactory() throws IOException {
10098
@SuppressWarnings("unchecked")
10199
public void testHandleAcceptAddsToOpenChannelsAndIsRemovedOnClose() throws IOException {
102100
SocketChannel rawChannel = SocketChannel.open();
103-
NioSocketChannel childChannel = new NioSocketChannel("", rawChannel, socketSelector);
101+
NioSocketChannel childChannel = new NioSocketChannel(rawChannel, socketSelector);
104102
childChannel.setContexts(mock(ReadContext.class), mock(WriteContext.class));
105103
when(channelFactory.acceptNioChannel(same(channel), same(socketSelector))).thenReturn(childChannel);
106104

‎test/framework/src/test/java/org/elasticsearch/transport/nio/SocketEventHandlerTests.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -57,7 +57,7 @@ public void setUpHandler() throws IOException {
5757
SocketSelector socketSelector = mock(SocketSelector.class);
5858
handler = new SocketEventHandler(logger, exceptionHandler, mock(OpenChannels.class));
5959
rawChannel = mock(SocketChannel.class);
60-
channel = new DoNotRegisterChannel("", rawChannel, socketSelector);
60+
channel = new DoNotRegisterChannel(rawChannel, socketSelector);
6161
readContext = mock(ReadContext.class);
6262
when(rawChannel.finishConnect()).thenReturn(true);
6363

‎test/framework/src/test/java/org/elasticsearch/transport/nio/channel/ChannelFactoryTests.java‎

Lines changed: 2 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -76,14 +76,12 @@ public void ensureClosed() throws IOException {
7676
public void testAcceptChannel() throws IOException {
7777
NioServerSocketChannel serverChannel = mock(NioServerSocketChannel.class);
7878
when(rawChannelFactory.acceptNioChannel(serverChannel)).thenReturn(rawChannel);
79-
when(serverChannel.getProfile()).thenReturn("parent-profile");
8079

8180
NioSocketChannel channel = channelFactory.acceptNioChannel(serverChannel, socketSelector);
8281

8382
verify(socketSelector).scheduleForRegistration(channel);
8483

8584
assertEquals(socketSelector, channel.getSelector());
86-
assertEquals("parent-profile", channel.getProfile());
8785
assertEquals(rawChannel, channel.getRawChannel());
8886
}
8987

@@ -106,7 +104,6 @@ public void testOpenChannel() throws IOException {
106104
verify(socketSelector).scheduleForRegistration(channel);
107105

108106
assertEquals(socketSelector, channel.getSelector());
109-
assertEquals("client-socket", channel.getProfile());
110107
assertEquals(rawChannel, channel.getRawChannel());
111108
}
112109

@@ -124,13 +121,11 @@ public void testOpenServerChannel() throws IOException {
124121
InetSocketAddress address = mock(InetSocketAddress.class);
125122
when(rawChannelFactory.openNioServerSocketChannel(same(address))).thenReturn(rawServerChannel);
126123

127-
String profile = "profile";
128-
NioServerSocketChannel channel = channelFactory.openNioServerSocketChannel(profile, address, acceptingSelector);
124+
NioServerSocketChannel channel = channelFactory.openNioServerSocketChannel(address, acceptingSelector);
129125

130126
verify(acceptingSelector).scheduleForRegistration(channel);
131127

132128
assertEquals(acceptingSelector, channel.getSelector());
133-
assertEquals(profile, channel.getProfile());
134129
assertEquals(rawServerChannel, channel.getRawChannel());
135130
}
136131

@@ -139,7 +134,7 @@ public void testOpenedServerChannelRejected() throws IOException {
139134
when(rawChannelFactory.openNioServerSocketChannel(same(address))).thenReturn(rawServerChannel);
140135
doThrow(new IllegalStateException()).when(acceptingSelector).scheduleForRegistration(any());
141136

142-
expectThrows(IllegalStateException.class, () -> channelFactory.openNioServerSocketChannel("", address, acceptingSelector));
137+
expectThrows(IllegalStateException.class, () -> channelFactory.openNioServerSocketChannel(address, acceptingSelector));
143138

144139
assertFalse(rawServerChannel.isOpen());
145140
}

0 commit comments

Comments
 (0)