/*
 * Copyright (C) 2005 NTT DATA Corporation
 * 
 */
package org.postgresforest.vm;

import java.sql.Connection;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;

import org.postgresforest.Driver;
import org.postgresforest.jdbc2.AbstractJdbc2Connection;


/**
 *　SQL実行スレッド管理クラス
 * @since 1.1
 */
public class StatementThreadMng {

	protected int m_max;
	protected LinkedList available = new LinkedList();
	protected HashMap m_execSeqMap = new HashMap();//コネクションごとのクエリー実行順格納用
	protected int used;

	/** ログ出力 @since 2.1 */
	protected LogUtil m_logUtil;

	public StatementThreadMng(int threadNum) {

		m_max = threadNum;

	}

	/**
	 * 
	 * @return
	 */
	synchronized public StatementSub getStatementSub(){

		StatementSub thread;

		while (true)
		{
			if (available.size() > 0)
			{
				thread = (StatementSub)available.removeFirst();
				used++;
				break;
			}

			if (m_max == 0 || used < m_max)
			{
				thread = new StatementSub(this);
				used++;
				break;
			}
			else
			{
				if (Driver.logInfo)
					m_logUtil.info("There is no empty thread. It stands by.");
				try
				{
					//Ver1.1R2 パフォーマンスチューニング
					//wait(5);			
					wait();			
				}
				catch (InterruptedException e)
				{}
			}
		}

		
		this.notify();

		if (Driver.logInfo)
			m_logUtil.info(thread.getName() + ": The thread was secured. ");

		return thread;
		
	}
	/**
	 * スレッド開放処理
	 * @param thread
	 */
	synchronized public void closeStatementSub(StatementSub thread){
		
		if (Driver.logDebug)
			m_logUtil.debug(thread.getName() + ": closeStatementSub() In");

		synchronized(thread){
		
			if(thread.isExecuted()){//すでに処理が終了している場合

				//スレッドの開放処理
				thread.close();
				
				//使用中から削除		
				used--;
				//利用可能にpush
				available.addFirst(thread);

				//Ver1.1R2 パフォーマンスチューニング 空きスレッドができたことを通知
				//Ver2.1
				this.notify();
				//this.notifyAll();

				if (Driver.logInfo)
					m_logUtil.info(thread.getName() + ": The thread was opened wide.");
	
			}else{//まだスレッドが処理中
				
				thread.setSelfClose(true);
				if (Driver.logInfo)
					m_logUtil.info(thread.getName() + ": Opening of a thread was reserved.");
			}

		}

		if (Driver.logDebug)
			m_logUtil.debug(thread.getName() + ": closeStatementSub() Out");
		
	}
	
	/**
	 * スレッド終了
	 *
	 */
	synchronized public void terminate(){
		
		if (Driver.logDebug)
			m_logUtil.debug("StatementThreadMng: terminate() Ready");

		//実行中スレッドを終了
		while(used > 0){
			try {
				if (Driver.logDebug)
					m_logUtil.debug("StatementThreadMng: terminate() Wait");
				//Ver1.1R パフォーマンスチューニング
				//wait(10);
				//Ver2.1 パフォーマンスチューニング
				//wait(5);
				wait();
			} catch (InterruptedException e) {}
		}

		if (Driver.logDebug)
			m_logUtil.debug("StatementThreadMng: terminate() Terminate Available");

		//待機中のスレッドを終了
		StatementSub thread;
		for (Iterator iter = available.iterator(); iter.hasNext();) {
			thread = (StatementSub) iter.next();
			thread.terminate();
		}

		if (Driver.logDebug)
			m_logUtil.debug("StatementThreadMng: terminate() End");
	}
	
	/**
	 * 実行中のスレッドがなくなるまで待機
	 */
	synchronized public void waitFree(){

		while (true)
		{

			if( used == 0){
				break; 
			}

			try {
				if (Driver.logDebug)
					Driver.debug("StatementThreadMng::waitFree() Waiting");
				wait();			
			} catch (InterruptedException e) {
				break;
			}
		}
	}
	/**
	 * @param util
	 */
	public void setLogUtil(LogUtil util) {
		m_logUtil = util;
	}

	/**
	 * @return
	 */
	public LogUtil getLogUtil() {
		return m_logUtil;
	}
	/**
	 * コネクション実行順設定
	 * @param obj（SQL実行スレッド）
	 */
	public void setExecSeq(Connection con,StatementSub thread){

		synchronized(m_execSeqMap) {
			LinkedList execSeq = ( LinkedList)m_execSeqMap.get(con);
			if(execSeq == null){
				execSeq = new LinkedList();
				execSeq.addLast(thread);
				m_execSeqMap.put(con,execSeq);
			}else{
				synchronized(execSeq){
					execSeq.addLast(thread);
				}
			}
		}
	}

	/**
	 * 指定したユーザDBへの処理待機リストを取得し、
	 * 指定したスレッドが実行中（リストの先頭に存在）だったら
	 * リストから削除する。
	 *
	 * @param Connection con 終了処理の対象とするユーザDBへのConnection
	 * @param StatementSub thread 実行を終了するスレッド
	 */
	public void endExecSeq(Connection con, StatementSub thread){
		/*
		 * ユーザDBへのConnectionの取得待ち（実行待ち）のスレッドを格納している
		 * 待機リストを取得。
		 */
		LinkedList execSeq = ( LinkedList)m_execSeqMap.get(con);
		 
		synchronized(execSeq){
			/*
			 * すでに待機リストが空だった場合は何もせず終了
			 *
			 * FIXME: すべきことはない？
			 */
			if ( execSeq.isEmpty() )
			{
				return;
			}

			/*
			 * リストの先頭のスレッドが実行中のものと同じであれば、
			 * そのスレッドをリストから削除。
			 */
			Object executing = execSeq.getFirst();
			if(thread.equals(executing)  ){
				execSeq.removeFirst();
			}
			else
			{
				/* FIXME: ここですべきことは？ */
			}

			/*
			 * 待機リスト内に実行待ちスレッドが残っていたら、
			 * リスト（全スレッドに）notify
			 */
			if(execSeq.size() > 0){
				execSeq.notifyAll();
			} 
 		}
		 
	}

	/**
	 * 指定したユーザDBへの処理待機リストを取得し、
	 * 指定したスレッドの実行順が来るまで待機する。
	 *
	 * @param Connection con 終了処理の対象とするユーザDBへのConnection
	 * @param StatementSub thread 実行を開始するスレッド
	 */
	public void waitExecSeq(Connection con,StatementSub thread){
		/*
		 * ユーザDBへのConnectionの取得待ち（実行待ち）のスレッドを格納している
		 * リンクリスト（行列）を取得。
		 */
		LinkedList execSeq = (LinkedList) m_execSeqMap.get(con);

		synchronized (execSeq) {

			while (execSeq.size() != 0) {
				/*
				 * 指定したスレッドが待機リストの先頭にいたら処理を開始
				 */
				if (thread.equals(execSeq.getFirst())) {
					return;
				}

				if (Driver.forestTestLog) {
					try {
						System.out.println(((Thread) thread).getName()
								+ ":CON_SEQ_WAIT:"
								+ ((AbstractJdbc2Connection) con).getURL());
					} catch (SQLException e1) {
					}
				}

				/*
				 * 待機リストが更新されるのを待つ。
				 * 通知されたらループの先頭に戻って再トライ。
				 */
				try {
					execSeq.wait();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}

		}
			

	}

}
