ThreadPool.java

package org.objectweb.jonas_ejb.container;

import java.util.ArrayList;

import org.objectweb.util.monolog.api.BasicLevel;

public class ThreadPool {

    public static final int DEFAULT_POOL_MIN_SIZE = 1;

    public static final int DEFAULT_POOL_MAX_SIZE = 100;

    protected static int poolnumber = 0;
    private static int threadnumber = 0;

    protected int poolsz; // current size of thread pool

    protected int maxpoolsz; // max size of thread pool
    protected int minpoolsz; // min size of thread pool

    protected ArrayList runnablelist = null; // List of runnable ready to run

    protected boolean valid = true; // set to false when pool is removed.
    
    protected int freeThreads;  // free threads ready to work

    private static final int MAX_WAITING_TIME = 20000;  // 20 sec.

    public ThreadPool() {
        this(DEFAULT_POOL_MIN_SIZE, DEFAULT_POOL_MAX_SIZE);
    }

    public ThreadPool(int size) {
        this(size, size > DEFAULT_POOL_MAX_SIZE ? size : DEFAULT_POOL_MAX_SIZE);
    }

    public ThreadPool(int minsz, int maxsz) {
        TraceEjb.thread.log(BasicLevel.DEBUG, "minpoolsize= " + minsz + " maxpoolsize= " + maxsz);
        runnablelist = new ArrayList(minsz);
        poolnumber++;
        for (poolsz = 0; poolsz < minsz; poolsz++) {
            ServerThread st = new ServerThread(this, threadnumber++, poolnumber);
            st.start();
        }
        minpoolsz = minsz;
        maxpoolsz = maxsz;
    }

    public synchronized void addRunnable(Runnable runner) {
        TraceEjb.thread.log(BasicLevel.DEBUG, "");
        runnablelist.add(runner);
        // The policy here is to create 1 more thread when there are more
        // waiting runnable than free threads. 
        // Do not create more than maxpoolsz threads.
        if (TraceEjb.isDebugThread()) {
            TraceEjb.thread.log(BasicLevel.DEBUG, "runnablelist size = " + runnablelist.size() + " with " + freeThreads
                    + " free Server threads");
        }
        if (poolsz < maxpoolsz && runnablelist.size() > freeThreads) {
            poolsz++;
            ServerThread st = new ServerThread(this, threadnumber++, poolnumber);
            st.start();
        } else {
            // Wake up 1 thread if it's already waiting for a Runnable
            notify();
        }
    }

    public synchronized Runnable nextRunnable() {
        Runnable run = null;
        boolean haswait = false;
        while (runnablelist.isEmpty()) {
            try {
                if ((haswait && freeThreads > minpoolsz) || !valid) {
                    poolsz--;
                    TraceEjb.thread.log(BasicLevel.DEBUG, "ending");
                    return null;
                }
                freeThreads++;
                wait(MAX_WAITING_TIME);
                freeThreads--;
                haswait = true;
            } catch (InterruptedException ex) {
                freeThreads--;
                poolsz--;
                TraceEjb.thread.log(BasicLevel.ERROR, "interrupted");
                return null;
            }
        }
        run = (Runnable) runnablelist.remove(0);
        TraceEjb.thread.log(BasicLevel.DEBUG, "got a runnable");
        return run;
    }

    public synchronized void stopThreads() {
        valid = false;
        notifyAll();
        poolnumber--;
    }
    
    protected class ServerThread extends Thread {

        ThreadPool pool;
        int number;

        ServerThread(ThreadPool p, int num, int pinb) {
            pool = p;
            number = num;
            setName("ServerThread-" + pinb + "/" + num);
        }

        public void run() {
            if (TraceEjb.isDebugThread()) {
                TraceEjb.thread.log(BasicLevel.DEBUG, getName() + " started");
            }
            while (true) {
                try {
                    // get next runnable, or wait if empty list.
                    if (TraceEjb.isDebugThread()) {
                        TraceEjb.thread.log(BasicLevel.DEBUG, getName() + " waiting");
                    }
                    Runnable runner = pool.nextRunnable();
                    if (runner == null) {
                        TraceEjb.thread.log(BasicLevel.DEBUG, getName() + " ended");
                        return;
                    }
                    if (TraceEjb.isDebugThread()) {
                        TraceEjb.thread.log(BasicLevel.DEBUG, getName() + " running");
                    }
                    runner.run();
                } catch (Exception ex) {
                }
            }
        }
    }
}

---