Pool.java

package org.objectweb.jonas.dbm;

import java.sql.SQLException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Map;

import javax.sql.XAConnection;
import javax.transaction.SystemException;
import javax.transaction.Transaction;

import org.objectweb.jonas.common.Log;
import org.objectweb.jonas.jdbc_xa.ConnectionImpl;
import org.objectweb.jonas.jdbc_xa.XADataSourceImpl;
import org.objectweb.jonas.jtm.TransactionService;
import org.objectweb.jonas.service.ServiceException;
import org.objectweb.jonas.service.ServiceManager;
import org.objectweb.transaction.jta.TransactionManager;
import org.objectweb.util.monolog.api.BasicLevel;
import org.objectweb.util.monolog.api.Logger;

public class Pool {

    // ----------------------------------------------------------------
    // Private data
    // ----------------------------------------------------------------

    private static Logger logger = null;

    private TransactionManager tm = null;

    private PoolKeeper poolKeeper;

    private Map xac2item = new HashMap();

    private LinkedList freeList = new LinkedList();

    private Map tx2item = new HashMap();

    private ConnectionManager cmgr;

    private XADataSourceImpl xads;

    private int waiterCount = 0;

    private long waitingTime = 0;

    private int busyMax = 0;

    private int busyMin = 0;

    // ----------------------------------------------------------------
    // Configuration
    // Each attibute have setter and getter, plus a default value.
    // ----------------------------------------------------------------

    private static final int NO_LIMIT = 99999;

    private static final long ONE_DAY = 1440L * 60L * 1000L;

    private static final int MAX_REMOVE_FREELIST = 10;

    private int poolMin = 0;

    public int getPoolMin() {
        return poolMin;
    }

    public synchronized void setPoolMin(int min) {
        if (poolMin != min) {
            poolMin = min;
            adjust();
        }
    }

    private int poolMax = NO_LIMIT;

    public int getPoolMax() {
        return poolMax;
    }

    public synchronized void setPoolMax(int max) {
        if (poolMax != max) {
            if (max < 0 || max > NO_LIMIT) {
                if (currentWaiters > 0) {
                    notify();
                }
                poolMax = NO_LIMIT;
            } else {
                if (currentWaiters > 0 && poolMax < max) {
                    notify();
                }
                poolMax = max;
                adjust();
            }
        }
    }

    private long maxAge = ONE_DAY;

    private int maxAgeMn;

    public int getMaxAge() {
        return maxAgeMn;
    }

    public long getMaxAgeMilli() {
        return maxAge;
    }

    public void setMaxAge(int mn) {
        maxAgeMn = mn;
        // set times in milliseconds
        maxAge = mn * 60L * 1000L;
    }

    private long maxOpenTime = ONE_DAY;

    private int maxOpenTimeMn;

    public int getMaxOpenTime() {
        return maxOpenTimeMn;
    }

    public long getMaxOpenTimeMilli() {
        return maxOpenTime;
    }

    public void setMaxOpenTime(int mn) {
        maxOpenTimeMn = mn;
        // set times in milliseconds
        maxOpenTime = mn * 60L * 1000L;
    }

    private long waiterTimeout = 10000;

    public int getMaxWaitTime() {
        return (int) (waiterTimeout / 1000L);
    }

    public void setMaxWaitTime(int sec) {
        waiterTimeout = sec * 1000L;
    }

    private int maxWaiters = 1000;

    public int getMaxWaiters() {
        return maxWaiters;
    }

    public void setMaxWaiters(int nb) {
        maxWaiters = nb;
    }

    private int samplingPeriod = 60; //default sampling period

    public int getSamplingPeriod() {
        return samplingPeriod;
    }

    public void setSamplingPeriod(int sec) {
        if (sec > 0) {
            samplingPeriod = sec;
            poolKeeper.setSamplingPeriod(sec);
        }
    }

    private int checkLevel = 1; // default = 1

    public int getCheckLevel() {
        return checkLevel;
    }

    public void setCheckLevel(int level) {
        checkLevel = level;
    }

    private String testStatement;

    public String getTestStatement() {
        return testStatement;
    }

    public void setTestStatement(String s) {
        testStatement = s;
    }

    // ----------------------------------------------------------------
    // Monitoring Attributes
    // Each attribute should have a get accessor.
    // ----------------------------------------------------------------

    private int busyMaxRecent = 0;

    public int getBusyMaxRecent() {
        return busyMaxRecent;
    }

    private int busyMinRecent = 0;

    public int getBusyMinRecent() {
        return busyMinRecent;
    }

    private int currentWaiters = 0;

    public int getCurrentWaiters() {
        return currentWaiters;
    }

    private int openedCount = 0;

    public int getOpenedCount() {
        return openedCount;
    }

    private int connectionFailures = 0;

