JAVA-1182: Throw error when synchronous call made on I/O thread.

Author: olim7t

changelog/README.md

@@ -3,6 +3,7 @@
 ### 2.0.12.2 (in progress)
 
 - [bug] JAVA-1179: Request objects should be copied when executed.
+- [improvement] JAVA-1182: Throw error when synchronous call made on I/O thread.
 
 
 ### 2.0.12.1
@@ -11,6 +12,7 @@
 - [improvement] JAVA-805: Document that metrics are null until Cluster is initialized.
 - [bug] JAVA-1072: Ensure defunct connections are properly evicted from the pool.
 
+
 ### 2.0.12
 
 - [bug] JAVA-950: Fix Cluster.connect with a case-sensitive keyspace.

driver-core/src/main/java/com/datastax/driver/core/AbstractSession.java

@@ -51,6 +51,7 @@ public ResultSet execute(String query, Object... values) {
  */
  @Override
  public ResultSet execute(Statement statement) {
+ checkNotInEventLoop();
  return executeAsync(statement).getUninterruptibly();
  }
 
@@ -75,6 +76,7 @@ public ResultSetFuture executeAsync(String query, Object... values) {
  */
  @Override
  public PreparedStatement prepare(String query) {
+ checkNotInEventLoop();
  try {
  return Uninterruptibles.getUninterruptibly(prepareAsync(query));
  } catch (ExecutionException e) {
@@ -87,6 +89,7 @@ public PreparedStatement prepare(String query) {
  */
  @Override
  public PreparedStatement prepare(RegularStatement statement) {
+ checkNotInEventLoop();
  try {
  return Uninterruptibles.getUninterruptibly(prepareAsync(statement));
  } catch (ExecutionException e) {
@@ -132,4 +135,13 @@ public void close() {
  Thread.currentThread().interrupt();
  }
  }
+
+ /**
+ * Checks that the current thread is not one of the Netty I/O threads used by the driver.
+ * <p/>
+ * This is called from the synchronous methods of this class to prevent deadlock issues.
+ */
+ protected void checkNotInEventLoop() {
+ // This method is concrete only to avoid a breaking change. See subclass for the actual implementation.
+ }
 }

driver-core/src/main/java/com/datastax/driver/core/Connection.java

@@ -679,7 +679,7 @@ public static class Factory {
 
  public final Timer timer;
 
- private final EventLoopGroup eventLoopGroup;
+ final EventLoopGroup eventLoopGroup;
  private final Class<? extends Channel> channelClass;
 
  private final ChannelGroup allChannels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

driver-core/src/main/java/com/datastax/driver/core/SessionManager.java

@@ -27,6 +27,7 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Lists;
 import com.google.common.util.concurrent.*;
+import io.netty.util.concurrent.EventExecutor;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -46,6 +47,9 @@ class SessionManager extends AbstractSession {
 
  private static final Logger logger = LoggerFactory.getLogger(Session.class);
 
+ private static final boolean CHECK_IO_DEADLOCKS = SystemProperties.getBoolean(
+ "com.datastax.driver.CHECK_IO_DEADLOCKS", true);
+
  final Cluster cluster;
  final ConcurrentMap<Host, HostConnectionPool> pools;
  final HostConnectionPool.PoolState poolsState;
@@ -611,6 +615,21 @@ void cleanupIdleConnections(long now) {
  }
  }
 
+ @Override
+ protected void checkNotInEventLoop() {
+ if (!CHECK_IO_DEADLOCKS)
+ return;
+ for (EventExecutor executor : cluster.manager.connectionFactory.eventLoopGroup) {
+ if (executor.inEventLoop()) {
+ throw new IllegalStateException(
+ "Detected a synchronous Session call (execute() or prepare()) on an I/O thread, " +
+ "this can cause deadlocks or unpredictable behavior. " +
+ "Make sure your Future callbacks only use async calls, or schedule them on a " +
+ "different executor.");
+ }
+ }
+ }
+
  private static class State implements Session.State {
 
  private final SessionManager session;

driver-core/src/main/java/com/datastax/driver/core/SystemProperties.java

@@ -21,7 +21,7 @@
 /**
  * Allows overriding internal settings via system properties.
  * <p/>
- * Warning: this is meant for integration tests only, NOT FOR PRODUCTION USE.
+ * This is generally reserved for tests or "expert" usage.
  */
 class SystemProperties {
  private static final Logger logger = LoggerFactory.getLogger(SystemProperties.class);
@@ -34,7 +34,7 @@ static int getInt(String key, int defaultValue) {
  }
  try {
  int value = Integer.parseInt(stringValue);
- logger.warn("{} is defined, using value {}", key, value);
+ logger.info("{} is defined, using value {}", key, value);
  return value;
  } catch (NumberFormatException e) {
  logger.warn("{} is defined but could not parse value {}, using default value {}", key, stringValue, defaultValue);
@@ -50,7 +50,7 @@ static boolean getBoolean(String key, boolean defaultValue) {
  }
  try {
  boolean value = Boolean.parseBoolean(stringValue);
- logger.warn("{} is defined, using value {}", key, value);
+ logger.info("{} is defined, using value {}", key, value);
  return value;
  } catch (NumberFormatException e) {
  logger.warn("{} is defined but could not parse value {}, using default value {}", key, stringValue, defaultValue);

driver-core/src/test/java/com/datastax/driver/core/AsyncQueryTest.java

@@ -17,6 +17,7 @@
 
 import com.datastax.driver.core.exceptions.InvalidQueryException;
 import com.google.common.base.Function;
+import com.google.common.base.Throwables;
 import com.google.common.collect.Lists;
 import com.google.common.util.concurrent.*;
 import org.testng.annotations.DataProvider;
@@ -27,6 +28,7 @@
 import java.util.concurrent.*;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.fail;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertTrue;
 
@@ -129,6 +131,26 @@ public void should_propagate_error_to_chained_query_if_session_init_fails() thro
  }
  }
 
+ @Test(groups = "short")
+ public void should_fail_when_synchronous_call_on_io_thread() throws Exception {
+ ResultSetFuture f = session.executeAsync("select release_version from system.local");
+ ListenableFuture<Void> f2 = Futures.transform(f, new Function<ResultSet, Void>() {
+ @Override
+ public Void apply(ResultSet input) {
+ session.execute("select release_version from system.local");
+ return null;
+ }
+ });
+ try {
+ f2.get();
+ fail("Expected a failed future");
+ } catch (Exception e) {
+ assertThat(Throwables.getRootCause(e))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("Detected a synchronous Session call");
+ }
+ }
+
  private ListenableFuture<Integer> connectAndQuery(String keyspace, Executor executor) {
  ListenableFuture<Session> sessionFuture = cluster.connectAsync(keyspace);
  ListenableFuture<ResultSet> queryFuture = Futures.transform(sessionFuture, new AsyncFunction<Session, ResultSet>() {