Dépôt officiel du code source de l'ERP OpenConcerto

/*

 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.

 *

 * Copyright 2011 OpenConcerto, by ILM Informatique. All rights reserved.

 *

 * The contents of this file are subject to the terms of the GNU General Public License Version 3

 * only ("GPL"). You may not use this file except in compliance with the License. You can obtain a

 * copy of the License at http://www.gnu.org/licenses/gpl-3.0.html See the License for the specific

 * language governing permissions and limitations under the License.

 *

 * When distributing the software, include this License Header Notice in each file.

 */

 package org.openconcerto.sql.model;

import org.openconcerto.sql.Log;

import org.openconcerto.sql.State;

import org.openconcerto.sql.request.SQLCache;

import org.openconcerto.sql.utils.SQLUtils;

import org.openconcerto.utils.CompareUtils;

import org.openconcerto.utils.ExceptionHandler;

import org.openconcerto.utils.ExceptionUtils;

import org.openconcerto.utils.RTInterruptedException;

import org.openconcerto.utils.ThreadFactory;

import org.openconcerto.utils.Tuple2;

import org.openconcerto.utils.cache.CacheResult;

import org.openconcerto.utils.cache.ICacheSupport;

import java.beans.PropertyChangeEvent;

import java.beans.PropertyChangeListener;

import java.sql.Connection;

import java.sql.ResultSet;

import java.sql.ResultSetMetaData;

import java.sql.SQLException;

import java.sql.SQLNonTransientException;

import java.sql.SQLTransientException;

import java.sql.Savepoint;

import java.sql.Statement;

import java.util.Arrays;

import java.util.Collections;

import java.util.HashMap;

import java.util.HashSet;

import java.util.Hashtable;

import java.util.List;

import java.util.Map;

import java.util.NoSuchElementException;

import java.util.Set;

import java.util.WeakHashMap;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.LinkedBlockingQueue;

import java.util.concurrent.ThreadPoolExecutor;

import java.util.concurrent.TimeUnit;

import java.util.concurrent.atomic.AtomicLong;

import java.util.concurrent.atomic.AtomicReference;

import java.util.logging.Level;

import org.apache.commons.dbcp.AbandonedConfig;

import org.apache.commons.dbcp.BasicDataSource;

import org.apache.commons.dbcp.ConnectionFactory;

import org.apache.commons.dbcp.PoolableConnection;

import org.apache.commons.dbcp.PoolableConnectionFactory;

import org.apache.commons.dbcp.PoolingConnection;

import org.apache.commons.dbcp.PoolingDataSource;

import org.apache.commons.dbcp.SQLNestedException;

import org.apache.commons.dbutils.BasicRowProcessor;

import org.apache.commons.dbutils.ResultSetHandler;

import org.apache.commons.dbutils.RowProcessor;

import org.apache.commons.dbutils.handlers.ArrayHandler;

import org.apache.commons.dbutils.handlers.ArrayListHandler;

import org.apache.commons.dbutils.handlers.MapHandler;

import org.apache.commons.dbutils.handlers.MapListHandler;

import org.apache.commons.dbutils.handlers.ScalarHandler;

import org.apache.commons.pool.KeyedObjectPool;

import org.apache.commons.pool.KeyedObjectPoolFactory;

import org.apache.commons.pool.ObjectPool;

import org.apache.commons.pool.impl.GenericObjectPool;

import org.h2.constant.ErrorCode;

import org.postgresql.util.GT;

import org.postgresql.util.PSQLException;

import net.jcip.annotations.GuardedBy;

import net.jcip.annotations.ThreadSafe;

/**

 * Une source de donnée SQL.

 *

 * @author ILM Informatique 10 juin 2004

 */

@ThreadSafe

public final class SQLDataSource extends BasicDataSource implements Cloneable {

    // MAYBE add a cache, but ATTN synchronized : one connection per thread, but only one shared DS

    /** A map of supported database systems and associated drivers. */

    static public final Map<SQLSystem, String> DRIVERS;

    static {

        DRIVERS = new HashMap<SQLSystem, String>();

        DRIVERS.put(SQLSystem.MYSQL, "com.mysql.jdbc.Driver");

        DRIVERS.put(SQLSystem.POSTGRESQL, "org.postgresql.Driver");

        DRIVERS.put(SQLSystem.DERBY, "org.apache.derby.jdbc.ClientDriver");

        DRIVERS.put(SQLSystem.H2, "org.h2.Driver");

        DRIVERS.put(SQLSystem.MSSQL, "com.microsoft.sqlserver.jdbc.SQLServerDriver");

    }

    // timeouts in seconds

    static public final int loginTimeOut = 15;

    static public final int socketTimeOut = 8 * 60;

    // in milliseconds

