honeymoose
/
OpenSearch
mirror of https://github.com/honeymoose/OpenSearch.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Cleanup TcpChannelFactory and remove classes (#28102) 

This commit is related to #27260. It moves the TcpChannelFactory into
NioTransport so that consumers do not have to be passed around.
Additionally it deletes an unused read handler.
This commit is contained in:
Tim Brooks 2018-01-08 10:18:19 -07:00 committed by GitHub
parent af3c2fcd65
commit ff3db0b50e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 42 additions and 137 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

57
plugins/transport-nio/src/main/java/org/elasticsearch/transport/nio/NioTransport.java
View File

@ -33,14 +33,17 @@ import org.elasticsearch.common.util.BigArrays;
import org.elasticsearch.common.util.PageCacheRecycler;
import org.elasticsearch.common.util.concurrent.EsExecutors;
import org.elasticsearch.indices.breaker.CircuitBreakerService;
import org.elasticsearch.nio.AcceptingSelector;
import org.elasticsearch.nio.AcceptorEventHandler;
import org.elasticsearch.nio.BytesReadContext;
import org.elasticsearch.nio.BytesWriteContext;
import org.elasticsearch.nio.ChannelFactory;
import org.elasticsearch.nio.InboundChannelBuffer;
import org.elasticsearch.nio.NioGroup;
import org.elasticsearch.nio.NioSocketChannel;
import org.elasticsearch.nio.ReadContext;
import org.elasticsearch.nio.SocketEventHandler;
import org.elasticsearch.nio.SocketSelector;
import org.elasticsearch.threadpool.ThreadPool;
import org.elasticsearch.transport.TcpChannel;
import org.elasticsearch.transport.TcpTransport;
@ -49,8 +52,9 @@ import org.elasticsearch.transport.Transports;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.concurrent.ConcurrentMap;
import java.util.function.Consumer;
import java.util.function.Supplier;
import static org.elasticsearch.common.settings.Setting.intSetting;
@ -110,13 +114,13 @@ public class NioTransport extends TcpTransport {
NioTransport.NIO_WORKER_COUNT.get(settings), SocketEventHandler::new);
ProfileSettings clientProfileSettings = new ProfileSettings(settings, "default");
clientChannelFactory = new TcpChannelFactory(clientProfileSettings, getContextSetter(), getServerContextSetter());
clientChannelFactory = new TcpChannelFactory(clientProfileSettings);
if (useNetworkServer) {
// loop through all profiles and start them up, special handling for default one
for (ProfileSettings profileSettings : profileSettings) {
String profileName = profileSettings.profileName;
TcpChannelFactory factory = new TcpChannelFactory(profileSettings, getContextSetter(), getServerContextSetter());
TcpChannelFactory factory = new TcpChannelFactory(profileSettings);
profileToChannelFactory.putIfAbsent(profileName, factory);
bindServer(profileSettings);
}
@ -143,29 +147,46 @@ public class NioTransport extends TcpTransport {
profileToChannelFactory.clear();
}
final void exceptionCaught(NioSocketChannel channel, Exception exception) {
private void exceptionCaught(NioSocketChannel channel, Exception exception) {
onException((TcpChannel) channel, exception);
}
private Consumer<TcpNioSocketChannel> getContextSetter() {
return (c) -> {
private void acceptChannel(NioSocketChannel channel) {
serverAcceptedChannel((TcpNioSocketChannel) channel);
}
private class TcpChannelFactory extends ChannelFactory<TcpNioServerSocketChannel, TcpNioSocketChannel> {
private final String profileName;
TcpChannelFactory(TcpTransport.ProfileSettings profileSettings) {
super(new RawChannelFactory(profileSettings.tcpNoDelay,
profileSettings.tcpKeepAlive,
profileSettings.reuseAddress,
Math.toIntExact(profileSettings.sendBufferSize.getBytes()),
Math.toIntExact(profileSettings.receiveBufferSize.getBytes())));
this.profileName = profileSettings.profileName;
}
@Override
public TcpNioSocketChannel createChannel(SocketSelector selector, SocketChannel channel) throws IOException {
TcpNioSocketChannel nioChannel = new TcpNioSocketChannel(profileName, channel, selector);
Supplier<InboundChannelBuffer.Page> pageSupplier = () -> {
Recycler.V<byte[]> bytes = pageCacheRecycler.bytePage(false);
return new InboundChannelBuffer.Page(ByteBuffer.wrap(bytes.v()), bytes::close);
};
ReadContext.ReadConsumer nioReadConsumer = channelBuffer ->
consumeNetworkReads(c, BytesReference.fromByteBuffers(channelBuffer.sliceBuffersTo(channelBuffer.getIndex())));
BytesReadContext readContext = new BytesReadContext(c, nioReadConsumer, new InboundChannelBuffer(pageSupplier));
c.setContexts(readContext, new BytesWriteContext(c), this::exceptionCaught);
};
}
consumeNetworkReads(nioChannel, BytesReference.fromByteBuffers(channelBuffer.sliceBuffersTo(channelBuffer.getIndex())));
BytesReadContext readContext = new BytesReadContext(nioChannel, nioReadConsumer, new InboundChannelBuffer(pageSupplier));
nioChannel.setContexts(readContext, new BytesWriteContext(nioChannel), NioTransport.this::exceptionCaught);
return nioChannel;
}
private void acceptChannel(NioSocketChannel channel) {
serverAcceptedChannel((TcpNioSocketChannel) channel);
}
private Consumer<TcpNioServerSocketChannel> getServerContextSetter() {
return (c) -> c.setAcceptContext(this::acceptChannel);
@Override
public TcpNioServerSocketChannel createServerChannel(AcceptingSelector selector, ServerSocketChannel channel) throws IOException {
TcpNioServerSocketChannel nioServerChannel = new TcpNioServerSocketChannel(profileName, channel, this, selector);
nioServerChannel.setAcceptContext(NioTransport.this::acceptChannel);
return nioServerChannel;
}
}
}

69
plugins/transport-nio/src/main/java/org/elasticsearch/transport/nio/TcpChannelFactory.java
View File

@ -1,69 +0,0 @@
/*
* 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.nio.AcceptingSelector;
import org.elasticsearch.nio.ChannelFactory;
import org.elasticsearch.nio.SocketSelector;
import org.elasticsearch.transport.TcpTransport;
import java.io.IOException;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.function.Consumer;
/**
* This is an implementation of {@link ChannelFactory} which returns channels that adhere to the
* {@link org.elasticsearch.transport.TcpChannel} interface. The channels will use the provided
* {@link TcpTransport.ProfileSettings}. The provided context setters will be called with the channel after
* construction.
*/
public class TcpChannelFactory extends ChannelFactory<TcpNioServerSocketChannel, TcpNioSocketChannel> {
private final Consumer<TcpNioSocketChannel> contextSetter;
private final Consumer<TcpNioServerSocketChannel> serverContextSetter;
private final String profileName;
TcpChannelFactory(TcpTransport.ProfileSettings profileSettings, Consumer<TcpNioSocketChannel> contextSetter,
Consumer<TcpNioServerSocketChannel> serverContextSetter) {
super(new RawChannelFactory(profileSettings.tcpNoDelay,
profileSettings.tcpKeepAlive,
profileSettings.reuseAddress,
Math.toIntExact(profileSettings.sendBufferSize.getBytes()),
Math.toIntExact(profileSettings.receiveBufferSize.getBytes())));
this.profileName = profileSettings.profileName;
this.contextSetter = contextSetter;
this.serverContextSetter = serverContextSetter;
}
@Override
public TcpNioSocketChannel createChannel(SocketSelector selector, SocketChannel channel) throws IOException {
TcpNioSocketChannel nioChannel = new TcpNioSocketChannel(profileName, channel, selector);
contextSetter.accept(nioChannel);
return nioChannel;
}
@Override
public TcpNioServerSocketChannel createServerChannel(AcceptingSelector selector, ServerSocketChannel channel) throws IOException {
TcpNioServerSocketChannel nioServerChannel = new TcpNioServerSocketChannel(profileName, channel, this, selector);
serverContextSetter.accept(nioServerChannel);
return nioServerChannel;
}
}

4
plugins/transport-nio/src/main/java/org/elasticsearch/transport/nio/TcpNioServerSocketChannel.java
View File

@ -22,6 +22,7 @@ package org.elasticsearch.transport.nio;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.common.bytes.BytesReference;
import org.elasticsearch.nio.AcceptingSelector;
import org.elasticsearch.nio.ChannelFactory;
import org.elasticsearch.nio.NioServerSocketChannel;
import org.elasticsearch.transport.TcpChannel;
@ -37,7 +38,8 @@ public class TcpNioServerSocketChannel extends NioServerSocketChannel implements
private final String profile;
TcpNioServerSocketChannel(String profile, ServerSocketChannel socketChannel, TcpChannelFactory channelFactory,
TcpNioServerSocketChannel(String profile, ServerSocketChannel socketChannel,
ChannelFactory<TcpNioServerSocketChannel, TcpNioSocketChannel> channelFactory,
AcceptingSelector selector) throws IOException {
super(socketChannel, channelFactory, selector);
this.profile = profile;

48
plugins/transport-nio/src/main/java/org/elasticsearch/transport/nio/TcpReadHandler.java
View File

@ -1,48 +0,0 @@
/*
* 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.bytes.BytesReference;
import org.elasticsearch.nio.NioSocketChannel;
import java.io.IOException;
public class TcpReadHandler {
private final String profile;
private final NioTransport transport;
TcpReadHandler(String profile, NioTransport transport) {
this.profile = profile;
this.transport = transport;
}
public void handleMessage(BytesReference reference, TcpNioSocketChannel channel, int messageBytesLength) {
try {
transport.messageReceived(reference, channel);
} catch (IOException e) {
handleException(channel, e);
}
}
public void handleException(NioSocketChannel channel, Exception e) {
transport.exceptionCaught(channel, e);
}
}

1
plugins/transport-nio/src/test/java/org/elasticsearch/transport/nio/NioTransportIT.java
View File

@ -128,5 +128,4 @@ public class NioTransportIT extends NioIntegTestCase {
}
}
}