core3/util/ThreadPool.java

/*
 * File    : ThreadPool.java
 * Created : 21-Nov-2003
 * By      : parg
 *
 * Azureus - a Java Bittorrent client
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU 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 General Public License for more details ( see the LICENSE file ).
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 */

package org.gudy.azureus2.core3.util;

/**
 * @author parg
 *
 */

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import org.gudy.azureus2.core3.config.COConfigurationManager;
import org.gudy.azureus2.core3.config.ParameterListener;


public class
ThreadPool
{
	private static final boolean	NAME_THREADS = Constants.IS_CVS_VERSION && System.getProperty( "az.thread.pool.naming.enable", "true" ).equals( "true" );

	private static final boolean	LOG_WARNINGS	= false;
	private static final int		WARN_TIME		= 10000;

	private static List		busy_pools			= new ArrayList();
	private static boolean	busy_pool_timer_set	= false;

	private static boolean	debug_thread_pool;
	private static boolean	debug_thread_pool_log_on;

	static{
		if ( System.getProperty("transitory.startup", "0").equals("0")){

			AEDiagnostics.addEvidenceGenerator(
				new AEDiagnosticsEvidenceGenerator()
				{
					public void
					generate(
						IndentWriter		writer )
					{
						writer.println( "Thread Pools" );

						try{
							writer.indent();

							List	pools;

							synchronized( busy_pools ){

								pools	= new ArrayList( busy_pools );
							}

							for (int i=0;i<pools.size();i++){

								((ThreadPool)pools.get(i)).generateEvidence( writer );
							}
						}finally{

							writer.exdent();
						}
					}
				});
		}
	}

	private static ThreadLocal		tls	=
		new ThreadLocal()
		{
			public Object
			initialValue()
			{
				return( null );
			}
		};

	protected static void
	checkAllTimeouts()
	{
		List	pools;

			// copy the busy pools to avoid potential deadlock due to synchronization
			// nestings

		synchronized( busy_pools ){

			pools	= new ArrayList( busy_pools );
		}

		for (int i=0;i<pools.size();i++){

			((ThreadPool)pools.get(i)).checkTimeouts();
		}
	}


	private String	name;
	private int		max_size;
	private int		thread_name_index	= 1;

	private long	execution_limit;

	private List	busy;
	private boolean	queue_when_full;
	private List	task_queue	= new ArrayList();

	private AESemaphore		thread_sem;
	private int				reserved_target;
	private int				reserved_actual;

	private int			thread_priority	= Thread.NORM_PRIORITY;
	private boolean		warn_when_full;

	private long		task_total;
	private long		task_total_last;
	private Average		task_average	= Average.getInstance( WARN_TIME, 120 );

	private boolean		log_cpu	= false || AEThread2.TRACE_TIMES;

	public
	ThreadPool(
		String	_name,
		int		_max_size )
	{
		this( _name, _max_size, false );
	}

	public
	ThreadPool(
		String	_name,
		int		_max_size,
		boolean	_queue_when_full )
	{
		name			= _name;
		max_size		= _max_size;
		queue_when_full	= _queue_when_full;

		thread_sem = new AESemaphore( "ThreadPool::" + name, _max_size );

		busy		= new ArrayList( _max_size );
	}

	private void
	generateEvidence(
		IndentWriter		writer )
	{
		writer.println( name + ": max=" + max_size +",qwf=" + queue_when_full + ",queue=" + task_queue.size() + ",busy=" + busy.size() + ",total=" + task_total + ":" + DisplayFormatters.formatDecimal(task_average.getDoubleAverage(),2) + "/sec");
	}

	public void
	setWarnWhenFull()
	{
		warn_when_full	= true;
	}

	public void
	setLogCPU()
	{
		log_cpu	= true;
	}

	public int
	getMaxThreads()
	{
		return( max_size );
	}

	public void
	setThreadPriority(
		int	_priority )
	{
		thread_priority	= _priority;
	}

	public void
	setExecutionLimit(
		long		millis )
	{
		synchronized( this ){

			execution_limit	= millis;
		}
	}

	public threadPoolWorker run(AERunnable runnable) {
		return( run(runnable, false, false));
	}


	/**
	 *
	 * @param runnable
	 * @param high_priority
	 *            inserts at front if tasks queueing
	 */
	public threadPoolWorker run(AERunnable runnable, boolean high_priority, boolean manualRelease) {

		if(manualRelease && !(runnable instanceof ThreadPoolTask))
			throw new IllegalArgumentException("manual release only allowed for ThreadPoolTasks");
		else if(manualRelease)
			((ThreadPoolTask)runnable).setManualRelease();

		// System.out.println( "Thread pool:" + name + " - sem = " + thread_sem.getValue() + ", queue = " + task_queue.size());

			// not queueing, grab synchronous sem here

		if ( !queue_when_full ){

			if ( !thread_sem.reserveIfAvailable()){

					// defend against recursive entry when in queuing mode (yes, it happens)

				threadPoolWorker	recursive_worker = (threadPoolWorker)tls.get();

				if ( recursive_worker == null || recursive_worker.getOwner() != this ){

						// do a blocking reserve here, not recursive

					checkWarning();

					thread_sem.reserve();

				}else{
						// run immediately

					if ( runnable instanceof ThreadPoolTask ){

						ThreadPoolTask task = (ThreadPoolTask)runnable;

						task.worker = recursive_worker;

						try{
							task.taskStarted();

							runIt( runnable );

							task.join();

						}finally{

							task.taskCompleted();
						}
					}else{

						runIt( runnable );
					}

					return( recursive_worker );
				}
			}
		}

		threadPoolWorker allocated_worker;

		synchronized( this ){

			if ( high_priority )
				task_queue.add( 0, runnable );
			else
				task_queue.add( runnable );

			// reserve if available is non-blocking

			if ( queue_when_full && !thread_sem.reserveIfAvailable()){

				allocated_worker	= null;

				checkWarning();

			}else{

				allocated_worker = new threadPoolWorker();

			}
		}

		return( allocated_worker );
	}

	protected void
	runIt(
		AERunnable	runnable )
	{
		if ( log_cpu ){

			long	start_cpu = log_cpu?AEJavaManagement.getThreadCPUTime():0;
			long	start_time	= SystemTime.getHighPrecisionCounter();

			runnable.run();

			if ( start_cpu > 0 ){

				long	end_cpu = log_cpu?AEJavaManagement.getThreadCPUTime():0;

				long	diff_cpu = ( end_cpu - start_cpu ) / 1000000;

				long	end_time	= SystemTime.getHighPrecisionCounter();

				long	diff_millis = ( end_time - start_time ) / 1000000;

				if ( diff_cpu > 10 || diff_millis > 10){

					System.out.println( TimeFormatter.milliStamp() + ": Thread: " + Thread.currentThread().getName() + ": " + runnable + " -> " + diff_cpu + "/" + diff_millis );
				}
			}
		}else{

			runnable.run();
		}
	}

	protected void checkWarning() {
		if (warn_when_full)
		{
			String task_names = "";
			try
			{
				synchronized (ThreadPool.this)
				{
					for (int i = 0; i < busy.size(); i++)
					{
						threadPoolWorker x = (threadPoolWorker) busy.get(i);
						AERunnable r = x.runnable;
						if (r != null)
						{
							String name;
							if (r instanceof ThreadPoolTask)
								name = ((ThreadPoolTask) r).getName();
							else
								name = r.getClass().getName();
							task_names += (task_names.length() == 0 ? "" : ",") + name;
						}
					}
				}
			} catch (Throwable e)
			{}
			Debug.out("Thread pool '" + getName() + "' is full (busy=" + task_names + ")");
			warn_when_full = false;
		}
	}

	public AERunnable[] getQueuedTasks() {
		synchronized (this)
		{
			AERunnable[] res = new AERunnable[task_queue.size()];
			task_queue.toArray(res);
			return (res);
		}
	}

	public int getQueueSize() {
		synchronized (this)
		{
			return task_queue.size();
		}
	}

	public boolean isQueued(AERunnable task) {
		synchronized (this)
		{
			return task_queue.contains(task);
		}
	}

	public AERunnable[]
	getRunningTasks()
	{
		List	runnables	= new ArrayList();

		synchronized( this ){

			Iterator	it = busy.iterator();

			while( it.hasNext()){

				threadPoolWorker	worker = (threadPoolWorker)it.next();

				AERunnable	runnable = worker.getRunnable();

				if ( runnable != null ){

					runnables.add( runnable );
				}
			}
		}

		AERunnable[]	res = new AERunnable[runnables.size()];

		runnables.toArray(res);

		return( res );
	}

	public int
  	getRunningCount()
  	{
  		int	res = 0;

  		synchronized( this ){

  			Iterator	it = busy.iterator();

  			while( it.hasNext()){

  				threadPoolWorker	worker = (threadPoolWorker)it.next();

  				AERunnable	runnable = worker.getRunnable();

  				if ( runnable != null ){

  					res++;
  				}
  			}
  		}

  		return( res );
  	}

	public boolean
	isFull()
	{
		return( thread_sem.getValue() == 0 );
	}

	public void
	setMaxThreads(
		int		max )
	{
		if ( max > max_size ){

			Debug.out( "should support this sometime..." );

			return;
		}

		setReservedThreadCount( max_size - max );
	}

	public void
	setReservedThreadCount(
		int		res )
	{
		synchronized( this ){

			if ( res < 0 ){

				res = 0;

			}else if ( res > max_size ){

				res = max_size;
			}

			int	 diff =  res - reserved_actual;

			while( diff < 0 ){

				thread_sem.release();

				reserved_actual--;

				diff++;
			}

			while( diff > 0 ){

				if ( thread_sem.reserveIfAvailable()){

					reserved_actual++;

					diff--;

				}else{

					break;
				}
			}

			reserved_target = res;
		}
	}

	protected void
	checkTimeouts()
	{
		synchronized( this ){

			long	diff = task_total - task_total_last;

			task_average.addValue( diff );

			task_total_last = task_total;

			if ( debug_thread_pool_log_on ){

				System.out.println( "ThreadPool '" + getName() + "'/" + thread_name_index + ": max=" + max_size + ",sem=[" + thread_sem.getString() + "],busy=" + busy.size() + ",queue=" + task_queue.size());
			}

			long	now = SystemTime.getMonotonousTime();

			for (int i=0;i<busy.size();i++){

				threadPoolWorker	x = (threadPoolWorker)busy.get(i);

				long	elapsed = now - x.run_start_time;

				if ( elapsed > ( (long)WARN_TIME * (x.warn_count+1))){

					x.warn_count++;

					if ( LOG_WARNINGS ){

						DebugLight.out( x.getWorkerName() + ": running, elapsed = " + elapsed + ", state = " + x.state );
					}

					if ( execution_limit > 0 && elapsed > execution_limit ){

						if ( LOG_WARNINGS ){

							DebugLight.out( x.getWorkerName() + ": interrupting" );
						}

						AERunnable r = x.runnable;

						if ( r != null ){

							try{
								if ( r instanceof ThreadPoolTask ){

									((ThreadPoolTask)r).interruptTask();

								}else{

									x.interrupt();
								}
							}catch( Throwable e ){

								DebugLight.printStackTrace( e );
							}
						}
					}
				}
			}
		}
	}

	public String getName() {
		return (name);
	}

	void releaseManual(ThreadPoolTask toRelease) {
		if( !toRelease.canManualRelease()){
			throw new IllegalStateException("task not manually releasable");
		}

		synchronized( this ){

			long elapsed = SystemTime.getMonotonousTime() - toRelease.worker.run_start_time;
			if (elapsed > WARN_TIME && LOG_WARNINGS)
				DebugLight.out(toRelease.worker.getWorkerName() + ": terminated, elapsed = " + elapsed + ", state = " + toRelease.worker.state);

			if ( !busy.remove(toRelease.worker)){

				throw new IllegalStateException("task already released");
			}

			// if debug is on we leave the pool registered so that we
			// can trace on the timeout events

			if (busy.size() == 0 && !debug_thread_pool){

				synchronized (busy_pools){

					busy_pools.remove(this);
				}
			}

			if ( busy.size() == 0){

				if ( reserved_target > reserved_actual ){

					reserved_actual++;

				}else{

					thread_sem.release();
				}
			}else{

				new threadPoolWorker();
			}
		}

	}

	public void registerThreadAsChild(threadPoolWorker parent)
	{
		if(tls.get() == null || tls.get() == parent)
			tls.set(parent);
		else
			throw new IllegalStateException("another parent is already set for this thread");
	}

	public void deregisterThreadAsChild(threadPoolWorker parent)
	{
		if(tls.get() == parent)
			tls.set(null);
		else
			throw new IllegalStateException("tls is not set to parent");
	}


	class threadPoolWorker extends AEThread2 {
		private final String		worker_name;
		private volatile AERunnable	runnable;
		private long				run_start_time;
		private int					warn_count;
		private String				state	= "<none>";

		protected threadPoolWorker()
		{
			super(NAME_THREADS?(name + " " + (thread_name_index)):name,true);
			thread_name_index++;
			setPriority(thread_priority);
			worker_name = this.getName();
			start();
		}

		public void run() {
			tls.set(threadPoolWorker.this);

			boolean autoRelease = true;

			try
			{
				do
				{
					try
					{
						synchronized (ThreadPool.this)
						{
							if (task_queue.size() > 0)
								runnable = (AERunnable) task_queue.remove(0);
							else
								break;
						}

						synchronized (ThreadPool.this)
						{
							run_start_time = SystemTime.getMonotonousTime();
							warn_count = 0;
							busy.add(threadPoolWorker.this);
							task_total++;
							if (busy.size() == 1)
							{
								synchronized (busy_pools)
								{
									if (!busy_pools.contains(ThreadPool.this))
									{
										busy_pools.add(ThreadPool.this);
										if (!busy_pool_timer_set)
										{
											// we have to defer this action rather
											// than running as a static initialiser
											// due to the dependency between
											// ThreadPool, Timer and ThreadPool again
											COConfigurationManager.addAndFireParameterListeners(new String[] { "debug.threadpool.log.enable", "debug.threadpool.debug.trace" }, new ParameterListener()
											{
												public void parameterChanged(String name) {
													debug_thread_pool = COConfigurationManager.getBooleanParameter("debug.threadpool.log.enable", false);
													debug_thread_pool_log_on = COConfigurationManager.getBooleanParameter("debug.threadpool.debug.trace", false);
												}
											});
											busy_pool_timer_set = true;
											SimpleTimer.addPeriodicEvent("ThreadPool:timeout", WARN_TIME, new TimerEventPerformer()
											{
												public void perform(TimerEvent event) {
													checkAllTimeouts();
												}
											});
										}
									}
								}
							}
						}

						if (runnable instanceof ThreadPoolTask)
						{
							ThreadPoolTask tpt = (ThreadPoolTask) runnable;
							tpt.worker = this;
							String task_name = NAME_THREADS?tpt.getName():null;
							try
							{
								if (task_name != null)
									setName(worker_name + "{" + task_name + "}");
								tpt.taskStarted();
								runIt(runnable);
							} finally
							{
								if (task_name != null)
									setName(worker_name);

								if(tpt.isAutoReleaseAndAllowManual())
									tpt.taskCompleted();
								else
								{
									autoRelease = false;
									break;
								}

							}
						} else
							runIt(runnable);

					} catch (Throwable e)
					{
						DebugLight.printStackTrace(e);
					} finally
					{
						if(autoRelease)
						{
							synchronized (ThreadPool.this)
							{
								long elapsed = SystemTime.getMonotonousTime() - run_start_time;
								if (elapsed > WARN_TIME && LOG_WARNINGS)
									DebugLight.out(getWorkerName() + ": terminated, elapsed = " + elapsed + ", state = " + state);

								busy.remove(threadPoolWorker.this);

								// if debug is on we leave the pool registered so that we
								// can trace on the timeout events
								if (busy.size() == 0 && !debug_thread_pool)
									synchronized (busy_pools)
									{
										busy_pools.remove(ThreadPool.this);
									}
							}
						}
					}
				} while (runnable != null);
			} catch (Throwable e)
			{
				DebugLight.printStackTrace(e);
			} finally
			{
				if ( autoRelease){

					synchronized (ThreadPool.this){

						if ( reserved_target > reserved_actual ){

							reserved_actual++;

						}else{

							thread_sem.release();
						}
					}
				}

				tls.set(null);
			}
		}

		public void setState(String _state) {
			//System.out.println( "state = " + _state );
			state = _state;
		}

		public String getState() {
			return (state);
		}

		protected String getWorkerName() {
			return (worker_name);
		}

		protected ThreadPool getOwner() {
			return (ThreadPool.this);
		}

		protected AERunnable getRunnable() {
			return (runnable);
		}
	}
}