    static public int QUERY_TUNING = 0;

    static public interface IgnoringRowProcessor extends RowProcessor {

        @Override

        public Map<String, Object> toMap(ResultSet rs) throws SQLException;

        /**

         * Convert the passed result set to a map, ignoring some columns.

         *

         * @param rs the result set.

         * @param toIgnore which columns' label to ignore.

         * @return a map with all columns of <code>rs</code> except <code>toIgnore</code>.

         * @throws SQLException if an error occurs while reading <code>rs</code>.

         */

        public Map<String, Object> toMap(ResultSet rs, Set<String> toIgnore) throws SQLException;

    }

    // ignoring case-sensitive processor

    static private class IgnoringCSRowProcessor extends BasicRowProcessor implements IgnoringRowProcessor {

        @Override

        public Map<String, Object> toMap(ResultSet rs) throws SQLException {

            return toMap(rs, Collections.<String> emptySet());

        }

        // on ne veut pas de CaseInsensitiveMap

        @Override

        public Map<String, Object> toMap(ResultSet rs, Set<String> toIgnore) throws SQLException {

            final Map<String, Object> result = new HashMap<String, Object>();

            final ResultSetMetaData rsmd = rs.getMetaData();

            final int cols = rsmd.getColumnCount();

            for (int i = 1; i <= cols; i++) {

                // ATTN use label not base name (eg "des" in "SELECT DESIGNATION as des FROM ...")

                final String label = rsmd.getColumnLabel(i);

                if (!toIgnore.contains(label))

                    result.put(label, rs.getObject(i));

            }

            return result;

        }

    }

    static public final IgnoringRowProcessor ROW_PROC = new IgnoringCSRowProcessor();

    // all thread safe

    public static final ColumnListHandler COLUMN_LIST_HANDLER = new ColumnListHandler();

    public static final ArrayListHandler ARRAY_LIST_HANDLER = new ArrayListHandler();

    public static final ListListHandlerGeneric<Object> LIST_LIST_HANDLER = ListListHandlerGeneric.create(Object.class, null);

    public static final ArrayHandler ARRAY_HANDLER = new ArrayHandler();

    public static final ScalarHandler SCALAR_HANDLER = new ScalarHandler();

    public static final MapListHandler MAP_LIST_HANDLER = new MapListHandler(ROW_PROC);

    public static final MapHandler MAP_HANDLER = new MapHandler(ROW_PROC);

    // Cache, linked to cacheEnable and tables

    @GuardedBy("this")

    private SQLCache<List<?>, Object> cache;

    @GuardedBy("this")

    private boolean cacheEnabled;

    private final PropertyChangeListener descL;

    // tables that can be used in queries (and thus can impact the cache)

    @GuardedBy("this")

    private Set<SQLTable> tables;

    private static int count = 0; // compteur de requetes

    private final DBSystemRoot sysRoot;

    private ConnectionFactory connectionFactory;

    // no need to synchronize multiple call to this attribute since we only access the

    // Thread.currentThread() key

    @GuardedBy("handlers")

    private final Map<Thread, HandlersStack> handlers;

    @GuardedBy("this")

    private ExecutorService exec = null;

    private final Object setInitialShemaLock = new String("initialShemaWriteLock");

    // linked to initialSchema and uptodate

    @GuardedBy("this")

    private boolean initialShemaSet;

    @GuardedBy("this")

    private String initialShema;

    // which Connection have the right default schema

    @GuardedBy("this")

    private final Map<Connection, Object> schemaUptodate;

    // which Connection aren't invalidated

    @GuardedBy("this")

    private final Map<Connection, Object> uptodate;

    private volatile int retryWait;

    @GuardedBy("this")

    private boolean blockWhenExhausted;

    @GuardedBy("this")

    private long softMinEvictableIdleTimeMillis;

    @GuardedBy("this")

    private int txIsolation;

    @GuardedBy("this")

    private Integer dbTxIsolation;

    @GuardedBy("this")

    private boolean checkOnceDBTxIsolation;

    private final Object testLock = new String("testLock");

    public SQLDataSource(DBSystemRoot sysRoot, String base, String login, String pass) {

        this(sysRoot, sysRoot.getServer().getURL(base), login, pass, Collections.<SQLTable> emptySet());

    }

