HArrayPool.java

package org.objectweb.jonas.resource.pool.lib;

import java.util.HashSet;
import java.util.Hashtable;
import java.util.LinkedList;
import java.util.Iterator;
import java.util.Vector;

import javax.resource.spi.ManagedConnection;

import org.objectweb.jonas.resource.pool.api.Pool;
import org.objectweb.jonas.resource.pool.api.PoolItemStats;
import org.objectweb.jonas.resource.pool.api.PoolMatchFactory;
import org.objectweb.util.monolog.api.BasicLevel;
import org.objectweb.util.monolog.api.Logger;

public class HArrayPool implements Pool {

    private Logger logger = null;

    private long timeout = 0;
    private PoolMatchFactory matchFactory = null;
    private HashSet freeList = null;
    private LinkedList linkList = null;
    private HashSet activeList = null;
    private Hashtable infoList = null;

    private static final int NO_LIMIT = -1;

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

    private static final int MAX_REMOVE_FREELIST = 10;

    private int busyMin = 0;

    private int busyMax = 0;

    private int initSize = -1;

    private long maxAge = 0;

    private long maxOpentime = 0;

    private int maxSize = -1;

    private int maxWaiters = 1000;

    private long maxWaitTimeout = 10000;

    private int minSize = 0;

    private HArrayPoolMonitor poolMonitor = null;

    private int samplingPeriod = 60; // defaultSamplingPeriod;

    private int waiterCount = 0;

    private long waitingTime = 0;


    public HArrayPool(Logger logger) {
        this.logger = logger;
        freeList = new HashSet();
        linkList = new LinkedList();
        activeList = new HashSet();
        infoList = new Hashtable();
    }

    // ----------------------------------------------------------------
    // Config properties (Getters & Setters)
    // ----------------------------------------------------------------

    public int getCurrentBusy() {
        return activeList.size();
    }

    public int getCurrentOpened() {
        return getSize();
    }

    public int getInitSize() {
        return initSize;
    }

    public synchronized void setInitSize(int initsize) throws Exception {
        if (matchFactory == null) {
            throw new Exception("The matchFactory is mandatory!");
        }
        if (initSize < 0 && initsize >= 0) {
            initSize = initsize;
            setInitSize();
        }
    }

    private void setInitSize() throws Exception {
        int initsize;

        if (initSize <= 0) {
            return;
        }

        if (maxSize < 0) {
            initsize = initSize;
        } else {
            initsize = initSize < maxSize ? initSize : maxSize;
        }
        initsize = initsize < minSize ? 0 : initsize - minSize;

        ManagedConnection res;
        for (int i = 0; i < initsize; i++) {
            res = createResource(null);
            freeList.add(res);
            linkList.addLast(res);
            infoList.put(res, new PoolItemStats());
        }
    }

    public long getMaxAge() {
        return maxAge;
    }

    public void setMaxAge(long maxAge) {
        this.maxAge = maxAge;
        if (maxOpentime == 0) {
            maxOpentime = maxAge;
        }
    }

    public long getMaxOpentime() {
        return maxOpentime;
    }

    public void setMaxOpentime(long mx) {
        maxOpentime = mx;
    }

    public int getMaxSize() {
        return maxSize;
    }

    public synchronized void setMaxSize(int maxsize) throws Exception {
        if (matchFactory == null) {
            throw new Exception("The matchFactory mandatory!!");
        }
        if (maxsize < minSize || maxsize == maxSize) {
            return;
        }

        // Determine if we need to adjust the pool
        if (maxsize < 0) {
            if (currentWaiters > 0) {
                notify();
            }
            maxSize = maxsize;
        } else {
            if (currentWaiters > 0 && maxSize < maxsize) {
                notify();
            }
            maxSize = maxsize;
            adjust();
        }
    }

    public int getMaxWaiters() {
        return maxWaiters;
    }

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

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

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

    public int getMinSize() {
        return minSize;
    }

    public synchronized void setMinSize(int minsize) throws Exception {
        if (matchFactory == null) {
            throw new Exception("A matchFactory is mandatory!");
        }
        if ((minsize < 0) || (minsize > maxSize) || (minsize == minSize)) {
            return;
        }
        if (minsize < minSize) {
            minSize = minsize;
            return;
        }
        minSize = minsize;
        adjust();
    }

    public int getSamplingPeriod() {
        return samplingPeriod;
    }

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

    public int getSize() {
        return (activeList.size() + freeList.size());
    }

    public long getTimeout() {
        return timeout;
    }

    public synchronized void setTimeout(long crto) {
    }