    public int getConnectionFailures() {
        return connectionFailures;
    }

    private int connectionLeaks = 0;

    public int getConnectionLeaks() {
        return connectionLeaks;
    }

    private int servedOpen = 0;

    public int getServedOpen() {
        return servedOpen;
    }

    private int rejectedFull = 0;

    public int getRejectedFull() {
        return rejectedFull;
    }

    private int rejectedTimeout = 0;

    public int getRejectedTimeout() {
        return rejectedTimeout;
    }

    private int rejectedOther = 0;

    public int getRejectedOther() {
        return rejectedOther;
    }

    public int getRejectedOpen() {
        return rejectedFull + rejectedTimeout + rejectedOther;
    }

    private int waitersHigh = 0;

    public int getWaitersHigh() {
        return waitersHigh;
    }

    private int waitersHighRecent = 0;

    public int getWaitersHighRecent() {
        return waitersHighRecent;
    }

    private int totalWaiterCount = 0;

    public int getWaiterCount() {
        return totalWaiterCount;
    }

    private long totalWaitingTime = 0;

    public long getWaitingTime() {
        return totalWaitingTime;
    }

    private long waitingHigh = 0;

    public long getWaitingHigh() {
        return waitingHigh;
    }

    private long waitingHighRecent = 0;

    public long getWaitingHighRecent() {
        return waitingHighRecent;
    }

    // ----------------------------------------------------------------
    // Constructors
    // ----------------------------------------------------------------

    public Pool(ConnectionManager cmgr, XADataSourceImpl xads) throws Exception {
        logger = Log.getLogger(Log.JONAS_DBM_PREFIX);
        this.cmgr = cmgr;
        this.xads = xads;
        try {
            // Modified for use with the gcj compiler
            ServiceManager sm = ServiceManager.getInstance();
            TransactionService ts = (TransactionService) sm.getTransactionService();
            tm = ts.getTransactionManager();
        } catch (Exception e) {
            throw e;
        }
        poolKeeper = new PoolKeeper(this);
        poolKeeper.start();
    }

    // ----------------------------------------------------------------
    // Public Methods
    // ----------------------------------------------------------------

    public int getCurrentOpened() {
        return xac2item.size();
    }

    public int getCurrentBusy() {
        return xac2item.size() - freeList.size();
    }

    public void recomputeBusy() {
        int busy = getCurrentBusy();
        if (busyMax < busy) {
            busyMax = busy;
        }
        if (busyMin > busy) {
            busyMin = busy;
        }
    }

    public int getCurrentInTx() {
        return tx2item.size();
    }