    private SQLDataSource(DBSystemRoot sysRoot, String url, String login, String pass, Set<SQLTable> tables) {

        this(sysRoot);

        final SQLSystem system = getSystem();

        if (!DRIVERS.containsKey(system))

            throw new IllegalArgumentException("unknown database system: " + system);

        this.setDriverClassName(DRIVERS.get(system));

        this.setUrl("jdbc:" + system.getJDBCName() + ":" + url);

        this.setUsername(login);

        this.setPassword(pass);

        this.setTables(tables);

        if (system == SQLSystem.MYSQL) {

            this.addConnectionProperty("transformedBitIsBoolean", "true");

            // by default allowMultiQueries, since it's more convenient (i.e. pass String around

            // instead of List<String>) and faster (less trips to the server, allow

            // SQLUtils.executeMultiple())

            this.addConnectionProperty("allowMultiQueries", "true");

        } else if (system == SQLSystem.MSSQL) {

            // Otherwise we get SQLState S0002 instead of 42S02 (needed at least in

            // SQLBase.getFwkMetadata())

            // http://msdn.microsoft.com/fr-fr/library/ms712451.aspx

            // http://technet.microsoft.com/en-us/library/ms378988(v=sql.105).aspx

            this.addConnectionProperty("xopenStates", "true");

            // see

            // https://connect.microsoft.com/SQLServer/feedback/details/295907/resultsetmetadata-gettablename-returns-null-or-inconsistent-results

            // http://social.msdn.microsoft.com/Forums/sqlserver/en-US/55e8cbb2-b11c-446e-93ab-dc30658caf99/resultsetmetadatagettablename-returns-instead-of-table-name?forum=sqldataaccess

            // 1. The statement that the resultset belongs to was created with

            // TYPE_SCROLL_INSENSITIVE or TYPE_SCROLL_SENSITIVE : The full table or view name will

            // be returned

            // 2. The statement that the resultset belongs to was created without specifying the

            // cursor type, or the cursor type is TYPE_FORWARD_ONLY : The full table or view name

            // will be returned if the column is a text, ntext, or image, else empty string.

            this.addConnectionProperty("selectMethod", "cursor");

        }

        this.setLoginTimeout(loginTimeOut);

        this.setSocketTimeout(socketTimeOut);

        this.setTCPKeepAlive(true);

        this.setRetryWait(7000);

        // ATTN DO NOT call execute() or any method that might create a connection

        // since at this point dsInit() has not been called and thus connection properties might be

        // missing (eg allowMultiQueries). And the faulty connection will stay in the pool.

    }

    @Override

    public final void setLoginTimeout(int timeout) {

        if (this.getSystem() == SQLSystem.MYSQL) {

            this.addConnectionProperty("connectTimeout", timeout + "000");

        } else if (this.getSystem() == SQLSystem.POSTGRESQL || this.getSystem() == SQLSystem.MSSQL) {

            this.addConnectionProperty("loginTimeout", timeout + "");

        } else {

            Log.get().warning("Ignoring login timeout for " + this);

        }

    }

    public final void setSocketTimeout(int timeout) {

        if (this.getSystem() == SQLSystem.MYSQL) {

            this.addConnectionProperty("socketTimeout", timeout + "000");

        } else if (this.getSystem() == SQLSystem.H2) {

            // org.h2.util.NetUtils.createSocket() doesn't use setSoTimeout(), so this is the next

            // best thing. But it isn't checked everywhere, see DbSettings.maxQueryTimeout

            this.addConnectionProperty("MAX_QUERY_TIMEOUT", timeout + "000");

        } else if (this.getSystem() == SQLSystem.POSTGRESQL) {

            this.addConnectionProperty("socketTimeout", timeout + "");

        } else {

            Log.get().log(getLogLevelForIgnoredTCPParam(), "Ignoring socket timeout for " + this);

        }

    }

    // if TCP isn't used or is used to connect to localhost, TCP should be quite robust

    private final Level getLogLevelForIgnoredTCPParam() {

        return SQLSyntax.get(this.sysRoot).isServerLocalhost(this.sysRoot.getServer()) ? Level.CONFIG : Level.WARNING;

    }

    public final void setTCPKeepAlive(final boolean b) {

        if (this.getSystem() == SQLSystem.POSTGRESQL || this.getSystem() == SQLSystem.MYSQL) {

            this.addConnectionProperty("tcpKeepAlive", String.valueOf(b));

        } else {

            Log.get().log(getLogLevelForIgnoredTCPParam(), "Ignoring TCP keep alive for " + this);

        }

    }

    /**

     * Set the number of milliseconds to wait before retrying if a connection fails to establish or

     * if a query fails to execute.

     *

     * @param retryWait the number of milliseconds to wait, negative to disable retrying.

     */

    public final void setRetryWait(int retryWait) {

        this.retryWait = retryWait;

    }

    synchronized void setTables(Set<SQLTable> tables) {

        // don't change the cache if we're only adding tables

        final boolean update = this.cache == null || !tables.containsAll(this.tables);

        this.tables = Collections.unmodifiableSet(new HashSet<SQLTable>(tables));

        if (update)

            updateCache();

    }

    private synchronized Set<SQLTable> getTables() {

        return this.tables;

    }