    // ----------------------------------------------------------------
    // 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;
    }

    // IMPLEMENTATION OF METHODS FROM THE Pool INTERFACE

    public synchronized Object getResource(Object hints)
        throws Exception {

        if (matchFactory == null) {
            throw new Exception("The matchFactory mandatory!!");
        }
        ManagedConnection res = null;
        long timetowait = maxWaitTimeout;
        long starttime = 0;
        while (res == null) {
            if (!freeList.isEmpty()) {
                try {
                    logger.log(BasicLevel.DEBUG, "Free entries available");
                    res = (ManagedConnection) matchFactory.matchResource(freeList, hints);
                    if (res != null) {
                        freeList.remove(res);
                        linkList.remove(res);
                        activeList.add(res);
                    }
                } catch (Exception ex) {
                    logger.log(BasicLevel.DEBUG, "Error from matchResource" + ex);
                }
            }
            if (res == null) {
                int curSize = activeList.size() + freeList.size();
                if (maxSize < 0 || curSize < maxSize) {
                    res = createResource(hints);
                    activeList.add(res);
                } else if (freeList.size() > 0) {
                    res = (ManagedConnection) linkList.getFirst();
                    matchFactory.releaseResource(res);
                    res.destroy();
                    freeList.remove(res);
                    linkList.remove(res);
                    infoList.remove(res);
                    // Create a new one and return it
                    res = createResource(hints);
                    activeList.add(res);
                } else {
                    boolean stoplooping = true;
                    // Determine if waiting is an option
                    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");
                            }
                            try {
                                wait(timetowait);
                            } catch (InterruptedException ign) {
                                logger.log(BasicLevel.WARN, "Interrupted");
                            } finally {
                                currentWaiters--;
                            }
                            long stoptime = System.currentTimeMillis();
                            long stillwaited = stoptime - starttime;
                            timetowait = maxWaitTimeout - stillwaited;
                            stoplooping = (timetowait <= 0);
                            if (stoplooping) {
                                // We have been woken up by the timeout.
                                totalWaiterCount++;
                                totalWaitingTime += stillwaited;
                                if (waitingTime < stillwaited) {
                                    waitingTime = stillwaited;
                                }
                            } else {
                                if (!freeList.isEmpty()) {
                                    // We have been notified by a connection release.
                                    logger.log(BasicLevel.DEBUG, "Notified after " + stillwaited);
                                    totalWaiterCount++;
                                    totalWaitingTime += stillwaited;
                                    if (waitingTime < stillwaited) {
                                        waitingTime = stillwaited;
                                    }
                                }
                                continue;
                            }
                        }
                    }
                    if (stoplooping && freeList.isEmpty()) {
                        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 Exception("No more connections");
                    }
                }
            }
        }
        infoList.put(res, new PoolItemStats());

        printLists();
        setItemStats(res);
        recomputeBusy();
        return res;
    }

    private ManagedConnection createResource(Object hints) throws Exception {
        ManagedConnection res = null;
        try {
            res = (ManagedConnection) matchFactory.createResource(hints);
            if (res == null) {
                Exception exc = new Exception("A null ManagedConnection was returned.");
                throw exc;
            }
            openedCount++;
        } catch (Exception ex) {
            connectionFailures++;
            rejectedOther++;
            logger.log(BasicLevel.DEBUG, "Cannot create new Connection: " + ex);
            throw ex;
        }
        logger.log(BasicLevel.DEBUG, "Created Resource: " + res);
        return res;
    }

    public synchronized void releaseResource(Object resource, boolean destroy, boolean adjustment)
        throws Exception {

        ManagedConnection res = (ManagedConnection) resource;
        if (matchFactory == null) {
            throw new Exception("The matchFactory mandatory!!");
        }
        if (activeList == null) {
            throw new Exception("No active resources to releases!!");
        }
        if (!activeList.contains(res)) {
            //Temp fix making the assumption that the item is already released
            return;
            //throw new Exception("Attempt to release inactive resource(" + res + ")");
        }
        activeList.remove(res);
        if (!destroy) {
            freeList.add(res);
            linkList.addLast(res);
            PoolItemStats pis = (PoolItemStats) infoList.get(res);
            if (pis != null) {
                pis.setTotalConnectionTime(System.currentTimeMillis() - pis.getStartTime());
            }
        } else {
            infoList.remove(res);
            res.destroy();
        }
        // Notify 1 thread waiting for a Connection.
        if (currentWaiters > 0) {
            notify();
        }
        if (adjustment) {
            adjust();
        }
    }

    public PoolMatchFactory getMatchFactory() {
        return matchFactory;
    }

    public synchronized void setMatchFactory(PoolMatchFactory pmf) {
        matchFactory = pmf;
    }

    public void startMonitor() {
        poolMonitor = new HArrayPoolMonitor(this);
        poolMonitor.start();
    }

    public void validateMCs() throws Exception {
        matchFactory.validateResource(freeList);
    }

    public synchronized void adjust() throws Exception {
        // Remove max aged elements in freelist
        // - Not more than MAX_REMOVE_FREELIST
        // - Don't reduce pool size less than minSize
        //logger.log(BasicLevel.DEBUG, "");
        int count = activeList.size() - minSize;
        long curTime = System.currentTimeMillis();
        if (count > 0) {
            if (count > MAX_REMOVE_FREELIST) {
                count = MAX_REMOVE_FREELIST;
            }
            Vector rList = new Vector();
            Iterator it = null;
            for (it = freeList.iterator(); it.hasNext();) {
                ManagedConnection res = (ManagedConnection) it.next();
                PoolItemStats pis = (PoolItemStats) infoList.get(res);
                if (maxOpentime > 0 && pis != null && pis.getMaxOpenTimeout() > 0
                        && curTime > pis.getMaxOpenTimeout()) {
                    rList.add(res);
                }
            }
            // Bug: 300351 Use the list built above to remove the MCs
            it = rList.iterator();
            while (it.hasNext()) {
                ManagedConnection res = (ManagedConnection) it.next();
                try {
                    matchFactory.releaseResource(res);
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
                freeList.remove(res);
                try {
                    res.destroy();
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
                linkList.remove(res);
            }
        }

        // Close (physically) connections lost (opened for too long time)
        curTime = System.currentTimeMillis();
        Vector aList = new Vector();
        Iterator it = null;
        for (it = activeList.iterator(); it.hasNext();) {
            ManagedConnection res = (ManagedConnection) it.next();
            PoolItemStats pis = (PoolItemStats) infoList.get(res);
            if (maxOpentime > 0 && pis != null && pis.getMaxOpenTimeout() > 0
                    && curTime > pis.getMaxOpenTimeout()) {
                aList.add(res);
            }
        }
        it = aList.iterator();
        while (it.hasNext()) {
            ManagedConnection item = (ManagedConnection) it.next();
            logger.log(BasicLevel.WARN, "close a timed out active connection");
            logger.log(BasicLevel.WARN, "MC = " + item);
            releaseResource(item, true, false);
            connectionLeaks++;
        }


        int curSize = activeList.size() + freeList.size();
        if (maxSize > 0 && maxSize < curSize) {
            // Removes as many free entries as possible
            int nbRemove = curSize - maxSize;
            if (freeList != null) {
                while (!freeList.isEmpty() && nbRemove > 0) {
                    ManagedConnection res = (ManagedConnection) linkList.getFirst();
                    matchFactory.releaseResource(res);
                    res.destroy();
                    freeList.remove(res);
                    linkList.remove(res);
                    infoList.remove(res);
                    nbRemove--;
                    curSize--;
                }
            }
        }

        // Recreate more Connections while minSize is not reached
        ManagedConnection res;
        while (minSize > curSize) {
            res = createResource(null);
            freeList.add(res);
            linkList.addLast(res);
            infoList.put(res, new PoolItemStats());
            curSize++;
        }

        recomputeBusy();
    }

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

    public void sampling() throws Exception {
        //matchFactory.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();
    }

    private void setItemStats(Object res) {
        PoolItemStats pis = (PoolItemStats) infoList.get(res);
        if (pis != null) {
            pis.incrementUses();
            long sTime = System.currentTimeMillis();
            pis.setStartTime(sTime);
            if (maxAge > 0 && pis.getMaxAgeTimeout() == 0) {
                pis.setMaxAgeTimeout(sTime + maxAge);
            }
            if (maxOpentime > 0) {
                pis.setMaxOpenTimeout(sTime + maxOpentime);
            }
        }
        servedOpen++;
    }

    void printLists() {
        int count = 0;
        if (logger.isLoggable(BasicLevel.DEBUG)) {
            logger.log(BasicLevel.DEBUG, "minSize=" + minSize + ", maxSize=" + maxSize
                                         + ",  freeSize=" + freeList.size());
            logger.log(BasicLevel.DEBUG, "activeList:");
            if (activeList == null) {
                logger.log(BasicLevel.DEBUG, " null");
            } else {
                Iterator it = activeList.iterator();
                while (it.hasNext() && ++count < 40) {
                    logger.log(BasicLevel.DEBUG, " " + (ManagedConnection) it.next());
                }
            }
            logger.log(BasicLevel.DEBUG, "freeList:");
            if (freeList == null) {
                logger.log(BasicLevel.DEBUG, " null");
            } else {
                count = 0;
                Iterator it = freeList.iterator();
                while (it.hasNext() && ++count < 40) {
                    logger.log(BasicLevel.DEBUG, " " + (ManagedConnection) it.next());
                }
            }
        }
    }

}

---