Merge pull request #57 from oracle/async-lock-2

Eliminating JDBC Lock Contention

Author: jeandelavarene

src/main/java/oracle/r2dbc/impl/AsyncLock.java

@@ -22,6 +22,7 @@
 package oracle.r2dbc.impl;
 
 import java.sql.Connection;
+import java.sql.Statement;
 import java.time.Duration;
 import java.util.LinkedList;
 import java.util.Queue;
@@ -114,18 +115,6 @@ public Batch add(String sql) {
    * are executed in the order they were added. Calling this method clears all
    * statements that have been added to the current batch.
    * </p><p>
-   * A {@code Result} emitted by the returned {@code Publisher} must be
-   * <a href="OracleStatementImpl.html#fully-consumed-result">
-   *   fully-consumed
-   * </a>
-   * before the next {@code Result} is emitted. This ensures that a command in
-   * the batch can not be executed while the {@code Result} of a previous
-   * command is consumed concurrently. It is a known limitation of the Oracle
-   * R2DBC Driver that concurrent operations on a single {@code Connection}
-   * will result in blocked threads. Deferring {@code Statement} execution
-   * until full consumption of the previous {@code Statement}'s {@code Result}
-   * is necessary in order to avoid blocked threads.
-   * </p><p>
    * If the execution of any statement in the sequence results in a failure,
    * then the returned publisher emits {@code onError} with an
    * {@link R2dbcException} that describes the failure, and all subsequent
@@ -147,42 +136,8 @@ public Publisher<OracleResultImpl> execute() {
     requireOpenConnection(jdbcConnection);
     Queue<OracleStatementImpl> currentStatements = statements;
     statements = new LinkedList<>();
-    return publishBatch(currentStatements);
-  }
-
-  /**
-   * <p>
-   * Publish a batch of {@code Result}s from {@code statements}. Each
-   * {@code Result} is published serially with the consumption of the
-   * previous {@code Result}.
-   * </p><p>
-   * This method returns an empty {@code Publisher} if {@code statements} is
-   * empty. Otherwise, this method dequeues the next {@code Statement} and
-   * executes it for a {@code Result}. After the {@code Result} has been
-   * fully consumed, this method is invoked recursively to publish the {@code
-   * Result}s of remaining {@code statements}.
-   * </p>
-   * @param statements A batch to executed.
-   * @return {@code Publisher} of {@code statements} {@code Result}s
-   */
-  private static Publisher<OracleResultImpl> publishBatch(
-    Queue<OracleStatementImpl> statements) {
-
-    OracleStatementImpl next = statements.poll();
-
-    if (next != null) {
-      AtomicReference<OracleResultImpl> lastResult =
-        new AtomicReference<>(null);
-
-      return Flux.from(next.execute())
-        .doOnNext(lastResult::set)
-        .concatWith(Flux.defer(() ->
-          Mono.from(lastResult.get().onConsumed())
-            .thenMany(publishBatch(statements))));
-    }
-    else {
-      return Mono.empty();
-    }
+    return Flux.fromIterable(currentStatements)
+      .concatMap(OracleStatementImpl::execute);
   }
 
 }

src/main/java/oracle/r2dbc/impl/OracleBatchImpl.java

@@ -470,6 +470,7 @@ public boolean isAutoCommit() {
   @Override
   public ConnectionMetadata getMetadata() {
     requireOpenConnection(jdbcConnection);
+    // TODO: Initialize this on construction, to avoid lock contention
     return new OracleConnectionMetadataImpl(
       fromJdbc(jdbcConnection::getMetaData));
   }
@@ -575,7 +576,7 @@ public Publisher<Void> rollbackTransactionToSavepoint(String name) {
   @Override
   public Publisher<Void> setAutoCommit(boolean autoCommit) {
     requireOpenConnection(jdbcConnection);
-    return Mono.defer(() -> fromJdbc(() -> {
+    return Mono.from(adapter.getLock().flatMap(() -> {
       if (autoCommit == jdbcConnection.getAutoCommit()) {
         return Mono.empty(); // No change
       }
@@ -589,7 +590,7 @@ else if (! autoCommit) {
         // Changing auto-commit from disabled to enabled. Commit in case
         // there is an active transaction.
         return Mono.from(commitTransaction())
-          .doOnSuccess(nil -> runJdbc(() ->
+          .concatWith(adapter.getLock().run(() ->
             jdbcConnection.setAutoCommit(true)));
       }
     }))

src/main/java/oracle/r2dbc/impl/OracleConnectionImpl.java

@@ -240,7 +240,7 @@ else if (sqlException instanceof SQLRecoverableException) {
    *   null.
    * @throws R2dbcException If the supplier throws a {@code SQLException}.
    */
-  static void runJdbc(ThrowingRunnable runnable)
+  static void runJdbc(JdbcRunnable runnable)
     throws R2dbcException {
     try {
       runnable.runOrThrow();
@@ -273,7 +273,7 @@ static void runJdbc(ThrowingRunnable runnable)
    * @return The output of the specified {@code supplier}.
    * @throws R2dbcException If the supplier throws a {@code SQLException}.
    */
-  static <T> T fromJdbc(ThrowingSupplier<T> supplier)
+  static <T> T fromJdbc(JdbcSupplier<T> supplier)
     throws R2dbcException {
     try {
       return supplier.getOrThrow();
@@ -330,7 +330,7 @@ private static String getSql(SQLException sqlException) {
    * </p>
    */
   @FunctionalInterface
-  interface ThrowingRunnable extends Runnable {
+  interface JdbcRunnable extends Runnable {
     /**
      * Runs to completion and returns normally, or throws a {@code SQLException}
      * if an error is encountered.
@@ -343,8 +343,7 @@ interface ThrowingRunnable extends Runnable {
      * R2dbcException} if an error is encountered.
      * @throws R2dbcException If the run does not complete due to an error.
      * @implNote The default implementation invokes
-     * {@link #runJdbc(ThrowingRunnable)} with this {@code
-     * ThrowingRunnable}.
+     * {@link #runJdbc(JdbcRunnable)} with this {@code JdbcRunnable}.
      */
     @Override
     default void run() throws R2dbcException {
@@ -362,7 +361,7 @@ default void run() throws R2dbcException {
    * @param <T> the type of values supplied by this supplier.
    */
   @FunctionalInterface
-  interface ThrowingSupplier<T> extends Supplier<T> {
+  interface JdbcSupplier<T> extends Supplier<T> {
     /**
      * Returns a value, or throws a {@code SQLException} if an error is
      * encountered.
@@ -376,8 +375,7 @@ interface ThrowingSupplier<T> extends Supplier<T> {
      * encountered.
      * @throws R2dbcException If a value is not returned due to an error.
      * @implNote The default implementation invokes
-     * {@link #fromJdbc(ThrowingSupplier)} (ThrowingRunnable)}
-     * with this {@code ThrowingSupplier}.
+     * {@link #fromJdbc(JdbcSupplier)} with this {@code JdbcSupplier}.
      */
     @Override
     default T get() throws R2dbcException {