    private synchronized void updateCache() {

        if (this.cache != null)

            this.cache.getSupp().die();

        this.cache = createCache(null);

        for (final HandlersStack s : this.handlers.values()) {

            s.updateCache();

        }

    }

    // the cache for committed data

    synchronized final SQLCache<List<?>, Object> getCommittedCache() {

        return this.cache;

    }

    final SQLCache<List<?>, Object> getCache() {

        // transactions are isolated from one another, so their caches should be too

        final HandlersStack stack = getHandlersStack();

        if (stack != null && stack.hasTransaction())

            return stack.getCache();

        else

            return this.getCommittedCache();

    }

    synchronized SQLCache<List<?>, Object> createCache(final TransactionPoint o) {

        final SQLCache<List<?>, Object> res;

        if (this.isCacheEnabled() && this.tables.size() > 0) {

            // the general cache should wait for transactions to end, but the cache of transactions

            // must not.

            final boolean committedCache = o == null;

            final Object desc = committedCache ? this : o;

            final ICacheSupport<SQLData> cacheSupp = committedCache ? null : this.cache.getSupp();

            res = new SQLCache<List<?>, Object>(cacheSupp, 30, 30, "results of " + desc.toString(), o) {

                @Override

                protected String getCacheSuppName(String cacheName) {

                    assert committedCache : "Creating extra ICacheSupport";

                    return SQLDataSource.this.toString();

                }

            };

        } else {

            res = null;

        }

        return res;

    }

    /**

     * Enable or disable the cache. ATTN if you enable the cache you must

     * {@link SQLTable#fire(SQLTableEvent) fire} table events, or use a class that does like

     * {@link SQLRowValues}.

     *

     * @param b <code>true</code> to enable the cache.

     */

    public final synchronized void setCacheEnabled(boolean b) {

        if (this.cacheEnabled != b) {

            this.cacheEnabled = b;

            updateCache();

        }

    }

    public final synchronized boolean isCacheEnabled() {

        return this.cacheEnabled;

    }

    /* pour le clonage */

    private SQLDataSource(DBSystemRoot sysRoot) {

        this.sysRoot = sysRoot;

        // on a besoin d'une implementation synchronisée

        this.handlers = new Hashtable<Thread, HandlersStack>();

        // weak, since this is only a hint to avoid initializing the connection

        // on each borrowal

        this.schemaUptodate = new WeakHashMap<Connection, Object>();

        this.uptodate = new WeakHashMap<Connection, Object>();

        this.initialShemaSet = false;

        this.initialShema = null;

        // used by #getNewConnection() when there's an exception and by #validateDBConnectivity()

        this.setValidationQuery("SELECT 1");

        // We must set a socket timeout high enough for large queries but the validation query

        // should return almost instantly. Not too low since overloaded links may have very high

        // latencies.

        this.setValidationQueryTimeout(6);

        // don't test on borrow, this would double the queries.

        this.setTestOnBorrow(false);

        // don't test on return, the connection was just used and this would double the queries.

        this.setTestOnReturn(false);

        // for now don't test on idle as our evictor is run quite often to promptly close unneeded

        // connections. If it is slowed down, we risk overloading the server or just hit its maximum

        // connection count. MAYBE enable when we upgrade to DBCP 2, since it depends on POOL 2

        // which has an EvictionPolicy, so for example we could test connections after a smaller

        // amount of time than minEvictableIdleTimeMillis.

        this.setTestWhileIdle(false);

        this.setInitialSize(3);

        // neither DOS the server...

        this.setBlockWhenExhausted(true);

        this.setMaxActive(12);

        // ... nor us

        this.setMaxWait(5000);

        // creating connections is quite costly so make sure we always have a couple free

        this.setMinIdle(2);

        // but not too much as it can lock out other users (the server has a max connection count)

        this.setMaxIdle(10);

        // check 5 connections every 4 seconds

        this.setTimeBetweenEvictionRunsMillis(4000);

        this.setNumTestsPerEvictionRun(5);

        // kill extra (above minIdle) connections after 40s

        this.setSoftMinEvictableIdleTimeMillis(TimeUnit.SECONDS.toMillis(40));

        // kill idle connections after 30 minutes (even if it means re-creating some new ones

        // immediately afterwards to ensure minIdle connections). For now it's a poor man's solution

        // for lacking testWhileIdle. After that time NAT tables expires, the path to the server

        // might be broken...

        this.setMinEvictableIdleTimeMillis(TimeUnit.MINUTES.toMillis(30));

        // the default of many systems

        this.txIsolation = Connection.TRANSACTION_READ_COMMITTED;

        // by definition unknown without a connection

        this.dbTxIsolation = null;

        // it's rare that DB configuration changes, and it's expensive to add a trip to the server

        // for each new connection

        this.checkOnceDBTxIsolation = true;

        // see #createDataSource() for properties not supported by this class

        this.tables = Collections.emptySet();

        this.descL = new PropertyChangeListener() {

            @Override

            public void propertyChange(PropertyChangeEvent evt) {

                if (evt.getPropertyName().equals("descendants")) {

                    // the dataSource must always have all tables, to listen to them for its cache

                    setTables(((DBSystemRoot) evt.getSource()).getDescs(SQLTable.class));

                }

            }

        };

        this.sysRoot.addListener(this.descL);

        this.cache = null;

        this.cacheEnabled = false;

    }

