Skip to content

Commit 4f5417f

Browse files
stIncMalestIncMale
authored
Fix how InterruptedExceptions are handled (#1192)
JAVA-4641
1 parent 90c7062 commit 4f5417f

File tree

24 files changed

+177
-121
lines changed

24 files changed

+177
-121
lines changed

‎driver-core/src/main/com/mongodb/MongoInterruptedException.java‎

Lines changed: 23 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -18,11 +18,32 @@
1818

1919
import com.mongodb.lang.Nullable;
2020

21+
import java.io.InputStream;
22+
import java.io.InterruptedIOException;
23+
import java.io.OutputStream;
24+
import java.net.Socket;
25+
import java.net.SocketAddress;
26+
import java.net.SocketException;
27+
import java.nio.channels.ClosedByInterruptException;
28+
import java.nio.channels.InterruptibleChannel;
29+
2130
/**
22-
* A non-checked exception indicating that the driver has been interrupted by a call to Thread.interrupt.
31+
* A driver-specific non-checked counterpart to {@link InterruptedException}.
32+
* Before this exception is thrown, the {@linkplain Thread#isInterrupted() interrupt status} of the thread will have been set
33+
* unless the {@linkplain #getCause() cause} is {@link InterruptedIOException}, in which case the driver leaves the status as is.
34+
* <p>
35+
* The Java SE API uses exceptions different from {@link InterruptedException} to communicate the same information:</p>
36+
* <ul>
37+
* <li>{@link InterruptibleChannel} uses {@link ClosedByInterruptException}.</li>
38+
* <li>{@link Socket#connect(SocketAddress)},
39+
* {@linkplain InputStream}/{@link OutputStream} obtained via {@link Socket#getInputStream()}/{@link Socket#getOutputStream()}
40+
* use either {@link ClosedByInterruptException} or {@link SocketException}.</li>
41+
* <li>There is also {@link InterruptedIOException}, which is documented to an extent as an IO-specific counterpart to
42+
* {@link InterruptedException}.</li>
43+
* </ul>
44+
* The driver strives to wrap those in {@link MongoInterruptedException} where relevant.
2345
*
2446
* @see Thread#interrupt()
25-
* @see InterruptedException
2647
*/
2748
public class MongoInterruptedException extends MongoException {
2849
private static final long serialVersionUID = -4110417867718417860L;

‎driver-core/src/main/com/mongodb/connection/netty/NettyStream.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,6 @@
1919
import com.mongodb.MongoClientException;
2020
import com.mongodb.MongoException;
2121
import com.mongodb.MongoInternalException;
22-
import com.mongodb.MongoInterruptedException;
2322
import com.mongodb.MongoSocketException;
2423
import com.mongodb.MongoSocketOpenException;
2524
import com.mongodb.MongoSocketReadTimeoutException;
@@ -71,6 +70,7 @@
7170
import static com.mongodb.assertions.Assertions.isTrueArgument;
7271
import static com.mongodb.internal.connection.SslHelper.enableHostNameVerification;
7372
import static com.mongodb.internal.connection.SslHelper.enableSni;
73+
import static com.mongodb.internal.thread.InterruptionUtil.interruptAndCreateMongoInterruptedException;
7474
import static java.util.Optional.ofNullable;
7575
import static java.util.concurrent.TimeUnit.MILLISECONDS;
7676

@@ -436,7 +436,7 @@ private void addSslHandler(final SocketChannel channel) {
436436

437437
private class InboundBufferHandler extends SimpleChannelInboundHandler<io.netty.buffer.ByteBuf> {
438438
@Override
439-
protected void channelRead0(final ChannelHandlerContext ctx, final io.netty.buffer.ByteBuf buffer) throws Exception {
439+
protected void channelRead0(final ChannelHandlerContext ctx, final io.netty.buffer.ByteBuf buffer) {
440440
handleReadResponse(buffer, null);
441441
}
442442

@@ -499,7 +499,7 @@ public T get() throws IOException {
499499
}
500500
return t;
501501
} catch (InterruptedException e) {
502-
throw new MongoInterruptedException("Interrupted", e);
502+
throw interruptAndCreateMongoInterruptedException("Interrupted", e);
503503
}
504504
}
505505
}

‎driver-core/src/main/com/mongodb/internal/Locks.java‎

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -16,11 +16,11 @@
1616

1717
package com.mongodb.internal;
1818

19-
import com.mongodb.MongoInterruptedException;
20-
2119
import java.util.concurrent.locks.Lock;
2220
import java.util.function.Supplier;
2321

22+
import static com.mongodb.internal.thread.InterruptionUtil.interruptAndCreateMongoInterruptedException;
23+
2424
/**
2525
* <p>This class is not part of the public API and may be removed or changed at any time</p>
2626
*/
@@ -45,8 +45,7 @@ public static <V, E extends Exception> V checkedWithLock(final Lock lock, final
4545
lock.unlock();
4646
}
4747
} catch (InterruptedException e) {
48-
Thread.currentThread().interrupt();
49-
throw new MongoInterruptedException("Interrupted waiting for lock", e);
48+
throw interruptAndCreateMongoInterruptedException("Interrupted waiting for lock", e);
5049
}
5150
}
5251