    public synchronized PoolItem openConnection(String user, Transaction tx) throws SQLException {
        PoolItem item = null;

        // If a Connection exists already for this tx, just return it.
        // If no transaction, never reuse a connection already used.
        if (tx != null) {
            item = (PoolItem) tx2item.get(tx);
            if (item != null) {
                logger.log(BasicLevel.DEBUG, "Reuse a Connection for same tx");
                item.open();
                servedOpen++;
                return item;
            }
        }

        // Loop until a valid item is found
        long timetowait = waiterTimeout;
        long starttime = 0;
        while (item == null) {
            // try to find an item in the free list
            if (freeList.isEmpty()) {
                // In case the pool is full, we must wait until a connection is released.
                if (xac2item.size() >= poolMax) {
                    boolean stoplooping = true;
                    // If a timeout has been specified, wait, unless maxWaiters is reached.
                    if (timetowait > 0) {
                        if (currentWaiters < maxWaiters) {
                            currentWaiters++;
                            // Store the maximum concurrent waiters
                            if (waiterCount < currentWaiters) {
                                waiterCount = currentWaiters;
                            }
                            if (starttime == 0) {
                                starttime = System.currentTimeMillis();
                                logger.log(BasicLevel.DEBUG, "Wait for a free Connection" + xac2item.size());
                            }
                            try {
                                wait(timetowait);
                            } catch (InterruptedException ign) {
                                logger.log(BasicLevel.WARN, "Interrupted");
                            } finally {
                                currentWaiters--;
                            }
                            long stoptime = System.currentTimeMillis();
                            long stillwaited = stoptime - starttime;
                            timetowait = waiterTimeout - stillwaited;
                            stoplooping = (timetowait <= 0);
                            if (stoplooping) {
                                // We have been waked up by the timeout.
                                totalWaiterCount++;
                                totalWaitingTime += stillwaited;
                                if (waitingTime < stillwaited) {
                                    waitingTime = stillwaited;
                                }
                            } else {
                                if (!freeList.isEmpty() || xac2item.size() < poolMax) {
                                    // We have been notified by a connection released.
                                    logger.log(BasicLevel.DEBUG, "Notified after " + stillwaited);
                                    totalWaiterCount++;
                                    totalWaitingTime += stillwaited;
                                    if (waitingTime < stillwaited) {
                                        waitingTime = stillwaited;
                                    }
                                }
                                continue;
                            }
                        }
                    }
                    if (stoplooping && freeList.isEmpty() && xac2item.size() >= poolMax) {
                        if (starttime > 0) {
                            rejectedTimeout++;
                            logger.log(BasicLevel.WARN, "Cannot create a Connection - timeout");
                        } else {
                            rejectedFull++;
                            logger.log(BasicLevel.WARN, "Cannot create a Connection");
                        }
                        throw new SQLException("No more connections in " + cmgr.getDatasourceName());
                    }
                    continue;
                }
                logger.log(BasicLevel.DEBUG, "empty free list: Create a new Connection");
                XAConnection xac = null;
                try {
                    // create a new XA Connection
                    xac = xads.getXAConnection();
                    openedCount++;
                } catch (SQLException e) {
                    connectionFailures++;
                    rejectedOther++;
                    logger.log(BasicLevel.WARN, "Cannot create new Connection for tx");
                    throw e;
                }
                item = new PoolItem(this, xac, user);
                // Register the connection manager as a ConnectionEventListener
                xac.addConnectionEventListener(cmgr);
                xac2item.put(xac, item);
            } else {
                item = (PoolItem) freeList.removeFirst();

                // Check the connection before reusing it
                if (checkLevel > 0) {
                    try {
                        ConnectionImpl conn = (ConnectionImpl) item.getXACon().getConnection();
                        if (conn.isPhysicallyClosed()) {
                            logger.log(BasicLevel.WARN, "The JDBC connection has been closed!");
                            destroyItem(item);
                            starttime = 0;
                            item = null;
                            continue;
                        }
                        if (checkLevel > 1) {
                            java.sql.Statement stmt = conn.createStatement();
                            stmt.execute(testStatement);
                            stmt.close();
                        }
                    } catch (Exception e) {
                        logger.log(BasicLevel.ERROR, "DataSource " + cmgr.getDatasourceName()
                                   + " error: removing invalid item:" + e);
                        destroyItem(item);
                        starttime = 0;
                        item = null;
                        continue;
                    }
                }
            }
        }
        recomputeBusy();
        item.setTx(tx);
        if (tx == null) {
            logger.log(BasicLevel.DEBUG, "Got a Connection - no TX: ");
        } else {
            logger.log(BasicLevel.DEBUG, "Got a Connection for TX: ");
            // register synchronization
            try {
                tx.registerSynchronization(item);
                tx2item.put(tx, item);  // only if registerSynchronization was OK.
            } catch (javax.transaction.RollbackException e) {
                logger.log(BasicLevel.WARN, "DataSource " + cmgr.getDatasourceName()
                           + " error: Pool item registered, but tx is rollback only");
            } catch (javax.transaction.SystemException e) {
                logger.log(BasicLevel.ERROR, "DataSource " + cmgr.getDatasourceName()
                           + " error in pool: system exception from transaction manager " + e.getMessage());
            } catch (IllegalStateException e) {
                // In case transaction has already committed, do as if no tx.
                logger.log(BasicLevel.WARN, "Got a Connection - committed TX: ");
                item.setTx(null);
            }
        }
        item.open();
        servedOpen++;
        return item;
    }

    public synchronized void freeConnections(Transaction tx) {
        if (logger.isLoggable(BasicLevel.DEBUG)) {
            logger.log(BasicLevel.DEBUG, "free connection for Tx = " + tx);
        }
        PoolItem item = (PoolItem) tx2item.remove(tx);
        if (item == null) {
            logger.log(BasicLevel.ERROR, "pool: no connection found to free for Tx = " + tx);
            return;
        }
        item.setTx(null);
        if (item.isOpen()) {
            // Connection may be not closed in case of stateful session bean that
            // keeps connection to be reused in another method of the bean.
            logger.log(BasicLevel.WARN, "WARNING: Connection not closed by caller");
            return;
        }
        freeItem(item);
    }

    public synchronized PoolItem closeConnection(XAConnection xac, int flag) {
        PoolItem item = (PoolItem) xac2item.get(xac);
        if (item == null) {
            logger.log(BasicLevel.ERROR, "Pool: no item for this xac!");
            return null;
        }
        // The connection will be available only if not associated
        // to a transaction. Else, it will be reusable only for the
        // same transaction.
        if (!item.close()) {
            return null;
        }
        if (item.getTx() != null) {
            logger.log(BasicLevel.DEBUG, "keep the Connection for this tx");
        } else {
            freeItem(item);
        }

        // delist Resource if in transaction
        Transaction tx = null;
        try {
            tx = tm.getTransaction();
        } catch (NullPointerException n) {
            // current is null: we are not in JOnAS Server.
            logger.log(BasicLevel.ERROR, "Pool: should not be used outside a JOnAS Server");
        } catch (SystemException e) {
            logger.log(BasicLevel.ERROR, "Pool: getTransaction failed:" + e);
        }
        if (tx != null && item.isClosed()) {
            try {
                tx.delistResource(xac.getXAResource(), flag);
            } catch (Exception e) {
                logger.log(BasicLevel.ERROR, "Pool: Exception while delisting resource:" + e);
            }
        }

        return item;
    }