    /**

     * Exécute la requête et retourne le résultat sous forme de liste de map. Si la requete va

     * retourner beaucoup de lignes, il est peut-être préférable d'utiliser un ResultSetHandler.

     *

     * @param query le requête à exécuter.

     * @return le résultat de la requête.

     * @see MapListHandler

     * @see #execute(String, ResultSetHandler)

     */

    public List execute(String query) {

        return (List) this.execute(query, MAP_LIST_HANDLER);

    }

    /**

     * Exécute la requête et retourne la première colonne uniquement.

     *

     * @param query le requête à exécuter.

     * @return le résultat de la requête.

     * @see ColumnListHandler

     * @see #execute(String, ResultSetHandler)

     */

    public List executeCol(String query) {

        return (List) this.execute(query, COLUMN_LIST_HANDLER);

    }

    /**

     * Exécute la requête et retourne le résultat sous forme de liste de tableau. Si la requete va

     * retourner beaucoup de lignes, il est peut-être préférable d'utiliser un ResultSetHandler.

     *

     * @param query le requête à exécuter.

     * @return le résultat de la requête.

     * @see ArrayListHandler

     * @see #execute(String, ResultSetHandler)

     */

    public List executeA(String query) {

        return (List) this.execute(query, ARRAY_LIST_HANDLER);

    }

    /**

     * Exécute la requête et retourne la première ligne du résultat sous forme de map.

     *

     * @param query le requête à exécuter.

     * @return le résultat de la requête.

     * @see MapHandler

     * @see #execute(String)

     */

    public Map<String, Object> execute1(String query) {

        return (Map<String, Object>) this.execute(query, MAP_HANDLER);

    }

    /**

     * Exécute la requête et retourne la première ligne du résultat sous forme de tableau.

     *

     * @param query le requête à exécuter.

     * @return le résultat de la requête.

     * @see ArrayHandler

     * @see #executeA(String)

     */

    public Object[] executeA1(String query) {

        return (Object[]) this.execute(query, ARRAY_HANDLER);

    }

    /**

     * Exécute la requête et retourne la valeur de la premiere colonne de la premiere ligne.

     *

     * @param query le requête à exécuter.

     * @return le résultat de la requête.

     */

    public Object executeScalar(String query) {

        return this.execute(query, SCALAR_HANDLER);

    }

    /**

     * Exécute la requête et passe le résultat au ResultSetHandler.

     *

     * @param query le requête à exécuter.

     * @param rsh le handler à utiliser, ou <code>null</code>.

     * @return le résultat du handler, <code>null</code> si rsh est <code>null</code>.

     * @see #execute(String)

     */

    public Object execute(String query, ResultSetHandler rsh) {

        return this.execute(query, rsh, null);

    }

    /**

     * Execute <code>query</code> within <code>c</code>, passing the result set to <code>rsh</code>.

     *

     * @param query the query to perform.

     * @param rsh what to do with the result, can be <code>null</code>.

     * @param changeState whether <code>query</code> changes the state of a connection.

     * @return the result of <code>rsh</code>, <code>null</code> if rsh or the resultSet is

     *         <code>null</code>.

     * @throws RTInterruptedException if the current thread is interrupted while waiting for the

     *         cache or for the database.

     */

    public final Object execute(final String query, final ResultSetHandler rsh, final boolean changeState) throws RTInterruptedException {

        return this.execute(query, rsh, changeState, null);

    }

    private Object execute(final String query, final ResultSetHandler rsh, final Connection c) throws RTInterruptedException {

        // false since the vast majority of request do NOT change the state

        return this.execute(query, rsh, false, c);

    }

    final List<Object> getCacheKey(final String query, final ResultSetHandler rsh) {

        return query.startsWith("SELECT") ? Arrays.asList(new Object[] { query, rsh }) : null;

    }

    /**

     * Execute <code>query</code> within <code>c</code>, passing the result set to <code>rsh</code>.

     *

     * @param query the query to perform.

     * @param rsh what to do with the result, can be <code>null</code>.

     * @param changeState whether <code>query</code> changes the state of a connection.

     * @param passedConn the sql connection to use.

     * @return the result of <code>rsh</code>, <code>null</code> if rsh or the resultSet is

     *         <code>null</code>.

     * @throws RTInterruptedException if the current thread is interrupted while waiting for the

     *         cache or for the database.

     */