‎driver-core/src/main/com/mongodb/internal/connection/AsynchronousChannelStream.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,6 @@
1818

1919
import com.mongodb.MongoException;
2020
import com.mongodb.MongoInternalException;
21-
import com.mongodb.MongoInterruptedException;
2221
import com.mongodb.MongoSocketReadException;
2322
import com.mongodb.MongoSocketReadTimeoutException;
2423
import com.mongodb.ServerAddress;
@@ -38,6 +37,7 @@
3837
import java.util.concurrent.atomic.AtomicReference;
3938

4039
import static com.mongodb.assertions.Assertions.assertTrue;
40+
import static com.mongodb.internal.thread.InterruptionUtil.interruptAndCreateMongoInterruptedException;
4141
import static java.util.concurrent.TimeUnit.MILLISECONDS;
4242

4343
/**
@@ -317,7 +317,7 @@ private T get(final String prefix) throws IOException {
317317
try {
318318
latch.await();
319319
} catch (InterruptedException e) {
320-
throw new MongoInterruptedException(prefix + " the AsynchronousSocketChannelStream failed", e);
320+
throw interruptAndCreateMongoInterruptedException(prefix + " the AsynchronousSocketChannelStream failed", e);
321321

322322
}
323323
if (error != null) {

‎driver-core/src/main/com/mongodb/internal/connection/BaseCluster.java‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,6 @@
1818

1919
import com.mongodb.MongoClientException;
2020
import com.mongodb.MongoIncompatibleDriverException;
21-
import com.mongodb.MongoInterruptedException;
2221
import com.mongodb.MongoTimeoutException;
2322
import com.mongodb.ServerAddress;
2423
import com.mongodb.connection.ClusterDescription;
@@ -60,6 +59,7 @@
6059
import static com.mongodb.internal.VisibleForTesting.AccessModifier.PRIVATE;
6160
import static com.mongodb.internal.connection.EventHelper.wouldDescriptionsGenerateEquivalentEvents;
6261
import static com.mongodb.internal.event.EventListenerHelper.singleClusterListener;
62+
import static com.mongodb.internal.thread.InterruptionUtil.interruptAndCreateMongoInterruptedException;
6363
import static java.lang.String.format;
6464
import static java.util.Arrays.asList;
6565
import static java.util.Comparator.comparingInt;
@@ -142,7 +142,7 @@ public ServerTuple selectServer(final ServerSelector serverSelector, final Opera
142142
}
143143

144144
} catch (InterruptedException e) {
145-
throw new MongoInterruptedException(format("Interrupted while waiting for a server that matches %s", serverSelector), e);
145+
throw interruptAndCreateMongoInterruptedException(format("Interrupted while waiting for a server that matches %s", serverSelector), e);
146146
}
147147
}
148148

@@ -211,7 +211,7 @@ public ClusterDescription getDescription() {
211211
}
212212
return curDescription;
213213
} catch (InterruptedException e) {
214-
throw new MongoInterruptedException("Interrupted while waiting to connect", e);
214+
throw interruptAndCreateMongoInterruptedException("Interrupted while waiting to connect", e);
215215
}
216216
}
217217

@@ -516,7 +516,7 @@ public void run() {
516516

517517
try {
518518
currentPhase.await(waitTimeNanos, NANOSECONDS);
519-
} catch (InterruptedException e) {
519+
} catch (InterruptedException closed) {
520520
// The cluster has been closed and the while loop will exit.
521521
}
522522
}

‎driver-core/src/main/com/mongodb/internal/connection/ConcurrentPool.java‎

Lines changed: 7 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,7 @@
4343
import static com.mongodb.assertions.Assertions.assertTrue;
4444
import static com.mongodb.assertions.Assertions.notNull;
4545
import static com.mongodb.internal.VisibleForTesting.AccessModifier.PRIVATE;
46+
import static com.mongodb.internal.thread.InterruptionUtil.interruptAndCreateMongoInterruptedException;
4647

4748
/**
4849
* A concurrent pool implementation.
@@ -411,7 +412,7 @@ boolean acquirePermit(final long timeout, final TimeUnit unit) throws MongoInter
411412
return false;
412413
}
413414
} catch (InterruptedException e) {
414-
throw new MongoInterruptedException(null, e);
415+
throw interruptAndCreateMongoInterruptedException(null, e);
415416
} finally {
416417
waitersEstimate.decrementAndGet();
417418
}
@@ -518,19 +519,20 @@ static void lockInterruptibly(final Lock lock) throws MongoInterruptedException
518519
try {
519520
lock.lockInterruptibly();
520521
} catch (InterruptedException e) {
521-
throw new MongoInterruptedException(null, e);
522+
throw interruptAndCreateMongoInterruptedException(null, e);
522523
}
523524
}
524525

525526
private static void lockInterruptiblyUnfair(final ReentrantLock lock) throws MongoInterruptedException {
526-
throwIfInterrupted();
527+
if (Thread.currentThread().isInterrupted()) {
528+
throw interruptAndCreateMongoInterruptedException(null, null);
529+
}
527530
// `ReentrantLock.tryLock` is unfair
528531
if (!lock.tryLock()) {
529532
try {
530533
lock.lockInterruptibly();
531534
} catch (InterruptedException e) {
532-
Thread.currentThread().interrupt();
533-
throw new MongoInterruptedException(null, e);
535+
throw interruptAndCreateMongoInterruptedException(null, e);
534536
}
535537
}
536538
}
@@ -541,10 +543,4 @@ static void lockUnfair(final ReentrantLock lock) {
541543
lock.lock();
542544
}
543545
}
544-
545-
private static void throwIfInterrupted() throws MongoInterruptedException {
546-
if (Thread.currentThread().isInterrupted()) {
547-
throw new MongoInterruptedException(null, null);
548-
}
549-
}
550546
}

‎driver-core/src/main/com/mongodb/internal/connection/DefaultConnectionPool.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -119,6 +119,7 @@
119119
import static com.mongodb.internal.logging.LogMessage.Entry.Name.SERVICE_ID;
120120
import static com.mongodb.internal.logging.LogMessage.Entry.Name.WAIT_QUEUE_TIMEOUT_MS;
121121
import static com.mongodb.internal.logging.LogMessage.Level.DEBUG;
122+
import static com.mongodb.internal.thread.InterruptionUtil.interruptAndCreateMongoInterruptedException;
122123
import static java.lang.String.format;
123124
import static java.util.concurrent.TimeUnit.MILLISECONDS;
124125
import static java.util.concurrent.TimeUnit.NANOSECONDS;
@@ -1178,7 +1179,7 @@ private long awaitNanos(final Condition condition, final long timeoutNanos) thro
11781179
return Math.max(0, condition.awaitNanos(timeoutNanos));
11791180
}
11801181
} catch (InterruptedException e) {
1181-
throw new MongoInterruptedException(null, e);
1182+
throw interruptAndCreateMongoInterruptedException(null, e);
11821183
}
11831184
}
11841185
}

‎driver-core/src/main/com/mongodb/internal/connection/DefaultDnsSrvRecordMonitor.java‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -106,7 +106,7 @@ public void run() {
106106

107107
try {
108108
Thread.sleep(getRescanFrequencyMillis());
109-
} catch (InterruptedException e) {
109+
} catch (InterruptedException closed) {
110110
// fall through
111111
}
112112
clusterType = dnsSrvRecordInitializer.getClusterType();
@@ -130,4 +130,3 @@ private Set<ServerAddress> createServerAddressSet(final List<String> resolvedHos
130130
}
131131
}
132132
}
133-

‎driver-core/src/main/com/mongodb/internal/connection/DefaultServerMonitor.java‎

Lines changed: 15 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -180,8 +180,8 @@ public void run() {
180180
}
181181
waitForNext();
182182
}
183-
} catch (MongoInterruptedException e) {
184-
// ignore
183+
} catch (InterruptedException | MongoInterruptedException closed) {
184+
// stop the monitor
185185
} catch (RuntimeException e) {
186186
LOGGER.error(format("Server monitor for %s exiting with exception", serverId), e);
187187
} finally {
@@ -285,21 +285,17 @@ private void logStateChange(final ServerDescription previousServerDescription,
285285
}
286286
}
287287

288-
private void waitForNext() {
289-
try {
290-
long timeRemaining = waitForSignalOrTimeout();
291-
if (timeRemaining > 0) {
292-
long timeWaiting = serverSettings.getHeartbeatFrequency(NANOSECONDS) - timeRemaining;
293-
long minimumNanosToWait = serverSettings.getMinHeartbeatFrequency(NANOSECONDS);
294-
if (timeWaiting < minimumNanosToWait) {
295-
long millisToSleep = MILLISECONDS.convert(minimumNanosToWait - timeWaiting, NANOSECONDS);
296-
if (millisToSleep > 0) {
297-
Thread.sleep(millisToSleep);
298-
}
288+
private void waitForNext() throws InterruptedException {
289+
long timeRemaining = waitForSignalOrTimeout();
290+
if (timeRemaining > 0) {
291+
long timeWaiting = serverSettings.getHeartbeatFrequency(NANOSECONDS) - timeRemaining;
292+
long minimumNanosToWait = serverSettings.getMinHeartbeatFrequency(NANOSECONDS);
293+
if (timeWaiting < minimumNanosToWait) {
294+
long millisToSleep = MILLISECONDS.convert(minimumNanosToWait - timeWaiting, NANOSECONDS);
295+
if (millisToSleep > 0) {
296+
Thread.sleep(millisToSleep);
299297
}
300298
}
301-
} catch (InterruptedException e) {
302-
// fall through
303299
}
304300
}
305301

@@ -429,6 +425,8 @@ public void run() {
429425
}
430426
waitForNext();
431427
}
428+
} catch (InterruptedException closed) {
429+
// stop the monitor
432430
} finally {
433431
if (connection != null) {
434432
connection.close();
@@ -453,12 +451,8 @@ private void pingServer(final InternalConnection connection) {
453451
}
454452
}
455453

456-
private void waitForNext() {
457-
try {
458-
Thread.sleep(serverSettings.getHeartbeatFrequency(MILLISECONDS));
459-
} catch (InterruptedException e) {
460-
// fall through
461-
}
454+
private void waitForNext() throws InterruptedException {
455+
Thread.sleep(serverSettings.getHeartbeatFrequency(MILLISECONDS));
462456
}
463457

464458
private String getHandshakeCommandName(final ServerDescription serverDescription) {

‎driver-core/src/main/com/mongodb/internal/connection/FutureAsyncCompletionHandler.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -18,13 +18,14 @@
1818

1919
import com.mongodb.MongoException;
2020
import com.mongodb.MongoInternalException;
21-
import com.mongodb.MongoInterruptedException;
2221
import com.mongodb.connection.AsyncCompletionHandler;
2322
import com.mongodb.lang.Nullable;
2423

2524
import java.io.IOException;
2625
import java.util.concurrent.CountDownLatch;
2726

27+
import static com.mongodb.internal.thread.InterruptionUtil.interruptAndCreateMongoInterruptedException;
28+
2829
class FutureAsyncCompletionHandler<T> implements AsyncCompletionHandler<T> {
2930
private final CountDownLatch latch = new CountDownLatch(1);
3031
private volatile T result;
@@ -58,7 +59,7 @@ private T get(final String prefix) throws IOException {
5859
try {
5960
latch.await();
6061
} catch (InterruptedException e) {
61-
throw new MongoInterruptedException(prefix + " the AsynchronousSocketChannelStream failed", e);
62+
throw interruptAndCreateMongoInterruptedException(prefix + " the AsynchronousSocketChannelStream failed", e);
6263

6364
}
6465
if (error != null) {

0 commit comments

Comments
 (0)