From e5209af6a65edf2c2a772e58aefa0056080a6f7f Mon Sep 17 00:00:00 2001 From: fhanik Date: Wed, 26 Nov 2008 01:14:32 +0000 Subject: [PATCH] Add the ability to retrieve connection object asynchronously git-svn-id: https://svn.apache.org/repos/asf/tomcat/trunk@720693 13f79535-47bb-0310-9956-ffa450edef68 --- modules/jdbc-pool/doc/jdbc-pool.xml | 34 +++++++++- .../apache/tomcat/jdbc/pool/ConnectionPool.java | 75 +++++++++++++++++++- .../apache/tomcat/jdbc/pool/DataSourceFactory.java | 2 + .../apache/tomcat/jdbc/pool/DataSourceProxy.java | 12 ++++ .../apache/tomcat/jdbc/test/DefaultTestCase.java | 2 +- .../org/apache/tomcat/jdbc/test/FairnessTest.java | 40 ++++++++++- .../tomcat/jdbc/test/SimplePOJOAsyncExample.java | 79 ++++++++++++++++++++++ .../apache/tomcat/jdbc/test/TwoDataSources.java | 6 +- 8 files changed, 242 insertions(+), 8 deletions(-) create mode 100644 modules/jdbc-pool/test/org/apache/tomcat/jdbc/test/SimplePOJOAsyncExample.java diff --git a/modules/jdbc-pool/doc/jdbc-pool.xml b/modules/jdbc-pool/doc/jdbc-pool.xml index 5fc06be70..ff8716e15 100644 --- a/modules/jdbc-pool/doc/jdbc-pool.xml +++ b/modules/jdbc-pool/doc/jdbc-pool.xml @@ -48,7 +48,9 @@
  • commons-dbcp uses static interfaces. This means you can't compile it with JDK 1.6, or if you run on JDK 1.6/1.7 you will get NoSuchMethodException for all the methods not implemented, even if the driver supports it.
  • The commons-dbcp has become fairly stagnant. Sparse updates, releases, and new feature support.
    • -
  • It's not worth rewriting over 60 classes, when something as a connection pool can be accomplished with as a much simpler implementation.
    • +
  • It's not worth rewriting over 60 classes, when something as a connection pool can be accomplished with as a much simpler implementation.
    • +
  • Tomcat jdbc pool implements a fairness option not available in commons-dbcp and still performs faster than commons-dbcp
    • +
  • Tomcat jdbc pool implements the ability retrieve a connection asynchronously, without adding additional threads to the library itself
  • Tomcat jdbc pool is a Tomcat module, it depends on Tomcat JULI, a simplified logging framework used in Tomcat.

    • @@ -70,6 +72,7 @@
  • Extremely simple, due to the very simplified implementation, the line count and source file count are very low, compare with c3p0 that has over 200 source files(last time we checked), Tomcat jdbc has a core of 8 files, the connection pool itself is about half that. As bugs may occur, they will be faster to track down, and easier to fix. Complexity reduction has been a focus from inception.
    • +
  • Asynchronous connection retrieval - you can queue your request for a connection and receive a Future<Connection> back.

    • @@ -87,6 +90,7 @@
  • validationInterval - in addition to running validations on connections, avoid running them too frequently.
  • jdbcInterceptors - flexible and pluggable interceptors to create any customizations around the pool, the query execution and the result set handling. More on this in the advanced section.
    • +
  • fairQueue - Set the fair flag to true to achieve thread fairness or to use asynchronous connection retrieval
    • @@ -322,6 +326,7 @@

    (boolean) Set to true if you wish that calls to getConnection should be treated fairly in a true FIFO fashion. This uses the org.apache.tomcat.jdbc.pool.FairBlockingQueue implementation for the list of the idle connections. The default value is false. + This flag is required when you want to use asynchronous connection retrieval.

    @@ -437,6 +442,32 @@     + +

    The Tomcat JDBC connection pool supports asynchronous connection retrieval without adding additional threads to the + pool library. It does this by adding a method to the data source called Future<Connection> getConnectionAsync(). + In order to use the async retrieval, two conditions must be met:
    + 1. You must configure the fairQueue property to be true.
    + 2. You will have to cast the data source to org.apache.tomcat.jdbc.pool.DataSource
    + An example of using the async feature is show below. + + + Connection con = null; + try { + Future<Connection> future = datasource.getConnectionAsync(); + while (!future.isDone()) { + System.out.println("Connection is not yet available. Do some background work"); + try { + Thread.sleep(100); //simulate work + }catch (InterruptedException x) { + Thread.currentThread().interrupted(); + } + } + con = future.get(); //should return instantly + Statement st = con.createStatement(); + ResultSet rs = st.executeQuery("select * from user"); + +

    +

    Interceptors are a powerful way to enable, disable or modify functionality on a specific connection or its sub components. There are many different use cases for when interceptors are useful. By default, and for performance reasons, the connection pool is stateless. @@ -499,6 +530,7 @@

    + diff --git a/modules/jdbc-pool/java/org/apache/tomcat/jdbc/pool/ConnectionPool.java b/modules/jdbc-pool/java/org/apache/tomcat/jdbc/pool/ConnectionPool.java index 4fea6e50d..751a12af6 100644 --- a/modules/jdbc-pool/java/org/apache/tomcat/jdbc/pool/ConnectionPool.java +++ b/modules/jdbc-pool/java/org/apache/tomcat/jdbc/pool/ConnectionPool.java @@ -28,14 +28,18 @@ import java.util.Queue; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; import java.util.concurrent.Callable; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import org.apache.juli.logging.Log; import org.apache.juli.logging.LogFactory; import org.apache.tomcat.jdbc.pool.jmx.ConnectionPoolMBean; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import javax.management.InstanceAlreadyExistsException; @@ -113,7 +117,12 @@ public class ConnectionPool { public Future getConnectionAsync() throws SQLException { - return null; + if (idle instanceof FairBlockingQueue) { + Future pcf = ((FairBlockingQueue)idle).pollAsync(); + return new ConnectionFuture(pcf); + } else { + throw new SQLException("Connection pool is misconfigured, doesn't support async retrieval. Set the 'fair' property to 'true'"); + } } /** @@ -726,7 +735,71 @@ public class ConnectionPool { log.warn("Unable to stop JMX integration for connection pool. Instance["+getName()+"].",x); } } + + /** + * Tread safe wrapper around a future for the regular queue + * This one retrieves the pooled connection object + * and performs the initialization according to + * interceptors and validation rules. + * This class is thread safe. + * @author fhanik + * + */ + protected class ConnectionFuture implements Future { + Future pcFuture = null; + AtomicBoolean configured = new AtomicBoolean(false); + CountDownLatch latch = new CountDownLatch(1); + Connection result = null; + SQLException cause = null; + public ConnectionFuture(Future pcf) { + this.pcFuture = pcf; + } + + public boolean cancel(boolean mayInterruptIfRunning) { + return pcFuture.cancel(mayInterruptIfRunning); + } + + public Connection get() throws InterruptedException, ExecutionException { + try { + return get(Long.MAX_VALUE, TimeUnit.MILLISECONDS); + }catch (TimeoutException x) { + throw new ExecutionException(x); + } + } + public Connection get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { + PooledConnection pc = pcFuture.get(timeout,unit); + if (pc!=null) { + if (result!=null) return result; + if (configured.compareAndSet(false, true)) { + try { + pc = borrowConnection(System.currentTimeMillis(),pc); + result = ConnectionPool.this.setupConnection(pc); + } catch (SQLException x) { + cause = x; + } finally { + latch.countDown(); + } + } else { + //if we reach here, another thread is configuring the actual connection + latch.await(timeout,unit); //this shouldn't block for long + } + if (result==null) throw new ExecutionException(cause); + return result; + } else { + return null; + } + } + + public boolean isCancelled() { + return pcFuture.isCancelled(); + } + + public boolean isDone() { + return pcFuture.isDone(); + } + + } protected class PoolCleaner extends Thread { protected ConnectionPool pool; diff --git a/modules/jdbc-pool/java/org/apache/tomcat/jdbc/pool/DataSourceFactory.java b/modules/jdbc-pool/java/org/apache/tomcat/jdbc/pool/DataSourceFactory.java index 6ad3a9bb5..7b94760b9 100644 --- a/modules/jdbc-pool/java/org/apache/tomcat/jdbc/pool/DataSourceFactory.java +++ b/modules/jdbc-pool/java/org/apache/tomcat/jdbc/pool/DataSourceFactory.java @@ -403,6 +403,8 @@ public class DataSourceFactory implements ObjectFactory { } public static DataSource getDataSource(org.apache.tomcat.jdbc.pool.DataSourceProxy dataSource) { + if (dataSource instanceof DataSource) return (DataSource)dataSource; + //only return a proxy if we didn't implement the DataSource interface DataSourceHandler handler = new DataSourceHandler(dataSource); DataSource ds = (DataSource)Proxy.newProxyInstance(DataSourceFactory.class.getClassLoader(), new Class[] {javax.sql.DataSource.class}, handler); return ds; diff --git a/modules/jdbc-pool/java/org/apache/tomcat/jdbc/pool/DataSourceProxy.java b/modules/jdbc-pool/java/org/apache/tomcat/jdbc/pool/DataSourceProxy.java index 0391001c7..f978f8319 100644 --- a/modules/jdbc-pool/java/org/apache/tomcat/jdbc/pool/DataSourceProxy.java +++ b/modules/jdbc-pool/java/org/apache/tomcat/jdbc/pool/DataSourceProxy.java @@ -20,6 +20,7 @@ import java.io.PrintWriter; import java.sql.Connection; import java.sql.SQLException; import java.util.Iterator; +import java.util.concurrent.Future; import org.apache.juli.logging.Log; import org.apache.juli.logging.LogFactory; @@ -93,6 +94,17 @@ public class DataSourceProxy { return createPool().getConnection(); return pool.getConnection(); } + + /** + * Invokes an sync operation to retrieve the connection. + * @return + * @throws SQLException + */ + public Future getConnectionAsync() throws SQLException { + if (pool == null) + return createPool().getConnectionAsync(); + return pool.getConnectionAsync(); + } /** * {@inheritDoc} diff --git a/modules/jdbc-pool/test/org/apache/tomcat/jdbc/test/DefaultTestCase.java b/modules/jdbc-pool/test/org/apache/tomcat/jdbc/test/DefaultTestCase.java index b7377b63b..148c1a811 100644 --- a/modules/jdbc-pool/test/org/apache/tomcat/jdbc/test/DefaultTestCase.java +++ b/modules/jdbc-pool/test/org/apache/tomcat/jdbc/test/DefaultTestCase.java @@ -56,7 +56,7 @@ public class DefaultTestCase extends TestCase { p.setMinIdle(threadcount); p.setLogAbandoned(false); p.setRemoveAbandoned(false); - datasource = new org.apache.tomcat.jdbc.pool.DataSourceProxy(); + datasource = new org.apache.tomcat.jdbc.pool.DataSource(); datasource.setPoolProperties(p); return datasource; } diff --git a/modules/jdbc-pool/test/org/apache/tomcat/jdbc/test/FairnessTest.java b/modules/jdbc-pool/test/org/apache/tomcat/jdbc/test/FairnessTest.java index 7dd7c5595..c8bfff9eb 100644 --- a/modules/jdbc-pool/test/org/apache/tomcat/jdbc/test/FairnessTest.java +++ b/modules/jdbc-pool/test/org/apache/tomcat/jdbc/test/FairnessTest.java @@ -17,6 +17,7 @@ package org.apache.tomcat.jdbc.test; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.sql.Connection; @@ -26,6 +27,7 @@ import java.sql.ResultSet; import javax.sql.DataSource; import org.apache.tomcat.jdbc.pool.DataSourceFactory; +import org.apache.tomcat.jdbc.pool.DataSourceProxy; /** * @author Filip Hanik @@ -143,11 +145,42 @@ public class FairnessTest extends DefaultTestCase { tearDown(); } + public void testPoolThreads20Connections10FairAsync() throws Exception { + System.out.println("Starting fairness - Tomcat JDBC - Fair - Async"); + init(); + this.datasource.getPoolProperties().setMaxActive(10); + this.datasource.getPoolProperties().setFairQueue(true); + this.threadcount = 20; + this.transferProperties(); + this.datasource.getConnection().close(); + latch = new CountDownLatch(threadcount); + long start = System.currentTimeMillis(); + TestThread[] threads = new TestThread[threadcount]; + for (int i=0; i cf = ((DataSourceProxy)d).getConnectionAsync(); + con = cf.get(); + } else { + con = d.getConnection(); + } long delta = System.nanoTime() - start; totalmax += delta; max = Math.max(delta, max); diff --git a/modules/jdbc-pool/test/org/apache/tomcat/jdbc/test/SimplePOJOAsyncExample.java b/modules/jdbc-pool/test/org/apache/tomcat/jdbc/test/SimplePOJOAsyncExample.java new file mode 100644 index 000000000..1935cae8b --- /dev/null +++ b/modules/jdbc-pool/test/org/apache/tomcat/jdbc/test/SimplePOJOAsyncExample.java @@ -0,0 +1,79 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.tomcat.jdbc.test; +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.Statement; +import java.util.concurrent.Future; + +import org.apache.tomcat.jdbc.pool.DataSource; +import org.apache.tomcat.jdbc.pool.PoolProperties; + +public class SimplePOJOAsyncExample { + + public static void main(String[] args) throws Exception { + PoolProperties p = new PoolProperties(); + p.setFairQueue(true); + p.setUrl("jdbc:mysql://localhost:3306/mysql?autoReconnect=true"); + p.setDriverClassName("com.mysql.jdbc.Driver"); + p.setUsername("root"); + p.setPassword("password"); + p.setJmxEnabled(true); + p.setTestWhileIdle(false); + p.setTestOnBorrow(true); + p.setValidationQuery("SELECT 1"); + p.setTestOnReturn(false); + p.setValidationInterval(30000); + p.setTimeBetweenEvictionRunsMillis(30000); + p.setMaxActive(100); + p.setInitialSize(10); + p.setMaxWait(10000); + p.setRemoveAbandonedTimeout(60); + p.setMinEvictableIdleTimeMillis(30000); + p.setMinIdle(10); + p.setLogAbandoned(true); + p.setRemoveAbandoned(true); + p.setJdbcInterceptors("org.apache.tomcat.jdbc.pool.interceptor.ConnectionState;org.apache.tomcat.jdbc.pool.interceptor.StatementFinalizer"); + DataSource datasource = new DataSource(); + datasource.setPoolProperties(p); + + Connection con = null; + try { + Future future = datasource.getConnectionAsync(); + while (!future.isDone()) { + System.out.println("Connection is not yet available. Do some background work"); + try { + Thread.sleep(100); //simulate work + }catch (InterruptedException x) { + Thread.currentThread().interrupted(); + } + } + con = future.get(); //should return instantly + Statement st = con.createStatement(); + ResultSet rs = st.executeQuery("select * from user"); + int cnt = 1; + while (rs.next()) { + System.out.println((cnt++)+". Host:" +rs.getString("Host")+" User:"+rs.getString("User")+" Password:"+rs.getString("Password")); + } + rs.close(); + st.close(); + } finally { + if (con!=null) try {con.close();}catch (Exception ignore) {} + } + } + +} \ No newline at end of file diff --git a/modules/jdbc-pool/test/org/apache/tomcat/jdbc/test/TwoDataSources.java b/modules/jdbc-pool/test/org/apache/tomcat/jdbc/test/TwoDataSources.java index 37a6f3880..febf8556a 100644 --- a/modules/jdbc-pool/test/org/apache/tomcat/jdbc/test/TwoDataSources.java +++ b/modules/jdbc-pool/test/org/apache/tomcat/jdbc/test/TwoDataSources.java @@ -34,14 +34,12 @@ public class TwoDataSources extends DefaultTestCase { this.assertTrue("Connection c2 should be working",false); } try { - c1.close(); - this.assertTrue("Connection should have been closed.",false); + this.assertTrue("Connection should have been closed.",c1.isClosed()); }catch (Exception x) { this.assertTrue("This is correct, c1 is closed",true); } try { - c2.close(); - this.assertTrue("Connection c2 should not have been closed.",true); + this.assertFalse("Connection c2 should not have been closed.",c2.isClosed()); }catch (Exception x) { this.assertTrue("Connection c2 should be working",false); } -- 2.11.0