    private Object execute(final String query, final ResultSetHandler rsh, final boolean changeState, final Connection passedConn) throws RTInterruptedException {

        final long timeMs = System.currentTimeMillis();

        final long time = System.nanoTime();

        // some systems refuse to execute nothing

        if (query.length() == 0) {

            SQLRequestLog.log(query, "Pas de requête.", timeMs, time);

            return null;

        }

        final IResultSetHandler irsh = rsh instanceof IResultSetHandler ? (IResultSetHandler) rsh : null;

        final boolean readCache = irsh == null || irsh.readCache();

        final boolean canWriteCache = irsh == null || irsh.canWriteCache();

        final SQLCache<List<?>, Object> cache = !readCache && !canWriteCache ? null : this.getCache();

        final List<Object> key = cache == null ? null : getCacheKey(query, rsh);

        final CacheResult<Object> cacheRes;

        if (key != null) {

            final Set<? extends SQLData> data = irsh == null || irsh.getCacheModifiers() == null ? this.getTables() : irsh.getCacheModifiers();

            cacheRes = cache.check(key, readCache, canWriteCache, data);

            if (cacheRes.getState() == CacheResult.State.INTERRUPTED)

                throw new RTInterruptedException("interrupted while waiting for the cache");

            else if (cacheRes.getState() == CacheResult.State.VALID) {

                // cache actif

                if (State.DEBUG)

                    State.INSTANCE.addCacheHit();

                SQLRequestLog.log(query, "En cache.", timeMs, time);

                return cacheRes.getRes();

            }

        } else {

            cacheRes = null;

        }

        Object result = null;

        QueryInfo info = null;

        final long afterCache = System.nanoTime();

        final long afterQueryInfo, afterExecute, afterHandle;

        int count = 0;

        try {

            info = new QueryInfo(query, changeState, passedConn);

            try {

                afterQueryInfo = System.nanoTime();

                final Object[] res = this.executeTwice(info);

                final Statement stmt = (Statement) res[0];

                ResultSet rs = (ResultSet) res[1];

                // TODO 1. rename #execute(String) to #executeN(String)

                // and make #execute(String) do #execute(String, null)

                // 2. let null rs pass to rsh

                // otherwise you write ds.execute("req", new ResultSetHandler() {

                // public Object handle(ResultSet rs) throws SQLException {

                // return "OK";

                // }

                // });

                // and OK won't be returned if "req" returns a null rs.

                afterExecute = System.nanoTime();

                if (rsh != null && rs != null) {

                    if (this.getSystem() == SQLSystem.DERBY || this.getSystem() == SQLSystem.POSTGRESQL) {

                        rs = new SQLResultSet(rs);

                    }

                    result = rsh.handle(rs);

                    count = SQLResultSet.getRowProcessedCount(rs);

                }

                afterHandle = System.nanoTime();

                stmt.close();

                // if key was added to the cache

                if (key != null) {

                    synchronized (this) {

                        putInCache(cache, irsh, cacheRes, result);

                    }

                }

                info.releaseConnection();

            } catch (SQLException exn) {

                // don't usually do a getSchema() as it access the db

                throw new IllegalStateException("Impossible d'accéder au résultat de " + query + "\n in " + this, exn);

            }

        } catch (RuntimeException e) {

            // for each #check() there must be a #removeRunning()

            // let the cache know we ain't gonna tell it the result

            if (cacheRes != null)

                cache.removeRunning(cacheRes);

            if (info != null)

                info.releaseConnection(e);

            throw e;

        }

        SQLRequestLog.log(query, "", info.getConnection(), timeMs, time, afterCache, afterQueryInfo, afterExecute, afterHandle, System.nanoTime(), count);

        return result;

    }

    private synchronized void putInCache(final SQLCache<List<?>, Object> cache, final IResultSetHandler irsh, final CacheResult<Object> cacheRes, Object result) {

        if (irsh != null && irsh.writeCache() || irsh == null && IResultSetHandler.shouldCache(result)) {

            cache.put(cacheRes, result);

        } else {

            cache.removeRunning(cacheRes);

        }

    }

    private synchronized final ExecutorService getExec() {

        if (this.exec == null) {

            // not daemon since we want the connections to be returned

            final ThreadFactory factory = new ThreadFactory(SQLDataSource.class.getSimpleName() + " " + this.toString() + " exec n° ", false);

            // a rather larger number of threads since all they do is wait severals seconds

            this.exec = new ThreadPoolExecutor(0, 32, 30L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(), factory);

        }

        return this.exec;

    }

