Skip to content

Commit 11fe52a

Browse files
authored
Add log warnings for long running event handling (#39729)
Recently we have had a number of test issues related to blocking activity occuring on the io thread. This commit adds a log warning for when handling event takes a >150 milliseconds. This is implemented for the MockNioTransport which is the transport used in ESIntegTestCase.
1 parent 160adf7 commit 11fe52a

File tree

8 files changed

+409
-111
lines changed

8 files changed

+409
-111
lines changed

libs/nio/src/main/java/org/elasticsearch/nio/EventHandler.java

Lines changed: 22 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -149,6 +149,15 @@ protected void writeException(SocketChannelContext context, Exception exception)
149149
context.handleException(exception);
150150
}
151151

152+
/**
153+
* This method is called when a task or listener attached to a channel is available to run.
154+
*
155+
* @param task to handle
156+
*/
157+
protected void handleTask(Runnable task) {
158+
task.run();
159+
}
160+
152161
/**
153162
* This method is called when a task or listener attached to a channel operation throws an exception.
154163
*
@@ -165,7 +174,11 @@ protected void taskException(Exception exception) {
165174
*/
166175
protected void postHandling(SocketChannelContext context) {
167176
if (context.selectorShouldClose()) {
168-
handleClose(context);
177+
try {
178+
handleClose(context);
179+
} catch (IOException e) {
180+
closeException(context, e);
181+
}
169182
} else {
170183
SelectionKey selectionKey = context.getSelectionKey();
171184
boolean currentlyWriteInterested = SelectionKeyUtils.isWriteInterested(selectionKey);
@@ -203,34 +216,30 @@ protected void uncaughtException(Exception exception) {
203216
*
204217
* @param context that should be closed
205218
*/
206-
protected void handleClose(ChannelContext<?> context) {
207-
try {
208-
context.closeFromSelector();
209-
} catch (IOException e) {
210-
closeException(context, e);
211-
}
219+
protected void handleClose(ChannelContext<?> context) throws IOException {
220+
context.closeFromSelector();
212221
assert context.isOpen() == false : "Should always be done as we are on the selector thread";
213222
}
214223

215224
/**
216225
* This method is called when an attempt to close a channel throws an exception.
217226
*
218-
* @param channel that was being closed
227+
* @param context that was being closed
219228
* @param exception that occurred
220229
*/
221-
protected void closeException(ChannelContext<?> channel, Exception exception) {
222-
channel.handleException(exception);
230+
protected void closeException(ChannelContext<?> context, Exception exception) {
231+
context.handleException(exception);
223232
}
224233

225234
/**
226235
* This method is called when handling an event from a channel fails due to an unexpected exception.
227236
* An example would be if checking ready ops on a {@link java.nio.channels.SelectionKey} threw
228237
* {@link java.nio.channels.CancelledKeyException}.
229238
*
230-
* @param channel that caused the exception
239+
* @param context that caused the exception
231240
* @param exception that was thrown
232241
*/
233-
protected void genericChannelException(ChannelContext<?> channel, Exception exception) {
234-
channel.handleException(exception);
242+
protected void genericChannelException(ChannelContext<?> context, Exception exception) {
243+
context.handleException(exception);
235244
}
236245
}

libs/nio/src/main/java/org/elasticsearch/nio/NioSelector.java

Lines changed: 16 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -265,11 +265,15 @@ void preSelect() {
265265
private void handleScheduledTasks(long nanoTime) {
266266
Runnable task;
267267
while ((task = taskScheduler.pollTask(nanoTime)) != null) {
268-
try {
269-
task.run();
270-
} catch (Exception e) {
271-
eventHandler.taskException(e);
272-
}
268+
handleTask(task);
269+
}
270+
}
271+
272+
private void handleTask(Runnable task) {
273+
try {
274+
eventHandler.handleTask(task);
275+
} catch (Exception e) {
276+
eventHandler.taskException(e);
273277
}
274278
}
275279

@@ -353,11 +357,7 @@ public void writeToChannel(WriteOperation writeOperation) {
353357
*/
354358
public <V> void executeListener(BiConsumer<V, Exception> listener, V value) {
355359
assertOnSelectorThread();
356-
try {
357-
listener.accept(value, null);
358-
} catch (Exception e) {
359-
eventHandler.taskException(e);
360-
}
360+
handleTask(() -> listener.accept(value, null));
361361
}
362362

363363
/**
@@ -369,11 +369,7 @@ public <V> void executeListener(BiConsumer<V, Exception> listener, V value) {
369369
*/
370370
public <V> void executeFailedListener(BiConsumer<V, Exception> listener, Exception exception) {
371371
assertOnSelectorThread();
372-
try {
373-
listener.accept(null, exception);
374-
} catch (Exception e) {
375-
eventHandler.taskException(e);
376-
}
372+
handleTask(() -> listener.accept(null, exception));
377373
}
378374

379375
private void cleanupPendingWrites() {
@@ -437,7 +433,11 @@ private void setUpNewChannels() {
437433
private void closePendingChannels() {
438434
ChannelContext<?> channelContext;
439435
while ((channelContext = channelsToClose.poll()) != null) {
440-
eventHandler.handleClose(channelContext);
436+
try {
437+
eventHandler.handleClose(channelContext);
438+
} catch (Exception e) {
439+
eventHandler.closeException(channelContext, e);
440+
}
441441
}
442442
}
443443

libs/nio/src/test/java/org/elasticsearch/nio/EventHandlerTests.java

Lines changed: 11 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
import java.nio.channels.SocketChannel;
3131
import java.util.ArrayList;
3232
import java.util.Collections;
33+
import java.util.concurrent.atomic.AtomicBoolean;
3334
import java.util.function.Consumer;
3435

3536
import static org.mockito.Matchers.same;
@@ -243,10 +244,16 @@ public void testPostHandlingWillRemoveWriteIfNecessary() throws IOException {
243244
assertEquals(SelectionKey.OP_READ, key.interestOps());
244245
}
245246

246-
public void testListenerExceptionCallsGenericExceptionHandler() throws IOException {
247-
RuntimeException listenerException = new RuntimeException();
248-
handler.taskException(listenerException);
249-
verify(genericExceptionHandler).accept(listenerException);
247+
public void testHandleTaskWillRunTask() throws Exception {
248+
AtomicBoolean isRun = new AtomicBoolean(false);
249+
handler.handleTask(() -> isRun.set(true));
250+
assertTrue(isRun.get());
251+
}
252+
253+
public void testTaskExceptionWillCallExceptionHandler() throws Exception {
254+
RuntimeException exception = new RuntimeException();
255+
handler.taskException(exception);
256+
verify(genericExceptionHandler).accept(exception);
250257
}
251258

252259
private class DoNotRegisterSocketContext extends BytesChannelContext {

libs/nio/src/test/java/org/elasticsearch/nio/NioSelectorTests.java

Lines changed: 43 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,7 @@
4141
import static org.mockito.Matchers.anyInt;
4242
import static org.mockito.Matchers.isNull;
4343
import static org.mockito.Matchers.same;
44+
import static org.mockito.Mockito.doAnswer;
4445
import static org.mockito.Mockito.doThrow;
4546
import static org.mockito.Mockito.mock;
4647
import static org.mockito.Mockito.times;
@@ -86,6 +87,10 @@ public void setUp() throws Exception {
8687
when(serverChannelContext.isOpen()).thenReturn(true);
8788
when(serverChannelContext.getSelector()).thenReturn(selector);
8889
when(serverChannelContext.getSelectionKey()).thenReturn(selectionKey);
90+
doAnswer(invocationOnMock -> {
91+
((Runnable) invocationOnMock.getArguments()[0]).run();
92+
return null;
93+
}).when(eventHandler).handleTask(any());
8994
}
9095

9196
@SuppressWarnings({"unchecked", "rawtypes"})
@@ -102,6 +107,23 @@ public void testQueueChannelForClosed() throws IOException {
102107
verify(eventHandler).handleClose(context);
103108
}
104109

110+
@SuppressWarnings({"unchecked", "rawtypes"})
111+
public void testCloseException() throws IOException {
112+
IOException ioException = new IOException();
113+
NioChannel channel = mock(NioChannel.class);
114+
ChannelContext context = mock(ChannelContext.class);
115+
when(channel.getContext()).thenReturn(context);
116+
when(context.getSelector()).thenReturn(selector);
117+
118+
selector.queueChannelClose(channel);
119+
120+
doThrow(ioException).when(eventHandler).handleClose(context);
121+
122+
selector.singleLoop();
123+
124+
verify(eventHandler).closeException(context, ioException);
125+
}
126+
105127
public void testNioDelayedTasksAreExecuted() throws IOException {
106128
AtomicBoolean isRun = new AtomicBoolean(false);
107129
long nanoTime = System.nanoTime() - 1;
@@ -113,9 +135,27 @@ public void testNioDelayedTasksAreExecuted() throws IOException {
113135
assertTrue(isRun.get());
114136
}
115137

138+
public void testTaskExceptionsAreHandled() {
139+
RuntimeException taskException = new RuntimeException();
140+
long nanoTime = System.nanoTime() - 1;
141+
Runnable task = () -> {
142+
throw taskException;
143+
};
144+
selector.getTaskScheduler().scheduleAtRelativeTime(task, nanoTime);
145+
146+
doAnswer((a) -> {
147+
task.run();
148+
return null;
149+
}).when(eventHandler).handleTask(same(task));
150+
151+
selector.singleLoop();
152+
verify(eventHandler).taskException(taskException);
153+
}
154+
116155
public void testDefaultSelectorTimeoutIsUsedIfNoTaskSooner() throws IOException {
117156
long delay = new TimeValue(15, TimeUnit.MINUTES).nanos();
118-
selector.getTaskScheduler().scheduleAtRelativeTime(() -> {}, System.nanoTime() + delay);
157+
selector.getTaskScheduler().scheduleAtRelativeTime(() -> {
158+
}, System.nanoTime() + delay);
119159

120160
selector.singleLoop();
121161
verify(rawSelector).select(300);
@@ -127,7 +167,8 @@ public void testSelectorTimeoutWillBeReducedIfTaskSooner() throws Exception {
127167
assertBusy(() -> {
128168
ArgumentCaptor<Long> captor = ArgumentCaptor.forClass(Long.class);
129169
long delay = new TimeValue(50, TimeUnit.MILLISECONDS).nanos();
130-
selector.getTaskScheduler().scheduleAtRelativeTime(() -> {}, System.nanoTime() + delay);
170+
selector.getTaskScheduler().scheduleAtRelativeTime(() -> {
171+
}, System.nanoTime() + delay);
131172
selector.singleLoop();
132173
verify(rawSelector).select(captor.capture());
133174
assertTrue(captor.getValue() > 0);
@@ -455,23 +496,4 @@ public void testCleanup() throws Exception {
455496
verify(eventHandler).handleClose(channelContext);
456497
verify(eventHandler).handleClose(unregisteredContext);
457498
}
458-
459-
public void testExecuteListenerWillHandleException() throws Exception {
460-
RuntimeException exception = new RuntimeException();
461-
doThrow(exception).when(listener).accept(null, null);
462-
463-
selector.executeListener(listener, null);
464-
465-
verify(eventHandler).taskException(exception);
466-
}
467-
468-
public void testExecuteFailedListenerWillHandleException() throws Exception {
469-
IOException ioException = new IOException();
470-
RuntimeException exception = new RuntimeException();
471-
doThrow(exception).when(listener).accept(null, ioException);
472-
473-
selector.executeFailedListener(listener, ioException);
474-
475-
verify(eventHandler).taskException(exception);
476-
}
477499
}

test/framework/src/main/java/org/elasticsearch/transport/nio/MockNioTransport.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -95,7 +95,7 @@ protected void doStart() {
9595
boolean success = false;
9696
try {
9797
nioGroup = new NioSelectorGroup(daemonThreadFactory(this.settings, TcpTransport.TRANSPORT_WORKER_THREAD_NAME_PREFIX), 2,
98-
(s) -> new TestingSocketEventHandler(this::onNonChannelException, s));
98+
(s) -> new TestEventHandler(this::onNonChannelException, s, System::nanoTime));
9999

100100
ProfileSettings clientProfileSettings = new ProfileSettings(settings, "default");
101101
clientChannelFactory = new MockTcpChannelFactory(true, clientProfileSettings, "client");

0 commit comments

Comments
 (0)