    public synchronized void sampling() {
        waitingHighRecent = waitingTime;
        if (waitingHigh < waitingTime) {
            waitingHigh = waitingTime;
        }
        waitingTime = 0;

        waitersHighRecent = waiterCount;
        if (waitersHigh < waiterCount) {
            waitersHigh = waiterCount;
        }
        waiterCount = 0;

        busyMaxRecent = busyMax;
        busyMax = getCurrentBusy();
        busyMinRecent = busyMin;
        busyMin = getCurrentBusy();
    }

    public synchronized void adjust() {
        logger.log(BasicLevel.DEBUG, "");

        // Remove max aged elements in freelist
        // - Not more than MAX_REMOVE_FREELIST
        // - Don't reduce pool size less than poolMin
        int count = xac2item.size() - poolMin;
        if (count > 0) {
            if (count > MAX_REMOVE_FREELIST) {
                count = MAX_REMOVE_FREELIST;
            }
            for (Iterator i = freeList.iterator(); i.hasNext();) {
                PoolItem item = (PoolItem) i.next();
                if (item.isAged()) {
                    logger.log(BasicLevel.DEBUG, "remove a timed out connection");
                    i.remove();
                    destroyItem(item);
                    count--;
                    if (count <= 0) {
                        break;
                    }
                }
            }
        }
        recomputeBusy();

        // Close (physically) connections lost (opened for too long time)
        for (Iterator i = xac2item.values().iterator(); i.hasNext();) {
            PoolItem item = (PoolItem) i.next();
            if (item.inactive()) {
                logger.log(BasicLevel.WARN, "close a timed out open connection");
                i.remove();
                // destroy item
                item.remove();
                connectionLeaks++;
                // Notify 1 thread waiting for a Connection.
                if (currentWaiters > 0) {
                    notify();
                }
            }
        }

        // Shrink the pool in case of max pool size
        // This occurs when max pool size has been reduced by jonas admin.
        if (poolMax != NO_LIMIT) {
            while (freeList.size() > poolMin && xac2item.size() > poolMax) {
                PoolItem item = (PoolItem) freeList.removeFirst();
                destroyItem(item);
            }
        }
        recomputeBusy();

        // Recreate more Connections while poolMin is not reached
        while (xac2item.size() < poolMin) {
            XAConnection xac = null;
            try {
                xac = xads.getXAConnection();
                openedCount++;
            } catch (SQLException e) {
                throw new ServiceException("Could not create " + poolMin + " items in the pool : " + e.toString());
            }
            // tx = null. Assumes maxage already configured.
            logger.log(BasicLevel.DEBUG, "Create a new available Connection");
            PoolItem item = new PoolItem(this, xac, null);
            freeList.addLast(item);
            xac2item.put(xac, item);
            xac.addConnectionEventListener(cmgr);
        }
    }

    public synchronized void closeAllConnections() {
        logger.log(BasicLevel.DEBUG, "");

        // Stop the pool keeper, since all connections will be closed.
        poolKeeper.stop();

        // Close physically all connections
        Iterator it = xac2item.keySet().iterator();
        try {
            while (it.hasNext()) {
                XAConnection xac = (XAConnection) it.next();
                xac.close();
            }
        } catch (java.sql.SQLException e) {
            logger.log(BasicLevel.ERROR, "Error while closing a Connection:" + e);
        }
    }

    // ----------------------------------------------------------------
    // Private Methods
    // ----------------------------------------------------------------

    private void freeItem(PoolItem item) {
        // Add it to the free list
        // Even if maxage is reached, because we avoids going under min pool size.
        // PoolKeeper will manage aged connections.
        freeList.addLast(item);
        if (logger.isLoggable(BasicLevel.DEBUG)) {
            logger.log(BasicLevel.DEBUG, "item added to freeList: " + freeList.size());
        }
        // Notify 1 thread waiting for a Connection.
        if (currentWaiters > 0) {
            notify();
        }
        recomputeBusy();
    }

    private void destroyItem(PoolItem item) {
        logger.log(BasicLevel.DEBUG, "remove an item in xac2item");
        xac2item.remove(item.getXACon());
        item.remove();
        // Notify 1 thread waiting for a Connection.
        if (currentWaiters > 0) {
            notify();
        }
        recomputeBusy();
    }

}

---