    private final class QueryInfo {

        private final String query;

        // whether query change the state of our connection

        private final boolean changeState;

        // can change if private

        private Connection c;

        // whether we acquired a new connection (and thus can do whatever we want with it)

        private final boolean privateConnection;

        QueryInfo(String query, boolean changeState, final Connection passedConn) {

            super();

            this.query = query;

            this.changeState = changeState;

            // if passedConn is provided use it, else we need to find one

            boolean acquiredConnection = false;

            final Connection foundConn;

            if (passedConn != null)

                foundConn = passedConn;

            else if (!handlingConnection()) {

                foundConn = getNewConnection();

                acquiredConnection = true;

            } else {

                final HandlersStack threadHandlers = getHandlersStack();

                if (!changeState || threadHandlers.isChangeAllowed()) {

                    foundConn = threadHandlers.getConnection();

                } else {

                    throw new IllegalStateException("the passed query change the connection's state and the current thread has a connection which will thus be changed."

                            + " A possible solution is to execute it in the setup() of a ConnectionHandler\n" + query);

                }

            }

            this.privateConnection = acquiredConnection;

            this.c = foundConn;

        }

        public final Connection getConnection() {

            return this.c;

        }

        public final String getQuery() {

            return this.query;

        }

        void releaseConnection(RuntimeException e) {

            // MySQL reste des fois bloqué dans SocketInputStream.socketRead0()

            // (le serveur ayant tué la query)

            if (e instanceof InterruptedQuery && getSystem() == SQLSystem.MYSQL) {

                final ExecutorThread thread = ((InterruptedQuery) e).getThread();

                if (this.privateConnection) {

                    if (this.changeState)

                        // no need to try to save the connection, it is no longer valid

                        this.releaseConnection();

                    else {

                        // test if the connection is still valid before returning it to the pool

                        getExec().execute(new Runnable() {

                            public void run() {

                                // on attend un peu

                                try {

                                    thread.join(1500);

                                    // pour voir si on meurt

                                    if (thread.isAlive()) {

                                        Log.get().warning(getFailedCancelMsg());

                                        closeConnection(getConnection());

                                    } else {

                                        // la connexion est ok, on la remet dans le pool

                                        returnConnection(getConnection());

                                    }

                                } catch (InterruptedException e) {

                                    // the datasource is closing

                                    Log.get().fine("Interrupted while joining " + getQuery());

                                    closeConnection(getConnection());

                                }

                            }

                        });

                    }

                } else {

                    // try to save the connection since it is used by others

                    try {

                        // clear the interrupt status set by InterruptedQuery

                        // so that we can wait on thread

                        Thread.interrupted();

                        thread.join(500);

                    } catch (InterruptedException e2) {

                        System.err.println("ignore, we are already interrupted");

                        e2.printStackTrace();

                    }

                    // remettre le flag pour les méthodes appelantes.

                    Thread.currentThread().interrupt();

                    // connection is still stuck

                    if (thread.isAlive()) {

                        throw new IllegalStateException(getFailedCancelMsg(), e);

                    } else

                        this.releaseConnection();

                }

            } else

                this.releaseConnection();

        }

        void releaseConnection() {

            // have we borrowed a connection, otherwise it is not our responsibility to release it.

            if (this.privateConnection) {

                if (this.changeState)

                    // the connection is no longer in a pristine state so close it

                    closeConnection(this.getConnection());

                else

                    // otherwise we can reuse it

                    returnConnection(this.getConnection());

            }

        }

        private final String getFailedCancelMsg() {

            return "cancel of " + System.identityHashCode(getConnection()) + " failed for " + getQuery();

        }

        final boolean canObtainNewConnection() {

            return this.privateConnection;

        }

        // an error has occured, try within another connection if possible

        final Connection obtainNewConnection() {

            if (!this.canObtainNewConnection()) {

                return null;

            } else {

                // ATTN should be sure that our connection was not already closed,

                // see #closeConnection()

                closeConnection(this.getConnection());

                this.c = borrowConnection(true);

                return this.getConnection();

            }

        }

        @Override

        public String toString() {

            return this.getClass().getSimpleName() + " private connection: " + this.privateConnection + " query: " + this.getQuery();

        }

    }

    /**

     * Whether the current thread has called {@link #useConnection(ConnectionHandler)}.

     *

     * @return <code>true</code> if within <code>useConnection()</code> and thus safe to call

     *         {@link #getConnection()}.

     */

    public final boolean handlingConnection() {

        return this.handlers.containsKey(Thread.currentThread());

    }

    private final HandlersStack getHandlersStack() {

        return this.handlers.get(Thread.currentThread());

    }

