honeymoose/OpenSearch
1
0
Fork 0
mirror of https://github.com/honeymoose/OpenSearch.git synced 2025-02-23 05:15:04 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add Checks for Closed Channel in Selector Loop (#39096) (#39439)

Browse Source 
* A few warnings could be observed in test logs about `NoSuchElementException` being thrown in `InboundChannelBuffer#sliceBuffersTo`.
These were the result of calls to this method after the relevant channel and hence the buffer was closed already as a result of a failed IO operation.
  * Fixed by adding the necessary guard statements to break out in these cases. I don't think there is a need here to do any additional error handling since `eventHandler.postHandling(channelContext);` at the end of the `processKey`
call in the main selection loop handles closing channels and invoking callbacks for writes that failed to go through already.
This commit is contained in:
Armin Braun 2019-02-27 11:28:30 +01:00 committed by GitHub
parent 1d0097b5e8
commit da9190be0a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 12 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
libs/nio/src/main/java/org/elasticsearch/nio/NioSelector.java
View File

@ -238,12 +238,13 @@ public class NioSelector implements Closeable {
} }
if (channelContext.isConnectComplete()) { if (channelContext.isConnectComplete()) {
if ((ops & SelectionKey.OP_WRITE) != 0) { if (channelContext.selectorShouldClose() == false) {
handleWrite(channelContext); if ((ops & SelectionKey.OP_WRITE) != 0) {
} handleWrite(channelContext);
}
if ((ops & SelectionKey.OP_READ) != 0) { if (channelContext.selectorShouldClose() == false && (ops & SelectionKey.OP_READ) != 0) {
handleRead(channelContext); handleRead(channelContext);
}
} }
} }
eventHandler.postHandling(channelContext); eventHandler.postHandling(channelContext);
@ -336,7 +337,9 @@ public class NioSelector implements Closeable {
} }
if (shouldFlushAfterQueuing) { if (shouldFlushAfterQueuing) {
handleWrite(context); if (context.selectorShouldClose() == false) {
handleWrite(context);
}
eventHandler.postHandling(context); eventHandler.postHandling(context);
} }
} }

2
libs/nio/src/main/java/org/elasticsearch/nio/SocketChannelContext.java
View File

@ -209,7 +209,7 @@ public abstract class SocketChannelContext extends ChannelContext<SocketChannel>
protected void handleReadBytes() throws IOException { protected void handleReadBytes() throws IOException {
int bytesConsumed = Integer.MAX_VALUE; int bytesConsumed = Integer.MAX_VALUE;
while (bytesConsumed > 0 && channelBuffer.getIndex() > 0) { while (isOpen() && bytesConsumed > 0 && channelBuffer.getIndex() > 0) {
bytesConsumed = readWriteHandler.consumeReads(channelBuffer); bytesConsumed = readWriteHandler.consumeReads(channelBuffer);
channelBuffer.release(bytesConsumed); channelBuffer.release(bytesConsumed);
} }

2
libs/nio/src/test/java/org/elasticsearch/nio/SocketChannelContextTests.java
View File

@ -476,7 +476,7 @@ public class SocketChannelContextTests extends ESTestCase {
@Override @Override
public boolean selectorShouldClose() { public boolean selectorShouldClose() {
return false; return isClosing.get();
} }
@Override @Override