org/objectweb/proactive/core/group/threadpool/ThreadPool.java

/* 
 * ################################################################
 * 
 * ProActive: The Java(TM) library for Parallel, Distributed, 
 *            Concurrent computing with Security and Mobility
 * 
 * Copyright (C) 1997-2007 INRIA/University of Nice-Sophia Antipolis
 * Contact: proactive@objectweb.org
 * 
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or any later version.
 *  
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 * 
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307
 * USA
 *  
 *  Initial developer(s):               The ProActive Team
 *                        http://www.inria.fr/oasis/ProActive/contacts.html
 *  Contributor(s): 
 * 
 * ################################################################
 */ 
package org.objectweb.proactive.core.group.threadpool;

import java.util.ArrayList;

import org.objectweb.proactive.core.group.AbstractProcessForGroup;


public class ThreadPool {

    private Thread[] threads = null;

    private ArrayList<AbstractProcessForGroup> pendingJobs = null;

    protected EndControler controler = new EndControler();

    private int memberToThreadRatio = 4;

    private int additionalThreads = 3;

    public ThreadPool() {
        this(1); //+3); //this.additionalThreads
    }

    public ThreadPool(int size) {
        this.threads = new ThreadInThePool[size];
        this.pendingJobs = new ArrayList<AbstractProcessForGroup>(size);
        for (int i = 0; i < this.threads.length; i++) {
            this.threads[i] = new ThreadInThePool(this);
            this.threads[i].start();
        }
    }

    protected void createThreads(int number) {
        this.threads = new ThreadInThePool[number];
        for (int i = 0; i < this.threads.length; i++) {
            this.threads[i] = new ThreadInThePool(this);
            this.threads[i].start();
        }
    }

    public void checkNumberOfThreads(int members) {
        int i;
        int f;

        // System.out.println("ThreadPool: there are " + members + " members in the pool");
        if (this.memberToThreadRatio != 0) {
            f = ((int) Math.ceil(((float) members) / ((float) this.memberToThreadRatio))) +
                this.additionalThreads;
        } else {
            f = this.additionalThreads;
        }

        // System.out.println("ThreadPool: we need " + f + " threads and we have " + this.threads.length);
        if (this.threads.length < f) {
            Thread[] tmp = new Thread[f];
            for (i = 0; i < this.threads.length; i++) {
                tmp[i] = this.threads[i];
            }
            for (; i < f; i++) {
                tmp[i] = new ThreadInThePool(this);
                tmp[i].start();
            }
            this.threads = tmp;
        } else if (this.threads.length > f) {
            Thread[] tmp = new Thread[f];
            for (i = 0; i < f; i++) {
                tmp[i] = this.threads[i];
            }
            for (; i < this.threads.length; i++) {
                this.threads[i].interrupt();
                this.threads[i] = null;
            }
            this.threads = tmp;
        }
    }

    public void ratio(int i) {
        this.memberToThreadRatio = i;
    }

    public void thread(int i) {
        this.additionalThreads = i;
    }

    public synchronized void addAJob(AbstractProcessForGroup r) {
        this.controler.jobStart();
        this.pendingJobs.add(r);
        this.notify();
    }

    public synchronized Runnable getJobForThePendingQueue() {
        try {
            //          System.out.println("ThreadPool.getJobForThePendingQueue() currently " + this.pendingJobs.size() + " in the queue" );
            while (!this.pendingJobs.iterator().hasNext()) {
                this.wait();
                //                      System.out.println("ThreadPool.getJobForThePendingQueue() woken currently " + this.pendingJobs.size() + " in the queue" );
            }

            //          System.out.println("ThreadPool.getJobForThePendingQueue() picking a job from the queue");
            Runnable r = (Runnable) this.pendingJobs.iterator().next();
            this.pendingJobs.remove(r);
            return r;
        } catch (InterruptedException e) {
            this.controler.jobFinish();
            return null;
        }
    }

    public void complete() {
        //this.controler.waitBegin();
        this.controler.waitDone();
    }

    //    /** Cleanly destroys a ThreadPool object */
    //    public void finalize() {
    //        this.controler.reset();
    //        for (int i = 0; i < threads.length; i++) {
    //            this.threads[i].interrupt();
    //            this.controler.jobStart();
    //            // this.threads[i].destroy();   // deprecated
    //        }
    //        this.controler.waitDone();
    //    }

    public void clean() {
        for (int i = 0; i < threads.length; i++) {
            this.threads[i].interrupt();
        }
    }
}

---