    /**

     * Use a single connection to execute <code>handler</code>.

     *

     * @param <T> type of return.

     * @param <X> type of exception.

     * @param handler what to do with the connection.

     * @return what <code>handler</code> returned.

     * @throws SQLException if an exception happens in setup() or restore().

     * @throws X if handle() throws an exception.

     * @see ConnectionHandler

     */

    public final <T, X extends Exception> T useConnection(ConnectionHandler<T, X> handler) throws SQLException, X {

        return this.useConnection(handler, false);

    }

    private final <T, X extends Exception> T useConnection(ConnectionHandler<T, X> handler, final boolean force) throws SQLException, X {

        final HandlersStack h;

        boolean connOutsidePool = false;

        if (!this.handlingConnection()) {

            Connection conn;

            try {

                conn = this.getNewConnection();

            } catch (NoSuchElementException e) {

                if (force) {

                    conn = this.connectionFactory.createConnection();

                    connOutsidePool = true;

                } else {

                    throw e;

                }

            }

            h = new HandlersStack(this, conn, handler);

            this.handlers.put(h.getThread(), h);

        } else if (handler.canRestoreState()) {

            h = this.getHandlersStack().push(handler);

        } else

            throw new IllegalStateException("this thread has already called useConnection() and thus expect its state, but the passed handler cannot restore state: " + handler);

        Connection conn = null;

        // before or after compute, RuntimeException or SQLException

        Exception beforeExn = null, afterExn = null;

        // X or SQLException

        Exception computeExn = null;

        try {

            conn = h.getConnection();

            h.setChangeAllowed(true);

            handler.setup(conn);

            h.setChangeAllowed(false);

            try {

                handler.compute(this);

            } catch (Exception e) {

                computeExn = e;

            }

        } catch (Exception e) {

            h.setChangeAllowed(false);

            beforeExn = e;

        }

        // in all cases (thanks to the above catch), try to restore the state

        // if conn is null setup() was never called

        boolean pristineState = conn == null;

        // don't bother trying to restore state if the connection has been invalidated (by a

        // recursive call)

        if (!pristineState && h.hasValidConnection() && handler.canRestoreState()) {

            h.setChangeAllowed(true);

            try {

                handler.restoreState(conn);

                pristineState = true;

            } catch (Exception e) {

                afterExn = e;

            }

            h.setChangeAllowed(false);

        }

        // ATTN conn can be null (return/closeConnection() accept it)

        if (h.pop()) {

            // remove if this thread has no handlers left

            this.handlers.remove(Thread.currentThread());

            if (connOutsidePool) {

                conn.close();

            } else if (pristineState) {

                this.returnConnection(h.getConnection());

            } else {

                this.closeConnection(h.invalidConnection());

            }

        } else {

            assert !connOutsidePool;

            // connection is still used

            if (!pristineState) {

                this.closeConnection(h.invalidConnection());

            }

            // else the top handler will release the connection

        }

        if (beforeExn != null) {

            assert computeExn == null : "Compute shouldn't be attempted if setup fails : " + beforeExn + " " + computeExn;

            if (afterExn != null) {

                throw new SQLException("could not restore state after failed setup : " + ExceptionUtils.getStackTrace(afterExn), beforeExn);

            } else {

                throw ExceptionUtils.throwExn(beforeExn, SQLException.class, RuntimeException.class);

            }

        } else if (afterExn != null) {

            if (computeExn != null) {

                throw new SQLException("could not restore state after successful setup : " + ExceptionUtils.getStackTrace(afterExn), computeExn);

            } else {

                throw ExceptionUtils.throwExn(afterExn, SQLException.class, RuntimeException.class);

            }

        } else {

            return handler.get();

        }

    }

    // this method create a Statement, don't forget to close it when you're done

    private Object[] executeTwice(QueryInfo queryInfo) throws SQLException {

        final String query = queryInfo.getQuery();

        Object[] res;

        try {

            res = executeOnce(query, queryInfo.getConnection());

        } catch (SQLException exn) {

            if (State.DEBUG)

                State.INSTANCE.addFailedRequest(query);

            // only retry for transient errors

            final boolean retry;

            if (exn instanceof SQLTransientException) {

                retry = true;

            } else if (exn instanceof SQLNonTransientException) {

                retry = false;

            } else if (getSystem() == SQLSystem.H2) {

                // 1. server was killed, maybe it will be restarted

                // 2. client network interface was brought down, maybe it will be brought up again

                retry = exn.getErrorCode() == ErrorCode.CONNECTION_BROKEN_1;

            } else if (getSystem() == SQLSystem.POSTGRESQL) {

                // Class 08 — Connection Exception (e.g. SocketException)

                // Class 57 — Operator Intervention (e.g. server shutdown)

                retry = exn.getSQLState().startsWith("08") || exn.getSQLState().startsWith("57");

            } else {