optimizethread.java

这是外国一个开源推理机

package org.openrdf.sesame.sailimpl.rdbms;

/**
 * <p>Title: </p>
 * <p>Description:
 * OptimizeThread class is used to collect the names of tables
 * which should be optimized and perform that operation in a scheduled manner.
 * It is neccessary to introduce such a handling of the table optimizations because
 * there is a significant delay in each SQL query over the table which is being
 * optimized when the size of the repository become larger than 1.5M statements
 * and repository is updated frequently (few times per minute).
 * </p>
 * <p>Copyright: Copyright (c) 2003</p>
 * <p>Company: Ontotext Lab., Sirma AI</p>
 * @author Damyan Ognyanoff
 * @version 1.0
 */
import java.sql.SQLException;
import java.util.HashSet;
import java.util.Iterator;

import org.openrdf.util.log.ThreadLog;

public class OptimizeThread implements Runnable {

	/**
	 * The basic time unit in ms.
	 */
	final static int UNIT = 1000;

	/**
	 * The Sail which associated RDBMS object to use for the real table optimization.
	 */
	private RdfSchemaRepository _sail;

	/**
	 * The active OptimizeThread instance which handles the requests for table optimization.
	 */
	private static OptimizeThread theOnlyOptimizeThread = null;

	/**
	 * How many minutes (time units x 60) to wait before actual table optimization to be performed.
	 */
	private static int minutes = 0;

	/**
	 * Flag indication that the current OptimizeThread to break the waiting loop and
	 * to perform the optimizations requests. Used in forceBreak() and run() methods.
	 */
	private static boolean bForceBreak = true;

	/**
	 * The time in ms on which the Thread is created. Initiaized in constructor.
	 */
	protected long ticksStart = 0;

	/**
	 * A set containing the names of the tables that should be optimized.
	 */
	protected HashSet tables;

	/**
	 * onShutDown() - break the waiting loop of the currently active OptimizeThread
	 * instance and force it to optimize the collected tables. It waits until the
	 * whole optimization finishes.
	 */
	public static void onShutDown() {
		if (theOnlyOptimizeThread != null) {
			theOnlyOptimizeThread.forceBreak();
		}
	}

	/**
	 * Changes the number of time units that the OptimizeThread should wait before
	 * actual optimization is performed.
	 * @param minutes - number of units x 60 to wait
	 */
	public static void setTimeout(int minutes) {
		OptimizeThread.minutes = minutes;
	}

	/**
	 * The entry point of the OptimizeThread class. if there is no pending optimization
	 * this method create a OptimizeThread instance and adds the table name to it.
	 * If the 'minutes' is zero no thread is created and optimization is performed immediatly
	 * @param sail - sail from which the request is made
	 * @param table - the name of the table to optimize
	 */
	public static void optimizeTable(RdfSchemaRepository sail, String table) {
		// in case of zero timie units - perform the optimization immediately
		if (minutes == 0) {
			try {
				sail.getRDBMS().optimizeTable(table);
			}
			catch (SQLException e) {
				ThreadLog.warning("SQLException while optimizing table " + table);
			}
			return;
		}

		// If there is no active instance, create one and register the table to it.
		if (theOnlyOptimizeThread == null) {
			theOnlyOptimizeThread = new OptimizeThread(sail);
			// Add the instance
			theOnlyOptimizeThread.add(table);
			// Start the OptimizeThread
			new Thread(theOnlyOptimizeThread).start();
		}
		else {
			// If there is an active instance, add the table to it
			theOnlyOptimizeThread.add(table);
		}
	}

	/**
	 * constuctor. it is protected
	 * @param sail
	 */
	protected OptimizeThread(RdfSchemaRepository sail) {
		_sail = sail;
		// create
		tables = new HashSet();
		ticksStart = System.currentTimeMillis();
	}

	/**
	 * Add a table for future optimization
	 * @param table
	 */
	synchronized protected void add(String table)	{
		long timeleft = (ticksStart+minutes*60*UNIT) - System.currentTimeMillis();
		System.out.println("table added: "+table+". expected to start after"+timeleft);
		tables.add(table);
	}

	/**
	 * commit() - optimizes the tables invoking the optimizeTable() method from
	 * the RDBMS instance associated with the Sail provided in the constructor.
	 */
	protected void commit()	{
		// Retrieve the RDBMS instance from the sail
		RDBMS rdbms = null;
		if (_sail != null) {
			rdbms = _sail.getRDBMS();
		}

		//enum the table names
		Iterator iter = tables.iterator();
		while (iter.hasNext()) {
			String table = (String)iter.next();
			try {
				// optimize it
				if (rdbms != null) {
					long start = System.currentTimeMillis();
					rdbms.optimizeTable(table);
					System.out.println(table+" optimized for "+(System.currentTimeMillis() - start)+"ms");
				}
				else {
					System.out.println(table);
				}
				if (semafor != null) {
					System.out.println(table+" optimized");
				}
			}
			catch (SQLException e) {
				ThreadLog.warning("SQLException during OPTIMIZE("+table+")");
			}
		}

		// notify the waiting thread(s) if forceBreak() was invoked
		if (semafor != null) {
			semafor.interrupt();
		}
	}

	/**
	 * run() - wait minutes*60 x timeunits ms before commiting the table optimization
	 */
	public void run() {
		synchronized (this) {
			// until not bForceBreak -- changed through forceBreak() method
			while (bForceBreak) {
				try {
					// wait a singe timeuint
					wait(UNIT);
				}
				catch (InterruptedException e) {
					notifyAll();
				}
				// if th ewaiting time elapsed - break the loop
				if ((ticksStart+minutes*60*UNIT) < System.currentTimeMillis()) {
					break;
				}
			}

			// ensure that the it is the main OptimizeThread instance
			if (theOnlyOptimizeThread == this) {
				theOnlyOptimizeThread = null;
			}

			// commit the operation
			commit();
		}
	}

	/**
	 * the semafor which is used to wait until actual commitinf is done so the
	 * forceBreak() to wait until that moment
	 */
	static Thread semafor = null;

	/**
	 * forces the loop breaking of the current OptimizeThread. so all the ables to be optimized
	 */
	public void forceBreak() {
		// set the loop break contition
		bForceBreak = false;

		if (semafor == null) {
			// set the calling thread to be signed in commit()
			try {
				semafor = Thread.currentThread();
				//suspend it until it is notified
				synchronized (semafor) {
					semafor.wait();
				}
			}
			catch (InterruptedException e) {
			}
		}

		// restore the loop break condition
		bForceBreak = true;
	}
}