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

Remove manual tracking of registered channels (#27445) 

This is related to #27260. Currently, every ESSelector keeps track of
all channels that are registered with it. ESSelector is just an
abstraction over a raw java nio selector. The java nio selector already
tracks its own selection keys. This commit removes our tracking and
relies on the java nio selector tracking.
This commit is contained in:
Tim Brooks 2017-11-17 16:20:09 -07:00 committed by GitHub
parent cc3be6ddda
commit ce45e29be7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 16 additions and 62 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
test/framework/src/main/java/org/elasticsearch/transport/nio/AcceptingSelector.java
View File

@ -22,13 +22,9 @@ package org.elasticsearch.transport.nio;
import org.elasticsearch.transport.nio.channel.NioServerSocketChannel;
import java.io.IOException;
import java.nio.channels.CancelledKeyException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.ClosedSelectorException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
/**
@ -93,7 +89,6 @@ public class AcceptingSelector extends ESSelector {
newChannel.register();
SelectionKey selectionKey = newChannel.getSelectionKey();
selectionKey.attach(newChannel);
addRegisteredChannel(newChannel);
eventHandler.serverChannelRegistered(newChannel);
} else {
eventHandler.registrationException(newChannel, new ClosedChannelException());

19
test/framework/src/main/java/org/elasticsearch/transport/nio/ESSelector.java
View File

@ -28,14 +28,13 @@ import java.nio.channels.CancelledKeyException;
import java.nio.channels.ClosedSelectorException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.Collections;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;
/**
* This is a basic selector abstraction used by {@link org.elasticsearch.transport.nio.NioTransport}. This
@ -56,7 +55,6 @@ public abstract class ESSelector implements Closeable {
private final CountDownLatch exitedLoop = new CountDownLatch(1);
private final AtomicBoolean isClosed = new AtomicBoolean(false);
private final PlainActionFuture<Boolean> isRunningFuture = PlainActionFuture.newFuture();
private final Set<NioChannel> registeredChannels = Collections.newSetFromMap(new ConcurrentHashMap<NioChannel, Boolean>());
private volatile Thread thread;
ESSelector(EventHandler eventHandler) throws IOException {
@ -134,7 +132,7 @@ public abstract class ESSelector implements Closeable {
void cleanupAndCloseChannels() {
cleanup();
channelsToClose.addAll(registeredChannels);
channelsToClose.addAll(selector.keys().stream().map(sk -> (NioChannel) sk.attachment()).collect(Collectors.toList()));
closePendingChannels();
}
@ -171,19 +169,6 @@ public abstract class ESSelector implements Closeable {
selector.wakeup();
}
public Set<NioChannel> getRegisteredChannels() {
return registeredChannels;
}
public void addRegisteredChannel(NioChannel channel) {
assert registeredChannels.contains(channel) == false : "Should only register channel once";
registeredChannels.add(channel);
}
public void removeRegisteredChannel(NioChannel channel) {
registeredChannels.remove(channel);
}
@Override
public void close() throws IOException {
if (isClosed.compareAndSet(false, true)) {

1
test/framework/src/main/java/org/elasticsearch/transport/nio/SocketSelector.java
View File

@ -171,7 +171,6 @@ public class SocketSelector extends ESSelector {
try {
if (newChannel.isOpen()) {
newChannel.register();
addRegisteredChannel(newChannel);
SelectionKey key = newChannel.getSelectionKey();
key.attach(newChannel);
eventHandler.handleRegistration(newChannel);

3
test/framework/src/main/java/org/elasticsearch/transport/nio/channel/AbstractNioChannel.java
View File

@ -115,9 +115,6 @@ public abstract class AbstractNioChannel<S extends SelectableChannel & NetworkCh
} catch (IOException e) {
closeContext.completeExceptionally(e);
throw e;
} finally {
// There is no problem with calling this multiple times
selector.removeRegisteredChannel(this);
}
}
}

21
test/framework/src/test/java/org/elasticsearch/transport/nio/AcceptingSelectorTests.java
View File

@ -20,7 +20,6 @@
package org.elasticsearch.transport.nio;
import org.elasticsearch.test.ESTestCase;
import org.elasticsearch.transport.nio.channel.NioChannel;
import org.elasticsearch.transport.nio.channel.NioServerSocketChannel;
import org.elasticsearch.transport.nio.utils.TestSelectionKey;
import org.junit.Before;
@ -30,8 +29,8 @@ import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.security.PrivilegedActionException;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.same;
@ -46,6 +45,7 @@ public class AcceptingSelectorTests extends ESTestCase {
private NioServerSocketChannel serverChannel;
private AcceptorEventHandler eventHandler;
private TestSelectionKey selectionKey;
private Selector rawSelector;
@Before
public void setUp() throws Exception {
@ -54,7 +54,7 @@ public class AcceptingSelectorTests extends ESTestCase {
eventHandler = mock(AcceptorEventHandler.class);
serverChannel = mock(NioServerSocketChannel.class);
Selector rawSelector = mock(Selector.class);
rawSelector = mock(Selector.class);
selector = new AcceptingSelector(eventHandler, rawSelector);
this.selector.setThread();
@ -71,9 +71,6 @@ public class AcceptingSelectorTests extends ESTestCase {
selector.preSelect();
verify(eventHandler).serverChannelRegistered(serverChannel);
Set<NioChannel> registeredChannels = selector.getRegisteredChannels();
assertEquals(1, registeredChannels.size());
assertTrue(registeredChannels.contains(serverChannel));
}
public void testClosedChannelWillNotBeRegistered() throws Exception {
@ -83,10 +80,6 @@ public class AcceptingSelectorTests extends ESTestCase {
selector.preSelect();
verify(eventHandler).registrationException(same(serverChannel), any(ClosedChannelException.class));
Set<NioChannel> registeredChannels = selector.getRegisteredChannels();
assertEquals(0, registeredChannels.size());
assertFalse(registeredChannels.contains(serverChannel));
}
public void testRegisterChannelFailsDueToException() throws Exception {
@ -98,10 +91,6 @@ public class AcceptingSelectorTests extends ESTestCase {
selector.preSelect();
verify(eventHandler).registrationException(serverChannel, closedChannelException);
Set<NioChannel> registeredChannels = selector.getRegisteredChannels();
assertEquals(0, registeredChannels.size());
assertFalse(registeredChannels.contains(serverChannel));
}
public void testAcceptEvent() throws IOException {
@ -128,7 +117,9 @@ public class AcceptingSelectorTests extends ESTestCase {
selector.preSelect();
assertEquals(1, selector.getRegisteredChannels().size());
TestSelectionKey key = new TestSelectionKey(0);
key.attach(serverChannel);
when(rawSelector.keys()).thenReturn(new HashSet<>(Collections.singletonList(key)));
selector.cleanupAndCloseChannels();

7
test/framework/src/test/java/org/elasticsearch/transport/nio/ESSelectorTests.java
View File

@ -51,19 +51,12 @@ public class ESSelectorTests extends ESTestCase {
public void testQueueChannelForClosed() throws IOException {
NioChannel channel = mock(NioChannel.class);
when(channel.getSelector()).thenReturn(selector);
selector.addRegisteredChannel(channel);
selector.queueChannelClose(channel);
assertEquals(1, selector.getRegisteredChannels().size());
selector.singleLoop();
verify(handler).handleClose(channel);
// Will be called in the channel close method
selector.removeRegisteredChannel(channel);
assertEquals(0, selector.getRegisteredChannels().size());
}
public void testSelectorClosedExceptionIsNotCaughtWhileRunning() throws IOException {

22
test/framework/src/test/java/org/elasticsearch/transport/nio/SocketSelectorTests.java
View File

@ -34,7 +34,8 @@ import java.nio.channels.ClosedChannelException;
import java.nio.channels.ClosedSelectorException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.Set;
import java.util.Collections;
import java.util.HashSet;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyInt;
@ -54,6 +55,7 @@ public class SocketSelectorTests extends ESTestCase {
private WriteContext writeContext;
private ActionListener<NioChannel> listener;
private NetworkBytesReference bufferReference = NetworkBytesReference.wrap(new BytesArray(new byte[1]));
private Selector rawSelector;
@Before
@SuppressWarnings("unchecked")
@ -65,7 +67,7 @@ public class SocketSelectorTests extends ESTestCase {
listener = mock(ActionListener.class);
selectionKey = new TestSelectionKey(0);
selectionKey.attach(channel);
Selector rawSelector = mock(Selector.class);
rawSelector = mock(Selector.class);
this.socketSelector = new SocketSelector(eventHandler, rawSelector);
this.socketSelector.setThread();
@ -83,10 +85,6 @@ public class SocketSelectorTests extends ESTestCase {
socketSelector.preSelect();
verify(eventHandler).handleRegistration(channel);
Set<NioChannel> registeredChannels = socketSelector.getRegisteredChannels();
assertEquals(1, registeredChannels.size());
assertTrue(registeredChannels.contains(channel));
}
public void testClosedChannelWillNotBeRegistered() throws Exception {
@ -97,10 +95,6 @@ public class SocketSelectorTests extends ESTestCase {
verify(eventHandler).registrationException(same(channel), any(ClosedChannelException.class));
verify(channel, times(0)).finishConnect();
Set<NioChannel> registeredChannels = socketSelector.getRegisteredChannels();
assertEquals(0, registeredChannels.size());
assertFalse(registeredChannels.contains(channel));
}
public void testRegisterChannelFailsDueToException() throws Exception {
@ -113,10 +107,6 @@ public class SocketSelectorTests extends ESTestCase {
verify(eventHandler).registrationException(channel, closedChannelException);
verify(channel, times(0)).finishConnect();
Set<NioChannel> registeredChannels = socketSelector.getRegisteredChannels();
assertEquals(0, registeredChannels.size());
assertFalse(registeredChannels.contains(channel));
}
public void testSuccessfullyRegisterChannelWillConnect() throws Exception {
@ -309,6 +299,10 @@ public class SocketSelectorTests extends ESTestCase {
socketSelector.queueWrite(new WriteOperation(mock(NioSocketChannel.class), networkBuffer, listener));
socketSelector.scheduleForRegistration(unRegisteredChannel);
TestSelectionKey testSelectionKey = new TestSelectionKey(0);
testSelectionKey.attach(channel);
when(rawSelector.keys()).thenReturn(new HashSet<>(Collections.singletonList(testSelectionKey)));
socketSelector.cleanupAndCloseChannels();
verify(listener).onFailure(any(ClosedSelectorException.class));