ThreadPool.java

Violations

Source code

/*
 * hgcommons 9
 * Hammurapi Group Common Library
 * Copyright (C) 2003  Hammurapi Group
 *
 * This program 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 of the License, or (at your option) any later version.
 *
 * This program 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
 *
 * URL: http://www.hammurapi.biz/hammurapi-biz/ef/xmenu/hammurapi-group/products/products/hgcommons/index.html
 * e-Mail: support@hammurapi.biz
 */
package biz.hammurapi.util;

import java.util.LinkedList;

import biz.hammurapi.config.ComponentBase;
import biz.hammurapi.config.ConfigurationException;
import biz.hammurapi.metrics.MeasurementConsumer;


/**
 * Distributes work among multiple threads.
 * @author Pavel Vlasov
 * @revision $Revision$
 */
public class ThreadPool extends ComponentBase implements Worker {
	
	private int numberOfThreads=10;
	private int priority=Thread.NORM_PRIORITY; 
	private ExceptionSink exceptionSink;
	private int maxQueue=10;
	private boolean stopped;
	
	public ThreadPool() {
		// Default constructor

	}
	
	/**
	 * @param numberOfThreads Number of threads to create.
	 * @param priority Threads priority.
	 * @param maxQueue Maximum number of jobs in execution queue. When
	 * execution queue reaches its maximum post() processes job in the invoking thread. Values <1 mean no limit.
	 * @param exceptionSink
	 */
	public ThreadPool(int numberOfThreads, int priority, int maxQueue, ExceptionSink exceptionSink) {
		super();
		this.numberOfThreads=numberOfThreads;
		this.priority=priority;
		this.exceptionSink=exceptionSink;
		this.maxQueue=maxQueue;
	}
	
	public ThreadPool(int numberOfThreads, int priority, int maxQueue, ExceptionSink exceptionSink, String name) {
		this(numberOfThreads, priority, maxQueue, exceptionSink);
		if (name!=null) {
			this.name=name;
		}
	}

	private LinkedList jobQueue=new LinkedList();
	private int[] threads={0};
	private String name = toString();
	
	public boolean post(Runnable job) {
		int queueSize;
		synchronized (jobQueue) {
			if (stopped) {
				return false;
			}
			
			queueSize=jobQueue.size();
			
			if (maxQueue==0 || queueSize<maxQueue) {
				// Add job to processing queue

				jobQueue.add(job);			
				jobQueue.notify(); // wake up one thread.

				
				addMeasurement("queue", queueSize, 0);
				addMeasurement("post", 1, 0);
				return true;
			}
		}
		
		addMeasurement("queue", queueSize, 0);
		addMeasurement("post", 1, 0);
		
		// process in current thread.

		process(job);
		return true;
	}
	
	private class WorkerThread extends Thread {
		
		public void run() {
			synchronized (threads) {
				threads[0]++;
			}
			
			try {
				while (true) {
					Runnable job;
					int queueSize;
					synchronized (jobQueue) {
						while (!stopped && jobQueue.isEmpty()) {
							try {
								jobQueue.wait();
							} catch (InterruptedException e) {
								if (exceptionSink==null) {
									e.printStackTrace();
								} else {
									exceptionSink.consume(this, e);
								}
								return;
							}								
						}
						
						if (stopped && jobQueue.isEmpty()) {
							return;
						}
						
						job=(Runnable) jobQueue.removeFirst();
						
						queueSize=jobQueue.size();
					}	
					
					addMeasurement("queue", queueSize, 0);
					
					process(job);
				}											
			} finally {
				synchronized (threads) {
					threads[0]--;
					if (threads[0]<=0) {
						threads.notifyAll();
					}
				}
			}
		}
	}

	public void start() throws ConfigurationException {
		for (int i=0; i<numberOfThreads; i++) {
			Thread th = new WorkerThread();			
			th.setPriority(priority);
			th.setName(this.name +"-pool-thread-"+i);
			th.start();
		}
	}
	
	/**
	 * Creates threads to replace terminated threads.
	 * Client posting jobs to thread pool may implement liveness check routines and
	 * terminate pool threads in a job hangs. This method allows the client code
	 * to replenish the thread pool after termination of one of worker threads.
	 */
	public void replenish() {
		if (!stopped) {
			synchronized (threads) {
				for (int i=threads[0]; i<numberOfThreads; i++) {
					Thread th = new WorkerThread();			
					th.setPriority(priority);
					th.setName(this.name +"-pool-thread-"+i);
					th.start();
				}				
			}
		}
	}

	public void stop() throws ConfigurationException {
		synchronized (threads) {			
			stopped=true;
			synchronized (jobQueue) {
				jobQueue.notifyAll();
			}
			
			while (threads[0]>0) {
				try {
					threads.wait();
				} catch (InterruptedException e) {
					throw new ConfigurationException("Stop() interrupted", e);
				}
			}
		}
	}

	/**
	 * @param job
	 */
	private void process(Runnable job) {
		long start=System.currentTimeMillis();
		try {
			job.run();
		} catch (Exception e) {
			if (exceptionSink==null) {
				e.printStackTrace();
			} else {
				exceptionSink.consume(job, e);
			}
		} finally {
			long now = System.currentTimeMillis();
			long duration = now - start;
			addMeasurement("run", duration, now);
			
			// Jobs can collect metrics if they implement MeasurementConsumer.

			if (job instanceof MeasurementConsumer) {
				((MeasurementConsumer) job).addMeasurement("run", duration, now);
			}
		}
	}

	public int getMaxQueue() {
		return maxQueue;
	}

	public void setMaxQueue(int maxQueue) {
		this.maxQueue = maxQueue;
	}

	public String getName() {
		return name;
	}

	public void setName(String name) {
		this.name = name;
	}

	public int getNumberOfThreads() {
		return numberOfThreads;
	}

	public void setNumberOfThreads(int numberOfThreads) {
		this.numberOfThreads = numberOfThreads;
	}

	public int getPriority() {
		return priority;
	}

	public void setPriority(int priority) {
		this.priority = priority;
	}
	
	public void setExceptionSink(ExceptionSink exceptionSink) {
		this.exceptionSink = exceptionSink;
	}

}